From 742169159987fe85c70ffa5fc9a44c240ce20cbb Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 4 Aug 2026 11:41:18 +0800 Subject: [PATCH] feat: validate redacted runtime evidence aggregate Co-Authored-By: Codex --- runtime_support.py | 212 ++++++++++++++++++++++++++++++++++ tests/test_runtime_support.py | 75 ++++++++++++ 2 files changed, 287 insertions(+) diff --git a/runtime_support.py b/runtime_support.py index 08450977..93c131ec 100644 --- a/runtime_support.py +++ b/runtime_support.py @@ -1,5 +1,6 @@ import hashlib import os +import re import time from collections.abc import Mapping from dataclasses import dataclass, field @@ -11,6 +12,23 @@ # Binance rate limits (public API: 1200 weight/min, order placement: 50 orders/10s) _BINANCE_ORDER_RATE_LIMIT_INTERVAL_SEC = 0.25 # max ~4 orders/sec _LAST_API_CALL_TS: float = 0.0 +RUNTIME_EVIDENCE_CONTRACT_VERSION = "qsl.runtime_evidence_aggregate.v1" +RECONCILIATION_STATUSES = frozenset({"MISSING", "MATCHED", "MISMATCHED"}) +_RUNTIME_EVIDENCE_FORBIDDEN_FIELDS = frozenset( + { + "api_key", + "api_secret", + "authorization", + "balances", + "credentials", + "headers", + "orders", + "positions", + "provider_rows", + "secret", + "token", + } +) def _rate_limit_pause(): @@ -22,6 +40,200 @@ def _rate_limit_pause(): _LAST_API_CALL_TS = time.monotonic() +def _is_sha256(value: Any) -> bool: + return isinstance(value, str) and bool(re.fullmatch(r"[0-9a-f]{64}", value.strip())) + + +def _is_git_revision(value: Any) -> bool: + return isinstance(value, str) and bool(re.fullmatch(r"[0-9a-f]{40}", value.strip())) + + +def _is_utc_timestamp(value: Any) -> bool: + if not isinstance(value, str) or not value.endswith("Z"): + return False + try: + datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return False + return True + + +def _append_missing_fields(payload: Mapping[str, Any], fields: tuple[str, ...], errors: list[str], label: str) -> None: + for field_name in fields: + if field_name not in payload: + errors.append(f"{label} missing field: {field_name}") + + +def _append_forbidden_field_errors(value: Any, errors: list[str]) -> None: + if isinstance(value, Mapping): + for field, nested_value in value.items(): + if str(field).lower() in _RUNTIME_EVIDENCE_FORBIDDEN_FIELDS: + errors.append(f"runtime_evidence_aggregate contains forbidden field: {field}") + _append_forbidden_field_errors(nested_value, errors) + elif isinstance(value, (list, tuple)): + for item in value: + _append_forbidden_field_errors(item, errors) + + +def _validate_release_identity(identity: Any, errors: list[str]) -> None: + label = "runtime_evidence_aggregate release_identity" + if not isinstance(identity, Mapping): + errors.append(f"{label} must be an object") + return + _append_missing_fields( + identity, + ( + "strategy_profile", + "mode", + "source_revision", + "input_timestamp", + "artifact_contract", + "artifact_version", + "artifacts", + ), + errors, + label, + ) + for field_name in ("strategy_profile", "mode", "artifact_contract", "artifact_version"): + if not isinstance(identity.get(field_name), str) or not identity[field_name].strip(): + errors.append(f"{label} {field_name} must be a non-empty string") + if not _is_git_revision(identity.get("source_revision")): + errors.append(f"{label} source_revision must be a 40-character lowercase git SHA") + if not _is_utc_timestamp(identity.get("input_timestamp")): + errors.append(f"{label} input_timestamp must be a UTC timestamp") + artifacts = identity.get("artifacts") + if not isinstance(artifacts, Mapping) or not artifacts: + errors.append(f"{label} artifacts must be a non-empty object") + return + for artifact_name, artifact in artifacts.items(): + if not isinstance(artifact_name, str) or not artifact_name.strip() or not isinstance(artifact, Mapping): + errors.append(f"{label} artifacts must contain named objects") + continue + if not _is_sha256(artifact.get("sha256")): + errors.append(f"{label} artifacts.{artifact_name}.sha256 must be a SHA-256 digest") + + +def _validate_reconciliation(reconciliation: Any, errors: list[str]) -> None: + label = "runtime_evidence_aggregate reconciliation" + if not isinstance(reconciliation, Mapping): + errors.append(f"{label} must be an object") + return + status = reconciliation.get("status") + if status not in RECONCILIATION_STATUSES: + errors.append(f"{label} status must be one of MISSING, MATCHED, MISMATCHED") + return + if status == "MATCHED": + for field in ("durable_receipt_sha256", "identity_sha256"): + if not _is_sha256(reconciliation.get(field)): + errors.append(f"{label}.MATCHED requires {field}") + errors.append(f"{label}.MATCHED is not valid for static acceptance") + elif status == "MISMATCHED": + for field in ("durable_receipt_sha256", "identity_sha256", "observed_identity_sha256"): + if not _is_sha256(reconciliation.get(field)): + errors.append(f"{label}.MISMATCHED requires {field}") + if reconciliation.get("identity_sha256") == reconciliation.get("observed_identity_sha256"): + errors.append(f"{label}.MISMATCHED identity digests must differ") + + +def validate_runtime_evidence_aggregate(aggregate: Any) -> dict[str, Any]: + """Validate a redacted, static-only runtime evidence aggregate.""" + errors: list[str] = [] + label = "runtime_evidence_aggregate" + if not isinstance(aggregate, Mapping): + return {"ok": False, "errors": [f"{label} must be an object"]} + + _append_forbidden_field_errors(aggregate, errors) + _append_missing_fields( + aggregate, + ( + "contract_version", + "release_identity", + "risk_engine", + "effective_exposure_cap", + "stop_breaker_evaluation", + "reconciliation", + "static_validation_only", + "execution_permitted", + "verified_active", + "fills_verified", + "capital_use_verified", + ), + errors, + label, + ) + if aggregate.get("contract_version") != RUNTIME_EVIDENCE_CONTRACT_VERSION: + errors.append(f"{label} contract_version must be {RUNTIME_EVIDENCE_CONTRACT_VERSION}") + _validate_release_identity(aggregate.get("release_identity"), errors) + + risk_engine = aggregate.get("risk_engine") + if not isinstance(risk_engine, Mapping): + errors.append(f"{label} risk_engine must be an object") + else: + if risk_engine.get("outcome") != "APPROVE": + errors.append(f"{label} risk_engine.outcome must be APPROVE") + if not isinstance(risk_engine.get("policy_version"), str) or not risk_engine["policy_version"].strip(): + errors.append(f"{label} risk_engine.policy_version must be a non-empty string") + + cap = aggregate.get("effective_exposure_cap") + if not isinstance(cap, Mapping): + errors.append(f"{label} effective_exposure_cap must be an object") + else: + value = cap.get("value") + if isinstance(value, bool) or not isinstance(value, (int, float)) or not 0 < value <= 1: + errors.append(f"{label} effective_exposure_cap.value must be in (0, 1]") + for field in ("mandate_version", "source"): + if not isinstance(cap.get(field), str) or not cap[field].strip(): + errors.append(f"{label} effective_exposure_cap.{field} must be a non-empty string") + + stop_breaker = aggregate.get("stop_breaker_evaluation") + if not isinstance(stop_breaker, Mapping): + errors.append(f"{label} stop_breaker_evaluation must be an object") + else: + if stop_breaker.get("stop_evaluated") is not True: + errors.append(f"{label} stop_breaker_evaluation.stop_evaluated must be true") + if stop_breaker.get("breaker_evaluated") is not True: + errors.append(f"{label} stop_breaker_evaluation.breaker_evaluated must be true") + if stop_breaker.get("outcome") != "CLEAR": + errors.append(f"{label} stop_breaker_evaluation.outcome must be CLEAR") + if not isinstance(stop_breaker.get("policy_version"), str) or not stop_breaker["policy_version"].strip(): + errors.append(f"{label} stop_breaker_evaluation.policy_version must be a non-empty string") + + _validate_reconciliation(aggregate.get("reconciliation"), errors) + for field in ("static_validation_only", "execution_permitted", "verified_active", "fills_verified", "capital_use_verified"): + expected = field == "static_validation_only" + if aggregate.get(field) is not expected: + errors.append(f"{label} {field} must be {str(expected).lower()} for static acceptance") + return {"ok": not errors, "errors": errors} + + +def build_runtime_evidence_aggregate( + *, + release_identity: Mapping[str, Any], + risk_engine: Mapping[str, Any], + effective_exposure_cap: Mapping[str, Any], + stop_breaker_evaluation: Mapping[str, Any], + reconciliation: Mapping[str, Any], +) -> dict[str, Any]: + """Build a fail-closed aggregate that cannot claim runtime activity.""" + aggregate = { + "contract_version": RUNTIME_EVIDENCE_CONTRACT_VERSION, + "release_identity": dict(release_identity), + "risk_engine": dict(risk_engine), + "effective_exposure_cap": dict(effective_exposure_cap), + "stop_breaker_evaluation": dict(stop_breaker_evaluation), + "reconciliation": dict(reconciliation), + "static_validation_only": True, + "execution_permitted": False, + "verified_active": False, + "fills_verified": False, + "capital_use_verified": False, + } + validation = validate_runtime_evidence_aggregate(aggregate) + if not validation["ok"]: + raise ValueError("Runtime evidence aggregate validation failed: " + "; ".join(validation["errors"])) + return aggregate + + @dataclass class ExecutionRuntime: dry_run: bool = False diff --git a/tests/test_runtime_support.py b/tests/test_runtime_support.py index ef6deb0e..3f96a9bf 100644 --- a/tests/test_runtime_support.py +++ b/tests/test_runtime_support.py @@ -14,15 +14,90 @@ from runtime_support import ( ExecutionRuntime, + build_runtime_evidence_aggregate, build_execution_report, finalize_notification_delivery, record_gating_event, runtime_notify, + validate_runtime_evidence_aggregate, ) from quant_platform_kit.common.runtime_target import build_runtime_target class TestBuildExecutionReport(unittest.TestCase): + @staticmethod + def runtime_evidence_inputs(): + return { + "release_identity": { + "strategy_profile": "crypto_live_pool_rotation", + "mode": "core_major", + "source_revision": "a" * 40, + "input_timestamp": "2026-03-13T00:00:00Z", + "artifact_contract": "crypto_live_pool_rotation.live_pool.v1", + "artifact_version": "2026-03-13-core_major", + "artifacts": {"live_pool": {"sha256": "b" * 64}}, + }, + "risk_engine": {"outcome": "APPROVE", "policy_version": "bootstrap_small_account_v2"}, + "effective_exposure_cap": { + "value": 0.5, + "mandate_version": "bootstrap_small_account_v2", + "source": "approved_risk_mandate", + }, + "stop_breaker_evaluation": { + "stop_evaluated": True, + "breaker_evaluated": True, + "outcome": "CLEAR", + "policy_version": "bootstrap_small_account_v2", + }, + "reconciliation": {"status": "MISSING"}, + } + + def test_runtime_evidence_aggregate_is_redacted_and_static_only(self): + aggregate = build_runtime_evidence_aggregate(**self.runtime_evidence_inputs()) + + self.assertTrue(validate_runtime_evidence_aggregate(aggregate)["ok"]) + self.assertFalse(aggregate["verified_active"]) + self.assertFalse(aggregate["fills_verified"]) + self.assertFalse(aggregate["capital_use_verified"]) + self.assertNotIn("orders", str(aggregate)) + + def test_runtime_evidence_aggregate_fails_closed_for_risk_reconciliation_and_sensitive_fields(self): + aggregate = build_runtime_evidence_aggregate(**self.runtime_evidence_inputs()) + aggregate["risk_engine"]["outcome"] = "REJECT" + aggregate["reconciliation"] = {"status": "MATCHED"} + aggregate["positions"] = [{"symbol": "BTCUSDT"}] + aggregate["release_identity"]["headers"] = {"authorization": "redacted"} + + validation = validate_runtime_evidence_aggregate(aggregate) + + self.assertFalse(validation["ok"]) + self.assertIn("runtime_evidence_aggregate risk_engine.outcome must be APPROVE", validation["errors"]) + self.assertIn( + "runtime_evidence_aggregate reconciliation.MATCHED requires durable_receipt_sha256", + validation["errors"], + ) + self.assertIn("runtime_evidence_aggregate contains forbidden field: positions", validation["errors"]) + self.assertIn("runtime_evidence_aggregate contains forbidden field: headers", validation["errors"]) + + def test_runtime_evidence_aggregate_rejects_static_matched_reconciliation(self): + matched_inputs = self.runtime_evidence_inputs() + matched_inputs["reconciliation"] = { + "status": "MATCHED", + "durable_receipt_sha256": "c" * 64, + "identity_sha256": "d" * 64, + } + mismatched_inputs = self.runtime_evidence_inputs() + mismatched_inputs["reconciliation"] = { + "status": "MISMATCHED", + "durable_receipt_sha256": "c" * 64, + "identity_sha256": "d" * 64, + "observed_identity_sha256": "e" * 64, + } + + with self.assertRaisesRegex(ValueError, "MATCHED is not valid for static acceptance"): + build_runtime_evidence_aggregate(**matched_inputs) + self.assertTrue(validate_runtime_evidence_aggregate(build_runtime_evidence_aggregate(**mismatched_inputs))["ok"]) + def test_report_contains_enrichment_fields(self): runtime = ExecutionRuntime(dry_run=True, run_id="test-001") report = build_execution_report(runtime)