diff --git a/python/scripts/activation_contract.py b/python/scripts/activation_contract.py new file mode 100644 index 0000000..c86b477 --- /dev/null +++ b/python/scripts/activation_contract.py @@ -0,0 +1,365 @@ +#!/usr/bin/env python3 +"""Validate the local-only QSL Activation v1 contract.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import math +import re +import sys +from datetime import UTC, datetime +from pathlib import Path +from typing import Any, Mapping + +from deployment_bundle_contract import BundleValidationError, parse_bundle_json, validate_bundle + +SCHEMA_ID = "qsl.activation.v1" +BUNDLE_SCHEMA_ID = "qsl.deployment_bundle.v1" +STAGES = ("DISABLED", "PAPER_DRY_RUN", "SHADOW", "LIMITED_LIVE", "FULL_LIVE") +_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$") +_REPOSITORY_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]*/[A-Za-z0-9][A-Za-z0-9_.-]*$") +_FORBIDDEN_KEY_PATTERN = re.compile( + r"credential|secret|token|password|cookie|jwt|private(?:[_-]?key)?|access[_-]?key|" + r"broker|order|capital|fill|matched|runtime[_-]?active|config[_-]?applied|applied", + re.IGNORECASE, +) +_URL_PATTERN = re.compile(r"[a-z][a-z0-9+.-]*://", re.IGNORECASE) +_REQUIRED_FIELDS = { + "schema", + "activation_id", + "created_at", + "digest_algorithm", + "contract_only", + "deployment_bundle", + "stage", + "effective_at", + "expires_at", + "human_authority", + "target", + "activation_sha256", +} +_BUNDLE_REFERENCE_FIELDS = {"schema", "bundle_id", "bundle_sha256"} +_AUTHORITY_FIELDS = {"stage", "authority_id", "authority_version", "authority_receipt_sha256"} +_TARGET_FIELDS = { + "platform", + "repository", + "revision", + "environment", + "account_alias", + "account_digest_sha256", +} + + +class ActivationValidationError(ValueError): + """Raised when an input is not a valid contract-only Activation.""" + + +def _fail(message: str) -> None: + raise ActivationValidationError(message) + + +def _reject_non_finite_or_null(value: Any, path: str = "activation") -> 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 = "activation") -> None: + if isinstance(value, Mapping): + for key, child in value.items(): + if _FORBIDDEN_KEY_PATTERN.search(key): + _fail(f"{path}.{key} is forbidden in an activation contract") + _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, *, allow_numeric_only: bool = True) -> str: + if not isinstance(value, str) or not _IDENTITY_PATTERN.fullmatch(value): + _fail(f"{path} must be a lowercase immutable identity") + if not allow_numeric_only and value.isdigit(): + _fail(f"{path} must be a non-sensitive alias, not a numeric account identifier") + 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 _parse_timestamp(value: Any, path: str) -> datetime: + if not isinstance(value, str) or not _TIMESTAMP_PATTERN.fullmatch(value): + _fail(f"{path} must be an RFC3339 UTC timestamp with whole seconds") + try: + parsed = datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ") + except ValueError as exc: + raise ActivationValidationError(f"{path} must be a valid calendar timestamp") from exc + return parsed.replace(tzinfo=UTC) + + +def _validate_bundle_reference(value: Any) -> Mapping[str, Any]: + reference = _expect_object(value, "deployment_bundle") + _expect_exact_keys(reference, _BUNDLE_REFERENCE_FIELDS, "deployment_bundle") + if reference["schema"] != BUNDLE_SCHEMA_ID: + _fail(f"deployment_bundle.schema must be {BUNDLE_SCHEMA_ID}") + _expect_identity(reference["bundle_id"], "deployment_bundle.bundle_id") + _expect_sha256(reference["bundle_sha256"], "deployment_bundle.bundle_sha256") + return reference + + +def _validate_authority(value: Any, path: str = "human_authority") -> Mapping[str, Any]: + authority = _expect_object(value, path) + _expect_exact_keys(authority, _AUTHORITY_FIELDS, path) + if authority["stage"] not in STAGES: + _fail(f"{path}.stage must be a canonical activation stage") + _expect_identity(authority["authority_id"], f"{path}.authority_id") + _expect_identity(authority["authority_version"], f"{path}.authority_version") + _expect_sha256(authority["authority_receipt_sha256"], f"{path}.authority_receipt_sha256") + return authority + + +def _validate_target(value: Any) -> Mapping[str, Any]: + target = _expect_object(value, "target") + _expect_exact_keys(target, _TARGET_FIELDS, "target") + _expect_identity(target["platform"], "target.platform") + if not isinstance(target["repository"], str) or not _REPOSITORY_PATTERN.fullmatch(target["repository"]): + _fail("target.repository must be an owner/repository identity, not a URL") + _expect_revision(target["revision"], "target.revision") + _expect_identity(target["environment"], "target.environment") + _expect_identity(target["account_alias"], "target.account_alias", allow_numeric_only=False) + _expect_sha256(target["account_digest_sha256"], "target.account_digest_sha256") + return target + + +def _validate_shape(activation: Any) -> tuple[Mapping[str, Any], datetime, datetime, datetime]: + _reject_non_finite_or_null(activation) + _reject_forbidden_material(activation) + root = _expect_object(activation, "activation") + _expect_exact_keys(root, _REQUIRED_FIELDS, "activation") + if root["schema"] != SCHEMA_ID: + _fail(f"schema must be {SCHEMA_ID}") + _expect_identity(root["activation_id"], "activation_id") + created_at = _parse_timestamp(root["created_at"], "created_at") + if root["digest_algorithm"] != _DIGEST_ALGORITHM: + _fail("digest_algorithm must be sha256") + if root["contract_only"] is not True: + _fail("contract_only must be true") + _validate_bundle_reference(root["deployment_bundle"]) + if root["stage"] not in STAGES: + _fail("stage must be a canonical activation stage") + effective_at = _parse_timestamp(root["effective_at"], "effective_at") + expires_at = _parse_timestamp(root["expires_at"], "expires_at") + if created_at > effective_at: + _fail("created_at must not be after effective_at") + if expires_at <= effective_at: + _fail("expires_at must be after effective_at") + authority = _validate_authority(root["human_authority"]) + if authority["stage"] != root["stage"]: + _fail("human authority stage must exactly match activation stage") + _validate_target(root["target"]) + _expect_sha256(root["activation_sha256"], "activation_sha256") + return root, created_at, effective_at, expires_at + + +def canonical_json(activation: Mapping[str, Any]) -> str: + """Return deterministic JSON with only the Activation self hash omitted.""" + if not isinstance(activation, Mapping): + _fail("activation must be an object") + content = dict(activation) + content.pop("activation_sha256", None) + try: + return json.dumps(content, sort_keys=True, separators=(",", ":"), ensure_ascii=True, allow_nan=False) + except (TypeError, ValueError) as exc: + raise ActivationValidationError("activation cannot be represented as canonical JSON") from exc + + +def calculate_activation_sha256(activation: Mapping[str, Any]) -> str: + return hashlib.sha256(canonical_json(activation).encode("utf-8")).hexdigest() + + +def _validate_expected_bundle(root: Mapping[str, Any], expected_bundle: Any) -> None: + if expected_bundle is None: + _fail("expected qsl.deployment_bundle.v1 is required") + try: + bundle = validate_bundle(expected_bundle) + except BundleValidationError as exc: + raise ActivationValidationError(f"expected bundle is invalid: {exc}") from exc + reference = root["deployment_bundle"] + expected_reference = { + "schema": bundle["schema"], + "bundle_id": bundle["bundle_id"], + "bundle_sha256": bundle["bundle_sha256"], + } + if reference != expected_reference: + _fail("deployment bundle reference does not match the exact expected bundle identity") + target = root["target"] + if target["platform"] != bundle["target"]["platform_id"]: + _fail("activation target does not match deployment bundle target") + if target["revision"] != bundle["dependencies"]["platform"]["revision"]: + _fail("activation platform revision does not match deployment bundle platform revision") + + +def _validate_expected_authority(root: Mapping[str, Any], expected_authority: Any) -> None: + if expected_authority is None: + return + authority = _validate_authority(expected_authority, "expected_authority") + if root["human_authority"] != authority: + _fail("human authority reference does not match the exact expected authority") + + +def _validate_previous_activation(root: Mapping[str, Any], previous_activation: Any) -> None: + if previous_activation is None: + return + previous, _, _, _ = _validate_shape(previous_activation) + if previous["activation_sha256"] != calculate_activation_sha256(previous): + _fail("previous activation_sha256 mismatch") + if previous["stage"] == root["stage"]: + return + previous_authority = previous["human_authority"] + current_authority = root["human_authority"] + reused = any( + previous_authority[field] == current_authority[field] + for field in ("authority_id", "authority_receipt_sha256") + ) + if reused: + _fail("cross-stage authority reuse or upgrade is forbidden") + + +def validate_activation( + activation: Any, + *, + as_of: str | None = None, + expected_bundle: Any, + expected_authority: Any = None, + previous_activation: Any = None, +) -> Mapping[str, Any]: + """Fail closed unless identity, authority, target, time window, and digest all match.""" + root, created_at, effective_at, expires_at = _validate_shape(activation) + expected_hash = calculate_activation_sha256(root) + if root["activation_sha256"] != expected_hash: + _fail("activation_sha256 mismatch") + _validate_expected_bundle(root, expected_bundle) + _validate_expected_authority(root, expected_authority) + _validate_previous_activation(root, previous_activation) + observed_at = datetime.now(UTC).replace(microsecond=0) if as_of is None else _parse_timestamp(as_of, "as_of") + if created_at > observed_at: + _fail("activation is stale or invalid because created_at is in the future") + if observed_at < effective_at: + _fail("activation is not yet effective") + if observed_at >= expires_at: + _fail("activation is expired") + 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 _load_json(text: str) -> Any: + try: + return json.loads( + text, + object_pairs_hook=_reject_duplicate_pairs, + parse_constant=lambda _: _fail("non-finite JSON value"), + ) + except json.JSONDecodeError as exc: + raise ActivationValidationError("invalid JSON") from exc + + +def parse_activation_json( + text: str, + *, + as_of: str | None = None, + expected_bundle: Any = None, + expected_authority: Any = None, + previous_activation: Any = None, +) -> Mapping[str, Any]: + value = _load_json(text) + return validate_activation( + value, + as_of=as_of, + expected_bundle=expected_bundle, + expected_authority=expected_authority, + previous_activation=previous_activation, + ) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--input", type=Path, required=True, help="contract-only Activation JSON") + parser.add_argument("--bundle", type=Path, required=True, help="exact DeploymentBundle JSON") + parser.add_argument("--as-of", help="inject canonical UTC validation time; defaults to current UTC") + parser.add_argument("--previous", type=Path, help="optional previous Activation used only to reject cross-stage authority reuse") + args = parser.parse_args(argv) + try: + bundle = parse_bundle_json(args.bundle.read_text(encoding="utf-8")) + previous = _load_json(args.previous.read_text(encoding="utf-8")) if args.previous else None + activation = parse_activation_json( + args.input.read_text(encoding="utf-8"), + as_of=args.as_of, + expected_bundle=bundle, + previous_activation=previous, + ) + except (OSError, ActivationValidationError, BundleValidationError) as exc: + print(f"activation validation failed: {exc}", file=sys.stderr) + return 1 + print( + json.dumps( + { + "activation_sha256": activation["activation_sha256"], + "contract_only": True, + "schema": activation["schema"], + }, + sort_keys=True, + ) + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/python/tests/test_activation_contract.py b/python/tests/test_activation_contract.py new file mode 100644 index 0000000..2aec5d8 --- /dev/null +++ b/python/tests/test_activation_contract.py @@ -0,0 +1,297 @@ +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] +SCRIPTS = ROOT / "scripts" + + +def _load_module(name: str): + spec = importlib.util.spec_from_file_location(name, SCRIPTS / f"{name}.py") + module = importlib.util.module_from_spec(spec) + assert spec.loader is not None + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +deployment_bundle_contract = _load_module("deployment_bundle_contract") +activation_contract = _load_module("activation_contract") + + +class ActivationContractTest(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-05T08:00: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 _activation(self, *, stage: str = "PAPER_DRY_RUN") -> dict[str, object]: + bundle = self._bundle() + activation: dict[str, object] = { + "schema": "qsl.activation.v1", + "activation_id": "activation.soxl-signal.ibkr-us.paper.20260805", + "created_at": "2026-08-05T09:00:00Z", + "digest_algorithm": "sha256", + "contract_only": True, + "deployment_bundle": { + "schema": bundle["schema"], + "bundle_id": bundle["bundle_id"], + "bundle_sha256": bundle["bundle_sha256"], + }, + "stage": stage, + "effective_at": "2026-08-05T10:00:00Z", + "expires_at": "2026-08-05T18:00:00Z", + "human_authority": { + "stage": stage, + "authority_id": f"human-authority.{stage.lower().replace('_', '-')}.20260805", + "authority_version": "v1", + "authority_receipt_sha256": self._sha("c"), + }, + "target": { + "platform": "interactive-brokers", + "repository": "QuantStrategyLab/InteractiveBrokersPlatform", + "revision": self._revision("8"), + "environment": "ibkr-paper", + "account_alias": "ibkr-research", + "account_digest_sha256": self._sha("d"), + }, + } + activation["activation_sha256"] = activation_contract.calculate_activation_sha256(activation) + return activation + + def _validate(self, activation: dict[str, object], **kwargs): + return activation_contract.validate_activation( + activation, + as_of="2026-08-05T12:00:00Z", + expected_bundle=kwargs.pop("expected_bundle", self._bundle()), + **kwargs, + ) + + def test_schema_is_closed_contract_only_and_uses_canonical_stages(self): + schema = json.loads((ROOT.parent / "schemas" / "qsl-activation.v1.schema.json").read_text()) + self.assertEqual(schema["$id"], "qsl.activation.v1") + self.assertFalse(schema["additionalProperties"]) + self.assertEqual(schema["properties"]["contract_only"], {"const": True}) + self.assertEqual( + schema["properties"]["stage"]["enum"], + ["DISABLED", "PAPER_DRY_RUN", "SHADOW", "LIMITED_LIVE", "FULL_LIVE"], + ) + self.assertIn("does not prove", schema["description"]) + for field in ("created_at", "effective_at", "expires_at"): + self.assertEqual(schema["properties"][field]["format"], "date-time") + + def test_valid_activation_binds_bundle_authority_target_and_digest(self): + activation = self._activation() + validated = self._validate(activation) + self.assertEqual(validated["activation_sha256"], activation_contract.calculate_activation_sha256(activation)) + self.assertEqual( + activation_contract.canonical_json(activation), + activation_contract.canonical_json(dict(reversed(activation.items()))), + ) + + def test_every_canonical_stage_requires_matching_human_authority(self): + for stage in ("DISABLED", "PAPER_DRY_RUN", "SHADOW", "LIMITED_LIVE", "FULL_LIVE"): + with self.subTest(stage=stage): + activation = self._activation(stage=stage) + self._validate(activation) + activation["human_authority"]["stage"] = "DISABLED" if stage != "DISABLED" else "SHADOW" + activation["activation_sha256"] = activation_contract.calculate_activation_sha256(activation) + with self.assertRaisesRegex(activation_contract.ActivationValidationError, "authority stage"): + self._validate(activation) + + def test_missing_unknown_and_invalid_contract_only_fields_fail_closed(self): + for mutate, message in ( + (lambda value: value.pop("target"), "missing required field"), + (lambda value: value.pop("human_authority"), "missing required field"), + (lambda value: value.update({"applied": True}), "forbidden"), + (lambda value: value.update({"contract_only": False}), "contract_only"), + ): + with self.subTest(message=message): + activation = self._activation() + mutate(activation) + activation["activation_sha256"] = activation_contract.calculate_activation_sha256(activation) + with self.assertRaisesRegex(activation_contract.ActivationValidationError, message): + self._validate(activation) + + def test_bundle_reference_and_platform_revision_mismatch_fail_closed(self): + for mutate, message in ( + (lambda value: value["deployment_bundle"].update({"bundle_sha256": self._sha("0")}), "bundle reference"), + (lambda value: value["deployment_bundle"].update({"bundle_id": "bundle.other"}), "bundle reference"), + (lambda value: value["target"].update({"platform": "binance-platform"}), "bundle target"), + (lambda value: value["target"].update({"revision": self._revision("0")}), "platform revision"), + ): + with self.subTest(message=message): + activation = self._activation() + mutate(activation) + activation["activation_sha256"] = activation_contract.calculate_activation_sha256(activation) + with self.assertRaisesRegex(activation_contract.ActivationValidationError, message): + self._validate(activation) + + def test_time_window_is_calendar_valid_ordered_current_and_injectable(self): + for field, timestamp in ( + ("created_at", "2026-02-30T09:00:00Z"), + ("effective_at", "2026-08-05T10:00:00+00:00"), + ("expires_at", "2026-13-05T18:00:00Z"), + ): + with self.subTest(field=field): + activation = self._activation() + activation[field] = timestamp + activation["activation_sha256"] = activation_contract.calculate_activation_sha256(activation) + with self.assertRaisesRegex(activation_contract.ActivationValidationError, "timestamp"): + self._validate(activation) + + activation = self._activation() + activation["expires_at"] = activation["effective_at"] + activation["activation_sha256"] = activation_contract.calculate_activation_sha256(activation) + with self.assertRaisesRegex(activation_contract.ActivationValidationError, "after effective_at"): + self._validate(activation) + with self.assertRaisesRegex(activation_contract.ActivationValidationError, "not yet effective"): + activation_contract.validate_activation( + self._activation(), as_of="2026-08-05T09:59:59Z", expected_bundle=self._bundle() + ) + with self.assertRaisesRegex(activation_contract.ActivationValidationError, "expired"): + activation_contract.validate_activation( + self._activation(), as_of="2026-08-05T18:00:00Z", expected_bundle=self._bundle() + ) + + def test_authority_expectation_and_cross_stage_reuse_fail_closed(self): + activation = self._activation() + expected = copy.deepcopy(activation["human_authority"]) + expected["authority_receipt_sha256"] = self._sha("0") + with self.assertRaisesRegex(activation_contract.ActivationValidationError, "authority reference"): + self._validate(activation, expected_authority=expected) + + previous = self._activation(stage="PAPER_DRY_RUN") + current = self._activation(stage="SHADOW") + current["human_authority"] = copy.deepcopy(previous["human_authority"]) + current["human_authority"]["stage"] = "SHADOW" + current["activation_sha256"] = activation_contract.calculate_activation_sha256(current) + with self.assertRaisesRegex(activation_contract.ActivationValidationError, "cross-stage authority reuse"): + self._validate(current, previous_activation=previous) + + fresh = self._activation(stage="SHADOW") + fresh["human_authority"]["authority_receipt_sha256"] = self._sha("e") + fresh["activation_sha256"] = activation_contract.calculate_activation_sha256(fresh) + self._validate(fresh, previous_activation=previous) + + def test_mutation_requires_recomputed_activation_digest(self): + activation = self._activation() + activation["target"]["environment"] = "ibkr-shadow" + with self.assertRaisesRegex(activation_contract.ActivationValidationError, "activation_sha256 mismatch"): + self._validate(activation) + + def test_digest_identity_and_repository_formats_are_strict(self): + for mutate, message in ( + (lambda value: value["target"].update({"account_digest_sha256": self._sha("A")}), "lowercase SHA-256"), + (lambda value: value["target"].update({"revision": "main"}), "40-character revision"), + (lambda value: value["target"].update({"repository": "https://github.com/org/repo"}), "forbidden URL"), + (lambda value: value["target"].update({"account_alias": "12345678"}), "account_alias"), + ): + with self.subTest(message=message): + activation = self._activation() + mutate(activation) + activation["activation_sha256"] = activation_contract.calculate_activation_sha256(activation) + with self.assertRaisesRegex(activation_contract.ActivationValidationError, message): + self._validate(activation) + + def test_secret_order_capital_fill_and_credential_url_material_fail_closed(self): + for key, value in ( + ("token", "not-a-real-token"), + ("password", "not-a-real-password"), + ("order_id", "no-order"), + ("capital", 0), + ("fill", {}), + ("credential_url", "https://user:password@example.invalid"), + ): + with self.subTest(key=key): + activation = self._activation() + activation[key] = value + activation["activation_sha256"] = activation_contract.calculate_activation_sha256(activation) + with self.assertRaisesRegex(activation_contract.ActivationValidationError, "forbidden"): + self._validate(activation) + + def test_non_finite_and_duplicate_json_keys_fail_closed(self): + activation = self._activation() + activation["unknown"] = float("nan") + with self.assertRaisesRegex(activation_contract.ActivationValidationError, "non-finite"): + self._validate(activation) + encoded = json.dumps(self._activation())[:-1] + ',"stage":"SHADOW"}' + with self.assertRaisesRegex(activation_contract.ActivationValidationError, "duplicate JSON key"): + activation_contract.parse_activation_json(encoded, as_of="2026-08-05T12:00:00Z") + + def test_activation_digest_excludes_only_its_own_hash(self): + activation = self._activation() + canonical = activation_contract.canonical_json(activation) + self.assertNotIn("activation_sha256", canonical) + replacement = copy.deepcopy(activation) + replacement["activation_sha256"] = self._sha("0") + self.assertEqual( + activation_contract.calculate_activation_sha256(activation), + activation_contract.calculate_activation_sha256(replacement), + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/schemas/qsl-activation.v1.schema.json b/schemas/qsl-activation.v1.schema.json new file mode 100644 index 0000000..484f66c --- /dev/null +++ b/schemas/qsl-activation.v1.schema.json @@ -0,0 +1,88 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "qsl.activation.v1", + "title": "QSL Activation v1", + "description": "Contract-only binding of a human-authorized stage and target to one exact qsl.deployment_bundle.v1 identity. It creates no authority, performs no apply, and does not prove configuration, runtime activity, observed disablement, reconciliation, orders, fills, or capital state.", + "type": "object", + "additionalProperties": false, + "required": [ + "schema", + "activation_id", + "created_at", + "digest_algorithm", + "contract_only", + "deployment_bundle", + "stage", + "effective_at", + "expires_at", + "human_authority", + "target", + "activation_sha256" + ], + "properties": { + "schema": {"const": "qsl.activation.v1"}, + "activation_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"]}, + "contract_only": {"const": true}, + "deployment_bundle": {"$ref": "#/$defs/deploymentBundleReference"}, + "stage": {"enum": ["DISABLED", "PAPER_DRY_RUN", "SHADOW", "LIMITED_LIVE", "FULL_LIVE"]}, + "effective_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$"}, + "expires_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$"}, + "human_authority": {"$ref": "#/$defs/humanAuthority"}, + "target": {"$ref": "#/$defs/target"}, + "activation_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}$"}, + "stage": { + "enum": ["DISABLED", "PAPER_DRY_RUN", "SHADOW", "LIMITED_LIVE", "FULL_LIVE"] + }, + "deploymentBundleReference": { + "type": "object", + "additionalProperties": false, + "required": ["schema", "bundle_id", "bundle_sha256"], + "properties": { + "schema": {"const": "qsl.deployment_bundle.v1"}, + "bundle_id": {"$ref": "#/$defs/identity"}, + "bundle_sha256": {"$ref": "#/$defs/sha256"} + } + }, + "humanAuthority": { + "type": "object", + "additionalProperties": false, + "required": ["stage", "authority_id", "authority_version", "authority_receipt_sha256"], + "properties": { + "stage": {"$ref": "#/$defs/stage"}, + "authority_id": {"$ref": "#/$defs/identity"}, + "authority_version": {"$ref": "#/$defs/identity"}, + "authority_receipt_sha256": {"$ref": "#/$defs/sha256"} + } + }, + "target": { + "type": "object", + "additionalProperties": false, + "required": [ + "platform", + "repository", + "revision", + "environment", + "account_alias", + "account_digest_sha256" + ], + "properties": { + "platform": {"$ref": "#/$defs/identity"}, + "repository": { + "type": "string", + "pattern": "^[A-Za-z0-9][A-Za-z0-9_.-]*/[A-Za-z0-9][A-Za-z0-9_.-]*$" + }, + "revision": {"$ref": "#/$defs/revision"}, + "environment": {"$ref": "#/$defs/identity"}, + "account_alias": {"$ref": "#/$defs/identity"}, + "account_digest_sha256": {"$ref": "#/$defs/sha256"} + } + } + } +}