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

add more

Browse Source
This commit is contained in:
larssand
2026-06-22 21:20:43 +02:00
parent d94cb20987
commit c4f2751ee4
3 changed files with 37 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
src/fgai/anomaly.py
View File

@@ -75,9 +75,10 @@ def _rate_per_minute(events: list[LogEvent]) -> tuple[float | None, float]:
def detect_source_anomalies(
events: list[LogEvent], *, limit: int = 20, baselines: dict[str, dict[str, object]] | None = None
events: list[LogEvent], *, limit: int = 20, baselines: dict[str, dict[str, object]] | None = None, field_deviations: dict[str, list[dict[str, object]]] | None = None
) -> list[AnomalyFinding]:
baselines = baselines or {}
field_deviations = field_deviations or {}
by_src: dict[str, list[LogEvent]] = defaultdict(list)
for event in events:
if event.src_ip:
@@ -217,6 +218,10 @@ def detect_source_anomalies(
score += 10
reasons.append("public source with repeated security-relevant events")
for deviation in field_deviations.get(src_ip, []):
score += int(deviation.get("score", 0))
reasons.append(str(deviation.get("reason", "stream field baseline deviation")))
if not reasons:
continue