from __future__ import annotations import ipaddress from collections import Counter, defaultdict from datetime import datetime from statistics import mean, pstdev from .logs import THREAT_ACTIONS, event_score, is_utm_event from .models import AnomalyFinding, LogEvent def _as_int(value: str | None) -> int: if not value: return 0 try: return int(float(value)) except ValueError: return 0 def _severity(score: int) -> str: if score >= 80: return "critical" if score >= 60: return "high" if score >= 35: return "medium" return "low" def _confidence(event_count: int, reason_count: int) -> str: if event_count >= 20 and reason_count >= 3: return "high" if event_count >= 5 and reason_count >= 2: return "medium" return "low" def _is_public_ip(value: str) -> bool: try: return ipaddress.ip_address(value).is_global except ValueError: return False def _event_timestamp(event: LogEvent) -> float | None: """Return the FortiGate event time in seconds when present.""" eventtime = event.fields.get("eventtime") if eventtime: try: value = float(eventtime) # Exports can use seconds, milliseconds, microseconds, or nanoseconds. while value > 10_000_000_000: value /= 1_000 return value except ValueError: pass date = event.fields.get("date") clock = event.fields.get("time") if date and clock: try: return datetime.fromisoformat(f"{date}T{clock}").timestamp() except ValueError: return None return None def _rate_per_minute(events: list[LogEvent]) -> tuple[float | None, float]: timestamps = [timestamp for event in events if (timestamp := _event_timestamp(event)) is not None] if len(timestamps) < 2: return None, 0.0 duration_seconds = max(timestamps) - min(timestamps) return len(timestamps) * 60 / max(1.0, duration_seconds), duration_seconds def detect_source_anomalies(events: list[LogEvent], *, limit: int = 20) -> list[AnomalyFinding]: by_src: dict[str, list[LogEvent]] = defaultdict(list) for event in events: if event.src_ip: by_src[event.src_ip].append(event) if not by_src: return [] event_counts = [len(src_events) for src_events in by_src.values()] distinct_dst_counts = [ len({event.fields.get("dstip") for event in src_events if event.fields.get("dstip")}) for src_events in by_src.values() ] byte_totals = [ sum(_as_int(event.fields.get("sentbyte")) + _as_int(event.fields.get("rcvdbyte")) for event in src_events) for src_events in by_src.values() ] hitcount_totals = [sum(_as_int(event.fields.get("hitcount")) for event in src_events) for src_events in by_src.values()] source_rates = [rate for src_events in by_src.values() if (rate := _rate_per_minute(src_events)[0]) is not None] avg_events = mean(event_counts) std_events = pstdev(event_counts) or 1.0 avg_dst = mean(distinct_dst_counts) std_dst = pstdev(distinct_dst_counts) or 1.0 avg_bytes = mean(byte_totals) std_bytes = pstdev(byte_totals) or 1.0 avg_hitcount = mean(hitcount_totals) std_hitcount = pstdev(hitcount_totals) or 1.0 avg_rate = mean(source_rates) if source_rates else 0.0 std_rate = (pstdev(source_rates) or 1.0) if source_rates else 1.0 findings: list[AnomalyFinding] = [] for src_ip, src_events in by_src.items(): event_count = len(src_events) distinct_dst = len({event.fields.get("dstip") for event in src_events if event.fields.get("dstip")}) distinct_services = len({event.fields.get("service") for event in src_events if event.fields.get("service")}) distinct_src_ports = len({event.fields.get("srcport") for event in src_events if event.fields.get("srcport")}) distinct_dst_ports = len({event.fields.get("dstport") for event in src_events if event.fields.get("dstport")}) total_bytes = sum(_as_int(event.fields.get("sentbyte")) + _as_int(event.fields.get("rcvdbyte")) for event in src_events) total_hitcount = sum(_as_int(event.fields.get("hitcount")) for event in src_events) max_hitcount = max((_as_int(event.fields.get("hitcount")) for event in src_events), default=0) event_rate, observed_duration = _rate_per_minute(src_events) 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") 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 event_z = (event_count - avg_events) / std_events if event_count >= 25 and event_z >= 2: points = min(25, 10 + int(event_z * 5)) score += points reasons.append(f"unusually high event volume for source ({event_count} events, z={event_z:.1f})") if event_rate is not None: rate_z = (event_rate - avg_rate) / std_rate if event_rate >= 20 and rate_z >= 2: points = min(25, 10 + int(rate_z * 5)) score += points reasons.append(f"unusually high log rate ({event_rate:.1f} events/min, z={rate_z:.1f})") dst_z = (distinct_dst - avg_dst) / std_dst if distinct_dst >= 10 and dst_z >= 2: points = min(25, 10 + int(dst_z * 5)) score += points reasons.append(f"source contacted unusually many destinations ({distinct_dst}, z={dst_z:.1f})") byte_z = (total_bytes - avg_bytes) / std_bytes if total_bytes >= 50_000_000 and byte_z >= 2: points = min(20, 8 + int(byte_z * 4)) score += points reasons.append(f"unusually high byte volume ({total_bytes} bytes, z={byte_z:.1f})") hitcount_z = (total_hitcount - avg_hitcount) / std_hitcount if total_hitcount >= 1_000 and hitcount_z >= 2: points = min(15, 5 + int(hitcount_z * 3)) score += points reasons.append(f"unusually high policy hitcount ({total_hitcount}, max event value {max_hitcount})") if event_count >= 5: deny_rate = deny_count / event_count if deny_count >= 10 and deny_rate >= 0.5: 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) score += points reasons.append(f"UTM/security detections observed ({utm_count} events)") if high_severity_count: score += min(20, high_severity_count * 8) reasons.append(f"high or critical severity events observed ({high_severity_count})") if distinct_services >= 8 and event_count >= 10: score += min(15, distinct_services) reasons.append(f"many distinct services used ({distinct_services})") if distinct_dst_ports >= 10 and event_count >= 10: score += min(15, distinct_dst_ports) reasons.append(f"many destination ports contacted ({distinct_dst_ports})") if _is_public_ip(src_ip) and (utm_count or deny_count >= 10): score += 10 reasons.append("public source with repeated security-relevant events") if not reasons: continue score = min(score, 100) findings.append( AnomalyFinding( subject=src_ip, score=score, severity=_severity(score), confidence=_confidence(event_count, len(reasons)), reasons=reasons, evidence={ "events": event_count, "distinct_destinations": distinct_dst, "distinct_services": distinct_services, "deny_or_threat_actions": deny_count, "utm_events": utm_count, "high_severity_events": high_severity_count, "total_bytes": total_bytes, "events_per_minute": round(event_rate, 2) if event_rate is not None else 0.0, "observed_duration_seconds": round(observed_duration, 2), "timed_events": sum(1 for event in src_events if _event_timestamp(event) is not None), "hitcount_total": total_hitcount, "hitcount_max": max_hitcount, "distinct_src_ports": distinct_src_ports, "distinct_dst_ports": distinct_dst_ports, "policy_count": len(policies), "implicit_deny_events": implicit_deny_count, }, ) ) return sorted(findings, key=lambda finding: finding.score, reverse=True)[:limit] def anomaly_summary(findings: list[AnomalyFinding]) -> dict[str, int]: counts = Counter(finding.severity for finding in findings) return { "total": len(findings), "critical": counts["critical"], "high": counts["high"], "medium": counts["medium"], "low": counts["low"], }