501 lines
19 KiB
Python
501 lines
19 KiB
Python
"""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",
|
|
]
|