163 lines
6.9 KiB
Python
163 lines
6.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Validate Nexus One AI specification fixtures; never validates production artifacts."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
|
|
from cryptography.hazmat.primitives import serialization
|
|
from cryptography.exceptions import InvalidSignature
|
|
from jsonschema import Draft202012Validator
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parent
|
|
SCHEMAS = ROOT.parent / "schemas"
|
|
MANIFEST_SCHEMA = json.loads((SCHEMAS / "artifact-manifest.schema.json").read_text())
|
|
SIGNATURE_SCHEMA = json.loads((SCHEMAS / "signature-envelope.schema.json").read_text())
|
|
PROFILE_RECEIPT_SCHEMA = json.loads(
|
|
(SCHEMAS / "profile-membership-verification-receipt.schema.json").read_text()
|
|
)
|
|
|
|
|
|
def load(name: str):
|
|
return json.loads((ROOT / name).read_text())
|
|
|
|
|
|
def canonical(value) -> bytes:
|
|
return (json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n").encode()
|
|
|
|
|
|
def mutate(value, changes):
|
|
result = copy.deepcopy(value)
|
|
for dotted, replacement in changes.items():
|
|
target = result
|
|
parts = dotted.split(".")
|
|
for part in parts[:-1]:
|
|
target = target[part]
|
|
target[parts[-1]] = replacement
|
|
return result
|
|
|
|
|
|
def semver(value: str):
|
|
match = re.fullmatch(r"(\d+)\.(\d+)\.(\d+)(?:-([0-9A-Za-z.-]+))?", value)
|
|
if not match:
|
|
raise AssertionError(f"unsupported semantic version: {value}")
|
|
major, minor, patch = map(int, match.group(1, 2, 3))
|
|
prerelease = match.group(4)
|
|
return major, minor, patch, (0, prerelease) if prerelease is not None else (1, "")
|
|
|
|
|
|
def validate_schemas():
|
|
Draft202012Validator.check_schema(MANIFEST_SCHEMA)
|
|
Draft202012Validator.check_schema(SIGNATURE_SCHEMA)
|
|
Draft202012Validator.check_schema(PROFILE_RECEIPT_SCHEMA)
|
|
manifest_validator = Draft202012Validator(MANIFEST_SCHEMA)
|
|
for name in (
|
|
"workstation-iso-manifest.valid.json",
|
|
"model-pack-manifest.valid.json",
|
|
"update-bundle-manifest.valid.json",
|
|
"trust-bundle-manifest.valid.json",
|
|
):
|
|
errors = sorted(manifest_validator.iter_errors(load(name)), key=lambda error: list(error.path))
|
|
if errors:
|
|
raise AssertionError(f"{name}: {errors[0].message}")
|
|
Draft202012Validator(SIGNATURE_SCHEMA).validate(load("workstation-iso-signature.valid.json"))
|
|
Draft202012Validator(PROFILE_RECEIPT_SCHEMA).validate(
|
|
load("python-profile-membership-receipt.structural-test.json")
|
|
)
|
|
|
|
|
|
def validate_signature_cases():
|
|
manifest = load("workstation-iso-manifest.valid.json")
|
|
envelope = load("workstation-iso-signature.valid.json")
|
|
public_key = serialization.load_pem_public_key((ROOT / "ed25519-test-public.pem").read_bytes())
|
|
cases = load("signature-cases.json")["cases"]
|
|
for case in cases:
|
|
candidate = mutate(manifest, case.get("mutation", {}))
|
|
body = canonical(candidate)
|
|
digest = "sha256:" + hashlib.sha256(body).hexdigest()
|
|
key_id = case.get("key_id", envelope["key_id"])
|
|
if key_id != envelope["key_id"]:
|
|
actual = "KEY_UNKNOWN"
|
|
elif digest != envelope["manifest_digest"]:
|
|
actual = "MANIFEST_DIGEST_MISMATCH"
|
|
else:
|
|
signature = base64.b64decode(case.get("signature", envelope["signature"]), validate=True)
|
|
try:
|
|
public_key.verify(signature, body)
|
|
actual = "VALID"
|
|
except InvalidSignature:
|
|
actual = "SIGNATURE_INVALID"
|
|
assert actual == case["expected"], f"{case['id']}: expected {case['expected']}, got {actual}"
|
|
|
|
|
|
def validate_artifact_signatures():
|
|
public_key = serialization.load_pem_public_key((ROOT / "ed25519-test-public.pem").read_bytes())
|
|
for prefix in ("workstation-iso", "model-pack", "update-bundle", "trust-bundle"):
|
|
manifest = load(f"{prefix}-manifest.valid.json")
|
|
envelope = load(f"{prefix}-signature.valid.json")
|
|
Draft202012Validator(SIGNATURE_SCHEMA).validate(envelope)
|
|
body = canonical(manifest)
|
|
digest = "sha256:" + hashlib.sha256(body).hexdigest()
|
|
assert digest == envelope["manifest_digest"], f"{prefix}: digest mismatch"
|
|
signature = base64.b64decode(envelope["signature"], validate=True)
|
|
public_key.verify(signature, body)
|
|
tampered = copy.deepcopy(manifest)
|
|
tampered["artifact"]["description"] = "TEST ONLY - tampered"
|
|
assert "sha256:" + hashlib.sha256(canonical(tampered)).hexdigest() != envelope["manifest_digest"]
|
|
|
|
|
|
def validate_compatibility_cases():
|
|
manifest = load("workstation-iso-manifest.valid.json")
|
|
suite = load("compatibility-cases.json")
|
|
for case in suite["cases"]:
|
|
candidate = mutate(manifest, case.get("mutation", {}))
|
|
context = mutate(suite["base_context"], case.get("context_mutation", {}))
|
|
trust = case.get("trust_mutation", {})
|
|
targets = candidate["targets"]
|
|
lifecycle = candidate["lifecycle"]
|
|
requirements = candidate["requirements"]
|
|
if trust.get("key_status") == "revoked":
|
|
actual = "KEY_REVOKED"
|
|
elif trust.get("signature_valid") is False:
|
|
actual = "SIGNATURE_INVALID"
|
|
elif trust.get("object_digests_valid") is False:
|
|
actual = "OBJECT_DIGEST_MISMATCH"
|
|
elif candidate["artifact"]["architecture"] not in ("multi", context["architecture"]):
|
|
actual = "ARCHITECTURE_INCOMPATIBLE"
|
|
elif context["category"] not in targets["categories"]:
|
|
actual = "CATEGORY_INCOMPATIBLE"
|
|
elif context["tier"] not in targets["tiers"]:
|
|
actual = "TIER_INCOMPATIBLE"
|
|
elif semver(lifecycle.get("to_version", context["platform_version"])) < semver(context["security_floor"]):
|
|
actual = "SECURITY_FLOOR_VIOLATION"
|
|
elif context["ubuntu_release"] != targets["ubuntu_release"]:
|
|
actual = "UBUNTU_RELEASE_INCOMPATIBLE"
|
|
elif targets.get("hardware_profiles") and context["hardware_profile"] not in targets["hardware_profiles"]:
|
|
actual = "HARDWARE_PROFILE_INCOMPATIBLE"
|
|
elif not set(requirements.get("required_capabilities", [])).issubset(context["capabilities"]):
|
|
actual = "CAPABILITY_MISSING"
|
|
elif context["available_bytes"] < requirements["required_free_bytes"]:
|
|
actual = "SPACE_INSUFFICIENT"
|
|
elif lifecycle["rollback_class"] != "none" and not context["rollback_prerequisite_verified"]:
|
|
actual = "ROLLBACK_PREREQUISITE_MISSING"
|
|
else:
|
|
actual = "COMPATIBLE"
|
|
assert actual == case["expected"], f"{case['id']}: expected {case['expected']}, got {actual}"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
validate_schemas()
|
|
validate_artifact_signatures()
|
|
validate_signature_cases()
|
|
validate_compatibility_cases()
|
|
print("Schema fixtures: PASS (Draft 2020-12)")
|
|
print("Artifact signatures: PASS (4/4)")
|
|
print("Signature negative cases: PASS (4/4)")
|
|
print("Compatibility cases: PASS (12/12)")
|