From 0b6159896d6d29fdfd7ffb99dffab44bbe372df9 Mon Sep 17 00:00:00 2001 From: Imran Siddique Date: Sun, 23 Aug 2026 09:53:53 -0700 Subject: [PATCH 1/2] security: bind receipt keys to trusted issuers --- .../tests/test_verify_receipts.py | 45 +++++++++++++++++-- embodied-action-receipts/verify_receipts.py | 41 +++++++++++++---- 2 files changed, 74 insertions(+), 12 deletions(-) diff --git a/embodied-action-receipts/tests/test_verify_receipts.py b/embodied-action-receipts/tests/test_verify_receipts.py index c0d6150..21bb6d9 100644 --- a/embodied-action-receipts/tests/test_verify_receipts.py +++ b/embodied-action-receipts/tests/test_verify_receipts.py @@ -1,15 +1,20 @@ from __future__ import annotations import json +import sys import unittest +from copy import deepcopy from pathlib import Path -import sys - ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) -from verify_receipts import verify_fixture # noqa: E402 +from generate_fixtures import sign +from verify_receipts import ( + load_trusted_keys, + receipt_hash, + verify_fixture, +) class ReceiptFixtureTests(unittest.TestCase): @@ -23,6 +28,40 @@ def test_rejected_receipt_is_valid_evidence(self) -> None: result = verify_fixture(ROOT / "fixtures" / "controller-rejected.json") self.assertEqual(result, {"result": "valid", "receipt_state": "rejected"}) + def test_explicit_empty_trust_store_trusts_nobody(self) -> None: + result = verify_fixture(ROOT / "fixtures" / "valid-chain.json", trusted_keys={}) + self.assertEqual(result, {"result": "invalid", "receipt_state": "untrusted"}) + + def test_trusted_key_cannot_claim_a_different_issuer(self) -> None: + fixture = json.loads((ROOT / "fixtures" / "valid-chain.json").read_text()) + forged = deepcopy(fixture["receipts"][0]) + forged.pop("signature") + forged["issuer"] = "spiffe://attacker.example/controller" + fixture["receipts"] = [sign(forged)] + path = ROOT / "fixtures" / ".test-wrong-issuer.json" + try: + path.write_text(json.dumps(fixture)) + result = verify_fixture(path, trusted_keys=load_trusted_keys()) + finally: + path.unlink(missing_ok=True) + self.assertEqual(result, {"result": "invalid", "receipt_state": "issuer_mismatch"}) + + def test_receipt_sequences_must_be_contiguous(self) -> None: + fixture = json.loads((ROOT / "fixtures" / "valid-chain.json").read_text()) + first = fixture["receipts"][0] + second = deepcopy(fixture["receipts"][1]) + second.pop("signature") + second["sequence"] = 3 + second["prev_receipt_hash"] = receipt_hash(first) + fixture["receipts"][1] = sign(second) + path = ROOT / "fixtures" / ".test-sequence-gap.json" + try: + path.write_text(json.dumps(fixture)) + result = verify_fixture(path) + finally: + path.unlink(missing_ok=True) + self.assertEqual(result, {"result": "invalid", "receipt_state": "sequence_mismatch"}) + if __name__ == "__main__": unittest.main() diff --git a/embodied-action-receipts/verify_receipts.py b/embodied-action-receipts/verify_receipts.py index fb43d8d..ea6adb9 100644 --- a/embodied-action-receipts/verify_receipts.py +++ b/embodied-action-receipts/verify_receipts.py @@ -6,16 +6,22 @@ import hashlib import json import sys +from dataclasses import dataclass from pathlib import Path from typing import Any from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey - ROOT = Path(__file__).parent +@dataclass(frozen=True) +class TrustedSigner: + issuer: str + public_key: Ed25519PublicKey + + def canonical_bytes(value: Any) -> bytes: return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() @@ -46,17 +52,25 @@ def receipt_hash(receipt: dict[str, Any]) -> str: return sha256_ref(receipt) -def load_trusted_keys(path: Path = ROOT / "trusted-keys.json") -> dict[str, Ed25519PublicKey]: +def load_trusted_keys(path: Path = ROOT / "trusted-keys.json") -> dict[str, TrustedSigner]: data = json.loads(path.read_text()) keys = {} for key_id, key in data["controller_signers"].items(): - keys[key_id] = Ed25519PublicKey.from_public_bytes(b64url_decode(key["public_key_b64url"])) + keys[key_id] = TrustedSigner( + issuer=key["issuer"], + public_key=Ed25519PublicKey.from_public_bytes( + b64url_decode(key["public_key_b64url"]) + ), + ) return keys -def verify_fixture(path: Path, trusted_keys: dict[str, Ed25519PublicKey] | None = None) -> dict[str, Any]: +def verify_fixture( + path: Path, trusted_keys: dict[str, TrustedSigner] | None = None +) -> dict[str, Any]: fixture = json.loads(path.read_text()) - trusted_keys = trusted_keys or load_trusted_keys() + if trusted_keys is None: + trusted_keys = load_trusted_keys() trace = fixture["trace"] action = fixture["action"] @@ -74,7 +88,11 @@ def verify_fixture(path: Path, trusted_keys: dict[str, Ed25519PublicKey] | None final_state = "absent" final_verdict = None - for receipt in sorted(receipts, key=lambda r: r["sequence"]): + for expected_sequence, receipt in enumerate( + sorted(receipts, key=lambda r: r["sequence"]), start=1 + ): + if receipt["sequence"] != expected_sequence: + return {"result": "invalid", "receipt_state": "sequence_mismatch"} if receipt["call_id"] != trace["cmcp_call_id"]: return {"result": "invalid", "receipt_state": "call_id_mismatch"} if receipt["trace_id"] != trace["trace_id"]: @@ -84,16 +102,21 @@ def verify_fixture(path: Path, trusted_keys: dict[str, Ed25519PublicKey] | None if receipt.get("prev_receipt_hash") != previous_hash: return {"result": "invalid", "receipt_state": "chain_mismatch"} - key = trusted_keys.get(receipt["issuer_key_id"]) - if key is None: + signer = trusted_keys.get(receipt["issuer_key_id"]) + if signer is None: return {"result": "invalid", "receipt_state": "untrusted"} + if receipt.get("issuer") != signer.issuer: + return {"result": "invalid", "receipt_state": "issuer_mismatch"} signature = receipt["signature"] if not signature.startswith("ed25519:"): return {"result": "invalid", "receipt_state": "signature_format"} try: - key.verify(b64url_decode(signature.removeprefix("ed25519:")), canonical_bytes(receipt_preimage(receipt))) + signer.public_key.verify( + b64url_decode(signature.removeprefix("ed25519:")), + canonical_bytes(receipt_preimage(receipt)), + ) except InvalidSignature: return {"result": "invalid", "receipt_state": "invalid_signature"} From 5533eb9cc92c615b760d9e81f032dac14af152db Mon Sep 17 00:00:00 2001 From: Imran Siddique Date: Sun, 23 Aug 2026 10:52:56 -0700 Subject: [PATCH 2/2] ci: fail closed when CodeQL analysis fails --- .github/workflows/codeql.yml | 1 - 1 file changed, 1 deletion(-) diff --git a/.github/workflows/codeql.yml b/.github/workflows/codeql.yml index a6ae7ed..ca2de24 100644 --- a/.github/workflows/codeql.yml +++ b/.github/workflows/codeql.yml @@ -38,6 +38,5 @@ jobs: - name: Perform CodeQL Analysis uses: github/codeql-action/analyze@v4 - continue-on-error: true with: category: /language:python