Public/fgAI
SHA256
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implemented cleanup/triage direction. ui, and baselinbe days

Browse Source
This commit is contained in:
larssand
2026-06-29 20:20:00 +02:00
parent d764c5a038
commit d540b8a77d
9 changed files with 298 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
src/fgai/threat_intel.py
View File

@@ -19,16 +19,28 @@ def is_public_ip(value: str | None) -> bool:
class ThreatIntelClient:
def __init__(self, *, cache_file: str = "state/threat-intel-cache.json", ttl_seconds: int | None = None, enabled: bool | None = None) -> None:
def __init__(
self,
*,
cache_file: str = "state/threat-intel-cache.json",
ttl_seconds: int | None = None,
enabled: bool | None = None,
provider: str | None = None,
abuseipdb_key: str | None = None,
virustotal_key: str | None = None,
daily_limit: int | None = None,
error_ttl_seconds: int | None = None,
abuseipdb_max_age_days: int | None = None,
) -> None:
self.enabled = os.getenv("FGAI_THREAT_INTEL", "").lower() in {"1", "true", "yes", "on"} if enabled is None else enabled
self.abuseipdb_key = os.getenv("ABUSEIPDB_API_KEY")
self.virustotal_key = os.getenv("VIRUSTOTAL_API_KEY")
self.provider = os.getenv("FGAI_THREAT_INTEL_PROVIDER", "auto").lower()
self.max_age_days = int(os.getenv("ABUSEIPDB_MAX_AGE_DAYS", "90"))
self.abuseipdb_key = abuseipdb_key if abuseipdb_key is not None else os.getenv("ABUSEIPDB_API_KEY")
self.virustotal_key = virustotal_key if virustotal_key is not None else os.getenv("VIRUSTOTAL_API_KEY")
self.provider = (provider if provider is not None else os.getenv("FGAI_THREAT_INTEL_PROVIDER", "auto")).lower()
self.max_age_days = abuseipdb_max_age_days if abuseipdb_max_age_days is not None else int(os.getenv("ABUSEIPDB_MAX_AGE_DAYS", "90"))
self.cache_path = Path(cache_file)
self.ttl_seconds = ttl_seconds if ttl_seconds is not None else int(os.getenv("FGAI_THREAT_INTEL_TTL_SECONDS", "604800"))
self.error_ttl_seconds = int(os.getenv("FGAI_THREAT_INTEL_ERROR_TTL_SECONDS", "3600"))
self.daily_limit = int(os.getenv("FGAI_THREAT_INTEL_DAILY_LIMIT", "100"))
self.error_ttl_seconds = error_ttl_seconds if error_ttl_seconds is not None else int(os.getenv("FGAI_THREAT_INTEL_ERROR_TTL_SECONDS", "3600"))
self.daily_limit = daily_limit if daily_limit is not None else int(os.getenv("FGAI_THREAT_INTEL_DAILY_LIMIT", "100"))
self.cache = self._read_cache()
def status(self) -> dict[str, object]:
@@ -196,9 +208,20 @@ class ThreatIntelClient:
def enrich_ips(
ips: list[str], *, cache_file: str = "state/threat-intel-cache.json", limit: int = 25, enabled: bool | None = None
ips: list[str], *, cache_file: str = "state/threat-intel-cache.json", limit: int = 25, enabled: bool | None = None, config: dict[str, object] | None = None
) -> dict[str, dict[str, object]]:
client = ThreatIntelClient(cache_file=cache_file, enabled=enabled)
config = config or {}
client = ThreatIntelClient(
cache_file=cache_file,
enabled=enabled,
provider=str(config.get("threat_intel_provider", "auto")),
abuseipdb_key=str(config.get("abuseipdb_api_key", "") or "") or None,
virustotal_key=str(config.get("virustotal_api_key", "") or "") or None,
daily_limit=int(config.get("threat_intel_daily_limit", 100) or 100),
ttl_seconds=int(config.get("threat_intel_ttl_seconds", 604800) or 604800),
error_ttl_seconds=int(config.get("threat_intel_error_ttl_seconds", 3600) or 3600),
abuseipdb_max_age_days=int(config.get("abuseipdb_max_age_days", 90) or 90),
)
enriched: dict[str, dict[str, object]] = {}
for ip in ips[:limit]:
enriched[ip] = client.lookup_ip(ip)