diff --git a/dns_sync/object_normalizer.py b/dns_sync/object_normalizer.py new file mode 100644 index 0000000..858a3e2 --- /dev/null +++ b/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 + ] diff --git a/tests/test_object_normalizer.py b/tests/test_object_normalizer.py new file mode 100644 index 0000000..29cca4f --- /dev/null +++ b/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()