fix(verify): enforce canonical base64url and reject private JWK members in kill_switch and claim verification (#804)

src/cmcp_verify/kill_switch.py

@@ -17,6 +17,7 @@
 import base64
 import binascii
 import hashlib
+import re
 from dataclasses import dataclass, field
 from typing import Any
 
@@ -38,8 +39,16 @@ class KillSwitchRefusalResult:
     errors: list[str] = field(default_factory=list)
 
 
+_B64URL_RE = re.compile(r"^[A-Za-z0-9_-]+$")
+
+
 def _b64url_decode(value: str) -> bytes:
-    return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))
+    if not isinstance(value, str) or not _B64URL_RE.fullmatch(value) or len(value) % 4 == 1:
+        raise ValueError("invalid canonical base64url")
+    raw = base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))
+    if base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii") != value:
+        raise ValueError("non-canonical base64url encoding")
+    return raw
 
 
 def verify_kill_switch_refusal(
@@ -100,6 +109,11 @@ def verify_kill_switch_refusal(
     if isinstance(identity, dict) and receipt.get("agent_id") != identity.get("agent_id"):
         errors.append("receipt names a different agent identity than the claim")
 
+    for req_field in ("agent_id", "session_id", "refused_at"):
+        val = receipt.get(req_field)
+        if not isinstance(val, str) or not val:
+            errors.append(f"receipt.{req_field} must be a non-empty string")
+
     return KillSwitchRefusalResult(
         valid=not errors,
         agent_id=receipt.get("agent_id"),

src/cmcp_verify/verify.py

@@ -102,6 +102,17 @@ def _leaf_public_key_pem(chain_pem: bytes) -> bytes:
 _EXTERNAL_EVIDENCE_ERROR = "EXTERNAL_EVIDENCE_VERIFICATION_FAILED"
 _EXTERNAL_EVIDENCE_HASH_RE = re.compile(r"^sha(256|384):[0-9a-f]+$")
 _ISSUER_KEY_ID_RE = re.compile(r"^[0-9a-f]{64}$")
+_B64URL_RE = re.compile(r"^[A-Za-z0-9_-]+$")
+_PRIVATE_JWK_PARAMS = frozenset({"d", "p", "q", "dp", "dq", "qi", "oth", "k"})
+
+
+def _canonical_b64url_decode(value: Any) -> bytes:
+    if not isinstance(value, str) or not _B64URL_RE.fullmatch(value) or len(value) % 4 == 1:
+        raise ValueError("must be canonical unpadded base64url")
+    raw = base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))
+    if base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii") != value:
+        raise ValueError("non-canonical base64url encoding")
+    return raw
 _EXTERNAL_EVIDENCE_TYPES = frozenset({
     "controller-execution-receipt/v1",
     "tee-signed-receipt",
@@ -187,9 +198,10 @@ def _canonical_json(claim_dict: dict[str, Any]) -> bytes:
 def _jwk_x_to_hex(x_b64: str) -> str | None:
     """Decode trace.cnf.jwk.x (base64url, no padding) to hex. Returns None on error."""
     try:
-        padding = 4 - (len(x_b64) % 4)
-        padded = x_b64 + ("=" * padding if padding != 4 else "")
-        return base64.urlsafe_b64decode(padded).hex()
+        raw = _canonical_b64url_decode(x_b64)
+        if len(raw) != 32:
+            return None
+        return raw.hex()
     except Exception:
         return None
 
@@ -211,24 +223,23 @@ def _platform_report_data(nonce: Any) -> str | None:
 def _verify_signature(claim: dict[str, Any]) -> tuple[bool, str | None]:
     """Verify the Ed25519 signature using the JWK public key in trace.cnf.jwk.x."""
     try:
-        x_b64: str = claim["trace"]["cnf"]["jwk"]["x"]
-        padding = 4 - (len(x_b64) % 4)
-        if padding != 4:
-            x_b64 += "=" * padding
-        pub_bytes = base64.urlsafe_b64decode(x_b64)
+        jwk = claim["trace"]["cnf"]["jwk"]
+        if not isinstance(jwk, dict):
+            return False, "trace.cnf.jwk must be a JSON object"
+        if _PRIVATE_JWK_PARAMS.intersection(jwk):
+            return False, "trace.cnf.jwk must not contain private or symmetric JWK parameters"
+        x_b64: str = jwk["x"]
+        pub_bytes = _canonical_b64url_decode(x_b64)
         pub_key = Ed25519PublicKey.from_public_bytes(pub_bytes)
-    except (KeyError, ValueError) as exc:
+    except (KeyError, TypeError, ValueError) as exc:
         return False, f"cannot parse trace.cnf.jwk.x: {exc}"
 
-    sig_b64: str = claim.get("signature", "")
-    if not sig_b64:
+    sig_b64 = claim.get("signature", "")
+    if not isinstance(sig_b64, str) or not sig_b64:
         return False, "signature field is empty"
 
     try:
-        padding = 4 - (len(sig_b64) % 4)
-        if padding != 4:
-            sig_b64 += "=" * padding
-        sig_bytes = base64.urlsafe_b64decode(sig_b64)
+        sig_bytes = _canonical_b64url_decode(sig_b64)
     except Exception as exc:
         return False, f"cannot decode signature: {exc}"
 
@@ -271,14 +282,17 @@ def _verify_key_binding(
         return None, "software-only mode -- TEE key binding not applicable"
 
     # Extract the public key bytes from cnf.jwk.x
-    x_b64 = claim.get("trace", {}).get("cnf", {}).get("jwk", {}).get("x", "")
+    jwk = claim.get("trace", {}).get("cnf", {}).get("jwk", {})
+    if not isinstance(jwk, dict) or _PRIVATE_JWK_PARAMS.intersection(jwk):
+        return False, "trace.cnf.jwk is malformed or contains private key parameters"
+    x_b64 = jwk.get("x", "")
     if not x_b64:
         return False, "trace.cnf.jwk.x is missing -- cannot verify key binding"
 
     try:
-        padding = 4 - (len(x_b64) % 4)
-        padded = x_b64 + ("=" * padding if padding != 4 else "")
-        base64.urlsafe_b64decode(padded)  # validate encoding; bytes not needed
+        raw_x = _canonical_b64url_decode(x_b64)
+        if len(raw_x) != 32:
+            return False, "trace.cnf.jwk.x must decode to 32 bytes"
     except Exception as exc:
         return False, f"cannot decode trace.cnf.jwk.x: {exc}"
 

tests/unit/test_kill_switch_evidence.py

@@ -192,3 +192,205 @@ def test_enabled_kill_switch_without_an_identity_refuses_to_start(tmp_path, monk
     with pytest.raises(SystemExit) as exc_info:
         run_startup(str(config_path))
     assert exc_info.value.code == 1
+
+
+def test_kill_switch_and_claim_verify_reject_non_canonical_b64url_and_private_jwk():
+    import base64
+    import hashlib
+    from datetime import UTC, datetime
+
+    from cryptography.hazmat.primitives.serialization import (
+        Encoding,
+        NoEncryption,
+        PrivateFormat,
+    )
+
+    from cmcp_runtime.audit.chain import AuditChain
+    from cmcp_runtime.audit.keys import SigningKey
+    from cmcp_runtime.audit.trace_claim import (
+        AttestationReportInfo,
+        CallGraphSummary,
+        CallSummary,
+        PolicyBundleInfo,
+        ToolCatalogInfo,
+        _to_dict,
+        generate_trace_claim,
+    )
+    from cmcp_verify.kill_switch import _b64url_decode
+    from cmcp_verify.verify import (
+        ApprovedHashes,
+        VerificationError,
+        VerificationStatus,
+        _canonical_json,
+        _verify_key_binding,
+        _verify_signature,
+        verify_trace_claim,
+    )
+
+    with pytest.raises(ValueError):
+        _b64url_decode("A" * 85 + "B")
+    with pytest.raises(ValueError):
+        _b64url_decode("A" * 42 + "B")
+
+    key = SigningKey()
+    chain = AuditChain("ks-test-session")
+    x_b64 = base64.urlsafe_b64encode(key.public_key_bytes).rstrip(b"=").decode("ascii")
+    jwk_fp = hashlib.sha256(
+        json.dumps(
+            {"crv": "Ed25519", "kty": "OKP", "x": x_b64},
+            separators=(",", ":"),
+            sort_keys=True,
+        ).encode()
+    ).digest()
+    report_data = (jwk_fp + hashlib.sha256(bytes.fromhex(chain.chain_root)).digest()).hex()
+
+    claim_obj = generate_trace_claim(
+        session_id="ks-test-session",
+        signing_key=key,
+        attestation_report=AttestationReportInfo(
+            provider="sev-snp",
+            measurement="ab" * 32,
+            report_data=report_data,
+            attestation_generated_at=datetime.now(tz=UTC).isoformat(),
+            attestation_validity_seconds=86400,
+        ),
+        policy_bundle=PolicyBundleInfo(
+            hash="sha256:" + "a" * 64,
+            enforcement_mode="enforcing",
+            policy_version="1.0.0",
+        ),
+        tool_catalog=ToolCatalogInfo(hash="sha256:" + "b" * 64),
+        call_summary=CallSummary(
+            tool_calls_total=0,
+            tool_calls_allowed=0,
+            tool_calls_denied=0,
+            tool_calls_faulted=0,
+            tools_invoked=[],
+            session_max_sensitivity="public",
+            call_graph_summary=CallGraphSummary(
+                compliance_domains_touched=[],
+                cross_boundary_events=[],
+            ),
+        ),
+        audit_chain_root=chain.chain_root,
+        audit_chain_tip=chain.chain_tip,
+        audit_chain_length=chain.length,
+        kill_switch_triggered=True,
+        do_sign=True,
+    )
+    claim = _to_dict(claim_obj)
+    approved = ApprovedHashes(
+        policy_bundle_hash="sha256:" + "a" * 64,
+        tool_catalog_hash="sha256:" + "b" * 64,
+    )
+
+    receipt = {
+        "type": REFUSAL_RECEIPT_TYPE,
+        "agent_id": _AGENT_ID,
+        "session_id": claim["gateway"]["session_id"],
+        "refused_at": datetime.now(tz=UTC).isoformat(),
+        "gateway_key": x_b64,
+        "claim_digest": "sha256:" + hashlib.sha256(_canonical_json(claim)).hexdigest(),
+    }
+    receipt["signature"] = (
+        base64.urlsafe_b64encode(key.sign(_canonical_json(receipt)))
+        .rstrip(b"=")
+        .decode("ascii")
+    )
+
+    # Intact signed control passes both public verifiers
+    control_res = verify_trace_claim(
+        claim, approved, trusted_public_key_hex=key.public_key_hex
+    )
+    assert "signature" in control_res.verified_fields
+    assert "public_key_binding" in control_res.verified_fields
+    assert verify_kill_switch_refusal(receipt, claim).valid is True
+
+    def _flip_pad_bit(b64_str: str) -> str:
+        alphabet = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_"
+        idx = alphabet.index(b64_str[-1])
+        return b64_str[:-1] + alphabet[idx ^ 1]
+
+    # Non-canonical signature alias (same decoded bytes, non-zero pad bit) fails verify_trace_claim
+    sig_alias_claim = copy.deepcopy(claim)
+    sig_alias_claim["signature"] = _flip_pad_bit(claim["signature"])
+    sig_res = verify_trace_claim(
+        sig_alias_claim, approved, trusted_public_key_hex=key.public_key_hex
+    )
+    assert "signature" in sig_res.unverified_fields
+    assert sig_res.failure_reason == VerificationError.SIGNATURE_INVALID
+
+    # Non-canonical key x alias fails verify_trace_claim and _verify_key_binding
+    key_alias_claim = copy.deepcopy(claim)
+    alias = _flip_pad_bit(x_b64)
+    assert base64.urlsafe_b64decode(alias + "=") == key.public_key_bytes
+    key_alias_claim["trace"]["cnf"]["jwk"]["x"] = alias
+    # Bind the altered textual JWK too, so a stale thumbprint cannot mask the guard.
+    alias_fp = hashlib.sha256(json.dumps(
+        {"crv": "Ed25519", "kty": "OKP", "x": alias},
+        separators=(",", ":"), sort_keys=True,
+    ).encode()).digest()
+    key_alias_claim["trace"]["runtime"]["nonce"] = base64.urlsafe_b64encode(
+        alias_fp + hashlib.sha256(bytes.fromhex(chain.chain_root)).digest()
+    ).rstrip(b"=").decode("ascii")
+    key_alias_claim["signature"] = base64.urlsafe_b64encode(
+        key.sign(_canonical_json(key_alias_claim))
+    ).rstrip(b"=").decode("ascii")
+    key_res = verify_trace_claim(
+        key_alias_claim, approved, trusted_public_key_hex=key.public_key_hex
+    )
+    assert "signature" in key_res.unverified_fields
+    assert _verify_key_binding(key_alias_claim, is_sw_only=False)[0] is False
+
+    # Freshly re-signed claim carrying private JWK member 'd' is rejected
+    priv_claim = copy.deepcopy(claim)
+    priv_raw = key._private.private_bytes(
+        Encoding.Raw, PrivateFormat.Raw, NoEncryption()
+    )
+    priv_claim["trace"]["cnf"]["jwk"]["d"] = (
+        base64.urlsafe_b64encode(priv_raw).rstrip(b"=").decode("ascii")
+    )
+    priv_claim["signature"] = (
+        base64.urlsafe_b64encode(key.sign(_canonical_json(priv_claim)))
+        .rstrip(b"=")
+        .decode("ascii")
+    )
+    priv_res = verify_trace_claim(
+        priv_claim, approved, trusted_public_key_hex=key.public_key_hex
+    )
+    assert priv_res.status != VerificationStatus.VERIFIED
+    assert "signature" not in priv_res.verified_fields
+    priv_receipt = copy.deepcopy(receipt)
+    priv_receipt["claim_digest"] = (
+        "sha256:" + hashlib.sha256(_canonical_json(priv_claim)).hexdigest()
+    )
+    priv_receipt["signature"] = (
+        base64.urlsafe_b64encode(key.sign(_canonical_json(priv_receipt)))
+        .rstrip(b"=")
+        .decode("ascii")
+    )
+    priv_ks_res = verify_kill_switch_refusal(priv_receipt, priv_claim)
+    assert priv_ks_res.valid is False
+    assert any("private" in e for e in priv_ks_res.errors)
+    ok, err = _verify_signature(priv_claim)
+    assert ok is False and err is not None and "private" in err
+    kb_ok, kb_err = _verify_key_binding(priv_claim, is_sw_only=False)
+    assert kb_ok is False and kb_err is not None and "private" in kb_err
+
+    # Empty or non-string required receipt fields and non-canonical receipt signature fail
+    for field in ("agent_id", "session_id", "refused_at"):
+        for bad_val in ("", None, 123):
+            bad_receipt = copy.deepcopy(receipt)
+            bad_receipt[field] = bad_val
+            bad_receipt["signature"] = (
+                base64.urlsafe_b64encode(key.sign(_canonical_json(bad_receipt)))
+                .rstrip(b"=")
+                .decode("ascii")
+            )
+            res = verify_kill_switch_refusal(bad_receipt, claim)
+            assert res.valid is False
+            assert any(f"receipt.{field}" in e for e in res.errors)
+
+    alias_receipt = copy.deepcopy(receipt)
+    alias_receipt["signature"] = _flip_pad_bit(receipt["signature"])
+    assert verify_kill_switch_refusal(alias_receipt, claim).valid is False