feat(auth): let BearerTokenAuth accept a non-bearer principal resolver (trusted-proxy / embedded identity) on both legs #1304

import json

from starlette.testclient import TestClient

from adcp.server import ADCPHandler
from adcp.server.auth import (
    REQUEST_STATE_PRINCIPAL,
    BearerTokenAuth,
    auth_context_factory,
    current_principal,
)
from adcp.server.serve import _build_mcp_and_a2a_app  # the builder serve(transport="both") uses

seen = []


class Seller(ADCPHandler):
    async def get_products(self, params, context=None):
        seen.append(context.caller_identity)
        return {"products": []}


class TrustedHeaderIdentity:
    """Adopter middleware: the trusted proxy has authenticated the caller."""

    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        principal = dict(scope.get("headers", ())).get(b"x-principal-id")
        if scope["type"] == "http" and principal:
            scope.setdefault("state", {})[REQUEST_STATE_PRINCIPAL] = principal.decode()
            current_principal.set(principal.decode())
        await self.app(scope, receive, send)


def build(allow_unauthenticated):
    app = _build_mcp_and_a2a_app(
        Seller(), name="seller", port=3001, host="0.0.0.0", instructions=None,
        test_controller=None, validation=None, stateless_http=True,
        enable_dns_rebinding_protection=False, context_factory=auth_context_factory,
        auth=BearerTokenAuth(validate_token=lambda token: None,
                             allow_unauthenticated=allow_unauthenticated),
    )
    return TrustedHeaderIdentity(app)


HEADERS = {"content-type": "application/json", "accept": "application/json, text/event-stream",
           "x-principal-id": "buyer-42", "A2A-Version": "1.0"}
PARAMS = {"buying_mode": "brief", "brief": "video"}
MCP_CALL = {"jsonrpc": "2.0", "id": 1, "method": "tools/call",
            "params": {"name": "get_products", "arguments": PARAMS}}
A2A_CALL = {"jsonrpc": "2.0", "id": 1, "method": "SendMessage", "params": {"message": {
    "messageId": "m1", "role": "ROLE_USER",
    "parts": [{"data": {"skill": "get_products", "parameters": PARAMS}}]}}}

for allow in (False, True):
    with TestClient(build(allow)) as client:
        for leg, path, body in (("MCP", "/mcp", MCP_CALL), ("A2A", "/", A2A_CALL)):
            seen.clear()
            response = client.post(path, headers=HEADERS, content=json.dumps(body))
            print(f"allow_unauthenticated={allow!s:5} {leg}: HTTP {response.status_code}, "
                  f"handler caller_identity={seen[0] if seen else '(handler not called)'!r}")

BearerTokenAuth(
    validate_token=...,
    resolve_principal=lambda request: Principal | None,  # sync or async
)