fix(transport): carry hardware attestation evidence over the reference wire codec

src/ca2a_runtime/transport/wire.py

@@ -9,6 +9,8 @@
 
 from __future__ import annotations
 
+import base64
+import re
 from typing import Any
 
 from ca2a_runtime.attestation import ChannelOffer
@@ -17,6 +19,45 @@
 from ca2a_runtime.provenance import DelegationRecord
 from ca2a_runtime.tee.base import AttestationReport
 
+# Unpadded base64url alphabet only, matching transport.a2a_adapter's convention:
+# padding is added back on decode, so an embedded "=" (or any other out-of-alphabet
+# character) is rejected as malformed rather than silently ignored.
+_BASE64URL_RE = re.compile(r"[A-Za-z0-9_-]*")
+
+# The AttestationReport fields that are *claims*: any peer can populate these
+# with any values, so they travel unconditionally.
+_CLAIM_FIELDS = ("platform", "measurement", "public_key", "nonce")
+
+# The AttestationReport fields that are *evidence*: absent on software-only
+# reports (and on older peers), present on every hardware provider (see
+# ca2a_runtime.tee.tpm/sev_snp/tdx), and required by a hardware Verifier
+# (e.g. ca2a_verify.tpm.tpm_verifier) to appraise a report at all. Omitted from
+# the wire body when absent, so a software-only offer's JSON is unchanged.
+_EVIDENCE_FIELDS = (
+    "raw_evidence",
+    "quote_signature",
+    "attestation_key_pem",
+    "attestation_key_chain_pem",
+)
+
+
+def _b64url_encode(value: bytes) -> str:
+    return base64.urlsafe_b64encode(value).decode("ascii").rstrip("=")
+
+
+def _b64url_decode(field: str, value: str) -> bytes:
+    """Decode a base64url string, failing closed on any non-base64url input."""
+    if not isinstance(value, str) or not _BASE64URL_RE.fullmatch(value):
+        raise TransportError(
+            f"{field} is not valid base64url",
+            detail=f"expected a base64url string, got {type(value).__name__}",
+        )
+    try:
+        padded = value + "=" * (-len(value) % 4)
+        return base64.urlsafe_b64decode(padded.encode("ascii"))
+    except (ValueError, UnicodeEncodeError) as exc:
+        raise TransportError(f"{field} is not valid base64url", detail=str(exc)) from exc
+
 
 def _record_to_dict(record: DelegationRecord) -> dict[str, Any]:
     body = record.body()
@@ -65,15 +106,24 @@ def serialize_channel_offer(offer: ChannelOffer, *, challenge: str | None = None
     ``challenge`` is the callee's half of a mutual exchange and is omitted when
     the callee issues none, so an older caller sees exactly the response it saw
     before. A caller that does not understand the field simply does not attest.
+
+    The report's evidence fields (``raw_evidence``, ``quote_signature``,
+    ``attestation_key_pem``, ``attestation_key_chain_pem``) are base64url-encoded
+    and included when present. Without them a hardware report cannot reach a
+    remote peer's ``Verifier`` at all: it fails closed with a misleading "no
+    quote to verify" rather than actually being appraised, even though the local
+    provider produced genuine evidence. They are omitted entirely (not sent as
+    null) when absent, so a software-only offer's JSON is byte-for-byte the same
+    as before this field existed.
     """
+    attestation: dict[str, Any] = {field: getattr(offer.report, field) for field in _CLAIM_FIELDS}
+    for field in _EVIDENCE_FIELDS:
+        value = getattr(offer.report, field)
+        if value is not None:
+            attestation[field] = _b64url_encode(value)
     body: dict[str, Any] = {
         "channel_public_key": offer.channel_public_key,
-        "attestation": {
-            "platform": offer.report.platform,
-            "measurement": offer.report.measurement,
-            "public_key": offer.report.public_key,
-            "nonce": offer.report.nonce,
-        },
+        "attestation": attestation,
     }
     if challenge is not None:
         body["challenge"] = challenge
@@ -89,11 +139,15 @@ def parse_channel_offer(data: dict[str, Any]) -> ChannelOffer:
     try:
         public_key = str(data["channel_public_key"])
         att = data["attestation"]
+        evidence = {
+            field: _b64url_decode(field, att[field]) for field in _EVIDENCE_FIELDS if field in att
+        }
         report = AttestationReport(
             platform=str(att["platform"]),
             measurement=str(att["measurement"]),
             public_key=str(att["public_key"]),
             nonce=str(att["nonce"]),
+            **evidence,
         )
     except (KeyError, TypeError) as exc:
         raise TransportError("malformed channel offer", detail=str(exc)) from exc

tests/unit/test_tpm_attest.py

@@ -29,7 +29,7 @@
 from cryptography.x509.oid import NameOID
 
 from ca2a_runtime.attestation import ChannelOffer, verify_offer
-from ca2a_runtime.errors import AttestationFailed, AttestationUnsupported
+from ca2a_runtime.errors import AttestationFailed, AttestationUnsupported, TransportError
 from ca2a_runtime.tee.base import AttestationReport
 from ca2a_runtime.tee.tpm import (
     TPM_GENERATED_VALUE,
@@ -38,6 +38,7 @@
     TpmQuote,
     tpm_qualifying_data,
 )
+from ca2a_runtime.transport import wire
 from ca2a_verify.tpm import (
     parse_tpmt_signature,
     tpm_verifier,
@@ -334,6 +335,78 @@ def test_tpm_offer_without_a_verifier_still_fails_closed() -> None:
         verify_offer(offer, expected_nonce=NONCE)
 
 
+def test_tpm_offer_survives_the_reference_transports_wire_codec() -> None:
+    """A hardware report must still reach assurance="hardware" after the exact
+    round trip the reference HTTP server/client run it through.
+
+    ``wire.serialize_channel_offer``/``parse_channel_offer`` is what
+    ``ca2a_runtime.transport.server`` sends on ``GET /.well-known/ca2a/channel``
+    and what ``ca2a_runtime.transport.client.handshake`` parses back. Encoding
+    only the claim fields (platform/measurement/public_key/nonce) and dropping
+    the evidence fields would make every hardware-attested offer unverifiable
+    over that transport, even though the provider produced a genuine quote.
+    """
+    report, root_pem = _report()
+    offer = ChannelOffer(channel_public_key=PUBLIC_KEY, report=report)
+
+    wire_body = wire.serialize_channel_offer(offer)
+    received = wire.parse_channel_offer(wire_body)
+
+    peer = verify_offer(received, expected_nonce=NONCE, verifier=tpm_verifier(root_pem))
+    assert peer.assurance == "hardware"
+    assert peer.measurement == "sha256:" + ("11" * 32)
+
+
+def test_software_only_offer_wire_body_has_no_evidence_keys() -> None:
+    """The evidence fields must be omitted, not sent as null, when absent -- so
+    a software-only offer's JSON is unchanged from before evidence traveled."""
+    bare = AttestationReport(
+        platform="software-only",
+        measurement="software-only-no-hardware-guarantee",
+        public_key=PUBLIC_KEY,
+        nonce=NONCE,
+    )
+    offer = ChannelOffer(channel_public_key=PUBLIC_KEY, report=bare)
+    body = wire.serialize_channel_offer(offer)
+    assert set(body["attestation"]) == {"platform", "measurement", "public_key", "nonce"}
+
+
+def test_parse_channel_offer_rejects_evidence_with_invalid_alphabet() -> None:
+    """A malformed peer (or an attacker) can put anything in the JSON body.
+    A character outside the base64url alphabet (here "!") must fail closed
+    with a clear TransportError, not an uncaught exception."""
+    body = {
+        "channel_public_key": PUBLIC_KEY,
+        "attestation": {
+            "platform": "tpm",
+            "measurement": "sha256:" + ("11" * 32),
+            "public_key": PUBLIC_KEY,
+            "nonce": NONCE,
+            "raw_evidence": "not-valid-base64url!!",
+        },
+    }
+    with pytest.raises(TransportError, match="raw_evidence is not valid base64url"):
+        wire.parse_channel_offer(body)
+
+
+def test_parse_channel_offer_rejects_evidence_with_bad_length() -> None:
+    """A string that only uses base64url-alphabet characters can still be an
+    invalid length (e.g. a single character can never be valid base64). That
+    must also fail closed with a TransportError, not a raw binascii.Error."""
+    body = {
+        "channel_public_key": PUBLIC_KEY,
+        "attestation": {
+            "platform": "tpm",
+            "measurement": "sha256:" + ("11" * 32),
+            "public_key": PUBLIC_KEY,
+            "nonce": NONCE,
+            "quote_signature": "A",
+        },
+    }
+    with pytest.raises(TransportError, match="quote_signature is not valid base64url"):
+        wire.parse_channel_offer(body)
+
+
 # ── collector checks ──────────────────────────────────────────────────────────