From b0d9857c380e28278a06704c310cbc5d7e9ef95f Mon Sep 17 00:00:00 2001 From: Lyncoln Date: Sat, 19 Sep 2026 19:05:24 -0400 Subject: [PATCH] Add verified generic plan execution --- dns_sync/executor.py | 226 ++++++++++++++++++++++++++ tests/test_executor.py | 350 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 576 insertions(+) create mode 100644 dns_sync/executor.py create mode 100644 tests/test_executor.py diff --git a/dns_sync/executor.py b/dns_sync/executor.py new file mode 100644 index 0000000..3a7b2c8 --- /dev/null +++ b/dns_sync/executor.py @@ -0,0 +1,226 @@ +from collections.abc import Iterable +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Protocol + +from .adapters import ( + CERTIFICATES, + DNS_RECORDS, + PROXY_HOSTS, + Capability, + CapabilityOperation, + Endpoint, + EndpointPlan, + NormalizedObject, + PlanAction, + PlanOperation, +) +from .persistence import SQLiteStateStore + + +_OBJECT_TYPE_ORDER = { + CERTIFICATES: 0, + PROXY_HOSTS: 1, + DNS_RECORDS: 2, +} + + +class ExecutionAdapter(Protocol): + endpoint: Endpoint + capabilities: frozenset[Capability] + + def apply(self, action: PlanAction) -> str | None: ... + + def verify(self, object: NormalizedObject) -> bool: ... + + +@dataclass(frozen=True) +class ExecutedAction: + endpoint_id: str + operation: PlanOperation + object_type: str + logical_key: str + + +@dataclass(frozen=True) +class ExecutionResult: + applied: tuple[ExecutedAction, ...] + skipped: tuple[PlanAction, ...] + + +def _action_sort_key( + action: PlanAction, +) -> tuple[int, str, str]: + desired = action.desired + if desired is None: + return (999, action.endpoint_id, "") + + return ( + _OBJECT_TYPE_ORDER.get(desired.object_type, 100), + action.endpoint_id, + desired.logical_key, + ) + + +def _required_capability( + adapter: ExecutionAdapter, + object_type: str, + operation: CapabilityOperation, +) -> None: + capability = Capability(object_type, operation) + if capability not in adapter.capabilities: + raise RuntimeError( + f"{adapter.endpoint.identity} lacks " + f"{operation.value} capability for {object_type}" + ) + + +def _preflight( + adapters: Iterable[ExecutionAdapter], + plans: Iterable[EndpointPlan], +) -> tuple[ + dict[str, ExecutionAdapter], + list[PlanAction], + list[PlanAction], +]: + adapters_by_endpoint: dict[str, ExecutionAdapter] = {} + for adapter in adapters: + endpoint_id = adapter.endpoint.identity + if endpoint_id in adapters_by_endpoint: + raise RuntimeError( + f"Duplicate execution adapter: {endpoint_id}" + ) + adapters_by_endpoint[endpoint_id] = adapter + + executable: list[PlanAction] = [] + skipped: list[PlanAction] = [] + + for plan in plans: + adapter = adapters_by_endpoint.get(plan.endpoint_id) + if adapter is None: + raise RuntimeError( + f"No adapter for endpoint plan {plan.endpoint_id}" + ) + + for action in plan.actions: + if action.endpoint_id != plan.endpoint_id: + raise RuntimeError( + "Plan action endpoint does not match its endpoint plan" + ) + + if action.operation in { + PlanOperation.NOOP, + PlanOperation.SKIP, + }: + skipped.append(action) + continue + + if action.operation not in { + PlanOperation.CREATE, + PlanOperation.UPDATE, + }: + raise RuntimeError( + "Unsupported executable plan operation: " + f"{action.operation.value}" + ) + + if action.desired is None: + raise RuntimeError( + "Executable plan action requires desired state" + ) + + object_type = action.desired.object_type + _required_capability( + adapter, + object_type, + CapabilityOperation.APPLY, + ) + _required_capability( + adapter, + object_type, + CapabilityOperation.VERIFY, + ) + executable.append(action) + + return adapters_by_endpoint, executable, skipped + + +def execute_endpoint_plans( + adapters: Iterable[ExecutionAdapter], + plans: Iterable[EndpointPlan], + store: SQLiteStateStore, + *, + executed_at: datetime | None = None, +) -> ExecutionResult: + timestamp = executed_at or datetime.now(timezone.utc) + if timestamp.tzinfo is None: + raise RuntimeError( + "Execution timestamp must be timezone-aware" + ) + timestamp = timestamp.astimezone(timezone.utc) + + adapters_by_endpoint, executable, skipped = _preflight( + adapters, + plans, + ) + applied: list[ExecutedAction] = [] + + for action in sorted(executable, key=_action_sort_key): + adapter = adapters_by_endpoint[action.endpoint_id] + desired = action.desired + + adapter.apply(action) + store.record_applied( + action.endpoint_id, + desired, + timestamp, + ) + + try: + verified = adapter.verify(desired) + except Exception as exc: + error = str(exc) + store.record_verification_failure( + action.endpoint_id, + desired, + error, + timestamp, + ) + raise RuntimeError( + f"{action.endpoint_id} " + f"{desired.object_type} " + f"{desired.logical_key}: verification failed" + ) from exc + + if not verified: + error = "adapter returned false" + store.record_verification_failure( + action.endpoint_id, + desired, + error, + timestamp, + ) + raise RuntimeError( + f"{action.endpoint_id} " + f"{desired.object_type} " + f"{desired.logical_key}: verification failed" + ) + + store.record_verification_success( + action.endpoint_id, + desired, + timestamp, + ) + applied.append( + ExecutedAction( + endpoint_id=action.endpoint_id, + operation=action.operation, + object_type=desired.object_type, + logical_key=desired.logical_key, + ) + ) + + return ExecutionResult( + applied=tuple(applied), + skipped=tuple(skipped), + ) diff --git a/tests/test_executor.py b/tests/test_executor.py new file mode 100644 index 0000000..9900f8c --- /dev/null +++ b/tests/test_executor.py @@ -0,0 +1,350 @@ +import unittest +from dataclasses import dataclass, field +from datetime import datetime, timezone + +from dns_sync.adapters import ( + CERTIFICATES, + DNS_RECORDS, + PROXY_HOSTS, + Capability, + CapabilityOperation, + Endpoint, + EndpointPlan, + NormalizedObject, + PlanAction, + PlanOperation, +) +from dns_sync.executor import execute_endpoint_plans +from dns_sync.persistence import SQLiteStateStore + + +NOW = datetime(2026, 9, 19, 23, 15, tzinfo=timezone.utc) + + +def object_for( + object_type: str, + logical_key: str, +) -> NormalizedObject: + if object_type == DNS_RECORDS: + payload = { + "hostname": logical_key, + "ip": "192.168.1.101", + "record_type": "A", + "wildcard": False, + } + elif object_type == CERTIFICATES: + payload = { + "domain_names": [logical_key], + "expires_on": "2026-12-15T00:00:00+00:00", + } + else: + payload = { + "domain_names": [logical_key], + "forward_scheme": "http", + "forward_host": "192.168.1.50", + "forward_port": 8080, + "enabled": True, + } + + return NormalizedObject( + object_type=object_type, + logical_key=logical_key, + payload=payload, + ) + + +def action( + endpoint_id: str, + operation: PlanOperation, + object_type: str = DNS_RECORDS, + logical_key: str = "app.example.com", +) -> PlanAction: + return PlanAction( + endpoint_id=endpoint_id, + operation=operation, + desired=object_for(object_type, logical_key), + ) + + +def capabilities( + *object_types: str, + verify: bool = True, +) -> frozenset[Capability]: + result = { + Capability( + object_type, + CapabilityOperation.APPLY, + ) + for object_type in object_types + } + if verify: + result.update( + Capability( + object_type, + CapabilityOperation.VERIFY, + ) + for object_type in object_types + ) + return frozenset(result) + + +@dataclass +class FakeAdapter: + endpoint: Endpoint + capabilities: frozenset[Capability] + verify_result: bool = True + apply_error: Exception | None = None + verify_error: Exception | None = None + events: list[tuple[str, str, str]] = field( + default_factory=list + ) + + def apply(self, plan_action): + desired = plan_action.desired + self.events.append( + ( + "apply", + desired.object_type, + desired.logical_key, + ) + ) + if self.apply_error is not None: + raise self.apply_error + return "test detail" + + def verify(self, object): + self.events.append( + ( + "verify", + object.object_type, + object.logical_key, + ) + ) + if self.verify_error is not None: + raise self.verify_error + return self.verify_result + + +class ExecutionTests(unittest.TestCase): + def test_successful_actions_are_applied_verified_and_persisted(self): + endpoint = Endpoint("verizon", "192.168.1.1:443") + adapter = FakeAdapter( + endpoint, + capabilities(DNS_RECORDS), + ) + plan = EndpointPlan( + endpoint.identity, + ( + action( + endpoint.identity, + PlanOperation.CREATE, + logical_key="add.example.com", + ), + action( + endpoint.identity, + PlanOperation.NOOP, + logical_key="keep.example.com", + ), + action( + endpoint.identity, + PlanOperation.UPDATE, + logical_key="update.example.com", + ), + ), + ) + + with SQLiteStateStore(":memory:") as store: + result = execute_endpoint_plans( + [adapter], + [plan], + store, + executed_at=NOW, + ) + persisted = store.load_applied_state( + endpoint.identity, + DNS_RECORDS, + ) + + self.assertEqual(len(result.applied), 2) + self.assertEqual(len(result.skipped), 1) + self.assertTrue(all(row.is_verified for row in persisted)) + self.assertEqual( + {row.object.logical_key for row in persisted}, + {"add.example.com", "update.example.com"}, + ) + self.assertEqual( + tuple(event[0] for event in adapter.events), + ("apply", "verify", "apply", "verify"), + ) + + def test_failed_verification_is_persisted_and_raises(self): + endpoint = Endpoint("verizon", "192.168.1.1:443") + adapter = FakeAdapter( + endpoint, + capabilities(DNS_RECORDS), + verify_result=False, + ) + plan = EndpointPlan( + endpoint.identity, + ( + action( + endpoint.identity, + PlanOperation.CREATE, + ), + ), + ) + + with SQLiteStateStore(":memory:") as store: + with self.assertRaisesRegex( + RuntimeError, + "verification failed", + ): + execute_endpoint_plans( + [adapter], + [plan], + store, + executed_at=NOW, + ) + persisted = store.load_applied_state() + + self.assertEqual(len(persisted), 1) + self.assertFalse(persisted[0].is_verified) + self.assertEqual( + persisted[0].verification_error, + "adapter returned false", + ) + + def test_apply_failure_does_not_record_applied_state(self): + endpoint = Endpoint("verizon", "192.168.1.1:443") + adapter = FakeAdapter( + endpoint, + capabilities(DNS_RECORDS), + apply_error=RuntimeError("write failed"), + ) + plan = EndpointPlan( + endpoint.identity, + ( + action( + endpoint.identity, + PlanOperation.CREATE, + ), + ), + ) + + with SQLiteStateStore(":memory:") as store: + with self.assertRaisesRegex(RuntimeError, "write failed"): + execute_endpoint_plans( + [adapter], + [plan], + store, + executed_at=NOW, + ) + self.assertEqual(store.load_applied_state(), []) + + def test_preflight_requires_verify_before_any_mutation(self): + endpoint = Endpoint("verizon", "192.168.1.1:443") + adapter = FakeAdapter( + endpoint, + capabilities(DNS_RECORDS, verify=False), + ) + plan = EndpointPlan( + endpoint.identity, + ( + action( + endpoint.identity, + PlanOperation.CREATE, + ), + ), + ) + + with SQLiteStateStore(":memory:") as store: + with self.assertRaisesRegex( + RuntimeError, + "lacks verify capability", + ): + execute_endpoint_plans( + [adapter], + [plan], + store, + executed_at=NOW, + ) + + self.assertEqual(adapter.events, []) + + def test_certificates_execute_before_proxy_hosts(self): + events: list[tuple[str, str, str]] = [] + endpoint = Endpoint("npm", "192.168.1.4:81") + adapter = FakeAdapter( + endpoint, + capabilities(CERTIFICATES, PROXY_HOSTS), + events=events, + ) + plan = EndpointPlan( + endpoint.identity, + ( + action( + endpoint.identity, + PlanOperation.CREATE, + PROXY_HOSTS, + "app.example.com", + ), + action( + endpoint.identity, + PlanOperation.CREATE, + CERTIFICATES, + "app.example.com", + ), + ), + ) + + with SQLiteStateStore(":memory:") as store: + execute_endpoint_plans( + [adapter], + [plan], + store, + executed_at=NOW, + ) + + self.assertEqual( + tuple((event[0], event[1]) for event in events), + ( + ("apply", CERTIFICATES), + ("verify", CERTIFICATES), + ("apply", PROXY_HOSTS), + ("verify", PROXY_HOSTS), + ), + ) + + def test_invalid_plan_fails_preflight_without_mutation(self): + endpoint = Endpoint("verizon", "192.168.1.1:443") + adapter = FakeAdapter( + endpoint, + capabilities(DNS_RECORDS), + ) + invalid = EndpointPlan( + endpoint.identity, + ( + action( + "verizon://different:443", + PlanOperation.CREATE, + ), + ), + ) + + with SQLiteStateStore(":memory:") as store: + with self.assertRaisesRegex( + RuntimeError, + "does not match", + ): + execute_endpoint_plans( + [adapter], + [invalid], + store, + executed_at=NOW, + ) + + self.assertEqual(adapter.events, []) + + +if __name__ == "__main__": + unittest.main()