🇫🇷 Version française 💻 Source code (AGPL): github.com/patlegu/cyber-agent-engine

The firewall reflex, applied to an AI decision

The first article in this series laid out the general thesis: an LLM that drives a production firewall must never cross, on its own, the boundary between intent and execution. The second showed how that boundary protects data — the model only ever sees tokens, never a real IP. This one tackles the other half of the problem: once the LLM has proposed an action on tokens, who decides whether it’s allowed to actually happen?

The answer lies in a reflex that’s nothing original in network security — it’s arguably the oldest rule in the trade: everything that isn’t explicitly authorized is denied. A well-configured firewall doesn’t list the forbidden packets, it lists the allowed ones, and closes the door on everything else. cyber-agent-engine applies exactly that principle to the layer that decides whether an intent proposed by an LLM can become a real action. No implicit allowlist, no “if no rule applies, assume it’s fine.” If nothing matches: deny.

evaluate: a pure function, no exception to the rule

The core of the mechanism sits in a single file, core/policy/engine.py, and its docstring sums up the stance in one sentence:

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)

Three properties deserve a closer look, because they’re precisely what makes this function auditable, and therefore trustworthy.

First matching rule wins. The policy list is walked in order, and the first Rule whose match applies determines the effect. That’s exactly the mental model of a firewall: the order of the rules IS the priority. No “most-specific-rule-wins” conflict resolution, no scoring, no aggregation of several matching rules — a single verdict, determined by position in the list. Which means reviewing a policy amounts to reading it top to bottom, like a Cisco ACL or an OPNsense ruleset, with no need to reason about hidden interactions between rules.

No matching rule → deny. The for loop ends without a return when nothing matched, and the function falls through to the last line: Verdict(effect="deny", ...). There is no code path that allows by default. An operator who forgets to write a rule for a new capability doesn’t accidentally allow it — they accidentally block it, which is the safe failure. That’s the very definition of fail-closed applied to code: the absence of a decision produces the most restrictive outcome, never the most permissive one.

A pure, deterministic function. evaluate reads no clock, no network, no mutable global state — it takes an Intention and a list[Rule], it returns a Verdict, always the same one for the same inputs. The module goes as far as spelling out in its docstring that conditions “compare structures (glob + eq/ne/in/nin/present/absent), with no code execution whatsoever.” You cannot write a rule that runs an arbitrary bit of Python to decide — so you also cannot inject a decision through a malicious rule. And the module also makes clear that the LLM cannot self-authorize: only the capability and args fields of the Intention are looked at by _match_applies, never rationale — the justification text the model produces to explain why it’s proposing the action. A compromised or hallucinating LLM can write “I have explicit administrator authorization” in its rationale all it wants; that field never enters the verdict computation.

The effect’s own typing is tightened to the strict necessary, in core/policy/models.py:

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

Three possible outcomes, nothing else. allow executes directly, deny stops everything, and approve — we’ll come back to it below — puts the decision in a human’s hands rather than immediately denying or granting it.

decide: validate, evaluate, audit, never out of order

evaluate alone isn’t enough: an intent can be malformed even before it’s checked against the policy (unknown capability, invalid arguments), and every decision — whatever it is — must leave a trace. That’s the role of core/decision.py, which orchestrates the three steps in a fixed order:

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 raises an exception if the capability doesn’t exist or if the arguments don’t match the expected schema — the failure is immediate and blocking, before the policy is even reached. Then evaluate produces the verdict. Then, systematically, sink.write writes an audit entry — whether the verdict is allow, deny, or approve. There is no code path where a decision is made without being logged: audit isn’t an option you switch on, it’s a step of the sequence itself. The module’s docstring notes that this function is extracted precisely so that the single-action orchestrator and the coordinator’s multi-step loop share “exactly the same logic (DRY), without duplicating the validation/evaluation/audit order” — a single place where this contract can break, not two implementations that could drift apart over time.

GatedLoop: four outcomes, no fifth

decide returns a verdict for one reasoning step. coordinator/loop.py chains these steps into a ReAct loop — propose, decide, execute, re-tokenize, repeat — and the final result of the entire loop is strictly one of four variants:

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: the LLM finished its task (Finish), the loop returns a summary and the accumulated results. Denied: an intent received a deny verdict, the loop stops right there — if verdict.effect == "deny": return Denied(reason=f"policy: {intention.capability}"). Failed: a technical incident (a proposer error, an exception during execution, or the max_steps limit reached) — never an unhandled exception bubbling up raw to the HTTP caller; the code explicitly comments “execution boundary: never an unhandled 500.” And Suspended: the fourth outcome, the one we’re interested in here, triggered by an approve verdict.

When evaluate returns approve for an intent — typically an action judged sensitive but not automatically to be forbidden, for instance modifying a production firewall rule rather than merely listing the current state — the loop neither executes it nor denies it. It suspends the entire 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)

All the state needed to resume exactly where the loop stopped is captured: the request’s tokens, a snapshot of the Vault (so the token→real-value mapping, needed to restore context on return), the history of steps already executed, the current step number, and an expiration date computed as self._clock() + self._ttl. The session_ttl parameter of GatedLoop’s constructor has an explicit default value in the code:

session_ttl: float = 300.0,

300 seconds. Past that delay without a human decision, SessionStore.get considers the session expired and resume returns Failed(reason="unknown or expired session") — an approval left hanging doesn’t stay exploitable forever, which closes an obvious attack window: an approval_id stolen weeks later is worthless.

ApprovalStore: an approval never resolved authorizes nothing

An approve verdict creates an Approval object in core/approval/store.py. Its module docstring states the same fail-closed logic as the policy engine.py, applied this time to the human rather than the rule:

“An approval that’s never resolved authorizes nothing (default pending → never executed).”

ApprovalStore exposes three operations, each doing exactly what its name promises, with no hidden side effect:

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 sets the initial state to pending (the default value of the state field on the Approval model) — and it’s that state which, by construction, never opens any execution authorization. An Approval that stays pending because nobody ever answered unlocks nothing; there’s no timer that flips it to approved on its own, and the associated session will expire after 300 seconds anyway. The only path to execution goes through approve() — and even then, under one precise condition.

The hash binds the approval to an exact intent, not to an identifier. create computes intention_hash(intention), a SHA-256 of the intent’s canonical serialization (sorted keys, independent of argument insertion order):

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) requires the caller to supply this hash and compares it byte for byte against the one computed at creation time. If they differ, the method raises ApprovalMismatch rather than approving anyway. The module’s docstring is explicit about the threat this targets: “approving X then presenting a different intent fails (against directive substitution).” Concretely, this closes a precise class of attack — the one where an operator sees and validates “list the NAT rules on wan1,” while a compromised component upstream has swapped, between display and confirmation, a different intent carrying the same approval_id but changed content (“delete the default deny rule”). Without the hash bound to the exact content, an approval_id alone proves nothing about what was shown to the human — only that an identifier was validated. With the hash, approving an identifier only authorizes the precise intent, byte for byte, from which that hash was computed.

ApprovalNotFound and ApprovalMismatch are the module’s only two exceptions — no third category of “questionable approval but let it through anyway.” _require raises ApprovalNotFound for any unknown approval_id, before even reaching the business logic of approve or reject.

The API: resume or reject, never guess

On the HTTP side, coordinator/app.py exposes two symmetric routes on the suspended GatedLoop, both protected by the same global auth dependency as the rest of the API (make_auth_dependency, a constant-time key comparison, loaded from 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))

Both are POST — resuming or rejecting an approval changes state, never a GET. resume delegates to GatedLoop.resume, which re-reads the session, checks via grant_approved that the Approval is indeed in the approved state (otherwise Denied(reason=f"approval in state {approval.state}") — a pending or a rejected one doesn’t pass), then does something worth quoting in full given how significant it is:

# 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)

The approval is marked executed and the session deleted before the call to execute(), not after. If execution fails partway through — a network timeout, an unreachable agent — a second call to /coordinator/resume/{id} won’t replay the action: the approval_id already no longer points to a usable approval. That’s the same fail-closed discipline applied at a third level, after the policy and the approval: accidental replay of a non-idempotent action (banning an IP twice isn’t a big deal, but not every action shares that property) is structurally prevented, not left to the discipline of an operator who might press the same button twice.

/coordinator/reject/{approval_id} is more direct: it calls ApprovalStore.reject, purges the session, writes an audit entry with event="rejected", and returns Denied(reason="rejected by the operator"). No execution possible on this path — the only exit verdict is a denial.

What this mechanism guarantees, and what it doesn’t

Stepping back through the four layers — evaluate with no permissive default rule, decide which systematically audits, GatedLoop which knows only four terminal outcomes, ApprovalStore which binds every authorization to the exact content of the intent that was shown — the resulting property is simple to state and hard to obtain any other way: nothing executes without an explicit verdict, and for everything classified as sensitive, without a human having seen and validated the precise content of what’s about to happen.

What this mechanism does not guarantee, however, deserves to be said just as clearly: it does not guarantee that the policy itself is well written. A too-broad allow rule on a capability glob (opnsense.* instead of opnsense.list_*) remains an open door — evaluate will faithfully apply a bad rule, because the function is pure and has no judgment on the quality of the policy it’s handed. That’s not a design flaw: it’s the expected separation of responsibilities. The engine guarantees that the policy, whatever it is, will never be bypassed by the LLM. Writing a good policy remains, as with any firewall, the operator’s responsibility.

flowchart TD
  P["Propose: intent (on tokens)"] --> D{"core.decide: evaluate + audit"}
  D -->|allow| X["Execute via the boundary"]
  D -->|"deny (default, fail-closed)"| S["Stop + audit"]
  D -->|approve| H["SUSPEND — Approval created"]
  H --> Op{"Operator"}
  Op -->|"resume + hash"| X
  Op -->|reject| S

This article is part of the cyber-agent-engine series. The first article lays out the five principles of the trust boundary; an upcoming article will show how these building blocks — policy, gated loop, approval, audit — assemble into a system operable in production.