"""Signed, provider-neutral backup and isolated-restore evidence.""" from __future__ import annotations from datetime import UTC, datetime from pathlib import Path import re from typing import Any, Mapping from urllib.parse import urlsplit from .distribution import ( DistributionError, load_bounded_json, verify_signed_document, ) MAX_BACKUP_EVIDENCE_BYTES = 1024 * 1024 MAX_BACKUP_KEYRING_BYTES = 1024 * 1024 DEFAULT_MAX_BACKUP_AGE_SECONDS = 24 * 60 * 60 MAX_COORDINATION_SKEW_SECONDS = 5 * 60 SHA256 = re.compile(r"^[0-9a-f]{64}$") TOKEN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$") IMAGE = re.compile(r"^[^@\s]+@sha256:[0-9a-f]{64}$") REFERENCE = re.compile(r"^[A-Za-z][A-Za-z0-9+.-]*:[^\s]{1,2040}$") def load_backup_evidence(path: Path) -> dict[str, Any]: return load_bounded_json(path, maximum_bytes=MAX_BACKUP_EVIDENCE_BYTES) def load_backup_keyring(path: Path) -> dict[str, Any]: return load_bounded_json(path, maximum_bytes=MAX_BACKUP_KEYRING_BYTES) def verify_backup_evidence( payload: Mapping[str, Any], keyring: Mapping[str, Any], *, installation_id: str, profile: str, release: Mapping[str, object], now: datetime | None = None, max_age_seconds: int = DEFAULT_MAX_BACKUP_AGE_SECONDS, openssl: str = "openssl", ) -> dict[str, object]: current = (now or datetime.now(UTC)).astimezone(UTC) values = _validate_payload(payload, now=current, max_age_seconds=max_age_seconds) if payload.get("installation_id") != installation_id: raise DistributionError("backup evidence belongs to another installation") subject = _object(payload.get("deployment_subject"), "deployment_subject") if subject.get("profile") != profile: raise DistributionError("backup evidence belongs to another deployment profile") evidence_release = _object(payload.get("release"), "release") for field in ( "channel", "version", "manifest_sha256", "composition_sha256", "api_image", "web_image", ): if evidence_release.get(field) != release.get(field): raise DistributionError( f"backup evidence does not match release field {field!r}" ) key_id = verify_signed_document( payload, keyring, purpose="govoplan-backup-evidence", label="backup evidence", now=current, openssl=openssl, ) recovery_point = _object(payload.get("recovery_point"), "recovery_point") restore = _object(payload.get("restore_drill"), "restore_drill") return { "evidence_id": payload["evidence_id"], "recovery_point_id": recovery_point["id"], "captured_at": recovery_point["captured_at"], "expires_at": payload["expires_at"], "restore_drill_id": restore["drill_id"], "restore_started_at": restore["started_at"], "restore_completed_at": restore["completed_at"], "measured_rpo_seconds": restore["measured_rpo_seconds"], "measured_rto_seconds": restore["measured_rto_seconds"], "signature_key_id": key_id, "component_count": values["component_count"], } def _validate_payload( payload: Mapping[str, Any], *, now: datetime, max_age_seconds: int, ) -> dict[str, int]: if max_age_seconds < 60 or max_age_seconds > 30 * 24 * 60 * 60: raise DistributionError("backup maximum age is out of bounds") _exact_keys( payload, { "schema_version", "evidence_id", "installation_id", "deployment_subject", "release", "recovery_point", "components", "restore_drill", "issued_at", "expires_at", "revoked", "signatures", }, "backup evidence", ) if payload.get("schema_version") != "1": raise DistributionError("unsupported backup evidence schema_version") _token(payload.get("evidence_id"), "evidence_id") _token(payload.get("installation_id"), "installation_id") issued = _timestamp(payload.get("issued_at"), "issued_at") expires = _timestamp(payload.get("expires_at"), "expires_at") if issued > now or expires <= issued or expires <= now: raise DistributionError("backup evidence is not currently valid") if payload.get("revoked") is not False: raise DistributionError("backup evidence is revoked") subject = _object(payload.get("deployment_subject"), "deployment_subject") _exact_keys(subject, {"profile", "topology", "subject_ref"}, "deployment_subject") if subject.get("profile") not in {"evaluation", "self-hosted"}: raise DistributionError("deployment_subject.profile is invalid") _token(subject.get("topology"), "deployment_subject.topology") _reference(subject.get("subject_ref"), "deployment_subject.subject_ref") release = _object(payload.get("release"), "release") _exact_keys( release, { "channel", "version", "manifest_sha256", "composition_sha256", "api_image", "web_image", }, "release", ) _token(release.get("channel"), "release.channel") _token(release.get("version"), "release.version") _sha256(release.get("manifest_sha256"), "release.manifest_sha256") _sha256(release.get("composition_sha256"), "release.composition_sha256") _image(release.get("api_image"), "release.api_image") _image(release.get("web_image"), "release.web_image") recovery = _object(payload.get("recovery_point"), "recovery_point") _exact_keys( recovery, {"id", "captured_at", "consistency", "rpo_seconds", "write_fence"}, "recovery_point", ) recovery_id = _token(recovery.get("id"), "recovery_point.id") captured = _timestamp(recovery.get("captured_at"), "recovery_point.captured_at") age = (now - captured).total_seconds() if age < 0 or age > max_age_seconds: raise DistributionError("backup recovery point is stale or in the future") if recovery.get("consistency") not in { "provider-atomic", "application-quiesced", "transaction-consistent", }: raise DistributionError("recovery_point.consistency is invalid") declared_rpo = _bounded_integer( recovery.get("rpo_seconds"), "recovery_point.rpo_seconds", maximum=30 * 24 * 60 * 60, ) fence = _object(recovery.get("write_fence"), "recovery_point.write_fence") _exact_keys( fence, {"mode", "token_sha256", "established_at"}, "recovery_point.write_fence", ) if fence.get("mode") not in { "provider-snapshot", "application-quiesce", "transaction-boundary", }: raise DistributionError("recovery_point.write_fence.mode is invalid") _sha256(fence.get("token_sha256"), "recovery_point.write_fence.token_sha256") established = _timestamp( fence.get("established_at"), "recovery_point.write_fence.established_at", ) if abs((captured - established).total_seconds()) > MAX_COORDINATION_SKEW_SECONDS: raise DistributionError( "backup write fence is not coordinated with recovery point" ) components = _object(payload.get("components"), "components") _exact_keys( components, {"database", "objects", "configuration", "key_custody"}, "components", ) captured_components = [ _database_component(components.get("database")), _objects_component(components.get("objects")), _configuration_component(components.get("configuration")), _key_custody_component(components.get("key_custody")), ] if any( abs((component_time - captured).total_seconds()) > MAX_COORDINATION_SKEW_SECONDS for component_time in captured_components ): raise DistributionError("backup components do not share one recovery point") restore = _object(payload.get("restore_drill"), "restore_drill") _exact_keys( restore, { "drill_id", "recovery_point_id", "started_at", "completed_at", "isolated_target_ref", "release_manifest_sha256", "migration_heads_sha256", "representative_object_manifest_sha256", "database_verified", "objects_verified", "configuration_verified", "key_custody_verified", "semantic_checks", "measured_rpo_seconds", "measured_rto_seconds", "evidence_ref", }, "restore_drill", ) _token(restore.get("drill_id"), "restore_drill.drill_id") if restore.get("recovery_point_id") != recovery_id: raise DistributionError("restore drill used another recovery point") started = _timestamp(restore.get("started_at"), "restore_drill.started_at") completed = _timestamp(restore.get("completed_at"), "restore_drill.completed_at") if started < captured or completed < started or completed > issued: raise DistributionError( "restore drill completion is outside evidence chronology" ) _reference(restore.get("isolated_target_ref"), "restore_drill.isolated_target_ref") _reference(restore.get("evidence_ref"), "restore_drill.evidence_ref") for field in ( "release_manifest_sha256", "migration_heads_sha256", "representative_object_manifest_sha256", ): _sha256(restore.get(field), f"restore_drill.{field}") if restore.get("release_manifest_sha256") != release.get("manifest_sha256"): raise DistributionError("restore drill used another immutable release") for field in ( "database_verified", "objects_verified", "configuration_verified", "key_custody_verified", ): if restore.get(field) is not True: raise DistributionError(f"restore_drill.{field} must be true") semantic = restore.get("semantic_checks") if not isinstance(semantic, list) or not semantic or len(semantic) > 128: raise DistributionError("restore_drill.semantic_checks must not be empty") seen_checks: set[str] = set() for index, raw in enumerate(semantic): check = _object(raw, f"restore_drill.semantic_checks[{index}]") _exact_keys( check, {"id", "status", "evidence_ref"}, f"restore_drill.semantic_checks[{index}]", ) check_id = _token(check.get("id"), f"semantic_checks[{index}].id") if check_id in seen_checks or check.get("status") != "passed": raise DistributionError("restore drill semantic checks are invalid") seen_checks.add(check_id) _reference(check.get("evidence_ref"), f"semantic_checks[{index}].evidence_ref") measured_rpo = _bounded_integer( restore.get("measured_rpo_seconds"), "restore_drill.measured_rpo_seconds", maximum=30 * 24 * 60 * 60, ) measured_rto = _bounded_integer( restore.get("measured_rto_seconds"), "restore_drill.measured_rto_seconds", maximum=30 * 24 * 60 * 60, ) if abs((completed - started).total_seconds() - measured_rto) > 5: raise DistributionError("restore drill RTO does not match its timestamps") if measured_rpo > declared_rpo: raise DistributionError( "restore drill exceeds the declared recovery point objective" ) _validate_signatures(payload.get("signatures")) return {"component_count": len(captured_components)} def _database_component(raw: object) -> datetime: value = _object(raw, "components.database") _exact_keys( value, { "provider", "artifact_ref", "artifact_sha256", "snapshot_id", "lsn", "protected", "encryption_key_ref", "captured_at", }, "components.database", ) _common_artifact(value, "components.database") _token(value.get("snapshot_id"), "components.database.snapshot_id") _bounded_text(value.get("lsn"), "components.database.lsn", maximum=256) return _timestamp(value.get("captured_at"), "components.database.captured_at") def _objects_component(raw: object) -> datetime: value = _object(raw, "components.objects") _exact_keys( value, { "provider", "artifact_ref", "manifest_sha256", "version_id", "object_count", "total_bytes", "protected", "encryption_key_ref", "captured_at", }, "components.objects", ) _token(value.get("provider"), "components.objects.provider") _reference(value.get("artifact_ref"), "components.objects.artifact_ref") _sha256(value.get("manifest_sha256"), "components.objects.manifest_sha256") _token(value.get("version_id"), "components.objects.version_id") _bounded_integer(value.get("object_count"), "components.objects.object_count") _bounded_integer(value.get("total_bytes"), "components.objects.total_bytes") _protected_key_reference(value, "components.objects") return _timestamp(value.get("captured_at"), "components.objects.captured_at") def _configuration_component(raw: object) -> datetime: value = _object(raw, "components.configuration") _exact_keys( value, { "artifact_ref", "sha256", "protected", "encryption_key_ref", "captured_at", }, "components.configuration", ) _reference(value.get("artifact_ref"), "components.configuration.artifact_ref") _sha256(value.get("sha256"), "components.configuration.sha256") _protected_key_reference(value, "components.configuration") return _timestamp(value.get("captured_at"), "components.configuration.captured_at") def _key_custody_component(raw: object) -> datetime: value = _object(raw, "components.key_custody") _exact_keys( value, {"provider", "keyset_ref", "keyset_version", "recoverable", "captured_at"}, "components.key_custody", ) _token(value.get("provider"), "components.key_custody.provider") _reference(value.get("keyset_ref"), "components.key_custody.keyset_ref") _token(value.get("keyset_version"), "components.key_custody.keyset_version") if value.get("recoverable") is not True: raise DistributionError("components.key_custody.recoverable must be true") return _timestamp(value.get("captured_at"), "components.key_custody.captured_at") def _common_artifact(value: Mapping[str, Any], label: str) -> None: _token(value.get("provider"), f"{label}.provider") _reference(value.get("artifact_ref"), f"{label}.artifact_ref") _sha256(value.get("artifact_sha256"), f"{label}.artifact_sha256") _protected_key_reference(value, label) def _protected_key_reference(value: Mapping[str, Any], label: str) -> None: if value.get("protected") is not True: raise DistributionError(f"{label}.protected must be true") _reference(value.get("encryption_key_ref"), f"{label}.encryption_key_ref") def _validate_signatures(raw: object) -> None: if not isinstance(raw, list) or not raw or len(raw) > 16: raise DistributionError("backup evidence signatures must not be empty") seen: set[str] = set() for index, item in enumerate(raw): signature = _object(item, f"signatures[{index}]") _exact_keys(signature, {"key_id", "algorithm", "value"}, f"signatures[{index}]") key_id = _token(signature.get("key_id"), f"signatures[{index}].key_id") if key_id in seen or signature.get("algorithm") != "ed25519": raise DistributionError("backup evidence signatures are invalid") seen.add(key_id) encoded = signature.get("value") if not isinstance(encoded, str) or len(encoded) > 256: raise DistributionError("backup evidence signature value is invalid") def _reference(raw: object, label: str) -> str: value = _bounded_text(raw, label, maximum=2048) if REFERENCE.fullmatch(value) is None or "BEGIN " in value.upper(): raise DistributionError(f"{label} must be an opaque provider reference") parsed = urlsplit(value) if parsed.username or parsed.password or parsed.query or parsed.fragment: raise DistributionError(f"{label} must not contain credentials or query data") return value def _object(raw: object, label: str) -> dict[str, Any]: if not isinstance(raw, dict) or not all(isinstance(key, str) for key in raw): raise DistributionError(f"{label} must be an object") return raw def _exact_keys(value: Mapping[str, Any], keys: set[str], label: str) -> None: if set(value) != keys: missing = sorted(keys - set(value)) extra = sorted(set(value) - keys) detail = [] if missing: detail.append("missing " + ", ".join(missing)) if extra: detail.append("unknown " + ", ".join(extra)) raise DistributionError(f"{label} has invalid fields: {'; '.join(detail)}") def _timestamp(raw: object, label: str) -> datetime: value = _bounded_text(raw, label, maximum=64) try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError as exc: raise DistributionError(f"{label} must be an RFC3339 timestamp") from exc if parsed.tzinfo is None: raise DistributionError(f"{label} must include a timezone") return parsed.astimezone(UTC) def _token(raw: object, label: str) -> str: value = _bounded_text(raw, label, maximum=128) if TOKEN.fullmatch(value) is None: raise DistributionError(f"{label} is invalid") return value def _sha256(raw: object, label: str) -> str: value = _bounded_text(raw, label, maximum=64) if SHA256.fullmatch(value) is None: raise DistributionError(f"{label} must be a lowercase SHA-256 digest") return value def _image(raw: object, label: str) -> str: value = _bounded_text(raw, label, maximum=300) if IMAGE.fullmatch(value) is None: raise DistributionError(f"{label} must be an OCI image pinned by sha256") return value def _bounded_text(raw: object, label: str, *, maximum: int) -> str: if not isinstance(raw, str) or not raw or len(raw) > maximum or "\n" in raw: raise DistributionError(f"{label} is invalid") return raw def _bounded_integer( raw: object, label: str, *, maximum: int = 2**63 - 1, ) -> int: if isinstance(raw, bool) or not isinstance(raw, int) or raw < 0 or raw > maximum: raise DistributionError(f"{label} is out of bounds") return raw __all__ = [ "DEFAULT_MAX_BACKUP_AGE_SECONDS", "MAX_BACKUP_EVIDENCE_BYTES", "MAX_BACKUP_KEYRING_BYTES", "load_backup_evidence", "load_backup_keyring", "verify_backup_evidence", ]