GovOPlaN/govoplan
1
0
Fork 0
Code Issues 10 Pull Requests Actions 3 Packages Projects Releases Wiki Activity

feat(assessment): verify installed composition evidence

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-22 18:34:02 +02:00
parent d0dc916837
commit 6c0b003dee
8 changed files with 2785 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

204
tools/assessments/capability-fit.py
View File

@@ -17,11 +17,18 @@ 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 render_review, review_capability_fit # noqa: E402
from govoplan_assessment import ( # noqa: E402
collect_installed_composition,
render_review,
review_capability_fit,
)
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
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
@@ -71,6 +78,58 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
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(
"--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(
"--json", action="store_true", help="Print the machine-readable review report."
)
@@ -88,6 +147,22 @@ def main(argv: list[str] | None = None) -> int:
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.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.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
@@ -96,8 +171,16 @@ def main(argv: list[str] | None = None) -> int:
raise SystemExit(
f"--trusted-keyring or ${TRUSTED_KEYRING_FILE_ENV} is required"
)
assessment = read_object(args.assessment, label="assessment")
schema = read_object(args.schema, label="schema")
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",
@@ -108,13 +191,75 @@ def main(argv: list[str] | None = None) -> int:
label="published keyring",
)
else:
catalog = read_object(args.catalog, label="catalog")
published_keyring = read_object(args.keyring, label="published keyring")
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)
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,
)
report = review_capability_fit(
assessment=assessment,
schema=schema,
@@ -124,6 +269,13 @@ def main(argv: list[str] | None = None) -> int:
workspace_root=None
if args.skip_tag_provenance
else args.workspace_root.resolve(),
installed_evidence=installed_evidence,
installed_evidence_schema=installed_evidence_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=parse_datetime(args.verification_time),
)
encoded = json.dumps(report, indent=2, sort_keys=True) + "\n"
if args.output is not None:
@@ -133,18 +285,54 @@ def main(argv: list[str] | None = None) -> int:
return {"current": 0, "blocked": 1, "review_required": 2}[report["status"]]
def read_object(path: Path | None, *, label: str) -> dict[str, object]:
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:
payload = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
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]) -> None:
if path is None:
raise SystemExit("Missing installed evidence output path")
encoded = json.dumps(payload, indent=2, sort_keys=True) + "\n"
if len(encoded.encode("utf-8")) > MAX_EVIDENCE_INPUT_BYTES:
raise SystemExit(
"Collected installed evidence exceeds the bounded output limit"
)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(encoded, encoding="utf-8")
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):