342 lines
13 KiB
Python
342 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
"""Fail-closed static eligibility gate for artifact execution tooling."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
SHA256_PATTERN = re.compile(r"[0-9a-f]{64}")
|
|
COMMIT_PATTERN = re.compile(r"[0-9a-f]{40}")
|
|
READY_DECISION = "READY_FOR_HARDENED_RUNTIME_DEPLOYMENT"
|
|
HARD_EFFECTS = {"PERSISTENT_WRITE", "UNBOUNDED_OR_UNKNOWN"}
|
|
ALLOWED_EFFECTS = {
|
|
"ALLOWED_APPLICATION_WRITE",
|
|
"BOUNDED_WATCHDOG",
|
|
"EXPLICIT_PROCESS_EXIT",
|
|
"EXPECTED_VOLATILE_RUNTIME_EFFECT",
|
|
"FAIL_CLOSED_TERMINATION",
|
|
"HASH_BOUND_SAME_FD",
|
|
"OS_RECLAIMED_ON_EXIT",
|
|
"RESTORED_BY_LOADER",
|
|
"PAYLOAD_PROCESS_LOCAL",
|
|
*HARD_EFFECTS,
|
|
}
|
|
EXPECTED_PAYLOAD_MANAGER = {
|
|
"base_commit": "cfbc70f30f419b09bf2b52283f7409e2d3117ee1",
|
|
"hardened_commit": "e23d94ff91233aa770e2342800c1467875bdef44",
|
|
"installed": False,
|
|
"release": "v0.3.1-chimera-controlled-phase07",
|
|
"reproducible": True,
|
|
"sha256": "8fecf8241a46246eddbd21e8bb4d875f5d76f1f4f4c6a11384df1f131aa5e5b1",
|
|
"size": 99560,
|
|
}
|
|
EXPECTED_LOADER = {
|
|
"base_commit": "699e8bcff03e91e8d6ca6eba281af25c5a58d8c2",
|
|
"hardened_commit": "197623058f509eddde18868dafcb92fdcac66464",
|
|
"installed": False,
|
|
"release": "v0.23-chimera-phase07",
|
|
"reproducible": True,
|
|
"sha256": "63e810982471eb40cae3a20aa9df9a0a02892f420e429874fae4e99aa400b561",
|
|
"size": 397000,
|
|
}
|
|
EXPECTED_SDK = {
|
|
"commit": "d2e2e585740362976a39fdd5ccf390f199a7bc37",
|
|
"release": "v0.41",
|
|
}
|
|
|
|
|
|
def hash_file(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as stream:
|
|
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def deny(reason_codes: list[str], artifact_sha256: str | None = None) -> int:
|
|
result: dict[str, Any] = {
|
|
"decision": "DENY",
|
|
"execution_authorized": False,
|
|
"reason_codes": sorted(set(reason_codes)),
|
|
"schema_version": 1,
|
|
}
|
|
if artifact_sha256 is not None:
|
|
result["artifact_sha256"] = artifact_sha256
|
|
print(json.dumps(result, sort_keys=True))
|
|
return 2
|
|
|
|
|
|
def load_json(path: Path) -> dict[str, Any]:
|
|
document = json.loads(path.read_text(encoding="utf-8"))
|
|
if not isinstance(document, dict):
|
|
raise ValueError(f"{path}: root must be an object")
|
|
return document
|
|
|
|
|
|
def validate_runtime_profile(
|
|
profile: dict[str, Any],
|
|
manifest: dict[str, Any],
|
|
firmware: str | None,
|
|
) -> list[str]:
|
|
if profile.get("schema_version") != 1:
|
|
raise ValueError("unsupported controlled runtime profile schema")
|
|
if profile.get("profile") != "controlled-ps5-runtime":
|
|
raise ValueError("unexpected controlled runtime profile")
|
|
|
|
reasons: list[str] = []
|
|
artifact = profile["artifact"]
|
|
execution = profile["execution"]
|
|
profile_firmware = profile["firmware"]
|
|
payload_manager = profile["payload_manager"]
|
|
loader = profile["loader"]
|
|
sdk = profile["sdk"]
|
|
budgets = profile["budgets"]
|
|
deployment = profile["deployment"]
|
|
effects = profile["effects"]
|
|
expected_volatile_effects = profile["expected_volatile_effects"]
|
|
blockers = profile["hard_blockers"]
|
|
if not all(
|
|
isinstance(item, dict)
|
|
for item in (
|
|
artifact,
|
|
execution,
|
|
profile_firmware,
|
|
payload_manager,
|
|
loader,
|
|
sdk,
|
|
budgets,
|
|
deployment,
|
|
)
|
|
):
|
|
raise ValueError("controlled runtime profile contains a malformed object")
|
|
if (
|
|
not isinstance(effects, list)
|
|
or not isinstance(expected_volatile_effects, list)
|
|
or not isinstance(blockers, list)
|
|
):
|
|
raise ValueError("controlled runtime profile arrays are malformed")
|
|
|
|
if profile.get("decision") != READY_DECISION:
|
|
reasons.append("RUNTIME_PROFILE_BLOCKED")
|
|
if (
|
|
profile.get("execution_authorized") is not False
|
|
or execution.get("authorized") is not False
|
|
or execution.get("transferred") is not False
|
|
or execution.get("executed") is not False
|
|
):
|
|
reasons.append("RUNTIME_PROFILE_EXECUTION_STATE_INVALID")
|
|
if execution.get("execution_eligible") is not True:
|
|
reasons.append("RUNTIME_PROFILE_EXECUTION_INELIGIBLE")
|
|
if artifact.get("built") is not True:
|
|
reasons.append("RUNTIME_PROFILE_ARTIFACT_NOT_BUILT")
|
|
if blockers:
|
|
reasons.append("RUNTIME_PROFILE_HARD_BLOCKERS_PRESENT")
|
|
|
|
manifest_artifact = manifest["artifact"]
|
|
manifest_source = manifest.get("source")
|
|
if not isinstance(manifest_source, dict):
|
|
raise ValueError("artifact manifest source record is required")
|
|
for key in ("id", "filename", "sha256", "size"):
|
|
if artifact.get(key) != manifest_artifact.get(key):
|
|
reasons.append("RUNTIME_PROFILE_ARTIFACT_MISMATCH")
|
|
source_commit = artifact.get("source_commit")
|
|
if (
|
|
not isinstance(source_commit, str)
|
|
or COMMIT_PATTERN.fullmatch(source_commit) is None
|
|
or source_commit != manifest_source.get("commit")
|
|
or manifest_source.get("dirty") is not False
|
|
):
|
|
reasons.append("RUNTIME_PROFILE_SOURCE_COMMIT_MISMATCH")
|
|
|
|
if firmware is None:
|
|
reasons.append("EXACT_FIRMWARE_REQUIRED")
|
|
if profile_firmware.get("exact") != "9.60" or firmware != "9.60":
|
|
reasons.append("FIRMWARE_MISMATCH")
|
|
if profile_firmware.get("evidence") != "jens_explicitly_confirmed_exact_9.60":
|
|
reasons.append("FIRMWARE_EVIDENCE_MISMATCH")
|
|
|
|
for key, value in EXPECTED_PAYLOAD_MANAGER.items():
|
|
if payload_manager.get(key) != value:
|
|
reasons.append("PAYLOAD_MANAGER_IDENTITY_MISMATCH")
|
|
for key, value in EXPECTED_LOADER.items():
|
|
if loader.get(key) != value:
|
|
reasons.append("LOADER_IDENTITY_MISMATCH")
|
|
for key, value in EXPECTED_SDK.items():
|
|
if sdk.get(key) != value:
|
|
reasons.append("SDK_IDENTITY_MISMATCH")
|
|
|
|
expected_budgets = {
|
|
"automatic_retry": False,
|
|
"filesystem_write_budget": "controlled_artifact_directory_only",
|
|
"payload_network_access": "none",
|
|
"persistent_write_budget": "controlled_artifact_removable",
|
|
}
|
|
for key, value in expected_budgets.items():
|
|
if budgets.get(key) != value:
|
|
reasons.append("RUNTIME_BUDGET_MISMATCH")
|
|
runtime = budgets.get("maximum_runtime_ms")
|
|
if (
|
|
not isinstance(runtime, int)
|
|
or isinstance(runtime, bool)
|
|
or not (1 <= runtime <= 2000)
|
|
):
|
|
reasons.append("RUNTIME_BUDGET_MISMATCH")
|
|
if deployment != {
|
|
"installed": False,
|
|
"ready_for_installation": True,
|
|
"rollback_prepared": True,
|
|
}:
|
|
reasons.append("DEPLOYMENT_STATE_MISMATCH")
|
|
|
|
classified_volatile_effects: list[str] = []
|
|
effect_ids: set[str] = set()
|
|
for item in effects:
|
|
if not isinstance(item, dict):
|
|
raise ValueError("controlled runtime effect must be an object")
|
|
effect_id = item.get("id")
|
|
if not isinstance(effect_id, str) or not effect_id or effect_id in effect_ids:
|
|
raise ValueError("controlled runtime effect IDs must be unique strings")
|
|
effect_ids.add(effect_id)
|
|
if item.get("classification") not in ALLOWED_EFFECTS:
|
|
raise ValueError("controlled runtime effect classification is invalid")
|
|
if item.get("classification") == "EXPECTED_VOLATILE_RUNTIME_EFFECT":
|
|
classified_volatile_effects.append(effect_id)
|
|
if item.get("classification") in HARD_EFFECTS:
|
|
reasons.append("RUNTIME_PROFILE_HARD_EFFECT")
|
|
if (
|
|
not all(
|
|
isinstance(item, str) and item for item in expected_volatile_effects
|
|
)
|
|
or len(set(expected_volatile_effects)) != len(expected_volatile_effects)
|
|
or sorted(expected_volatile_effects) != sorted(classified_volatile_effects)
|
|
):
|
|
reasons.append("RUNTIME_PROFILE_VOLATILE_EFFECTS_MISMATCH")
|
|
|
|
return reasons
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--manifest", type=Path, required=True)
|
|
parser.add_argument("--denylist", type=Path, required=True)
|
|
parser.add_argument("--artifact", type=Path)
|
|
parser.add_argument("--runtime-profile", type=Path)
|
|
parser.add_argument("--firmware")
|
|
args = parser.parse_args()
|
|
|
|
try:
|
|
manifest = load_json(args.manifest)
|
|
denylist = load_json(args.denylist)
|
|
if manifest.get("schema_version") != 1:
|
|
raise ValueError("unsupported artifact manifest schema")
|
|
if denylist.get("schema_version") != 1:
|
|
raise ValueError("unsupported denylist schema")
|
|
if denylist.get("hash_algorithm") != "sha256":
|
|
raise ValueError("denylist hash algorithm must be sha256")
|
|
if denylist.get("fail_closed") is not True:
|
|
raise ValueError("denylist must be fail-closed")
|
|
|
|
artifact_record = manifest["artifact"]
|
|
execution = manifest["execution"]
|
|
if not isinstance(artifact_record, dict) or not isinstance(execution, dict):
|
|
raise ValueError("manifest artifact and execution records must be objects")
|
|
digest = artifact_record["sha256"]
|
|
if not isinstance(digest, str) or SHA256_PATTERN.fullmatch(digest) is None:
|
|
raise ValueError("manifest artifact SHA-256 is invalid")
|
|
if not isinstance(execution.get("execution_eligible"), bool):
|
|
raise ValueError("execution_eligible must be an explicit boolean")
|
|
if any(
|
|
execution.get(key) is not False
|
|
for key in ("authorized", "transferred", "executed")
|
|
):
|
|
reasons = ["MANIFEST_EXECUTION_STATE_INVALID"]
|
|
else:
|
|
reasons = []
|
|
if not isinstance(artifact_record.get("filename"), str):
|
|
raise ValueError("artifact filename must be a string")
|
|
if (
|
|
not isinstance(artifact_record.get("size"), int)
|
|
or artifact_record["size"] < 1
|
|
):
|
|
raise ValueError("artifact size must be a positive integer")
|
|
|
|
blocked_hashes: set[str] = set()
|
|
entries = denylist["entries"]
|
|
if not isinstance(entries, list) or not entries:
|
|
raise ValueError("denylist entries must be a non-empty array")
|
|
for entry in entries:
|
|
entry_digest = entry["sha256"]
|
|
if not isinstance(entry_digest, str) or SHA256_PATTERN.fullmatch(
|
|
entry_digest
|
|
) is None:
|
|
raise ValueError("denylist contains an invalid SHA-256")
|
|
if (
|
|
entry.get("status") != "BLOCKED"
|
|
or entry.get("permanent") is not True
|
|
or entry.get("execution_eligible") is not False
|
|
):
|
|
raise ValueError("denylist entry is not permanently blocked")
|
|
if entry_digest in blocked_hashes:
|
|
raise ValueError("denylist contains a duplicate SHA-256")
|
|
blocked_hashes.add(entry_digest)
|
|
|
|
if execution["execution_eligible"] is not True:
|
|
reasons.append("MANIFEST_EXECUTION_INELIGIBLE")
|
|
if digest in blocked_hashes:
|
|
reasons.append("ARTIFACT_PERMANENTLY_DENYLISTED")
|
|
|
|
if execution["execution_eligible"] is True:
|
|
if args.runtime_profile is None:
|
|
reasons.append("CONTROLLED_RUNTIME_PROFILE_REQUIRED")
|
|
else:
|
|
profile = load_json(args.runtime_profile)
|
|
reasons.extend(
|
|
validate_runtime_profile(profile, manifest, args.firmware)
|
|
)
|
|
elif args.runtime_profile is not None:
|
|
profile = load_json(args.runtime_profile)
|
|
reasons.extend(validate_runtime_profile(profile, manifest, args.firmware))
|
|
|
|
if args.artifact is not None:
|
|
artifact = args.artifact.resolve(strict=True)
|
|
if not artifact.is_file():
|
|
raise ValueError("artifact is not a regular file")
|
|
if artifact.name != artifact_record["filename"]:
|
|
reasons.append("ARTIFACT_FILENAME_MISMATCH")
|
|
if artifact.stat().st_size != artifact_record["size"]:
|
|
reasons.append("ARTIFACT_SIZE_MISMATCH")
|
|
if hash_file(artifact) != digest:
|
|
reasons.append("ARTIFACT_DIGEST_MISMATCH")
|
|
else:
|
|
reasons.append("ARTIFACT_BYTES_NOT_SUPPLIED")
|
|
|
|
if reasons:
|
|
return deny(reasons, digest)
|
|
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"artifact_sha256": digest,
|
|
"decision": "PASS_STATIC_DEPLOYMENT_ELIGIBILITY_GATE",
|
|
"execution_authorized": False,
|
|
"reason_codes": [],
|
|
"schema_version": 1,
|
|
},
|
|
sort_keys=True,
|
|
)
|
|
)
|
|
return 0
|
|
except (KeyError, OSError, TypeError, ValueError, json.JSONDecodeError):
|
|
return deny(["INVALID_OR_INCOMPLETE_POLICY_INPUT"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|