diff --git a/python/scripts/deployment_bundle_contract.py b/python/scripts/deployment_bundle_contract.py new file mode 100644 index 0000000..3f61d36 --- /dev/null +++ b/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()) diff --git a/python/tests/test_deployment_bundle_contract.py b/python/tests/test_deployment_bundle_contract.py new file mode 100644 index 0000000..00d5538 --- /dev/null +++ b/python/tests/test_deployment_bundle_contract.py @@ -0,0 +1,183 @@ +from __future__ import annotations + +import copy +import importlib.util +import json +import sys +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +MODULE_PATH = ROOT / "scripts" / "deployment_bundle_contract.py" +MODULE_SPEC = importlib.util.spec_from_file_location("deployment_bundle_contract", MODULE_PATH) +deployment_bundle_contract = importlib.util.module_from_spec(MODULE_SPEC) +assert MODULE_SPEC.loader is not None +sys.modules[MODULE_SPEC.name] = deployment_bundle_contract +MODULE_SPEC.loader.exec_module(deployment_bundle_contract) + + +class DeploymentBundleContractTest(unittest.TestCase): + @staticmethod + def _sha(character: str) -> str: + return character * 64 + + @staticmethod + def _revision(character: str) -> str: + return character * 40 + + def _bundle(self) -> dict[str, object]: + bundle: dict[str, object] = { + "schema": "qsl.deployment_bundle.v1", + "bundle_id": "bundle.soxl-signal.ibkr-us.20260805", + "created_at": "2026-08-05T15:30:00Z", + "digest_algorithm": "sha256", + "strategy": { + "id": "soxl-signal", + "source_id": "us-equity-strategies", + "revision": self._revision("a"), + "artifact_sha256": self._sha("b"), + }, + "profile": { + "id": "research-profile", + "revision": self._revision("c"), + "artifact_sha256": self._sha("d"), + }, + "config": { + "id": "ibkr-us-config", + "revision": self._revision("e"), + "artifact_sha256": self._sha("f"), + }, + "evidence": { + "id": "soxl-evidence", + "revision": self._revision("1"), + "artifact_sha256": self._sha("2"), + }, + "target": {"id": "ibkr-us", "platform_id": "interactive-brokers"}, + "dependencies": { + "qpk": { + "id": "quant-platform-kit", + "revision": self._revision("3"), + "artifact_sha256": self._sha("4"), + }, + "strategy": { + "id": "us-equity-strategies", + "revision": self._revision("a"), + "artifact_sha256": self._sha("5"), + }, + "pipeline": { + "id": "crypto-live-pool-pipelines", + "revision": self._revision("6"), + "artifact_sha256": self._sha("7"), + }, + "platform": { + "id": "interactive-brokers", + "revision": self._revision("8"), + "artifact_sha256": self._sha("9"), + }, + }, + } + bundle["bundle_sha256"] = deployment_bundle_contract.calculate_bundle_sha256(bundle) + return bundle + + def test_schema_is_closed_and_declares_utc_rfc3339_timestamp(self): + schema = json.loads((ROOT.parent / "schemas" / "qsl-deployment-bundle.v1.schema.json").read_text()) + self.assertEqual(schema["$id"], "qsl.deployment_bundle.v1") + self.assertFalse(schema["additionalProperties"]) + self.assertEqual(schema["properties"]["created_at"]["format"], "date-time") + self.assertEqual(schema["properties"]["created_at"]["pattern"], "^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z$") + self.assertFalse(schema["properties"]["dependencies"]["additionalProperties"]) + self.assertEqual(set(schema["properties"]["dependencies"]["required"]), {"qpk", "strategy", "pipeline", "platform"}) + + def test_valid_bundle_has_deterministic_canonical_digest(self): + bundle = self._bundle() + validated = deployment_bundle_contract.validate_bundle(bundle) + self.assertEqual(validated["bundle_sha256"], deployment_bundle_contract.calculate_bundle_sha256(bundle)) + self.assertEqual( + deployment_bundle_contract.canonical_json(bundle), + deployment_bundle_contract.canonical_json(dict(reversed(bundle.items()))), + ) + + def test_mutation_requires_a_recomputed_bundle_digest(self): + bundle = self._bundle() + bundle["config"]["revision"] = self._revision("0") + with self.assertRaisesRegex(deployment_bundle_contract.BundleValidationError, "bundle_sha256 mismatch"): + deployment_bundle_contract.validate_bundle(bundle) + + def test_strategy_revision_must_equal_dependency_before_hash(self): + bundle = self._bundle() + bundle["strategy"]["revision"] = self._revision("0") + with self.assertRaisesRegex(deployment_bundle_contract.BundleValidationError, "strategy.revision"): + deployment_bundle_contract.validate_bundle(bundle) + + def test_stale_or_invalid_identity_fails_closed_even_with_recomputed_digest(self): + bundle = self._bundle() + bundle["strategy"]["source_id"] = "obsolete-strategy-source" + bundle["bundle_sha256"] = deployment_bundle_contract.calculate_bundle_sha256(bundle) + with self.assertRaisesRegex(deployment_bundle_contract.BundleValidationError, "strategy.source_id"): + deployment_bundle_contract.validate_bundle(bundle) + + def test_invalid_calendar_timestamps_fail_closed(self): + for timestamp in ( + "2026-13-01T00:00:00Z", + "2026-04-31T00:00:00Z", + "2025-02-29T00:00:00Z", + "2024-02-30T00:00:00Z", + "2026-01-01T24:00:00Z", + "2026-01-01T23:60:00Z", + "2026-01-01T23:59:60Z", + "2026-01-01T00:00:00+08:00", + ): + with self.subTest(timestamp=timestamp): + bundle = self._bundle() + bundle["created_at"] = timestamp + with self.assertRaisesRegex(deployment_bundle_contract.BundleValidationError, "timestamp"): + deployment_bundle_contract.validate_bundle(bundle) + + def test_unknown_field_and_uppercase_digest_fail_closed(self): + bundle = self._bundle() + bundle["unexpected"] = "value" + bundle["bundle_sha256"] = deployment_bundle_contract.calculate_bundle_sha256(bundle) + with self.assertRaisesRegex(deployment_bundle_contract.BundleValidationError, "unknown field"): + deployment_bundle_contract.validate_bundle(bundle) + bundle = self._bundle() + bundle["evidence"]["artifact_sha256"] = self._sha("A") + bundle["bundle_sha256"] = deployment_bundle_contract.calculate_bundle_sha256(bundle) + with self.assertRaisesRegex(deployment_bundle_contract.BundleValidationError, "lowercase SHA-256"): + deployment_bundle_contract.validate_bundle(bundle) + + def test_secret_and_authority_bearing_data_fail_closed(self): + for key, value in ( + ("token", "not-a-real-token"), + ("activation", "allowed"), + ("artifact_url", "https://user:password@example.invalid/artifact"), + ): + with self.subTest(key=key): + bundle = self._bundle() + bundle[key] = value + bundle["bundle_sha256"] = deployment_bundle_contract.calculate_bundle_sha256(bundle) + with self.assertRaisesRegex(deployment_bundle_contract.BundleValidationError, "forbidden"): + deployment_bundle_contract.validate_bundle(bundle) + + def test_non_finite_values_and_duplicate_json_keys_fail_closed(self): + bundle = self._bundle() + bundle["unknown"] = float("nan") + with self.assertRaisesRegex(deployment_bundle_contract.BundleValidationError, "non-finite"): + deployment_bundle_contract.validate_bundle(bundle) + encoded = json.dumps(self._bundle())[:-1] + ',"bundle_id":"duplicate"}' + with self.assertRaisesRegex(deployment_bundle_contract.BundleValidationError, "duplicate JSON key"): + deployment_bundle_contract.parse_bundle_json(encoded) + + def test_bundle_digest_excludes_only_its_own_hash_field(self): + bundle = self._bundle() + canonical = deployment_bundle_contract.canonical_json(bundle) + self.assertNotIn("bundle_sha256", canonical) + reordered = copy.deepcopy(bundle) + reordered["bundle_sha256"] = "0" * 64 + self.assertEqual( + deployment_bundle_contract.calculate_bundle_sha256(bundle), + deployment_bundle_contract.calculate_bundle_sha256(reordered), + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/schemas/qsl-deployment-bundle.v1.schema.json b/schemas/qsl-deployment-bundle.v1.schema.json new file mode 100644 index 0000000..26e6ac3 --- /dev/null +++ b/schemas/qsl-deployment-bundle.v1.schema.json @@ -0,0 +1,79 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "qsl.deployment_bundle.v1", + "title": "QSL Deployment Bundle v1", + "description": "Immutable desired deployment identity. This contract does not authorize activation, apply, or runtime behavior.", + "type": "object", + "additionalProperties": false, + "required": [ + "schema", + "bundle_id", + "created_at", + "digest_algorithm", + "strategy", + "profile", + "config", + "evidence", + "target", + "dependencies", + "bundle_sha256" + ], + "properties": { + "schema": {"const": "qsl.deployment_bundle.v1"}, + "bundle_id": {"$ref": "#/$defs/identity"}, + "created_at": {"type": "string", "format": "date-time", "pattern": "^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z$"}, + "digest_algorithm": {"enum": ["sha256"]}, + "strategy": {"$ref": "#/$defs/strategyIdentity"}, + "profile": {"$ref": "#/$defs/artifactIdentity"}, + "config": {"$ref": "#/$defs/artifactIdentity"}, + "evidence": {"$ref": "#/$defs/artifactIdentity"}, + "target": {"$ref": "#/$defs/targetIdentity"}, + "dependencies": { + "type": "object", + "additionalProperties": false, + "required": ["qpk", "strategy", "pipeline", "platform"], + "properties": { + "qpk": {"$ref": "#/$defs/artifactIdentity"}, + "strategy": {"$ref": "#/$defs/artifactIdentity"}, + "pipeline": {"$ref": "#/$defs/artifactIdentity"}, + "platform": {"$ref": "#/$defs/artifactIdentity"} + } + }, + "bundle_sha256": {"$ref": "#/$defs/sha256"} + }, + "$defs": { + "identity": {"type": "string", "pattern": "^[a-z][a-z0-9]*(?:[._-][a-z0-9]+)*$"}, + "revision": {"type": "string", "pattern": "^[0-9a-f]{40}$"}, + "sha256": {"type": "string", "pattern": "^[0-9a-f]{64}$"}, + "artifactIdentity": { + "type": "object", + "additionalProperties": false, + "required": ["id", "revision", "artifact_sha256"], + "properties": { + "id": {"$ref": "#/$defs/identity"}, + "revision": {"$ref": "#/$defs/revision"}, + "artifact_sha256": {"$ref": "#/$defs/sha256"} + } + }, + "strategyIdentity": { + "type": "object", + "additionalProperties": false, + "required": ["id", "source_id", "revision", "artifact_sha256"], + "properties": { + "id": {"$ref": "#/$defs/identity"}, + "source_id": {"$ref": "#/$defs/identity"}, + "revision": {"$ref": "#/$defs/revision"}, + "artifact_sha256": {"$ref": "#/$defs/sha256"} + } + }, + "targetIdentity": { + "type": "object", + "additionalProperties": false, + "required": ["id", "platform_id"], + "properties": { + "id": {"$ref": "#/$defs/identity"}, + "platform_id": {"$ref": "#/$defs/identity"} + } + } + } +}