diff --git a/dns_sync/adapters.py b/dns_sync/adapters.py index 63fe686..4ac9249 100644 --- a/dns_sync/adapters.py +++ b/dns_sync/adapters.py @@ -19,6 +19,7 @@ class CapabilityOperation(StrEnum): class PlanOperation(StrEnum): CREATE = "create" UPDATE = "update" + REPLACE = "replace" DELETE = "delete" NOOP = "noop" SKIP = "skip" @@ -157,16 +158,69 @@ class NormalizedObject: ) +@dataclass(frozen=True) +class RetireRequest: + object: NormalizedObject + + def __post_init__(self) -> None: + if self.object.local_key is None: + raise ValueError( + "Retired object requires an endpoint-local key" + ) + + @dataclass(frozen=True) class ApplyResult: object: NormalizedObject detail: str | None = None + retirements: tuple[RetireRequest, ...] = () def __post_init__(self) -> None: if self.object.local_key is None: raise ValueError( "Applied object requires an endpoint-local key" ) + + retirements = tuple(self.retirements) + + if any( + not isinstance(retirement, RetireRequest) + for retirement in retirements + ): + raise ValueError( + "ApplyResult retirements must be RetireRequest instances" + ) + + seen: set[tuple[str, str]] = set() + + for retirement in retirements: + retired = retirement.object + identity = ( + retired.object_type, + retired.local_key, + ) + + if identity in seen: + raise ValueError( + "ApplyResult retirements cannot contain duplicates" + ) + + seen.add(identity) + + if ( + retired.object_type == self.object.object_type + and retired.local_key == self.object.local_key + ): + raise ValueError( + "ApplyResult cannot retire the newly applied object" + ) + + object.__setattr__( + self, + "retirements", + retirements, + ) + if self.detail is not None: detail = self.detail.strip() object.__setattr__( diff --git a/dns_sync/executor.py b/dns_sync/executor.py index 35c50eb..7a57826 100644 --- a/dns_sync/executor.py +++ b/dns_sync/executor.py @@ -13,6 +13,7 @@ from .adapters import ( ObjectReference, PlanAction, PlanOperation, + RetireRequest, ) from .dependencies import order_plan_actions from .persistence import SQLiteStateStore @@ -26,6 +27,8 @@ class ExecutionAdapter(Protocol): def verify(self, object: NormalizedObject) -> bool: ... + def retire(self, request: RetireRequest) -> None: ... + @dataclass(frozen=True) class ExecutedAction: @@ -271,6 +274,30 @@ def _validated_apply_result( "desired references" ) + if ( + action.operation != PlanOperation.REPLACE + and result.retirements + ): + raise RuntimeError( + "Only REPLACE actions may request deferred retirement" + ) + + for retirement in result.retirements: + retired = retirement.object + + if retired.local_key is None: + raise RuntimeError( + "Deferred retirement requires endpoint-local identity" + ) + + if ( + retired.object_type == applied.object_type + and retired.local_key == applied.local_key + ): + raise RuntimeError( + "Deferred retirement cannot target applied object" + ) + return result @@ -330,6 +357,7 @@ def _preflight( if action.operation not in { PlanOperation.CREATE, PlanOperation.UPDATE, + PlanOperation.REPLACE, }: raise RuntimeError( "Unsupported executable plan operation: " @@ -376,6 +404,10 @@ def execute_endpoint_plans( plans, ) applied: list[ExecutedAction] = [] + pending_retirements: list[ + tuple[str, RetireRequest] + ] = [] + local_references: dict[ str, _LocalReferenceIndex, @@ -451,6 +483,14 @@ def execute_endpoint_plans( authoritative=True, ) + for retirement in result.retirements: + pending_retirements.append( + ( + action.endpoint_id, + retirement, + ) + ) + applied.append( ExecutedAction( endpoint_id=action.endpoint_id, @@ -460,6 +500,19 @@ def execute_endpoint_plans( ) ) + for endpoint_id, retirement in pending_retirements: + adapter = adapters_by_endpoint[endpoint_id] + + retire = getattr(adapter, "retire", None) + + if retire is None: + raise RuntimeError( + f"{endpoint_id} adapter does not support " + "deferred retirement" + ) + + retire(retirement) + return ExecutionResult( applied=tuple(applied), skipped=tuple(skipped), diff --git a/dns_sync/plugins/adapters/npm.py b/dns_sync/plugins/adapters/npm.py index c080612..c70818d 100644 --- a/dns_sync/plugins/adapters/npm.py +++ b/dns_sync/plugins/adapters/npm.py @@ -11,6 +11,7 @@ from ...adapters import ( ObjectReference, PlanAction, PlanOperation, + RetireRequest, ) from ...certificates.material import CertificateMaterial from ...npm_client import NpmClient @@ -602,6 +603,12 @@ class Plugin: ): return self._create_certificate(desired) + if ( + desired.object_type == CERTIFICATES + and action.operation == PlanOperation.REPLACE + ): + return self._replace_certificate(action) + if ( desired.object_type == PROXY_HOSTS and action.operation == PlanOperation.CREATE @@ -619,6 +626,65 @@ class Plugin: f"{action.operation} for {desired.object_type}" ) + def _replace_certificate( + self, + action: PlanAction, + ) -> ApplyResult: + desired = action.desired + current = action.current + + if desired is None or desired.object_type != CERTIFICATES: + raise RuntimeError( + "NPM certificate replace requires desired certificate" + ) + + if current is None or current.object_type != CERTIFICATES: + raise RuntimeError( + "NPM certificate replace requires current certificate" + ) + + if current.local_key is None: + raise RuntimeError( + "NPM certificate replace requires current local id" + ) + + if current.logical_key != desired.logical_key: + raise RuntimeError( + "NPM certificate replace identity mismatch" + ) + + result = self._create_certificate(desired) + + return ApplyResult( + object=result.object, + detail="replaced certificate", + retirements=( + RetireRequest( + object=current, + ), + ), + ) + + def retire( + self, + request: RetireRequest, + ) -> None: + retired = request.object + + if retired.object_type != CERTIFICATES: + raise RuntimeError( + "NPM deferred retirement only supports certificates" + ) + + if retired.local_key is None: + raise RuntimeError( + "NPM certificate retirement requires local id" + ) + + self.client.delete_certificate( + retired.local_key + ) + def _create_certificate( self, desired: NormalizedObject, diff --git a/dns_sync/reconciliation.py b/dns_sync/reconciliation.py index 2cb8cb5..c8f1831 100644 --- a/dns_sync/reconciliation.py +++ b/dns_sync/reconciliation.py @@ -38,6 +38,7 @@ class PlanningAdapter(Protocol): class ReconciliationPolicy(Protocol): object_type: str + mutation_operation: PlanOperation def reconcile( self, @@ -181,6 +182,7 @@ def group_observations( class DnsUnionPolicy: object_type = DNS_RECORDS + mutation_operation = PlanOperation.UPDATE def reconcile( self, @@ -229,6 +231,7 @@ class DnsUnionPolicy: class CertificateNewestPolicy: object_type = CERTIFICATES + mutation_operation = PlanOperation.REPLACE _PORTABLE_FIELDS = ( "domain_names", @@ -247,20 +250,38 @@ class CertificateNewestPolicy: ) @staticmethod - def _expiry_key( - object: NormalizedObject, + def _date_key( + value: object, + field: str, ) -> tuple[int, str]: - expires_on = object.payload.get("expires_on") - - if expires_on is None: + if value is None: return (0, "") - if not isinstance(expires_on, str): + if not isinstance(value, str): raise ReconciliationConflict( - "Certificate expires_on must be normalized text" + f"Certificate {field} must be normalized text" ) - return (1, expires_on) + return (1, value) + + @classmethod + def _newest_key( + cls, + object: NormalizedObject, + ) -> tuple[ + tuple[int, str], + tuple[int, str], + ]: + return ( + cls._date_key( + object.payload.get("valid_from"), + "valid_from", + ), + cls._date_key( + object.payload.get("expires_on"), + "expires_on", + ), + ) def reconcile( self, @@ -273,14 +294,14 @@ class CertificateNewestPolicy: ) newest_key = max( - self._expiry_key(observation.object) + self._newest_key(observation.object) for observation in observations ) newest = [ observation for observation in observations - if self._expiry_key(observation.object) == newest_key + if self._newest_key(observation.object) == newest_key ] baseline = self._portable_state(newest[0].object) @@ -296,7 +317,7 @@ class CertificateNewestPolicy: raise ReconciliationConflict( f"Conflicting {CERTIFICATES} observations for " - f"{logical_key!r} at newest expiry from " + f"{logical_key!r} at newest issuance from " + ", ".join(sorted(conflicting_endpoints)) ) @@ -324,6 +345,7 @@ class CertificateNewestPolicy: class ProxyHostConsensusPolicy: object_type = PROXY_HOSTS + mutation_operation = PlanOperation.UPDATE def reconcile( self, @@ -540,7 +562,11 @@ def build_endpoint_plans( operation = ( PlanOperation.NOOP if equivalent - else PlanOperation.UPDATE + else ( + policy.mutation_operation + if policy is not None + else PlanOperation.UPDATE + ) ) actions.append( @@ -552,6 +578,49 @@ def build_endpoint_plans( ) ) + replaced = { + ( + action.desired.object_type, + action.desired.logical_key, + ) + for action in actions + if ( + action.operation == PlanOperation.REPLACE + and action.desired is not None + ) + } + + if replaced: + promoted: list[PlanAction] = [] + + for action in actions: + desired = action.desired + + if ( + action.operation == PlanOperation.NOOP + and desired is not None + and any( + ( + reference.object_type, + reference.logical_key, + ) + in replaced + for reference in desired.references + if reference.logical_key is not None + ) + ): + action = replace( + action, + operation=PlanOperation.UPDATE, + reason=( + "referenced object is being replaced" + ), + ) + + promoted.append(action) + + actions = promoted + plans.append( EndpointPlan( endpoint_id=endpoint_id, diff --git a/tests/test_executor.py b/tests/test_executor.py index ed9d210..9156c13 100644 --- a/tests/test_executor.py +++ b/tests/test_executor.py @@ -16,6 +16,7 @@ from dns_sync.adapters import ( Observation, PlanAction, PlanOperation, + RetireRequest, ) from dns_sync.executor import execute_endpoint_plans from dns_sync.persistence import SQLiteStateStore @@ -117,6 +118,9 @@ class FakeAdapter: events: list[tuple[str, str, str]] = field( default_factory=list ) + retired: list[RetireRequest] = field( + default_factory=list + ) def apply(self, plan_action): desired = plan_action.desired @@ -163,6 +167,16 @@ class FakeAdapter: raise self.verify_error return self.verify_result + def retire(self, request): + self.events.append( + ( + "retire", + request.object.object_type, + request.object.logical_key, + ) + ) + self.retired.append(request) + class ExecutionTests(unittest.TestCase): def test_successful_actions_are_applied_verified_and_persisted(self): @@ -216,6 +230,190 @@ class ExecutionTests(unittest.TestCase): ("apply", "verify", "apply", "verify"), ) + def test_replace_rebinds_dependent_before_retiring_old_object(self): + endpoint = Endpoint( + "npm", + "192.168.1.4:81", + ) + + class ReplacementAdapter(FakeAdapter): + def apply(self, plan_action): + result = super().apply(plan_action) + + if ( + plan_action.operation + == PlanOperation.REPLACE + ): + return ApplyResult( + object=result.object, + detail="replacement", + retirements=( + RetireRequest( + object=plan_action.current, + ), + ), + ) + + return result + + adapter = ReplacementAdapter( + endpoint, + capabilities( + CERTIFICATES, + PROXY_HOSTS, + ), + apply_local_keys={ + ( + CERTIFICATES, + "app.example.com", + ): "84", + ( + 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", + }, + ) + + desired_certificate = NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + 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_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.REPLACE, + desired=desired_certificate, + current=old_certificate, + ), + PlanAction( + endpoint_id=endpoint.identity, + operation=PlanOperation.UPDATE, + desired=desired_proxy, + current=old_proxy, + ), + ), + ) + + with SQLiteStateStore(":memory:") as store: + store.replace_observations( + endpoint.identity, + CERTIFICATES, + [ + Observation( + endpoint.identity, + old_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", + "apply", + "verify", + "retire", + ), + ) + + proxy_apply = adapter.applied_actions[1] + + self.assertEqual( + proxy_apply.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 5c3b830..a57b7ab 100644 --- a/tests/test_planning.py +++ b/tests/test_planning.py @@ -500,6 +500,121 @@ class EndpointPlanningTests(unittest.TestCase): desired.references, ) + def test_older_certificate_is_replace_and_dependent_proxy_is_update(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"], + "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", + ), + ), + ) + + current_certificate = NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + local_key="67", + payload={ + "domain_names": ["app.example.com"], + "valid_from": "2026-08-20T00:00:00+00:00", + "expires_on": "2026-11-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, + current_certificate, + ), + observed( + endpoint.identity, + current_proxy, + ), + ], + ) + + self.assertEqual(len(plans), 1) + + actions = plans[0].actions + + self.assertEqual( + [ + ( + action.operation, + action.desired.object_type, + ) + for action in actions + ], + [ + ( + PlanOperation.REPLACE, + CERTIFICATES, + ), + ( + PlanOperation.UPDATE, + PROXY_HOSTS, + ), + ], + ) + + self.assertEqual( + actions[1].reason, + "referenced object is being replaced", + ) + + 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 ed9be0f..31a5367 100644 --- a/tests/test_reconciliation.py +++ b/tests/test_reconciliation.py @@ -326,7 +326,7 @@ class DnsUnionPolicyTests(unittest.TestCase): ) self.assertIsNone(desired[0].local_key) - def test_certificate_same_expiry_conflict_fails_closed(self): + def test_certificate_policy_prefers_newer_valid_from_over_expiry(self): observations = [ Observation( endpoint_id="npm://node-one:81", @@ -337,7 +337,7 @@ class DnsUnionPolicyTests(unittest.TestCase): payload={ "domain_names": ["app.example.com"], "valid_from": "2026-08-01T00:00:00+00:00", - "expires_on": "2026-12-01T00:00:00+00:00", + "expires_on": "2027-01-01T00:00:00+00:00", }, ), observed_at=NOW, @@ -350,19 +350,25 @@ class DnsUnionPolicyTests(unittest.TestCase): 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", + "valid_from": "2026-09-15T00:00:00+00:00", + "expires_on": "2026-12-15T00:00:00+00:00", }, ), observed_at=NOW, ), ] - with self.assertRaisesRegex( - ReconciliationConflict, - "newest expiry", - ): - reconcile_observations(observations) + desired = reconcile_observations(observations) + + self.assertEqual(len(desired), 1) + self.assertEqual( + desired[0].payload["valid_from"], + "2026-09-15T00:00:00+00:00", + ) + self.assertEqual( + desired[0].payload["expires_on"], + "2026-12-15T00:00:00+00:00", + ) def test_proxy_consensus_ignores_different_endpoint_local_certificate_ids(self): observations = []