194 lines
7.7 KiB
Python
194 lines
7.7 KiB
Python
"""Detector helpers for BGP anomaly generation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections import Counter, defaultdict
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from app.models.bgp_anomaly import BGPAnomaly
|
|
|
|
|
|
def detect_origin_change_anomalies(
|
|
*,
|
|
source: str,
|
|
snapshot_id: int | None,
|
|
task_id: int | None,
|
|
events: list[dict[str, Any]],
|
|
previous_origin_map: dict[str, set[int]],
|
|
) -> list[BGPAnomaly]:
|
|
prefix_to_origins: defaultdict[str, set[int]] = defaultdict(set)
|
|
for event in events:
|
|
metadata = event.get("metadata") or {}
|
|
prefix = metadata.get("prefix")
|
|
origin_asn = metadata.get("origin_asn")
|
|
if prefix and origin_asn is not None:
|
|
prefix_to_origins[str(prefix)].add(int(origin_asn))
|
|
|
|
anomalies: list[BGPAnomaly] = []
|
|
for prefix, origins in prefix_to_origins.items():
|
|
historic = previous_origin_map.get(prefix, set())
|
|
new_origins = sorted(origin for origin in origins if origin not in historic)
|
|
if not historic or not new_origins:
|
|
continue
|
|
|
|
for new_origin in new_origins:
|
|
sample_event = next(
|
|
(
|
|
event
|
|
for event in events
|
|
if (event.get("metadata") or {}).get("prefix") == prefix
|
|
and int((event.get("metadata") or {}).get("origin_asn") or -1) == new_origin
|
|
),
|
|
{},
|
|
)
|
|
sample_metadata = sample_event.get("metadata") or {}
|
|
sample_enrichment = sample_metadata.get("enrichment") or {}
|
|
anomalies.append(
|
|
BGPAnomaly(
|
|
snapshot_id=snapshot_id,
|
|
task_id=task_id,
|
|
source=source,
|
|
anomaly_type="origin_change",
|
|
severity="critical",
|
|
status="active",
|
|
entity_key=f"origin_change:{prefix}:{new_origin}",
|
|
prefix=prefix,
|
|
origin_asn=sorted(historic)[0],
|
|
new_origin_asn=new_origin,
|
|
peer_scope=[],
|
|
started_at=datetime.now(UTC),
|
|
confidence=0.86,
|
|
summary=f"Prefix {prefix} is now originated by AS{new_origin}, outside the current baseline.",
|
|
evidence={
|
|
"previous_origins": sorted(historic),
|
|
"current_origins": sorted(origins),
|
|
"events": [sample_metadata] if sample_metadata else [],
|
|
"origin_asn_profile": sample_enrichment.get("origin_asn_profile"),
|
|
"new_origin_asn_profile": sample_enrichment.get("new_origin_asn_profile"),
|
|
"rpki_validation": sample_enrichment.get("rpki_validation"),
|
|
"prefix_scope": sample_enrichment.get("prefix_scope"),
|
|
"impacted_regions": sample_enrichment.get("prefix_scope", {}).get("regions", []),
|
|
},
|
|
)
|
|
)
|
|
|
|
return anomalies
|
|
|
|
|
|
def detect_more_specific_burst_anomalies(
|
|
*,
|
|
source: str,
|
|
snapshot_id: int | None,
|
|
task_id: int | None,
|
|
events: list[dict[str, Any]],
|
|
) -> list[BGPAnomaly]:
|
|
prefix_to_more_specifics: defaultdict[str, list[dict[str, Any]]] = defaultdict(list)
|
|
for event in events:
|
|
metadata = event.get("metadata") or {}
|
|
prefix = metadata.get("prefix")
|
|
enrichment = metadata.get("enrichment") or {}
|
|
if prefix and enrichment.get("is_more_specific"):
|
|
prefix_to_more_specifics[str(prefix).split("/")[0]].append(event)
|
|
|
|
anomalies: list[BGPAnomaly] = []
|
|
for root_prefix, more_specifics in prefix_to_more_specifics.items():
|
|
if len(more_specifics) < 2:
|
|
continue
|
|
|
|
sample = more_specifics[0].get("metadata") or {}
|
|
sample_enrichment = sample.get("enrichment") or {}
|
|
anomalies.append(
|
|
BGPAnomaly(
|
|
snapshot_id=snapshot_id,
|
|
task_id=task_id,
|
|
source=source,
|
|
anomaly_type="more_specific_burst",
|
|
severity="high",
|
|
status="active",
|
|
entity_key=f"more_specific_burst:{root_prefix}:{len(more_specifics)}",
|
|
prefix=sample.get("prefix"),
|
|
origin_asn=sample.get("origin_asn"),
|
|
new_origin_asn=None,
|
|
peer_scope=sorted(
|
|
{
|
|
str(item.get("metadata", {}).get("collector") or "")
|
|
for item in more_specifics
|
|
if item.get("metadata", {}).get("collector")
|
|
}
|
|
),
|
|
started_at=datetime.now(UTC),
|
|
confidence=0.72,
|
|
summary=f"{len(more_specifics)} more-specific announcements clustered around {root_prefix}.",
|
|
evidence={
|
|
"events": [item.get("metadata") for item in more_specifics[:10]],
|
|
"rpki_validation": sample_enrichment.get("rpki_validation"),
|
|
"origin_asn_profile": sample_enrichment.get("origin_asn_profile"),
|
|
"prefix_scope": sample_enrichment.get("prefix_scope"),
|
|
"impacted_regions": sample_enrichment.get("prefix_scope", {}).get("regions", []),
|
|
},
|
|
)
|
|
)
|
|
|
|
return anomalies
|
|
|
|
|
|
def detect_mass_withdrawal_anomalies(
|
|
*,
|
|
source: str,
|
|
snapshot_id: int | None,
|
|
task_id: int | None,
|
|
events: list[dict[str, Any]],
|
|
) -> list[BGPAnomaly]:
|
|
withdrawal_counter: Counter[tuple[str, int | None]] = Counter()
|
|
for event in events:
|
|
metadata = event.get("metadata") or {}
|
|
prefix = metadata.get("prefix")
|
|
if prefix and metadata.get("event_type") == "withdrawal":
|
|
withdrawal_counter[(str(prefix), metadata.get("origin_asn"))] += 1
|
|
|
|
anomalies: list[BGPAnomaly] = []
|
|
for (prefix, origin_asn), count in withdrawal_counter.items():
|
|
if count < 3:
|
|
continue
|
|
sample_event = next(
|
|
(
|
|
event
|
|
for event in events
|
|
if (event.get("metadata") or {}).get("prefix") == prefix
|
|
and (event.get("metadata") or {}).get("event_type") == "withdrawal"
|
|
),
|
|
{},
|
|
)
|
|
sample_metadata = sample_event.get("metadata") or {}
|
|
sample_enrichment = sample_metadata.get("enrichment") or {}
|
|
|
|
anomalies.append(
|
|
BGPAnomaly(
|
|
snapshot_id=snapshot_id,
|
|
task_id=task_id,
|
|
source=source,
|
|
anomaly_type="mass_withdrawal",
|
|
severity="high" if count < 8 else "critical",
|
|
status="active",
|
|
entity_key=f"mass_withdrawal:{prefix}:{origin_asn}:{count}",
|
|
prefix=prefix,
|
|
origin_asn=origin_asn,
|
|
new_origin_asn=None,
|
|
peer_scope=[],
|
|
started_at=datetime.now(UTC),
|
|
confidence=min(0.55 + (count * 0.05), 0.95),
|
|
summary=f"{count} withdrawal events observed for {prefix} in the current ingest window.",
|
|
evidence={
|
|
"withdrawal_count": count,
|
|
"events": [sample_metadata] if sample_metadata else [],
|
|
"origin_asn_profile": sample_enrichment.get("origin_asn_profile"),
|
|
"rpki_validation": sample_enrichment.get("rpki_validation"),
|
|
"prefix_scope": sample_enrichment.get("prefix_scope"),
|
|
"impacted_regions": sample_enrichment.get("prefix_scope", {}).get("regions", []),
|
|
},
|
|
)
|
|
)
|
|
|
|
return anomalies
|