feat: add immutable deployment bundle contract

Co-Authored-By: Codex <noreply@openai.com>

Author: Pigbibi

python/scripts/deployment_bundle_contract.py

@@ -0,0 +1,224 @@
+#!/usr/bin/env python3
+"""Validate the local-only immutable QSL DeploymentBundle v1 contract."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import math
+import re
+import sys
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Mapping
+
+SCHEMA_ID = "qsl.deployment_bundle.v1"
+_DIGEST_ALGORITHM = "sha256"
+_IDENTITY_PATTERN = re.compile(r"^[a-z][a-z0-9]*(?:[._-][a-z0-9]+)*$")
+_REVISION_PATTERN = re.compile(r"^[0-9a-f]{40}$")
+_SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$")
+_TIMESTAMP_PATTERN = re.compile(r"^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z$")
+_FORBIDDEN_KEY_PATTERN = re.compile(
+    r"credential|secret|token|password|cookie|jwt|private|access[_-]?key|broker|account|order|capital|"
+    r"activation|apply|runtime|configured|live[_-]?ready|promotion|matched|fill",
+    re.IGNORECASE,
+)
+_URL_PATTERN = re.compile(r"[a-z][a-z0-9+.-]*://", re.IGNORECASE)
+_REQUIRED_FIELDS = {
+    "schema",
+    "bundle_id",
+    "created_at",
+    "digest_algorithm",
+    "strategy",
+    "profile",
+    "config",
+    "evidence",
+    "target",
+    "dependencies",
+    "bundle_sha256",
+}
+_REQUIRED_ARTIFACT_FIELDS = {"id", "revision", "artifact_sha256"}
+
+
+class BundleValidationError(ValueError):
+    """Raised when an input is not a valid immutable deployment bundle."""
+
+
+def _fail(message: str) -> None:
+    raise BundleValidationError(message)
+
+
+def _reject_non_finite_or_null(value: Any, path: str = "bundle") -> None:
+    if value is None:
+        _fail(f"{path} must not be null")
+    if isinstance(value, float) and not math.isfinite(value):
+        _fail(f"{path} contains a non-finite number")
+    if isinstance(value, Mapping):
+        for key, child in value.items():
+            if not isinstance(key, str):
+                _fail(f"{path} contains a non-string key")
+            _reject_non_finite_or_null(child, f"{path}.{key}")
+    elif isinstance(value, list):
+        for index, child in enumerate(value):
+            _reject_non_finite_or_null(child, f"{path}[{index}]")
+
+
+def _reject_forbidden_material(value: Any, path: str = "bundle") -> None:
+    if isinstance(value, Mapping):
+        for key, child in value.items():
+            if _FORBIDDEN_KEY_PATTERN.search(key):
+                _fail(f"{path}.{key} is forbidden in a deployment bundle")
+            _reject_forbidden_material(child, f"{path}.{key}")
+    elif isinstance(value, list):
+        for index, child in enumerate(value):
+            _reject_forbidden_material(child, f"{path}[{index}]")
+    elif isinstance(value, str) and _URL_PATTERN.search(value):
+        _fail(f"{path} contains a forbidden URL")
+
+
+def _expect_object(value: Any, path: str) -> Mapping[str, Any]:
+    if not isinstance(value, Mapping):
+        _fail(f"{path} must be an object")
+    return value
+
+
+def _expect_exact_keys(value: Mapping[str, Any], expected: set[str], path: str) -> None:
+    missing = sorted(expected - set(value))
+    unknown = sorted(set(value) - expected)
+    if missing:
+        _fail(f"{path} missing required field(s): {', '.join(missing)}")
+    if unknown:
+        _fail(f"{path} has unknown field(s): {', '.join(unknown)}")
+
+
+def _expect_identity(value: Any, path: str) -> str:
+    if not isinstance(value, str) or not _IDENTITY_PATTERN.fullmatch(value):
+        _fail(f"{path} must be a lowercase immutable identity")
+    return value
+
+
+def _expect_revision(value: Any, path: str) -> str:
+    if not isinstance(value, str) or not _REVISION_PATTERN.fullmatch(value):
+        _fail(f"{path} must be a lowercase 40-character revision")
+    return value
+
+
+def _expect_sha256(value: Any, path: str) -> str:
+    if not isinstance(value, str) or not _SHA256_PATTERN.fullmatch(value):
+        _fail(f"{path} must be a lowercase SHA-256 digest")
+    return value
+
+
+def _expect_timestamp(value: Any) -> str:
+    if not isinstance(value, str) or not _TIMESTAMP_PATTERN.fullmatch(value):
+        _fail("created_at must be an RFC3339 UTC timestamp with whole seconds")
+    try:
+        datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ")
+    except ValueError as exc:
+        raise BundleValidationError("created_at must be a valid calendar timestamp") from exc
+    return value
+
+
+def _validate_artifact_identity(value: Any, path: str, *, strategy: bool = False) -> Mapping[str, Any]:
+    identity = _expect_object(value, path)
+    expected = _REQUIRED_ARTIFACT_FIELDS | ({"source_id"} if strategy else set())
+    _expect_exact_keys(identity, expected, path)
+    _expect_identity(identity["id"], f"{path}.id")
+    if strategy:
+        _expect_identity(identity["source_id"], f"{path}.source_id")
+    _expect_revision(identity["revision"], f"{path}.revision")
+    _expect_sha256(identity["artifact_sha256"], f"{path}.artifact_sha256")
+    return identity
+
+
+def _validate_shape(bundle: Any) -> Mapping[str, Any]:
+    _reject_non_finite_or_null(bundle)
+    _reject_forbidden_material(bundle)
+    root = _expect_object(bundle, "bundle")
+    _expect_exact_keys(root, _REQUIRED_FIELDS, "bundle")
+    if root["schema"] != SCHEMA_ID:
+        _fail(f"schema must be {SCHEMA_ID}")
+    _expect_identity(root["bundle_id"], "bundle_id")
+    _expect_timestamp(root["created_at"])
+    if root["digest_algorithm"] != _DIGEST_ALGORITHM:
+        _fail("digest_algorithm must be sha256")
+    strategy = _validate_artifact_identity(root["strategy"], "strategy", strategy=True)
+    _validate_artifact_identity(root["profile"], "profile")
+    _validate_artifact_identity(root["config"], "config")
+    _validate_artifact_identity(root["evidence"], "evidence")
+    target = _expect_object(root["target"], "target")
+    _expect_exact_keys(target, {"id", "platform_id"}, "target")
+    _expect_identity(target["id"], "target.id")
+    _expect_identity(target["platform_id"], "target.platform_id")
+    dependencies = _expect_object(root["dependencies"], "dependencies")
+    _expect_exact_keys(dependencies, {"qpk", "strategy", "pipeline", "platform"}, "dependencies")
+    for name in ("qpk", "strategy", "pipeline", "platform"):
+        _validate_artifact_identity(dependencies[name], f"dependencies.{name}")
+    if strategy["source_id"] != dependencies["strategy"]["id"]:
+        _fail("strategy.source_id must match dependencies.strategy.id")
+    if strategy["revision"] != dependencies["strategy"]["revision"]:
+        _fail("strategy.revision must match dependencies.strategy.revision")
+    if target["platform_id"] != dependencies["platform"]["id"]:
+        _fail("target.platform_id must match dependencies.platform.id")
+    _expect_sha256(root["bundle_sha256"], "bundle_sha256")
+    return root
+
+
+def canonical_json(bundle: Mapping[str, Any]) -> str:
+    """Return the deterministic JSON representation with only the self hash omitted."""
+    if not isinstance(bundle, Mapping):
+        _fail("bundle must be an object")
+    content = dict(bundle)
+    content.pop("bundle_sha256", None)
+    try:
+        return json.dumps(content, sort_keys=True, separators=(",", ":"), ensure_ascii=True, allow_nan=False)
+    except (TypeError, ValueError) as exc:
+        raise BundleValidationError("bundle cannot be represented as canonical JSON") from exc
+
+
+def calculate_bundle_sha256(bundle: Mapping[str, Any]) -> str:
+    return hashlib.sha256(canonical_json(bundle).encode("utf-8")).hexdigest()
+
+
+def validate_bundle(bundle: Any) -> Mapping[str, Any]:
+    """Fail closed unless the exact immutable content matches its declared digest."""
+    root = _validate_shape(bundle)
+    expected = calculate_bundle_sha256(root)
+    if root["bundle_sha256"] != expected:
+        _fail("bundle_sha256 mismatch")
+    return root
+
+
+def _reject_duplicate_pairs(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
+    result: dict[str, Any] = {}
+    for key, value in pairs:
+        if key in result:
+            _fail(f"duplicate JSON key: {key}")
+        result[key] = value
+    return result
+
+
+def parse_bundle_json(text: str) -> Mapping[str, Any]:
+    try:
+        value = json.loads(text, object_pairs_hook=_reject_duplicate_pairs, parse_constant=lambda _: _fail("non-finite JSON value"))
+    except json.JSONDecodeError as exc:
+        raise BundleValidationError("invalid JSON") from exc
+    return validate_bundle(value)
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--input", type=Path, required=True, help="immutable bundle JSON to validate locally")
+    args = parser.parse_args(argv)
+    try:
+        bundle = parse_bundle_json(args.input.read_text(encoding="utf-8"))
+    except (OSError, BundleValidationError) as exc:
+        print(f"deployment bundle validation failed: {exc}", file=sys.stderr)
+        return 1
+    print(json.dumps({"bundle_sha256": bundle["bundle_sha256"], "schema": bundle["schema"]}, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())