107 lines
4.6 KiB
Python
107 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Plan fail-closed Python runtime profile selection without mutating the host."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
|
|
|
|
class PlanError(ValueError):
|
|
def __init__(self, code: str, message: str):
|
|
super().__init__(message)
|
|
self.code = code
|
|
|
|
|
|
def choose_profile(probe: dict, requested: str) -> str:
|
|
if probe.get("architecture") != "amd64":
|
|
raise PlanError("ARCHITECTURE_UNSUPPORTED", "Offline Workstation profile requires AMD64")
|
|
nvidia = probe.get("nvidia", {})
|
|
ready = all(nvidia.get(key) is True for key in ("present", "supported", "driver_ready"))
|
|
if requested == "cpu-only":
|
|
return "cpu-only"
|
|
if requested == "nvidia-cu128":
|
|
if not ready:
|
|
raise PlanError("NVIDIA_PROFILE_NOT_READY", "requested NVIDIA profile is not ready")
|
|
return "nvidia-cu128"
|
|
if requested != "auto":
|
|
raise PlanError("PROFILE_UNKNOWN", f"unknown requested profile: {requested}")
|
|
return "nvidia-cu128" if ready else "cpu-only"
|
|
|
|
|
|
def plan(probe: dict, receipt: dict, requested: str, operation_id: str, journal: dict | None,
|
|
allow_test_receipt: bool = False) -> dict:
|
|
if receipt.get("schema") != "com.cezen.nexus-one.profile-membership-verification-receipt.v1":
|
|
raise PlanError("MEMBERSHIP_RECEIPT_INVALID", "membership receipt schema is invalid")
|
|
if receipt.get("result") != "verified":
|
|
raise PlanError("MEMBERSHIP_UNVERIFIED", "signed profile membership is not verified")
|
|
if receipt.get("environment") != "production" and not allow_test_receipt:
|
|
raise PlanError("TEST_RECEIPT_FORBIDDEN", "test receipt is forbidden for production planning")
|
|
if receipt.get("namespace") != "offline-workstation/python-runtime-profiles":
|
|
raise PlanError("MEMBERSHIP_RECEIPT_INVALID", "membership namespace is invalid")
|
|
if not receipt.get("manifest_sha256"):
|
|
raise PlanError("MEMBERSHIP_UNVERIFIED", "verified membership lacks manifest identity")
|
|
selected = choose_profile(probe, requested)
|
|
if selected not in receipt.get("profiles", []):
|
|
raise PlanError("PROFILE_NOT_IN_SIGNED_MEMBERSHIP", f"{selected} is not signed membership")
|
|
|
|
if journal:
|
|
state = journal.get("state")
|
|
previous = journal.get("selected_profile")
|
|
if state != "rolled_back":
|
|
if journal.get("operation_id") != operation_id:
|
|
raise PlanError("ACTIVE_OPERATION_EXISTS", "another profile operation is active")
|
|
if previous != selected:
|
|
raise PlanError(
|
|
"PROFILE_CHANGE_REQUIRES_ROLLBACK",
|
|
f"journal selected {previous}; requested selection is {selected}",
|
|
)
|
|
return {
|
|
"action": "resume",
|
|
"operation_id": operation_id,
|
|
"selected_profile": selected,
|
|
"resume_from": state,
|
|
"manifest_sha256": receipt["manifest_sha256"],
|
|
"mutation_allowed_after_journal_write": True,
|
|
}
|
|
if journal.get("operation_id") == operation_id:
|
|
raise PlanError("NEW_OPERATION_REQUIRED", "rolled-back selection requires a new operation id")
|
|
|
|
return {
|
|
"action": "start",
|
|
"operation_id": operation_id,
|
|
"selected_profile": selected,
|
|
"journal_state": "profile_selected",
|
|
"checkpoint_required": True,
|
|
"manifest_sha256": receipt["manifest_sha256"],
|
|
"mutation_allowed_after_journal_write": True,
|
|
}
|
|
|
|
|
|
def load(path: Path | None) -> dict | None:
|
|
return json.loads(path.read_text(encoding="utf-8")) if path else None
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--probe", type=Path, required=True)
|
|
parser.add_argument("--membership-receipt", type=Path, required=True)
|
|
parser.add_argument("--requested", choices=("auto", "cpu-only", "nvidia-cu128"), default="auto")
|
|
parser.add_argument("--operation-id", required=True)
|
|
parser.add_argument("--journal", type=Path)
|
|
parser.add_argument("--allow-test-receipt", action="store_true")
|
|
args = parser.parse_args()
|
|
try:
|
|
result = plan(load(args.probe), load(args.membership_receipt), args.requested,
|
|
args.operation_id, load(args.journal), args.allow_test_receipt)
|
|
except PlanError as error:
|
|
print(json.dumps({"allowed": False, "code": error.code, "message": str(error)}, sort_keys=True))
|
|
return 1
|
|
print(json.dumps({"allowed": True, **result}, sort_keys=True))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|