Limites de sécurité dans les intégrations MCP

Une intégration MCP connecte un processus de décision contrôlé par une IA à des outils, des ressources et des prompts exposés par des serveurs. Cette connexion constitue une limite de sécurité, et pas seulement une commodité d’API. Les arguments d’outils générés par le modèle doivent être considérés comme des entrées non fiables, car un modèle peut mal comprendre une demande, suivre un contenu malveillant ou sélectionner un outil aux conséquences imprévues. Les données renvoyées par un serveur MCP sont également non fiables : elles peuvent contenir des informations sensibles, des structures malformées ou des instructions conçues pour influencer le comportement ultérieur du modèle.

Un modèle de menace utile distingue l’utilisateur, l’application hôte, le modèle, le client MCP, le serveur MCP et le système en aval. L’authentification établit une identité, tandis que l’autorisation détermine ce que cette identité peut faire. Le modèle ne doit jamais hériter de toutes les permissions détenues par le processus de l’application. Par exemple, un assistant du support peut lire la fiche d’un client, mais il ne doit pas disposer d’un accès illimité à la base de données ni de la permission d’émettre des remboursements sans un contrôle de politique supplémentaire.

La confiance doit être accordée à chaque opération individuelle plutôt qu’à un serveur entier. Un serveur peut exposer un outil de recherche de produits inoffensif aux côtés d’un outil destructif de suppression de compte. Enregistrez uniquement les capacités nécessaires à l’application, classez-les selon leur effet et maintenez les outils à fort impact indisponibles tant que le workflow actuel ne les exige pas. Une approbation humaine est appropriée lorsqu’une action modifie de l’argent, des accès, un statut juridique, des données de production ou une communication externe.

Validation et permissions selon le principe du moindre privilège

Validez les arguments des outils immédiatement avant leur exécution, même si la définition de l’outil MCP décrit déjà un schéma d’entrée. Les descriptions aident un modèle à formuler une requête, mais elles n’appliquent pas la politique de l’application. La validation doit rejeter les champs manquants, les champs inattendus, les types incorrects, les longueurs excessives, les identifiants invalides et les valeurs situées en dehors des plages autorisées. Les contraintes métier sont aussi importantes que les contraintes structurelles : un montant de remboursement syntaxiquement valide peut tout de même dépasser la limite d’approbation de l’utilisateur actuel.

L’exemple Python exécutable suivant implémente une petite limite de sécurité autour d’un adaptateur d’outil MCP hypothétique. Il utilise uniquement la bibliothèque standard. La fonction de l’adaptateur représente l’appel qu’un client MCP réel effectuerait via son SDK ou son transport. La passerelle valide les arguments, vérifie les permissions de l’appelant et exige une approbation pour une opération qui modifie l’état.

from dataclasses import dataclass
from typing import Any, Callable

@dataclass(frozen=True)
class Principal:
    user_id: str
    permissions: frozenset[str]

TOOL_POLICY = {
    "customer.read": {"permission": "customer:read", "mutating": False},
    "refund.create": {"permission": "refund:create", "mutating": True},
}

def validate_arguments(tool: str, arguments: dict[str, Any]) -> dict[str, Any]:
    if tool == "customer.read":
        if set(arguments) != {"customer_id"}:
            raise ValueError("customer.read requires only customer_id")
        customer_id = arguments["customer_id"]
        if not isinstance(customer_id, str) or not customer_id.startswith("cus_"):
            raise ValueError("customer_id must start with cus_")
        return {"customer_id": customer_id}

    if tool == "refund.create":
        if set(arguments) != {"payment_id", "amount_cents"}:
            raise ValueError("refund.create requires payment_id and amount_cents")
        payment_id = arguments["payment_id"]
        amount = arguments["amount_cents"]
        if not isinstance(payment_id, str) or not payment_id.startswith("pay_"):
            raise ValueError("payment_id must start with pay_")
        if isinstance(amount, bool) or not isinstance(amount, int):
            raise ValueError("amount_cents must be an integer")
        if amount < 1 or amount > 10_000:
            raise ValueError("amount_cents is outside the allowed range")
        return {"payment_id": payment_id, "amount_cents": amount}

    raise ValueError("tool is not registered")

def secure_call(
    principal: Principal,
    tool: str,
    arguments: dict[str, Any],
    approved: bool,
    adapter: Callable[[str, dict[str, Any]], dict[str, Any]],
) -> dict[str, Any]:
    policy = TOOL_POLICY.get(tool)
    if policy is None:
        raise PermissionError("tool is not allowed")
    if policy["permission"] not in principal.permissions:
        raise PermissionError("principal lacks the required permission")
    if policy["mutating"] and not approved:
        raise PermissionError("explicit approval is required")

    validated = validate_arguments(tool, arguments)
    result = adapter(tool, validated)
    if not isinstance(result, dict):
        raise ValueError("tool returned an invalid result")
    return result

def demo_adapter(tool: str, arguments: dict[str, Any]) -> dict[str, Any]:
    return {"tool": tool, "accepted": True, "arguments": arguments}

if __name__ == "__main__":
    principal = Principal(
        user_id="user_42",
        permissions=frozenset({"customer:read", "refund:create"}),
    )
    output = secure_call(
        principal,
        "refund.create",
        {"payment_id": "pay_123", "amount_cents": 2500},
        approved=True,
        adapter=demo_adapter,
    )
    print(output)

L’entrée contient un principal, un nom d’outil, un objet d’arguments et une décision d’approbation. La sortie est un dictionnaire validé renvoyé par l’adaptateur. Remplacer approved par false entraîne l’échec de la demande de remboursement avant son exécution. La transmission d’un argument supplémentaire, d’un identifiant malformé ou d’un montant supérieur à la limite de la politique échoue également. En production, les permissions doivent provenir d’un système d’identité de confiance, et l’approbation doit être représentée par un artefact vérifié par le serveur plutôt que par une valeur booléenne générée par le modèle.

Exécution fiable des outils

Chaque opération MCP doit avoir un délai d’expiration. Sans cela, un serveur indisponible ou une dépendance en aval peut monopoliser les workers indéfiniment et bloquer le workflow de l’agent. Utilisez des délais distincts pour la connexion et l’opération lorsque la pile cliente le permet, et conservez un délai global pour la requête de l’utilisateur. L’annulation doit se propager de l’application au client MCP afin que les tâches abandonnées ne continuent pas à consommer des ressources.

Les tentatives répétées nécessitent de connaître les effets de bord. La lecture d’une ressource ou l’exécution d’une recherche déterministe peut souvent être retentée après une défaillance de transport transitoire. La création d’un paiement, d’un remboursement, d’un ticket ou d’un message ne doit pas être retentée à l’aveugle, car la première tentative peut avoir abouti même si la réponse a été perdue. Pour les outils qui modifient l’état, utilisez une clé d’idempotence stable si le système en aval le permet. Sinon, vérifiez l’état de l’opération avant de réessayer.

Les échecs doivent être classés en erreurs de validation, d’autorisation, de délai d’expiration, de transport, de serveur, de limitation de débit et d’application. Un message sûr peut être présenté à l’utilisateur, par exemple pour lui indiquer que l’action n’a pas pu être effectuée, tandis que la télémétrie interne consigne la catégorie exploitable. N’exposez au modèle ni traces de pile, ni adresses de serveur, ni jetons d’accès, ni réponses brutes des systèmes en aval. Un disjoncteur peut interrompre temporairement les appels vers un serveur en échec persistant, mais il doit compléter les délais d’expiration et la supervision, et non les remplacer.

Journalisation structurée et traçage

L’observabilité doit permettre de savoir quelle opération a été tentée, qui l’a initiée, si la politique l’a autorisée, combien de temps elle a pris et pourquoi elle a échoué. Utilisez des événements structurés plutôt que des messages en texte libre. Les champs utiles comprennent un horodatage, un identifiant de trace, un identifiant de requête, un identifiant du principal, le nom du serveur, le nom de l’outil, le résultat, la catégorie d’erreur, le nombre de tentatives et la durée. Propagez le même contexte de trace dans l’application, le client MCP, le serveur et le service en aval lorsque ces composants prennent en charge le traçage.

Par défaut, ne consignez pas l’intégralité des arguments et des résultats des outils. Ils peuvent contenir des données client, des identifiants, le contenu de documents ou des prompts de modèle. Privilégiez une liste d’autorisation de métadonnées sûres, telles que le type d’enregistrement, le nombre de résultats ou la taille de la charge utile. Masquez les secrets à la frontière de journalisation et protégez les journaux au moyen de contrôles d’accès et de limites de conservation. Le hachage d’un identifiant peut faciliter la corrélation, mais les identifiants prévisibles peuvent tout de même être retrouvés ; le hachage avec clé est donc plus sûr lorsqu’une pseudonymisation est nécessaire.

Les métriques doivent inclure le volume de requêtes, le taux de réussite, le nombre d’appels refusés, les échecs de validation, les percentiles de latence, les délais d’expiration, le nombre de tentatives et la disponibilité des serveurs. Les alertes doivent refléter l’impact sur les utilisateurs plutôt qu’un bruit isolé. Par exemple, déclenchez une alerte lorsque le taux d’échec d’un outil critique reste élevé pendant une période significative. Les enregistrements d’audit des actions sensibles doivent être durables et indiquer la décision de politique ainsi que les éléments attestant l’approbation, sans stocker le raisonnement caché du modèle.

Une passerelle fiable et observable

L’exemple suivant ajoute des délais d’expiration, des tentatives prudentes, une journalisation structurée et un masquage à la frontière présentée précédemment. Il ne retente que les outils marqués comme non modificateurs. Les threads Python ne peuvent pas arrêter de force le travail sous-jacent ; l’adaptateur doit donc également configurer des délais au niveau du transport. Le délai d’expiration de la passerelle empêche malgré tout le workflow appelant d’attendre indéfiniment et montre où placer ce contrôle.

import concurrent.futures
import json
import time
import uuid
from typing import Any, Callable

SAFE_TO_RETRY = {"customer.read"}

def emit_event(event: dict[str, Any]) -> None:
    print(json.dumps(event, separators=(",", ":"), sort_keys=True))

def call_with_controls(
    tool: str,
    arguments: dict[str, Any],
    adapter: Callable[[str, dict[str, Any]], dict[str, Any]],
    timeout_seconds: float = 1.0,
    max_attempts: int = 2,
) -> dict[str, Any]:
    request_id = str(uuid.uuid4())
    attempts = max_attempts if tool in SAFE_TO_RETRY else 1

    for attempt in range(1, attempts + 1):
        started = time.monotonic()
        try:
            with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
                future = pool.submit(adapter, tool, arguments)
                result = future.result(timeout=timeout_seconds)

            duration_ms = round((time.monotonic() - started) * 1000, 2)
            emit_event({
                "event": "mcp_tool_call",
                "request_id": request_id,
                "tool": tool,
                "attempt": attempt,
                "outcome": "success",
                "duration_ms": duration_ms,
                "argument_keys": sorted(arguments.keys()),
            })
            return result
        except concurrent.futures.TimeoutError:
            category = "timeout"
        except ConnectionError:
            category = "transport"
        except Exception:
            category = "internal"

        duration_ms = round((time.monotonic() - started) * 1000, 2)
        emit_event({
            "event": "mcp_tool_call",
            "request_id": request_id,
            "tool": tool,
            "attempt": attempt,
            "outcome": "failure",
            "error_category": category,
            "duration_ms": duration_ms,
            "argument_keys": sorted(arguments.keys()),
        })

        if attempt == attempts:
            raise RuntimeError(f"tool call failed: {category}")
        time.sleep(0.1 * attempt)

    raise RuntimeError("unreachable")

def stable_adapter(tool: str, arguments: dict[str, Any]) -> dict[str, Any]:
    return {"customer_id": arguments["customer_id"], "status": "active"}

if __name__ == "__main__":
    result = call_with_controls(
        "customer.read",
        {"customer_id": "cus_123"},
        stable_adapter,
    )
    print(json.dumps(result))

L’exécution du programme produit un événement de journal structuré, puis le résultat de l’adaptateur. Le journal consigne les noms des arguments, mais pas leurs valeurs, ce qui réduit le risque d’exposition accidentelle de données. Une implémentation de production devrait remplacer l’affichage par l’exporteur de télémétrie de l’organisation, joindre un identifiant de trace existant et conserver en interne l’exception typée d’origine. La couche destinée au modèle devrait recevoir un objet d’erreur stable et assaini, qui l’aide à déterminer s’il faut interroger l’utilisateur, choisir un autre outil ou s’arrêter.

Tests et préparation opérationnelle

Les tests de sécurité doivent tenter d'utiliser des outils inconnus, des permissions manquantes, des arguments malformés, des chaînes trop volumineuses, des champs inattendus, des valeurs d'approbation falsifiées et du contenu malveillant renvoyé par un serveur. Les tests de fiabilité doivent simuler des réponses lentes, des connexions interrompues, des limitations de débit, des défaillances partielles et des résultats ambigus pour les appels qui modifient des données. Vérifiez que la logique de nouvelle tentative ne duplique jamais un effet de bord et que la gestion des délais d'expiration ne signale pas à tort comme définitivement échouée une transaction qui a peut-être abouti.

Les tests d'observabilité sont tout aussi importants. Confirmez que chaque requête possède un identifiant de corrélation, que les appels refusés génèrent des événements d'audit, que les mesures de latence utilisent une horloge monotone et que les valeurs sensibles n'apparaissent jamais dans les journaux. Générez des défaillances contrôlées et vérifiez que les tableaux de bord et les alertes les affichent. Une alerte qui n'a jamais été testée n'est qu'une supposition concernant le pipeline de supervision.

Une intégration MCP prête pour la production place donc une passerelle contrôlée entre l'intention du modèle et les capacités réelles. Cette passerelle valide les deux directions, applique une politique de privilèges minimaux, obtient une approbation pour les actions lourdes de conséquences, fixe des échéances, n'effectue de nouvelles tentatives que pour les opérations sûres, assainit les erreurs et émet une télémétrie corrélée. Ces contrôles rendent le système plus sûr sans compter sur le modèle pour appliquer les règles de sécurité, et fournissent les éléments nécessaires au diagnostic des défaillances lorsque les applications réelles se comportent différemment d'une démonstration de développement.