🇬🇧 English version 💻 Code source (AGPL) : github.com/patlegu/cyber-agent-engine

Le réflexe firewall, appliqué à une décision d’IA

Le premier article de cette série posait la thèse générale : un LLM qui pilote un pare-feu de production ne doit jamais franchir seul la frontière entre intention et exécution. Le deuxième montrait comment cette frontière protège les données — le modèle ne voit que des jetons, jamais une IP réelle. Celui-ci s’attaque à l’autre moitié du problème : une fois que le LLM a proposé une action sur des jetons, qui décide si elle a le droit de se produire ?

La réponse tient dans un réflexe qui n’a rien d’original en sécurité réseau — c’est même la règle la plus ancienne du métier : tout ce qui n’est pas explicitement autorisé est refusé. Un firewall bien configuré ne liste pas les paquets interdits, il liste les paquets autorisés, et referme la porte sur tout le reste. cyber-agent-engine applique exactement ce principe à la couche qui décide si une intention proposée par un LLM peut devenir une action réelle. Pas d’allowlist implicite, pas de « si aucune règle ne s’applique, on suppose que c’est bon ». Si rien ne matche : deny.

evaluate : une fonction pure, sans exception à la règle

Le cœur du mécanisme tient dans un seul fichier, core/policy/engine.py, et sa docstring résume la position en une phrase :

def evaluate(intention: Intention, policy: list[Rule]) -> Verdict:
    """Confronte l'intention à la politique. Défaut deny (fail-closed)."""
    for rule in policy:
        if _match_applies(rule.match, intention):
            return Verdict(effect=rule.effect, matched_rule=rule, intention=intention)
    return Verdict(effect="deny", matched_rule=None, intention=intention)

Trois propriétés valent la peine d’être détaillées, parce que c’est précisément ce qui rend cette fonction auditable et donc digne de confiance.

Première règle qui matche gagne. La liste policy est parcourue dans l’ordre, et le premier Rule dont le match s’applique détermine l’effet. Exactement le modèle mental d’un firewall : l’ordre des règles EST la priorité. Pas de résolution de conflit à la « la règle la plus spécifique gagne », pas de scoring, pas d’agrégation de plusieurs règles qui matchent — un seul verdict, déterminé par la position dans la liste. Ça veut dire que relire une politique revient à la lire de haut en bas, comme une ACL Cisco ou un ruleset OPNsense, sans devoir raisonner sur des interactions cachées entre règles.

Aucune règle → deny. La boucle for se termine sans return dans le cas où rien n’a matché, et la fonction retombe sur la dernière ligne : Verdict(effect="deny", ...). Il n’existe pas de chemin de code qui autorise par défaut. Un opérateur qui oublie d’écrire une règle pour une nouvelle capacité ne l’autorise pas par accident — il la bloque par accident, ce qui est l’échec sûr. C’est la définition même de fail-closed appliquée au code : l’absence de décision produit le résultat le plus restrictif, jamais le plus permissif.

Fonction pure et déterministe. evaluate ne lit ni horloge, ni réseau, ni état mutable global — elle prend une Intention et une list[Rule], elle rend un Verdict, toujours le même pour les mêmes entrées. Le module va jusqu’à préciser dans sa docstring que les conditions « comparent des structures (glob + eq/ne/in/nin/present/absent), sans aucune exécution de code ». On ne peut pas écrire une règle qui exécute un bout de Python arbitraire pour décider — donc on ne peut pas non plus injecter une décision via une règle malicieuse. Et le module précise aussi que le LLM ne peut pas s’auto-autoriser : seuls capability et args de l’Intention sont regardés par _match_applies, jamais rationale — le texte de justification que le modèle produit pour expliquer pourquoi il propose l’action. Un LLM compromis ou halluciné a beau écrire « j’ai l’autorisation explicite de l’administrateur » dans son rationale, ce champ n’entre jamais dans le calcul du verdict.

Le typage de l’effet lui-même est resserré au strict nécessaire, dans core/policy/models.py :

Effect = Literal["allow", "approve", "deny"]

Trois issues possibles, rien d’autre. allow exécute directement, deny arrête tout, et approve — on y revient plus bas — met la décision entre les mains d’un humain plutôt que de la refuser ou de l’accorder immédiatement.

decide : valider, évaluer, auditer, jamais dans le désordre

evaluate seule ne suffit pas : une intention peut être malformée avant même d’être confrontée à la politique (capacité inconnue, arguments invalides), et chaque décision — quelle qu’elle soit — doit laisser une trace. C’est le rôle de core/decision.py, qui orchestre les trois étapes dans un ordre fixe :

def decide(
    intention: Intention,
    *,
    catalog: CapabilityCatalog,
    policy: list[Rule],
    sink: AuditSink,
    event: str = "policy_decision",
) -> Verdict:
    """Valide l'intention (lève si capacité/args invalides), évalue, audite, renvoie."""
    catalog.validate_intention(intention)
    verdict = evaluate(intention, policy)
    sink.write(entry_from_verdict(verdict, event=event))
    return verdict

catalog.validate_intention lève une exception si la capacité n’existe pas ou si les arguments ne correspondent pas au schéma attendu — l’échec est immédiat et bloquant, avant même d’atteindre la politique. Puis evaluate produit le verdict. Puis, systématiquement, sink.write écrit une entrée d’audit — que le verdict soit allow, deny ou approve. Il n’existe pas de chemin où une décision est prise sans être journalisée : l’audit n’est pas une option qu’on active, c’est une étape de la séquence elle-même. Le module précise dans sa docstring que cette fonction est extraite précisément pour que l’orchestrateur mono-action et la boucle multi-pas du coordinateur partagent « exactement la même logique (DRY), sans dupliquer l’ordre validation/évaluation/audit » — un seul endroit où ce contrat peut se briser, pas deux implémentations qui pourraient diverger avec le temps.

GatedLoop : quatre issues, aucune cinquième

decide rend un verdict pour un pas de raisonnement. coordinator/loop.py enchaîne ces pas dans une boucle ReAct — proposer, décider, exécuter, re-tokeniser, recommencer — et le résultat final de toute la boucle est strictement l’une de quatre variantes :

class Completed(BaseModel):
    summary: str
    results: list[dict[str, Any]]

class Suspended(BaseModel):
    approval_id: str

class Denied(BaseModel):
    reason: str

class Failed(BaseModel):
    reason: str

LoopResult = Completed | Suspended | Denied | Failed

Completed : le LLM a terminé sa tâche (Finish), la boucle rend un résumé et les résultats accumulés. Denied : une intention a reçu un verdict deny, la boucle s’arrête là — if verdict.effect == "deny": return Denied(reason=f"policy: {intention.capability}"). Failed : un incident technique (erreur du proposer, exception à l’exécution, ou la limite de max_steps atteinte) — jamais une exception non gérée qui remonterait brute jusqu’à l’appelant HTTP, le code commente explicitement « frontière d’exécution : jamais de 500 non géré ». Et Suspended : la quatrième issue, celle qui nous intéresse ici, déclenchée par le verdict approve.

Quand evaluate rend approve pour une intention — typiquement une action jugée sensible mais pas automatiquement à proscrire, par exemple modifier une règle de pare-feu en production plutôt que simplement lister l’état courant — la boucle ne l’exécute pas et ne la refuse pas non plus. Elle suspend l’intégralité de la session :

if verdict.effect == "approve":
    sid = self._new_id()
    self._approvals.create(intention, approval_id=sid)
    self._sessions.save(SessionState(
        id=sid, request_tokens=request_tokens, vault_snapshot=vault.snapshot(),
        history=history, step=step, expires_at=self._clock() + self._ttl,
        results=results,
        rule_reason=(verdict.matched_rule.reason if verdict.matched_rule else None),
    ))
    return Suspended(approval_id=sid)

Tout l’état nécessaire pour reprendre exactement là où la boucle s’est arrêtée est capturé : les jetons de la requête, un instantané du Vault (donc la table de correspondance jeton→valeur réelle, nécessaire pour retrouver le contexte au retour), l’historique des pas déjà exécutés, le numéro du pas courant, et une date d’expiration calculée comme self._clock() + self._ttl. Le paramètre session_ttl du constructeur de GatedLoop a une valeur par défaut explicite dans le code :

session_ttl: float = 300.0,

300 secondes. Passé ce délai sans décision humaine, SessionStore.get considère la session expirée et resume rend Failed(reason="unknown or expired session") — une approbation qui traîne ne reste pas indéfiniment exploitable, ce qui referme une fenêtre d’attaque évidente : un approval_id volé des semaines plus tard ne sert plus à rien.

ApprovalStore : une approbation jamais résolue n’autorise rien

Le verdict approve crée un objet Approval dans core/approval/store.py. Sa docstring de module pose la même logique fail-closed que l’engine.py de la politique, appliquée cette fois à l’humain plutôt qu’à la règle :

« Une approbation jamais résolue n’autorise rien (défaut pending → jamais exécuté). »

ApprovalStore expose trois opérations, chacune correspondant exactement à ce que son nom promet, sans effet de bord caché :

def create(self, intention: Intention, approval_id: str | None = None) -> Approval:
    ...
    ap = Approval(id=approval_id, intention=intention, intention_hash=intention_hash(intention))
    self._by_id[approval_id] = ap
    return ap

def approve(self, approval_id: str, provided_hash: str) -> Approval:
    ap = self._require(approval_id)
    if provided_hash != ap.intention_hash:
        raise ApprovalMismatch(approval_id)
    updated = ap.model_copy(update={"state": "approved"})
    self._by_id[approval_id] = updated
    return updated

def reject(self, approval_id: str) -> Approval:
    ap = self._require(approval_id)
    updated = ap.model_copy(update={"state": "rejected"})
    self._by_id[approval_id] = updated
    return updated

create fixe l’état initial à pending (valeur par défaut du champ state sur le modèle Approval) — et c’est cet état qui, par construction, n’ouvre jamais aucune autorisation d’exécution. Un Approval qui reste pending parce que personne n’a jamais répondu ne débloque rien ; il n’y a aucun minuteur qui le fait basculer vers approved tout seul, et de toute façon la session associée expirera au bout des 300 secondes. Le seul chemin vers l’exécution passe par approve() — et encore, à une condition précise.

Le hash lie l’approbation à une intention exacte, pas à un identifiant. create calcule intention_hash(intention), un SHA-256 de la sérialisation canonique de l’intention (clés triées, indépendant de l’ordre d’insertion des arguments) :

def intention_hash(intention: Intention) -> str:
    """Hash canonique de l'intention : clés triées, insensible à l'ordre d'insertion des args."""
    canonical = json.dumps(intention.model_dump(), sort_keys=True, ensure_ascii=False)
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

approve(approval_id, provided_hash) exige que l’appelant fournisse ce hash et le compare bit à bit à celui calculé à la création. S’ils diffèrent, la méthode lève ApprovalMismatch plutôt que d’approuver quand même. La docstring du module est explicite sur la menace visée : « approuver X puis présenter une intention différente échoue (contre la substitution de directive) ». Concrètement, ça ferme une classe d’attaque précise — celle où un opérateur voit et valide « lister les règles NAT sur wan1 », pendant qu’un composant compromis en amont a substitué, entre l’affichage et la confirmation, une intention différente portant le même approval_id mais un contenu changé (« supprimer la règle par défaut deny »). Sans le hash lié au contenu exact, un approval_id seul ne prouve rien sur ce qui a été montré à l’humain — seulement qu’un identifiant a été validé. Avec le hash, approuver un identifiant n’autorise que l’intention précise, octet pour octet, dont ce hash a été calculé.

ApprovalNotFound et ApprovalMismatch sont les deux seules exceptions du module — pas de troisième catégorie « approbation douteuse mais on laisse passer ». _require lève ApprovalNotFound pour tout approval_id inconnu, avant même d’atteindre la logique métier de approve ou reject.

L’API : reprendre ou rejeter, jamais deviner

Côté HTTP, coordinator/app.py expose deux routes symétriques sur la GatedLoop suspendue, toutes deux protégées par la même dépendance d’auth globale que le reste de l’API (make_auth_dependency, une comparaison de clé en temps constant, chargée depuis COORDINATOR_API_KEY) :

@app.post("/coordinator/resume/{approval_id}", dependencies=[Depends(require_auth)])
async def resume(approval_id: str, request: Request) -> dict[str, Any]:
    return _serialize(await request.app.state.loop.resume(approval_id))

@app.post("/coordinator/reject/{approval_id}", dependencies=[Depends(require_auth)])
async def reject(approval_id: str, request: Request) -> dict[str, Any]:
    return _serialize(request.app.state.loop.reject(approval_id))

Les deux sont des POST — reprendre ou rejeter une approbation change un état, jamais un GET. resume délègue à GatedLoop.resume, qui relit la session, vérifie via grant_approved que l’Approval est bien dans l’état approved (sinon Denied(reason=f"approval in state {approval.state}") — un pending ou un rejected ne passe pas), puis fait un geste qu’il vaut la peine de citer intégralement tant il est significatif :

# Consommer l'approbation AVANT d'exécuter : anti-rejeu fail-closed. Une panne
# transitoire de l'agent pendant `execute` ne doit jamais laisser une session
# approuvée rejouable (un ban n'est pas idempotent, il ne doit jamais s'exécuter deux fois).
self._approvals.mark_executed(approval_id)
self._sessions.delete(approval_id)

L’approbation est marquée executed et la session supprimée avant l’appel à execute(), pas après. Si l’exécution échoue en cours de route — timeout réseau, agent injoignable — un deuxième appel à /coordinator/resume/{id} ne rejouera pas l’action : l’approval_id ne pointe déjà plus vers une approbation utilisable. C’est la même discipline fail-closed appliquée à un troisième niveau, après la politique et l’approbation : le rejeu accidentel d’une action non idempotente (bannir une IP deux fois n’est pas grave, mais toutes les actions ne partagent pas cette propriété) est structurellement empêché, pas laissé à la discipline de l’opérateur qui appuierait deux fois sur le même bouton.

/coordinator/reject/{approval_id} est plus direct : il appelle ApprovalStore.reject, purge la session, écrit une entrée d’audit event="rejected", et rend Denied(reason="rejected by the operator"). Aucune exécution possible sur ce chemin — le seul verdict de sortie est un refus.

Ce que ce mécanisme garantit, et ce qu’il ne garantit pas

En remontant les quatre couches — evaluate sans règle par défaut permissive, decide qui audite systématiquement, GatedLoop qui ne connaît que quatre issues terminales, ApprovalStore qui lie chaque autorisation au contenu exact de l’intention montrée — la propriété qui en résulte est simple à énoncer et difficile à obtenir autrement : rien ne s’exécute sans un verdict explicite, et pour tout ce qui est classé sensible, sans qu’un humain ait vu et validé le contenu précis de ce qui va se produire.

Ce que ce mécanisme ne garantit pas, en revanche, mérite d’être dit aussi clairement : il ne garantit pas que la politique elle-même est bien écrite. Une règle allow trop large sur un glob de capacité (opnsense.* au lieu de opnsense.list_*) reste une porte ouverte — evaluate appliquera fidèlement une mauvaise règle, parce que la fonction est pure et n’a aucun jugement sur la qualité de la politique qu’on lui donne. Ce n’est pas un défaut de conception : c’est la séparation attendue des responsabilités. Le moteur garantit que la politique, quelle qu’elle soit, ne sera jamais contournée par le LLM. Écrire une bonne politique reste, comme pour tout firewall, la responsabilité de l’opérateur.

flowchart TD
  P["Proposer : intention (sur jetons)"] --> D{"core.decide : evaluate + audit"}
  D -->|allow| X["Exécuter via la frontière"]
  D -->|"deny (défaut, fail-closed)"| S["Stop + audit"]
  D -->|approve| H["SUSPEND — Approval créé"]
  H --> Op{"Opérateur"}
  Op -->|"resume + hash"| X
  Op -->|reject| S

Cet article fait partie de la série sur cyber-agent-engine. Le premier article pose les cinq principes de la frontière de confiance ; un article à venir montrera comment ces briques — politique, boucle gatée, approbation, audit — s’assemblent en un système opérable en production.