From 4ab488a135e516cc141e61bc268fcd08d52c8c0a Mon Sep 17 00:00:00 2001 From: Raghuvar Karthik Durga <177662415+drk-ram@users.noreply.github.com> Date: Sat, 31 Oct 2026 10:52:56 -0500 Subject: [PATCH] fix(verify): enforce canonical base64url and reject private JWK members in kill_switch and claim verification (#804) - Enforce canonical unpadded base64url in cmcp_verify.kill_switch._b64url_decode and require non-empty string agent_id, session_id, and refused_at on refusal receipts. - Reject RFC 7517/7518/8037 private or symmetric JWK parameters on trace.cnf.jwk and enforce canonical unpadded base64url on trace.cnf.jwk.x and signature in cmcp_verify.verify. Closes #804 Signed-off-by: Raghuvar Karthik Durga <177662415+drk-ram@users.noreply.github.com> --- src/cmcp_verify/kill_switch.py | 16 +- src/cmcp_verify/verify.py | 52 +++--- tests/unit/test_kill_switch_evidence.py | 202 ++++++++++++++++++++++++ 3 files changed, 250 insertions(+), 20 deletions(-) diff --git a/src/cmcp_verify/kill_switch.py b/src/cmcp_verify/kill_switch.py index b7ab1734..ad3152f4 100644 --- a/src/cmcp_verify/kill_switch.py +++ b/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"), diff --git a/src/cmcp_verify/verify.py b/src/cmcp_verify/verify.py index 38eac917..b7a754f8 100644 --- a/src/cmcp_verify/verify.py +++ b/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}" diff --git a/tests/unit/test_kill_switch_evidence.py b/tests/unit/test_kill_switch_evidence.py index c9d57b2a..76b58586 100644 --- a/tests/unit/test_kill_switch_evidence.py +++ b/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