125 lines
4.1 KiB
Python
125 lines
4.1 KiB
Python
#!/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())
|