diff --git a/dns_sync/reconciliation.py b/dns_sync/reconciliation.py index 4c6bfec..2cb8cb5 100644 --- a/dns_sync/reconciliation.py +++ b/dns_sync/reconciliation.py @@ -1,14 +1,18 @@ from collections import defaultdict from collections.abc import Iterable, Mapping, Sequence +from dataclasses import replace from typing import Protocol from .adapters import ( + CERTIFICATES, + PROXY_HOSTS, Capability, CapabilityOperation, DNS_RECORDS, Endpoint, EndpointPlan, NormalizedObject, + ObjectReference, Observation, PlanAction, PlanOperation, @@ -41,6 +45,12 @@ class ReconciliationPolicy(Protocol): observations: Sequence[Observation], ) -> NormalizedObject: ... + def equivalent( + self, + current: NormalizedObject, + desired: NormalizedObject, + ) -> bool: ... + def _observation_sort_key( observation: Observation, @@ -52,6 +62,94 @@ def _observation_sort_key( ) +def logicalize_observation_references( + observations: Iterable[Observation], +) -> tuple[Observation, ...]: + rows = tuple(observations) + + local_targets: dict[ + tuple[str, str, str], + str, + ] = {} + + for observation in rows: + object = observation.object + + if object.local_key is None: + continue + + identity = ( + observation.endpoint_id, + object.object_type, + object.local_key, + ) + + existing = local_targets.get(identity) + + if ( + existing is not None + and existing != object.logical_key + ): + raise ReconciliationConflict( + "Conflicting endpoint-local object identity for " + f"{observation.endpoint_id} " + f"{object.object_type} " + f"local key {object.local_key!r}" + ) + + local_targets[identity] = object.logical_key + + logicalized: list[Observation] = [] + + for observation in rows: + object = observation.object + + references = [] + + for reference in object.references: + if reference.logical_key is not None: + references.append(reference) + continue + + target = ( + observation.endpoint_id, + reference.object_type, + reference.local_key, + ) + logical_key = local_targets.get(target) + + if logical_key is None: + raise ReconciliationConflict( + "Unresolved endpoint-local reference from " + f"{observation.endpoint_id} " + f"{object.object_type} " + f"{object.logical_key!r}: " + f"{reference.relation} -> " + f"{reference.object_type} " + f"local key {reference.local_key!r}" + ) + + references.append( + ObjectReference( + relation=reference.relation, + object_type=reference.object_type, + logical_key=logical_key, + ) + ) + + logicalized.append( + replace( + observation, + object=replace( + object, + references=tuple(references), + ), + ) + ) + + return tuple(logicalized) + + def group_observations( observations: Iterable[Observation], ) -> dict[ObservationGroupKey, tuple[Observation, ...]]: @@ -116,9 +214,179 @@ class DnsUnionPolicy: payload=dict(first.payload), ) + def equivalent( + self, + current: NormalizedObject, + desired: NormalizedObject, + ) -> bool: + return ( + current.object_type == desired.object_type + and current.logical_key == desired.logical_key + and current.payload == desired.payload + and current.references == desired.references + ) + + +class CertificateNewestPolicy: + object_type = CERTIFICATES + + _PORTABLE_FIELDS = ( + "domain_names", + "valid_from", + "expires_on", + ) + + @classmethod + def _portable_state( + cls, + object: NormalizedObject, + ) -> tuple[object, ...]: + return tuple( + object.payload.get(field) + for field in cls._PORTABLE_FIELDS + ) + + @staticmethod + def _expiry_key( + object: NormalizedObject, + ) -> tuple[int, str]: + expires_on = object.payload.get("expires_on") + + if expires_on is None: + return (0, "") + + if not isinstance(expires_on, str): + raise ReconciliationConflict( + "Certificate expires_on must be normalized text" + ) + + return (1, expires_on) + + def reconcile( + self, + logical_key: str, + observations: Sequence[Observation], + ) -> NormalizedObject: + if not observations: + raise RuntimeError( + "Certificate policy requires at least one observation" + ) + + newest_key = max( + self._expiry_key(observation.object) + for observation in observations + ) + + newest = [ + observation + for observation in observations + if self._expiry_key(observation.object) == newest_key + ] + + baseline = self._portable_state(newest[0].object) + + conflicting_endpoints = { + observation.endpoint_id + for observation in newest + if self._portable_state(observation.object) != baseline + } + + if conflicting_endpoints: + conflicting_endpoints.add(newest[0].endpoint_id) + + raise ReconciliationConflict( + f"Conflicting {CERTIFICATES} observations for " + f"{logical_key!r} at newest expiry from " + + ", ".join(sorted(conflicting_endpoints)) + ) + + winner = newest[0].object + + return NormalizedObject( + object_type=CERTIFICATES, + logical_key=logical_key, + payload=dict(winner.payload), + ) + + def equivalent( + self, + current: NormalizedObject, + desired: NormalizedObject, + ) -> bool: + return ( + current.object_type == CERTIFICATES + and desired.object_type == CERTIFICATES + and current.logical_key == desired.logical_key + and self._portable_state(current) + == self._portable_state(desired) + ) + + +class ProxyHostConsensusPolicy: + object_type = PROXY_HOSTS + + def reconcile( + self, + logical_key: str, + observations: Sequence[Observation], + ) -> NormalizedObject: + if not observations: + raise RuntimeError( + "Proxy-host policy requires at least one observation" + ) + + first = observations[0].object + + conflicting_endpoints = { + observation.endpoint_id + for observation in observations + if ( + observation.object.payload != first.payload + or observation.object.references != first.references + ) + } + + if conflicting_endpoints: + conflicting_endpoints.add(observations[0].endpoint_id) + + raise ReconciliationConflict( + f"Conflicting {PROXY_HOSTS} observations for " + f"{logical_key!r} from " + + ", ".join(sorted(conflicting_endpoints)) + ) + + for reference in first.references: + if reference.local_key is not None: + raise ReconciliationConflict( + "Proxy-host desired state cannot contain " + "endpoint-local references" + ) + + return NormalizedObject( + object_type=PROXY_HOSTS, + logical_key=logical_key, + payload=dict(first.payload), + references=first.references, + ) + + def equivalent( + self, + current: NormalizedObject, + desired: NormalizedObject, + ) -> bool: + return ( + current.object_type == PROXY_HOSTS + and desired.object_type == PROXY_HOSTS + and current.logical_key == desired.logical_key + and current.payload == desired.payload + and current.references == desired.references + ) + DEFAULT_POLICIES: Mapping[str, ReconciliationPolicy] = { DNS_RECORDS: DnsUnionPolicy(), + CERTIFICATES: CertificateNewestPolicy(), + PROXY_HOSTS: ProxyHostConsensusPolicy(), } @@ -133,8 +401,12 @@ def reconcile_observations( ) desired: list[NormalizedObject] = [] + logicalized = logicalize_observation_references( + observations + ) + for (object_type, logical_key), rows in ( - group_observations(observations).items() + group_observations(logicalized).items() ): policy = selected_policies.get(object_type) if policy is None: @@ -156,6 +428,7 @@ def build_endpoint_plans( adapters: Iterable[PlanningAdapter], desired_objects: Iterable[NormalizedObject], observations: Iterable[Observation], + policies: Mapping[str, ReconciliationPolicy] | None = None, ) -> tuple[EndpointPlan, ...]: desired_by_identity: dict[ tuple[str, str], @@ -180,11 +453,22 @@ def build_endpoint_plans( ) desired_by_identity[identity] = desired + selected_policies = ( + DEFAULT_POLICIES + if policies is None + else policies + ) + current_by_identity: dict[ tuple[str, str, str], list[Observation], ] = defaultdict(list) - for observation in observations: + + logicalized_observations = ( + logicalize_observation_references(observations) + ) + + for observation in logicalized_observations: current_by_identity[ ( observation.endpoint_id, @@ -239,10 +523,25 @@ def build_endpoint_plans( ) if current is None: operation = PlanOperation.CREATE - elif current.payload == desired.payload: - operation = PlanOperation.NOOP else: - operation = PlanOperation.UPDATE + policy = selected_policies.get(object_type) + + if policy is None: + equivalent = ( + current.payload == desired.payload + and current.references == desired.references + ) + else: + equivalent = policy.equivalent( + current, + desired, + ) + + operation = ( + PlanOperation.NOOP + if equivalent + else PlanOperation.UPDATE + ) actions.append( PlanAction( diff --git a/tests/test_planning.py b/tests/test_planning.py index e92f68a..5c3b830 100644 --- a/tests/test_planning.py +++ b/tests/test_planning.py @@ -289,6 +289,217 @@ class EndpointPlanningTests(unittest.TestCase): ) + def test_certificate_provider_difference_is_noop_when_portable_state_matches(self): + endpoint = Endpoint("npm", "node-two:81") + + npm_adapter = FakeAdapter( + endpoint=endpoint, + capabilities=frozenset( + { + Capability( + CERTIFICATES, + CapabilityOperation.APPLY, + ), + } + ), + ) + + desired = NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + payload={ + "domain_names": ["app.example.com"], + "provider": "letsencrypt", + "nice_name": "source", + "valid_from": "2026-09-01T00:00:00+00:00", + "expires_on": "2026-12-01T00:00:00+00:00", + }, + ) + + current = Observation( + endpoint_id=endpoint.identity, + object=NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + local_key="84", + payload={ + "domain_names": ["app.example.com"], + "provider": "other", + "nice_name": "imported copy", + "valid_from": "2026-09-01T00:00:00+00:00", + "expires_on": "2026-12-01T00:00:00+00:00", + }, + ), + observed_at=NOW, + ) + + plans = build_endpoint_plans( + [npm_adapter], + [desired], + [current], + ) + + self.assertEqual( + plans[0].actions[0].operation, + PlanOperation.NOOP, + ) + + def test_proxy_reference_change_is_update_even_when_payload_matches(self): + endpoint = Endpoint("npm", "node-two:81") + + npm_adapter = FakeAdapter( + endpoint=endpoint, + capabilities=frozenset( + { + Capability( + PROXY_HOSTS, + CapabilityOperation.APPLY, + ), + } + ), + ) + + desired = NormalizedObject( + object_type=PROXY_HOSTS, + logical_key="app.example.com", + payload={ + "domain_names": ["app.example.com"], + "forward_scheme": "http", + "forward_host": "backend.internal", + "forward_port": 8080, + }, + references=( + ObjectReference( + relation="certificate", + object_type=CERTIFICATES, + logical_key="new.example.com", + ), + ), + ) + + observations = [ + Observation( + endpoint_id=endpoint.identity, + object=NormalizedObject( + object_type=CERTIFICATES, + logical_key="old.example.com", + local_key="67", + payload={ + "domain_names": ["old.example.com"], + }, + ), + observed_at=NOW, + ), + Observation( + endpoint_id=endpoint.identity, + object=NormalizedObject( + object_type=PROXY_HOSTS, + logical_key="app.example.com", + local_key="12", + payload=dict(desired.payload), + references=( + ObjectReference( + relation="certificate", + object_type=CERTIFICATES, + local_key="67", + ), + ), + ), + observed_at=NOW, + ), + ] + + plans = build_endpoint_plans( + [npm_adapter], + [desired], + observations, + ) + + self.assertEqual( + plans[0].actions[0].operation, + PlanOperation.UPDATE, + ) + + def test_planner_logicalizes_matching_local_reference_to_noop(self): + endpoint = Endpoint("npm", "node-two:81") + + npm_adapter = FakeAdapter( + endpoint=endpoint, + capabilities=frozenset( + { + Capability( + PROXY_HOSTS, + CapabilityOperation.APPLY, + ), + } + ), + ) + + desired = NormalizedObject( + object_type=PROXY_HOSTS, + logical_key="app.example.com", + payload={ + "domain_names": ["app.example.com"], + "forward_scheme": "http", + "forward_host": "backend.internal", + "forward_port": 8080, + }, + references=( + ObjectReference( + relation="certificate", + object_type=CERTIFICATES, + logical_key="app.example.com", + ), + ), + ) + + observations = [ + Observation( + endpoint_id=endpoint.identity, + object=NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + local_key="84", + payload={ + "domain_names": ["app.example.com"], + }, + ), + observed_at=NOW, + ), + Observation( + endpoint_id=endpoint.identity, + object=NormalizedObject( + object_type=PROXY_HOSTS, + logical_key="app.example.com", + local_key="12", + payload=dict(desired.payload), + references=( + ObjectReference( + relation="certificate", + object_type=CERTIFICATES, + local_key="84", + ), + ), + ), + observed_at=NOW, + ), + ] + + plans = build_endpoint_plans( + [npm_adapter], + [desired], + observations, + ) + + self.assertEqual( + plans[0].actions[0].operation, + PlanOperation.NOOP, + ) + self.assertEqual( + plans[0].actions[0].current.references, + desired.references, + ) + def test_duplicate_endpoint_objects_fail_without_payload_output(self): desired = [ dns_object("app.example.com", "192.168.1.101") diff --git a/tests/test_reconciliation.py b/tests/test_reconciliation.py index 7363775..ed9be0f 100644 --- a/tests/test_reconciliation.py +++ b/tests/test_reconciliation.py @@ -4,12 +4,15 @@ from datetime import datetime, timezone from dns_sync.adapters import ( CERTIFICATES, DNS_RECORDS, + PROXY_HOSTS, NormalizedObject, + ObjectReference, Observation, ) from dns_sync.reconciliation import ( ReconciliationConflict, group_observations, + logicalize_observation_references, reconcile_observations, ) @@ -166,12 +169,12 @@ class DnsUnionPolicyTests(unittest.TestCase): def test_missing_policy_fails_explicitly(self): observation = Observation( - endpoint_id="npm://node-one:81", + endpoint_id="generic://node-one", object=NormalizedObject( - object_type=CERTIFICATES, + object_type="unknown_objects", logical_key="app.example.com", local_key="67", - payload={"expires_on": "2026-12-15"}, + payload={}, ), observed_at=NOW, ) @@ -182,6 +185,298 @@ class DnsUnionPolicyTests(unittest.TestCase): ): reconcile_observations([observation]) + def test_local_certificate_references_become_logical_references(self): + observations = [ + Observation( + endpoint_id="npm://node-one:81", + object=NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + local_key="67", + payload={ + "domain_names": ["app.example.com"], + "valid_from": "2026-09-01T00:00:00+00:00", + "expires_on": "2026-12-01T00:00:00+00:00", + }, + ), + observed_at=NOW, + ), + Observation( + endpoint_id="npm://node-one:81", + object=NormalizedObject( + object_type=PROXY_HOSTS, + logical_key="app.example.com", + local_key="12", + payload={ + "domain_names": ["app.example.com"], + "forward_scheme": "http", + "forward_host": "backend.internal", + "forward_port": 8080, + "enabled": True, + }, + references=( + ObjectReference( + relation="certificate", + object_type=CERTIFICATES, + local_key="67", + ), + ), + ), + observed_at=NOW, + ), + ] + + logicalized = logicalize_observation_references( + observations + ) + + proxy = next( + row.object + for row in logicalized + if row.object.object_type == PROXY_HOSTS + ) + + self.assertEqual( + proxy.references, + ( + ObjectReference( + relation="certificate", + object_type=CERTIFICATES, + logical_key="app.example.com", + ), + ), + ) + + def test_unresolved_local_reference_fails_closed(self): + observation = Observation( + endpoint_id="npm://node-one:81", + object=NormalizedObject( + object_type=PROXY_HOSTS, + logical_key="app.example.com", + local_key="12", + payload={ + "domain_names": ["app.example.com"], + "forward_scheme": "http", + "forward_host": "backend.internal", + "forward_port": 8080, + }, + references=( + ObjectReference( + relation="certificate", + object_type=CERTIFICATES, + local_key="999", + ), + ), + ), + observed_at=NOW, + ) + + with self.assertRaisesRegex( + ReconciliationConflict, + "Unresolved endpoint-local reference", + ): + reconcile_observations([observation]) + + def test_certificate_policy_selects_newest_expiry(self): + observations = [ + Observation( + endpoint_id="npm://node-one:81", + object=NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + local_key="10", + payload={ + "domain_names": ["app.example.com"], + "provider": "other", + "nice_name": "older", + "valid_from": "2026-06-01T00:00:00+00:00", + "expires_on": "2026-09-01T00:00:00+00:00", + }, + ), + observed_at=NOW, + ), + Observation( + endpoint_id="npm://node-two:81", + object=NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + local_key="44", + payload={ + "domain_names": ["app.example.com"], + "provider": "letsencrypt", + "nice_name": "newer", + "valid_from": "2026-09-01T00:00:00+00:00", + "expires_on": "2026-12-01T00:00:00+00:00", + }, + ), + observed_at=NOW, + ), + ] + + desired = reconcile_observations(observations) + + self.assertEqual(len(desired), 1) + self.assertEqual( + desired[0].payload["expires_on"], + "2026-12-01T00:00:00+00:00", + ) + self.assertEqual( + desired[0].payload["provider"], + "letsencrypt", + ) + self.assertIsNone(desired[0].local_key) + + def test_certificate_same_expiry_conflict_fails_closed(self): + observations = [ + Observation( + endpoint_id="npm://node-one:81", + object=NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + local_key="10", + payload={ + "domain_names": ["app.example.com"], + "valid_from": "2026-08-01T00:00:00+00:00", + "expires_on": "2026-12-01T00:00:00+00:00", + }, + ), + observed_at=NOW, + ), + Observation( + endpoint_id="npm://node-two:81", + object=NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + local_key="44", + payload={ + "domain_names": ["app.example.com"], + "valid_from": "2026-09-01T00:00:00+00:00", + "expires_on": "2026-12-01T00:00:00+00:00", + }, + ), + observed_at=NOW, + ), + ] + + with self.assertRaisesRegex( + ReconciliationConflict, + "newest expiry", + ): + reconcile_observations(observations) + + def test_proxy_consensus_ignores_different_endpoint_local_certificate_ids(self): + observations = [] + + for endpoint_id, certificate_id, proxy_id in ( + ("npm://node-one:81", "67", "12"), + ("npm://node-two:81", "84", "31"), + ): + observations.extend( + [ + Observation( + endpoint_id=endpoint_id, + object=NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + local_key=certificate_id, + payload={ + "domain_names": ["app.example.com"], + "valid_from": "2026-09-01T00:00:00+00:00", + "expires_on": "2026-12-01T00:00:00+00:00", + }, + ), + observed_at=NOW, + ), + Observation( + endpoint_id=endpoint_id, + object=NormalizedObject( + object_type=PROXY_HOSTS, + logical_key="app.example.com", + local_key=proxy_id, + payload={ + "domain_names": ["app.example.com"], + "forward_scheme": "http", + "forward_host": "backend.internal", + "forward_port": 8080, + "enabled": True, + }, + references=( + ObjectReference( + relation="certificate", + object_type=CERTIFICATES, + local_key=certificate_id, + ), + ), + ), + observed_at=NOW, + ), + ] + ) + + desired = reconcile_observations(observations) + + proxy = next( + object + for object in desired + if object.object_type == PROXY_HOSTS + ) + + self.assertEqual( + proxy.references, + ( + ObjectReference( + relation="certificate", + object_type=CERTIFICATES, + logical_key="app.example.com", + ), + ), + ) + + def test_proxy_consensus_rejects_backend_drift_without_payload_output(self): + observations = [ + Observation( + endpoint_id="npm://node-one:81", + object=NormalizedObject( + object_type=PROXY_HOSTS, + logical_key="app.example.com", + local_key="12", + payload={ + "domain_names": ["app.example.com"], + "forward_scheme": "http", + "forward_host": "one.internal", + "forward_port": 8080, + }, + ), + observed_at=NOW, + ), + Observation( + endpoint_id="npm://node-two:81", + object=NormalizedObject( + object_type=PROXY_HOSTS, + logical_key="app.example.com", + local_key="31", + payload={ + "domain_names": ["app.example.com"], + "forward_scheme": "http", + "forward_host": "two.internal", + "forward_port": 8080, + }, + ), + observed_at=NOW, + ), + ] + + with self.assertRaises(ReconciliationConflict) as raised: + reconcile_observations(observations) + + message = str(raised.exception) + + self.assertIn("app.example.com", message) + self.assertIn("npm://node-one:81", message) + self.assertIn("npm://node-two:81", message) + self.assertNotIn("one.internal", message) + self.assertNotIn("two.internal", message) + def test_explicit_empty_policy_registry_uses_no_defaults(self): observation = dns_observation( "npm://node-one:81",