govoplan/tools/assessments/generate-authority-keypair.py

#!/usr/bin/env python3
"""Generate an independently held Ed25519 assessment-authority keypair."""

from __future__ import annotations

import argparse
import base64
from datetime import UTC, datetime, timedelta
import json
import os
from pathlib import Path
import re
import stat

from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey


KEY_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,159}$")
PROOF_SCOPES = (
    "target_environment",
    "external_providers",
    "accessibility",
    "privacy",
    "security",
    "operations",
    "recovery",
    "production_approval",
)
PURPOSES = {
    "proof": (
        "govoplan.capability-fit-proof-authorities",
        "./capability-fit-proof-authority-keyring.schema.json",
    ),
    "installer": (
        "govoplan.installer-receipt-authorities",
        "./installer-receipt-authority-keyring.schema.json",
    ),
}


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--purpose", choices=tuple(PURPOSES), required=True)
    parser.add_argument("--key-id", required=True)
    parser.add_argument(
        "--scope",
        action="append",
        choices=PROOF_SCOPES,
        default=[],
        help="Authorized proof scope; repeat as needed. Not used for installer keys.",
    )
    parser.add_argument("--private-key", type=Path, required=True)
    parser.add_argument("--keyring", type=Path, required=True)
    parser.add_argument(
        "--valid-days",
        type=int,
        default=365,
        help="Validity from generation time (default: 365 days).",
    )
    parser.add_argument(
        "--status",
        choices=("active", "next"),
        default="active",
    )
    args = parser.parse_args(argv)

    if not KEY_ID_PATTERN.fullmatch(args.key_id):
        parser.error("--key-id must be a valid opaque identifier")
    if args.valid_days < 1 or args.valid_days > 3660:
        parser.error("--valid-days must be between 1 and 3660")
    scopes = _resolve_scopes(parser, purpose=args.purpose, scopes=args.scope)

    private_path = args.private_key.expanduser().resolve()
    keyring_path = args.keyring.expanduser().resolve()
    _require_fresh_output(parser, private_path, label="private key")
    _require_fresh_output(parser, keyring_path, label="keyring")
    _require_private_directory(parser, private_path.parent)
    _require_output_directory(parser, keyring_path.parent)

    private_key = Ed25519PrivateKey.generate()
    private_bytes = private_key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=serialization.NoEncryption(),
    )
    public_bytes = private_key.public_key().public_bytes(
        encoding=serialization.Encoding.Raw,
        format=serialization.PublicFormat.Raw,
    )
    public_base64 = base64.b64encode(public_bytes).decode("ascii")
    now = datetime.now(UTC).replace(microsecond=0)
    not_after = now + timedelta(days=args.valid_days)
    purpose, schema = PURPOSES[args.purpose]
    keyring = {
        "$schema": schema,
        "schema_version": "0.1.0",
        "purpose": purpose,
        "keys": [
            {
                "key_id": args.key_id,
                "status": args.status,
                "public_key": public_base64,
                "allowed_scopes": scopes,
                "not_before": _rfc3339(now),
                "not_after": _rfc3339(not_after),
            }
        ],
    }

    _write_new_private_file(private_path, private_bytes)
    try:
        _write_new_private_file(
            keyring_path,
            (json.dumps(keyring, indent=2, sort_keys=True) + "\n").encode("utf-8"),
        )
    except BaseException:
        private_path.unlink(missing_ok=True)
        keyring_path.unlink(missing_ok=True)
        raise

    print(f"private_key={private_path}")
    print(f"keyring={keyring_path}")
    print(f"key_id={args.key_id}")
    print(f"allowed_scopes={','.join(scopes)}")
    return 0


def _resolve_scopes(
    parser: argparse.ArgumentParser, *, purpose: str, scopes: list[str]
) -> list[str]:
    if purpose == "installer":
        if scopes:
            parser.error("installer authorities do not accept --scope")
        return ["installed_release_origin"]
    unique = list(dict.fromkeys(scopes))
    if not unique:
        parser.error("proof authorities require at least one --scope")
    return unique


def _require_fresh_output(
    parser: argparse.ArgumentParser, path: Path, *, label: str
) -> None:
    if path.exists() or path.is_symlink():
        parser.error(f"{label.capitalize()} output already exists: {path}")


def _require_private_directory(
    parser: argparse.ArgumentParser, directory: Path
) -> None:
    _require_output_directory(parser, directory)
    mode = stat.S_IMODE(directory.stat().st_mode)
    if mode & (stat.S_IRWXG | stat.S_IRWXO):
        parser.error(
            "Private-key parent directory must not be accessible by group or others"
        )


def _require_output_directory(
    parser: argparse.ArgumentParser, directory: Path
) -> None:
    try:
        metadata = directory.lstat()
    except OSError as exc:
        parser.error(f"Output parent directory is unavailable: {directory}")
        raise AssertionError from exc
    if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode):
        parser.error(f"Output parent must be a real directory: {directory}")


def _write_new_private_file(path: Path, payload: bytes) -> None:
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
    if hasattr(os, "O_NOFOLLOW"):
        flags |= os.O_NOFOLLOW
    descriptor = os.open(path, flags, 0o600)
    try:
        with os.fdopen(descriptor, "wb", closefd=False) as handle:
            handle.write(payload)
            handle.flush()
            os.fsync(handle.fileno())
        metadata = os.fstat(descriptor)
        if not stat.S_ISREG(metadata.st_mode) or stat.S_IMODE(metadata.st_mode) != 0o600:
            raise OSError("Authority output could not be secured")
    finally:
        os.close(descriptor)


def _rfc3339(value: datetime) -> str:
    return value.isoformat().replace("+00:00", "Z")


if __name__ == "__main__":
    raise SystemExit(main())