452 lines
16 KiB
Python
Executable File
452 lines
16 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Re-run a capability-fit assessment against a trusted release catalog."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import sys
|
|
|
|
|
|
META_ROOT = Path(__file__).resolve().parents[2]
|
|
ASSESSMENT_TOOLS_ROOT = META_ROOT / "tools" / "assessments"
|
|
RELEASE_TOOLS_ROOT = META_ROOT / "tools" / "release"
|
|
for tools_root in (ASSESSMENT_TOOLS_ROOT, RELEASE_TOOLS_ROOT):
|
|
if str(tools_root) not in sys.path:
|
|
sys.path.insert(0, str(tools_root))
|
|
|
|
from govoplan_assessment import ( # noqa: E402
|
|
collect_installed_composition,
|
|
render_review,
|
|
review_capability_fit,
|
|
)
|
|
from govoplan_assessment.atomic_io import ( # noqa: E402
|
|
AtomicJsonWriteError,
|
|
atomic_write_json,
|
|
)
|
|
from govoplan_assessment.evidence import validate_payload # noqa: E402
|
|
from govoplan_release.catalog import DEFAULT_PUBLIC_BASE_URL, fetch_json # noqa: E402
|
|
|
|
|
|
TRUSTED_KEYRING_FILE_ENV = "GOVOPLAN_MODULE_PACKAGE_CATALOG_TRUSTED_KEYS_FILE"
|
|
MAX_ASSESSMENT_INPUT_BYTES = 16 * 1024 * 1024
|
|
MAX_EVIDENCE_INPUT_BYTES = 4 * 1024 * 1024
|
|
MAX_REPORT_OUTPUT_BYTES = 16 * 1024 * 1024
|
|
OPAQUE_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,159}$")
|
|
|
|
|
|
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
description="Validate a fit assessment and identify conclusions needing review after release drift."
|
|
)
|
|
parser.add_argument(
|
|
"--assessment",
|
|
type=Path,
|
|
default=META_ROOT / "docs" / "capability-fit-current.json",
|
|
)
|
|
parser.add_argument(
|
|
"--schema",
|
|
type=Path,
|
|
default=META_ROOT / "docs" / "capability-fit.schema.json",
|
|
)
|
|
source = parser.add_mutually_exclusive_group(required=True)
|
|
source.add_argument(
|
|
"--public",
|
|
action="store_true",
|
|
help="Use the fixed public GovOPlaN stable catalog and keyring URLs.",
|
|
)
|
|
source.add_argument(
|
|
"--catalog", type=Path, help="Read a catalog from a local file."
|
|
)
|
|
parser.add_argument(
|
|
"--keyring",
|
|
type=Path,
|
|
help="Published or candidate keyring whose hash is pinned by --catalog.",
|
|
)
|
|
parser.add_argument(
|
|
"--trusted-keyring",
|
|
type=Path,
|
|
help=(
|
|
"Independently pinned local keyring used as the catalog signature trust "
|
|
f"root; defaults to ${TRUSTED_KEYRING_FILE_ENV}."
|
|
),
|
|
)
|
|
parser.add_argument(
|
|
"--workspace-root",
|
|
type=Path,
|
|
default=META_ROOT.parent,
|
|
help="Workspace containing repository checkouts for local tag provenance.",
|
|
)
|
|
parser.add_argument(
|
|
"--skip-tag-provenance",
|
|
action="store_true",
|
|
help="Compare signed metadata without checking local annotated tags.",
|
|
)
|
|
installed = parser.add_mutually_exclusive_group()
|
|
installed.add_argument(
|
|
"--installed-evidence",
|
|
type=Path,
|
|
help="Validate an existing bounded installed-composition evidence document.",
|
|
)
|
|
installed.add_argument(
|
|
"--collect-installed-evidence",
|
|
type=Path,
|
|
help=(
|
|
"Collect the current interpreter's GovOPlaN distributions, write the "
|
|
"bounded evidence document, and include it in this review. Loading "
|
|
"module entry points executes installed GovOPlaN manifest factories."
|
|
),
|
|
)
|
|
parser.add_argument(
|
|
"--installed-evidence-schema",
|
|
type=Path,
|
|
default=META_ROOT / "docs" / "installed-composition-evidence.schema.json",
|
|
)
|
|
parser.add_argument(
|
|
"--installer-receipt",
|
|
type=Path,
|
|
help="Signed installer receipt binding installed payloads to this catalog.",
|
|
)
|
|
parser.add_argument(
|
|
"--installer-receipt-schema",
|
|
type=Path,
|
|
default=META_ROOT / "docs" / "installer-receipt.schema.json",
|
|
)
|
|
parser.add_argument(
|
|
"--installer-authority-keyring",
|
|
type=Path,
|
|
help=(
|
|
"Independently provisioned, role-scoped trust root for installer "
|
|
"receipt signatures."
|
|
),
|
|
)
|
|
parser.add_argument(
|
|
"--installer-authority-keyring-schema",
|
|
type=Path,
|
|
default=META_ROOT
|
|
/ "docs"
|
|
/ "installer-receipt-authority-keyring.schema.json",
|
|
)
|
|
parser.add_argument(
|
|
"--boundary-evidence",
|
|
type=Path,
|
|
help=(
|
|
"Externally issued target/provider/production proof bound to the "
|
|
"installed-evidence digest."
|
|
),
|
|
)
|
|
parser.add_argument(
|
|
"--boundary-evidence-schema",
|
|
type=Path,
|
|
default=META_ROOT / "docs" / "capability-fit-boundary-evidence.schema.json",
|
|
)
|
|
parser.add_argument(
|
|
"--boundary-authority-keyring",
|
|
type=Path,
|
|
help=(
|
|
"Independently provisioned keyring authorizing proof signers for "
|
|
"specific target/provider/production scopes."
|
|
),
|
|
)
|
|
parser.add_argument(
|
|
"--boundary-authority-keyring-schema",
|
|
type=Path,
|
|
default=META_ROOT
|
|
/ "docs"
|
|
/ "capability-fit-proof-authority-keyring.schema.json",
|
|
)
|
|
parser.add_argument(
|
|
"--verification-time",
|
|
help="Optional ISO-8601 time for reproducible boundary-proof verification.",
|
|
)
|
|
parser.add_argument(
|
|
"--expected-external-provider-subject",
|
|
help=(
|
|
"Opaque provider-environment ID that external-provider proof must match; "
|
|
"never supply a URL, credential, or endpoint identifier."
|
|
),
|
|
)
|
|
parser.add_argument(
|
|
"--json", action="store_true", help="Print the machine-readable review report."
|
|
)
|
|
parser.add_argument(
|
|
"--output",
|
|
type=Path,
|
|
help="Also write the machine-readable review report to this path.",
|
|
)
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = parse_args(argv)
|
|
if args.catalog is not None and args.keyring is None:
|
|
raise SystemExit("--keyring is required with --catalog")
|
|
if args.public and args.keyring is not None:
|
|
raise SystemExit("--keyring cannot be combined with --public")
|
|
if (args.boundary_evidence is None) != (args.boundary_authority_keyring is None):
|
|
raise SystemExit(
|
|
"--boundary-evidence and --boundary-authority-keyring are required together"
|
|
)
|
|
if (args.installer_receipt is None) != (
|
|
args.installer_authority_keyring is None
|
|
):
|
|
raise SystemExit(
|
|
"--installer-receipt and --installer-authority-keyring are required together"
|
|
)
|
|
if args.installer_receipt is not None and (
|
|
args.installed_evidence is None and args.collect_installed_evidence is None
|
|
):
|
|
raise SystemExit(
|
|
"--installer-receipt requires --installed-evidence or --collect-installed-evidence"
|
|
)
|
|
if args.boundary_evidence is not None and (
|
|
args.installed_evidence is None and args.collect_installed_evidence is None
|
|
):
|
|
raise SystemExit(
|
|
"--boundary-evidence requires --installed-evidence or --collect-installed-evidence"
|
|
)
|
|
if (
|
|
args.expected_external_provider_subject is not None
|
|
and OPAQUE_ID_PATTERN.fullmatch(args.expected_external_provider_subject) is None
|
|
):
|
|
raise SystemExit(
|
|
"--expected-external-provider-subject must be a bounded opaque ID"
|
|
)
|
|
if (
|
|
args.collect_installed_evidence is not None
|
|
and args.output is not None
|
|
and args.collect_installed_evidence.resolve() == args.output.resolve()
|
|
):
|
|
raise SystemExit("Evidence and review output paths must differ")
|
|
configured_trusted_keyring = os.getenv(TRUSTED_KEYRING_FILE_ENV, "").strip()
|
|
trusted_keyring_path = args.trusted_keyring or (
|
|
Path(configured_trusted_keyring) if configured_trusted_keyring else None
|
|
)
|
|
if trusted_keyring_path is None:
|
|
raise SystemExit(
|
|
f"--trusted-keyring or ${TRUSTED_KEYRING_FILE_ENV} is required"
|
|
)
|
|
assessment = read_object(
|
|
args.assessment,
|
|
label="assessment",
|
|
max_bytes=MAX_ASSESSMENT_INPUT_BYTES,
|
|
)
|
|
schema = read_object(
|
|
args.schema,
|
|
label="schema",
|
|
max_bytes=MAX_ASSESSMENT_INPUT_BYTES,
|
|
)
|
|
if args.public:
|
|
catalog = fetch_public_json(
|
|
f"{DEFAULT_PUBLIC_BASE_URL}/catalogs/v1/channels/stable.json",
|
|
label="catalog",
|
|
)
|
|
published_keyring = fetch_public_json(
|
|
f"{DEFAULT_PUBLIC_BASE_URL}/catalogs/v1/keyring.json",
|
|
label="published keyring",
|
|
)
|
|
else:
|
|
catalog = read_object(
|
|
args.catalog, label="catalog", max_bytes=MAX_ASSESSMENT_INPUT_BYTES
|
|
)
|
|
published_keyring = read_object(
|
|
args.keyring,
|
|
label="published keyring",
|
|
max_bytes=MAX_EVIDENCE_INPUT_BYTES,
|
|
)
|
|
trusted_keyring = read_object(
|
|
trusted_keyring_path,
|
|
label="trusted keyring",
|
|
max_bytes=MAX_EVIDENCE_INPUT_BYTES,
|
|
)
|
|
|
|
installed_evidence = None
|
|
installed_evidence_schema = None
|
|
if (
|
|
args.installed_evidence is not None
|
|
or args.collect_installed_evidence is not None
|
|
):
|
|
installed_evidence_schema = read_object(
|
|
args.installed_evidence_schema,
|
|
label="installed evidence schema",
|
|
max_bytes=MAX_EVIDENCE_INPUT_BYTES,
|
|
)
|
|
if args.installed_evidence is not None:
|
|
installed_evidence = read_object(
|
|
args.installed_evidence,
|
|
label="installed evidence",
|
|
max_bytes=MAX_EVIDENCE_INPUT_BYTES,
|
|
)
|
|
else:
|
|
installed_evidence = collect_installed_composition(assessment=assessment)
|
|
collection_schema_errors = validate_payload(
|
|
payload=installed_evidence,
|
|
schema=installed_evidence_schema,
|
|
)
|
|
if collection_schema_errors:
|
|
raise SystemExit(
|
|
"Collected installed evidence did not satisfy its bounded schema"
|
|
)
|
|
write_object(
|
|
args.collect_installed_evidence,
|
|
installed_evidence,
|
|
max_bytes=MAX_EVIDENCE_INPUT_BYTES,
|
|
label="installed evidence",
|
|
)
|
|
|
|
boundary_evidence = None
|
|
boundary_evidence_schema = None
|
|
boundary_authority_keyring = None
|
|
boundary_authority_keyring_schema = None
|
|
if args.boundary_evidence is not None:
|
|
boundary_evidence = read_object(
|
|
args.boundary_evidence,
|
|
label="boundary evidence",
|
|
max_bytes=MAX_EVIDENCE_INPUT_BYTES,
|
|
)
|
|
boundary_evidence_schema = read_object(
|
|
args.boundary_evidence_schema,
|
|
label="boundary evidence schema",
|
|
max_bytes=MAX_EVIDENCE_INPUT_BYTES,
|
|
)
|
|
boundary_authority_keyring = read_object(
|
|
args.boundary_authority_keyring,
|
|
label="boundary authority keyring",
|
|
max_bytes=MAX_EVIDENCE_INPUT_BYTES,
|
|
)
|
|
boundary_authority_keyring_schema = read_object(
|
|
args.boundary_authority_keyring_schema,
|
|
label="boundary authority keyring schema",
|
|
max_bytes=MAX_EVIDENCE_INPUT_BYTES,
|
|
)
|
|
|
|
installer_receipt = None
|
|
installer_receipt_schema = None
|
|
installer_authority_keyring = None
|
|
installer_authority_keyring_schema = None
|
|
if args.installer_receipt is not None:
|
|
installer_receipt = read_object(
|
|
args.installer_receipt,
|
|
label="installer receipt",
|
|
max_bytes=MAX_EVIDENCE_INPUT_BYTES,
|
|
)
|
|
installer_receipt_schema = read_object(
|
|
args.installer_receipt_schema,
|
|
label="installer receipt schema",
|
|
max_bytes=MAX_EVIDENCE_INPUT_BYTES,
|
|
)
|
|
installer_authority_keyring = read_object(
|
|
args.installer_authority_keyring,
|
|
label="installer authority keyring",
|
|
max_bytes=MAX_EVIDENCE_INPUT_BYTES,
|
|
)
|
|
installer_authority_keyring_schema = read_object(
|
|
args.installer_authority_keyring_schema,
|
|
label="installer authority keyring schema",
|
|
max_bytes=MAX_EVIDENCE_INPUT_BYTES,
|
|
)
|
|
|
|
verification_time = parse_datetime(args.verification_time)
|
|
report = review_capability_fit(
|
|
assessment=assessment,
|
|
schema=schema,
|
|
catalog=catalog,
|
|
published_keyring=published_keyring,
|
|
trusted_keyring=trusted_keyring,
|
|
workspace_root=None
|
|
if args.skip_tag_provenance
|
|
else args.workspace_root.resolve(),
|
|
installed_evidence=installed_evidence,
|
|
installed_evidence_schema=installed_evidence_schema,
|
|
installer_receipt=installer_receipt,
|
|
installer_receipt_schema=installer_receipt_schema,
|
|
installer_authority_keyring=installer_authority_keyring,
|
|
installer_authority_keyring_schema=installer_authority_keyring_schema,
|
|
boundary_evidence=boundary_evidence,
|
|
boundary_evidence_schema=boundary_evidence_schema,
|
|
boundary_authority_keyring=boundary_authority_keyring,
|
|
boundary_authority_keyring_schema=boundary_authority_keyring_schema,
|
|
verification_time=verification_time,
|
|
installed_evidence_mode=(
|
|
"direct_local"
|
|
if args.collect_installed_evidence is not None
|
|
else "imported_unsigned"
|
|
),
|
|
expected_external_provider_subject=args.expected_external_provider_subject,
|
|
)
|
|
encoded = json.dumps(report, indent=2, sort_keys=True) + "\n"
|
|
if args.output is not None:
|
|
write_object(
|
|
args.output,
|
|
report,
|
|
max_bytes=MAX_REPORT_OUTPUT_BYTES,
|
|
label="review report",
|
|
)
|
|
sys.stdout.write(encoded if args.json else render_review(report))
|
|
return {"current": 0, "blocked": 1, "review_required": 2}[report["status"]]
|
|
|
|
|
|
def read_object(path: Path | None, *, label: str, max_bytes: int) -> dict[str, object]:
|
|
if path is None:
|
|
raise SystemExit(f"Missing {label} path")
|
|
try:
|
|
if path.stat().st_size > max_bytes:
|
|
raise SystemExit(
|
|
f"{label.capitalize()} exceeds the {max_bytes}-byte input limit"
|
|
)
|
|
with path.open("rb") as handle:
|
|
encoded = handle.read(max_bytes + 1)
|
|
if len(encoded) > max_bytes:
|
|
raise SystemExit(
|
|
f"{label.capitalize()} exceeds the {max_bytes}-byte input limit"
|
|
)
|
|
payload = json.loads(encoded.decode("utf-8"))
|
|
except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
|
|
raise SystemExit(f"Could not read {label} {path}: {exc}") from exc
|
|
if not isinstance(payload, dict):
|
|
raise SystemExit(f"{label.capitalize()} must be a JSON object: {path}")
|
|
return payload
|
|
|
|
|
|
def write_object(
|
|
path: Path | None,
|
|
payload: dict[str, object],
|
|
*,
|
|
max_bytes: int,
|
|
label: str,
|
|
) -> None:
|
|
if path is None:
|
|
raise SystemExit(f"Missing {label} output path")
|
|
try:
|
|
atomic_write_json(path, payload, max_bytes=max_bytes)
|
|
except AtomicJsonWriteError as exc:
|
|
raise SystemExit(f"Could not securely write {label}") from exc
|
|
|
|
|
|
def parse_datetime(value: str | None):
|
|
if value is None:
|
|
return None
|
|
from datetime import datetime
|
|
|
|
try:
|
|
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
except ValueError as exc:
|
|
raise SystemExit("--verification-time must be an ISO-8601 timestamp") from exc
|
|
if parsed.tzinfo is None:
|
|
raise SystemExit("--verification-time must include a timezone")
|
|
return parsed
|
|
|
|
|
|
def fetch_public_json(url: str, *, label: str) -> dict[str, object]:
|
|
result = fetch_json(url)
|
|
if result.get("ok") is not True or not isinstance(result.get("payload"), dict):
|
|
raise SystemExit(f"Could not fetch fixed public {label} endpoint.")
|
|
return result["payload"]
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|