351 lines
14 KiB
Python
351 lines
14 KiB
Python
"""Shared helpers for BGP collectors."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
from collections import defaultdict
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.bgp_anomaly import BGPAnomaly
|
|
from app.models.bgp_observation import BGPObservation
|
|
from app.models.collected_data import CollectedData
|
|
from app.services.bgp_incidents import create_bgp_incidents_for_anomalies
|
|
from app.services.bgp_detectors import (
|
|
detect_mass_withdrawal_anomalies,
|
|
detect_more_specific_burst_anomalies,
|
|
detect_origin_change_anomalies,
|
|
detect_path_flap_anomalies,
|
|
detect_route_leak_anomalies,
|
|
)
|
|
from app.services.bgp_enrichment import enrich_bgp_events_for_batch, extract_bgp_network_fields
|
|
|
|
|
|
RIPE_RIS_COLLECTOR_COORDS: dict[str, dict[str, Any]] = {
|
|
"rrc00": {"city": "Amsterdam", "country": "Netherlands", "latitude": 52.3676, "longitude": 4.9041},
|
|
"rrc01": {"city": "London", "country": "United Kingdom", "latitude": 51.5072, "longitude": -0.1276},
|
|
"rrc03": {"city": "Amsterdam", "country": "Netherlands", "latitude": 52.3676, "longitude": 4.9041},
|
|
"rrc04": {"city": "Geneva", "country": "Switzerland", "latitude": 46.2044, "longitude": 6.1432},
|
|
"rrc05": {"city": "Vienna", "country": "Austria", "latitude": 48.2082, "longitude": 16.3738},
|
|
"rrc06": {"city": "Otemachi", "country": "Japan", "latitude": 35.686, "longitude": 139.7671},
|
|
"rrc07": {"city": "Stockholm", "country": "Sweden", "latitude": 59.3293, "longitude": 18.0686},
|
|
"rrc10": {"city": "Milan", "country": "Italy", "latitude": 45.4642, "longitude": 9.19},
|
|
"rrc11": {"city": "New York", "country": "United States", "latitude": 40.7128, "longitude": -74.006},
|
|
"rrc12": {"city": "Frankfurt", "country": "Germany", "latitude": 50.1109, "longitude": 8.6821},
|
|
"rrc13": {"city": "Moscow", "country": "Russia", "latitude": 55.7558, "longitude": 37.6173},
|
|
"rrc14": {"city": "Palo Alto", "country": "United States", "latitude": 37.4419, "longitude": -122.143},
|
|
"rrc15": {"city": "Sao Paulo", "country": "Brazil", "latitude": -23.5558, "longitude": -46.6396},
|
|
"rrc16": {"city": "Miami", "country": "United States", "latitude": 25.7617, "longitude": -80.1918},
|
|
"rrc18": {"city": "Barcelona", "country": "Spain", "latitude": 41.3874, "longitude": 2.1686},
|
|
"rrc19": {"city": "Johannesburg", "country": "South Africa", "latitude": -26.2041, "longitude": 28.0473},
|
|
"rrc20": {"city": "Zurich", "country": "Switzerland", "latitude": 47.3769, "longitude": 8.5417},
|
|
"rrc21": {"city": "Paris", "country": "France", "latitude": 48.8566, "longitude": 2.3522},
|
|
"rrc22": {"city": "Bucharest", "country": "Romania", "latitude": 44.4268, "longitude": 26.1025},
|
|
"rrc23": {"city": "Singapore", "country": "Singapore", "latitude": 1.3521, "longitude": 103.8198},
|
|
"rrc24": {"city": "Montevideo", "country": "Uruguay", "latitude": -34.9011, "longitude": -56.1645},
|
|
"rrc25": {"city": "Amsterdam", "country": "Netherlands", "latitude": 52.3676, "longitude": 4.9041},
|
|
"rrc26": {"city": "Dubai", "country": "United Arab Emirates", "latitude": 25.2048, "longitude": 55.2708},
|
|
}
|
|
|
|
|
|
def _safe_int(value: Any) -> int | None:
|
|
try:
|
|
if value in (None, ""):
|
|
return None
|
|
return int(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _parse_timestamp(value: Any) -> datetime:
|
|
if isinstance(value, datetime):
|
|
return value.astimezone(UTC) if value.tzinfo else value.replace(tzinfo=UTC)
|
|
|
|
if isinstance(value, (int, float)):
|
|
return datetime.fromtimestamp(value, tz=UTC)
|
|
|
|
if isinstance(value, str) and value:
|
|
normalized = value.replace("Z", "+00:00")
|
|
parsed = datetime.fromisoformat(normalized)
|
|
return parsed.astimezone(UTC) if parsed.tzinfo else parsed.replace(tzinfo=UTC)
|
|
|
|
return datetime.now(UTC)
|
|
|
|
|
|
def _normalize_as_path(raw_path: Any) -> list[int]:
|
|
if raw_path in (None, ""):
|
|
return []
|
|
if isinstance(raw_path, list):
|
|
return [asn for asn in (_safe_int(item) for item in raw_path) if asn is not None]
|
|
if isinstance(raw_path, str):
|
|
parts = raw_path.replace("{", "").replace("}", "").split()
|
|
return [asn for asn in (_safe_int(item) for item in parts) if asn is not None]
|
|
return []
|
|
|
|
|
|
def normalize_bgp_event(payload: dict[str, Any], *, project: str) -> dict[str, Any]:
|
|
raw_message = payload.get("raw_message", payload)
|
|
raw_path = (
|
|
payload.get("path")
|
|
or payload.get("as_path")
|
|
or payload.get("attrs", {}).get("path")
|
|
or payload.get("attrs", {}).get("as_path")
|
|
or []
|
|
)
|
|
as_path = _normalize_as_path(raw_path)
|
|
|
|
raw_type = str(payload.get("event_type") or payload.get("type") or payload.get("msg_type") or "").lower()
|
|
if raw_type in {"a", "announce", "announcement"}:
|
|
event_type = "announcement"
|
|
elif raw_type in {"w", "withdraw", "withdrawal"}:
|
|
event_type = "withdrawal"
|
|
elif raw_type in {"r", "rib"}:
|
|
event_type = "rib"
|
|
else:
|
|
event_type = raw_type or "announcement"
|
|
|
|
prefix = str(payload.get("prefix") or payload.get("prefixes") or payload.get("target_prefix") or "").strip()
|
|
if prefix.startswith("[") and prefix.endswith("]"):
|
|
prefix = prefix[1:-1]
|
|
|
|
timestamp = _parse_timestamp(payload.get("timestamp") or payload.get("time") or payload.get("ts"))
|
|
collector = str(payload.get("collector") or payload.get("host") or payload.get("router") or "unknown")
|
|
peer_asn = _safe_int(payload.get("peer_asn") or payload.get("peer"))
|
|
peer_ip = payload.get("peer_ip") or payload.get("peer_address")
|
|
if peer_ip in (None, ""):
|
|
peer_candidate = payload.get("peer")
|
|
peer_ip = str(peer_candidate) if isinstance(peer_candidate, str) and ":" in peer_candidate else peer_candidate
|
|
origin_asn = _safe_int(payload.get("origin_asn")) or (as_path[-1] if as_path else None)
|
|
source_material = "|".join(
|
|
[
|
|
collector,
|
|
str(peer_asn or ""),
|
|
prefix,
|
|
event_type,
|
|
timestamp.isoformat(),
|
|
",".join(str(asn) for asn in as_path),
|
|
]
|
|
)
|
|
source_id = hashlib.sha1(source_material.encode("utf-8")).hexdigest()[:24]
|
|
|
|
collector_location = RIPE_RIS_COLLECTOR_COORDS.get(collector, {})
|
|
network_fields = extract_bgp_network_fields(prefix)
|
|
metadata = {
|
|
"project": project,
|
|
"collector": collector,
|
|
"peer_asn": peer_asn,
|
|
"peer_ip": peer_ip,
|
|
"event_type": event_type,
|
|
"prefix": prefix,
|
|
"origin_asn": origin_asn,
|
|
"as_path": as_path,
|
|
"communities": payload.get("communities")
|
|
or payload.get("community")
|
|
or payload.get("attrs", {}).get("communities")
|
|
or [],
|
|
"next_hop": payload.get("next_hop") or payload.get("attrs", {}).get("next_hop"),
|
|
"med": payload.get("med") or payload.get("attrs", {}).get("med"),
|
|
"local_pref": payload.get("local_pref") or payload.get("attrs", {}).get("local_pref"),
|
|
"timestamp": timestamp.isoformat(),
|
|
"as_path_length": len(as_path),
|
|
"visibility_weight": 1,
|
|
"collector_location": collector_location,
|
|
"raw_message": raw_message,
|
|
"prefix_family": network_fields.get("prefix_family"),
|
|
"prefix_length": network_fields.get("prefix_length"),
|
|
"prefix_supernet": network_fields.get("prefix_supernet"),
|
|
"is_more_specific": network_fields.get("is_more_specific", False),
|
|
}
|
|
|
|
return {
|
|
"source_id": source_id,
|
|
"name": prefix or f"{collector}:{event_type}",
|
|
"title": f"{event_type} {prefix}".strip(),
|
|
"description": f"{collector} observed {event_type} for {prefix}".strip(),
|
|
"reference_date": timestamp.isoformat(),
|
|
"country": collector_location.get("country"),
|
|
"city": collector_location.get("city"),
|
|
"latitude": collector_location.get("latitude"),
|
|
"longitude": collector_location.get("longitude"),
|
|
"metadata": metadata,
|
|
}
|
|
|
|
|
|
async def save_bgp_observations_for_batch(
|
|
db: AsyncSession,
|
|
*,
|
|
source: str,
|
|
snapshot_id: int | None,
|
|
task_id: int | None,
|
|
events: list[dict[str, Any]],
|
|
) -> int:
|
|
if not events:
|
|
return 0
|
|
|
|
ingest_batch_id = f"{source}:{task_id or 'adhoc'}:{snapshot_id or 'nosnapshot'}"
|
|
created = 0
|
|
|
|
for event in events:
|
|
metadata = event.get("metadata", {}) or {}
|
|
collector_location = metadata.get("collector_location") or {}
|
|
observed_at = _parse_timestamp(
|
|
metadata.get("timestamp") or event.get("reference_date")
|
|
)
|
|
|
|
db.add(
|
|
BGPObservation(
|
|
snapshot_id=snapshot_id,
|
|
task_id=task_id,
|
|
source=source,
|
|
ingest_batch_id=ingest_batch_id,
|
|
source_event_id=event.get("source_id"),
|
|
collector=metadata.get("collector"),
|
|
peer_asn=_safe_int(metadata.get("peer_asn")),
|
|
peer_ip=metadata.get("peer_ip"),
|
|
prefix=metadata.get("prefix"),
|
|
event_type=str(metadata.get("event_type") or "announcement"),
|
|
as_path=metadata.get("as_path") or [],
|
|
origin_asn=_safe_int(metadata.get("origin_asn")),
|
|
next_hop=metadata.get("next_hop"),
|
|
communities=metadata.get("communities") or [],
|
|
observed_at=observed_at,
|
|
collector_geo=collector_location,
|
|
raw_payload=metadata.get("raw_message") or {},
|
|
note=event.get("description"),
|
|
)
|
|
)
|
|
created += 1
|
|
|
|
if created:
|
|
await db.commit()
|
|
|
|
return created
|
|
|
|
|
|
async def create_bgp_anomalies_for_batch(
|
|
db: AsyncSession,
|
|
*,
|
|
source: str,
|
|
snapshot_id: int | None,
|
|
task_id: int | None,
|
|
events: list[dict[str, Any]],
|
|
) -> int:
|
|
if not events:
|
|
return 0
|
|
|
|
enriched_events = await enrich_bgp_events_for_batch(
|
|
db,
|
|
source=source,
|
|
events=events,
|
|
)
|
|
|
|
prefixes = {
|
|
event["metadata"].get("prefix")
|
|
for event in enriched_events
|
|
if event.get("metadata", {}).get("prefix")
|
|
}
|
|
previous_origin_map: dict[str, set[int]] = defaultdict(set)
|
|
|
|
if prefixes:
|
|
previous_query = await db.execute(
|
|
select(CollectedData).where(
|
|
CollectedData.source == source,
|
|
CollectedData.snapshot_id != snapshot_id,
|
|
CollectedData.extra_data["prefix"].as_string().in_(sorted(prefixes)),
|
|
)
|
|
)
|
|
for record in previous_query.scalars().all():
|
|
metadata = record.extra_data or {}
|
|
prefix = metadata.get("prefix")
|
|
origin = _safe_int(metadata.get("origin_asn"))
|
|
if prefix and origin is not None:
|
|
previous_origin_map[prefix].add(origin)
|
|
|
|
pending_anomalies = [
|
|
*detect_origin_change_anomalies(
|
|
source=source,
|
|
snapshot_id=snapshot_id,
|
|
task_id=task_id,
|
|
events=enriched_events,
|
|
previous_origin_map=previous_origin_map,
|
|
),
|
|
*detect_more_specific_burst_anomalies(
|
|
source=source,
|
|
snapshot_id=snapshot_id,
|
|
task_id=task_id,
|
|
events=enriched_events,
|
|
),
|
|
*detect_mass_withdrawal_anomalies(
|
|
source=source,
|
|
snapshot_id=snapshot_id,
|
|
task_id=task_id,
|
|
events=enriched_events,
|
|
),
|
|
*detect_route_leak_anomalies(
|
|
source=source,
|
|
snapshot_id=snapshot_id,
|
|
task_id=task_id,
|
|
events=enriched_events,
|
|
),
|
|
*detect_path_flap_anomalies(
|
|
source=source,
|
|
snapshot_id=snapshot_id,
|
|
task_id=task_id,
|
|
events=enriched_events,
|
|
),
|
|
]
|
|
|
|
if not pending_anomalies:
|
|
return 0
|
|
|
|
existing_result = await db.execute(
|
|
select(BGPAnomaly.entity_key).where(
|
|
BGPAnomaly.entity_key.in_([item.entity_key for item in pending_anomalies])
|
|
)
|
|
)
|
|
existing_keys = {row[0] for row in existing_result.fetchall()}
|
|
existing_anomalies: list[BGPAnomaly] = []
|
|
if existing_keys:
|
|
existing_anomaly_result = await db.execute(
|
|
select(BGPAnomaly).where(BGPAnomaly.entity_key.in_(sorted(existing_keys)))
|
|
)
|
|
existing_anomalies = existing_anomaly_result.scalars().all()
|
|
|
|
created = 0
|
|
created_anomalies: list[BGPAnomaly] = []
|
|
refreshed_anomalies: list[BGPAnomaly] = []
|
|
existing_map = {item.entity_key: item for item in existing_anomalies if item.entity_key}
|
|
for anomaly in pending_anomalies:
|
|
if anomaly.entity_key in existing_keys:
|
|
existing = existing_map.get(anomaly.entity_key)
|
|
if existing is not None:
|
|
existing.severity = anomaly.severity
|
|
existing.status = anomaly.status
|
|
existing.summary = anomaly.summary
|
|
existing.confidence = anomaly.confidence
|
|
existing.peer_scope = anomaly.peer_scope
|
|
existing.evidence = anomaly.evidence
|
|
existing.new_origin_asn = anomaly.new_origin_asn
|
|
existing.origin_asn = anomaly.origin_asn
|
|
refreshed_anomalies.append(existing)
|
|
continue
|
|
db.add(anomaly)
|
|
created_anomalies.append(anomaly)
|
|
created += 1
|
|
|
|
if created or refreshed_anomalies:
|
|
await db.commit()
|
|
incident_seed_anomalies = [*created_anomalies, *refreshed_anomalies]
|
|
if incident_seed_anomalies:
|
|
await create_bgp_incidents_for_anomalies(
|
|
db,
|
|
source=source,
|
|
snapshot_id=snapshot_id,
|
|
task_id=task_id,
|
|
anomalies=incident_seed_anomalies,
|
|
)
|
|
return created
|