diff --git a/dns_sync/persistence.py b/dns_sync/persistence.py new file mode 100644 index 0000000..d6eb5a6 --- /dev/null +++ b/dns_sync/persistence.py @@ -0,0 +1,358 @@ +import json +import sqlite3 +from collections.abc import Iterable, Mapping +from dataclasses import dataclass +from datetime import datetime, timezone +from os import PathLike +from typing import Any + +from .adapters import NormalizedObject, Observation + + +_FORBIDDEN_PERSISTED_KEYS = frozenset( + { + "private_key", + "certificate_key", + "certificate", + "dns_provider_credentials", + "secret", + } +) + + +@dataclass(frozen=True) +class ObservationStatus: + endpoint_id: str + object_type: str + last_attempt_at: datetime + last_success_at: datetime | None + last_error: str | None + + @property + def is_live(self) -> bool: + return self.last_error is None + + +def _utc(value: datetime, label: str) -> datetime: + if value.tzinfo is None: + raise RuntimeError(f"{label} must be timezone-aware") + return value.astimezone(timezone.utc) + + +def _timestamp(value: datetime, label: str) -> str: + return _utc(value, label).isoformat() + + +def _parse_timestamp(value: str | None) -> datetime | None: + if value is None: + return None + parsed = datetime.fromisoformat(value) + if parsed.tzinfo is None: + raise RuntimeError( + f"Persisted timestamp is not timezone-aware: {value!r}" + ) + return parsed.astimezone(timezone.utc) + + +def _validate_safe_payload(value: Any, path: str = "payload") -> None: + if isinstance(value, Mapping): + for raw_key, child in value.items(): + key = str(raw_key).strip().lower().replace("-", "_") + if key in _FORBIDDEN_PERSISTED_KEYS: + raise RuntimeError( + f"Refusing to persist sensitive field {path}.{key}" + ) + _validate_safe_payload(child, f"{path}.{key}") + elif isinstance(value, (list, tuple)): + for index, child in enumerate(value): + _validate_safe_payload(child, f"{path}[{index}]") + + +def _serialize_payload(payload: Mapping[str, Any]) -> str: + _validate_safe_payload(payload) + try: + return json.dumps( + payload, + sort_keys=True, + separators=(",", ":"), + ) + except (TypeError, ValueError) as exc: + raise RuntimeError( + f"Object payload is not JSON serializable: {exc}" + ) from exc + + +def _deserialize_payload(value: str) -> dict[str, Any]: + try: + payload = json.loads(value) + except json.JSONDecodeError as exc: + raise RuntimeError("Persisted object payload is invalid JSON") from exc + + if not isinstance(payload, dict): + raise RuntimeError("Persisted object payload is not a mapping") + return payload + + +class SQLiteStateStore: + def __init__(self, path: str | PathLike[str]): + self.path = str(path) + self.connection = sqlite3.connect(self.path) + self.connection.row_factory = sqlite3.Row + self.connection.execute("PRAGMA foreign_keys = ON") + self.connection.execute("PRAGMA busy_timeout = 5000") + if self.path != ":memory:": + self.connection.execute("PRAGMA journal_mode = WAL") + self._initialize() + + def _initialize(self) -> None: + with self.connection: + self.connection.executescript( + """ + CREATE TABLE IF NOT EXISTS schema_meta ( + version INTEGER NOT NULL + ); + + INSERT INTO schema_meta(version) + SELECT 1 + WHERE NOT EXISTS ( + SELECT 1 FROM schema_meta + ); + + CREATE TABLE IF NOT EXISTS observations ( + endpoint_id TEXT NOT NULL, + object_type TEXT NOT NULL, + logical_key TEXT NOT NULL, + local_key TEXT NOT NULL, + payload_json TEXT NOT NULL, + observed_at TEXT NOT NULL, + PRIMARY KEY ( + endpoint_id, + object_type, + local_key + ) + ); + + CREATE INDEX IF NOT EXISTS + observations_by_logical_identity + ON observations ( + object_type, + logical_key + ); + + CREATE TABLE IF NOT EXISTS observation_status ( + endpoint_id TEXT NOT NULL, + object_type TEXT NOT NULL, + last_attempt_at TEXT NOT NULL, + last_success_at TEXT, + last_error TEXT, + PRIMARY KEY ( + endpoint_id, + object_type + ) + ); + """ + ) + + def close(self) -> None: + self.connection.close() + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, traceback): + self.close() + + def replace_observations( + self, + endpoint_id: str, + object_type: str, + observations: Iterable[Observation], + snapshot_at: datetime, + ) -> None: + rows = list(observations) + snapshot_timestamp = _timestamp( + snapshot_at, + "Observation snapshot timestamp", + ) + prepared: list[tuple[str, str, str, str, str, str]] = [] + local_keys: set[str] = set() + + for observation in rows: + normalized = observation.object + if observation.endpoint_id != endpoint_id: + raise RuntimeError( + "Observation endpoint does not match snapshot endpoint" + ) + if normalized.object_type != object_type: + raise RuntimeError( + "Observation object type does not match snapshot type" + ) + if normalized.local_key is None: + raise RuntimeError( + "Observed objects require an endpoint-local key" + ) + if normalized.local_key in local_keys: + raise RuntimeError( + f"Duplicate endpoint-local key: {normalized.local_key}" + ) + + local_keys.add(normalized.local_key) + prepared.append( + ( + endpoint_id, + object_type, + normalized.logical_key, + normalized.local_key, + _serialize_payload(normalized.payload), + _timestamp( + observation.observed_at, + "Observation timestamp", + ), + ) + ) + + with self.connection: + self.connection.execute( + """ + DELETE FROM observations + WHERE endpoint_id = ? AND object_type = ? + """, + (endpoint_id, object_type), + ) + self.connection.executemany( + """ + INSERT INTO observations ( + endpoint_id, + object_type, + logical_key, + local_key, + payload_json, + observed_at + ) VALUES (?, ?, ?, ?, ?, ?) + """, + prepared, + ) + self.connection.execute( + """ + INSERT INTO observation_status ( + endpoint_id, + object_type, + last_attempt_at, + last_success_at, + last_error + ) VALUES (?, ?, ?, ?, NULL) + ON CONFLICT(endpoint_id, object_type) DO UPDATE SET + last_attempt_at = excluded.last_attempt_at, + last_success_at = excluded.last_success_at, + last_error = NULL + """, + ( + endpoint_id, + object_type, + snapshot_timestamp, + snapshot_timestamp, + ), + ) + + def record_discovery_failure( + self, + endpoint_id: str, + object_type: str, + error: str, + attempted_at: datetime, + ) -> None: + attempted_timestamp = _timestamp( + attempted_at, + "Discovery attempt timestamp", + ) + with self.connection: + self.connection.execute( + """ + INSERT INTO observation_status ( + endpoint_id, + object_type, + last_attempt_at, + last_success_at, + last_error + ) VALUES (?, ?, ?, NULL, ?) + ON CONFLICT(endpoint_id, object_type) DO UPDATE SET + last_attempt_at = excluded.last_attempt_at, + last_error = excluded.last_error + """, + ( + endpoint_id, + object_type, + attempted_timestamp, + str(error), + ), + ) + + def load_observations( + self, + object_type: str | None = None, + ) -> list[Observation]: + sql = """ + SELECT + endpoint_id, + object_type, + logical_key, + local_key, + payload_json, + observed_at + FROM observations + """ + parameters: tuple[str, ...] = () + if object_type is not None: + sql += " WHERE object_type = ?" + parameters = (object_type,) + sql += """ + ORDER BY + endpoint_id, + object_type, + logical_key, + local_key + """ + + rows = self.connection.execute(sql, parameters).fetchall() + return [ + Observation( + 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"]), + ), + observed_at=_parse_timestamp(row["observed_at"]), + ) + for row in rows + ] + + def observation_status( + self, + endpoint_id: str, + object_type: str, + ) -> ObservationStatus | None: + row = self.connection.execute( + """ + SELECT + endpoint_id, + object_type, + last_attempt_at, + last_success_at, + last_error + FROM observation_status + WHERE endpoint_id = ? AND object_type = ? + """, + (endpoint_id, object_type), + ).fetchone() + if row is None: + return None + + return ObservationStatus( + endpoint_id=row["endpoint_id"], + object_type=row["object_type"], + last_attempt_at=_parse_timestamp(row["last_attempt_at"]), + last_success_at=_parse_timestamp(row["last_success_at"]), + last_error=row["last_error"], + ) diff --git a/tests/test_persistence.py b/tests/test_persistence.py new file mode 100644 index 0000000..c46c24b --- /dev/null +++ b/tests/test_persistence.py @@ -0,0 +1,257 @@ +import tempfile +import unittest +from datetime import datetime, timezone +from pathlib import Path + +from dns_sync.adapters import ( + CERTIFICATES, + NormalizedObject, + Observation, +) +from dns_sync.persistence import SQLiteStateStore + + +ENDPOINT = "npm://192.168.1.3:81" +T1 = datetime(2026, 9, 19, 20, 0, tzinfo=timezone.utc) +T2 = datetime(2026, 9, 19, 20, 5, tzinfo=timezone.utc) +T3 = datetime(2026, 9, 19, 20, 10, tzinfo=timezone.utc) + + +def certificate( + local_key: str, + expires_on: str, + observed_at: datetime = T1, + payload=None, +) -> Observation: + return Observation( + endpoint_id=ENDPOINT, + object=NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + local_key=local_key, + payload=payload + or { + "domain_names": ["app.example.com"], + "expires_on": expires_on, + }, + ), + observed_at=observed_at, + ) + + +class ObservationPersistenceTests(unittest.TestCase): + def test_snapshot_preserves_duplicate_logical_identity(self): + rows = [ + certificate("63", "2026-12-01T00:00:00+00:00"), + certificate("67", "2026-12-15T00:00:00+00:00"), + ] + + with SQLiteStateStore(":memory:") as store: + store.replace_observations( + ENDPOINT, + CERTIFICATES, + rows, + T1, + ) + loaded = store.load_observations(CERTIFICATES) + + self.assertEqual( + tuple(item.object.local_key for item in loaded), + ("63", "67"), + ) + self.assertEqual( + {item.object.logical_key for item in loaded}, + {"app.example.com"}, + ) + + def test_live_snapshot_atomically_replaces_previous_rows(self): + with SQLiteStateStore(":memory:") as store: + store.replace_observations( + ENDPOINT, + CERTIFICATES, + [ + certificate("63", "old"), + certificate("67", "newer"), + ], + T1, + ) + store.replace_observations( + ENDPOINT, + CERTIFICATES, + [ + certificate( + "67", + "newest", + observed_at=T2, + ) + ], + T2, + ) + loaded = store.load_observations(CERTIFICATES) + status = store.observation_status( + ENDPOINT, + CERTIFICATES, + ) + + self.assertEqual(len(loaded), 1) + self.assertEqual(loaded[0].object.local_key, "67") + self.assertEqual( + loaded[0].object.payload["expires_on"], + "newest", + ) + self.assertTrue(status.is_live) + self.assertEqual(status.last_success_at, T2) + self.assertIsNone(status.last_error) + + def test_empty_live_snapshot_removes_previous_rows(self): + with SQLiteStateStore(":memory:") as store: + store.replace_observations( + ENDPOINT, + CERTIFICATES, + [certificate("63", "old")], + T1, + ) + store.replace_observations( + ENDPOINT, + CERTIFICATES, + [], + T2, + ) + + self.assertEqual( + store.load_observations(CERTIFICATES), + [], + ) + self.assertTrue( + store.observation_status( + ENDPOINT, + CERTIFICATES, + ).is_live + ) + + def test_failed_discovery_preserves_cache_until_live_recovery(self): + with SQLiteStateStore(":memory:") as store: + store.replace_observations( + ENDPOINT, + CERTIFICATES, + [certificate("63", "cached")], + T1, + ) + store.record_discovery_failure( + ENDPOINT, + CERTIFICATES, + "connection refused", + T2, + ) + + cached = store.load_observations(CERTIFICATES) + failed = store.observation_status( + ENDPOINT, + CERTIFICATES, + ) + + store.replace_observations( + ENDPOINT, + CERTIFICATES, + [ + certificate( + "67", + "live", + observed_at=T3, + ) + ], + T3, + ) + recovered = store.observation_status( + ENDPOINT, + CERTIFICATES, + ) + + self.assertEqual(len(cached), 1) + self.assertEqual(cached[0].object.local_key, "63") + self.assertFalse(failed.is_live) + self.assertEqual(failed.last_success_at, T1) + self.assertEqual(failed.last_attempt_at, T2) + self.assertEqual(failed.last_error, "connection refused") + self.assertTrue(recovered.is_live) + self.assertEqual(recovered.last_success_at, T3) + + def test_sensitive_payload_is_rejected_before_cache_changes(self): + with SQLiteStateStore(":memory:") as store: + store.replace_observations( + ENDPOINT, + CERTIFICATES, + [certificate("63", "safe")], + T1, + ) + unsafe = certificate( + "67", + "unsafe", + observed_at=T2, + payload={ + "domain_names": ["app.example.com"], + "nested": { + "private_key": "must-never-persist", + }, + }, + ) + + with self.assertRaisesRegex( + RuntimeError, + "sensitive field", + ): + store.replace_observations( + ENDPOINT, + CERTIFICATES, + [unsafe], + T2, + ) + + cached = store.load_observations(CERTIFICATES) + + self.assertEqual(len(cached), 1) + self.assertEqual(cached[0].object.local_key, "63") + + def test_duplicate_local_keys_are_rejected(self): + duplicate = certificate("63", "duplicate") + with SQLiteStateStore(":memory:") as store: + with self.assertRaisesRegex( + RuntimeError, + "Duplicate endpoint-local key", + ): + store.replace_observations( + ENDPOINT, + CERTIFICATES, + [duplicate, duplicate], + T1, + ) + + def test_file_database_survives_reopen(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "state.sqlite3" + + with SQLiteStateStore(path) as store: + store.replace_observations( + ENDPOINT, + CERTIFICATES, + [certificate("63", "durable")], + T1, + ) + + with SQLiteStateStore(path) as reopened: + loaded = reopened.load_observations(CERTIFICATES) + status = reopened.observation_status( + ENDPOINT, + CERTIFICATES, + ) + + self.assertEqual(len(loaded), 1) + self.assertEqual( + loaded[0].object.payload["expires_on"], + "durable", + ) + self.assertEqual(status.last_success_at, T1) + + +if __name__ == "__main__": + unittest.main()