From 812cb4d740e815e08abaa0167aeac3079dc79fcd Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 29 Aug 2026 15:06:30 +0800 Subject: [PATCH] feat: build hash-bound M0 publisher envelopes Co-Authored-By: Codex --- ...m0_research_publisher_envelope_contract.md | 79 ++++ .../build_m0_research_publisher_envelope.py | 418 ++++++++++++++++++ ...st_build_m0_research_publisher_envelope.py | 232 ++++++++++ ...research-publisher-envelope.v1.schema.json | 46 ++ 4 files changed, 775 insertions(+) create mode 100644 docs/m0_research_publisher_envelope_contract.md create mode 100644 python/scripts/build_m0_research_publisher_envelope.py create mode 100644 python/tests/test_build_m0_research_publisher_envelope.py create mode 100644 schemas/qsl-m0-research-publisher-envelope.v1.schema.json diff --git a/docs/m0_research_publisher_envelope_contract.md b/docs/m0_research_publisher_envelope_contract.md new file mode 100644 index 0000000..d3fd753 --- /dev/null +++ b/docs/m0_research_publisher_envelope_contract.md @@ -0,0 +1,79 @@ +# M0 研究发布封套 v1 + +`python/scripts/build_m0_research_publisher_envelope.py` 是 M0 研究台账的 +**离线**发布构建器。它只接受一个已闭合的 +`qsl_m0_research_source_snapshot.v1` 文件，并调用本仓 +`m0_research_ledger` validator/aggregator 生成 +`qsl_m0_research_ledger.v1`。它不导入 selector、平台配置、调度器、券商、 +策略开关或执行组件。 + +输出为严格的 `qsl_m0_research_publisher_envelope.v1`： + +```json +{ + "schema_version": "qsl_m0_research_publisher_envelope.v1", + "producer": { + "repository": "QuantStrategyLab/QuantRuntimeSettings", + "revision": "<40-char immutable git revision>" + }, + "source_artifact": { + "repository": "QuantStrategyLab/QuantAdvisorResearch", + "revision": "<40-char immutable git revision>", + "run_id": "", + "artifact_id": "", + "sha256": "" + }, + "ledger_sha256": "", + "ledger": { "...": "qsl_m0_research_ledger.v1" } +} +``` + +`producer` 和 `source_artifact` 只能包含以上字段；不能附带 token、账户、 +平台、策略、权重、仓位、订单或运行时目标。`ledger_sha256` 使用 UTF-8 的 +canonical JSON（键排序、紧凑分隔符、禁止 NaN）计算。台账的 +`generated_at` 和 `computed_at` 必须相同，均为精确到秒的 UTC `Z` 时间戳。 +因此同一 source artifact、metadata 和 `--now` 总会生成字节相同的封套。 + +## 默认离线构建 + +```bash +python3 python/scripts/build_m0_research_publisher_envelope.py \ + --source-snapshot /safe/input/m0-source.json \ + --output /safe/output/m0-envelope.json \ + --source-artifact-repository QuantStrategyLab/QuantAdvisorResearch \ + --source-artifact-revision 0123456789abcdef0123456789abcdef01234567 \ + --source-artifact-run-id 123456789 \ + --source-artifact-id m0-source-snapshot \ + --source-artifact-sha256 "$(sha256sum /safe/input/m0-source.json | awk '{print $1}')" \ + --producer-repository QuantStrategyLab/QuantRuntimeSettings \ + --producer-revision 89abcdef0123456789abcdef0123456789abcdef \ + --now 2026-08-29T12:00:00Z +``` + +默认模式只写 `--output` 指定的本地 JSON；没有网络调用，也不会读取任何 +environment variable。输入源文件限制为 2 MiB、拒绝重复 JSON key，且其原始 +字节 SHA-256 必须与显式 `--source-artifact-sha256` 一致。metadata 中的 revision +均要求 40 位小写 git SHA。上述 shell 中的 `sha256sum` 只是操作员生成显式 +metadata 的便利方式，构建器不会执行 shell 或命令替换。 + +## 明确选择的发布 + +发布不是默认行为。只有传入 `--publish` **且**同时存在两个专用环境变量时， +构建器才会在成功写入本地封套后对 HTTPS endpoint 进行一次 POST： + +```bash +export QSL_M0_RESEARCH_LEDGER_PUBLISH_URL='https://research-console.example/api/internal/m0' +export QSL_M0_RESEARCH_LEDGER_PUBLISH_TOKEN='dedicated-publisher-token' + +python3 python/scripts/build_m0_research_publisher_envelope.py ... --publish +``` + +URL 必须是无用户名、无密码、无 query、无 fragment 的 HTTPS URL。token 只从 +`QSL_M0_RESEARCH_LEDGER_PUBLISH_TOKEN` 读取，作为 HTTP `Authorization: Bearer` +header；它从不写进封套、标准输出、错误信息或日志。该工具不接受 token CLI 参数， +也不会读取 broker、平台、策略、运行时或通用控制平面凭据。缺少任一专用环境变量， +或 POST 失败，都会 fail closed。 + +发布 endpoint 只是研究资料接收端：接收者仍必须重验 schema、artifact metadata、 +`ledger_sha256` 和 `ledger.policy` 的 `research_only/no_order` 固定值。接收、展示或 +排队研究任务都不能构成 P4/P5/P6、Shadow、Paper 或 live 授权。 diff --git a/python/scripts/build_m0_research_publisher_envelope.py b/python/scripts/build_m0_research_publisher_envelope.py new file mode 100644 index 0000000..4ce6f0e --- /dev/null +++ b/python/scripts/build_m0_research_publisher_envelope.py @@ -0,0 +1,418 @@ +#!/usr/bin/env python3 +"""Build a hash-bound, research-only M0 publisher envelope. + +This is deliberately an *offline* builder. It accepts one closed +``qsl_m0_research_source_snapshot.v1`` file and explicit immutable artifact +metadata, then asks the local M0 validator/aggregator to produce a ledger. +The result is written as canonical local JSON. A network POST is possible +only when ``--publish`` is set and the two dedicated publish environment +variables are present; this module never discovers or reads broker, strategy, +runtime, or general-purpose credentials. +""" + +from __future__ import annotations + +import argparse +import copy +import datetime as dt +import hashlib +import json +import os +import re +import sys +import urllib.error +import urllib.parse +import urllib.request +from collections.abc import Mapping, Sequence +from pathlib import Path +from typing import Any + +from m0_research_ledger import ( + M0_AUTHORITY, + M0_LEDGER_SCHEMA, + M0_NEXT_STEP, + M0ResearchLedgerValidationError, + aggregate_m0_research_sources, +) + + +PUBLISHER_ENVELOPE_SCHEMA = "qsl_m0_research_publisher_envelope.v1" +PUBLISH_URL_ENV = "QSL_M0_RESEARCH_LEDGER_PUBLISH_URL" +PUBLISH_TOKEN_ENV = "QSL_M0_RESEARCH_LEDGER_PUBLISH_TOKEN" +MAX_SOURCE_SNAPSHOT_BYTES = 2 * 1024 * 1024 + +_REPOSITORY = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]*/[A-Za-z0-9][A-Za-z0-9_.-]*$") +_REVISION = re.compile(r"^[0-9a-f]{40}$") +_SHA256 = re.compile(r"^[0-9a-f]{64}$") +_IDENTIFIER = re.compile(r"^[A-Za-z0-9._:/-]{1,128}$") +_CANONICAL_TIMESTAMP = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$") + + +class M0ResearchPublisherEnvelopeError(ValueError): + """Raised when an envelope cannot be built or safely published.""" + + +def canonical_json(value: object) -> str: + """Return the only JSON representation used for M0 envelope hashes.""" + + try: + return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False) + except (TypeError, ValueError) as exc: + raise M0ResearchPublisherEnvelopeError("m0_envelope_not_canonicalizable") from exc + + +def calculate_ledger_sha256(ledger: Mapping[str, Any]) -> str: + """Hash the complete ledger, never an abbreviated display projection.""" + + if not isinstance(ledger, Mapping): + raise M0ResearchPublisherEnvelopeError("ledger_invalid") + return hashlib.sha256(canonical_json(dict(ledger)).encode("utf-8")).hexdigest() + + +def _exact_mapping(value: object, fields: frozenset[str], label: str) -> dict[str, Any]: + if not isinstance(value, Mapping) or set(value) != fields: + raise M0ResearchPublisherEnvelopeError(f"{label}_keys_invalid") + return dict(value) + + +def _require_repository(value: object, label: str) -> str: + if not isinstance(value, str) or not _REPOSITORY.fullmatch(value): + raise M0ResearchPublisherEnvelopeError(f"{label}_invalid") + return value + + +def _require_revision(value: object, label: str) -> str: + if not isinstance(value, str) or not _REVISION.fullmatch(value): + raise M0ResearchPublisherEnvelopeError(f"{label}_invalid") + return value + + +def _require_sha256(value: object, label: str) -> str: + if not isinstance(value, str) or not _SHA256.fullmatch(value): + raise M0ResearchPublisherEnvelopeError(f"{label}_invalid") + return value + + +def _require_identifier(value: object, label: str) -> str: + if not isinstance(value, str) or not _IDENTIFIER.fullmatch(value): + raise M0ResearchPublisherEnvelopeError(f"{label}_invalid") + return value + + +def canonical_timestamp(value: object, label: str = "now") -> str: + """Accept only UTC whole-second timestamps and return their canonical form.""" + + if not isinstance(value, str) or not _CANONICAL_TIMESTAMP.fullmatch(value): + raise M0ResearchPublisherEnvelopeError(f"{label}_invalid") + try: + parsed = dt.datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError as exc: + raise M0ResearchPublisherEnvelopeError(f"{label}_invalid") from exc + if parsed.tzinfo is None or parsed.utcoffset() != dt.timedelta(0): + raise M0ResearchPublisherEnvelopeError(f"{label}_invalid") + return parsed.replace(microsecond=0).isoformat().replace("+00:00", "Z") + + +def _current_canonical_timestamp() -> str: + return dt.datetime.now(tz=dt.UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z") + + +def _reject_duplicate_keys(pairs: Sequence[tuple[str, Any]]) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + raise ValueError("duplicate JSON key") + result[key] = value + return result + + +def load_source_snapshot(path: Path) -> tuple[dict[str, Any], str]: + """Load one bounded source artifact and return its payload plus byte hash.""" + + try: + raw = path.read_bytes() + except OSError as exc: + raise M0ResearchPublisherEnvelopeError("source_snapshot_unreadable") from exc + if not raw or len(raw) > MAX_SOURCE_SNAPSHOT_BYTES: + raise M0ResearchPublisherEnvelopeError("source_snapshot_size_invalid") + try: + payload = json.loads(raw.decode("utf-8"), object_pairs_hook=_reject_duplicate_keys) + except (UnicodeDecodeError, ValueError, json.JSONDecodeError) as exc: + raise M0ResearchPublisherEnvelopeError("source_snapshot_json_invalid") from exc + if not isinstance(payload, dict): + raise M0ResearchPublisherEnvelopeError("source_snapshot_object_invalid") + return payload, hashlib.sha256(raw).hexdigest() + + +def build_source_artifact_metadata( + *, + repository: str, + revision: str, + run_id: str, + artifact_id: str, + sha256: str, +) -> dict[str, str]: + """Close the provenance of exactly one source snapshot artifact.""" + + return { + "repository": _require_repository(repository, "source_artifact_repository"), + "revision": _require_revision(revision, "source_artifact_revision"), + "run_id": _require_identifier(run_id, "source_artifact_run_id"), + "artifact_id": _require_identifier(artifact_id, "source_artifact_id"), + "sha256": _require_sha256(sha256, "source_artifact_sha256"), + } + + +def _validate_ledger(ledger: object, *, expected_timestamp: str) -> dict[str, Any]: + value = _exact_mapping( + ledger, + frozenset( + { + "schema_version", + "generated_at", + "computed_at", + "data_status", + "summary", + "subjects", + "policy", + "errors", + } + ), + "ledger", + ) + if value["schema_version"] != M0_LEDGER_SCHEMA: + raise M0ResearchPublisherEnvelopeError("ledger_schema_invalid") + if ( + canonical_timestamp(value["generated_at"], "ledger_generated_at") != expected_timestamp + or canonical_timestamp(value["computed_at"], "ledger_computed_at") != expected_timestamp + ): + raise M0ResearchPublisherEnvelopeError("ledger_time_invalid") + policy = _exact_mapping( + value["policy"], + frozenset({"authority", "no_order", "permitted_next_step", "notice"}), + "ledger_policy", + ) + if ( + policy["authority"] != M0_AUTHORITY + or policy["no_order"] is not True + or policy["permitted_next_step"] != M0_NEXT_STEP + ): + raise M0ResearchPublisherEnvelopeError("ledger_policy_invalid") + return copy.deepcopy(value) + + +def build_m0_research_publisher_envelope( + *, + source_snapshot: object, + source_artifact: Mapping[str, Any], + producer_repository: str, + producer_revision: str, + now: str, +) -> dict[str, Any]: + """Aggregate one immutable M0 source artifact into a strict publish envelope.""" + + timestamp = canonical_timestamp(now) + producer = { + "repository": _require_repository(producer_repository, "producer_repository"), + "revision": _require_revision(producer_revision, "producer_revision"), + } + artifact = _exact_mapping( + source_artifact, + frozenset({"repository", "revision", "run_id", "artifact_id", "sha256"}), + "source_artifact", + ) + normalized_artifact = build_source_artifact_metadata(**artifact) + try: + ledger = aggregate_m0_research_sources([source_snapshot], now=timestamp) + except M0ResearchLedgerValidationError as exc: + raise M0ResearchPublisherEnvelopeError("m0_source_snapshot_invalid") from exc + normalized_ledger = _validate_ledger(ledger, expected_timestamp=timestamp) + envelope = { + "schema_version": PUBLISHER_ENVELOPE_SCHEMA, + "producer": producer, + "source_artifact": normalized_artifact, + "ledger_sha256": calculate_ledger_sha256(normalized_ledger), + "ledger": normalized_ledger, + } + return validate_m0_research_publisher_envelope(envelope) + + +def validate_m0_research_publisher_envelope(payload: object) -> dict[str, Any]: + """Verify a strict, non-executable M0 envelope before any optional POST.""" + + envelope = _exact_mapping( + payload, + frozenset({"schema_version", "producer", "source_artifact", "ledger_sha256", "ledger"}), + "publisher_envelope", + ) + if envelope["schema_version"] != PUBLISHER_ENVELOPE_SCHEMA: + raise M0ResearchPublisherEnvelopeError("publisher_envelope_schema_invalid") + producer = _exact_mapping(envelope["producer"], frozenset({"repository", "revision"}), "producer") + normalized_producer = { + "repository": _require_repository(producer["repository"], "producer_repository"), + "revision": _require_revision(producer["revision"], "producer_revision"), + } + source_artifact = _exact_mapping( + envelope["source_artifact"], + frozenset({"repository", "revision", "run_id", "artifact_id", "sha256"}), + "source_artifact", + ) + normalized_artifact = build_source_artifact_metadata(**source_artifact) + ledger = envelope["ledger"] + if not isinstance(ledger, Mapping): + raise M0ResearchPublisherEnvelopeError("ledger_invalid") + # This validator is intentionally narrow: the only ledger accepted by the + # builder is generated by the local M0 aggregator above. It still binds + # the no-order policy and canonical timestamps before a publish attempt. + generated_at = ledger.get("generated_at") + expected_timestamp = canonical_timestamp(generated_at, "ledger_generated_at") + normalized_ledger = _validate_ledger(ledger, expected_timestamp=expected_timestamp) + expected_digest = calculate_ledger_sha256(normalized_ledger) + if _require_sha256(envelope["ledger_sha256"], "ledger_sha256") != expected_digest: + raise M0ResearchPublisherEnvelopeError("ledger_sha256_mismatch") + return { + "schema_version": PUBLISHER_ENVELOPE_SCHEMA, + "producer": normalized_producer, + "source_artifact": normalized_artifact, + "ledger_sha256": expected_digest, + "ledger": normalized_ledger, + } + + +def _publish_url_from_environment(environ: Mapping[str, str]) -> tuple[str, str]: + """Read only the two dedicated publisher variables, never ambient credentials.""" + + url = environ.get(PUBLISH_URL_ENV) + token = environ.get(PUBLISH_TOKEN_ENV) + if not isinstance(url, str) or not url or not isinstance(token, str) or not token: + raise M0ResearchPublisherEnvelopeError("m0_publish_environment_missing") + parsed = urllib.parse.urlsplit(url) + if ( + parsed.scheme != "https" + or not parsed.netloc + or parsed.username is not None + or parsed.password is not None + or parsed.query + or parsed.fragment + ): + raise M0ResearchPublisherEnvelopeError("m0_publish_url_invalid") + return url, token + + +def publish_m0_research_publisher_envelope( + envelope: Mapping[str, Any], *, environ: Mapping[str, str] | None = None +) -> None: + """POST one validated envelope through the dedicated, opt-in publication route.""" + + validated = validate_m0_research_publisher_envelope(envelope) + url, token = _publish_url_from_environment(os.environ if environ is None else environ) + request = urllib.request.Request( + url, + data=canonical_json(validated).encode("utf-8"), + method="POST", + headers={ + "Authorization": f"Bearer {token}", + "Content-Type": "application/json", + "Accept": "application/json", + }, + ) + try: + with urllib.request.urlopen(request, timeout=15) as response: # nosec B310 - URL is HTTPS validated above. + status = response.getcode() + except (urllib.error.HTTPError, urllib.error.URLError, OSError) as exc: + raise M0ResearchPublisherEnvelopeError("m0_publish_failed") from exc + if not isinstance(status, int) or status < 200 or status >= 300: + raise M0ResearchPublisherEnvelopeError("m0_publish_failed") + + +def _parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Build a local, no-order M0 research publisher envelope.") + parser.add_argument("--source-snapshot", required=True, type=Path) + parser.add_argument("--output", required=True, type=Path) + parser.add_argument("--source-artifact-repository", required=True) + parser.add_argument("--source-artifact-revision", required=True) + parser.add_argument("--source-artifact-run-id", required=True) + parser.add_argument("--source-artifact-id", required=True) + parser.add_argument("--source-artifact-sha256", required=True) + parser.add_argument("--producer-repository", required=True) + parser.add_argument("--producer-revision", required=True) + parser.add_argument( + "--now", + help="Canonical UTC timestamp (YYYY-MM-DDTHH:MM:SSZ); pass explicitly for reproducible output.", + ) + parser.add_argument( + "--publish", + action="store_true", + help=( + "POST only after writing the local envelope; requires dedicated " + f"{PUBLISH_URL_ENV} and {PUBLISH_TOKEN_ENV} environment variables." + ), + ) + return parser.parse_args(argv) + + +def main(argv: Sequence[str] | None = None) -> int: + args = _parse_args(argv) + now = canonical_timestamp(args.now) if args.now else _current_canonical_timestamp() + snapshot, snapshot_sha256 = load_source_snapshot(args.source_snapshot) + if snapshot_sha256 != _require_sha256(args.source_artifact_sha256, "source_artifact_sha256"): + raise M0ResearchPublisherEnvelopeError("source_artifact_sha256_mismatch") + artifact = build_source_artifact_metadata( + repository=args.source_artifact_repository, + revision=args.source_artifact_revision, + run_id=args.source_artifact_run_id, + artifact_id=args.source_artifact_id, + sha256=snapshot_sha256, + ) + envelope = build_m0_research_publisher_envelope( + source_snapshot=snapshot, + source_artifact=artifact, + producer_repository=args.producer_repository, + producer_revision=args.producer_revision, + now=now, + ) + if args.publish: + # Validate the dedicated environment before writing so a typo cannot + # leave a local file that an operator mistakes for an attempted POST. + _publish_url_from_environment(os.environ) + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(canonical_json(envelope) + "\n", encoding="utf-8") + if args.publish: + publish_m0_research_publisher_envelope(envelope) + print( + json.dumps( + { + "output": str(args.output), + "ledger_sha256": envelope["ledger_sha256"], + "published": bool(args.publish), + }, + sort_keys=True, + ) + ) + return 0 + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except M0ResearchPublisherEnvelopeError as exc: + print(str(exc), file=sys.stderr) + raise SystemExit(2) from exc + + +__all__ = [ + "MAX_SOURCE_SNAPSHOT_BYTES", + "M0ResearchPublisherEnvelopeError", + "PUBLISHER_ENVELOPE_SCHEMA", + "PUBLISH_TOKEN_ENV", + "PUBLISH_URL_ENV", + "build_m0_research_publisher_envelope", + "build_source_artifact_metadata", + "calculate_ledger_sha256", + "canonical_json", + "canonical_timestamp", + "load_source_snapshot", + "main", + "publish_m0_research_publisher_envelope", + "validate_m0_research_publisher_envelope", +] diff --git a/python/tests/test_build_m0_research_publisher_envelope.py b/python/tests/test_build_m0_research_publisher_envelope.py new file mode 100644 index 0000000..59c4977 --- /dev/null +++ b/python/tests/test_build_m0_research_publisher_envelope.py @@ -0,0 +1,232 @@ +from __future__ import annotations + +import hashlib +import importlib.util +import io +import json +import sys +import tempfile +import unittest +from contextlib import redirect_stdout +from pathlib import Path +from unittest.mock import patch + + +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 + + +_load_module("m0_research_ledger") +publisher = _load_module("build_m0_research_publisher_envelope") + + +class _Response: + def __init__(self, status: int): + self.status = status + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, traceback): + return False + + def getcode(self): + return self.status + + +class M0ResearchPublisherEnvelopeTest(unittest.TestCase): + def _snapshot(self) -> dict[str, object]: + return { + "schema_version": "qsl_m0_research_source_snapshot.v1", + "source_id": "quant-advisor-research", + "source_report_digest": "a" * 64, + "generated_at": "2026-08-20T12:01:00Z", + "computed_at": "2026-08-20T12:02:00Z", + "data_status": "ready", + "hypotheses": [ + { + "schema_version": "qsl.m0_research_hypothesis.v1", + "artifact_type": "research_hypothesis", + "authority": "research_only", + "no_order": True, + "hypothesis_id": "m0r-semiconductor-1", + "as_of": "2026-08-20", + "generated_at": "2026-08-20T12:00:00Z", + "expires_at": "2026-08-27T12:00:00Z", + "subject": {"kind": "asset_idea", "identifier": "SOXX"}, + "research_context": { + "state": "candidate", + "primary_horizon": "medium", + "suitable_horizons": ["medium"], + "source_confidence": "high", + "source_style": "mixed_research", + "theme_ids": ["semiconductors"], + }, + "evidence": { + "source_entry_digest": "b" * 64, + "evidence_ref_count": 3, + "risk_note_count": 1, + }, + "provenance": { + "source_project": "QuantAdvisorResearch", + "source_schema_version": "6", + "source_contract_version": "model_recommendations.v6", + "source_report_digest": "a" * 64, + "source_input_digest": "c" * 64, + }, + "permitted_next_step": "research_validation_only", + } + ], + "errors": [], + } + + def _artifact(self, sha256: str) -> dict[str, str]: + return publisher.build_source_artifact_metadata( + repository="QuantStrategyLab/QuantAdvisorResearch", + revision="d" * 40, + run_id="123456789", + artifact_id="m0-source-snapshot", + sha256=sha256, + ) + + def _arguments(self, source: Path, output: Path, sha256: str) -> list[str]: + return [ + "--source-snapshot", + str(source), + "--output", + str(output), + "--source-artifact-repository", + "QuantStrategyLab/QuantAdvisorResearch", + "--source-artifact-revision", + "d" * 40, + "--source-artifact-run-id", + "123456789", + "--source-artifact-id", + "m0-source-snapshot", + "--source-artifact-sha256", + sha256, + "--producer-repository", + "QuantStrategyLab/QuantRuntimeSettings", + "--producer-revision", + "e" * 40, + "--now", + "2026-08-21T12:00:00Z", + ] + + def test_build_is_deterministic_hash_bound_and_research_only(self): + source = self._snapshot() + artifact = self._artifact("f" * 64) + first = publisher.build_m0_research_publisher_envelope( + source_snapshot=source, + source_artifact=artifact, + producer_repository="QuantStrategyLab/QuantRuntimeSettings", + producer_revision="e" * 40, + now="2026-08-21T12:00:00Z", + ) + second = publisher.build_m0_research_publisher_envelope( + source_snapshot=json.loads(json.dumps(source)), + source_artifact=dict(reversed(artifact.items())), + producer_repository="QuantStrategyLab/QuantRuntimeSettings", + producer_revision="e" * 40, + now="2026-08-21T12:00:00Z", + ) + self.assertEqual(publisher.canonical_json(first), publisher.canonical_json(second)) + self.assertEqual(first["schema_version"], "qsl_m0_research_publisher_envelope.v1") + self.assertEqual(first["ledger"]["generated_at"], "2026-08-21T12:00:00Z") + self.assertEqual(first["ledger"]["computed_at"], "2026-08-21T12:00:00Z") + self.assertEqual(first["ledger"]["policy"]["authority"], "research_only") + self.assertTrue(first["ledger"]["policy"]["no_order"]) + self.assertEqual(first["ledger_sha256"], publisher.calculate_ledger_sha256(first["ledger"])) + self.assertEqual(publisher.validate_m0_research_publisher_envelope(first), first) + + def test_envelope_validation_rejects_digest_or_execution_policy_tampering(self): + envelope = publisher.build_m0_research_publisher_envelope( + source_snapshot=self._snapshot(), + source_artifact=self._artifact("f" * 64), + producer_repository="QuantStrategyLab/QuantRuntimeSettings", + producer_revision="e" * 40, + now="2026-08-21T12:00:00Z", + ) + tampered_digest = json.loads(json.dumps(envelope)) + tampered_digest["ledger_sha256"] = "0" * 64 + with self.assertRaisesRegex(publisher.M0ResearchPublisherEnvelopeError, "ledger_sha256_mismatch"): + publisher.validate_m0_research_publisher_envelope(tampered_digest) + + tampered_policy = json.loads(json.dumps(envelope)) + tampered_policy["ledger"]["policy"]["no_order"] = False + with self.assertRaisesRegex(publisher.M0ResearchPublisherEnvelopeError, "ledger_policy_invalid"): + publisher.validate_m0_research_publisher_envelope(tampered_policy) + + def test_cli_default_is_local_only_and_binds_the_exact_source_bytes(self): + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + source = root / "source.json" + output = root / "envelope.json" + raw = json.dumps(self._snapshot(), ensure_ascii=False, indent=2).encode("utf-8") + source.write_bytes(raw) + sha256 = hashlib.sha256(raw).hexdigest() + with patch.object(publisher.urllib.request, "urlopen", side_effect=AssertionError("network called")): + with redirect_stdout(io.StringIO()): + self.assertEqual(publisher.main(self._arguments(source, output, sha256)), 0) + envelope = json.loads(output.read_text(encoding="utf-8")) + self.assertEqual(envelope["source_artifact"]["sha256"], sha256) + self.assertEqual(envelope["ledger_sha256"], publisher.calculate_ledger_sha256(envelope["ledger"])) + + missing_output = root / "missing.json" + with self.assertRaisesRegex(publisher.M0ResearchPublisherEnvelopeError, "source_artifact_sha256_mismatch"): + publisher.main(self._arguments(source, missing_output, "0" * 64)) + self.assertFalse(missing_output.exists()) + + def test_publish_requires_dedicated_environment_and_never_serializes_token(self): + envelope = publisher.build_m0_research_publisher_envelope( + source_snapshot=self._snapshot(), + source_artifact=self._artifact("f" * 64), + producer_repository="QuantStrategyLab/QuantRuntimeSettings", + producer_revision="e" * 40, + now="2026-08-21T12:00:00Z", + ) + with self.assertRaisesRegex(publisher.M0ResearchPublisherEnvelopeError, "m0_publish_environment_missing"): + publisher.publish_m0_research_publisher_envelope(envelope, environ={}) + + secret = "dedicated-publisher-token" + captured = [] + + def fake_urlopen(request, timeout): + captured.append((request, timeout)) + return _Response(202) + + with patch.object(publisher.urllib.request, "urlopen", side_effect=fake_urlopen): + publisher.publish_m0_research_publisher_envelope( + envelope, + environ={ + publisher.PUBLISH_URL_ENV: "https://research-console.example/api/internal/m0", + publisher.PUBLISH_TOKEN_ENV: secret, + "BROKER_API_TOKEN": "must-not-be-read", + }, + ) + self.assertEqual(len(captured), 1) + self.assertEqual(captured[0][1], 15) + self.assertNotIn(secret, publisher.canonical_json(envelope)) + self.assertNotIn("BROKER_API_TOKEN", publisher.canonical_json(envelope)) + + with self.assertRaisesRegex(publisher.M0ResearchPublisherEnvelopeError, "m0_publish_url_invalid"): + publisher.publish_m0_research_publisher_envelope( + envelope, + environ={ + publisher.PUBLISH_URL_ENV: "https://research-console.example/api?token=not-allowed", + publisher.PUBLISH_TOKEN_ENV: secret, + }, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/schemas/qsl-m0-research-publisher-envelope.v1.schema.json b/schemas/qsl-m0-research-publisher-envelope.v1.schema.json new file mode 100644 index 0000000..5847d31 --- /dev/null +++ b/schemas/qsl-m0-research-publisher-envelope.v1.schema.json @@ -0,0 +1,46 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://quantstrategylab.github.io/QuantRuntimeSettings/schemas/qsl-m0-research-publisher-envelope.v1.schema.json", + "title": "QSL M0 Research Publisher Envelope v1", + "description": "Hash-bound transport for a locally aggregated M0 research ledger. It cannot express allocation, runtime, platform, broker, or execution authority.", + "type": "object", + "additionalProperties": false, + "required": ["schema_version", "producer", "source_artifact", "ledger_sha256", "ledger"], + "properties": { + "schema_version": { "const": "qsl_m0_research_publisher_envelope.v1" }, + "producer": { "$ref": "#/$defs/producer" }, + "source_artifact": { "$ref": "#/$defs/sourceArtifact" }, + "ledger_sha256": { "$ref": "#/$defs/sha256" }, + "ledger": { "$ref": "qsl-m0-research-ledger.v1.schema.json" } + }, + "$defs": { + "repository": { + "type": "string", + "pattern": "^[A-Za-z0-9][A-Za-z0-9_.-]*/[A-Za-z0-9][A-Za-z0-9_.-]*$" + }, + "revision": { "type": "string", "pattern": "^[0-9a-f]{40}$" }, + "sha256": { "type": "string", "pattern": "^[0-9a-f]{64}$" }, + "identifier": { "type": "string", "pattern": "^[A-Za-z0-9._:/-]{1,128}$" }, + "producer": { + "type": "object", + "additionalProperties": false, + "required": ["repository", "revision"], + "properties": { + "repository": { "$ref": "#/$defs/repository" }, + "revision": { "$ref": "#/$defs/revision" } + } + }, + "sourceArtifact": { + "type": "object", + "additionalProperties": false, + "required": ["repository", "revision", "run_id", "artifact_id", "sha256"], + "properties": { + "repository": { "$ref": "#/$defs/repository" }, + "revision": { "$ref": "#/$defs/revision" }, + "run_id": { "$ref": "#/$defs/identifier" }, + "artifact_id": { "$ref": "#/$defs/identifier" }, + "sha256": { "$ref": "#/$defs/sha256" } + } + } + } +}