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

policy 0 is deny

Browse Source
This commit is contained in:
larssand
2026-06-18 23:07:02 +02:00
parent 6530e51700
commit 6117013c6e
3 changed files with 36 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
src/fgai/anomaly.py
View File

@@ -76,7 +76,8 @@ def detect_source_anomalies(events: list[LogEvent], *, limit: int = 20) -> list[
deny_count = sum(1 for event in src_events if event.action in THREAT_ACTIONS)
utm_count = sum(1 for event in src_events if is_utm_event(event))
high_severity_count = sum(1 for event in src_events if event.severity in {"critical", "high", "alert", "emergency"})
policies = {event.fields.get("policyid") for event in src_events if event.fields.get("policyid")}
policies = {event.fields.get("policyid") for event in src_events if event.fields.get("policyid") and event.fields.get("policyid") != "0"}
implicit_deny_count = sum(1 for event in src_events if event.fields.get("policyid") == "0")
reasons: list[str] = []
score = 0
@@ -105,6 +106,10 @@ def detect_source_anomalies(events: list[LogEvent], *, limit: int = 20) -> list[
score += min(20, 8 + int(deny_rate * 12))
reasons.append(f"high deny/threat-action rate ({deny_count}/{event_count})")
if implicit_deny_count >= 10:
score += min(15, 5 + implicit_deny_count // 10)
reasons.append(f"implicit FortiGate deny/drop hits observed (policyid=0, {implicit_deny_count} events)")
if utm_count:
utm_score = sum(event_score(event) for event in src_events if is_utm_event(event))
points = min(35, 5 + utm_score)
@@ -143,6 +148,7 @@ def detect_source_anomalies(events: list[LogEvent], *, limit: int = 20) -> list[
"high_severity_events": high_severity_count,
"total_bytes": total_bytes,
"policy_count": len(policies),
"implicit_deny_events": implicit_deny_count,
},
)
)

15
src/fgai/recommendations.py
View File

@@ -25,6 +25,10 @@ def _top_values(events: list[LogEvent], field: str, limit: int = 5) -> list[str]
return [value for value, _ in counter.most_common(limit)]
def _top_policy_values(events: list[LogEvent], limit: int = 5) -> list[str]:
return [value for value in _top_values(events, "policyid", limit=limit + 1) if value != "0"][:limit]
def _events_by_src(events: list[LogEvent]) -> dict[str, list[LogEvent]]:
grouped: dict[str, list[LogEvent]] = defaultdict(list)
for event in events:
@@ -47,7 +51,7 @@ def build_recommendations(
if not src_events:
continue
policy_ids = _top_values(src_events, "policyid")
policy_ids = _top_policy_values(src_events)
services = _top_values(src_events, "service")
dst_ips = [event.dst_ip for event in src_events if event.dst_ip]
public_dst = [ip for ip in _top_values(src_events, "dstip", limit=10) if is_public_ip(ip)]
@@ -63,7 +67,14 @@ def build_recommendations(
score = min(100, score + 20)
reasons.append(f"threat intelligence hit ({'; '.join(bad_reputation[:3])})")
if is_public_ip(anomaly.subject) and anomaly.score >= 60:
if anomaly.evidence.get("implicit_deny_events", 0) and not policy_ids:
title = "Implicit deny/drop traffic observed"
action = (
"FortiGate policyid=0 is the implicit deny/drop path, not an editable firewall policy. "
"If this traffic is expected, create a narrow explicit allow policy above the deny using the observed "
"source, destination, and service. If it is not expected, keep the deny and investigate or reduce noisy logging."
)
elif is_public_ip(anomaly.subject) and anomaly.score >= 60:
title = "Quarantine or block suspicious public source"
action = (
"Inspect the matching FortiGate logs and policy IDs, then quarantine the source IP "