diff --git a/dns_sync/executor.py b/dns_sync/executor.py index 81809e7..4daa588 100644 --- a/dns_sync/executor.py +++ b/dns_sync/executor.py @@ -314,6 +314,17 @@ def _required_capability( ) +def _required_retirement_support( + adapter: ExecutionAdapter, +) -> None: + retire = getattr(adapter, "retire", None) + if not callable(retire): + raise RuntimeError( + f"{adapter.endpoint.identity} adapter does not support " + "deferred retirement" + ) + + def _preflight( adapters: Iterable[ExecutionAdapter], plans: Iterable[EndpointPlan], @@ -345,6 +356,9 @@ def _preflight( f"No adapter for endpoint plan {plan.endpoint_id}" ) + if plan.retirements: + _required_retirement_support(adapter) + for retirement in plan.retirements: planned_retirements.append( ( @@ -381,6 +395,9 @@ def _preflight( "Executable plan action requires desired state" ) + if action.operation == PlanOperation.REPLACE: + _required_retirement_support(adapter) + object_type = action.desired.object_type _required_capability( adapter, diff --git a/dns_sync/orchestrator.py b/dns_sync/orchestrator.py index ff65c2d..8c929be 100644 --- a/dns_sync/orchestrator.py +++ b/dns_sync/orchestrator.py @@ -21,6 +21,7 @@ from .persistence import SQLiteStateStore from .projections import project_proxy_hosts_to_dns from .reconciliation import ( build_endpoint_plans, + merge_desired_with_previous, reconcile_observations, ) @@ -315,6 +316,17 @@ def discover_reconcile_plan( else [] ) + if certificate_authoritative: + previous_certificates = [ + row.object + for row in store.load_desired_state(CERTIFICATES) + ] + + npm_desired = merge_desired_with_previous( + npm_desired, + previous_certificates, + ) + proxy_observations = _authoritative_observations( observations, live_pairs, diff --git a/dns_sync/plugins/adapters/npm.py b/dns_sync/plugins/adapters/npm.py index c70818d..faa40e7 100644 --- a/dns_sync/plugins/adapters/npm.py +++ b/dns_sync/plugins/adapters/npm.py @@ -43,6 +43,10 @@ class Plugin: PROXY_HOSTS, CapabilityOperation.DISCOVER, ), + Capability( + PROXY_HOSTS, + CapabilityOperation.APPLY, + ), Capability( PROXY_HOSTS, CapabilityOperation.VERIFY, @@ -51,6 +55,10 @@ class Plugin: CERTIFICATES, CapabilityOperation.DISCOVER, ), + Capability( + CERTIFICATES, + CapabilityOperation.APPLY, + ), Capability( CERTIFICATES, CapabilityOperation.VERIFY, diff --git a/dns_sync/reconciliation.py b/dns_sync/reconciliation.py index e61be0a..4627da2 100644 --- a/dns_sync/reconciliation.py +++ b/dns_sync/reconciliation.py @@ -54,6 +54,46 @@ class ReconciliationPolicy(Protocol): ) -> bool: ... +def merge_desired_with_previous( + fresh: Iterable[NormalizedObject], + previous: Iterable[NormalizedObject], + policies: Mapping[str, ReconciliationPolicy] | None = None, +) -> list[NormalizedObject]: + selected_policies = ( + DEFAULT_POLICIES + if policies is None + else policies + ) + + previous_by_identity = { + (object.object_type, object.logical_key): object + for object in previous + } + + merged: list[NormalizedObject] = [] + + for candidate in fresh: + identity = ( + candidate.object_type, + candidate.logical_key, + ) + prior = previous_by_identity.get(identity) + policy = selected_policies.get(candidate.object_type) + + merger = ( + getattr(policy, "merge_desired", None) + if policy is not None + else None + ) + + if prior is not None and callable(merger): + candidate = merger(prior, candidate) + + merged.append(candidate) + + return merged + + def _observation_sort_key( observation: Observation, ) -> tuple[str, str, str]: @@ -343,6 +383,26 @@ class CertificateNewestPolicy: == self._portable_state(desired) ) + def merge_desired( + self, + previous: NormalizedObject, + fresh: NormalizedObject, + ) -> NormalizedObject: + if ( + previous.object_type != CERTIFICATES + or fresh.object_type != CERTIFICATES + or previous.logical_key != fresh.logical_key + ): + raise ReconciliationConflict( + "Certificate desired-state merge requires " + "matching certificate identities" + ) + + if self._newest_key(previous) > self._newest_key(fresh): + return previous + + return fresh + @staticmethod def _equivalent_selection_key( object: NormalizedObject, diff --git a/tests/test_executor.py b/tests/test_executor.py index 6eae350..5b1a87a 100644 --- a/tests/test_executor.py +++ b/tests/test_executor.py @@ -655,6 +655,148 @@ class ExecutionTests(unittest.TestCase): ) self.assertEqual(store.load_applied_state(), []) + def test_preflight_requires_retire_for_planned_retirement_before_mutation( + self, + ): + endpoint = Endpoint("npm", "192.168.1.4:81") + inner = FakeAdapter( + endpoint, + capabilities(CERTIFICATES), + ) + + class NoRetireAdapter: + def __init__(self, wrapped): + self.endpoint = wrapped.endpoint + self.capabilities = wrapped.capabilities + self.wrapped = wrapped + + def apply(self, plan_action): + return self.wrapped.apply(plan_action) + + def verify(self, object): + return self.wrapped.verify(object) + + adapter = NoRetireAdapter(inner) + + old_certificate = NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + local_key="67", + payload={ + "domain_names": ["app.example.com"], + }, + ) + + plan = EndpointPlan( + endpoint.identity, + ( + action( + endpoint.identity, + PlanOperation.CREATE, + object_type=CERTIFICATES, + logical_key="other.example.com", + ), + ), + retirements=( + RetireRequest( + object=old_certificate, + ), + ), + ) + + with SQLiteStateStore(":memory:") as store: + with self.assertRaisesRegex( + RuntimeError, + "does not support deferred retirement", + ): + execute_endpoint_plans( + [adapter], + [plan], + store, + executed_at=NOW, + ) + + self.assertEqual( + store.load_applied_state(), + [], + ) + + self.assertEqual(inner.events, []) + self.assertEqual(inner.applied_actions, []) + + def test_preflight_requires_retire_for_replace_before_mutation( + self, + ): + endpoint = Endpoint("npm", "192.168.1.4:81") + inner = FakeAdapter( + endpoint, + capabilities(CERTIFICATES), + ) + + class NoRetireAdapter: + def __init__(self, wrapped): + self.endpoint = wrapped.endpoint + self.capabilities = wrapped.capabilities + self.wrapped = wrapped + + def apply(self, plan_action): + return self.wrapped.apply(plan_action) + + def verify(self, object): + return self.wrapped.verify(object) + + adapter = NoRetireAdapter(inner) + + current = NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + local_key="67", + payload={ + "domain_names": ["app.example.com"], + "expires_on": "2026-11-01T00:00:00+00:00", + }, + ) + desired = NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + payload={ + "domain_names": ["app.example.com"], + "expires_on": "2026-12-01T00:00:00+00:00", + }, + ) + + plan = EndpointPlan( + endpoint.identity, + ( + PlanAction( + endpoint_id=endpoint.identity, + operation=PlanOperation.REPLACE, + desired=desired, + current=current, + ), + ), + ) + + with SQLiteStateStore(":memory:") as store: + with self.assertRaisesRegex( + RuntimeError, + "does not support deferred retirement", + ): + execute_endpoint_plans( + [adapter], + [plan], + store, + executed_at=NOW, + ) + + self.assertEqual( + store.load_applied_state(), + [], + ) + + self.assertEqual(inner.events, []) + self.assertEqual(inner.applied_actions, []) + def test_preflight_requires_verify_before_any_mutation(self): endpoint = Endpoint("verizon", "192.168.1.1:443") adapter = FakeAdapter( diff --git a/tests/test_npm_adapter.py b/tests/test_npm_adapter.py index a85ceb9..4c3c2ee 100644 --- a/tests/test_npm_adapter.py +++ b/tests/test_npm_adapter.py @@ -94,7 +94,7 @@ class NpmAdapterTests(unittest.TestCase): self.assertIs(adapters[0].client, first_client) self.assertIs(adapters[1].client, second_client) - def test_capabilities_cover_discovery_and_verify_only(self): + def test_capabilities_cover_discovery_apply_and_verify(self): adapter = self.adapter(Mock()) self.assertEqual( @@ -105,6 +105,10 @@ class NpmAdapterTests(unittest.TestCase): PROXY_HOSTS, CapabilityOperation.DISCOVER, ), + Capability( + PROXY_HOSTS, + CapabilityOperation.APPLY, + ), Capability( PROXY_HOSTS, CapabilityOperation.VERIFY, @@ -113,6 +117,10 @@ class NpmAdapterTests(unittest.TestCase): CERTIFICATES, CapabilityOperation.DISCOVER, ), + Capability( + CERTIFICATES, + CapabilityOperation.APPLY, + ), Capability( CERTIFICATES, CapabilityOperation.VERIFY, @@ -121,21 +129,6 @@ class NpmAdapterTests(unittest.TestCase): ), ) - self.assertNotIn( - Capability( - PROXY_HOSTS, - CapabilityOperation.APPLY, - ), - adapter.capabilities, - ) - self.assertNotIn( - Capability( - CERTIFICATES, - CapabilityOperation.APPLY, - ), - adapter.capabilities, - ) - def test_discovers_proxy_hosts_with_endpoint_local_key(self): client = Mock() client.proxy_hosts.return_value = [ diff --git a/tests/test_orchestrator_authority.py b/tests/test_orchestrator_authority.py index 3eecd16..057dfa4 100644 --- a/tests/test_orchestrator_authority.py +++ b/tests/test_orchestrator_authority.py @@ -240,6 +240,109 @@ class ObservationAuthorityTests(unittest.TestCase): ) ) + def test_persisted_newer_certificate_cannot_regress_to_older_live_certificate( + self, + ): + endpoint = Endpoint( + "npm", + "192.168.1.3:81", + ) + + live = FakeAdapter( + endpoint=endpoint, + capabilities=capability_set( + CERTIFICATES, + apply=True, + ), + discovered={ + CERTIFICATES: [ + discovered_certificate( + "20", + valid_from=( + "2026-09-15T00:00:00+00:00" + ), + expires_on=( + "2026-12-15T00:00:00+00:00" + ), + ) + ] + }, + ) + + persisted = normalized_certificate( + "99", + valid_from=( + "2026-09-19T00:00:00+00:00" + ), + expires_on=( + "2026-12-19T00:00:00+00:00" + ), + ) + + persisted = NormalizedObject( + object_type=persisted.object_type, + logical_key=persisted.logical_key, + payload=dict(persisted.payload), + ) + + with SQLiteStateStore(":memory:") as store: + store.replace_desired_state( + CERTIFICATES, + [persisted], + T1, + ) + + result = discover_reconcile_plan( + [live], + store, + target_ip="192.168.1.101", + run_at=T2, + ) + + stored = store.load_desired_state( + CERTIFICATES + ) + + certificates = [ + object + for object in result.desired + if object.object_type == CERTIFICATES + ] + + self.assertEqual(len(certificates), 1) + self.assertEqual( + certificates[0].payload["valid_from"], + "2026-09-19T00:00:00+00:00", + ) + self.assertEqual( + certificates[0].payload["expires_on"], + "2026-12-19T00:00:00+00:00", + ) + + self.assertEqual(len(stored), 1) + self.assertEqual( + stored[0].object, + persisted, + ) + + self.assertEqual(len(result.plans), 1) + self.assertEqual( + result.plans[0].endpoint_id, + endpoint.identity, + ) + self.assertEqual( + len(result.plans[0].actions), + 1, + ) + self.assertEqual( + result.plans[0].actions[0].operation.value, + "replace", + ) + self.assertEqual( + result.plans[0].actions[0].desired, + persisted, + ) + def test_all_failed_sources_preserve_persisted_desired_without_planning(self): npm_endpoint = Endpoint( "npm",