diff --git a/dns_sync/persistence.py b/dns_sync/persistence.py index d6eb5a6..507405a 100644 --- a/dns_sync/persistence.py +++ b/dns_sync/persistence.py @@ -33,6 +33,29 @@ class ObservationStatus: return self.last_error is None +@dataclass(frozen=True) +class DesiredState: + object: NormalizedObject + selected_at: datetime + + +@dataclass(frozen=True) +class AppliedState: + endpoint_id: str + object: NormalizedObject + applied_at: datetime + verification_attempted_at: datetime | None + verified_at: datetime | None + verification_error: str | None + + @property + def is_verified(self) -> bool: + return ( + self.verified_at is not None + and self.verification_error is None + ) + + def _utc(value: datetime, label: str) -> datetime: if value.tzinfo is None: raise RuntimeError(f"{label} must be timezone-aware") @@ -153,6 +176,63 @@ class SQLiteStateStore: """ ) + versions = self.connection.execute( + "SELECT version FROM schema_meta" + ).fetchall() + if len(versions) != 1: + raise RuntimeError( + "SQLite state schema has invalid version metadata" + ) + + version = versions[0]["version"] + if version not in (1, 2): + raise RuntimeError( + "Unsupported SQLite state schema version: " + f"{version!r}" + ) + + if version == 1: + self.connection.executescript( + """ + CREATE TABLE desired_state ( + object_type TEXT NOT NULL, + logical_key TEXT NOT NULL, + payload_json TEXT NOT NULL, + selected_at TEXT NOT NULL, + PRIMARY KEY ( + object_type, + logical_key + ) + ); + + CREATE TABLE applied_state ( + endpoint_id TEXT NOT NULL, + object_type TEXT NOT NULL, + logical_key TEXT NOT NULL, + local_key TEXT, + payload_json TEXT NOT NULL, + applied_at TEXT NOT NULL, + verification_attempted_at TEXT, + verified_at TEXT, + verification_error TEXT, + PRIMARY KEY ( + endpoint_id, + object_type, + logical_key + ) + ); + + CREATE INDEX applied_state_by_object + ON applied_state ( + object_type, + logical_key + ); + + UPDATE schema_meta + SET version = 2; + """ + ) + def close(self) -> None: self.connection.close() @@ -328,6 +408,291 @@ class SQLiteStateStore: for row in rows ] + def replace_desired_state( + self, + object_type: str, + objects: Iterable[NormalizedObject], + selected_at: datetime, + ) -> None: + rows = list(objects) + selected_timestamp = _timestamp( + selected_at, + "Desired-state selection timestamp", + ) + prepared: list[tuple[str, str, str, str]] = [] + logical_keys: set[str] = set() + + for normalized in rows: + if normalized.object_type != object_type: + raise RuntimeError( + "Desired object type does not match snapshot type" + ) + if normalized.local_key is not None: + raise RuntimeError( + "Canonical desired objects cannot contain " + "an endpoint-local key" + ) + if normalized.logical_key in logical_keys: + raise RuntimeError( + "Duplicate desired logical key: " + f"{normalized.logical_key}" + ) + + logical_keys.add(normalized.logical_key) + prepared.append( + ( + object_type, + normalized.logical_key, + _serialize_payload(normalized.payload), + selected_timestamp, + ) + ) + + with self.connection: + self.connection.execute( + """ + DELETE FROM desired_state + WHERE object_type = ? + """, + (object_type,), + ) + self.connection.executemany( + """ + INSERT INTO desired_state ( + object_type, + logical_key, + payload_json, + selected_at + ) VALUES (?, ?, ?, ?) + """, + prepared, + ) + + def load_desired_state( + self, + object_type: str | None = None, + ) -> list[DesiredState]: + sql = """ + SELECT + object_type, + logical_key, + payload_json, + selected_at + FROM desired_state + """ + parameters: tuple[str, ...] = () + if object_type is not None: + sql += " WHERE object_type = ?" + parameters = (object_type,) + sql += " ORDER BY object_type, logical_key" + + rows = self.connection.execute(sql, parameters).fetchall() + return [ + DesiredState( + object=NormalizedObject( + object_type=row["object_type"], + logical_key=row["logical_key"], + payload=_deserialize_payload(row["payload_json"]), + ), + selected_at=_parse_timestamp(row["selected_at"]), + ) + for row in rows + ] + + def record_applied( + self, + endpoint_id: str, + object: NormalizedObject, + applied_at: datetime, + ) -> None: + if not endpoint_id: + raise RuntimeError("Applied-state endpoint_id cannot be empty") + + payload_json = _serialize_payload(object.payload) + applied_timestamp = _timestamp( + applied_at, + "Application timestamp", + ) + + with self.connection: + self.connection.execute( + """ + INSERT INTO applied_state ( + endpoint_id, + object_type, + logical_key, + local_key, + payload_json, + applied_at, + verification_attempted_at, + verified_at, + verification_error + ) VALUES (?, ?, ?, ?, ?, ?, NULL, NULL, NULL) + ON CONFLICT( + endpoint_id, + object_type, + logical_key + ) DO UPDATE SET + local_key = excluded.local_key, + payload_json = excluded.payload_json, + applied_at = excluded.applied_at, + verification_attempted_at = NULL, + verified_at = NULL, + verification_error = NULL + """, + ( + endpoint_id, + object.object_type, + object.logical_key, + object.local_key, + payload_json, + applied_timestamp, + ), + ) + + def _record_verification( + self, + endpoint_id: str, + object: NormalizedObject, + attempted_at: datetime, + *, + succeeded: bool, + error: str | None, + ) -> None: + attempted_timestamp = _timestamp( + attempted_at, + "Verification timestamp", + ) + payload_json = _serialize_payload(object.payload) + + if succeeded: + verified_at = attempted_timestamp + verification_error = None + else: + verified_at = None + verification_error = str( + error or "verification failed" + ) + + with self.connection: + cursor = self.connection.execute( + """ + UPDATE applied_state + SET + verification_attempted_at = ?, + verified_at = COALESCE(?, verified_at), + verification_error = ? + WHERE + endpoint_id = ? + AND object_type = ? + AND logical_key = ? + AND payload_json = ? + """, + ( + attempted_timestamp, + verified_at, + verification_error, + endpoint_id, + object.object_type, + object.logical_key, + payload_json, + ), + ) + if cursor.rowcount != 1: + raise RuntimeError( + "Verification does not match the currently " + "applied object" + ) + + def record_verification_success( + self, + endpoint_id: str, + object: NormalizedObject, + verified_at: datetime, + ) -> None: + self._record_verification( + endpoint_id, + object, + verified_at, + succeeded=True, + error=None, + ) + + def record_verification_failure( + self, + endpoint_id: str, + object: NormalizedObject, + error: str, + attempted_at: datetime, + ) -> None: + self._record_verification( + endpoint_id, + object, + attempted_at, + succeeded=False, + error=error, + ) + + def load_applied_state( + self, + endpoint_id: str | None = None, + object_type: str | None = None, + ) -> list[AppliedState]: + clauses: list[str] = [] + parameters: list[str] = [] + + if endpoint_id is not None: + clauses.append("endpoint_id = ?") + parameters.append(endpoint_id) + if object_type is not None: + clauses.append("object_type = ?") + parameters.append(object_type) + + sql = """ + SELECT + endpoint_id, + object_type, + logical_key, + local_key, + payload_json, + applied_at, + verification_attempted_at, + verified_at, + verification_error + FROM applied_state + """ + if clauses: + sql += " WHERE " + " AND ".join(clauses) + sql += """ + ORDER BY + endpoint_id, + object_type, + logical_key + """ + + rows = self.connection.execute( + sql, + tuple(parameters), + ).fetchall() + return [ + AppliedState( + endpoint_id=row["endpoint_id"], + object=NormalizedObject( + object_type=row["object_type"], + logical_key=row["logical_key"], + local_key=row["local_key"], + payload=_deserialize_payload(row["payload_json"]), + ), + applied_at=_parse_timestamp(row["applied_at"]), + verification_attempted_at=_parse_timestamp( + row["verification_attempted_at"] + ), + verified_at=_parse_timestamp(row["verified_at"]), + verification_error=row["verification_error"], + ) + for row in rows + ] + def observation_status( self, endpoint_id: str, diff --git a/tests/test_reconciliation_persistence.py b/tests/test_reconciliation_persistence.py new file mode 100644 index 0000000..891f44d --- /dev/null +++ b/tests/test_reconciliation_persistence.py @@ -0,0 +1,240 @@ +import sqlite3 +import tempfile +import unittest +from datetime import datetime, timezone +from pathlib import Path + +from dns_sync.adapters import CERTIFICATES, NormalizedObject +from dns_sync.persistence import SQLiteStateStore + + +ENDPOINT = "npm://192.168.1.3:81" +T1 = datetime(2026, 9, 19, 21, 0, tzinfo=timezone.utc) +T2 = datetime(2026, 9, 19, 21, 5, tzinfo=timezone.utc) +T3 = datetime(2026, 9, 19, 21, 10, tzinfo=timezone.utc) + + +def certificate( + expires_on: str, + local_key: str | None = None, +) -> NormalizedObject: + return NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + local_key=local_key, + payload={ + "domain_names": ["app.example.com"], + "expires_on": expires_on, + }, + ) + + +class DesiredStatePersistenceTests(unittest.TestCase): + def test_snapshot_round_trips_and_can_be_cleared(self): + desired = certificate("2026-12-15T00:00:00+00:00") + + with SQLiteStateStore(":memory:") as store: + store.replace_desired_state( + CERTIFICATES, + [desired], + T1, + ) + loaded = store.load_desired_state(CERTIFICATES) + + store.replace_desired_state( + CERTIFICATES, + [], + T2, + ) + cleared = store.load_desired_state(CERTIFICATES) + + self.assertEqual(len(loaded), 1) + self.assertEqual(loaded[0].object, desired) + self.assertEqual(loaded[0].selected_at, T1) + self.assertEqual(cleared, []) + + def test_endpoint_local_identity_is_rejected(self): + with SQLiteStateStore(":memory:") as store: + with self.assertRaisesRegex( + RuntimeError, + "endpoint-local key", + ): + store.replace_desired_state( + CERTIFICATES, + [certificate("new", local_key="67")], + T1, + ) + + def test_sensitive_payload_is_rejected_before_snapshot_changes(self): + safe = certificate("safe") + unsafe = NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + payload={ + "domain_names": ["app.example.com"], + "nested": { + "private_key": "must-never-persist", + }, + }, + ) + + with SQLiteStateStore(":memory:") as store: + store.replace_desired_state( + CERTIFICATES, + [safe], + T1, + ) + + with self.assertRaisesRegex( + RuntimeError, + "sensitive field", + ): + store.replace_desired_state( + CERTIFICATES, + [unsafe], + T2, + ) + + loaded = store.load_desired_state(CERTIFICATES) + + self.assertEqual(len(loaded), 1) + self.assertEqual(loaded[0].object, safe) + self.assertEqual(loaded[0].selected_at, T1) + + +class AppliedStatePersistenceTests(unittest.TestCase): + def test_apply_requires_successful_verification(self): + first = certificate("first", local_key="91") + second = certificate("second", local_key="91") + + with SQLiteStateStore(":memory:") as store: + store.record_applied(ENDPOINT, first, T1) + before = store.load_applied_state( + ENDPOINT, + CERTIFICATES, + )[0] + + store.record_verification_success( + ENDPOINT, + first, + T2, + ) + verified = store.load_applied_state( + ENDPOINT, + CERTIFICATES, + )[0] + + store.record_applied(ENDPOINT, second, T3) + reapplied = store.load_applied_state( + ENDPOINT, + CERTIFICATES, + )[0] + + self.assertFalse(before.is_verified) + self.assertIsNone(before.verified_at) + + self.assertTrue(verified.is_verified) + self.assertEqual(verified.verified_at, T2) + self.assertEqual(verified.verification_attempted_at, T2) + + self.assertEqual( + reapplied.object.payload["expires_on"], + "second", + ) + self.assertEqual(reapplied.applied_at, T3) + self.assertFalse(reapplied.is_verified) + self.assertIsNone(reapplied.verification_attempted_at) + self.assertIsNone(reapplied.verified_at) + + def test_stale_payload_cannot_verify_current_applied_state(self): + old = certificate("old", local_key="91") + current = certificate("current", local_key="91") + + with SQLiteStateStore(":memory:") as store: + store.record_applied(ENDPOINT, current, T1) + + with self.assertRaisesRegex( + RuntimeError, + "does not match", + ): + store.record_verification_success( + ENDPOINT, + old, + T2, + ) + + state = store.load_applied_state( + ENDPOINT, + CERTIFICATES, + )[0] + + self.assertFalse(state.is_verified) + self.assertIsNone(state.verification_attempted_at) + self.assertIsNone(state.verified_at) + + def test_failed_verification_is_not_reported_as_verified(self): + applied = certificate("current", local_key="91") + + with SQLiteStateStore(":memory:") as store: + store.record_applied(ENDPOINT, applied, T1) + store.record_verification_success( + ENDPOINT, + applied, + T2, + ) + store.record_verification_failure( + ENDPOINT, + applied, + "target no longer matches", + T3, + ) + state = store.load_applied_state( + ENDPOINT, + CERTIFICATES, + )[0] + + self.assertFalse(state.is_verified) + self.assertEqual(state.verified_at, T2) + self.assertEqual(state.verification_attempted_at, T3) + self.assertEqual( + state.verification_error, + "target no longer matches", + ) + + +class SchemaMigrationTests(unittest.TestCase): + def test_version_one_database_is_migrated_to_version_two(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "state.sqlite3" + connection = sqlite3.connect(path) + connection.execute( + "CREATE TABLE schema_meta (version INTEGER NOT NULL)" + ) + connection.execute( + "INSERT INTO schema_meta(version) VALUES (1)" + ) + connection.commit() + connection.close() + + with SQLiteStateStore(path) as store: + version = store.connection.execute( + "SELECT version FROM schema_meta" + ).fetchone()[0] + tables = { + row[0] + for row in store.connection.execute( + """ + SELECT name + FROM sqlite_master + WHERE type = 'table' + """ + ) + } + + self.assertEqual(version, 2) + self.assertIn("desired_state", tables) + self.assertIn("applied_state", tables) + + +if __name__ == "__main__": + unittest.main()