diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 94b588b7..98732d35 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -450,6 +450,33 @@ jobs: curl -fsS http://127.0.0.1:8877/data/reality_cases.json | grep -q "async-rithmic-53" kill "$(cat /tmp/counterproof-demo.pid)" + + external-measurement-capsule-replay: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: "3.11" + + - name: Install CounterProof + run: | + python -m pip install --upgrade pip + pip install -e . + + - name: Fetch pinned CSOAI claim-watch capsule batch + run: | + curl -fsSL --retry 3 --retry-delay 1 \ + -o /tmp/csoai-claim-watch-capsules.jsonl.gz \ + https://raw.githubusercontent.com/CSOAI-ORG/councilof-ai/85df6bc2ed76450c171d6982bfd39e94d80982d2/public/measurement-capsules/v0.2/claim_watch/capsules.jsonl.gz + + - name: Independently replay public capsule ids and Merkle root + run: | + python scripts/interop/replay_csoai_claim_watch.py \ + /tmp/csoai-claim-watch-capsules.jsonl.gz + + validate-example-skills: runs-on: ubuntu-latest steps: diff --git a/docs/interop/CSOAI_CLAIM_WATCH_REPLAY.md b/docs/interop/CSOAI_CLAIM_WATCH_REPLAY.md new file mode 100644 index 00000000..cb520e83 --- /dev/null +++ b/docs/interop/CSOAI_CLAIM_WATCH_REPLAY.md @@ -0,0 +1,27 @@ +# CSOAI claim_watch external Merkle replay + +CounterProof pins one public external batch from: + +- repository: `CSOAI-ORG/councilof-ai` +- commit: `85df6bc2ed76450c171d6982bfd39e94d80982d2` +- path: `public/measurement-capsules/v0.2/claim_watch/leaves.json` + +The published batch contains one capsule id: + +`007b0e5ccbb3f86ac587d645dd8df74c689765267eadc09736eb58bf8be95314` + +Applying the draft's RFC 6962/9162 leaf rule: + +`SHA-256(0x00 || capsule_id_bytes)` + +reproduces the published Merkle root: + +`5e6440b6f9fbccdc2dca66aff3c79c02f649ccb56cd795d0f043b651dd50c32e` + +This is a **real external Merkle replay**, not a synthetic fixture. + +It is still only a partial result. The gzip capsule bytes are published in the +same upstream directory, but the current connector cannot decode binary +repository content. Until CounterProof recomputes the capsule id from those +published capsule bytes, it must not claim the Measurement Capsule draft's full +independent-batch-reproduction success condition. diff --git a/docs/interop/MEASUREMENT_CAPSULE_INDEPENDENT_VERIFIER.md b/docs/interop/MEASUREMENT_CAPSULE_INDEPENDENT_VERIFIER.md new file mode 100644 index 00000000..b79bb8d2 --- /dev/null +++ b/docs/interop/MEASUREMENT_CAPSULE_INDEPENDENT_VERIFIER.md @@ -0,0 +1,53 @@ +# Independent SCITT Measurement Capsule verifier + +Status: experimental. + +This module implements the identifier and batch-integrity parts of +`draft-templeman-scitt-measurement-capsule-00` from the public text. + +It does **not** import or call the CSOAI prototype builder/verifier. + +Implemented: + +- RFC 8785 JCS canonicalization using the independent Trail of Bits + `rfc8785` package; +- `capsule_id = SHA-256(JCS(capsule without capsule_id))`; +- exact-JCS stored-line verification; +- draft Section 7.1 no-decision/no-authority surface checks; +- digest-only `sources` admission; +- preservation of `UNCHECKABLE` as a measurement state; +- RFC 9162 Merkle Tree Hash using: + - leaf hash `SHA-256(0x00 || capsule_id_bytes)`; + - node hash `SHA-256(0x01 || left || right)`; + - capsule ids sorted in ascending byte order; + - duplicate refusal; +- expected root/count verification for a published batch. + +Not implemented: + +- COSE_Sign1; +- SCITT Transparency Service registration/Receipt verification; +- OpenTimestamps or Rekor verification; +- measurement-instrument correctness; +- CSOAI-specific per-kind semantic vocabularies beyond the generic draft rules. + +## External experiment target + +The draft's Section 13 defines a success outcome where an implementation by +another party, written from the text, recomputes identifiers and roots of a +published batch. + +This verifier is intended to attempt exactly that outcome. + +At the time this code was added, the public draft/index were readable but the +raw `capsules.jsonl.gz` / `leaves.json` batch bytes were not retrievable +through the current automation environment. Therefore this repository must not +claim the Section 13 outcome until a real published batch is supplied and the +root matches. + +## Claim boundary + +A successful Merkle recomputation proves only that the supplied capsule bytes +bind to the supplied root under the draft algorithm. It does not prove that the +measurements are true, independent, unbiased, authorised, or endorsed by SCITT +or the IETF. diff --git a/examples/interop/csoai-claim-watch-leaves-2026-10-01.json b/examples/interop/csoai-claim-watch-leaves-2026-10-01.json new file mode 100644 index 00000000..02adfcaf --- /dev/null +++ b/examples/interop/csoai-claim-watch-leaves-2026-10-01.json @@ -0,0 +1,17 @@ +{ + "source_repository": "CSOAI-ORG/councilof-ai", + "source_commit": "85df6bc2ed76450c171d6982bfd39e94d80982d2", + "source_path": "public/measurement-capsules/v0.2/claim_watch/leaves.json", + "batch": "claim_watch", + "kind": "measurement.claim_watch", + "n": 1, + "leaves": [ + "007b0e5ccbb3f86ac587d645dd8df74c689765267eadc09736eb58bf8be95314" + ], + "expected_merkle_root": "5e6440b6f9fbccdc2dca66aff3c79c02f649ccb56cd795d0f043b651dd50c32e", + "claim_boundary": [ + "This fixture pins public CSOAI leaf/root data.", + "Matching this root does not prove the capsule_id was independently recomputed from capsule bytes.", + "Full draft experiment success still requires replay of the published capsule bytes." + ] +} diff --git a/pyproject.toml b/pyproject.toml index 9edb46a2..2d359c9e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -36,6 +36,7 @@ dependencies = [ "click>=8.1", "rich>=13.0", "pydantic>=2.0", + "rfc8785>=0.1.4", ] [project.urls] diff --git a/scripts/interop/replay_csoai_claim_watch.py b/scripts/interop/replay_csoai_claim_watch.py new file mode 100644 index 00000000..43cae8f8 --- /dev/null +++ b/scripts/interop/replay_csoai_claim_watch.py @@ -0,0 +1,55 @@ +"""Replay one pinned public CSOAI Measurement Capsule batch. + +This script is intentionally small and network-agnostic: CI fetches the exact +commit-pinned gzip bytes, this script verifies the bytes and draft-level +identifier/root semantics. + +It does not claim CSOAI or IETF endorsement. +""" + +from __future__ import annotations + +import argparse +import gzip +import hashlib +from pathlib import Path + +from skill_factory.evolution.measurement_capsule_verify import verify_jsonl_batch + + +EXPECTED_GZIP_SHA256 = "70451eda2d8166b9a61084f5c01b8ff57210a7016d97ab0320b4c730f2b827ed" +EXPECTED_ROOT = "5e6440b6f9fbccdc2dca66aff3c79c02f649ccb56cd795d0f043b651dd50c32e" +EXPECTED_COUNT = 1 + + +def replay(path: Path) -> None: + raw = path.read_bytes() + digest = hashlib.sha256(raw).hexdigest() + if digest != EXPECTED_GZIP_SHA256: + raise SystemExit( + f"gzip sha256 mismatch: got={digest} expected={EXPECTED_GZIP_SHA256}" + ) + + data = gzip.decompress(raw) + lines = data.splitlines(keepends=True) + result = verify_jsonl_batch( + lines, + expected_merkle_root=EXPECTED_ROOT, + expected_n_capsules=EXPECTED_COUNT, + ) + + print("CSOAI claim_watch replay: PASS") + print(f"capsules={result.n_capsules}") + print(f"merkle_root={result.merkle_root}") + print(f"capsule_id={result.capsule_ids[0]}") + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("capsules_gz", type=Path) + args = parser.parse_args() + replay(args.capsules_gz) + + +if __name__ == "__main__": + main() diff --git a/skill_factory/evolution/measurement_capsule_verify.py b/skill_factory/evolution/measurement_capsule_verify.py new file mode 100644 index 00000000..56d4d231 --- /dev/null +++ b/skill_factory/evolution/measurement_capsule_verify.py @@ -0,0 +1,294 @@ +"""Independent verifier for SCITT Measurement Capsule draft semantics. + +This module is intentionally implemented from the public draft text and RFC +primitives rather than importing the CSOAI prototype builder or verifier. + +Scope: +- recompute capsule_id using RFC 8785/JCS with capsule_id excluded; +- validate stored JSONL lines are the exact JCS bytes for their capsule; +- recompute RFC 9162 Merkle Tree Hash over capsule_id bytes sorted ascending; +- preserve the draft's measurement-only / no-decision boundary. + +It does not verify COSE, Transparency Service receipts, OpenTimestamps, or the +measurement instrument itself. +""" + +from __future__ import annotations + +import hashlib +import json +import re +from dataclasses import dataclass +from typing import Any, Iterable, Sequence + +import rfc8785 + + +CAPSULE_SCHEMA = "csoai.measurement-capsule/0.2" +AUTHORITY_NONE = ( + "NONE: measurement only; this capsule grants and records no execution authority" +) +_SHA256_HEX = re.compile(r"^[0-9a-f]{64}$") + +_FORBIDDEN_MEMBER_STEMS = { + "decision", + "allow", + "allowed", + "allowlist", + "hold", + "reject", + "approve", + "approved", + "approval", + "admit", + "admission", + "authority", + "authorization", + "authorisation", + "permit", + "permission", + "grant", + "enforce", + "enforcement", + "action", + "recommended_action", + "gate", + "gate_result", +} + +_FORBIDDEN_EXACT_VALUES = { + "ALLOW", + "HOLD", + "REJECT", + "DENY", + "APPROVE", + "APPROVED", + "ADMIT", + "ADMITTED", + "BLOCK", + "PERMIT", + "GRANT", +} + + +class MeasurementCapsuleError(ValueError): + """Raised when a capsule violates the draft-level verification boundary.""" + + +@dataclass(frozen=True) +class CapsuleVerification: + capsule_id: str + canonical_bytes: bytes + stored_line_matches_jcs: bool + + +@dataclass(frozen=True) +class BatchVerification: + n_capsules: int + merkle_root: str + capsule_ids: tuple[str, ...] + duplicate_ids: tuple[str, ...] + + +def _canonical_without_id(capsule: dict[str, Any]) -> bytes: + payload = dict(capsule) + payload.pop("capsule_id", None) + try: + return rfc8785.dumps(payload) + except Exception as exc: # library-specific canonicalization subclasses vary + raise MeasurementCapsuleError(f"JCS canonicalization failed: {exc}") from exc + + +def recompute_capsule_id(capsule: dict[str, Any]) -> str: + """Recompute draft Section 4 capsule_id from JCS bytes.""" + if not isinstance(capsule, dict): + raise TypeError("capsule must be a JSON object") + return hashlib.sha256(_canonical_without_id(capsule)).hexdigest() + + +def _walk_no_decision_surface(value: Any, *, path: tuple[str, ...] = ()) -> None: + if isinstance(value, dict): + for key, child in value.items(): + key_text = str(key) + if key_text != "authority_state": + normalized = key_text.lower().replace("-", "_") + if normalized in _FORBIDDEN_MEMBER_STEMS: + where = ".".join((*path, key_text)) + raise MeasurementCapsuleError( + f"forbidden decision/authority member at {where}" + ) + _walk_no_decision_surface(child, path=(*path, key_text)) + elif isinstance(value, list): + for index, child in enumerate(value): + _walk_no_decision_surface(child, path=(*path, str(index))) + elif isinstance(value, str) and value in _FORBIDDEN_EXACT_VALUES: + where = ".".join(path) + raise MeasurementCapsuleError( + f"forbidden decision-like string value at {where or ''}" + ) + + +def _validate_sources(value: Any, *, path: str = "sources") -> None: + if value is None: + return + if isinstance(value, str): + digest = value.removeprefix("sha256:") + if not re.fullmatch(r"[0-9a-f]{40,128}", digest): + raise MeasurementCapsuleError( + f"{path} contains a non-digest string" + ) + return + if isinstance(value, list): + for index, child in enumerate(value): + _validate_sources(child, path=f"{path}[{index}]") + return + if isinstance(value, dict): + for key, child in value.items(): + _validate_sources(child, path=f"{path}.{key}") + return + raise MeasurementCapsuleError( + f"{path} contains evidence bytes/free-text instead of digests" + ) + + +def validate_capsule_shape(capsule: dict[str, Any]) -> None: + if capsule.get("schema") != CAPSULE_SCHEMA: + raise MeasurementCapsuleError( + f"unsupported capsule schema: {capsule.get('schema')!r}" + ) + kind = capsule.get("kind") + if not isinstance(kind, str) or not re.fullmatch(r"measurement\.[a-z0-9_]+", kind): + raise MeasurementCapsuleError("invalid measurement kind") + if capsule.get("authority_state") != AUTHORITY_NONE: + raise MeasurementCapsuleError("authority_state must be measurement-only NONE") + capsule_id = capsule.get("capsule_id") + if not isinstance(capsule_id, str) or not _SHA256_HEX.fullmatch(capsule_id): + raise MeasurementCapsuleError("capsule_id must be 64 lowercase hex characters") + state = capsule.get("measurement_state") + if not isinstance(state, str) or not (1 <= len(state) <= 64): + raise MeasurementCapsuleError("measurement_state must be a non-empty short string") + limitations = capsule.get("limitations") + if not isinstance(limitations, list): + raise MeasurementCapsuleError("limitations must be a list") + _validate_sources(capsule.get("sources")) + _walk_no_decision_surface(capsule) + + +def verify_capsule(capsule: dict[str, Any], *, stored_line: bytes | None = None) -> CapsuleVerification: + """Verify identifier, no-decision boundary, and optional canonical stored line.""" + validate_capsule_shape(capsule) + expected = recompute_capsule_id(capsule) + if capsule["capsule_id"] != expected: + raise MeasurementCapsuleError( + f"capsule_id mismatch: stored={capsule['capsule_id']} recomputed={expected}" + ) + + canonical_full = rfc8785.dumps(capsule) + line_matches = True + if stored_line is not None: + candidate = stored_line[:-1] if stored_line.endswith(b"\n") else stored_line + line_matches = candidate == canonical_full + if not line_matches: + raise MeasurementCapsuleError( + "stored capsule line is not the exact JCS serialization" + ) + + return CapsuleVerification( + capsule_id=expected, + canonical_bytes=canonical_full, + stored_line_matches_jcs=line_matches, + ) + + +def _leaf_hash(capsule_id_hex: str) -> bytes: + if not _SHA256_HEX.fullmatch(capsule_id_hex): + raise MeasurementCapsuleError(f"invalid capsule id leaf: {capsule_id_hex!r}") + return hashlib.sha256(b"\x00" + bytes.fromhex(capsule_id_hex)).digest() + + +def _node_hash(left: bytes, right: bytes) -> bytes: + return hashlib.sha256(b"\x01" + left + right).digest() + + +def _largest_power_of_two_less_than(n: int) -> int: + if n < 2: + raise ValueError("n must be >= 2") + return 1 << ((n - 1).bit_length() - 1) + + +def _mth_from_ids(sorted_ids: Sequence[str]) -> bytes: + n = len(sorted_ids) + if n == 0: + return hashlib.sha256(b"").digest() + if n == 1: + return _leaf_hash(sorted_ids[0]) + k = _largest_power_of_two_less_than(n) + return _node_hash( + _mth_from_ids(sorted_ids[:k]), + _mth_from_ids(sorted_ids[k:]), + ) + + +def recompute_batch_merkle_root(capsule_ids: Iterable[str]) -> BatchVerification: + """Recompute RFC 9162 Merkle Tree Hash over sorted capsule_id bytes.""" + ids = tuple(capsule_ids) + counts: dict[str, int] = {} + for item in ids: + if not isinstance(item, str) or not _SHA256_HEX.fullmatch(item): + raise MeasurementCapsuleError(f"invalid capsule_id: {item!r}") + counts[item] = counts.get(item, 0) + 1 + duplicates = tuple(sorted(item for item, count in counts.items() if count > 1)) + if duplicates: + raise MeasurementCapsuleError( + "duplicate capsule_id values are not permitted in a batch: " + + ", ".join(duplicates) + ) + + ordered = tuple(sorted(ids, key=lambda item: bytes.fromhex(item))) + root = _mth_from_ids(ordered).hex() + return BatchVerification( + n_capsules=len(ordered), + merkle_root=root, + capsule_ids=ordered, + duplicate_ids=(), + ) + + +def verify_jsonl_batch( + lines: Iterable[bytes], + *, + expected_merkle_root: str | None = None, + expected_n_capsules: int | None = None, +) -> BatchVerification: + """Verify canonical capsule lines and recompute the published batch root.""" + ids: list[str] = [] + previous: str | None = None + for line_number, raw in enumerate(lines, start=1): + if not raw.strip(): + raise MeasurementCapsuleError( + f"blank capsule line at {line_number}" + ) + try: + capsule = json.loads(raw) + except json.JSONDecodeError as exc: + raise MeasurementCapsuleError( + f"invalid JSON at line {line_number}: {exc}" + ) from exc + result = verify_capsule(capsule, stored_line=raw) + if previous is not None and result.capsule_id <= previous: + raise MeasurementCapsuleError( + "capsule file is not strictly sorted by capsule_id" + ) + previous = result.capsule_id + ids.append(result.capsule_id) + + batch = recompute_batch_merkle_root(ids) + if expected_n_capsules is not None and batch.n_capsules != expected_n_capsules: + raise MeasurementCapsuleError( + f"capsule count mismatch: got={batch.n_capsules} expected={expected_n_capsules}" + ) + if expected_merkle_root is not None and batch.merkle_root != expected_merkle_root: + raise MeasurementCapsuleError( + f"Merkle root mismatch: got={batch.merkle_root} expected={expected_merkle_root}" + ) + return batch diff --git a/tests/unit/test_csoai_claim_watch_replay.py b/tests/unit/test_csoai_claim_watch_replay.py new file mode 100644 index 00000000..98800488 --- /dev/null +++ b/tests/unit/test_csoai_claim_watch_replay.py @@ -0,0 +1,24 @@ +import json +from pathlib import Path + +from skill_factory.evolution.measurement_capsule_verify import recompute_batch_merkle_root + + +FIXTURE=Path("examples/interop/csoai-claim-watch-leaves-2026-10-01.json") + + +def test_replay_real_csoai_claim_watch_merkle_root_from_public_leaf(): + payload=json.loads(FIXTURE.read_text(encoding="utf-8")) + + result=recompute_batch_merkle_root(payload["leaves"]) + + assert result.n_capsules==payload["n"] + assert result.merkle_root==payload["expected_merkle_root"] + + +def test_real_leaf_replay_fixture_does_not_claim_capsule_byte_reproduction(): + payload=json.loads(FIXTURE.read_text(encoding="utf-8")) + boundary=" ".join(payload["claim_boundary"]).lower() + + assert "does not prove" in boundary + assert "capsule bytes" in boundary diff --git a/tests/unit/test_measurement_capsule_verify.py b/tests/unit/test_measurement_capsule_verify.py new file mode 100644 index 00000000..4c224c3b --- /dev/null +++ b/tests/unit/test_measurement_capsule_verify.py @@ -0,0 +1,138 @@ +import hashlib + +import pytest +import rfc8785 + +from skill_factory.evolution.measurement_capsule_verify import ( + AUTHORITY_NONE, + MeasurementCapsuleError, + recompute_batch_merkle_root, + recompute_capsule_id, + verify_capsule, + verify_jsonl_batch, +) + + +def _capsule(subject, state="MEASURED"): + capsule={ + "schema":"csoai.measurement-capsule/0.2", + "kind":"measurement.public_signal", + "subject_id":subject, + "claim":{"statement":"published value equals observed value"}, + "declared":{"value":"1"}, + "observed":{"value":"1"}, + "differential":{"equal":True}, + "sources":{"declared_sha256":"a"*64,"observed_sha256":"b"*64}, + "measurement_state":state, + "authority_state":AUTHORITY_NONE, + "effect_reference":None, + "observed_at":"2026-10-08T00:00:00Z", + "correction_pointer":None, + "limitations":["synthetic draft-level verifier test only"], + "capsule_id":"", + } + capsule["capsule_id"]=recompute_capsule_id(capsule) + return capsule + + +def test_capsule_id_uses_rfc8785_with_capsule_id_excluded(): + capsule=_capsule("example:one") + without=dict(capsule) + without.pop("capsule_id") + expected=hashlib.sha256(rfc8785.dumps(without)).hexdigest() + + assert capsule["capsule_id"]==expected + assert verify_capsule(capsule).capsule_id==expected + + +def test_exact_jcs_line_is_required_when_stored_line_is_supplied(): + capsule=_capsule("example:one") + canonical=rfc8785.dumps(capsule)+b"\n" + + assert verify_capsule(capsule,stored_line=canonical).stored_line_matches_jcs + + pretty=(__import__("json").dumps(capsule,indent=2)+"\n").encode() + with pytest.raises(MeasurementCapsuleError,match="exact JCS"): + verify_capsule(capsule,stored_line=pretty) + + +def test_unc_checkable_is_preserved_as_measurement_state(): + capsule=_capsule("example:one",state="UNCHECKABLE") + + result=verify_capsule(capsule) + + assert result.capsule_id==capsule["capsule_id"] + + +def test_decision_like_member_is_rejected(): + capsule=_capsule("example:one") + capsule["declared"]["approval"]="APPROVED" + capsule["capsule_id"]=recompute_capsule_id(capsule) + + with pytest.raises(MeasurementCapsuleError,match="forbidden"): + verify_capsule(capsule) + + +def test_sources_must_be_digests_not_urls_or_free_text(): + capsule=_capsule("example:one") + capsule["sources"]["source"]="https://example.com/raw-evidence" + capsule["capsule_id"]=recompute_capsule_id(capsule) + + with pytest.raises(MeasurementCapsuleError,match="non-digest"): + verify_capsule(capsule) + + +def test_rfc9162_single_leaf_root_has_leaf_domain_separator(): + leaf="11"*32 + expected=hashlib.sha256(b"\x00"+bytes.fromhex(leaf)).hexdigest() + + assert recompute_batch_merkle_root([leaf]).merkle_root==expected + + +def test_rfc9162_two_leaf_root_has_node_domain_separator(): + first="11"*32 + second="22"*32 + left=hashlib.sha256(b"\x00"+bytes.fromhex(first)).digest() + right=hashlib.sha256(b"\x00"+bytes.fromhex(second)).digest() + expected=hashlib.sha256(b"\x01"+left+right).hexdigest() + + assert recompute_batch_merkle_root([second,first]).merkle_root==expected + + +def test_duplicate_capsule_ids_are_rejected(): + leaf="11"*32 + + with pytest.raises(MeasurementCapsuleError,match="duplicate"): + recompute_batch_merkle_root([leaf,leaf]) + + +def test_jsonl_batch_requires_sorted_canonical_lines_and_expected_root(): + first=_capsule("example:a") + second=_capsule("example:b") + capsules=sorted([first,second],key=lambda item:item["capsule_id"]) + lines=[rfc8785.dumps(item)+b"\n" for item in capsules] + expected=recompute_batch_merkle_root( + [item["capsule_id"] for item in capsules] + ) + + result=verify_jsonl_batch( + lines, + expected_merkle_root=expected.merkle_root, + expected_n_capsules=2, + ) + + assert result==expected + + +def test_jsonl_batch_rejects_unsorted_capsules(): + first=_capsule("example:a") + second=_capsule("example:b") + capsules=sorted( + [first,second], + key=lambda item:item["capsule_id"], + reverse=True, + ) + lines=[rfc8785.dumps(item)+b"\n" for item in capsules] + + with pytest.raises(MeasurementCapsuleError,match="strictly sorted"): + verify_jsonl_batch(lines)