Sécurité entre agents : authentification et confiance

Qui a le droit d'appeler qui ?

Sans protection, n'importe quel programme qui atteint le port 8102 peut demander des prix, et n'importe qui atteignant 8100 peut consommer vos appels au modèle. On distingue :

• authentification : prouver qui appelle (un jeton, une identité de service) ;
• autorisation : décider ce que cet appelant a le droit de faire ;
• confiance dans les réponses : ce qu'un agent distant renvoie reste une donnée à vérifier, pas un ordre.

sequenceDiagram
  participant O as "Orchestrateur"
  participant MW as "Middleware BearerAuth"
  participant P as "Agent tarification"
  O->>MW: "GET /.well-known/agent-card.json"
  MW->>P: "chemin public : laissé passer"
  P-->>O: "Agent Card"
  O->>MW: "POST / + Authorization: Bearer jeton"
  MW->>MW: "hmac.compare_digest(reçu, attendu)"
  alt "jeton valide"
    MW->>P: "requête JSON-RPC"
    P-->>O: "tâche"
  else "absent ou faux"
    MW-->>O: "401 unauthorized"
  end

Côté serveur : un middleware ASGI de quinze lignes

class BearerAuth:
    """Middleware ASGI : l'Agent Card reste publique, l'appel JSON-RPC exige le jeton."""

    def __init__(self, app, token: str):
        self.app, self.expected = app, f"Bearer {token}".encode()

    async def __call__(self, scope, receive, send):
        if scope["type"] == "http" and not scope["path"].endswith(CARD_PATH):
            got = dict(scope["headers"]).get(b"authorization", b"")
            if not hmac.compare_digest(got, self.expected):  # comparaison à temps constant
                await send({"type": "http.response.start", "status": 401,
                            "headers": [(b"content-type", b"application/json")]})
                await send({"type": "http.response.body", "body": b'{"error":"unauthorized"}'})
                return
        await self.app(scope, receive, send)

Trois détails comptent :

1. hmac.compare_digest compare en temps constant : un == ordinaire s'arrête au premier caractère différent, ce qui peut, en théorie, laisser deviner le jeton en mesurant les temps de réponse.
2. Le jeton est refusé s'il fait moins de 16 caractères (a2a_token() lève une erreur) : un jeton faible ne doit pas démarrer en silence.
3. La carte reste publique pour la découverte. Si elle doit être privée, le RemoteA2aAgent d'ADK 2.10 envoie aussi le jeton lors de la lecture de la carte.

Côté client : auth_scheme et auth_credential

def bearer_auth(token: str):
    """Schéma + identifiant : ADK ajoute « Authorization: Bearer … » à chaque appel A2A."""
    cred = AuthCredential(auth_type=AuthCredentialTypes.HTTP,
                          http=HttpAuth(scheme="bearer", credentials=HttpCredentials(token=token)))
    return HTTPBearer(), cred

ADK résout l'identifiant une fois par invocation et l'ajoute à la lecture de la carte et aux messages. Le code installé dérive aussi une clé de cache par agent distant, pour que le jeton d'un agent ne soit jamais envoyé à un autre hôte. Au lancement, ADK affiche des avertissements [EXPERIMENTAL] (PLUGGABLE_AUTH, CredentialManager) : la fonctionnalité marche, mais son API peut changer.

Résultat obtenu

Le 2026-10-15, scénario 2 de e2e_a2a.py, et journal du serveur stock pendant les essais :

2) Appel sans jeton
   HTTP 401

INFO:     127.0.0.1:35282 - "POST / HTTP/1.1" 401 Unauthorized
INFO:     127.0.0.1:35284 - "POST / HTTP/1.1" 200 OK

Les tests sans clé couvrent les trois cas : sans jeton (401), mauvais jeton (401), bon jeton (le middleware laisse passer et c'est le protocole qui répond).

En production : des identités, pas un secret partagé

Un jeton partagé a deux défauts : tous les appelants se ressemblent (impossible de savoir qui a appelé), et une fuite oblige à changer le secret partout. Sur Google Cloud :

Avec Cloud Run déployé en --no-allow-unauthenticated, c'est Cloud Run qui vérifie le jeton d'identité avant votre code : le middleware du lab devient inutile (il entrerait même en conflit, puisque les deux utilisent l'en-tête Authorization). L'orchestrateur obtient un jeton d'identité dont l'audience est l'URL du service appelé, par exemple avec google.oauth2.id_token.fetch_id_token(requête, url_du_service), et l'envoie à chaque appel. Ce jeton expire : il faut le renouveler, ce qu'un identifiant statique passé à auth_credential ne fait pas.

La confiance dans les réponses

Être authentifié ne rend pas une réponse fiable. L'orchestrateur relaie à son modèle le texte des spécialistes : si un spécialiste est compromis ou bogué, son texte peut contenir n'importe quoi. Règles de base :

• n'accordez à l'orchestrateur aucun outil à effet (paiement, suppression) qu'un texte d'agent distant pourrait déclencher sans contrôle en code ;
• traitez les réponses comme des données : validez les nombres côté code quand c'est possible ;
• limitez ce que les spécialistes renvoient : l'historique des tâches expose leurs appels d'outils internes (leçon 2 du module 1).

Exercice : un jeton par appelant

Faites évoluer BearerAuth pour accepter plusieurs jetons, chacun associé au nom d'un appelant, et journaliser qui appelle.

class BearerAuth:
    def __init__(self, app, tokens: dict[str, str]):   # {"orchestrator": "…", "batch": "…"}
        self.app = app
        self.expected = {f"Bearer {t}".encode(): name for name, t in tokens.items()}

    async def __call__(self, scope, receive, send):
        if scope["type"] == "http" and not scope["path"].endswith(CARD_PATH):
            got = dict(scope["headers"]).get(b"authorization", b"")
            caller = next((n for tok, n in self.expected.items() if hmac.compare_digest(got, tok)), None)
            if caller is None:
                ...  # même réponse 401 qu'avant
                return
            scope.setdefault("state", {})["caller"] = caller
            logging.info("appel A2A de %s", caller)
        await self.app(scope, receive, send)

Je sais faire

• Protéger un serveur A2A par un middleware à comparaison en temps constant
• Configurer auth_scheme et auth_credential sur un RemoteA2aAgent
• Expliquer pourquoi passer à IAM en production
• Traiter la réponse d'un agent distant comme une donnée, pas comme un ordre