Enforce signed backup evidence before migrations

2026-08-03 02:01:13 +02:00
parent 2c515f73c2
commit cbbe08d912
16 changed files with 2157 additions and 52 deletions
@@ -0,0 +1,124 @@
+#!/usr/bin/env python3
+"""Sign and validate provider-produced GovOPlaN backup evidence."""
+
+from __future__ import annotations
+
+import argparse
+import base64
+import hashlib
+from pathlib import Path
+import re
+import stat
+from typing import Any
+
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from govoplan_deploy.backup_evidence import (
+    MAX_BACKUP_EVIDENCE_BYTES,
+    load_backup_keyring,
+    verify_backup_evidence,
+)
+from govoplan_deploy.bundle import atomic_write
+from govoplan_deploy.distribution import (
+    canonical_json,
+    canonical_signed_payload,
+    load_bounded_json,
+)
+
+
+KEY_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(
+        description=(
+            "Sign a provider-produced backup/restore evidence document and "
+            "validate it against an independently managed public keyring."
+        )
+    )
+    parser.add_argument("--input", type=Path, required=True)
+    parser.add_argument("--output", type=Path, required=True)
+    parser.add_argument("--trusted-keyring", type=Path, required=True)
+    parser.add_argument(
+        "--signing-key",
+        action="append",
+        required=True,
+        metavar="KEY_ID=PRIVATE_PEM",
+        help="Ed25519 signer; may be repeated during key rotation.",
+    )
+    parser.add_argument(
+        "--replace-signatures",
+        action="store_true",
+        help="Replace existing signatures instead of rejecting the input.",
+    )
+    args = parser.parse_args()
+
+    source = args.input.expanduser().resolve()
+    payload = load_bounded_json(source, maximum_bytes=MAX_BACKUP_EVIDENCE_BYTES)
+    existing = payload.get("signatures")
+    if existing not in (None, []) and not args.replace_signatures:
+        raise SystemExit("input already contains signatures; use --replace-signatures")
+
+    signers = [_load_signer(value) for value in args.signing_key]
+    if len({key_id for key_id, _ in signers}) != len(signers):
+        raise SystemExit("duplicate signing key id")
+    payload["signatures"] = []
+    signed = canonical_signed_payload(payload)
+    payload["signatures"] = [
+        {
+            "key_id": key_id,
+            "algorithm": "ed25519",
+            "value": base64.b64encode(private_key.sign(signed)).decode("ascii"),
+        }
+        for key_id, private_key in signers
+    ]
+
+    keyring = load_backup_keyring(args.trusted_keyring.expanduser().resolve())
+    release = payload.get("release")
+    if not isinstance(release, dict):
+        raise SystemExit("input release must be an object")
+    verify_backup_evidence(
+        payload,
+        keyring,
+        installation_id=str(payload.get("installation_id") or ""),
+        profile=str(
+            _object(payload.get("deployment_subject"), "deployment_subject").get(
+                "profile"
+            )
+            or ""
+        ),
+        release=release,
+    )
+    encoded = canonical_json(payload)
+    output = args.output.expanduser().resolve()
+    atomic_write(output, encoded, mode=0o600)
+    print(f"Wrote {output}")
+    print(f"SHA256 {hashlib.sha256(encoded).hexdigest()}")
+    return 0
+
+
+def _load_signer(value: str) -> tuple[str, Ed25519PrivateKey]:
+    key_id, separator, raw_path = value.partition("=")
+    if not separator or KEY_ID.fullmatch(key_id) is None or not raw_path:
+        raise SystemExit("--signing-key must use KEY_ID=/path/to/private.pem")
+    path = Path(raw_path).expanduser().resolve()
+    mode = stat.S_IMODE(path.stat().st_mode)
+    if mode & 0o077:
+        raise SystemExit(
+            f"private signing key must not be group/world accessible: {path}"
+        )
+    private_key = serialization.load_pem_private_key(path.read_bytes(), password=None)
+    if not isinstance(private_key, Ed25519PrivateKey):
+        raise SystemExit(f"signing key is not Ed25519: {path}")
+    return key_id, private_key
+
+
+def _object(value: object, label: str) -> dict[str, Any]:
+    if not isinstance(value, dict):
+        raise SystemExit(f"input {label} must be an object")
+    return value
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())