prototype(verifier_token,execution_appraisal): enforce canonical 32-byte nonces, validate RelyingParty.challenge/verify_proof inputs, and reject non-finite JWT JSON (#629)

prototype/cmcp_gate.py

@@ -187,6 +187,10 @@ def _prove(
         nonce = credentials["nonce"]
         if not isinstance(nonce, str) or len(nonce) != 43:
             raise v.ProfileError("nonce_shape")
+        try:
+            self._decode(nonce)
+        except v.ProfileError as exc:
+            raise v.ProfileError("nonce_shape") from exc
         proof = self._decode(credentials["proof"])
         record = self._store.consume_challenge(nonce)
         if record is None:

prototype/execution_appraisal.py

@@ -39,6 +39,10 @@ def _pairs(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
     return result
 
 
+def _no_nonfinite(_token: str) -> Any:
+    raise ExecutionDenied("jwt_framing")
+
+
 def _decode(value: Any) -> bytes:
     if not isinstance(value, str) or not re.fullmatch(r"[A-Za-z0-9_-]+", value):
         raise ExecutionDenied("invalid_base64url")
@@ -96,7 +100,9 @@ def authenticate_token(token: str, jwks: dict[str, Any]) -> dict[str, Any]:
     if len(parts) != 3:
         raise ExecutionDenied("jwt_framing")
     try:
-        header = json.loads(_decode(parts[0]), object_pairs_hook=_pairs)
+        header = json.loads(
+            _decode(parts[0]), object_pairs_hook=_pairs, parse_constant=_no_nonfinite
+        )
     except (UnicodeDecodeError, json.JSONDecodeError) as error:
         raise ExecutionDenied("jwt_framing") from error
     if not isinstance(header, dict) or header.get("alg") != "RS256":
@@ -133,7 +139,9 @@ def authenticate_token(token: str, jwks: dict[str, Any]) -> dict[str, Any]:
     except (InvalidSignature, ValueError) as error:
         raise ExecutionDenied("signature") from error
     try:
-        claims = json.loads(_decode(parts[1]), object_pairs_hook=_pairs)
+        claims = json.loads(
+            _decode(parts[1]), object_pairs_hook=_pairs, parse_constant=_no_nonfinite
+        )
     except (UnicodeDecodeError, json.JSONDecodeError) as error:
         raise ExecutionDenied("jwt_framing") from error
     if not isinstance(claims, dict):
@@ -157,7 +165,12 @@ def appraise_launch(
     try:
         if not isinstance(packet, dict):
             raise ExecutionDenied("malformed_or_unauthenticated_evidence")
-        if type(now) is not int or now < 0 or type(policy.max_age) is not int or policy.max_age < 1:
+        if (
+            type(now) is not int
+            or not 0 <= now <= 9007199254740991
+            or type(policy.max_age) is not int
+            or policy.max_age < 1
+        ):
             raise ExecutionDenied("clock_or_policy")
         if not re.fullmatch(r"sha256:[0-9a-f]{64}", policy.image_digest):
             raise ExecutionDenied("policy_digest")
@@ -179,6 +192,8 @@ def appraise_launch(
         if any(type(value) is not int for value in times):
             raise ExecutionDenied("time_type")
         issued, not_before, expiry = (cast(int, value) for value in times)
+        if any(not 0 <= t <= 9007199254740991 for t in (issued, not_before, expiry)):
+            raise ExecutionDenied("time_bounds")
         if not (issued <= now < expiry and not_before <= now and issued <= expiry):
             raise ExecutionDenied("time_bounds")
         if now - issued >= policy.max_age:

prototype/verifier_token.py

@@ -234,6 +234,12 @@ class Proof(Closed):
     issued_at: Time
     expires_at: Time
 
+    @model_validator(mode="after")
+    def check_canonical_nonce(self) -> Proof:
+        if len(_b64url(self.nonce)) != 32:
+            raise ValueError("noncanonical_nonce")
+        return self
+
 
 class Receipt(Closed):
     profile: Literal["urn:agentrust:trace:decision-receipt:experimental-v1"]
@@ -757,6 +763,22 @@ def verify_proof(
     challenge_expires_at: int,
     now: int,
 ) -> Proof:
+    if (
+        type(now) is not int
+        or type(challenge_issued_at) is not int
+        or type(challenge_expires_at) is not int
+        or not 0 <= now <= 9007199254740991
+        or not 0 <= challenge_issued_at <= 9007199254740991
+        or not 0 <= challenge_expires_at <= 9007199254740991
+        or not isinstance(nonce, str)
+        or not re.fullmatch(r"[A-Za-z0-9_-]{43}", nonce)
+    ):
+        raise ProfileError("verification_inputs_invalid")
+    try:
+        if len(_b64url(nonce)) != 32:
+            raise ValueError("noncanonical_nonce")
+    except ValueError as exc:
+        raise ProfileError("verification_inputs_invalid") from exc
     proof: Proof = authenticate(proof_bytes, PROOF_PROFILE, holder_public(token.cnf), Proof)
     if (
         proof.nonce != nonce
@@ -793,9 +815,23 @@ def __init__(self, context: VerificationContext) -> None:
     def challenge(
         self, *, nonce: str, session_id: str, action_digest: str, issued_at: int, expires_at: int
     ) -> None:
-        if not re.fullmatch(r"[A-Za-z0-9_-]{43}", nonce):
+        if not isinstance(nonce, str) or not re.fullmatch(r"[A-Za-z0-9_-]{43}", nonce):
             raise ProfileError("nonce_shape")
-        if not issued_at < expires_at <= issued_at + 30:
+        try:
+            if len(_b64url(nonce)) != 32:
+                raise ValueError("noncanonical_nonce")
+        except ValueError as exc:
+            raise ProfileError("nonce_shape") from exc
+        try:
+            TypeAdapter(Identifier).validate_python(session_id, strict=True)
+            TypeAdapter(Digest).validate_python(action_digest, strict=True)
+        except ValueError as exc:
+            raise ProfileError("input_invalid") from exc
+        if (
+            type(issued_at) is not int
+            or type(expires_at) is not int
+            or not 0 <= issued_at < expires_at <= min(issued_at + 30, 9007199254740991)
+        ):
             raise ProfileError("challenge_interval")
         with self._lock:
             self._used_nonces = {n: end for n, end in self._used_nonces.items() if end > issued_at}

tests/test_629_independent_rejections.py

@@ -0,0 +1,103 @@
+"""Supplemental independent controls for the three review gaps in #634."""
+
+from dataclasses import replace
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from prototype import execution_appraisal as appraisal
+from prototype import verifier_token as v
+from prototype.cmcp_gate import CMCPTraceGate
+from tests.test_execution_appraisal import CHALLENGE, NOW, case
+from tests.test_verifier_token_profile import context, fixture, g, proof, unb64
+
+__all__ = ["case", "fixture"]
+
+
+@pytest.mark.parametrize("value", [float("nan"), float("inf"), float("-inf")])
+def test_authenticated_claims_body_rejects_nonfinite_json(case, value):
+    claims, keys, _, packet, _ = case
+    assert appraisal.authenticate_token(packet()["token"], keys)["iat"] == NOW
+    bad = dict(claims, extra={"nested": [value]})
+    signed = packet(bad)
+    with pytest.raises(appraisal.ExecutionDenied, match="jwt_framing"):
+        appraisal.authenticate_token(signed["token"], keys)
+
+
+@pytest.mark.parametrize(
+    ("field", "value"), [("iat", -1), ("nbf", -1), ("exp", 2**53)]
+)
+def test_signed_claim_time_bounds_are_reached_independently(case, field, value):
+    claims, keys, policy, packet, _ = case
+    policy = replace(policy, max_age=2000)
+    assert appraisal.appraise_launch(
+        packet(), keys, policy, expected_challenge=CHALLENGE, now=NOW
+    )["observed_digest"] == policy.image_digest
+    with pytest.raises(appraisal.ExecutionDenied, match="time_bounds"):
+        appraisal.appraise_launch(
+            packet(dict(claims, **{field: value})),
+            keys, policy, expected_challenge=CHALLENGE, now=NOW,
+        )
+
+
+def test_gate_refuses_nonce_alias_before_store_consumption(fixture, tmp_path, monkeypatch):
+    monkeypatch.setattr("prototype.cmcp_gate.secrets.token_urlsafe", lambda _n: "A" * 43)
+    raw = unb64(fixture["envelope_b64"])
+    gate = CMCPTraceGate(
+        context(fixture),
+        database=tmp_path / "gate.sqlite",
+        gateway_key=Ed25519PrivateKey.from_private_bytes(bytes(range(96, 128))),
+        gateway_issuer="spiffe://example.test/gateway",
+        gateway_policy_digest="sha256:" + "a" * 64,
+        clock=lambda: g.NOW,
+    )
+    action = {"domain": "cmcp:trace-admission:experimental-v1", "session_id": "session-1"}
+    challenge = gate.challenge(raw, session_id="session-1", action=action)
+    payload = {
+        "profile": v.PROOF_PROFILE, "nonce": challenge["nonce"],
+        "token_digest": v.digest(raw), "token_id": challenge["token_id"],
+        "audience": challenge["audience"], "session_id": challenge["session_id"],
+        "action_digest": challenge["action_digest"], "issued_at": challenge["issued_at"],
+        "expires_at": challenge["expires_at"],
+    }
+    credentials = {
+        "nonce": challenge["nonce"],
+        "proof": CMCPTraceGate._encode(g.envelope(payload, g.HOLDER)),
+    }
+    calls = []
+    original = gate._store.consume_challenge
+
+    def observe(nonce):
+        calls.append(nonce)
+        return original(nonce)
+
+    monkeypatch.setattr(gate._store, "consume_challenge", observe)
+    with pytest.raises(v.ProfileError, match="nonce_shape"):
+        gate.admit(raw, dict(credentials, nonce="A" * 42 + "B"), session_id="session-1")
+    assert calls == []
+    assert gate.admit(raw, credentials, session_id="session-1")["generation"] == 1
+    assert calls == ["A" * 43]
+
+
+@pytest.mark.parametrize(
+    ("change", "code"),
+    [
+        ({"nonce": 123}, "nonce_shape"),
+        ({"issued_at": False, "expires_at": 1}, "challenge_interval"),
+        ({"issued_at": 0, "expires_at": True}, "challenge_interval"),
+        ({"issued_at": 2**53 - 10, "expires_at": 2**53 + 5}, "challenge_interval"),
+    ],
+)
+def test_challenge_refuses_each_new_input(fixture, change, code):
+    body, _ = proof(fixture)
+    kwargs = {
+        "nonce": "A" * 43,
+        "session_id": "session-1",
+        "action_digest": body["action_digest"],
+        "issued_at": g.NOW,
+        "expires_at": g.NOW + 30,
+    }
+    v.RelyingParty(context(fixture)).challenge(**kwargs)
+    with pytest.raises(v.ProfileError, match=f"^{code}$"):
+        v.RelyingParty(context(fixture)).challenge(**{**kwargs, **change})
+

tests/test_verifier_token_api_contract.py

@@ -169,3 +169,124 @@ def test_issuer_configuration_refuses_malformed_fields(fixture, field):
     for value in [None, {}, [], True]:
         with pytest.raises(v.ProfileError, match="issuer_configuration_invalid"):
             replace(issuer, **{field: value})
+
+
+def test_canonical_32_byte_nonce_enforced_in_proof_and_relying_party(fixture):
+    proof_body, signed_proof = proof(fixture)
+    ctx = context(fixture)
+    rp = v.RelyingParty(ctx)
+    # Valid 32-byte canonical base64url nonce succeeds
+    rp.challenge(
+        nonce="A" * 43,
+        session_id="session-1",
+        action_digest=proof_body["action_digest"],
+        issued_at=g.NOW,
+        expires_at=g.NOW + 30,
+    )
+    # 43-char string with non-zero unused pad bits ("A"*42 + "B") is rejected
+    with pytest.raises(v.ProfileError, match="nonce_shape"):
+        rp.challenge(
+            nonce="A" * 42 + "B",
+            session_id="session-1",
+            action_digest=proof_body["action_digest"],
+            issued_at=g.NOW,
+            expires_at=g.NOW + 30,
+        )
+    bad_proof = dict(proof_body, nonce="A" * 42 + "B")
+    with pytest.raises(ValueError):
+        v.Proof.model_validate(bad_proof)
+
+
+def test_relying_party_challenge_and_verify_proof_reject_boolean_or_invalid_inputs(fixture):
+    raw = unb64(fixture["envelope_b64"])
+    token = v.Token.model_validate(fixture["token"])
+    ctx = context(fixture)
+    proof_body, signed_proof = proof(fixture)
+    rp = v.RelyingParty(ctx)
+
+    for idx, bad_session in enumerate(("", 123, None)):
+        with pytest.raises(v.ProfileError, match="input_invalid"):
+            rp.challenge(
+                nonce=f"{idx}A" + "A" * 41,
+                session_id=bad_session,  # type: ignore[arg-type]
+                action_digest=proof_body["action_digest"],
+                issued_at=g.NOW,
+                expires_at=g.NOW + 30,
+            )
+
+    with pytest.raises(v.ProfileError, match="input_invalid"):
+        rp.challenge(
+            nonce="Q" * 43,
+            session_id="session-1",
+            action_digest="not-a-digest",
+            issued_at=g.NOW,
+            expires_at=g.NOW + 30,
+        )
+
+    for idx, (bad_issued, bad_expires) in enumerate(
+        ((False, True), (-10, -5), (g.NOW, True), (g.NOW, 2**53))
+    ):
+        with pytest.raises(v.ProfileError, match="challenge_interval"):
+            rp.challenge(
+                nonce=f"{idx}Q" + "A" * 41,
+                session_id="session-1",
+                action_digest=proof_body["action_digest"],
+                issued_at=bad_issued,
+                expires_at=bad_expires,
+            )
+
+    for kw, bad_val in (
+        ("now", True),
+        ("challenge_issued_at", False),
+        ("challenge_expires_at", True),
+        ("now", 2**53),
+        ("nonce", "A" * 42 + "B"),
+    ):
+        kwargs = {
+            "proof_bytes": signed_proof,
+            "token_bytes": raw,
+            "token": token,
+            "nonce": proof_body["nonce"],
+            "audience": ctx.audience,
+            "session_id": "session-1",
+            "action_digest": proof_body["action_digest"],
+            "challenge_issued_at": 1,
+            "challenge_expires_at": 2,
+            "now": 1,
+        }
+        kwargs[kw] = bad_val
+        with pytest.raises(v.ProfileError, match="verification_inputs_invalid"):
+            v.verify_proof(**kwargs)  # type: ignore[arg-type]
+
+
+def test_execution_appraisal_rejects_nonfinite_jwt_json_and_unsafe_integer_times():
+    import base64
+    from prototype.execution_appraisal import (
+        ExecutionDenied,
+        LaunchPolicy,
+        appraise_launch,
+        authenticate_token,
+    )
+
+    def _b64u(data: bytes) -> str:
+        return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")
+
+    bad_header = _b64u(b'{"alg":"RS256","kid":"k1","x":NaN}') + "." + _b64u(b"{}") + ".AAAA"
+    with pytest.raises(ExecutionDenied, match="jwt_framing"):
+        authenticate_token(bad_header, {"keys": []})
+
+    policy = LaunchPolicy(
+        audience="aud-1",
+        image_digest="sha256:" + "a" * 64,
+        args=("run",),
+        subject="https://www.googleapis.com/compute/v1/projects/p1/zones/us-central1-a/instances/i1",
+        instance_id="12345",
+    )
+    with pytest.raises(ExecutionDenied, match="clock_or_policy"):
+        appraise_launch(
+            {},
+            {"keys": []},
+            policy,
+            expected_challenge="a" * 64,
+            now=2**53,
+        )