ci: fail closed when CodeQL analysis fails

.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

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()

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"}