171 lines
7.2 KiB
Python
171 lines
7.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Collect live installed payloads and issue an independently signed receipt."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
|
|
META_ROOT = Path(__file__).resolve().parents[2]
|
|
for tools_root in (META_ROOT / "tools" / "assessments", META_ROOT / "tools" / "release"):
|
|
if str(tools_root) not in sys.path:
|
|
sys.path.insert(0, str(tools_root))
|
|
|
|
from cryptography.hazmat.primitives import serialization # noqa: E402
|
|
from cryptography.hazmat.primitives.asymmetric.ed25519 import ( # noqa: E402
|
|
Ed25519PrivateKey,
|
|
)
|
|
|
|
from govoplan_assessment import collect_installed_composition # noqa: E402
|
|
from govoplan_assessment.atomic_io import ( # noqa: E402
|
|
AtomicJsonWriteError,
|
|
atomic_write_json,
|
|
)
|
|
from govoplan_assessment.capability_fit import ( # noqa: E402
|
|
canonical_hash,
|
|
review_capability_fit,
|
|
)
|
|
from govoplan_assessment.evidence import validate_payload # noqa: E402
|
|
from govoplan_assessment.installer_receipt import ( # noqa: E402
|
|
issue_installer_receipt,
|
|
)
|
|
|
|
|
|
MAX_INPUT_BYTES = 16 * 1024 * 1024
|
|
MAX_OUTPUT_BYTES = 4 * 1024 * 1024
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--assessment", type=Path, required=True)
|
|
parser.add_argument("--assessment-schema", type=Path, default=META_ROOT / "docs" / "capability-fit.schema.json")
|
|
parser.add_argument("--catalog", type=Path, required=True)
|
|
parser.add_argument("--keyring", type=Path, required=True)
|
|
parser.add_argument("--trusted-keyring", type=Path, required=True)
|
|
parser.add_argument("--receipt-id", required=True)
|
|
parser.add_argument("--signing-key", required=True, metavar="KEY_ID=/PATH/PRIVATE.pem")
|
|
parser.add_argument(
|
|
"--python-artifact",
|
|
action="append",
|
|
default=[],
|
|
required=True,
|
|
metavar="PACKAGE=/PATH/TO/CONSUMED.whl",
|
|
help="Exact wheel consumed by this installer; repeat for every package.",
|
|
)
|
|
parser.add_argument("--evidence-output", type=Path, required=True)
|
|
parser.add_argument("--receipt-output", type=Path, required=True)
|
|
parser.add_argument("--installed-evidence-schema", type=Path, default=META_ROOT / "docs" / "installed-composition-evidence.schema.json")
|
|
parser.add_argument("--receipt-schema", type=Path, default=META_ROOT / "docs" / "installer-receipt.schema.json")
|
|
args = parser.parse_args(argv)
|
|
if args.evidence_output.resolve() == args.receipt_output.resolve():
|
|
parser.error("evidence and receipt output paths must differ")
|
|
|
|
assessment = read_object(args.assessment, label="assessment")
|
|
assessment_schema = read_object(args.assessment_schema, label="assessment schema")
|
|
catalog = read_object(args.catalog, label="catalog")
|
|
keyring = read_object(args.keyring, label="published keyring")
|
|
trusted_keyring = read_object(args.trusted_keyring, label="trusted keyring")
|
|
evidence_schema = read_object(args.installed_evidence_schema, label="installed evidence schema")
|
|
receipt_schema = read_object(args.receipt_schema, label="installer receipt schema")
|
|
release = catalog.get("release")
|
|
if not isinstance(release, dict) or release.get("keyring_sha256") != canonical_hash(keyring):
|
|
parser.error("catalog does not pin the supplied published keyring")
|
|
|
|
evidence = collect_installed_composition(assessment=assessment)
|
|
if validate_payload(payload=evidence, schema=evidence_schema):
|
|
parser.error("live installed observation does not satisfy its bounded schema")
|
|
report = review_capability_fit(
|
|
assessment=assessment,
|
|
schema=assessment_schema,
|
|
catalog=catalog,
|
|
published_keyring=keyring,
|
|
trusted_keyring=trusted_keyring,
|
|
workspace_root=None,
|
|
installed_evidence=evidence,
|
|
installed_evidence_schema=evidence_schema,
|
|
installed_evidence_mode="direct_local",
|
|
)
|
|
required_scopes = (
|
|
"catalog_signature_and_keyring",
|
|
"catalog_signature_and_trusted_keyring",
|
|
"published_keyring_hash",
|
|
"release_metadata",
|
|
"installed_artifacts",
|
|
"installed_record_integrity",
|
|
)
|
|
if any(report["proof_scope"].get(scope, {}).get("valid") is not True for scope in required_scopes):
|
|
parser.error("trusted catalog and live installed-composition checks must pass before receipt issuance")
|
|
|
|
key_id, private_key = read_signing_key(args.signing_key)
|
|
consumed_artifacts = parse_package_paths(tuple(args.python_artifact))
|
|
receipt = issue_installer_receipt(
|
|
assessment=assessment,
|
|
catalog=catalog,
|
|
installed_evidence=evidence,
|
|
receipt_id=args.receipt_id,
|
|
key_id=key_id,
|
|
private_key=private_key,
|
|
consumed_artifacts=consumed_artifacts,
|
|
same_process_observation=True,
|
|
)
|
|
if validate_payload(payload=receipt, schema=receipt_schema):
|
|
parser.error("issued installer receipt does not satisfy its bounded schema")
|
|
write_object(args.evidence_output, evidence, label="installed evidence")
|
|
write_object(args.receipt_output, receipt, label="installer receipt")
|
|
print(json.dumps({"receipt_id": args.receipt_id, "evidence_output": str(args.evidence_output), "receipt_output": str(args.receipt_output)}, sort_keys=True))
|
|
return 0
|
|
|
|
|
|
def read_object(path: Path, *, label: str) -> dict[str, object]:
|
|
try:
|
|
if path.stat().st_size > MAX_INPUT_BYTES:
|
|
raise ValueError("input exceeds size limit")
|
|
encoded = path.read_bytes()
|
|
payload = json.loads(encoded.decode("utf-8"))
|
|
except (OSError, UnicodeDecodeError, json.JSONDecodeError, ValueError) as exc:
|
|
raise SystemExit(f"Could not read {label}") from exc
|
|
if not isinstance(payload, dict):
|
|
raise SystemExit(f"{label.capitalize()} must be a JSON object")
|
|
return payload
|
|
|
|
|
|
def read_signing_key(value: str) -> tuple[str, Ed25519PrivateKey]:
|
|
key_id, separator, path_text = value.partition("=")
|
|
if not separator or not key_id or not path_text:
|
|
raise SystemExit("--signing-key must use KEY_ID=/path/to/private.pem")
|
|
try:
|
|
key = serialization.load_pem_private_key(Path(path_text).expanduser().read_bytes(), password=None)
|
|
except (OSError, ValueError) as exc:
|
|
raise SystemExit("Could not read installer signing key") from exc
|
|
if not isinstance(key, Ed25519PrivateKey):
|
|
raise SystemExit("Installer signing key must be Ed25519")
|
|
return key_id, key
|
|
|
|
|
|
def parse_package_paths(values: tuple[str, ...]) -> dict[str, Path]:
|
|
result: dict[str, Path] = {}
|
|
for value in values:
|
|
package, separator, path_text = value.partition("=")
|
|
package = package.strip()
|
|
path_text = path_text.strip()
|
|
if not separator or not package or not path_text or package in result:
|
|
raise SystemExit(
|
|
"--python-artifact must uniquely use PACKAGE=/path/to/consumed.whl"
|
|
)
|
|
result[package] = Path(path_text).expanduser()
|
|
return result
|
|
|
|
|
|
def write_object(path: Path, payload: dict[str, object], *, label: str) -> None:
|
|
try:
|
|
atomic_write_json(path, payload, max_bytes=MAX_OUTPUT_BYTES)
|
|
except AtomicJsonWriteError as exc:
|
|
raise SystemExit(f"Could not securely write {label}") from exc
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|