126 lines
5.2 KiB
Python
126 lines
5.2 KiB
Python
from __future__ import annotations
|
|
|
|
from collections import Counter, defaultdict
|
|
|
|
from .models import AnomalyFinding, LogEvent, Recommendation
|
|
from .threat_intel import is_public_ip
|
|
|
|
|
|
def _severity(score: int) -> str:
|
|
if score >= 80:
|
|
return "critical"
|
|
if score >= 60:
|
|
return "high"
|
|
if score >= 35:
|
|
return "medium"
|
|
return "low"
|
|
|
|
|
|
def _top_values(events: list[LogEvent], field: str, limit: int = 5) -> list[str]:
|
|
counter: Counter[str] = Counter()
|
|
for event in events:
|
|
value = event.fields.get(field)
|
|
if value:
|
|
counter[value] += 1
|
|
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:
|
|
if event.src_ip:
|
|
grouped[event.src_ip].append(event)
|
|
return grouped
|
|
|
|
|
|
def build_recommendations(
|
|
events: list[LogEvent],
|
|
anomalies: list[AnomalyFinding],
|
|
reputation: dict[str, dict[str, object]] | None = None,
|
|
) -> list[Recommendation]:
|
|
reputation = reputation or {}
|
|
grouped = _events_by_src(events)
|
|
recommendations: list[Recommendation] = []
|
|
|
|
for anomaly in anomalies:
|
|
src_events = grouped.get(anomaly.subject, [])
|
|
if not src_events:
|
|
continue
|
|
|
|
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)]
|
|
bad_reputation = [
|
|
f"{ip}:score={intel.get('score')} status={intel.get('status')}"
|
|
for ip, intel in reputation.items()
|
|
if int(intel.get("score", 0) or 0) >= 50 and (ip == anomaly.subject or ip in dst_ips)
|
|
]
|
|
|
|
score = anomaly.score
|
|
reasons = list(anomaly.reasons)
|
|
if bad_reputation:
|
|
score = min(100, score + 20)
|
|
reasons.append(f"threat intelligence hit ({'; '.join(bad_reputation[:3])})")
|
|
|
|
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 "
|
|
"temporarily if the traffic is unsolicited or UTM-confirmed. Convert to a permanent "
|
|
"address object/block only after reputation and business impact are verified."
|
|
)
|
|
elif public_dst and bad_reputation:
|
|
title = "Investigate risky destination reputation"
|
|
action = (
|
|
"Inspect the affected internal client, DNS history, and policy path. Consider blocking "
|
|
"the destination with an address object or ISDB/category control if reputation remains malicious."
|
|
)
|
|
elif anomaly.evidence.get("distinct_destinations", 0) >= 10:
|
|
title = "Investigate scan-like or fan-out traffic"
|
|
action = (
|
|
"Inspect the source host and the matching policies. If this is not expected discovery or monitoring, "
|
|
"limit allowed destinations/services and add IPS/application control on the policy."
|
|
)
|
|
elif anomaly.evidence.get("distinct_services", 0) >= 8:
|
|
title = "Restrict broad service usage"
|
|
action = (
|
|
"Review the matching policy services. Replace broad service objects such as ALL with the observed "
|
|
"business-required services only."
|
|
)
|
|
elif anomaly.evidence.get("utm_events", 0):
|
|
title = "Review UTM-triggering traffic"
|
|
action = (
|
|
"Inspect the IPS/AV/WebFilter event details and policy. Keep or enable UTM profiles on this policy, "
|
|
"and tighten source/destination scope if the traffic is not expected."
|
|
)
|
|
else:
|
|
title = "Review traffic anomaly"
|
|
action = "Inspect the related policy and host behavior before changing enforcement."
|
|
|
|
recommendations.append(
|
|
Recommendation(
|
|
subject=anomaly.subject,
|
|
score=min(100, score),
|
|
severity=_severity(min(100, score)),
|
|
title=title,
|
|
recommendation=action,
|
|
reasons=reasons,
|
|
related_policy_ids=policy_ids,
|
|
related_services=services,
|
|
)
|
|
)
|
|
|
|
return sorted(recommendations, key=lambda item: item.score, reverse=True)
|