From bd73d498285c727bc2dd5238b06c48625104e896 Mon Sep 17 00:00:00 2001 From: ciregenz Date: Mon, 15 Jun 2026 14:22:48 -0700 Subject: [PATCH] [eric] mcp-gate: property-test the activation gate (forwarded => activated) + Z3 formal proof --- backend/tests/formal/README.md | 20 +++++++ backend/tests/formal/mcp_gate_proof.py | 83 ++++++++++++++++++++++++++ backend/tests/test_v2_invariants.py | 44 ++++++++++++++ 3 files changed, 147 insertions(+) create mode 100644 backend/tests/formal/README.md create mode 100644 backend/tests/formal/mcp_gate_proof.py diff --git a/backend/tests/formal/README.md b/backend/tests/formal/README.md new file mode 100644 index 00000000..5bb7cf13 --- /dev/null +++ b/backend/tests/formal/README.md @@ -0,0 +1,20 @@ +# Formal proofs + +Machine-checked proofs of safety/security invariants that the unit/property +tests can only *sample*. A property test tries thousands of cases; an SMT proof +is exhaustive over the modeled domain (assert the negation, `unsat` => theorem). + +Not wired into prod or CI, and excluded from the packaged build (under `tests/`). +Run manually: + +``` +pip install z3-solver +python backend/tests/formal/mcp_gate_proof.py +``` + +- **`mcp_gate_proof.py`** , the MCP dispatch-gate invariant (`agent_manager._build_mcp_servers`): + a gated session forwards a server *only if* it was activated, an empty + activation list forwards zero, and a denied server is never forwarded. Sampled + by `tests/test_v2_invariants.py::test_mcp_gate_only_forwards_activated_servers`; + proven for all inputs here. The script also refutes a deliberately-buggy gate + (activation check dropped) so the proof can't be vacuous. diff --git a/backend/tests/formal/mcp_gate_proof.py b/backend/tests/formal/mcp_gate_proof.py new file mode 100644 index 00000000..6bf03dc6 --- /dev/null +++ b/backend/tests/formal/mcp_gate_proof.py @@ -0,0 +1,83 @@ +"""Formal proof (Z3 / SMT) of the MCP dispatch-gate security invariant. + +The product rule "MCP tools are reachable only after MCPActivate" is enforced at +dispatch in agent_manager._build_mcp_servers: for a gated session a server is +forwarded to the model only if its sanitized name is in session.active_mcps. + +tests/test_v2_invariants.py::test_mcp_gate_only_forwards_activated_servers +SAMPLES that contract (400 random cases). This SMT proof is exhaustive over the +modeled domain: we assert the negation of each property and ask Z3 for a +counterexample. `unsat` means none can exist, so the property is a theorem, +true for every possible input, not just the ones a test happened to try. + +Not wired into prod or CI. Run manually: + pip install z3-solver && python backend/tests/formal/mcp_gate_proof.py +""" + +from z3 import And, Bool, Implies, Not, Or, Solver, sat, unsat + + +def forwarded(installed, allowed, denied, active_is_none, active_t): + """Faithful model of the gate decision for one arbitrary server `t` + (agent_manager.py:165-203). A server ships to the model iff it is an + installed+configured MCP tool, passes the permission gate, isn't fully + denied, and EITHER the session is legacy (active_mcps is None) OR the + server is in active_mcps. Proving it for an arbitrary symbolic `t` proves + it for all servers.""" + return And(installed, allowed, Not(denied), Or(active_is_none, active_t)) + + +def buggy_forwarded(installed, allowed, denied, active_is_none, active_t): + """The same gate with the activation check dropped, used to show the proof + has teeth: Z3 must be able to refute the no-leak property for this variant.""" + return And(installed, allowed, Not(denied)) + + +def prove(name: str, claim) -> bool: + """`claim` should be valid (true for every input). Proven by showing its + negation is unsatisfiable.""" + s = Solver() + s.add(Not(claim)) + if s.check() == unsat: + print(f" PROVED: {name}") + return True + print(f" FAILED: {name} counterexample: {s.model()}") + return False + + +def main() -> None: + installed = Bool("installed") + allowed = Bool("allowed") + denied = Bool("denied") + active_is_none = Bool("active_is_none") # legacy session (no activation gate) + active_t = Bool("active_t") # server t is in active_mcps + fwd = forwarded(installed, allowed, denied, active_is_none, active_t) + gated = Not(active_is_none) + + print("Proving MCP dispatch-gate invariants (exhaustive over all inputs):") + ok = True + # A. No leak: a gated session never forwards a non-activated server. + ok &= prove("gated => (forwarded(t) -> activated(t))", + Implies(And(gated, fwd), active_t)) + # B. Empty activation => zero servers (no t is active, so none ship). + ok &= prove("gated & !activated(t) => !forwarded(t)", + Implies(And(gated, Not(active_t)), Not(fwd))) + # C. The permission gate still binds: a denied server is never forwarded. + ok &= prove("denied(t) => !forwarded(t)", Implies(denied, Not(fwd))) + + # Teeth: the buggy gate (activation check dropped) MUST be refutable, else + # the proof above would be vacuous. + print("Sanity-checking the proof has teeth (a buggy gate must be refuted):") + bug = buggy_forwarded(installed, allowed, denied, active_is_none, active_t) + s = Solver() + s.add(Not(Implies(And(gated, bug), active_t))) + assert s.check() == sat, "buggy gate should leak but Z3 couldn't refute it" + print(f" REFUTED (as expected): a gate without the activation check leaks; " + f"counterexample = {s.model()}") + + print("\nALL GATE PROPERTIES PROVED" if ok else "\nPROOF FAILED") + raise SystemExit(0 if ok else 1) + + +if __name__ == "__main__": + main() diff --git a/backend/tests/test_v2_invariants.py b/backend/tests/test_v2_invariants.py index 51d146ed..6ea61908 100644 --- a/backend/tests/test_v2_invariants.py +++ b/backend/tests/test_v2_invariants.py @@ -550,6 +550,50 @@ def test_error_classify_gemini_resource_exhausted_is_transient(): assert not _is_transient_capacity_error(Exception("403 permission denied")) +@pytest.mark.asyncio +async def test_mcp_gate_only_forwards_activated_servers(): + """Dispatch-layer security invariant (the non-bypassable enforcement of + 'MCP tools only via MCPActivate'): for a GATED session (active_mcps is a + list), _build_mcp_servers forwards ONLY servers whose sanitized name is in + active_mcps; an empty list forwards ZERO; None is the legacy all-allowed + path. The model cannot reach an unactivated server no matter what it asks + for. Property-checked over random installed sets and random activation + subsets, plus the two boundary cases.""" + import random + from types import SimpleNamespace + from backend.apps.agents.agent_manager import AgentManager + mgr = AgentManager() + names = ["gmail", "drive", "slack", "reddit", "notion", "airtable"] + + def installed(): + return [SimpleNamespace(name=n, mcp_config={"x": 1}, enabled=True, + auth_status="configured", auth_type="apikey") for n in names] + + # allowed_tools == get_all_tool_names() bypasses the (separate) permission + # gate so we isolate the ACTIVATION gate. _sanitize_server_name -> identity. + with patch("backend.apps.agents.agent_manager.load_all_tools", side_effect=installed), \ + patch("backend.apps.agents.agent_manager.get_all_tool_names", return_value=["__ALL__"]), \ + patch("backend.apps.agents.agent_manager._sanitize_server_name", side_effect=lambda n: n), \ + patch("backend.apps.agents.agent_manager._is_fully_denied", return_value=False), \ + patch("backend.apps.agents.agent_manager.derive_mcp_config", side_effect=lambda t: {"command": "x"}): + allowed = ["__ALL__"] + # Boundary 1: empty activation list -> zero servers, always. + assert await mgr._build_mcp_servers(allowed, active_mcps=[]) == {} + # Boundary 2: None (legacy) -> permission gate only, all forwarded. + assert set((await mgr._build_mcp_servers(allowed, active_mcps=None)).keys()) == set(names) + # Property: forwarded set is ALWAYS a subset of the activated set, and + # equals exactly the activated-and-installed intersection. + rng = random.Random(1234) + for _ in range(400): + active = rng.sample(names, rng.randint(0, len(names))) + # throw in a bogus name the gate must never invent a server for + if rng.random() < 0.3: + active = active + ["ghost-not-installed"] + forwarded = set((await mgr._build_mcp_servers(allowed, active_mcps=active)).keys()) + assert forwarded <= set(active), f"leaked {forwarded - set(active)} for active={active}" + assert forwarded == (set(active) & set(names)), f"mismatch for active={active}" + + def test_banned_models_not_offered(): """Claude Fable (banned) and Gemini 3.1 Pro (no working lane: AG can't serve it, AI Studio key 429s pro-preview) were pulled from the picker. Guard so a