From bb68d0e38bc9ba61508a26096ea72d71727ab286 Mon Sep 17 00:00:00 2001 From: DeepZone Date: Tue, 19 May 2026 19:02:26 +0100 Subject: [PATCH] Harden registry origin extraction across IRR payload shapes --- backend/app/services/registry_checker.py | 96 +++++++++++++++++++++--- backend/tests/test_registry_checker.py | 75 ++++++++++++++++++ 2 files changed, 159 insertions(+), 12 deletions(-) diff --git a/backend/app/services/registry_checker.py b/backend/app/services/registry_checker.py index bbe5bee..639f4bc 100644 --- a/backend/app/services/registry_checker.py +++ b/backend/app/services/registry_checker.py @@ -1,4 +1,5 @@ from app.core.status import CheckStatus +import re class RegistryChecker: @@ -84,23 +85,94 @@ def check(self, prefix: str, origin_as: str | None, whois_payload: dict) -> dict } def _extract_route_origins(self, payload: dict) -> set[str]: - objects = payload.get("data", {}).get("irr_records") or payload.get("data", {}).get("records") or [] origins: set[str] = set() + data = payload.get("data", {}) if isinstance(payload, dict) else {} - for obj in objects: - route_seen = False - current_origin: str | None = None + for record in self._walk_records(data): + route_seen, normalized_origin = self._extract_fields_from_record(record) + if route_seen and normalized_origin: + origins.add(normalized_origin) - for field in obj if isinstance(obj, list) else []: - key = str(field.get("key", "")).lower() - value = str(field.get("value", "")).strip().upper() + return origins + + def _walk_records(self, data: dict): + sources = [] + if isinstance(data, dict): + sources.extend([data.get("irr_records"), data.get("records")]) + if "fields" in data: + sources.append(data.get("fields")) + for source in sources: + yield from self._walk_node(source) + + def _walk_node(self, node): + if isinstance(node, dict): + if isinstance(node.get("fields"), list): + yield node + else: + for value in node.values(): + yield from self._walk_node(value) + return + + if isinstance(node, list): + if self._looks_like_field_list(node): + yield node + else: + for item in node: + yield from self._walk_node(item) + + def _looks_like_field_list(self, node: list) -> bool: + return bool(node) and all(isinstance(item, dict) and "key" in item for item in node) + + def _extract_fields_from_record(self, record) -> tuple[bool, str | None]: + route_seen = False + current_origin: str | None = None + text_blobs: list[str] = [] + fields = [] + if isinstance(record, dict): + maybe_fields = record.get("fields") + if isinstance(maybe_fields, list): + fields = maybe_fields + else: + fields = [record] + elif isinstance(record, list): + fields = record + + for field in fields: + if isinstance(field, dict): + key = str(field.get("key", "")).strip().lower() + value = str(field.get("value", "")).strip() if key in {"route", "route6"} and value: route_seen = True - if key == "origin" and value.startswith("AS"): - current_origin = value + elif key == "origin": + normalized = self._normalize_origin(value) + if normalized: + current_origin = normalized + text_blobs.extend([key, value]) + elif isinstance(field, str): + text_blobs.append(field) - if route_seen and current_origin: - origins.add(current_origin) + if route_seen and current_origin: + return True, current_origin - return origins + combined_text = " ".join(part for part in text_blobs if part) + if not route_seen and re.search(r"\broute6?\b\s*:", combined_text, re.IGNORECASE): + route_seen = True + if not route_seen and re.search(r"\broute6?\b\s+\S+", combined_text, re.IGNORECASE): + route_seen = True + + if route_seen and not current_origin: + origin_match = re.search(r"\borigin\b\s*:?\s*(AS)?\s*(\d+)\b", combined_text, re.IGNORECASE) + if origin_match: + current_origin = self._normalize_origin(origin_match.group(0)) + + return route_seen, current_origin + + def _normalize_origin(self, value: str) -> str | None: + if not value: + return None + cleaned = str(value).strip().upper() + match = re.search(r"\b(AS)?\s*(\d+)\b", cleaned) + if not match: + return None + return f"AS{match.group(2)}" diff --git a/backend/tests/test_registry_checker.py b/backend/tests/test_registry_checker.py index f819f11..1eb79ce 100644 --- a/backend/tests/test_registry_checker.py +++ b/backend/tests/test_registry_checker.py @@ -40,3 +40,78 @@ def test_route_object_origin_mismatch_is_critical(): def test_empty_data_is_unknown(): result = RegistryChecker().check("193.0.6.0/24", "AS3333", {}) assert result["status"] == "UNKNOWN" + + +def test_irr_records_structure_finds_origin(): + payload = { + "data": { + "irr_records": [ + [ + {"key": "route", "value": "193.0.6.0/24"}, + {"key": "origin", "value": "as3333"}, + ] + ] + } + } + assert RegistryChecker()._extract_route_origins(payload) == {"AS3333"} + + +def test_fields_structure_finds_origin(): + payload = { + "data": { + "records": [ + { + "type": "route", + "fields": [ + {"key": "route", "value": "193.0.6.0/24"}, + {"key": "origin", "value": "3333"}, + ], + } + ] + } + } + assert RegistryChecker()._extract_route_origins(payload) == {"AS3333"} + + +def test_nested_structure_finds_origin(): + payload = { + "data": { + "records": [ + [ + [ + {"key": "route6", "value": "2001:db8::/32"}, + {"key": "origin", "value": "AS3333"}, + ] + ] + ] + } + } + assert RegistryChecker()._extract_route_origins(payload) == {"AS3333"} + + +def test_origin_without_route_is_ignored(): + payload = {"data": {"records": [[{"key": "origin", "value": "AS3333"}]]}} + assert RegistryChecker()._extract_route_origins(payload) == set() + + +def test_route_without_origin_has_no_origin(): + payload = {"data": {"records": [[{"key": "route", "value": "193.0.6.0/24"}]]}} + assert RegistryChecker()._extract_route_origins(payload) == set() + + +def test_string_origin_extraction_requires_route_in_same_record(): + payload = { + "data": { + "records": [ + [{"key": "remarks", "value": "route: 193.0.6.0/24 origin: AS3333"}], + [{"key": "remarks", "value": "origin: AS64500"}], + ] + } + } + assert RegistryChecker()._extract_route_origins(payload) == {"AS3333"} + + +def test_unknown_structure_does_not_crash_and_is_warning_with_data(): + payload = {"data": {"records": [{"unexpected": {"nested": 1}}]}} + result = RegistryChecker().check("193.0.6.0/24", "AS3333", payload) + assert result["status"] == "WARNING"