diff --git a/dns_sync/adapters.py b/dns_sync/adapters.py index 4ac9249..82a96d5 100644 --- a/dns_sync/adapters.py +++ b/dns_sync/adapters.py @@ -252,3 +252,35 @@ class PlanAction: class EndpointPlan: endpoint_id: str actions: tuple[PlanAction, ...] + retirements: tuple[RetireRequest, ...] = () + + def __post_init__(self) -> None: + retirements = tuple(self.retirements) + + if any( + not isinstance(retirement, RetireRequest) + for retirement in retirements + ): + raise ValueError( + "EndpointPlan retirements must be " + "RetireRequest instances" + ) + + identities = [ + ( + retirement.object.object_type, + retirement.object.local_key, + ) + for retirement in retirements + ] + + if len(set(identities)) != len(identities): + raise ValueError( + "EndpointPlan retirements cannot contain duplicates" + ) + + object.__setattr__( + self, + "retirements", + retirements, + ) diff --git a/dns_sync/executor.py b/dns_sync/executor.py index 7a57826..81809e7 100644 --- a/dns_sync/executor.py +++ b/dns_sync/executor.py @@ -321,6 +321,7 @@ def _preflight( dict[str, ExecutionAdapter], list[PlanAction], list[PlanAction], + list[tuple[str, RetireRequest]], ]: adapters_by_endpoint: dict[str, ExecutionAdapter] = {} for adapter in adapters: @@ -333,6 +334,9 @@ def _preflight( executable: list[PlanAction] = [] skipped: list[PlanAction] = [] + planned_retirements: list[ + tuple[str, RetireRequest] + ] = [] for plan in plans: adapter = adapters_by_endpoint.get(plan.endpoint_id) @@ -341,6 +345,14 @@ def _preflight( f"No adapter for endpoint plan {plan.endpoint_id}" ) + for retirement in plan.retirements: + planned_retirements.append( + ( + plan.endpoint_id, + retirement, + ) + ) + for action in plan.actions: if action.endpoint_id != plan.endpoint_id: raise RuntimeError( @@ -382,7 +394,12 @@ def _preflight( ) executable.append(action) - return adapters_by_endpoint, executable, skipped + return ( + adapters_by_endpoint, + executable, + skipped, + planned_retirements, + ) def execute_endpoint_plans( @@ -399,14 +416,20 @@ def execute_endpoint_plans( ) timestamp = timestamp.astimezone(timezone.utc) - adapters_by_endpoint, executable, skipped = _preflight( + ( + adapters_by_endpoint, + executable, + skipped, + planned_retirements, + ) = _preflight( adapters, plans, ) + applied: list[ExecutedAction] = [] pending_retirements: list[ tuple[str, RetireRequest] - ] = [] + ] = list(planned_retirements) local_references: dict[ str, @@ -421,6 +444,18 @@ def execute_endpoint_plans( ) ) + for action in skipped: + if ( + action.operation == PlanOperation.NOOP + and action.current is not None + and action.current.local_key is not None + ): + _remember_local_object( + local_references[action.endpoint_id], + action.current, + authoritative=True, + ) + for action in order_plan_actions(executable): adapter = adapters_by_endpoint[action.endpoint_id] resolved_action = _resolve_action_references( @@ -500,7 +535,23 @@ def execute_endpoint_plans( ) ) + retired_identities: set[ + tuple[str, str, str] + ] = set() + for endpoint_id, retirement in pending_retirements: + retired = retirement.object + identity = ( + endpoint_id, + retired.object_type, + retired.local_key, + ) + + if identity in retired_identities: + continue + + retired_identities.add(identity) + adapter = adapters_by_endpoint[endpoint_id] retire = getattr(adapter, "retire", None) diff --git a/dns_sync/reconciliation.py b/dns_sync/reconciliation.py index c8f1831..e61be0a 100644 --- a/dns_sync/reconciliation.py +++ b/dns_sync/reconciliation.py @@ -16,6 +16,7 @@ from .adapters import ( Observation, PlanAction, PlanOperation, + RetireRequest, ) from .dependencies import order_plan_actions @@ -342,6 +343,81 @@ class CertificateNewestPolicy: == self._portable_state(desired) ) + @staticmethod + def _equivalent_selection_key( + object: NormalizedObject, + ) -> tuple[int, str]: + # Preserve a native ACME-renewable certificate over an + # imported replica when both contain the same portable + # certificate state. + provider = object.payload.get("provider") + + return ( + 1 if provider == "letsencrypt" else 0, + object.local_key or "", + ) + + def select_current( + self, + observations: Sequence[Observation], + desired: NormalizedObject, + ) -> tuple[ + NormalizedObject, + tuple[RetireRequest, ...], + ]: + if not observations: + raise PlanningConflict( + "Certificate current-state selection requires " + "at least one observation" + ) + + candidates = [ + observation.object + for observation in observations + ] + + if any( + candidate.local_key is None + for candidate in candidates + ): + raise PlanningConflict( + "Observed certificate requires endpoint-local identity" + ) + + matching = [ + candidate + for candidate in candidates + if self.equivalent( + candidate, + desired, + ) + ] + + if matching: + selected = max( + matching, + key=self._equivalent_selection_key, + ) + else: + # No local copy is already desired. Replace the newest + # local certificate and retire every other stale copy + # after the replacement and its dependents verify. + selected = max( + candidates, + key=lambda candidate: ( + self._newest_key(candidate), + candidate.local_key or "", + ), + ) + + retirements = tuple( + RetireRequest(object=candidate) + for candidate in candidates + if candidate.local_key != selected.local_key + ) + + return selected, retirements + class ProxyHostConsensusPolicy: object_type = PROXY_HOSTS @@ -481,13 +557,31 @@ def build_endpoint_plans( else policies ) + raw_observations = tuple(observations) + current_by_identity: dict[ tuple[str, str, str], list[Observation], ] = defaultdict(list) + raw_current_by_identity: dict[ + tuple[str, str, str], + list[Observation], + ] = defaultdict(list) + + for observation in raw_observations: + raw_current_by_identity[ + ( + observation.endpoint_id, + observation.object.object_type, + observation.object.logical_key, + ) + ].append(observation) + logicalized_observations = ( - logicalize_observation_references(observations) + logicalize_observation_references( + raw_observations + ) ) for observation in logicalized_observations: @@ -521,6 +615,10 @@ def build_endpoint_plans( continue actions: list[PlanAction] = [] + planned_retirements: list[ + RetireRequest + ] = [] + for identity, desired in sorted( desired_by_identity.items() ): @@ -532,22 +630,42 @@ def build_endpoint_plans( (endpoint_id, object_type, logical_key), [], ) + policy = selected_policies.get(object_type) + if len(current_rows) > 1: - raise PlanningConflict( - f"Multiple endpoint objects for {endpoint_id} " - f"{object_type} {logical_key!r}" + selector = ( + getattr( + policy, + "select_current", + None, + ) + if policy is not None + else None ) - current = ( - current_rows[0].object - if current_rows - else None - ) + if selector is None: + raise PlanningConflict( + f"Multiple endpoint objects for {endpoint_id} " + f"{object_type} {logical_key!r}" + ) + + current, retirements = selector( + tuple(current_rows), + desired, + ) + + planned_retirements.extend( + retirements + ) + else: + current = ( + current_rows[0].object + if current_rows + else None + ) if current is None: operation = PlanOperation.CREATE else: - policy = selected_policies.get(object_type) - if policy is None: equivalent = ( current.payload == desired.payload @@ -590,15 +708,22 @@ def build_endpoint_plans( ) } - if replaced: + retiring_local = { + ( + retirement.object.object_type, + retirement.object.local_key, + ) + for retirement in planned_retirements + } + + if replaced or retiring_local: promoted: list[PlanAction] = [] for action in actions: desired = action.desired - if ( - action.operation == PlanOperation.NOOP - and desired is not None + logical_dependency_replaced = ( + desired is not None and any( ( reference.object_type, @@ -608,13 +733,51 @@ def build_endpoint_plans( for reference in desired.references if reference.logical_key is not None ) + ) + + raw_rows = ( + raw_current_by_identity.get( + ( + endpoint_id, + desired.object_type, + desired.logical_key, + ), + [], + ) + if desired is not None + else [] + ) + + local_dependency_retiring = any( + ( + reference.object_type, + reference.local_key, + ) + in retiring_local + for row in raw_rows + for reference in row.object.references + if reference.local_key is not None + ) + + if ( + action.operation == PlanOperation.NOOP + and ( + logical_dependency_replaced + or local_dependency_retiring + ) ): + reason = ( + "referenced object is being replaced" + if logical_dependency_replaced + else ( + "referenced local object is being retired" + ) + ) + action = replace( action, operation=PlanOperation.UPDATE, - reason=( - "referenced object is being replaced" - ), + reason=reason, ) promoted.append(action) @@ -625,6 +788,9 @@ def build_endpoint_plans( EndpointPlan( endpoint_id=endpoint_id, actions=order_plan_actions(actions), + retirements=tuple( + planned_retirements + ), ) ) diff --git a/tests/test_executor.py b/tests/test_executor.py index 9156c13..6eae350 100644 --- a/tests/test_executor.py +++ b/tests/test_executor.py @@ -414,6 +414,183 @@ class ExecutionTests(unittest.TestCase): ) + def test_restart_recovery_rebinds_before_planned_retirement(self): + endpoint = Endpoint( + "npm", + "192.168.1.4:81", + ) + + adapter = FakeAdapter( + endpoint, + capabilities( + CERTIFICATES, + PROXY_HOSTS, + ), + apply_local_keys={ + ( + PROXY_HOSTS, + "app.example.com", + ): "12", + }, + ) + + old_certificate = NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + local_key="67", + payload={ + "domain_names": ["app.example.com"], + "valid_from": "2026-08-01T00:00:00+00:00", + "expires_on": "2026-11-01T00:00:00+00:00", + }, + ) + + new_certificate = NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + local_key="84", + payload={ + "domain_names": ["app.example.com"], + "valid_from": "2026-09-01T00:00:00+00:00", + "expires_on": "2026-12-01T00:00:00+00:00", + }, + ) + + old_proxy = NormalizedObject( + object_type=PROXY_HOSTS, + logical_key="app.example.com", + local_key="12", + payload={ + "domain_names": ["app.example.com"], + "forward_scheme": "https", + "forward_host": "backend.internal", + "forward_port": 443, + "enabled": True, + }, + references=( + ObjectReference( + relation="certificate", + object_type=CERTIFICATES, + local_key="67", + ), + ), + ) + + desired_certificate = NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + payload=dict(new_certificate.payload), + ) + + desired_proxy = NormalizedObject( + object_type=PROXY_HOSTS, + logical_key="app.example.com", + payload=dict(old_proxy.payload), + references=( + ObjectReference( + relation="certificate", + object_type=CERTIFICATES, + logical_key="app.example.com", + ), + ), + ) + + plan = EndpointPlan( + endpoint.identity, + ( + PlanAction( + endpoint_id=endpoint.identity, + operation=PlanOperation.NOOP, + desired=desired_certificate, + current=new_certificate, + ), + PlanAction( + endpoint_id=endpoint.identity, + operation=PlanOperation.UPDATE, + desired=desired_proxy, + current=old_proxy, + reason=( + "referenced local object is being retired" + ), + ), + ), + retirements=( + RetireRequest( + object=old_certificate, + ), + ), + ) + + with SQLiteStateStore(":memory:") as store: + store.replace_observations( + endpoint.identity, + CERTIFICATES, + [ + Observation( + endpoint.identity, + old_certificate, + NOW, + ), + Observation( + endpoint.identity, + new_certificate, + NOW, + ), + ], + NOW, + ) + + store.replace_observations( + endpoint.identity, + PROXY_HOSTS, + [ + Observation( + endpoint.identity, + old_proxy, + NOW, + ) + ], + NOW, + ) + + execute_endpoint_plans( + [adapter], + [plan], + store, + executed_at=NOW, + ) + + self.assertEqual( + tuple(event[0] for event in adapter.events), + ( + "apply", + "verify", + "retire", + ), + ) + + proxy_action = adapter.applied_actions[0] + + self.assertEqual( + proxy_action.desired.references, + ( + ObjectReference( + relation="certificate", + object_type=CERTIFICATES, + local_key="84", + ), + ), + ) + + self.assertEqual( + adapter.retired, + [ + RetireRequest( + object=old_certificate, + ) + ], + ) + def test_failed_verification_is_persisted_and_raises(self): endpoint = Endpoint("verizon", "192.168.1.1:443") adapter = FakeAdapter( diff --git a/tests/test_planning.py b/tests/test_planning.py index a57b7ab..395d8d4 100644 --- a/tests/test_planning.py +++ b/tests/test_planning.py @@ -615,6 +615,217 @@ class EndpointPlanningTests(unittest.TestCase): ) + def test_restart_duplicate_certificate_uses_desired_copy_and_rebinds_proxy(self): + endpoint = Endpoint( + "generic", + "node.example:443", + ) + + generic_adapter = adapter( + "generic", + "node.example:443", + CERTIFICATES, + PROXY_HOSTS, + ) + + desired_certificate = NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + payload={ + "domain_names": ["app.example.com"], + "provider": "letsencrypt", + "valid_from": "2026-09-20T00:00:00+00:00", + "expires_on": "2026-12-20T00:00:00+00:00", + }, + ) + + desired_proxy = NormalizedObject( + object_type=PROXY_HOSTS, + logical_key="app.example.com", + payload={ + "domain_names": ["app.example.com"], + "forward_scheme": "https", + "forward_host": "backend.internal", + "forward_port": 443, + "enabled": True, + }, + references=( + ObjectReference( + relation="certificate", + object_type=CERTIFICATES, + logical_key="app.example.com", + ), + ), + ) + + old_certificate = NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + local_key="67", + payload={ + "domain_names": ["app.example.com"], + "provider": "other", + "valid_from": "2026-08-20T00:00:00+00:00", + "expires_on": "2026-11-20T00:00:00+00:00", + }, + ) + + new_certificate = NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + local_key="84", + payload={ + "domain_names": ["app.example.com"], + "provider": "other", + "valid_from": "2026-09-20T00:00:00+00:00", + "expires_on": "2026-12-20T00:00:00+00:00", + }, + ) + + current_proxy = NormalizedObject( + object_type=PROXY_HOSTS, + logical_key="app.example.com", + local_key="12", + payload=dict(desired_proxy.payload), + references=( + ObjectReference( + relation="certificate", + object_type=CERTIFICATES, + local_key="67", + ), + ), + ) + + plans = build_endpoint_plans( + [generic_adapter], + [ + desired_certificate, + desired_proxy, + ], + [ + observed( + endpoint.identity, + old_certificate, + ), + observed( + endpoint.identity, + new_certificate, + ), + observed( + endpoint.identity, + current_proxy, + ), + ], + ) + + self.assertEqual(len(plans), 1) + + plan = plans[0] + + by_type = { + action.desired.object_type: action + for action in plan.actions + } + + self.assertEqual( + by_type[CERTIFICATES].operation, + PlanOperation.NOOP, + ) + self.assertEqual( + by_type[CERTIFICATES].current.local_key, + "84", + ) + + self.assertEqual( + by_type[PROXY_HOSTS].operation, + PlanOperation.UPDATE, + ) + self.assertEqual( + by_type[PROXY_HOSTS].reason, + "referenced local object is being retired", + ) + + self.assertEqual( + tuple( + retirement.object.local_key + for retirement in plan.retirements + ), + ("67",), + ) + + def test_equivalent_duplicate_preserves_letsencrypt_copy(self): + endpoint = Endpoint( + "generic", + "node.example:443", + ) + + generic_adapter = adapter( + "generic", + "node.example:443", + CERTIFICATES, + ) + + desired = NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + payload={ + "domain_names": ["app.example.com"], + "valid_from": "2026-09-20T00:00:00+00:00", + "expires_on": "2026-12-20T00:00:00+00:00", + }, + ) + + imported = NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + local_key="90", + payload={ + "domain_names": ["app.example.com"], + "provider": "other", + "valid_from": "2026-09-20T00:00:00+00:00", + "expires_on": "2026-12-20T00:00:00+00:00", + }, + ) + + native = NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + local_key="12", + payload={ + "domain_names": ["app.example.com"], + "provider": "letsencrypt", + "valid_from": "2026-09-20T00:00:00+00:00", + "expires_on": "2026-12-20T00:00:00+00:00", + }, + ) + + plans = build_endpoint_plans( + [generic_adapter], + [desired], + [ + observed(endpoint.identity, imported), + observed(endpoint.identity, native), + ], + ) + + action = plans[0].actions[0] + + self.assertEqual( + action.operation, + PlanOperation.NOOP, + ) + self.assertEqual( + action.current.local_key, + "12", + ) + self.assertEqual( + tuple( + retirement.object.local_key + for retirement in plans[0].retirements + ), + ("90",), + ) + def test_duplicate_endpoint_objects_fail_without_payload_output(self): desired = [ dns_object("app.example.com", "192.168.1.101")