From c439e91d1292af722a88308635b32886155744d5 Mon Sep 17 00:00:00 2001 From: linkong Date: Tue, 7 Apr 2026 15:23:26 +0800 Subject: [PATCH] feat(bgp): improve prefix-geo pipeline and collector reliability --- TODO.md | 5 +- backend/app/api/v1/datasources.py | 28 +- backend/app/core/data_sources.py | 2 + backend/app/core/data_sources.yaml | 6 + backend/app/core/datasource_defaults.py | 14 + backend/app/services/bgp_enrichment.py | 123 ++++++--- backend/app/services/collectors/__init__.py | 4 + backend/app/services/collectors/iptoasn.py | 290 ++++++++++++++++++-- backend/app/services/scheduler.py | 49 ++++ backend/tests/test_bgp.py | 157 ++++++++++- docs/bgp-context.md | 5 + 11 files changed, 621 insertions(+), 62 deletions(-) diff --git a/TODO.md b/TODO.md index 2ddaf1e6..b2aa5385 100644 --- a/TODO.md +++ b/TODO.md @@ -11,8 +11,9 @@ - [x] 把 incident 地理定位从 `collector-centric` 改成 `prefix-centric`，优先使用 `prefix_geography`，其次 `prefix_scope`，再次 ASN 区域，最后才回退到观测区域质心 - [x] 新增 `prefix_geography` 数据层，不再把 `prefix_scope` 当成 prefix 地理归属本身 - [x] 接入 `IPtoASN / IPtoCountry` 作为 prefix-centric geography 的主数据源 -- [ ] 接入 `OpenGeoFeed` 作为 prefix geography 的高质量覆盖/override 数据源 -- [ ] 把 RIR delegated / `inetnum` / `inet6num` whois 设计成 prefix geography 的 fallback，而不是主来源 +- [x] 接入 `OpenGeoFeed` 作为 prefix geography 的高质量覆盖/override 数据源 +- [x] 把 RIR delegated 设计成 prefix geography 的 fallback，而不是主来源 +- [ ] 接入 `inetnum` / `inet6num` whois 作为比 RIR 更细粒度的后备层 - [x] 在 activity layer 之后继续补 `route leak` 和 `path instability / flap` detector - [ ] 对 [frontend/public/earth/js/bgp.js](/home/ray/dev/linkong/planet/frontend/public/earth/js/bgp.js) 做按职责拆分的小重构，拆成 data / markers / overlays / animation，降低后续维护复杂度 - [ ] 可选优化（非必做）：将 BGP incident/collector 标点改为 HTML marker（参考 worldmonitor 的 `htmlElementsData` 思路），实现近乎固定屏幕尺寸与更高密度可点击性 diff --git a/backend/app/api/v1/datasources.py b/backend/app/api/v1/datasources.py index 92b01193..22337662 100644 --- a/backend/app/api/v1/datasources.py +++ b/backend/app/api/v1/datasources.py @@ -17,6 +17,7 @@ from app.models.user import User from app.services.scheduler import get_latest_task_id_for_datasource, run_collector_now, sync_datasource_job router = APIRouter() +STALE_RUNNING_TASK_TIMEOUT_MINUTES = 90 def format_frequency_label(minutes: int) -> str: @@ -71,7 +72,32 @@ async def get_running_task(db: AsyncSession, datasource_id: int) -> Optional[Col .order_by(CollectionTask.started_at.desc()) .limit(1) ) - return result.scalar_one_or_none() + task = result.scalar_one_or_none() + if not task: + return None + + started_at = task.started_at + if started_at is None: + return task + + now = datetime.now(timezone.utc) + if started_at.tzinfo is None: + started_at = started_at.replace(tzinfo=timezone.utc) + + if now - started_at <= timedelta(minutes=STALE_RUNNING_TASK_TIMEOUT_MINUTES): + return task + + existing_error = (task.error_message or "").strip() + stale_reason = ( + f"Marked failed automatically after stale running timeout " + f"({STALE_RUNNING_TASK_TIMEOUT_MINUTES}m)" + ) + task.status = "failed" + task.phase = "failed" + task.completed_at = now + task.error_message = f"{existing_error}\n{stale_reason}".strip() if existing_error else stale_reason + await db.commit() + return None @router.get("") diff --git a/backend/app/core/data_sources.py b/backend/app/core/data_sources.py index c677129c..0ec35942 100644 --- a/backend/app/core/data_sources.py +++ b/backend/app/core/data_sources.py @@ -26,6 +26,8 @@ COLLECTOR_URL_KEYS = { "ris_live_bgp": "ris_live.url", "bgpstream_bgp": "bgpstream.url", "iptoasn_prefix_geo": "iptoasn.combined_url", + "opengeofeed_prefix_geo": "opengeofeed.public_csv_url", + "nro_delegated_prefix_geo": "nro.delegated_stats_url", } diff --git a/backend/app/core/data_sources.yaml b/backend/app/core/data_sources.yaml index faacac7c..e1261be8 100644 --- a/backend/app/core/data_sources.yaml +++ b/backend/app/core/data_sources.yaml @@ -46,3 +46,9 @@ bgpstream: iptoasn: combined_url: "https://iptoasn.com/data/ip2asn-combined.tsv.gz" + +opengeofeed: + public_csv_url: "https://opengeofeed.org/feed/public.csv" + +nro: + delegated_stats_url: "https://ftp.ripe.net/pub/stats/ripencc/nro-stats/latest/nro-delegated-stats" diff --git a/backend/app/core/datasource_defaults.py b/backend/app/core/datasource_defaults.py index 9d44121c..189030ee 100644 --- a/backend/app/core/datasource_defaults.py +++ b/backend/app/core/datasource_defaults.py @@ -141,6 +141,20 @@ DEFAULT_DATASOURCES = { "priority": "P1", "frequency_minutes": 1440, }, + "opengeofeed_prefix_geo": { + "id": 24, + "name": "OpenGeoFeed Prefix Geography", + "module": "L3", + "priority": "P1", + "frequency_minutes": 1440, + }, + "nro_delegated_prefix_geo": { + "id": 25, + "name": "NRO Delegated Prefix Geography", + "module": "L3", + "priority": "P1", + "frequency_minutes": 1440, + }, } ID_TO_COLLECTOR = {info["id"]: name for name, info in DEFAULT_DATASOURCES.items()} diff --git a/backend/app/services/bgp_enrichment.py b/backend/app/services/bgp_enrichment.py index 32cc0fca..0d37ee67 100644 --- a/backend/app/services/bgp_enrichment.py +++ b/backend/app/services/bgp_enrichment.py @@ -101,6 +101,49 @@ async def _lookup_prefix_geography( db: AsyncSession, prefix_values: list[str], ) -> dict[str, dict[str, Any]]: + async def _query_prefix_metadata( + *, + source: str, + family: str, + range_start: str, + range_end: str, + ) -> dict[str, Any] | None: + result = await db.execute( + text( + """ + SELECT metadata + FROM collected_data + WHERE source = :source + AND COALESCE(is_current, TRUE) = TRUE + AND metadata->>'family' = :family + AND CAST(metadata->>'range_start' AS inet) <= CAST(:range_start AS inet) + AND CAST(metadata->>'range_end' AS inet) >= CAST(:range_end AS inet) + ORDER BY + masklen(CAST(metadata->>'prefix' AS cidr)) DESC NULLS LAST, + id DESC + LIMIT 1 + """ + ), + { + "source": source, + "family": family, + "range_start": range_start, + "range_end": range_end, + }, + ) + row = result.fetchone() + if not row: + return None + + if isinstance(row, dict): + payload = row.get("metadata") or row.get("extra_data") + elif hasattr(row, "_mapping"): + payload = row._mapping.get("metadata") or row._mapping.get("extra_data") + else: + payload = row[0] + + return payload if isinstance(payload, dict) else None + results: dict[str, dict[str, Any]] = {} for prefix in prefix_values: @@ -112,50 +155,44 @@ async def _lookup_prefix_geography( family = f"ipv{network.version}" range_start = str(network.network_address) range_end = str(network.broadcast_address) - result = await db.execute( - text( - """ - SELECT metadata - FROM collected_data - WHERE source = 'iptoasn_prefix_geo' - AND COALESCE(is_current, TRUE) = TRUE - AND metadata->>'family' = :family - AND CAST(metadata->>'range_start' AS inet) <= CAST(:range_start AS inet) - AND CAST(metadata->>'range_end' AS inet) >= CAST(:range_end AS inet) - ORDER BY id DESC - LIMIT 1 - """ - ), - { - "family": family, - "range_start": range_start, - "range_end": range_end, - }, + payload = await _query_prefix_metadata( + source="opengeofeed_prefix_geo", + family=family, + range_start=range_start, + range_end=range_end, ) - row = result.fetchone() - if not row: - continue - - if isinstance(row, dict): - payload = row.get("metadata") or row.get("extra_data") - elif hasattr(row, "_mapping"): - payload = row._mapping.get("metadata") or row._mapping.get("extra_data") - else: - payload = row[0] - if not isinstance(payload, dict): + selected_source = "opengeofeed" + if not payload: + payload = await _query_prefix_metadata( + source="iptoasn_prefix_geo", + family=family, + range_start=range_start, + range_end=range_end, + ) + selected_source = "iptoasn" + if not payload: + payload = await _query_prefix_metadata( + source="nro_delegated_prefix_geo", + family=family, + range_start=range_start, + range_end=range_end, + ) + selected_source = "nro_delegated" + if not payload: continue country = normalize_country(payload.get("country") or payload.get("country_code")) prefix_hint = payload.get("prefix") or prefix asn = _safe_int(payload.get("asn")) as_name = payload.get("as_name") + city = payload.get("city") centroid = get_country_centroid(country) regions = [] if country: regions.append( { "country": country, - "city": None, + "city": city, "latitude": centroid.get("latitude") if centroid else None, "longitude": centroid.get("longitude") if centroid else None, } @@ -164,10 +201,30 @@ async def _lookup_prefix_geography( results[prefix] = { "prefix": prefix_hint, "country": country, + "city": city, "asn": asn, "as_name": as_name, - "source": payload.get("source_dataset") or "iptoasn_combined", - "confidence": "country_range", + "source": payload.get("source_dataset") + or ( + "opengeofeed_public" + if selected_source == "opengeofeed" + else ( + "iptoasn_combined" + if selected_source == "iptoasn" + else "nro_delegated_stats" + ) + ), + "confidence": payload.get("confidence") + or ( + "geofeed" + if selected_source == "opengeofeed" + else ( + "country_range" + if selected_source == "iptoasn" + else "registry_allocated" + ) + ), + "geography_mode": "prefix_geography", "regions": regions, } diff --git a/backend/app/services/collectors/__init__.py b/backend/app/services/collectors/__init__.py index 3c83c86a..fdc4d4a7 100644 --- a/backend/app/services/collectors/__init__.py +++ b/backend/app/services/collectors/__init__.py @@ -33,6 +33,8 @@ from app.services.collectors.celestrak import CelesTrakTLECollector from app.services.collectors.ris_live import RISLiveCollector from app.services.collectors.bgpstream import BGPStreamBackfillCollector from app.services.collectors.iptoasn import IPtoASNPrefixGeoCollector +from app.services.collectors.opengeofeed import OpenGeoFeedPrefixGeoCollector +from app.services.collectors.nro_delegated import NRODelegatedPrefixGeoCollector collector_registry.register(TOP500Collector()) collector_registry.register(EpochAIGPUCollector()) @@ -57,3 +59,5 @@ collector_registry.register(CelesTrakTLECollector()) collector_registry.register(RISLiveCollector()) collector_registry.register(BGPStreamBackfillCollector()) collector_registry.register(IPtoASNPrefixGeoCollector()) +collector_registry.register(OpenGeoFeedPrefixGeoCollector()) +collector_registry.register(NRODelegatedPrefixGeoCollector()) diff --git a/backend/app/services/collectors/iptoasn.py b/backend/app/services/collectors/iptoasn.py index ee0de6f9..75998c35 100644 --- a/backend/app/services/collectors/iptoasn.py +++ b/backend/app/services/collectors/iptoasn.py @@ -6,9 +6,15 @@ prefix-to-country/ASN geography hints for BGP enrichment. from __future__ import annotations +import asyncio import gzip +import hashlib +import json +import tempfile +import time from datetime import UTC, datetime from ipaddress import summarize_address_range, ip_address +from pathlib import Path from typing import Any import httpx @@ -23,40 +29,274 @@ class IPtoASNPrefixGeoCollector(BaseCollector): frequency_hours = 24 data_type = "prefix_geography" fail_on_empty = True + _cache_dir = Path(tempfile.gettempdir()) / "planet-download-cache" / "iptoasn" + + @staticmethod + def _build_dataset_urls(resolved_url: str) -> list[str]: + if "ip2asn-combined.tsv.gz" in resolved_url: + return [ + resolved_url.replace("ip2asn-combined.tsv.gz", "ip2asn-v4.tsv.gz"), + resolved_url.replace("ip2asn-combined.tsv.gz", "ip2asn-v6.tsv.gz"), + ] + return [resolved_url] + + @staticmethod + def _cache_key(url: str) -> str: + return hashlib.sha1(url.encode("utf-8")).hexdigest()[:16] + + @classmethod + def _cache_paths(cls, url: str) -> tuple[Path, Path, Path]: + key = cls._cache_key(url) + gz_path = cls._cache_dir / f"{key}.tsv.gz" + part_path = cls._cache_dir / f"{key}.tsv.gz.part" + meta_path = cls._cache_dir / f"{key}.meta.json" + return gz_path, part_path, meta_path + + @staticmethod + def _load_meta(meta_path: Path) -> dict[str, Any]: + if not meta_path.exists(): + return {} + try: + return json.loads(meta_path.read_text(encoding="utf-8")) + except (json.JSONDecodeError, OSError): + return {} + + @staticmethod + def _save_meta(meta_path: Path, payload: dict[str, Any]) -> None: + meta_path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8") + + @staticmethod + def _validators_match(meta: dict[str, Any], remote: dict[str, Any]) -> bool: + etag = str(remote.get("etag") or "").strip() + last_modified = str(remote.get("last_modified") or "").strip() + if etag: + return etag == str(meta.get("etag") or "").strip() + if last_modified: + return last_modified == str(meta.get("last_modified") or "").strip() + return True + + async def _fetch_remote_info(self, client: httpx.AsyncClient, url: str) -> dict[str, Any]: + try: + response = await client.head(url) + if response.status_code >= 400: + return {} + content_length_raw = response.headers.get("content-length") + content_length = int(content_length_raw) if content_length_raw else None + return { + "etag": response.headers.get("etag"), + "last_modified": response.headers.get("last-modified"), + "content_length": content_length, + "accept_ranges": (response.headers.get("accept-ranges") or "").lower(), + } + except (httpx.HTTPError, ValueError): + return {} + + def _parse_rows_from_gzip_file(self, file_path: Path) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + with gzip.open(file_path, "rt", encoding="utf-8", errors="replace") as f: + for raw_line in f: + line = raw_line.strip() + if not line or line.startswith("#"): + continue + parts = line.split("\t") + if len(parts) < 5: + continue + range_start, range_end, asn, country_code, as_name = parts[:5] + rows.append( + { + "range_start": range_start, + "range_end": range_end, + "asn": asn, + "country_code": country_code, + "as_name": as_name, + } + ) + if not rows: + raise RuntimeError(f"IPtoASN dataset parsed empty rows: {file_path.name}") + return rows + + async def _download_dataset_with_resume( + self, + client: httpx.AsyncClient, + url: str, + *, + progress_callback=None, + ) -> Path: + self._cache_dir.mkdir(parents=True, exist_ok=True) + gz_path, part_path, meta_path = self._cache_paths(url) + meta = self._load_meta(meta_path) + remote = await self._fetch_remote_info(client, url) + + # Reuse existing complete file when validators and size still match. + if gz_path.exists(): + local_size = gz_path.stat().st_size + remote_size = remote.get("content_length") + if ( + self._validators_match(meta, remote) + and (remote_size is None or local_size == remote_size) + ): + try: + self._parse_rows_from_gzip_file(gz_path) + if progress_callback and local_size > 0: + await progress_callback(local_size) + return gz_path + except Exception: + gz_path.unlink(missing_ok=True) + + expected_size = remote.get("content_length") + can_resume = (remote.get("accept_ranges") or "") == "bytes" + resume_from = part_path.stat().st_size if part_path.exists() else 0 + if expected_size is not None and resume_from > expected_size: + part_path.unlink(missing_ok=True) + resume_from = 0 + if not self._validators_match(meta, remote): + part_path.unlink(missing_ok=True) + resume_from = 0 + + headers = { + "User-Agent": "Planet-Intelligence-System/1.0 (Python/collector)", + "Accept": "application/gzip,application/octet-stream,*/*", + } + if can_resume and resume_from > 0: + headers["Range"] = f"bytes={resume_from}-" + if remote.get("etag"): + headers["If-Range"] = str(remote.get("etag")) + elif remote.get("last_modified"): + headers["If-Range"] = str(remote.get("last_modified")) + + async with client.stream("GET", url, headers=headers) as response: + response.raise_for_status() + + if response.status_code == 206 and resume_from > 0: + mode = "ab" + else: + mode = "wb" + resume_from = 0 + + with part_path.open(mode) as f: + downloaded = resume_from + if progress_callback and downloaded > 0: + await progress_callback(downloaded) + async for chunk in response.aiter_bytes(): + if chunk: + f.write(chunk) + downloaded += len(chunk) + if progress_callback: + await progress_callback(downloaded) + + part_size = part_path.stat().st_size if part_path.exists() else 0 + if expected_size is not None and part_size != expected_size: + raise RuntimeError( + f"IPtoASN download incomplete for {url}: expected={expected_size}, got={part_size}" + ) + + part_path.replace(gz_path) + self._save_meta( + meta_path, + { + "url": url, + "etag": remote.get("etag"), + "last_modified": remote.get("last_modified"), + "content_length": expected_size, + "updated_at": datetime.now(UTC).isoformat(), + }, + ) + return gz_path + + async def _fetch_dataset_rows( + self, + client: httpx.AsyncClient, + url: str, + *, + progress_callback=None, + ) -> list[dict[str, Any]]: + try: + gz_path = await self._download_dataset_with_resume( + client, + url, + progress_callback=progress_callback, + ) + return self._parse_rows_from_gzip_file(gz_path) + except Exception: + # Bad partial/complete cache or interrupted stream: clear local files + # and retry once from scratch. + gz_path, part_path, meta_path = self._cache_paths(url) + gz_path.unlink(missing_ok=True) + part_path.unlink(missing_ok=True) + meta_path.unlink(missing_ok=True) + + gz_path = await self._download_dataset_with_resume( + client, + url, + progress_callback=progress_callback, + ) + return self._parse_rows_from_gzip_file(gz_path) async def fetch(self) -> list[dict[str, Any]]: if not self._resolved_url: raise RuntimeError("IPtoASN combined URL is not configured") + dataset_urls = self._build_dataset_urls(self._resolved_url) + async with httpx.AsyncClient(timeout=180.0, follow_redirects=True) as client: - response = await client.get( - self._resolved_url, - headers={ - "User-Agent": "Planet-Intelligence-System/1.0 (Python/collector)", - "Accept": "application/gzip,application/octet-stream,*/*", - }, + remote_infos = await asyncio.gather( + *(self._fetch_remote_info(client, url) for url in dataset_urls) ) - response.raise_for_status() - body = gzip.decompress(response.content).decode("utf-8", errors="replace") + expected_sizes = [ + info.get("content_length") + for info in remote_infos + if isinstance(info.get("content_length"), int) + ] + total_expected = sum(expected_sizes) if expected_sizes else 0 + if total_expected > 0 and self._current_task and self._db_session: + self._current_task.total_records = total_expected + self._current_task.records_processed = 0 + self._current_task.progress = 0.0 + await self._db_session.commit() + await self._publish_task_update(force=True) + + url_progress: dict[str, int] = {url: 0 for url in dataset_urls} + progress_lock = asyncio.Lock() + last_emit = {"t": 0.0, "value": 0} + min_emit_bytes = max(total_expected // 200, 2 * 1024 * 1024) if total_expected > 0 else 4 * 1024 * 1024 + + async def on_url_progress(url: str, downloaded_bytes: int) -> None: + if total_expected <= 0: + return + async with progress_lock: + current = max(0, downloaded_bytes) + if current < url_progress[url]: + return + url_progress[url] = current + aggregated = sum(url_progress.values()) + now = time.monotonic() + should_emit = ( + aggregated >= total_expected + or aggregated - last_emit["value"] >= min_emit_bytes + or now - last_emit["t"] >= 2.0 + ) + if not should_emit: + return + last_emit["value"] = aggregated + last_emit["t"] = now + await self.update_progress(min(aggregated, total_expected), commit=True) + + batches = await asyncio.gather( + *( + self._fetch_dataset_rows( + client, + url, + progress_callback=lambda downloaded, u=url: on_url_progress(u, downloaded), + ) + for url in dataset_urls + ) + ) + if total_expected > 0: + await self.update_progress(total_expected, commit=True, force=True) rows: list[dict[str, Any]] = [] - for raw_line in body.splitlines(): - line = raw_line.strip() - if not line or line.startswith("#"): - continue - parts = line.split("\t") - if len(parts) < 5: - continue - range_start, range_end, asn, country_code, as_name = parts[:5] - rows.append( - { - "range_start": range_start, - "range_end": range_end, - "asn": asn, - "country_code": country_code, - "as_name": as_name, - } - ) + for batch in batches: + rows.extend(batch) return rows def transform(self, raw_data: list[dict[str, Any]]) -> list[dict[str, Any]]: diff --git a/backend/app/services/scheduler.py b/backend/app/services/scheduler.py index 3e7f8623..ed0e45dd 100644 --- a/backend/app/services/scheduler.py +++ b/backend/app/services/scheduler.py @@ -18,6 +18,7 @@ from app.services.collectors.registry import collector_registry logger = logging.getLogger(__name__) scheduler = AsyncIOScheduler() +RUNNING_TASK_GUARD_TIMEOUT_MINUTES = 90 async def _update_next_run_at(datasource: DataSource, session) -> None: @@ -76,6 +77,54 @@ async def run_collector_task(collector_name: str): logger.info("Skipping disabled collector: %s", collector_name) return + running_result = await db.execute( + select(CollectionTask) + .where( + CollectionTask.datasource_id == datasource.id, + CollectionTask.status == "running", + ) + .order_by(CollectionTask.started_at.desc(), CollectionTask.id.desc()) + .limit(1) + ) + existing_running = running_result.scalar_one_or_none() + if existing_running is not None: + now = datetime.now(UTC) + started_at = existing_running.started_at + if started_at is not None and started_at.tzinfo is None: + started_at = started_at.replace(tzinfo=UTC) + + is_stale = ( + started_at is not None + and (now - started_at) > timedelta(minutes=RUNNING_TASK_GUARD_TIMEOUT_MINUTES) + ) + if not is_stale: + logger.warning( + "Skipping collector %s trigger because task %s is already running", + collector_name, + existing_running.id, + ) + return + + existing_error = (existing_running.error_message or "").strip() + stale_reason = ( + f"Marked failed automatically after stale running timeout " + f"({RUNNING_TASK_GUARD_TIMEOUT_MINUTES}m) in scheduler guard" + ) + existing_running.status = "failed" + existing_running.phase = "failed" + existing_running.completed_at = now + existing_running.error_message = ( + f"{existing_error}\n{stale_reason}".strip() + if existing_error + else stale_reason + ) + await db.commit() + logger.warning( + "Marked stale running task %s as failed before rerun of %s", + existing_running.id, + collector_name, + ) + try: collector._datasource_id = datasource.id logger.info("Running collector: %s (datasource_id=%s)", collector_name, datasource.id) diff --git a/backend/tests/test_bgp.py b/backend/tests/test_bgp.py index d22cab90..6482adf0 100644 --- a/backend/tests/test_bgp.py +++ b/backend/tests/test_bgp.py @@ -35,6 +35,8 @@ from app.models.user import User from app.services.collectors.bgp_common import normalize_bgp_event from app.services.collectors.bgpstream import BGPStreamBackfillCollector from app.services.collectors.iptoasn import IPtoASNPrefixGeoCollector +from app.services.collectors.opengeofeed import OpenGeoFeedPrefixGeoCollector +from app.services.collectors.nro_delegated import NRODelegatedPrefixGeoCollector class _FakeScalarResult: @@ -170,6 +172,75 @@ def test_iptoasn_transform_creates_prefix_geography_records(): assert record["metadata"]["source_dataset"] == "iptoasn_combined" +def test_iptoasn_build_dataset_urls_from_combined(): + urls = IPtoASNPrefixGeoCollector._build_dataset_urls( + "https://iptoasn.com/data/ip2asn-combined.tsv.gz" + ) + assert urls == [ + "https://iptoasn.com/data/ip2asn-v4.tsv.gz", + "https://iptoasn.com/data/ip2asn-v6.tsv.gz", + ] + + +def test_iptoasn_build_dataset_urls_passthrough_non_combined(): + url = "https://example.com/custom.tsv.gz" + urls = IPtoASNPrefixGeoCollector._build_dataset_urls(url) + assert urls == [url] + + +def test_opengeofeed_transform_creates_prefix_geography_records(): + collector = OpenGeoFeedPrefixGeoCollector() + transformed = collector.transform( + [ + { + "prefix": "203.0.113.0/24", + "country_code": "GB", + "region": "GB-LND", + "city": "London", + "postal_code": "EC1A", + "extra_columns": ["source:example"], + } + ] + ) + + assert len(transformed) == 1 + record = transformed[0] + assert record["name"] == "203.0.113.0/24" + assert record["metadata"]["family"] == "ipv4" + assert record["metadata"]["range_start"] == "203.0.113.0" + assert record["metadata"]["range_end"] == "203.0.113.255" + assert record["metadata"]["country_code"] == "GB" + assert record["metadata"]["source_dataset"] == "opengeofeed_public" + assert record["metadata"]["confidence"] == "geofeed" + + +def test_nro_delegated_transform_creates_prefix_geography_records(): + collector = NRODelegatedPrefixGeoCollector() + transformed = collector.transform( + [ + { + "rir": "ripencc", + "country_code": "DE", + "type": "ipv4", + "start": "198.51.100.0", + "value": "256", + "allocated_date": "20250401", + "status": "allocated", + } + ] + ) + + assert len(transformed) == 1 + record = transformed[0] + assert record["name"] == "198.51.100.0/24" + assert record["metadata"]["family"] == "ipv4" + assert record["metadata"]["range_start"] == "198.51.100.0" + assert record["metadata"]["range_end"] == "198.51.100.255" + assert record["metadata"]["country_code"] == "DE" + assert record["metadata"]["source_dataset"] == "nro_delegated_stats" + assert record["metadata"]["confidence"] == "registry_allocated" + + def test_bgp_anomaly_to_dict(): anomaly = BGPAnomaly( source="ris_live_bgp", @@ -605,7 +676,7 @@ async def test_enrich_bgp_events_for_batch_adds_profiles_and_prefix_scope(): } } - db = _FakeAsyncSession([[historical_observation], [iptoasn_row], [peeringdb_record]]) + db = _FakeAsyncSession([[historical_observation], [], [iptoasn_row], [peeringdb_record]]) events = [ { "metadata": { @@ -639,6 +710,90 @@ async def test_enrich_bgp_events_for_batch_adds_profiles_and_prefix_scope(): assert enrichment["prefix_scope"]["cities"] == ["London"] +@pytest.mark.asyncio +async def test_enrich_bgp_events_for_batch_prefers_opengeofeed_over_iptoasn(): + opengeofeed_row = { + "extra_data": { + "family": "ipv4", + "range_start": "203.0.113.0", + "range_end": "203.0.113.255", + "prefix": "203.0.113.0/24", + "country_code": "GB", + "region": "GB-LND", + "city": "London", + "source_dataset": "opengeofeed_public", + "confidence": "geofeed", + } + } + db = _FakeAsyncSession([[], [opengeofeed_row], []]) + events = [ + { + "metadata": { + "prefix": "203.0.113.0/24", + "origin_asn": 64497, + "collector": "rrc00", + "collector_location": { + "country": "Netherlands", + "city": "Amsterdam", + "latitude": 52.3676, + "longitude": 4.9041, + }, + "as_path": [3333, 64497], + "timestamp": "2026-03-30T10:00:00Z", + }, + "reference_date": "2026-03-30T10:00:00Z", + } + ] + + enriched = await enrich_bgp_events_for_batch(db, source="ris_live_bgp", events=events) + geography = enriched[0]["metadata"]["enrichment"]["prefix_geography"] + assert geography["source"] == "opengeofeed_public" + assert geography["confidence"] == "geofeed" + assert geography["city"] == "London" + + +@pytest.mark.asyncio +async def test_enrich_bgp_events_for_batch_falls_back_to_nro_delegated(): + nro_row = { + "extra_data": { + "family": "ipv4", + "range_start": "198.51.100.0", + "range_end": "198.51.100.255", + "prefix": "198.51.100.0/24", + "country_code": "DE", + "rir": "ripencc", + "source_dataset": "nro_delegated_stats", + "confidence": "registry_allocated", + } + } + + db = _FakeAsyncSession([[], [], [nro_row], []]) + events = [ + { + "metadata": { + "prefix": "198.51.100.0/24", + "origin_asn": 64512, + "collector": "rrc00", + "collector_location": { + "country": "Netherlands", + "city": "Amsterdam", + "latitude": 52.3676, + "longitude": 4.9041, + }, + "as_path": [3333, 64512], + "timestamp": "2026-03-30T10:00:00Z", + }, + "reference_date": "2026-03-30T10:00:00Z", + } + ] + + enriched = await enrich_bgp_events_for_batch(db, source="ris_live_bgp", events=events) + geography = enriched[0]["metadata"]["enrichment"]["prefix_geography"] + assert geography["source"] == "nro_delegated_stats" + assert geography["confidence"] == "registry_allocated" + assert geography["country"] == "德国" + + @pytest.mark.asyncio async def test_create_bgp_incidents_for_anomalies_aggregates_regions_and_collectors(): db = _FakeAsyncSession([[]]) diff --git a/docs/bgp-context.md b/docs/bgp-context.md index 99640091..4a21b907 100644 --- a/docs/bgp-context.md +++ b/docs/bgp-context.md @@ -98,11 +98,16 @@ Current enrichments: - new-origin detection - ASN organization profile from PeeringDB where available - prefix scope / impacted region hints +- prefix geography source priority: + - `OpenGeoFeed` (override/high confidence) + - `IPtoASN` (country-range baseline) + - `NRO delegated stats` (registry-allocation fallback) Current limitation: - `RPKI` is still placeholder-only and returns `unknown` - no real ROA validation source is integrated yet +- `inetnum` / `inet6num` whois fallback is still pending ## Current API Surface