Certificate Mismatch¶

# Got token with cert1
with open("client1.crt") as f:
    cert1 = f.read()

token = request_token_with_mtls(cert="client1.crt", key="client1.key")

# Later, rotated to cert2 but still using old token
response = requests.post(
    "https://arcp.example.com/agents/register",
    headers={"Authorization": f"Bearer {token}"},  # Bound to cert1
    cert=("client2.crt", "client2.key"),  # Using cert2
    json=registration_data
)
# Result: mtls-binding-mismatch error

# Agent has certificates for different purposes
cert_production = "prod-client.crt"
cert_staging = "staging-client.crt"

# Got token with production cert
token = request_token_with_mtls(cert=cert_production, key="prod-client.key")

# Accidentally used staging cert for request
response = requests.post(
    url,
    headers={"Authorization": f"Bearer {token}"},
    cert=(cert_staging, "staging-client.key"),  # Wrong cert!
    json=data
)

# Load balancer terminates TLS with one cert
# App receives different cert in X-SSL-Client-Cert header
# Token was bound to client's original cert
# Result: mismatch

class MTLSSession:
    def __init__(self, cert_file, key_file, ca_file):
        """Maintain consistent mTLS session."""
        self.cert_file = cert_file
        self.key_file = key_file
        self.ca_file = ca_file
        self.token = None
        self.spki_hash = self._calculate_spki_hash()

    def _calculate_spki_hash(self):
        """Calculate SPKI hash of certificate."""
        from cryptography import x509
        from cryptography.hazmat.backends import default_backend
        import hashlib
        import base64

        with open(self.cert_file, 'rb') as f:
            cert = x509.load_pem_x509_certificate(f.read(), default_backend())

        spki = cert.public_key().public_bytes(
            encoding=serialization.Encoding.DER,
            format=serialization.PublicFormat.SubjectPublicKeyInfo
        )

        hash_bytes = hashlib.sha256(spki).digest()
        return base64.urlsafe_b64encode(hash_bytes).decode().rstrip('=')

    def get_token(self):
        """Get mTLS-bound token."""
        if not self.token or self._is_token_expired():
            # Request token with mTLS
            response = requests.post(
                "https://arcp.example.com/auth/agent/request_temp_token",
                cert=(self.cert_file, self.key_file),
                verify=self.ca_file,
                json={
                    "agent_id": "my-agent",
                    "agent_type": "processing",
                    "agent_key": "secret-key"
                }
            )
            self.token = response.json()["temp_token"]
        return self.token

    def make_request(self, method, url, **kwargs):
        """Make request with consistent mTLS."""
        token = self.get_token()

        # Use SAME certificate
        return requests.request(
            method,
            url,
            headers={
                "Authorization": f"Bearer {token}",
                **kwargs.pop("headers", {})
            },
            cert=(self.cert_file, self.key_file),  # Same cert!
            verify=self.ca_file,
            **kwargs
        )

import hashlib
import base64
from cryptography import x509
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization
import jwt

def calculate_cert_spki_hash(cert_file):
    """Calculate SPKI hash of certificate."""
    with open(cert_file, 'rb') as f:
        cert = x509.load_pem_x509_certificate(f.read(), default_backend())

    # Get public key SPKI
    spki = cert.public_key().public_bytes(
        encoding=serialization.Encoding.DER,
        format=serialization.PublicFormat.SubjectPublicKeyInfo
    )

    # SHA-256 hash
    hash_bytes = hashlib.sha256(spki).digest()

    # Base64url encode
    return base64.urlsafe_b64encode(hash_bytes).decode().rstrip('=')

# Verify match
token_payload = jwt.decode(token, options={"verify_signature": False})
token_spki = token_payload["cnf"]["x5t#S256"]
cert_spki = calculate_cert_spki_hash("client.crt")

assert token_spki == cert_spki, f"Certificate mismatch!\nToken: {token_spki}\nCert: {cert_spki}"

# If you need to rotate certificates
old_cert = ("old-client.crt", "old-client.key")
new_cert = ("new-client.crt", "new-client.key")

# Get new token with new certificate
response = requests.post(
    "https://arcp.example.com/auth/agent/request_temp_token",
    cert=new_cert,
    verify="ca.crt",
    json={
        "agent_id": "my-agent",
        "agent_type": "processing",
        "agent_key": "secret-key"
    }
)

new_token = response.json()["temp_token"]

# Use new certificate with new token
response = requests.post(
    "https://arcp.example.com/agents/register",
    headers={"Authorization": f"Bearer {new_token}"},
    cert=new_cert,
    verify="ca.crt",
    json=registration_data
)

from examples.agents.mtls_helper import MTLSClientHelper

# Helper automatically maintains certificate consistency
helper = MTLSClientHelper(
    cert_file="client.crt",
    key_file="client.key",
    ca_file="ca.crt"
)

# Get token (uses helper's cert)
token = helper.request_temp_token(
    agent_id="my-agent",
    agent_type="processing",
    agent_key="secret-key"
)

# Make requests (uses same cert)
response = helper.post(
    "https://arcp.example.com/agents/register",
    json=registration_data
)

# SPKI Hash = Base64url(SHA256(SubjectPublicKeyInfo))
cert → public_key → SPKI (DER) → SHA-256 → Base64url → hash

# ARCP validates:
1. Extract x5t#S256 from token's cnf claim
2. Extract client certificate from request
3. Calculate SPKI hash from certificate
4. Compare: token_spki == cert_spki

# Safe certificate rotation process:

# 1. Generate new certificate
new_cert, new_key = generate_new_certificate()

# 2. Request new token with new certificate
new_token = request_token_with_mtls(new_cert, new_key)

# 3. Switch to new certificate and token
session.update(cert=new_cert, key=new_key, token=new_token)

# 4. Revoke old certificate (optional)
revoke_certificate(old_cert)

import jwt

# Decode token to see binding
payload = jwt.decode(token, options={"verify_signature": False})
print("Token SPKI:", payload.get("cnf", {}).get("x5t#S256"))

def debug_mtls_binding(token, cert_file):
    """Debug mTLS binding mismatch."""
    # Token SPKI
    token_payload = jwt.decode(token, options={"verify_signature": False})
    token_spki = token_payload.get("cnf", {}).get("x5t#S256")

    # Certificate SPKI
    cert_spki = calculate_cert_spki_hash(cert_file)

    print(f"Token SPKI:  {token_spki}")
    print(f"Cert SPKI:   {cert_spki}")
    print(f"Match:       {token_spki == cert_spki}")

    if token_spki != cert_spki:
        print("\n⚠️  Mismatch detected!")
        print("Token is bound to a different certificate.")
        print("You need to request a new token with the current certificate.")