Normalize endpoint observations centrally

dns_sync/object_normalizer.py

@@ -0,0 +1,215 @@
+from collections.abc import Iterable
+from datetime import datetime, timezone
+from typing import Any
+
+from .adapters import (
+    CERTIFICATES,
+    DNS_RECORDS,
+    PROXY_HOSTS,
+    DiscoveredObject,
+    NormalizedObject,
+    Observation,
+)
+from .certificates.normalizer import (
+    normalize_certificate_domains,
+    parse_npm_datetime,
+)
+from .normalizer import normalize_hostname, normalize_ipv4
+
+
+_PROXY_BOOLEAN_FIELDS = (
+    "ssl_forced",
+    "caching_enabled",
+    "block_exploits",
+    "allow_websocket_upgrade",
+    "http2_support",
+    "hsts_enabled",
+    "hsts_subdomains",
+    "enabled",
+)
+
+
+def _normalized_domains(values: Any, label: str) -> tuple[str, ...]:
+    if not isinstance(values, (list, tuple)):
+        raise RuntimeError(f"{label} domain_names must be a list")
+
+    domains: set[str] = set()
+    for raw_domain in values:
+        hostname, _ = normalize_hostname(
+            str(raw_domain),
+            allow_wildcard=True,
+        )
+        domains.add(hostname)
+
+    if not domains:
+        raise RuntimeError(f"{label} has no usable domain names")
+
+    return tuple(sorted(domains))
+
+
+def _normalized_bool(value: Any) -> bool:
+    return value is True or value == 1
+
+
+def _normalized_port(value: Any, label: str) -> int:
+    try:
+        port = int(value)
+    except (TypeError, ValueError) as exc:
+        raise RuntimeError(f"{label} has invalid forward_port") from exc
+
+    if not 1 <= port <= 65535:
+        raise RuntimeError(
+            f"{label} forward_port must be between 1 and 65535"
+        )
+    return port
+
+
+def _iso_datetime(value: Any) -> str | None:
+    parsed = parse_npm_datetime(value)
+    return parsed.isoformat() if parsed is not None else None
+
+
+def _normalize_dns_record(
+    discovered: DiscoveredObject,
+) -> NormalizedObject:
+    payload = discovered.payload
+    hostname, wildcard = normalize_hostname(
+        str(payload.get("hostname") or ""),
+        allow_wildcard=True,
+    )
+    record_type = str(payload.get("record_type") or "A").upper()
+
+    if record_type != "A":
+        raise RuntimeError(
+            f"{hostname}: unsupported DNS record type {record_type}"
+        )
+
+    normalized_payload = {
+        "hostname": hostname,
+        "ip": normalize_ipv4(str(payload.get("ip") or ""), hostname),
+        "record_type": record_type,
+        "wildcard": wildcard,
+    }
+    return NormalizedObject(
+        object_type=DNS_RECORDS,
+        logical_key=hostname,
+        local_key=discovered.local_key,
+        payload=normalized_payload,
+    )
+
+
+def _normalize_proxy_host(
+    discovered: DiscoveredObject,
+) -> NormalizedObject:
+    payload = discovered.payload
+    label = f"NPM proxy host {discovered.local_key}"
+    domains = _normalized_domains(
+        payload.get("domain_names"),
+        label,
+    )
+
+    forward_scheme = str(
+        payload.get("forward_scheme") or ""
+    ).strip().lower()
+    forward_host = str(
+        payload.get("forward_host") or ""
+    ).strip()
+
+    if not forward_scheme:
+        raise RuntimeError(f"{label} has no forward_scheme")
+    if not forward_host:
+        raise RuntimeError(f"{label} has no forward_host")
+
+    certificate_id = payload.get("certificate_id")
+    if certificate_id in (None, "", 0, "0"):
+        certificate_local_key = None
+    else:
+        certificate_local_key = str(certificate_id)
+
+    normalized_payload: dict[str, Any] = {
+        "domain_names": list(domains),
+        "forward_scheme": forward_scheme,
+        "forward_host": forward_host,
+        "forward_port": _normalized_port(
+            payload.get("forward_port"),
+            label,
+        ),
+        "certificate_local_key": certificate_local_key,
+    }
+    for field in _PROXY_BOOLEAN_FIELDS:
+        normalized_payload[field] = _normalized_bool(
+            payload.get(field)
+        )
+
+    return NormalizedObject(
+        object_type=PROXY_HOSTS,
+        logical_key=",".join(domains),
+        local_key=discovered.local_key,
+        payload=normalized_payload,
+    )
+
+
+def _normalize_certificate(
+    discovered: DiscoveredObject,
+) -> NormalizedObject:
+    payload = discovered.payload
+    identity = normalize_certificate_domains(
+        payload.get("domain_names")
+    )
+    normalized_payload = {
+        "domain_names": list(identity.domains),
+        "provider": str(payload.get("provider") or ""),
+        "nice_name": str(payload.get("nice_name") or ""),
+        "valid_from": _iso_datetime(payload.get("valid_from")),
+        "expires_on": _iso_datetime(payload.get("expires_on")),
+        "cluster_managed": _normalized_bool(
+            payload.get("cluster_managed")
+        ),
+        "acme_renew": _normalized_bool(payload.get("acme_renew")),
+        "dns_challenge": _normalized_bool(
+            payload.get("dns_challenge")
+        ),
+    }
+    return NormalizedObject(
+        object_type=CERTIFICATES,
+        logical_key=identity.key,
+        local_key=discovered.local_key,
+        payload=normalized_payload,
+    )
+
+
+def normalize_discovered_object(
+    discovered: DiscoveredObject,
+) -> NormalizedObject:
+    if discovered.object_type == DNS_RECORDS:
+        return _normalize_dns_record(discovered)
+    if discovered.object_type == PROXY_HOSTS:
+        return _normalize_proxy_host(discovered)
+    if discovered.object_type == CERTIFICATES:
+        return _normalize_certificate(discovered)
+
+    raise RuntimeError(
+        f"No normalizer for object type {discovered.object_type!r}"
+    )
+
+
+def normalize_observations(
+    endpoint_id: str,
+    discovered: Iterable[DiscoveredObject],
+    observed_at: datetime | None = None,
+) -> list[Observation]:
+    if not endpoint_id:
+        raise RuntimeError("Observation endpoint_id cannot be empty")
+
+    timestamp = observed_at or datetime.now(timezone.utc)
+    if timestamp.tzinfo is None:
+        raise RuntimeError("Observation timestamp must be timezone-aware")
+
+    return [
+        Observation(
+            endpoint_id=endpoint_id,
+            object=normalize_discovered_object(item),
+            observed_at=timestamp.astimezone(timezone.utc),
+        )
+        for item in discovered
+    ]

tests/test_object_normalizer.py

@@ -0,0 +1,203 @@
+import unittest
+from datetime import datetime, timedelta, timezone
+
+from dns_sync.adapters import (
+    CERTIFICATES,
+    DNS_RECORDS,
+    PROXY_HOSTS,
+    DiscoveredObject,
+)
+from dns_sync.object_normalizer import (
+    normalize_discovered_object,
+    normalize_observations,
+)
+
+
+class CertificateNormalizationTests(unittest.TestCase):
+    def test_same_domains_share_identity_but_keep_local_keys(self):
+        rows = [
+            DiscoveredObject(
+                object_type=CERTIFICATES,
+                local_key="63",
+                payload={
+                    "domain_names": [
+                        "APP.EXAMPLE.COM",
+                        "example.com.",
+                    ],
+                    "provider": "letsencrypt",
+                    "expires_on": "2026-12-01T00:00:00Z",
+                },
+            ),
+            DiscoveredObject(
+                object_type=CERTIFICATES,
+                local_key="67",
+                payload={
+                    "domain_names": [
+                        "example.com",
+                        "app.example.com",
+                    ],
+                    "provider": "other",
+                    "expires_on": "2026-12-15T00:00:00Z",
+                },
+            ),
+        ]
+
+        observations = normalize_observations(
+            "npm://192.168.1.3:81",
+            rows,
+        )
+
+        self.assertEqual(
+            observations[0].object.logical_key,
+            "app.example.com,example.com",
+        )
+        self.assertEqual(
+            observations[0].object.logical_key,
+            observations[1].object.logical_key,
+        )
+        self.assertEqual(
+            tuple(item.object.local_key for item in observations),
+            ("63", "67"),
+        )
+        self.assertEqual(
+            observations[1].object.payload["expires_on"],
+            "2026-12-15T00:00:00+00:00",
+        )
+
+
+class ProxyHostNormalizationTests(unittest.TestCase):
+    def test_proxy_host_identity_and_payload_are_deterministic(self):
+        discovered = DiscoveredObject(
+            object_type=PROXY_HOSTS,
+            local_key="12",
+            payload={
+                "domain_names": [
+                    "WWW.EXAMPLE.COM.",
+                    "example.com",
+                    "www.example.com",
+                ],
+                "forward_scheme": "HTTP",
+                "forward_host": "backend.internal",
+                "forward_port": "8080",
+                "certificate_id": 67,
+                "ssl_forced": 1,
+                "enabled": True,
+            },
+        )
+
+        normalized = normalize_discovered_object(discovered)
+
+        self.assertEqual(
+            normalized.logical_key,
+            "example.com,www.example.com",
+        )
+        self.assertEqual(
+            normalized.payload["domain_names"],
+            ["example.com", "www.example.com"],
+        )
+        self.assertEqual(normalized.payload["forward_scheme"], "http")
+        self.assertEqual(normalized.payload["forward_port"], 8080)
+        self.assertEqual(
+            normalized.payload["certificate_local_key"],
+            "67",
+        )
+        self.assertTrue(normalized.payload["ssl_forced"])
+        self.assertTrue(normalized.payload["enabled"])
+        self.assertFalse(normalized.payload["hsts_enabled"])
+
+    def test_invalid_forward_port_is_rejected(self):
+        discovered = DiscoveredObject(
+            object_type=PROXY_HOSTS,
+            local_key="12",
+            payload={
+                "domain_names": ["example.com"],
+                "forward_scheme": "http",
+                "forward_host": "backend.internal",
+                "forward_port": 70000,
+            },
+        )
+
+        with self.assertRaisesRegex(RuntimeError, "between 1 and 65535"):
+            normalize_discovered_object(discovered)
+
+
+class DnsNormalizationTests(unittest.TestCase):
+    def test_dns_record_normalizes_hostname_and_ipv4(self):
+        discovered = DiscoveredObject(
+            object_type=DNS_RECORDS,
+            local_key="4",
+            payload={
+                "hostname": "APP.EXAMPLE.COM.",
+                "ip": "192.168.001.101",
+                "record_type": "a",
+            },
+        )
+
+        with self.assertRaisesRegex(RuntimeError, "valid IP address"):
+            normalize_discovered_object(discovered)
+
+        valid = DiscoveredObject(
+            object_type=DNS_RECORDS,
+            local_key="4",
+            payload={
+                "hostname": "APP.EXAMPLE.COM.",
+                "ip": "192.168.1.101",
+                "record_type": "a",
+            },
+        )
+        normalized = normalize_discovered_object(valid)
+
+        self.assertEqual(normalized.logical_key, "app.example.com")
+        self.assertEqual(normalized.payload["record_type"], "A")
+        self.assertEqual(normalized.payload["ip"], "192.168.1.101")
+
+
+class ObservationNormalizationTests(unittest.TestCase):
+    def test_batch_uses_one_utc_timestamp(self):
+        observed_at = datetime(
+            2026,
+            9,
+            19,
+            16,
+            0,
+            tzinfo=timezone(timedelta(hours=-4)),
+        )
+        row = DiscoveredObject(
+            object_type=DNS_RECORDS,
+            local_key="1",
+            payload={
+                "hostname": "example.com",
+                "ip": "192.168.1.101",
+            },
+        )
+
+        observations = normalize_observations(
+            "verizon://192.168.1.1",
+            [row, row],
+            observed_at=observed_at,
+        )
+
+        self.assertEqual(
+            {item.observed_at.isoformat() for item in observations},
+            {"2026-09-19T20:00:00+00:00"},
+        )
+
+    def test_unsupported_type_and_naive_timestamp_are_rejected(self):
+        unsupported = DiscoveredObject(
+            object_type="unknown",
+            local_key="1",
+            payload={},
+        )
+        with self.assertRaisesRegex(RuntimeError, "No normalizer"):
+            normalize_discovered_object(unsupported)
+
+        with self.assertRaisesRegex(RuntimeError, "timezone-aware"):
+            normalize_observations(
+                "npm://192.168.1.3:81",
+                [],
+                observed_at=datetime(2026, 9, 19),
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()