From 069d1513ee6f2c17767e31d70f5264d93f985c3e Mon Sep 17 00:00:00 2001 From: henleda Date: Wed, 29 Jul 2026 20:50:06 -0500 Subject: [PATCH] feat(mcp): K1 MCP server mode MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `vpcopilot mcp [--write]` serves the pipeline as Model Context Protocol tools over stdio, so an agent session gets a band-aid proposal inline. Hand-rolled newline-delimited JSON-RPC 2.0 — no new dependency, so it works with no extra install, unlike `console`. Confirmed connected to a real MCP client. Ten read/scan tools always: scan_result, patches_list, ledger, impact, deps, simulation_result, drift, verify_bundle, and scan_start/scan_status (a scan takes minutes, so it is start-then-poll, as the console already does it). Five mutating tools — apply, pr, retire, reconcile, simulate — are ABSENT from tools/list unless writes are explicitly enabled via --write or VPCOPILOT_MCP_WRITE=1. Deliberate, and documented: - apply/pr/retire default to dry_run=true, inverting every module default. The CLI and console each pass a choice a human made at a keyboard; an MCP call is issued by a model, so the default must change nothing. - `simulate` is classified as MUTATING though the roadmap listed it read-only: a simulation attaches a throwaway policy to a load balancer. Cleaning up after itself makes it safe, not read-only. - `apply` takes a policy NAME, not a path, validated as a slug and checked to resolve inside /policies (the J2 precedent). - stdout carries the protocol and nothing else: serve() points sys.stdout at stderr for its lifetime and writes frames to a private handle, so a stray print anywhere beneath it cannot corrupt the stream. run_pipeline, survey_report and drift.check all default log=print. Behaviour change to a shared write path, deliberate and pinned both ways: the G2 blast-radius gate had exactly one production caller — the console — so `vpcopilot apply --from-scan` would attach an over-broad policy the console refuses, and the --allow-overbroad flag ROADMAP.md described did not exist. It moved into simulate.promotion_gate, called by both apply paths, and the CLI gained the flag. K1's acceptance ("the same gate as the CLI and console") could not otherwise be honest. Still warn-with-audited-override, never a veto. Also fixed: apply_from_scan(create_only=True) returned before the only guard_lb call, so it wrote a policy object to the tenant unguarded; write_result erased a policy's blast-radius flag when a narrower replay did not measure it, silently disabling the gate for it. 681 tests pass offline, ruff clean, coverage 79%. Co-Authored-By: Claude Opus 5 (1M context) --- ROADMAP.md | 132 ++++- docs/AUDIT.md | 7 +- docs/USAGE.md | 89 +++ src/vpcopilot/apply.py | 23 +- src/vpcopilot/cli.py | 34 +- src/vpcopilot/console/app.py | 27 +- src/vpcopilot/mcp.py | 997 +++++++++++++++++++++++++++++++++ src/vpcopilot/refiner.py | 9 +- src/vpcopilot/schemas.py | 4 + src/vpcopilot/simulate.py | 66 ++- tests/test_console_simulate.py | 20 +- tests/test_mcp.py | 809 ++++++++++++++++++++++++++ tests/test_simulate.py | 166 ++++++ 13 files changed, 2343 insertions(+), 40 deletions(-) create mode 100644 src/vpcopilot/mcp.py create mode 100644 tests/test_mcp.py diff --git a/ROADMAP.md b/ROADMAP.md index 1d50270..a617d7b 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -760,25 +760,121 @@ sees the trail. J1–J4 are the open `BACKLOG.md` evidence entries, scheduled; * ## Phase K — Reach developers without the console -- [ ] **K1** MCP server mode. (M, P1) Depends on G2 and I1. - Expose the read-only surface as MCP tools so an agent session gets a band-aid proposal - inline: `scan`, `triage`, `generate`, `simulate`, `patches list`. - - Acceptance: apply, pr, retire, and reconcile are absent from the tool list unless - explicitly enabled in config; **enabling them does not bypass the human gate** — a write - tool still routes through the same gate and guardrails as the CLI and console; tool schemas - document every argument; the server calls the same module functions as the CLI and console. - - **Reconciled:** `simulate` needs G2 and `patches list` needs I1; the item declared no - dependencies. - - Surfaces: `src/vpcopilot/mcp.py`, `vpcopilot mcp`. +- [x] **K1** MCP server mode. (M, P1) — **DONE:** `vpcopilot mcp [--write]`. `mcp.py` is a + hand-rolled Model Context Protocol server over stdio — **no new dependency** — exposing ten + read/scan tools always and five mutating ones only when writes are explicitly enabled. + **Verified against a real MCP client**: registered with Claude Code, `✔ Connected`, tools + enumerated, and driven end to end over a real subprocess pipe against the live H2 run data and + live OSV. 53 tests. + - **Acceptance, as met:** apply/pr/retire/reconcile (and `simulate`) are **absent** from + `tools/list` unless enabled ✅ — absent rather than present-and-refusing, because a tool an agent + can see is a tool it will try; every write tool calls the same module function the CLI and + console call ✅, so it inherits `guard_lb`, `PROTECTED_POLICIES`, `drift.preflight`, the G2 + blast-radius gate, rollback-unless-`keep` and a centrally-stamped audit record rather than + reimplementing any of them; every argument of every tool carries a description ✅, pinned by a + test that walks the schemas rather than by review. + - **The acceptance criterion could not be met as written, and fixing that is the largest change + here.** "The same gate and guardrails as the CLI and console" presumes the two agree. They did + not: `simulate.promotion_block` — the G2 blast-radius gate — had exactly **one** production + caller, `console/app.py`, so `vpcopilot apply --from-scan` would attach an over-broad policy the + console refuses with a 409, and the `--allow-overbroad` flag **this file described at line 143 + did not exist**. Same shape as I1's `--force-probe`, whose guard lived only in the CLI and left + the console able to mass-replay every destructive exploit. So the check moved into + `simulate.promotion_gate`, called by **both** `apply.apply_from_scan` and + `refiner.refine_apply_service_policy` (the latter is the default for `--from-scan` and the + console's Mitigate button, so gating only the former would have gated nothing anyone uses), and + the CLI gained the flag. **This is a deliberate behaviour change to a shared write path**: a CLI + apply now refuses an over-broad policy unless `--allow-overbroad`, and writes the + `simulate_override` audit record it previously never wrote. Still warn-with-audited-override, not + a machine veto (the G2/I2 precedent). Pinned in both directions, including that a policy with no + simulation applies exactly as before. + - **`simulate` is not read-only, and the item listed it as such.** G2's simulation creates a + throwaway `-vpcsim` policy object, **attaches it to the load balancer**, replays through it + and deletes it. Cleaning up after itself makes it safe, not read-only, so it sits behind the same + opt-in as `apply`; `simulation_result` is the ungated way to read a previous run's numbers. The + three-tier taxonomy this forced — `READ` / `WRITES_OUT` (a scan, additive, tenant untouched) / + `MUTATES` — drives the MCP annotations from one `Access` value per tool, so `readOnlyHint` and + `destructiveHint` cannot drift from the truth. + - **stdout belongs to the protocol, and this codebase is the worst possible tenant for that.** The + spec forbids writing anything to stdout that is not an MCP message, and `run_pipeline`, + `survey_report` and `drift.check` all default `log=print`, with `rprint` used throughout the CLI. + Passing `log=` at every call site is a convention, and a convention is what the next call site + forgets — so `serve()` reassigns `sys.stdout` to stderr for its lifetime and writes frames to a + private handle, which makes a stray `print` **anywhere beneath it** structurally incapable of + corrupting the stream. Verified empirically that `rich` follows the swap (it resolves + `sys.stdout` at write time rather than binding it at import), and pinned by a test that prints + from inside a tool. + - **A scan is start-then-poll, not one blocking call.** A scan takes minutes and an MCP call is + request/response, so `scan_start` returns a job id and `scan_status` tails the log with a `since` + cursor — the same shape the console already uses, with none of its FastAPI coupling. The log sink + is a list, so progress is returned to the caller rather than written anywhere. + - **Decisions.** *Stdlib, not the official SDK* (which is available, at 2.0.0): the surface K1 needs + is `initialize` / `tools/list` / `tools/call` / `ping`, hand-rolling it keeps `vpcopilot mcp` + working with no extra install unlike `console`, it is testable offline by feeding frames to + `handle()`, and it avoids pinning a major-version API that churns under a committed demo. The + interop risk is real but bounded, and it was retired by testing against a real client rather than + by argument. *Opt-in is `--write` or `VPCOPILOT_MCP_WRITE=1`*, not a key in `agents.yaml` — + `config.py` is an agent-model registry and a feature flag does not belong in it; authoring the + client config is the human action, exercised once, which is the argument I1 made about the + crontab. *`apply`/`pr`/`retire` default to `dry_run=True`*, inverting every module default, + because the CLI and console each pass a choice a human made at a keyboard and an MCP call is + issued by a model. *`force_probe` is not exposed at all* — its guard needs a single `--finding` + because replaying every destructive exploit at once is not something to do by accident, and a + model deciding to pass it is exactly that accident. *`apply` takes a policy **name**, not a + path*, derived against the run directory (the J2 precedent), validated as a slug and checked to + resolve inside `/policies`; traversal already failed because the mandatory `service_policy.` + prefix makes the first segment a directory that must exist, which is luck rather than design. + - **What the opt-in cannot do is supply the human.** MCP clients are expected to confirm tool calls + with a user, but that is client behaviour this server can neither enforce nor verify — stated in + `docs/USAGE.md` rather than implied, and the reason the write tools are off by default. + - **Fixed en route (pre-existing):** `apply_from_scan(create_only=True)` returned before + `apply_service_policy`, the only caller of `guard_lb`, so it wrote a policy object into the tenant + with neither the protected-LB check nor the drift preflight. It attaches nothing, so no traffic + changed — but a persistent write against a protected target should not be the one path that skips + the guard. `guard_lb` is now unconditional at the top of `apply_from_scan`; it is a pure check, so + the other paths are unchanged. + - **Found by adversarial review, before shipping** (18 raised across five failure dimensions; 12 + verified, of which 3 were confirmed outright and 9 were refuted **because they had already been + fixed mid-review** — the skeptics were reading the patched tree, as happened in H2 — plus 6 lower + -severity ones triaged afterwards). The two that mattered: + - **A malformed `tools/call` killed the server outright, with zero frames written.** JSON-RPC + permits positional (array) `params`, and a list is truthy, so `params.get(...)` raised straight + out of `serve()`'s loop; a non-string tool name did the same through an unhashable dict lookup. + The client waits forever on a request that will never be answered and every later request is + lost with it. Dying silently is the worst available failure for a transport. Both inputs are now + invalid-params, and — the structural half — the loop wraps `handle()` and answers `-32603` + rather than ending, so a bug not yet written cannot kill the connection either. + - **A narrower replay erased an earlier policy's blast-radius flag.** `simulate --policy B` + filters the candidates and `write_result` overwrote `simulation.json` wholesale, so a policy A + an earlier run had flagged lost `blocked_promotion` — and the gate above went quiet for it. An + operator who simulated everything, saw A flagged, then re-simulated only B would find A + applying with no warning: a guard erased as a side effect of measuring something else, which is + I1's "a band-aid could vouch for its own removal" in a new place. Entries this run did not + measure are now carried forward stamped `carried_from`, so the gate keeps firing and nothing + passes an old number off as fresh. Pre-existing in G2; the gate move is what made it load-bearing. + - **A regression this change introduced, caught here:** the legacy `POST /api/apply` — still + served though the UI no longer calls it — began enforcing the moved gate while `ApplyReq` + carried no `allow_overbroad`, turning warn-with-audited-override into an unoverridable machine + veto on that one surface. All four call sites of the two gated functions now expose the flag. + - Plus: a `tools/call` `TypeError` was reported as *invalid arguments*, which would send an agent + round a loop retrying arguments against a fault inside a tool (there is now deliberately no + `except TypeError`, because `validate_args` makes an argument-binding error unreachable and a + test pins that every documented property is a real parameter); `scan_status` read the log twice + and could advance its cursor past what it returned, losing lines a client could never re-request + (proven at 565 of 20000 polls); a corrupt `findings.json` rendered as `[]` — the H2 confusion, + reproduced in new code — so an unreadable member is now `null` and named in `unreadable`; + `impact`/`ledger`/`patches_list` answered a **nonexistent** run directory with confident zeros; + `scan_start` accepted a path that does not exist, which `run_pipeline`'s own docstring calls + "the failure mode not to extend"; two scans into one run directory interleaved their artifacts; + a notification whose method was a request method was answered with an unsolicited `id: null`; + stdin was decoded with the process locale rather than the mandated UTF-8; and the `drift` tool's + description promised a shadowing check that does not run without `policy`. + - **Reconciled, and confirmed accurate:** `triage` and `generate` have **no module function to + share** — both take a live `Harness` plus pydantic models with no CLI or console twin — so v1 + exposes them only as stages inside `scan_start` rather than inventing twins for them. `vpcopilot + mcp` matches the flat command set; there is no `serve` verb. - Note: pairs with the vendor's own Distributed Cloud MCP server effort. Keep them independent. - This one exposes the pipeline, not the tenant. - - **Reconciled:** there is no `serve` command — `console` (`cli.py:499`) is the only launcher, so - a second verb would be a new convention; `vpcopilot mcp` matches the flat command set. Two of - the five listed tools have **no module function to share**: `triage` and `generate` exist only - as agent entry points taking a live `Harness` plus pydantic models (`agents/triage.py:57`, - `agents/generate.py:94`), with no CLI or console twin. Either scope v1 to surfaces that exist, - or add a sub-item creating those twins first. The dependency is also transitive — G2 is itself - gated on the undecided G1. + This one exposes the pipeline, not the tenant — `mcp.py` never imports `xc`, pinned by a test. - [ ] **K2** GitHub Action. (M, P2) Depends on G2. Scan the diff on a pull request and comment each new finding above a severity threshold diff --git a/docs/AUDIT.md b/docs/AUDIT.md index 886c82b..f8cc53b 100644 --- a/docs/AUDIT.md +++ b/docs/AUDIT.md @@ -117,7 +117,7 @@ Everything else is per-action detail. | `apply_skipped_no_change` | gate | The policy asked for was already the attached one. Nothing was pushed — no LB PUT, no snapshot, no run artifact | `lb` `policy` | | `drift_block` | gate | The apply was **refused**: an ALLOW inside the policy matched the exploit before its DENY, so under FIRST_MATCH the band-aid would have attached cleanly and blocked nothing | `lb` `policy` `conflicts` | | `drift_override` | gate | The same conflict, applied anyway via `--force` / the console's **apply anyway**. The override is the point of the entry | `lb` `policy` `conflicts` | -| `simulate_override` | gate | A policy that shadow simulation flagged as over-broad was promoted anyway | `finding_id` `policy` `lb` `block_rate` `threshold` `reason` | +| `simulate_override` | gate | A policy that shadow simulation flagged as over-broad was promoted anyway. Written by `simulate.promotion_gate`, which **both** apply paths call — it used to be written only by the console, so a CLI apply neither refused nor recorded the override (K1) | `finding_id` `policy` `lb` `block_rate` `threshold` `reason` | | `escalation` | reconcile | A band-aid outlived its TTL with no merged cure. The control is **left in place** (`kept: true`) — an escalation is a notification, never a removal | `finding_id` `lb` `control` `policy` `cure_url` `cure_state` `applied_at` `expires_at` `ttl_hours` `age_hours` `escalation_count` `kept` `notified` `trigger` `pass_id` + denormalized `title` `vuln_class` `severity` | | `fix_ineffective` | reconcile | The cure PR merged, but the exploit still reproduces **at the origin** — the code fix did not work. The band-aid is held | `finding_id` `lb` `control` `policy` `cure_url` `reason` `kept` `origin_probe` `trigger` `pass_id` + denormalized finding fields | | `reconcile_retire` | reconcile | An unattended pass detached a band-aid after proving at origin that the exploit no longer reproduces. Written **alongside** the normal `retire` entry, not instead of it, so every existing consumer of `retire` keeps working | `finding_id` `lb` `control` `cure_url` `origin_probe` `trigger` `pass_id` | @@ -147,7 +147,10 @@ Notes read from the source: given `--apply`. A report-only pass that finds an overdue patch still writes `escalation` — the notification is the point — but it never writes `reconcile_retire`. A pass where nothing changed writes nothing at all, so a nightly cron does not grow the log by N lines a night forever. -- Every reconcile record carries `pass_id` and `trigger` (`cli` / `console` / `cron`). Cron invokes +- Every reconcile record carries `pass_id` and `trigger` (`cli` / `console` / `cron` / `mcp`). An + agent session driving the MCP server (K1) records `mcp`, so the trail can say a reconcile pass came + from an agent rather than a person — the fact a reviewer most wants and the one the log could not + previously express. Cron invokes the CLI, so a scheduled pass is marked by exporting `VPCOPILOT_RECONCILE_TRIGGER=cron` in the crontab; without it a nightly pass is indistinguishable from someone typing the command. `run_id` cannot identify a pass — it is the identity of the out dir, and `audit.record` strips a diff --git a/docs/USAGE.md b/docs/USAGE.md index 791e35f..2a76d68 100644 --- a/docs/USAGE.md +++ b/docs/USAGE.md @@ -238,6 +238,27 @@ In the console the same check runs inside the Mitigate job, so its warnings appe A refusal renders an **apply anyway** button, and `no_change` renders as its own outcome rather than a pass or a fail. +### Pre-apply blast-radius gate (G2) + +If a simulation (§ *Blast radius*) found the policy would block too much of the recorded traffic, +applying it needs an explicit override: + +```sh +vpcopilot apply --from-scan out/policies/.json --lb --url --allow-overbroad +``` + +Without it the apply refuses and names the rate and threshold; with it the apply proceeds and writes +a `simulate_override` audit record carrying the finding, the policy, the LB, the rate, the threshold +and the actor. Silent when nothing was simulated — G2 adds a check, never a prerequisite, so an +operator who never runs `simulate` sees exactly the behaviour they saw before it existed. + +**This gate used to exist on only one surface**, and that is worth stating because it changes CLI +behaviour. `simulate.promotion_block` had a single production caller — the console — so +`vpcopilot apply --from-scan` would happily attach an over-broad policy the console refused, and the +`--allow-overbroad` flag this documentation described did not exist. The check now lives in +`simulate.promotion_gate`, called by both apply paths, so the CLI, the console and the MCP server +share one copy and cannot drift. A guard in one surface is not a guard. + ## 5. Open the code-fix PR (the cure) ```sh vpcopilot pr --repo owner/name --base --path-prefix [--finding ] [--dry-run] @@ -416,6 +437,74 @@ header carries a live model switcher, and each step is deep-linkable (`#mitigate | **⑦ Benchmark** | build a model-tagged report from this run, then compare models side by side per target app | | **⚙ Setup** | credentials (writes `.env`), XC status, the per-agent model wiring, and the report buttons | +## 8. MCP server mode (K1) + +The same pipeline as MCP tools over stdio, so an agent session gets a band-aid proposal inline +instead of shelling out. No extra install — the transport is stdlib. + +```sh +vpcopilot mcp # read-only (default) +vpcopilot mcp --write # also expose apply, pr, retire, reconcile, simulate +``` + +Register it with any MCP client. For Claude Code: + +```sh +claude mcp add vpcopilot -- /path/to/.venv/bin/python -m vpcopilot.cli mcp +``` + +**Read-only by default, and the write tools are *absent* rather than present-and-refusing.** A tool +an agent can see is a tool it will try, so enabling them is an explicit act: `--write`, or +`VPCOPILOT_MCP_WRITE=1`. Authoring the client config that does it is the human action, exercised +once — the same argument `reconcile --apply` makes about the crontab. + +| Tool | What it does | Costs | +|---|---|---| +| `scan_result` | the band-aid proposal from a finished run: findings, triage, generated policies, cures, dependency funnel | nothing | +| `patches_list` | live band-aids with age, TTL remaining, cure state, escalations | nothing | +| `ledger` · `impact` | the four-state lifecycle; the headline numbers | nothing | +| `deps` | what a `--manifest` scan would find, without a model call | reaches OSV.dev | +| `simulation_result` | a previous blast-radius replay's numbers | nothing | +| `drift` | live LB vs last snapshot vs proposed, read-only | XC credentials | +| `verify_bundle` | re-check an evidence bundle against its own manifest | nothing | +| `scan_start` · `scan_status` | start a scan, then poll it | **model calls**, minutes | +| `apply` · `pr` · `retire` · `reconcile` · `simulate` | *only with writes enabled* | mutates | + +**Three things are deliberate.** + +*`apply`, `pr` and `retire` default to `dry_run=true`* — the opposite of every module function, whose +default is a real run. The CLI and console each pass a choice a human made at a keyboard; an MCP call +is issued by a model, so the default has to be the one that changes nothing, and applying for real +has to be a second explicit call. `reconcile` is report-only unless `apply=true`, as on the CLI. + +*`simulate` is a write tool, though the roadmap listed it as read-only.* A simulation creates a +throwaway policy object, attaches it to the load balancer, replays through it and deletes it again. +Cleaning up after itself makes it safe, not read-only. `simulation_result` is the ungated way to read +the numbers. + +*`apply` takes a policy **name**, not a path*, and derives the artifact from the run directory — an +interface that accepts a caller-supplied filesystem path is an arbitrary-file reader, and a tool a +model invokes is a worse place for one than an endpoint a human drives (the J2 precedent). The name +must be a generated slug; anything carrying a path separator is refused. + +**What the opt-in does not do is supply the human.** MCP clients are expected to confirm tool calls +with a user, but that is the client's behaviour, not something this server can enforce or verify — +which is exactly why the write tools are off by default. What the server *can* guarantee is that a +write tool calls the same module function the CLI and console call, so it inherits `guard_lb` for a +protected load balancer, `PROTECTED_POLICIES` for a protected name, `drift.preflight` for drift and a +self-shadowing DENY, the G2 blast-radius gate, rollback-unless-`keep`, and an audit record whose +identity is stamped centrally. Reconcile passes `trigger="mcp"`, so the trail says an agent session +did it. + +`force_probe` is deliberately not exposed at all: its guard requires a single `--finding` because +replaying every destructive exploit at once is not something to do by accident, and a model deciding +to pass it is exactly that accident. + +**stdout carries the protocol and nothing else.** The server points `sys.stdout` at stderr for its +lifetime and writes frames to a private handle, so a stray `print` anywhere beneath it — the pipeline +defaults `log=print`, and `rprint` is used throughout the CLI — lands on stderr, which the MCP spec +reserves for logging, instead of corrupting the message stream. + **Run settings** — the collapsible bar shown on the action steps (**Mitigate / Cure / Retire**): LB · validate URL · PR repo · base · path prefix, plus **dry-run** (on by default), **refine** + attempts, **keep live**, and **allow protected LB**. Its summary line spells out the mode you're diff --git a/src/vpcopilot/apply.py b/src/vpcopilot/apply.py index 109cbc1..39ba51a 100644 --- a/src/vpcopilot/apply.py +++ b/src/vpcopilot/apply.py @@ -366,7 +366,8 @@ def apply_from_scan(artifact_path: str, lb: str, target_url: str, *, name: str | create_only: bool = False, dry_run: bool = False, keep: bool = False, allow_protected: bool = False, probe: bool = False, retries: int = 8, wait_seconds: int = 8, finding_id: str | None = None, force: bool = False, - out_dir: str = "out", log: Callable = print) -> dict: + allow_overbroad: bool = False, out_dir: str = "out", + log: Callable = print) -> dict: """End-to-end from a generated artifact: create the policy in XC (if missing), then attach -> validate -> rollback via apply_service_policy. Guarded against clobbering a protected policy. @@ -376,6 +377,14 @@ def apply_from_scan(artifact_path: str, lb: str, target_url: str, *, name: str | PUTs, so "reports no_change and writes nothing" has to short-circuit earlier than the mutation. `force=True` re-applies anyway (to re-validate a policy that is already attached).""" xc = XC() + # The protected-LB guardrail, unconditionally. It used to be reached only via + # `apply_service_policy`, which `create_only` returns before ever calling — so + # `apply_from_scan(lb="nimbus-www", create_only=True)` wrote a real policy object into the + # tenant with neither this check nor the drift preflight. It attaches nothing, so no traffic + # changed, but a persistent write against a protected target should not be the one path that + # skips the guard. `guard_lb` is a pure check, so `apply_service_policy` calling it again below + # is harmless and the non-create_only path is unchanged. + guard_lb(lb, allow_protected=allow_protected, dry_run=dry_run) art = json.loads(Path(artifact_path).read_text()) # Normalize to an XC create body: {metadata:{name,namespace,...}, spec:{...}}. # Generated artifacts vary — some are full {metadata, spec} objects, some a bare spec. @@ -401,6 +410,18 @@ def apply_from_scan(artifact_path: str, lb: str, target_url: str, *, name: str | # to precede `ApplyContext.load()`, which writes a snapshot on every call, so "no_change writes # nothing" is true of the run directory as well as the LB. `create_only` makes no attachment, # so there is nothing for it to gate. + # G2 — the blast-radius gate, now in the module so the CLI, the console and the MCP server + # cannot disagree about it. Runs before the CREATE for the same reason the drift check does. + from .simulate import promotion_gate + # The audit record's join key: unlike the refiner, this path never resolves `finding_id` from the + # ledger, so an override recorded here would lose the one field that ties it to a finding. + # Resolved for the RECORD only — binding it to `finding_id` itself would change which probe + # `apply_service_policy` validates against, which is not this change's business. + from . import ledger as _ledger + _fid = finding_id or _ledger.find_finding_for_policy(out_dir, policy_name) + promotion_gate(out_dir, policy_name, allow_overbroad=allow_overbroad, dry_run=dry_run, + finding_id=_fid, lb=lb, log=log) + if not dry_run and not create_only: from .drift import preflight d = preflight(lb, policy_name, out_dir=out_dir, force=force, xc=xc, log=log, spec=spec, diff --git a/src/vpcopilot/cli.py b/src/vpcopilot/cli.py index d757e6e..1192723 100644 --- a/src/vpcopilot/cli.py +++ b/src/vpcopilot/cli.py @@ -233,6 +233,7 @@ def apply( probe_token: str = typer.Option(None, "--probe-token", help="bearer token for validation instead of user/pass (or VPCOPILOT_PROBE_TOKEN)"), finding: str = typer.Option(None, "--finding", help="finding id whose probe (out/probes.json) validates this policy; overrides the ledger lookup"), force: bool = typer.Option(False, "--force", help="apply despite the pre-apply drift check: re-attach a policy that is already attached, or push past a conflicting ALLOW on another attached policy"), + allow_overbroad: bool = typer.Option(False, "--allow-overbroad", help="apply despite the G2 blast-radius gate: a policy a simulation found too broad. Writes a simulate_override audit record"), out: str = typer.Option("out", help="output directory"), ): """Gated apply: (create from scan) -> snapshot -> self-test -> attach -> validate -> refine/rollback.""" @@ -250,10 +251,12 @@ def apply( from .refiner import refine_apply_service_policy res = refine_apply_service_policy(from_scan, lb, url, name=name, keep=keep, allow_protected=allow_protected_lb, max_refine=refine_attempts, - finding_id=finding, force=force, out_dir=out, log=logf) + finding_id=finding, force=force, + allow_overbroad=allow_overbroad, out_dir=out, log=logf) elif from_scan: from .apply import apply_from_scan - res = apply_from_scan(from_scan, lb, url, name=name, create_only=create_only, force=force, **kw) + res = apply_from_scan(from_scan, lb, url, name=name, create_only=create_only, force=force, + allow_overbroad=allow_overbroad, **kw) else: from .apply import apply_service_policy res = apply_service_policy(lb, policy, url, **kw) @@ -846,6 +849,33 @@ def console(host: str = typer.Option("127.0.0.1", help="bind host"), uvicorn.run("vpcopilot.console.app:app", host=host, port=port, log_level="warning") +@app.command() +def mcp(write: bool = typer.Option(None, "--write/--no-write", + help="expose the mutating tools (apply, pr, retire, reconcile, " + "simulate). Off unless this flag or VPCOPILOT_MCP_WRITE=1. " + "They still route through the same gates as the CLI, and " + "apply/pr/retire default to a dry run")): + """Serve the pipeline as MCP tools over stdio, for an agent session (K1). + + Read-only by default: read a finished run (scan_result, impact, patches_list, ledger), survey + dependencies against OSV (deps), inspect a load balancer (drift), verify an evidence bundle, and + start or poll a scan. The mutating tools are ABSENT from the tool list unless writes are + enabled — authoring the client config that enables them is the human action, exercised once, the + same argument `reconcile --apply` makes about the crontab. + + Nothing is printed here: stdout carries the protocol and only the protocol. Progress and errors + go to stderr, which the MCP spec reserves for exactly that.""" + import sys as _sys + + from .mcp import serve + enabled = write if write is not None else \ + os.environ.get("VPCOPILOT_MCP_WRITE", "").lower() in ("1", "true", "yes") + # stderr, never stdout — one stray character on stdout desynchronises the client. + print(f"vpcopilot mcp: serving on stdio, writes {'ENABLED' if enabled else 'off'}", + file=_sys.stderr, flush=True) + serve(enable_writes=enabled) + + def main(): load_dotenv() # pull provider keys (ANTHROPIC_API_KEY, etc.) from .env app() diff --git a/src/vpcopilot/console/app.py b/src/vpcopilot/console/app.py index 771c992..4ad8fc4 100644 --- a/src/vpcopilot/console/app.py +++ b/src/vpcopilot/console/app.py @@ -773,27 +773,20 @@ def _dispatch_action(body: ActionReq, log): if c == "service_policy": # G2 gate: a simulated policy found too broad WARNS and requires an explicit override. # Silent when nothing was simulated — G2 adds a check, never a prerequisite. - from ..simulate import promotion_block - over = promotion_block(str(OUT), body.policy_name) if body.policy_name else None - if over and not body.dry_run: - if not body.allow_overbroad: - raise HTTPException(409, f"simulation says this policy {over.get('reason')}. " - "Re-run with 'allow overbroad' to apply it anyway.") - log(f"⚠ overbroad override: {over.get('reason')}") - from ..audit import record as _audit - _audit(str(OUT), "simulate_override", finding_id=body.finding_id, - policy=body.policy_name, lb=body.lb, block_rate=over.get("block_rate"), - threshold=over.get("threshold"), reason=over.get("reason")) + # K1 moved the check itself into `simulate.promotion_gate`, called by BOTH apply paths, so + # the CLI and the MCP server get it too — it used to live only here. The message and the + # resulting job state are unchanged: `_run_action` catches the raise and reports + # `state="error"` carrying the "allow overbroad" text, exactly as before. art = str(OUT / "policies" / f"service_policy.{body.policy_name}.json") if body.refine and not body.dry_run: from ..refiner import refine_apply_service_policy return refine_apply_service_policy(art, body.lb, body.url, finding_id=body.finding_id, name=body.policy_name, keep=body.keep, allow_protected=body.allow_protected_lb, max_refine=body.refine_attempts, config_path=_active_config, force=body.force, - out_dir=str(OUT), log=log) + allow_overbroad=body.allow_overbroad, out_dir=str(OUT), log=log) return A.apply_from_scan(art, body.lb, body.url, name=body.policy_name, dry_run=body.dry_run, keep=body.keep, allow_protected=body.allow_protected_lb, force=body.force, - out_dir=str(OUT), log=log) + allow_overbroad=body.allow_overbroad, out_dir=str(OUT), log=log) if c == "malicious_user": return A.apply_malicious_user(body.lb, **kw) if c == "rate_limit": @@ -878,6 +871,10 @@ class ApplyReq(BaseModel): refine_attempts: int | None = None allow_protected_lb: bool = False force: bool = False + # K1: this older endpoint is still served even though the UI now posts to /api/action. Moving the + # G2 gate into the module meant it started enforcing here too — and without this field there was + # no way to override it, turning a warn-with-audited-override into a machine veto on one surface. + allow_overbroad: bool = False @app.post("/api/apply") @@ -890,11 +887,13 @@ def do_apply(body: ApplyReq): return refine_apply_service_policy(art, body.lb, body.url, name=body.name, keep=body.keep, allow_protected=body.allow_protected_lb, max_refine=body.refine_attempts, force=body.force, + allow_overbroad=body.allow_overbroad, out_dir=str(OUT), log=lambda m: None) from ..apply import apply_from_scan return apply_from_scan(art, body.lb, body.url, name=body.name, create_only=body.create_only, dry_run=body.dry_run, keep=body.keep, force=body.force, - allow_protected=body.allow_protected_lb, out_dir=str(OUT), + allow_protected=body.allow_protected_lb, + allow_overbroad=body.allow_overbroad, out_dir=str(OUT), log=lambda m: None) except Exception as e: # noqa: BLE001 raise HTTPException(400, str(e)) diff --git a/src/vpcopilot/mcp.py b/src/vpcopilot/mcp.py new file mode 100644 index 0000000..0654df1 --- /dev/null +++ b/src/vpcopilot/mcp.py @@ -0,0 +1,997 @@ +"""K1 — MCP server mode. The pipeline's read-only surface as Model Context Protocol tools, so an +agent session gets a band-aid proposal inline instead of shelling out to the CLI. + +Three things about this module are deliberate and load-bearing. + +**No new dependency.** The stdio transport is newline-delimited JSON-RPC 2.0, and the whole surface +K1 needs is `initialize`, `tools/list`, `tools/call` and `ping`. Hand-rolling that is ~150 lines and +means `vpcopilot mcp` works with no extra install — unlike `console`, which needs the `console` +extra — and it is testable offline by feeding frames to `handle()`, with no network and no fakes +beyond the ones `tests/` already has. The official SDK is at 2.0.0; pinning a major-version API +that churns under a committed demo buys interop we can get by testing against a real client. + +**stdout belongs to the protocol, and nothing else.** The spec is explicit: the server MUST NOT +write anything to stdout that is not a valid MCP message. This codebase logs to stdout everywhere — +`rprint` throughout `cli.py`, and `run_pipeline`/`survey_report`/`drift.check` all default +`log=print`. One stray line corrupts the stream and the client drops the connection. Passing +`log=` at every call site would be a convention, and a convention is exactly what gets forgotten by +the next call site. So `serve()` reassigns `sys.stdout` to stderr for the life of the process and +keeps a private handle for frames: a stray `print` ANYWHERE beneath us is structurally incapable of +corrupting the protocol. Pinned by a test that prints from inside a tool. + +**Read-only is asserted, not assumed.** `readOnlyHint` defaults to *false* and `destructiveHint` +defaults to *true* in the MCP schema, so every hint here is set explicitly from a single +`Access` classification per tool rather than left to a default a reader would have to look up. +""" +from __future__ import annotations + +import json +import re +import sys +import threading +import traceback +import uuid +from collections.abc import Callable +from dataclasses import dataclass, field +from enum import Enum +from typing import Any + +from . import __version__ + +# The protocol revision this server implements. On a mismatch the spec says respond with a version +# we DO support rather than erroring, and let the client decide whether to disconnect. +PROTOCOL_VERSION = "2025-06-18" +SERVER_NAME = "vpcopilot" + +# JSON-RPC error codes we actually emit. +PARSE_ERROR = -32700 +INVALID_REQUEST = -32600 +METHOD_NOT_FOUND = -32601 +INVALID_PARAMS = -32602 +INTERNAL_ERROR = -32603 + +# A generated policy name, as `generate` emits and `apply` writes to disk: a slug that starts +# alphanumeric. Used to keep a caller-supplied name from becoming a path. `..` is excluded +# separately — it satisfies the charset (dots are legal inside a slug) but is never a real name. +_SAFE_NAME = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}") + + +def _is_safe_name(name: str) -> bool: + return bool(_SAFE_NAME.fullmatch(name)) and ".." not in name + + +class Access(Enum): + """What a tool does to the world. Drives the MCP annotations deterministically, so the hints + cannot drift from the truth by someone editing one and not the other. + + READ — reads disk and/or the network; changes nothing anywhere. One documented caveat: + `deps` will populate the OSV advisory cache if — and only if — the operator set + `VPCOPILOT_ADVISORY_CACHE`. That is an opt-in, idempotent write outside the run + directory that cannot affect the tenant, the ledger or any artifact, so the tool + still reports `readOnlyHint: true`; marking it mutating would make clients prompt + for a dependency lookup, which is worse and less true. + WRITES_OUT — writes only into the run directory (a scan). Additive, never destructive, and + nothing on the tenant is touched. + MUTATES — changes something outside this machine: a load balancer, a GitHub PR, the ledger's + live state. Absent from the tool list unless writes are explicitly enabled. + """ + + READ = "read" + WRITES_OUT = "writes_out" + MUTATES = "mutates" + + +@dataclass +class Tool: + name: str + title: str + description: str + schema: dict + fn: Callable[..., Any] + access: Access = Access.READ + # openWorldHint: does it reach an unbounded external world (OSV, the tenant, GitHub)? + open_world: bool = False + + def definition(self) -> dict: + """The `tools/list` entry. Every argument carries a description — an acceptance criterion, + and the only documentation an agent session ever sees.""" + return { + "name": self.name, + "title": self.title, + "description": self.description, + "inputSchema": self.schema, + "annotations": { + "title": self.title, + "readOnlyHint": self.access is Access.READ, + # Meaningful only when readOnlyHint is false. A scan only adds artifacts to a run + # dir; a MUTATES tool can take protection off a live load balancer. + "destructiveHint": self.access is Access.MUTATES, + "idempotentHint": False, + "openWorldHint": self.open_world, + }, + } + + +def _out_schema(extra: dict | None = None, *, required: list[str] | None = None) -> dict: + """Most tools are "read this run directory". Shared so the wording of `out` is identical in + every schema rather than retyped eight times.""" + props = {"out": {"type": "string", "default": "out", + "description": "run directory written by a previous scan (default: out)"}} + props.update(extra or {}) + return {"type": "object", "properties": props, "required": required or []} + + +# --------------------------------------------------------------------------- tool implementations +# Each returns a plain JSON-serialisable dict or list. Each is handed a `log` sink that appends to a +# list, never stdout — belt as well as the braces of the sys.stdout swap in `serve()`. + +def _require_run_dir(out: str) -> None: + """A run directory that does not exist is not an empty one. + + `list_patches`, `ledger.load` and `impact` all answer a missing directory with zeros, which reads + as "no live band-aids" / "nothing mitigated" — a confident all-clear for a run that was never + made. A typo'd `out` would have produced the most reassuring possible answer.""" + from pathlib import Path + if not Path(out).is_dir(): + raise Declined(f"no run directory at {out!r} — that is not the same as a run with nothing " + "in it. Check the path, or start one with scan_start.") + + +def _tool_patches_list(out: str = "out", expired_only: bool = False, **_) -> dict: + from .reconcile import list_patches + _require_run_dir(out) + rows = list_patches(out) + if expired_only: + rows = [r for r in rows if r["expired"]] + return {"out": out, "count": len(rows), "patches": rows} + + +def _tool_ledger(out: str = "out", **_) -> dict: + from .ledger import load + _require_run_dir(out) + entries = load(out) + return {"out": out, "count": len(entries), "entries": list(entries.values())} + + +def _tool_impact(out: str = "out", **_) -> dict: + from .impact import impact + _require_run_dir(out) + return impact(out) + + +def _tool_scan_result(out: str = "out", **_) -> dict: + """The band-aid proposal itself, read back from a finished run: what was found, how each finding + was triaged, and which XC config was generated for it. This is the payload K1 exists to put in + front of an agent session.""" + import json as _json + from pathlib import Path + + # ABSENT and UNREADABLE are different answers and must not render the same way. A member that is + # not there is normal — not every run writes every artifact — and gets the default. A member that + # exists and will not parse is a fact we failed to establish: swallowing the error returned `[]`, + # which is indistinguishable from a run that genuinely found nothing. That is the same confusion + # H2 exists to prevent, so an unreadable member becomes `null` (not `[]`) and is named in + # `unreadable`, where nothing can mistake it for an empty result. + unreadable: list[dict] = [] + + def rj(name, default): + p = Path(out) / name + if not p.is_file(): + return default + try: + return _json.loads(p.read_text()) + except (OSError, _json.JSONDecodeError) as e: + unreadable.append({"member": name, "error": str(e)}) + return None + + summary = rj("summary.json", {}) + if not summary: + # Refusing to guess: an empty dict here would read as "a clean scan", which is the one + # answer this project never lets a missing input produce. + raise Declined( + f"summary.json in {out!r} could not be read: {unreadable[0]['error']}" + if unreadable else + f"no scan results in {out!r} — nothing has been scanned into it yet " + "(run scan_start, or point `out` at a directory that holds summary.json)") + res = {"out": out, "summary": summary, "findings": rj("findings.json", []), + "triage": rj("triage.json", []), "policies": rj("policies.json", []), + "remediations": rj("remediations.json", []), + "correlations": rj("correlations.json", []), + "dependencies": rj("dependencies.json", None)} + if unreadable: + res["unreadable"] = unreadable + res["caveat"] = (f"{len(unreadable)} artifact(s) exist but could not be parsed and are null " + "below, NOT empty: " + ", ".join(u["member"] for u in unreadable)) + return res + + +def _tool_deps(manifest: list[str] | None = None, min_severity: str = "high", + max_advisories: int = 25, include_dev: bool = False, log=None, **_) -> dict: + from .inputs.deps import survey_report + paths = [m for m in (manifest or []) if str(m).strip()] + if not paths: + raise Declined("give at least one manifest path (requirements.txt, package-lock.json " + "or pom.xml)") + if min_severity not in ("critical", "high", "medium", "low"): + raise Declined("min_severity must be one of critical, high, medium, low") + return survey_report(paths, min_severity=min_severity, max_advisories=max_advisories, + include_dev=include_dev, log=log or (lambda m: None)) + + +def _tool_simulation_result(out: str = "out", **_) -> dict: + from .simulate import load_result + res = load_result(out) + if not res: + raise Declined(f"no simulation.json in {out!r} — no blast-radius replay has run for this " + "run directory. A simulation attaches a throwaway policy to a spare load " + "balancer, so it is not something this server does on its own.") + return res + + +def _tool_drift(lb: str = "", out: str = "out", control: str | None = None, + policy: str | None = None, finding: str | None = None, log=None, **_) -> dict: + """Read-only by construction: `drift.check` issues no PUT and writes no snapshot (I2).""" + import json as _json + from pathlib import Path + + from .drift import check + if not str(lb).strip(): + raise Declined("lb is required — name the load balancer to inspect") + exploit = None + if finding: + from .apply import _load_probe + exploit = (_load_probe(out, finding) or {}).get("exploit") + art = Path(out, "policies", f"service_policy.{policy}.json") if policy else None + spec = _json.loads(art.read_text()) if art and art.is_file() else None + return check(lb, out_dir=out, control=control, policy_name=policy, exploit=exploit, + spec=spec, log=log or (lambda m: None)) + + +def _tool_verify_bundle(path: str = "", pubkey: str | None = None, log=None, **_) -> dict: + from .export import verify_bundle + if not str(path).strip(): + raise Declined("path is required — point at an exported evidence bundle (.zip)") + return verify_bundle(path, pubkey=pubkey, log=log or (lambda m: None)) + + +class Declined(Exception): + """A tool that cannot establish a fact says so. Surfaced to the client as a tool-execution + error (`isError: true`) with the reason, never as a traceback and never as an empty result that + would read like a clean answer.""" + + +# --------------------------------------------------------------------------- the scan job registry +# A scan takes minutes and an MCP tool call is request/response, so `scan_start` returns a job id +# immediately and `scan_status` polls — the same start-then-poll shape the console already uses for +# scan/apply/reconcile, but with none of its FastAPI coupling. + +_JOBS: dict[str, dict] = {} +_JOBS_LOCK = threading.Lock() + + +def _tool_scan_start(repo: str | None = None, cve: str | None = None, spec: str | None = None, + manifest: list[str] | None = None, out: str = "out", + min_confidence: float = 0.5, max_files: int = 200, max_bytes: int = 60_000, + draft_code_fixes: bool = True, min_severity: str = "high", + max_advisories: int = 25, include_dev: bool = False, + config_path: str | None = None, **_) -> dict: + manifests = [m for m in (manifest or []) if str(m).strip()] + repo, cve, spec = (str(x).strip() if x else "" for x in (repo, cve, spec)) + # Same input rules the CLI and the console enforce, checked here so the failure is a clean + # decline rather than a traceback out of run_pipeline's own guard. + if cve and (repo or spec or manifests): + raise Declined("a CVE scan cannot be combined with a repo, a spec or a manifest") + if not (repo or cve or spec or manifests): + raise Declined("give a repo path, a CVE/GHSA id, an OpenAPI spec, or a dependency manifest") + # A path that does not exist produced a completed scan with zero findings and a durable + # summary.json saying so — a clean bill of health for something never read. `run_pipeline`'s own + # docstring calls this out as "the failure mode not to extend", so it is checked before the job + # is even started rather than discovered minutes later in a log. + from pathlib import Path + for label, p in (("repo", repo), ("spec", spec), *[("manifest", m) for m in manifests]): + if p and not Path(p).exists(): + raise Declined(f"{label} path {p!r} does not exist — a scan of nothing would write a " + "summary saying nothing was found, which is not the same answer") + + job_id = f"scan-{uuid.uuid4().hex[:12]}" + job = {"id": job_id, "state": "running", "log": [], "summary": None, "error": None, + "out": out} + with _JOBS_LOCK: + # Two scans into ONE run directory interleave their artifacts: both write findings.json, + # triage.json, policies/ and the ledger, and the loser's half-written state is + # indistinguishable from a complete run. An agent session polling one job would read the + # other's numbers. Refusing is the only honest answer, and it names the job to wait for. + clash = next((j for j in _JOBS.values() + if j["state"] == "running" and j["out"] == out), None) + if clash is not None: + raise Declined( + f"a scan is already running into {out!r} (job {clash['id']}) — two scans would " + "interleave their artifacts in one run directory. Wait for it with scan_status, or " + "pass a different `out`.") + _JOBS[job_id] = job + + def run(): + from .pipeline import run_pipeline + try: + summary = run_pipeline( + repo or None, out_dir=out, config_path=config_path, + min_confidence=min_confidence, max_files=max_files, max_bytes=max_bytes, + draft_code_fixes=draft_code_fixes, + # The log sink is a list, never stdout. `serve()` also swaps sys.stdout, so this is + # the second of two independent guarantees. + log=lambda m: job["log"].append(str(m)), + advisory=cve or None, spec_path=spec or None, + manifest_paths=manifests or None, min_severity=min_severity, + max_advisories=max_advisories, include_dev=include_dev) + job["summary"] = summary + job["state"] = "done" + except Exception as e: # noqa: BLE001 — a failed scan is a reported state, not a crash + job["error"] = f"{type(e).__name__}: {e}" + job["log"].append(f"scan failed: {job['error']}") + job["state"] = "failed" + + t = threading.Thread(target=run, name=job_id, daemon=True) + t.start() + return {"job_id": job_id, "state": "running", "out": out, + "note": "poll scan_status with this job_id; a scan spends model calls and usually " + "takes minutes"} + + +def _tool_scan_status(job_id: str = "", since: int = 0, **_) -> dict: + with _JOBS_LOCK: + job = _JOBS.get(str(job_id)) + if job is None: + known = sorted(_JOBS) + raise Declined(f"no such scan job {job_id!r}" + + (f" — known jobs: {', '.join(known)}" if known else + " — no scan has been started by this server")) + # ONE snapshot of the log, and both the slice and the cursor come from it. Reading + # `job["log"]` twice — slicing it, then taking its length for `log_next` — is a lost-update + # race: the worker thread appends between the two reads, `log_next` counts lines the slice did + # not return, and a client tailing with `since=log_next` never sees them. Silently dropping + # progress lines is the kind of wrong answer nobody notices. + snap = list(job["log"]) + frm = max(0, int(since)) + state, summary, error = job["state"], job["summary"], job["error"] + return {"job_id": job["id"], "state": state, "out": job["out"], + "log_from": frm, "log_next": len(snap), "log": snap[frm:], + "summary": summary, "error": error} + + +# --------------------------------------------------------------------------- registry + +def build_tools(*, enable_writes: bool = False) -> list[Tool]: + """The tool list. Write tools are ABSENT unless explicitly enabled — not present-and-refusing, + because a tool an agent can see is a tool it will try, and the acceptance criterion is absence. + """ + tools = [ + Tool("scan_result", "Scan result", + "The band-aid proposal from a finished scan: every verified finding, how triage routed " + "it to an F5 XC control (or to code only), the generated policy artifacts, the code " + "cures, and the dependency funnel when the scan used --manifest. Reads a run directory; " + "spends nothing. Declines if the directory holds no scan.", + _out_schema(), _tool_scan_result), + Tool("patches_list", "Live band-aids", + "Every live band-aid with its age, TTL remaining, cure state and escalation count. A " + "pure read of the ledger — no tenant call, no GitHub call, no exploit fired — so it is " + "cheap enough to poll.", + _out_schema({"expired_only": {"type": "boolean", "default": False, + "description": "only patches past their TTL"}}), + _tool_patches_list), + Tool("ledger", "Remediation ledger", + "The lifecycle of every finding: found -> mitigated -> remediated -> retired, with the " + "control attached and the cure PR. Pure ledger read.", + _out_schema(), _tool_ledger), + Tool("impact", "Headline numbers", + "The numbers the report and console render: exploitable vulns, how many are mitigated " + "live, mean time to mitigate against normal change-control days, drafted code-fix PRs, " + "and dependency upgrades that no PR can cover. Pure read of the run dir and ledger.", + _out_schema(), _tool_impact), + Tool("deps", "Dependency survey", + "What a --manifest scan WOULD find, without spending a single model call: parses " + "requirements.txt / package-lock.json / pom.xml, asks OSV.dev which pinned packages " + "have advisories, and returns the whole funnel — including every entry it could not " + "pin and every advisory held back by the filters, each with the reason. Needs no " + "credentials and no model. Reaches api.osv.dev.", + {"type": "object", + "properties": { + "manifest": {"type": "array", "items": {"type": "string"}, + "description": "one or more manifest paths: requirements.txt, " + "package-lock.json, pom.xml"}, + "min_severity": {"type": "string", "enum": ["critical", "high", "medium", "low"], + "default": "high", + "description": "floor a scan would resolve at; below it an " + "advisory is listed but never sent to an agent"}, + "max_advisories": {"type": "integer", "default": 25, + "description": "cap a scan would apply to the agent stage " + "(0 = no cap); shared round-robin across " + "packages, not consumed in sort order"}, + "include_dev": {"type": "boolean", "default": False, + "description": "also count dev/test-scoped dependencies, which " + "are not in the request path"}}, + "required": ["manifest"]}, + _tool_deps, open_world=True), + Tool("simulation_result", "Blast radius", + "The would-block result of a previous shadow simulation: per candidate policy, how " + "many recorded requests it would have blocked, the rate, a sample, and whether it " + "tripped the blast-radius threshold. Reads simulation.json — running a simulation " + "attaches a throwaway policy to a spare load balancer, so this server never does it.", + _out_schema(), _tool_simulation_result), + Tool("drift", "LB drift and conflicts", + "What is on the load balancer now, versus what the last run's snapshot recorded, " + "versus what you are about to push. Reports operator hand-edits as a field-level diff. " + "Pass `policy` as well to also check the candidate for an ALLOW rule that would shadow " + "its own DENY — without it there are no rules to walk and that check does not run. " + "Read-only against the tenant: no PUT, no snapshot written, nothing in the run dir " + "touched. Needs XC credentials.", + _out_schema({ + "lb": {"type": "string", "description": "load balancer to inspect"}, + "control": {"type": "string", + "description": "the control you are about to apply, e.g. " + "service_policy, waf, rate_limit"}, + "policy": {"type": "string", + "description": "service-policy name you are about to attach; also " + "locates the generated artifact for the shadowing check"}, + "finding": {"type": "string", + "description": "use this finding's recorded exploit for the shadowing " + "check"}}, required=["lb"]), + _tool_drift, open_world=True), + Tool("verify_bundle", "Verify evidence bundle", + "Re-read an exported evidence bundle and check every member digest against its own " + "manifest, plus the minisign signature when a public key is supplied. Reports each " + "member as ok / mismatch / missing / unlisted. A bundle with no signature verifies its " + "digests and says the signature is absent rather than failing.", + {"type": "object", + "properties": { + "path": {"type": "string", "description": "path to the bundle .zip"}, + "pubkey": {"type": "string", + "description": "minisign public key, obtained out of band, to check " + "the signature too"}}, + "required": ["path"]}, + _tool_verify_bundle), + # Writes only into the run directory: additive, and nothing on the tenant is touched. + Tool("scan_start", "Start a scan", + "Run the pipeline: discover -> verify -> triage -> generate XC band-aids -> draft the " + "code cure. Read-only with respect to the tenant — it never touches a load balancer — " + "but it SPENDS MODEL CALLS, writes artifacts into the run directory, and usually takes " + "minutes, so it returns a job_id immediately. Poll scan_status. Inputs: a repo path, a " + "CVE/GHSA id (exclusive), an OpenAPI spec and/or dependency manifests (additive).", + {"type": "object", + "properties": { + "repo": {"type": "string", + "description": "path to the source directory to scan"}, + "cve": {"type": "string", + "description": "a CVE or GHSA id to scan instead of a repo; cannot be " + "combined with repo, spec or manifest"}, + "spec": {"type": "string", + "description": "path to an OpenAPI spec; alone it scans the contract, " + "with a repo it also reports spec/code drift"}, + "manifest": {"type": "array", "items": {"type": "string"}, + "description": "dependency manifests, additive with repo and spec"}, + "out": {"type": "string", "default": "out", + "description": "run directory to write artifacts into"}, + "min_confidence": {"type": "number", "default": 0.5, + "description": "drop verified findings below this confidence"}, + "max_files": {"type": "integer", "default": 200, + "description": "cap on files read from the repo"}, + "max_bytes": {"type": "integer", "default": 60000, + "description": "cap on bytes read per file"}, + "draft_code_fixes": {"type": "boolean", "default": True, + "description": "also draft the code cure; off saves roughly " + "half the tokens and yields band-aids only"}, + "min_severity": {"type": "string", "enum": ["critical", "high", "medium", "low"], + "default": "high", + "description": "manifest advisories: floor for reaching the " + "resolve agent"}, + "max_advisories": {"type": "integer", "default": 25, + "description": "manifest advisories: cap on the agent stage " + "(0 = no cap)"}, + "include_dev": {"type": "boolean", "default": False, + "description": "manifest advisories: resolve dev/test-scoped " + "dependencies too"}, + "config_path": {"type": "string", + "description": "config/agents*.yaml to run with; omit for the " + "default model config"}}, + "required": []}, + _tool_scan_start, access=Access.WRITES_OUT, open_world=True), + Tool("scan_status", "Scan progress", + "Poll a scan started by scan_start: its state (running / done / failed), the new log " + "lines since `since`, and the summary once it finishes. Pass log_next back as `since` " + "to tail without re-reading what you already have.", + {"type": "object", + "properties": { + "job_id": {"type": "string", "description": "id returned by scan_start"}, + "since": {"type": "integer", "default": 0, + "description": "number of log lines already seen; use log_next from " + "the previous call"}}, + "required": ["job_id"]}, + _tool_scan_status, access=Access.WRITES_OUT), + ] + if enable_writes: + tools += _write_tools() + return tools + + +def _tool_apply(policy_name: str = "", lb: str = "", url: str = "", finding_id: str | None = None, + dry_run: bool = True, keep: bool = False, refine: bool = True, + allow_protected_lb: bool = False, force: bool = False, + allow_overbroad: bool = False, out: str = "out", log=None, **_) -> dict: + """Attach a generated service policy, validate it live, roll back unless it passed and `keep`. + + Two deliberate differences from the CLI, both narrowing: + + `dry_run` defaults to **True** here where every module function defaults it False. The CLI and + the console each pass an explicit choice made by a human at a keyboard; an MCP tool call is + issued by a model, so the default has to be the one that changes nothing. Applying for real is a + second, explicit call. + + It takes a policy **name**, not a path, and derives the artifact from the run directory — the J2 + precedent: an endpoint that accepts a caller-supplied filesystem path is an arbitrary-file + reader, and a tool invoked by a model is a worse place for one than an endpoint a human drives. + """ + from pathlib import Path + + from . import apply as A + log = log or (lambda m: None) + if not str(policy_name).strip(): + raise Declined("policy_name is required — the name of a generated service policy; " + "list them with scan_result") + if not str(lb).strip(): + raise Declined("lb is required — the load balancer to apply to") + if not str(url).strip(): + raise Declined("url is required — the live host the band-aid is validated against") + # A generated policy name is a slug (`deny-jndi-log4shell-headers`). Anything else is not a + # policy name, and a name carrying path separators would let the derived artifact path leave the + # run directory — so an attacker-chosen JSON could be read and pushed to the tenant as a spec. + # Traversal happens to fail today because the mandatory `service_policy.` prefix makes the first + # segment a directory that must exist, which is luck rather than a design. Both halves are + # checked: the name, and that the path it produced is still inside the run's policies dir. + if not _is_safe_name(policy_name): + raise Declined(f"{policy_name!r} is not a policy name — expected a generated slug of " + "letters, digits, dots, dashes and underscores") + pol_dir = (Path(out) / "policies").resolve() + art = pol_dir / f"service_policy.{policy_name}.json" + if not art.resolve().is_relative_to(pol_dir): + raise Declined(f"{policy_name!r} resolves outside {out!r}/policies — refusing") + if not art.is_file(): + raise Declined(f"no generated artifact for policy {policy_name!r} in {out!r} — " + "scan_result lists the policies this run produced") + if refine and not dry_run: + from .refiner import refine_apply_service_policy + return refine_apply_service_policy( + str(art), lb, url, finding_id=finding_id, name=policy_name, keep=keep, + allow_protected=allow_protected_lb, force=force, allow_overbroad=allow_overbroad, + out_dir=out, log=log) + return A.apply_from_scan(str(art), lb, url, name=policy_name, finding_id=finding_id, + dry_run=dry_run, keep=keep, allow_protected=allow_protected_lb, + force=force, allow_overbroad=allow_overbroad, out_dir=out, log=log) + + +def _tool_pr(finding_id: str = "", repo: str = "", base: str = "main", path_prefix: str = "", + dry_run: bool = True, out: str = "out", log=None, **_) -> dict: + import json as _json + from pathlib import Path + + from .pr import open_pr + if not str(finding_id).strip(): + raise Declined("finding_id is required") + if not str(repo).strip(): + raise Declined("repo is required — the owner/name GitHub slug to open the PR against") + p = Path(out) / "remediations.json" + rems = _json.loads(p.read_text()) if p.is_file() else [] + r = next((x for x in rems if x.get("finding_id") == finding_id), None) + if r is None: + raise Declined(f"no remediation for {finding_id!r} in {out!r}") + if r.get("kind") == "dependency_upgrade": + # H1/H2: the cure is a version bump in someone else's package. `pr.py` declines this too; + # saying so here means the tool does not look like it merely failed. + raise Declined(f"{finding_id} is a dependency advisory, not a code finding — its cure is " + f"{r.get('summary')!r}, which nobody can open a PR against this repo for") + return open_pr(r, repo, base=base, path_prefix=path_prefix, dry_run=dry_run, out_dir=out, + log=log or (lambda m: None)) + + +def _tool_retire(finding_id: str = "", force: bool = False, dry_run: bool = True, + allow_protected_lb: bool = False, out: str = "out", log=None, **_) -> dict: + from .retire import retire_finding + if not str(finding_id).strip(): + raise Declined("finding_id is required") + return retire_finding(out, finding_id, force=force, dry_run=dry_run, + allow_protected=allow_protected_lb, log=log or (lambda m: None)) + + +def _tool_reconcile(apply: bool = False, finding_id: str | None = None, + allow_protected_lb: bool = False, out: str = "out", log=None, **_) -> dict: + """Report-only unless `apply` — the same default the CLI has, for the same reason (I1). + + `force_probe` is deliberately NOT exposed. Its guard lives in the module and needs a single + `--finding`, because replaying every destructive exploit at once is not something to do by + accident; a model deciding to pass it would be exactly that accident.""" + from .reconcile import reconcile + return reconcile(out, apply=apply, finding_id=finding_id, trigger="mcp", + allow_protected=allow_protected_lb, log=log or (lambda m: None)) + + +def _tool_simulate(policy_name: str | None = None, lb: str = "", url: str = "", logs: str = "", + threshold: float | None = None, max_records: int | None = None, + out: str = "out", log=None, **_) -> dict: + """Replay a recorded traffic sample against candidate band-aids and report the would-block set. + + **This is a write tool, and the roadmap listed it as read-only.** G2's simulation creates a + throwaway `-vpcsim` policy object, attaches it to the load balancer, replays through it and + deletes it again. That is a real mutation of a real tenant — the fact that it cleans up after + itself makes it safe, not read-only — so it sits behind the same explicit opt-in as apply. Use + `simulation_result` to read a previous run's numbers without touching anything.""" + from .simulate import candidates_from_out, simulate_policies, write_result + from .traffic import load as load_traffic + log = log or (lambda m: None) + if not str(lb).strip() or not str(url).strip(): + raise Declined("lb and url are both required — the spare load balancer to replay through") + if not str(logs).strip(): + raise Declined("logs is required — a HAR or JSONL traffic sample to replay. Reading the " + "tenant's own request logs is a CLI-only path (`simulate --from-tenant`), " + "because it needs a time window an operator chooses.") + cands = candidates_from_out(out, policy_name) + if not cands: + raise Declined(f"no service_policy artifacts in {out!r} — nothing to simulate") + records, redacted = load_traffic(logs) + if not records: + raise Declined(f"no records ingested from {logs!r}") + kw = {} if threshold is None else {"threshold": threshold} + res = simulate_policies(cands, records, lb=lb, url=url, out_dir=out, max_records=max_records, + source=f"file:{logs}", redacted=redacted, log=log, **kw) + write_result(out, res) + return res.model_dump() if hasattr(res, "model_dump") else dict(res) + + +def _write_tools() -> list[Tool]: + """The mutating surface, present only when the operator started the server with writes enabled. + + Every one of these calls the SAME module function the CLI and console call, so it inherits the + guardrails rather than reimplementing them: `guard_lb` for a protected load balancer, + `PROTECTED_POLICIES` for a protected policy name, `drift.preflight` for operator drift and a + self-shadowing DENY, `simulate.promotion_gate` for blast radius, rollback-unless-`keep`, and an + audit record with identity stamped centrally by `audit.record`. K1 moved the blast-radius gate + out of the console and into the module precisely so this sentence could be true. + + What the server opt-in does NOT do is supply the human. MCP clients are expected to confirm tool + calls with a user, but that is the client's behaviour and not something this server can enforce + or verify — which is the whole reason these tools are off by default and the reason `dry_run` + defaults to True on `apply`, `pr` and `retire`.""" + return [ + Tool("apply", "Apply a band-aid", + "Attach a generated service policy to a load balancer, validate against the finding's " + "own exploit and a legitimate request, and roll back unless it passed and keep=true. " + "DEFAULTS TO dry_run=true: pass dry_run=false to change the tenant. Routes through the " + "same drift, blast-radius, protected-LB and protected-policy gates as the CLI and " + "console. Takes a policy NAME; the artifact is read from the run directory.", + _out_schema({ + "policy_name": {"type": "string", + "description": "name of a generated service policy (see " + "scan_result)"}, + "lb": {"type": "string", "description": "load balancer to attach to"}, + "url": {"type": "string", + "description": "live host the band-aid is validated against"}, + "finding_id": {"type": "string", + "description": "finding whose recorded probe validates this policy; " + "defaults to the ledger lookup"}, + "dry_run": {"type": "boolean", "default": True, + "description": "true (the default) changes nothing; false applies for " + "real"}, + "keep": {"type": "boolean", "default": False, + "description": "leave the policy attached when validation passes; the " + "default rolls back even on success"}, + "refine": {"type": "boolean", "default": True, + "description": "refine the policy until it actually blocks the exploit " + "(ignored when dry_run)"}, + "allow_protected_lb": {"type": "boolean", "default": False, + "description": "permit a load balancer listed in " + "VPCOPILOT_PROTECTED_LBS"}, + "force": {"type": "boolean", "default": False, + "description": "apply despite the pre-apply drift check"}, + "allow_overbroad": {"type": "boolean", "default": False, + "description": "apply despite the blast-radius gate; writes a " + "simulate_override audit record"}}, + required=["policy_name", "lb", "url"]), + _tool_apply, access=Access.MUTATES, open_world=True), + Tool("pr", "Open the cure PR", + "Open the code-fix pull request for a finding — the cure the band-aid is buying time " + "for. DEFAULTS TO dry_run=true. Declines for a dependency advisory, whose cure is a " + "version bump in someone else's package that no PR against this repo can make.", + _out_schema({ + "finding_id": {"type": "string", "description": "finding to open the cure PR for"}, + "repo": {"type": "string", "description": "GitHub slug, owner/name"}, + "base": {"type": "string", "default": "main", + "description": "base branch to open the PR against"}, + "path_prefix": {"type": "string", + "description": "prefix to prepend to the patched file's path, when " + "the scanned directory is not the repo root"}, + "dry_run": {"type": "boolean", "default": True, + "description": "true (the default) previews the PR without creating " + "anything on GitHub"}}, + required=["finding_id", "repo"]), + _tool_pr, access=Access.MUTATES, open_world=True), + Tool("retire", "Retire a band-aid", + "Detach a band-aid once its cure has shipped, moving the finding to retired. DEFAULTS " + "TO dry_run=true. Refuses unless the cure PR is merged, unless force=true.", + _out_schema({ + "finding_id": {"type": "string", "description": "finding whose band-aid to retire"}, + "force": {"type": "boolean", "default": False, + "description": "retire without a merged cure PR"}, + "dry_run": {"type": "boolean", "default": True, + "description": "true (the default) changes nothing"}, + "allow_protected_lb": {"type": "boolean", "default": False, + "description": "permit a protected load balancer"}}, + required=["finding_id"]), + _tool_retire, access=Access.MUTATES, open_world=True), + Tool("reconcile", "Reconcile live band-aids", + "Walk every live band-aid: check its cure PR, re-fire its exploit at the ORIGIN, and " + "act — retire when the cure merged and the exploit is gone, hold and report " + "fix_ineffective when it merged and the exploit still reproduces, escalate when the " + "TTL passed with no merged cure. REPORT-ONLY unless apply=true. Every branch that " + "cannot establish a fact holds the band-aid and says why.", + _out_schema({ + "apply": {"type": "boolean", "default": False, + "description": "false (the default) reports and changes nothing; true " + "detaches a band-aid whose cure is proven"}, + "finding_id": {"type": "string", + "description": "reconcile a single finding instead of every live " + "patch"}, + "allow_protected_lb": {"type": "boolean", "default": False, + "description": "permit a protected load balancer"}}), + _tool_reconcile, access=Access.MUTATES, open_world=True), + Tool("simulate", "Measure blast radius", + "Replay a recorded traffic sample against candidate band-aids on a spare load balancer " + "and report what each WOULD block before anything reaches the gate. This MUTATES the " + "tenant: it creates a throwaway policy object, attaches it, replays, then deletes it — " + "cleaning up after itself makes it safe, not read-only. Use simulation_result to read a " + "previous run's numbers without touching anything.", + _out_schema({ + "policy_name": {"type": "string", + "description": "simulate only this candidate; omit for all of the " + "run's service policies"}, + "lb": {"type": "string", + "description": "spare, non-production load balancer to replay through"}, + "url": {"type": "string", "description": "live host that load balancer fronts"}, + "logs": {"type": "string", + "description": "HAR or JSONL traffic sample to replay; values are redacted " + "at parse time"}, + "threshold": {"type": "number", + "description": "would-block rate above which a policy is marked " + "blocked_promotion; defaults to " + "VPCOPILOT_SIM_THRESHOLD"}, + "max_records": {"type": "integer", + "description": "cap on records replayed per candidate"}}, + required=["lb", "url", "logs"]), + _tool_simulate, access=Access.MUTATES, open_world=True), + ] + + +# --------------------------------------------------------------------------- JSON-RPC plumbing + +@dataclass +class Server: + enable_writes: bool = False + tools: dict[str, Tool] = field(default_factory=dict) + initialized: bool = False + + def __post_init__(self): + if not self.tools: + self.tools = {t.name: t for t in build_tools(enable_writes=self.enable_writes)} + + +_JSON_TYPES: dict[str, tuple] = { + "string": (str,), "integer": (int,), "number": (int, float), + "boolean": (bool,), "array": (list,), "object": (dict,), +} + + +def validate_args(tool: Tool, args: dict) -> str | None: + """Check `args` against the tool's own schema. Returns an error message, or None if it passes. + + The spec says a server MUST validate tool inputs, and there is a sharper reason here: every tool + function takes `**_` so it can be handed a `log` sink uniformly, which means an unrecognised + argument would be silently swallowed. A client sending `output=` instead of `out=` would then get + a confident answer about the DEFAULT run directory — a wrong answer that looks like a right one, + which is the failure this codebase refuses everywhere else. So the schema is authoritative: it is + both the documentation an agent reads and the gate its arguments pass through, and the two + therefore cannot disagree. + """ + props: dict = tool.schema.get("properties", {}) + for name in tool.schema.get("required", []): + if name not in args: + return f"missing required argument {name!r} for {tool.name}" + for name, value in args.items(): + spec = props.get(name) + if spec is None: + close = ", ".join(sorted(props)) + return f"unknown argument {name!r} for {tool.name} — accepted arguments: {close}" + want = spec.get("type") + allowed = _JSON_TYPES.get(want) + # bool is a subclass of int in Python; a JSON boolean is not an integer or a number. + if allowed and (not isinstance(value, allowed) + or (want in ("integer", "number") and isinstance(value, bool))): + return (f"argument {name!r} for {tool.name} must be a {want}, " + f"got {type(value).__name__}") + if spec.get("enum") and value not in spec["enum"]: + return (f"argument {name!r} for {tool.name} must be one of " + f"{', '.join(map(str, spec['enum']))} — got {value!r}") + if want == "array" and spec.get("items", {}).get("type"): + it = _JSON_TYPES.get(spec["items"]["type"]) + if it and not all(isinstance(v, it) for v in value): + return f"every item in {name!r} for {tool.name} must be a {spec['items']['type']}" + return None + + +def _result(mid, payload) -> dict: + return {"jsonrpc": "2.0", "id": mid, "result": payload} + + +def _error(mid, code: int, message: str, data=None) -> dict: + err = {"code": code, "message": message} + if data is not None: + err["data"] = data + return {"jsonrpc": "2.0", "id": mid, "error": err} + + +def _tool_result(payload, *, is_error: bool = False) -> dict: + """A tool result carries the data twice on purpose: `structuredContent` for a client that can + use it, and the same JSON serialised into a text block, which the spec asks for so a + text-only client is not left with nothing. `default=str` means one unexpected value degrades to + a string instead of taking the server down mid-frame.""" + text = payload if isinstance(payload, str) else json.dumps(payload, default=str) + out: dict[str, Any] = {"content": [{"type": "text", "text": text}], "isError": is_error} + if not is_error and not isinstance(payload, str): + out["structuredContent"] = payload if isinstance(payload, dict) else {"result": payload} + return out + + +def handle(msg: dict, srv: Server) -> dict | None: + """One JSON-RPC message in, one response out — or None for a notification, which by definition + gets no reply. Pure: no streams, so tests drive it directly.""" + mid = msg.get("id") + method = msg.get("method") + is_notification = "id" not in msg + + # A notification carries no id and MUST NOT be answered — whatever its method. Checked once, + # here, rather than per branch: without it `{"method":"initialize"}` with no id was answered with + # an unsolicited `id: null` frame, and a client that is not expecting a response desynchronises + # on the next one it reads. + if is_notification: + if method == "notifications/initialized": + srv.initialized = True + return None + + if method == "initialize": + # Version negotiation: if the client asked for something else, answer with what we DO + # support and let it decide, rather than erroring the connection out. + return _result(mid, { + "protocolVersion": PROTOCOL_VERSION, + "capabilities": {"tools": {"listChanged": False}}, + "serverInfo": {"name": SERVER_NAME, "title": "Virtual Patch Copilot", + "version": __version__}, + "instructions": + "Virtual Patch Copilot routes vulnerability findings to F5 Distributed Cloud " + "controls (the band-aid) and to a code fix (the cure). Read a finished run with " + "scan_result, impact and patches_list; survey dependencies with deps; start new " + "work with scan_start then poll scan_status. Tools that would change a load " + "balancer, open a PR or retire a control are absent unless the operator started " + "this server with writes enabled — the human gate is not something a tool call " + "can satisfy on its own.", + }) + + if method == "ping": + return _result(mid, {}) + + if method == "tools/list": + return _result(mid, {"tools": [t.definition() for t in srv.tools.values()]}) + + if method == "tools/call": + params = msg.get("params") or {} + # JSON-RPC permits positional (array) params, and a list is truthy — so `params.get` would + # raise straight out of the loop. Likewise a non-string `name` is unhashable and would blow + # up in the dict lookup below. Both are client-controlled, so both are invalid-params. + if not isinstance(params, dict): + return _error(mid, INVALID_PARAMS, "params must be an object") + name = params.get("name") + if not isinstance(name, str): + return _error(mid, INVALID_PARAMS, "params.name must be a string naming a tool") + args = params.get("arguments") or {} + tool = srv.tools.get(name) + if tool is None: + # An unknown tool is a protocol error, per the spec — not a tool-execution error. + known = ", ".join(sorted(srv.tools)) + return _error(mid, INVALID_PARAMS, f"Unknown tool: {name}", + {"available": sorted(srv.tools), "hint": f"available tools: {known}"}) + if not isinstance(args, dict): + return _error(mid, INVALID_PARAMS, "arguments must be an object") + bad = validate_args(tool, args) + if bad: + # "Invalid arguments" is a protocol error in the spec, not a tool-execution error: the + # call never happened, so reporting it as a tool result would make a client typo + # indistinguishable from a real failure inside a tool that ran. + return _error(mid, INVALID_PARAMS, bad) + lines: list[str] = [] + try: + payload = tool.fn(log=lines.append, **args) + except Declined as e: + # The honest decline: a fact could not be established, and saying so is the answer. + return _result(mid, _tool_result(f"declined: {e}", is_error=True)) + # There is deliberately NO `except TypeError` here. It looks like the natural home for a bad + # argument, but `validate_args` above has already rejected every unknown name and wrong type, + # and a test pins that every documented property is a real parameter — so an argument-binding + # TypeError is unreachable. What a TypeError actually means at this point is a bug INSIDE a + # tool that ran, and reporting that as "invalid arguments" would send an agent round a loop + # retrying different arguments against a fault they cannot influence. + except Exception as e: # noqa: BLE001 — never let one tool call kill the server + detail = "".join(traceback.format_exception_only(type(e), e)).strip() + if lines: + detail += "\nlog:\n" + "\n".join(lines[-20:]) + return _result(mid, _tool_result(f"{name} failed: {detail}", is_error=True)) + if lines and isinstance(payload, dict): + payload = {**payload, "log": lines} + return _result(mid, _tool_result(payload)) + + return _error(mid, METHOD_NOT_FOUND, f"Method not found: {method}") + + +def serve(inp=None, outp=None, *, enable_writes: bool = False, swap_stdout: bool = True) -> None: + """Read newline-delimited JSON-RPC from `inp`, write responses to `outp`. + + `swap_stdout` is the structural guarantee described in the module docstring: `sys.stdout` is + pointed at stderr for the duration, and frames go to a private handle. Any `print` beneath us — + in the pipeline, in an agent, in a library — then lands on stderr, which the spec explicitly + allows the server to use for logging, instead of corrupting the message stream. + """ + inp = inp or sys.stdin + frames = outp or sys.stdout + # The transport mandates UTF-8; Python opens the real stdio streams with the PROCESS LOCALE, so + # on a box with a non-UTF-8 locale a client sending a non-ASCII path or advisory summary would + # get a decode error instead of an answer. Only the real streams are reconfigured — a StringIO + # from a test has no encoding to set. + for stream in (inp, frames): + try: + stream.reconfigure(encoding="utf-8") # type: ignore[union-attr] + except (AttributeError, ValueError, OSError): + pass + srv = Server(enable_writes=enable_writes) + saved = sys.stdout + if swap_stdout: + sys.stdout = sys.stderr + try: + for line in inp: + line = line.strip() + if not line: + continue + try: + msg = json.loads(line) + except json.JSONDecodeError as e: + _write(frames, _error(None, PARSE_ERROR, f"Parse error: {e}")) + continue + if not isinstance(msg, dict): + _write(frames, _error(None, INVALID_REQUEST, "Request must be a JSON object")) + continue + # `handle` is defensive, but the loop does not rely on it being perfect. An unhandled + # exception here used to propagate out of `serve()` and end the process with NO frame + # written at all: the client waits forever on a request that will never be answered, and + # every later request is lost with it. A malformed `params` did exactly that. Dying + # silently is the worst available failure for a transport, so the loop answers with an + # internal error and keeps serving — the same reasoning as the per-tool guard, one level + # up, and it covers bugs not yet written. + try: + resp = handle(msg, srv) + except Exception as e: # noqa: BLE001 + detail = "".join(traceback.format_exception_only(type(e), e)).strip() + resp = _error(msg.get("id"), INTERNAL_ERROR, f"internal error: {detail}") + if resp is not None: + _write(frames, resp) + finally: + if swap_stdout: + sys.stdout = saved + + +def _write(stream, payload: dict) -> None: + """One message, one line. `json.dumps` never emits a raw newline without `indent`, which is why + none is passed — the spec requires messages to contain no embedded newlines.""" + stream.write(json.dumps(payload, default=str) + "\n") + stream.flush() diff --git a/src/vpcopilot/refiner.py b/src/vpcopilot/refiner.py index 2648435..a085965 100644 --- a/src/vpcopilot/refiner.py +++ b/src/vpcopilot/refiner.py @@ -71,7 +71,7 @@ def refine_apply_service_policy(artifact_path: str, lb: str, target_url: str, *, allow_protected: bool = False, config_path: str | None = None, retries: int = 6, wait_seconds: int = 8, records: list | None = None, sim_threshold: float | None = None, - force: bool = False, + force: bool = False, allow_overbroad: bool = False, out_dir: str = "out", log: Callable = print) -> dict: """Create/attach a service policy, validate it live, and refine-until-it-works (or give up honestly). Returns passed + attempts + before/after; persists the WORKING spec to the artifact. @@ -105,6 +105,13 @@ def refine_apply_service_policy(artifact_path: str, lb: str, target_url: str, *, finding = _load_finding(out_dir, finding_id) probe = _load_probe(out_dir, finding_id) + # G2 — this is the DEFAULT path for both the CLI's `--from-scan` and the console's Mitigate + # button, so the blast-radius gate has to be here as well as in `apply_from_scan`; a guard the + # primary UX skips is not a guard. Same reasoning as the drift preflight below. + from .simulate import promotion_gate + promotion_gate(out_dir, policy_name, allow_overbroad=allow_overbroad, + finding_id=finding_id, lb=lb, log=log) + from .drift import preflight d = preflight(lb, policy_name, out_dir=out_dir, force=force, xc=xc, log=log, spec=spec, exploit=(probe or {}).get("exploit"), refine=True) diff --git a/src/vpcopilot/schemas.py b/src/vpcopilot/schemas.py index 60aee57..673680a 100644 --- a/src/vpcopilot/schemas.py +++ b/src/vpcopilot/schemas.py @@ -255,6 +255,10 @@ class PolicySimulation(BaseModel): top_paths: list[list] = Field(default_factory=list, description="[[path, count], ...] of blocks") top_user_agents: list[list] = Field(default_factory=list) error: str = "" + carried_from: str = Field( + "", description="set when this entry came from an EARLIER replay and was preserved through a " + "later, narrower one (`simulate --policy X`). Empty means it was measured by " + "the run whose metadata heads this artifact.") class SimulationResult(BaseModel): diff --git a/src/vpcopilot/simulate.py b/src/vpcopilot/simulate.py index fd9bf6d..e3ff723 100644 --- a/src/vpcopilot/simulate.py +++ b/src/vpcopilot/simulate.py @@ -277,9 +277,38 @@ def candidates_from_out(out_dir: str, policy: str | None = None) -> list[dict]: def write_result(out_dir: str, res: SimulationResult) -> str: + """Write `simulation.json`, PRESERVING any policy this run did not measure. + + A narrower replay used to erase the wider one. `simulate --policy B` filters the candidates to B + and this function overwrote the artifact wholesale, so a policy A that an earlier run had flagged + `blocked_promotion` lost its flag — and `promotion_block`, the pre-apply blast-radius gate, went + quiet for it. An operator who simulated everything, saw A flagged, then re-simulated only B would + find A applying with no warning at all: a guard erased as a side effect of measuring something + else, which is the shape of I1's "a band-aid could vouch for its own removal". + + So entries absent from this run are carried forward and stamped `carried_from` with the timestamp + of the run that did measure them. The gate keeps firing, and nothing pretends the number is fresh. + """ p = Path(out_dir) / "simulation.json" p.parent.mkdir(parents=True, exist_ok=True) - p.write_text(json.dumps(res.model_dump(), indent=2)) + payload = res.model_dump() + prev = load_result(out_dir) + if prev: + fresh = {pol.get("policy_name") for pol in payload.get("policies") or []} + carried = [] + for pol in prev.get("policies") or []: + if pol.get("policy_name") in fresh: + continue + pol = {**pol, "carried_from": pol.get("carried_from") or prev.get("ts", "")} + carried.append(pol) + if carried: + payload["policies"] = (payload.get("policies") or []) + carried + payload["caveats"] = (payload.get("caveats") or []) + [ + f"{len(carried)} policy result(s) carried forward from an earlier replay and NOT " + f"measured by this one ({', '.join(c['policy_name'] for c in carried)}) — each " + "carries `carried_from`. They are preserved so the pre-apply blast-radius gate does " + "not go quiet for a policy this narrower run did not look at."] + p.write_text(json.dumps(payload, indent=2)) return str(p) @@ -302,3 +331,38 @@ def promotion_block(out_dir: str, policy_name: str) -> dict | None: if p.get("policy_name") == policy_name and p.get("blocked_promotion"): return p return None + + +def promotion_gate(out_dir: str, policy_name: str | None, *, allow_overbroad: bool = False, + dry_run: bool = False, finding_id: str | None = None, lb: str | None = None, + log: Callable = print) -> None: + """Enforce the G2 blast-radius decision. Raises unless the override is explicit. + + **K1 moved this out of the console and into the module, and that is a behaviour change worth + stating.** `promotion_block` shipped with exactly one production caller — `console/app.py` — so + the guard the roadmap describes existed on one of two surfaces: `vpcopilot apply --from-scan` + would happily attach an over-broad policy that the console refuses, and the `--allow-overbroad` + flag ROADMAP.md claims never existed. That is the same shape as I1's `--force-probe`, whose guard + lived only in the CLI and left the console able to mass-replay every destructive exploit. A guard + in one surface is not a guard, and K1 could not honestly claim its write tools "route through the + same gate as the CLI and console" while the two disagreed about what the gate was. + + Still a warn-with-audited-override, never a machine veto (the G2/I2 precedent): exceeding the + threshold requires `--allow-overbroad`, and taking the override writes a `simulate_override` + audit record naming the rate, the threshold and the actor. Silent when nothing was simulated — + G2 adds a check, not a prerequisite.""" + if dry_run or not policy_name: + return + over = promotion_block(out_dir, policy_name) + if over is None: + return + if not allow_overbroad: + raise RuntimeError( + f"simulation says this policy {over.get('reason')}. Re-run with 'allow overbroad' " + f"(CLI: --allow-overbroad) to apply it anyway." + ) + log(f"⚠ overbroad override: {over.get('reason')}") + from .audit import record + record(out_dir, "simulate_override", finding_id=finding_id, policy=policy_name, lb=lb, + block_rate=over.get("block_rate"), threshold=over.get("threshold"), + reason=over.get("reason")) diff --git a/tests/test_console_simulate.py b/tests/test_console_simulate.py index 34ff40a..660b373 100644 --- a/tests/test_console_simulate.py +++ b/tests/test_console_simulate.py @@ -12,6 +12,18 @@ def _client(): return TestClient(A.app) +def _artifact(out, name="deny-wide"): + """K1 moved the G2 gate out of `_dispatch_action` and into `simulate.promotion_gate`, called by + both apply paths so the CLI and the MCP server get it too. It therefore fires after the artifact + is read rather than before — the gate keys on the policy name, which the artifact can supply — + so a test reaching it needs the artifact a real run would have generated. Nothing reaches the + tenant: the gate raises before any XC call that mutates.""" + d = out / "policies" + d.mkdir(parents=True, exist_ok=True) + (d / f"service_policy.{name}.json").write_text(json.dumps( + {"metadata": {"name": name}, "spec": {"rules": []}})) + + def _sim(out, **over): (out / "simulation.json").write_text(json.dumps({ "ts": "2026-07-27T12:00:00Z", "lb": "vpcopilot-lab", "records": 200, "records_replayed": 200, @@ -41,15 +53,21 @@ def test_simulate_endpoint_is_empty_before_any_run(tmp_path, monkeypatch): # ---- the gate: warn + explicit override, audited ---- def test_an_overbroad_policy_is_refused_without_the_override(tmp_path, monkeypatch): _sim(tmp_path) + _artifact(tmp_path) monkeypatch.setattr(A, "OUT", tmp_path) r = _client().post("/api/action", json={"control": "service_policy", "policy_name": "deny-wide", "finding_id": "f-1", "lb": "lab", "dry_run": False}) assert r.status_code == 200 # the job starts… job = r.json()["job"] - for _ in range(50): + # The gate now sits a little further into the job (after the artifact read), so a busy poll with + # no sleep can finish before the worker thread does. Nothing reaches the tenant either way: + # `XC()` only builds an httpx client, and the raise precedes every request. + import time + for _ in range(200): s = _client().get(f"/api/action?job={job}").json() if s["state"] != "running": break + time.sleep(0.02) assert s["state"] == "error" and "allow overbroad" in s["error"] diff --git a/tests/test_mcp.py b/tests/test_mcp.py new file mode 100644 index 0000000..0e8fed2 --- /dev/null +++ b/tests/test_mcp.py @@ -0,0 +1,809 @@ +"""K1 — MCP server mode. Offline throughout: every test drives the JSON-RPC layer directly or over +StringIO streams. No network, no tenant, no model.""" +from __future__ import annotations + +import io +import json + +import pytest + +from vpcopilot import mcp + + +def _drive(messages: list[dict], *, enable_writes: bool = False, swap_stdout: bool = False) -> list[dict]: + """Feed newline-delimited JSON-RPC through `serve` and parse what comes back.""" + inp = io.StringIO("".join(json.dumps(m) + "\n" for m in messages)) + out = io.StringIO() + mcp.serve(inp, out, enable_writes=enable_writes, swap_stdout=swap_stdout) + return [json.loads(ln) for ln in out.getvalue().splitlines() if ln.strip()] + + +def _init() -> dict: + return {"jsonrpc": "2.0", "id": 1, "method": "initialize", + "params": {"protocolVersion": mcp.PROTOCOL_VERSION, "capabilities": {}, + "clientInfo": {"name": "test", "version": "1"}}} + + +def _call(name, args=None, mid=9) -> dict: + return {"jsonrpc": "2.0", "id": mid, "method": "tools/call", + "params": {"name": name, "arguments": args or {}}} + + +# ============================================================ lifecycle +def test_initialize_declares_tools_and_our_protocol_version(): + (r,) = _drive([_init()]) + res = r["result"] + assert res["protocolVersion"] == mcp.PROTOCOL_VERSION + assert res["capabilities"]["tools"] is not None + assert res["serverInfo"]["name"] == "vpcopilot" + assert res["instructions"] # an agent session's only orientation + + +def test_a_client_asking_for_another_version_gets_ours_not_an_error(): + """The spec: if the server does not support the requested version it MUST respond with one it + does support, and let the client decide whether to disconnect. Erroring the connection out + instead would make every future protocol bump a hard break.""" + msg = _init() + msg["params"]["protocolVersion"] = "1999-01-01" + (r,) = _drive([msg]) + assert "error" not in r + assert r["result"]["protocolVersion"] == mcp.PROTOCOL_VERSION + + +def test_notifications_get_no_reply_at_all(): + """A JSON-RPC notification has no id and MUST NOT be answered. Answering one makes a client + that is not expecting a frame desynchronise.""" + out = _drive([{"jsonrpc": "2.0", "method": "notifications/initialized"}, + {"jsonrpc": "2.0", "method": "notifications/something/unknown"}, + _init()]) + assert [r["id"] for r in out] == [1] + + +def test_ping_is_answered_when_it_is_a_request(): + out = _drive([{"jsonrpc": "2.0", "id": 7, "method": "ping"}]) + assert out == [{"jsonrpc": "2.0", "id": 7, "result": {}}] + + +# ============================================================ framing +def test_every_frame_is_one_line_of_valid_json_with_no_embedded_newline(): + """stdio framing: messages are newline-delimited and MUST NOT contain embedded newlines. A + tool result carrying a multi-line log is the obvious way to break this by accident.""" + inp = io.StringIO(json.dumps(_init()) + "\n" + json.dumps(_call("scan_result", {"out": "nope"})) + "\n") + out = io.StringIO() + mcp.serve(inp, out, swap_stdout=False) + raw = out.getvalue() + assert raw.endswith("\n") + lines = raw.splitlines() + assert len(lines) == 2 # exactly one frame per request + for ln in lines: + json.loads(ln) # parses standalone + + +def test_malformed_json_is_a_parse_error_and_the_loop_survives(): + inp = io.StringIO("{not json\n" + json.dumps(_init()) + "\n") + out = io.StringIO() + mcp.serve(inp, out, swap_stdout=False) + got = [json.loads(ln) for ln in out.getvalue().splitlines()] + assert got[0]["error"]["code"] == mcp.PARSE_ERROR and got[0]["id"] is None + assert got[1]["result"]["protocolVersion"] # still serving afterwards + + +def test_a_non_object_request_is_rejected_without_killing_the_server(): + inp = io.StringIO("[1,2,3]\n" + json.dumps(_init()) + "\n") + out = io.StringIO() + mcp.serve(inp, out, swap_stdout=False) + got = [json.loads(ln) for ln in out.getvalue().splitlines()] + assert got[0]["error"]["code"] == mcp.INVALID_REQUEST + assert len(got) == 2 + + +def test_an_unknown_method_is_method_not_found(): + (r,) = _drive([{"jsonrpc": "2.0", "id": 3, "method": "resources/list"}]) + assert r["error"]["code"] == mcp.METHOD_NOT_FOUND + + +# ============================================================ stdout is the protocol's +def test_a_tool_that_prints_to_stdout_cannot_corrupt_the_stream(monkeypatch, capsys): + """The load-bearing guarantee. `run_pipeline`, `survey_report` and `drift.check` all default + `log=print`, and `rprint` is used throughout the CLI — one stray line on stdout and the client + drops the connection. Passing `log=` at every call site is a convention; `serve()` reassigning + `sys.stdout` is a guarantee that also covers a print we did not write.""" + def noisy(**kw): + print("this would corrupt the protocol stream") + return {"ok": True} + + monkeypatch.setattr(mcp, "build_tools", lambda **kw: [ + mcp.Tool("noisy", "Noisy", "prints to stdout", {"type": "object", "properties": {}}, noisy)]) + inp = io.StringIO(json.dumps(_call("noisy")) + "\n") + out = io.StringIO() + mcp.serve(inp, out, swap_stdout=True) + # the frame stream holds exactly one clean message… + (frame,) = [json.loads(ln) for ln in out.getvalue().splitlines()] + assert frame["result"]["structuredContent"] == {"ok": True} + # …and the stray line went to stderr, which the spec explicitly allows for logging + assert "corrupt the protocol stream" in capsys.readouterr().err + + +def test_serve_restores_stdout_afterwards(monkeypatch): + """A test process (or a CLI that keeps running) must not be left with stdout pointing at + stderr.""" + import sys + before = sys.stdout + mcp.serve(io.StringIO(""), io.StringIO(), swap_stdout=True) + assert sys.stdout is before + + +# ============================================================ the tool contract +def test_every_tool_documents_every_argument(): + """Acceptance criterion, made mechanical: "tool schemas document every argument". A schema is + the only documentation an agent session ever sees, so a missing description is a real defect, + not a style nit.""" + for t in mcp.build_tools(enable_writes=True): + d = t.definition() + assert d["description"].strip(), f"{t.name} has no description" + assert d["inputSchema"]["type"] == "object" + for arg, spec in d["inputSchema"]["properties"].items(): + assert spec.get("description", "").strip(), f"{t.name}.{arg} has no description" + assert spec.get("type"), f"{t.name}.{arg} has no type" + + +def test_read_only_tools_assert_the_hint_rather_than_relying_on_the_default(): + """`readOnlyHint` defaults to FALSE and `destructiveHint` defaults to TRUE in the MCP schema, so + an omitted annotation describes the opposite of a read. Every hint here is derived from one + `Access` value per tool, so the two cannot drift apart.""" + by_name = {t.name: t for t in mcp.build_tools(enable_writes=True)} + for name in ("scan_result", "patches_list", "ledger", "impact", "deps", "simulation_result", + "drift", "verify_bundle"): + ann = by_name[name].definition()["annotations"] + assert ann["readOnlyHint"] is True, name + assert ann["destructiveHint"] is False, name + # a scan writes artifacts into the run dir, so it is not read-only — but it is additive and + # never touches the tenant, so it is not destructive either + for name in ("scan_start", "scan_status"): + ann = by_name[name].definition()["annotations"] + assert ann["readOnlyHint"] is False, name + assert ann["destructiveHint"] is False, name + + +def test_tool_names_are_unique_and_stable(): + names = [t.name for t in mcp.build_tools(enable_writes=True)] + assert len(names) == len(set(names)) + # the read-only set is the contract an agent session depends on; renaming one is a breaking change + assert {"scan_result", "patches_list", "ledger", "impact", "deps", "simulation_result", + "drift", "verify_bundle", "scan_start", "scan_status"} <= set(names) + + +def test_an_unknown_tool_is_a_protocol_error_not_a_tool_error(): + """The spec puts "unknown tool" under protocol errors. Returning isError instead would make a + typo indistinguishable from a real failure inside a tool that does exist.""" + out = _drive([_call("no_such_tool")]) + assert out[0]["error"]["code"] == mcp.INVALID_PARAMS + assert "available" in out[0]["error"]["data"] + + +def test_an_unknown_argument_is_rejected_rather_than_silently_ignored(): + """Every tool function takes `**_` so it can be handed a `log` sink uniformly, which means an + unrecognised argument would be swallowed. A client sending `output=` instead of `out=` would then + get a confident answer about the DEFAULT run directory — a wrong answer wearing the shape of a + right one. The schema is the gate as well as the documentation.""" + out = _drive([_call("patches_list", {"output": "/somewhere/else"})]) + assert out[0]["error"]["code"] == mcp.INVALID_PARAMS + assert "unknown argument 'output'" in out[0]["error"]["message"] + assert "out" in out[0]["error"]["message"] # names what it should have been + + +def test_a_missing_required_argument_is_invalid_params(): + out = _drive([_call("drift", {}), _call("verify_bundle", {}, mid=10), + _call("scan_status", {}, mid=11)]) + for r in out: + assert r["error"]["code"] == mcp.INVALID_PARAMS + assert "missing required argument" in r["error"]["message"] + + +def test_argument_types_are_checked_against_the_schema(): + cases = [ + ("patches_list", {"out": 7}, "must be a string"), + ("patches_list", {"expired_only": "yes"}, "must be a boolean"), + ("scan_start", {"repo": "./a", "max_files": 1.5}, "must be a integer"), + ("scan_start", {"repo": "./a", "manifest": [1, 2]}, "must be a string"), + ("deps", {"manifest": ["r.txt"], "min_severity": "urgent"}, "must be one of"), + ] + for i, (name, args, expect) in enumerate(cases): + (r,) = _drive([_call(name, args, mid=100 + i)]) + assert r["error"]["code"] == mcp.INVALID_PARAMS, (name, args) + assert expect in r["error"]["message"], (name, args, r["error"]["message"]) + + +def test_a_boolean_is_not_accepted_where_a_number_is_required(): + """`bool` is a subclass of `int` in Python, so a naive isinstance check would let + `max_files=True` through and then use it as the integer 1.""" + (r,) = _drive([_call("scan_start", {"repo": "./a", "max_files": True})]) + assert r["error"]["code"] == mcp.INVALID_PARAMS and "must be a integer" in r["error"]["message"] + + +def test_valid_arguments_still_pass(tmp_path): + (r,) = _drive([_call("patches_list", {"out": str(tmp_path), "expired_only": True})]) + assert "error" not in r and r["result"]["isError"] is False + + +def test_arguments_must_be_an_object(): + out = _drive([{"jsonrpc": "2.0", "id": 4, "method": "tools/call", + "params": {"name": "patches_list", "arguments": [1, 2]}}]) + assert out[0]["error"]["code"] == mcp.INVALID_PARAMS + + +# ============================================================ declining, not guessing +def test_reading_a_run_dir_that_holds_no_scan_declines_instead_of_looking_clean(tmp_path): + """The invariant this project is built on: "we did not check this" must never render the same + way as "this is clean". An empty summary would have read as a scan that found nothing.""" + out = _drive([_call("scan_result", {"out": str(tmp_path)})]) + res = out[0]["result"] + assert res["isError"] is True + assert "declined" in res["content"][0]["text"] + assert "nothing has been scanned" in res["content"][0]["text"] + assert "structuredContent" not in res # nothing that could be mistaken for a result + + +def test_a_missing_simulation_declines_and_says_why_it_will_not_run_one(tmp_path): + out = _drive([_call("simulation_result", {"out": str(tmp_path)})]) + txt = out[0]["result"]["content"][0]["text"] + assert out[0]["result"]["isError"] is True + assert "no simulation.json" in txt and "throwaway policy" in txt + + +def test_deps_declines_on_an_empty_manifest_list(): + """`required` is satisfied — the key is present — so the schema lets it through and the module + guard is what catches it. A bad severity is caught earlier, by the schema's enum.""" + out = _drive([_call("deps", {"manifest": []})]) + assert out[0]["result"]["isError"] + assert "at least one manifest" in out[0]["result"]["content"][0]["text"] + + +def test_drift_and_verify_bundle_decline_on_a_blank_value_that_passed_the_schema(): + """The schema catches a MISSING argument; these catch one that is present and empty, which a + JSON-Schema `required` check waves through. Both guards live in the module, so a direct Python + caller gets the same answer as an MCP client.""" + out = _drive([_call("drift", {"lb": " "}), _call("verify_bundle", {"path": ""}, mid=10)]) + assert out[0]["result"]["isError"] and "lb is required" in out[0]["result"]["content"][0]["text"] + assert out[1]["result"]["isError"] and "path is required" in out[1]["result"]["content"][0]["text"] + + +def test_a_tool_that_raises_is_reported_not_fatal(monkeypatch): + def boom(**kw): + raise RuntimeError("the tenant said no") + + monkeypatch.setattr(mcp, "build_tools", lambda **kw: [ + mcp.Tool("boom", "Boom", "raises", {"type": "object", "properties": {}}, boom)]) + out = _drive([_call("boom"), {"jsonrpc": "2.0", "id": 5, "method": "ping"}]) + assert out[0]["result"]["isError"] is True + assert "the tenant said no" in out[0]["result"]["content"][0]["text"] + assert out[1] == {"jsonrpc": "2.0", "id": 5, "result": {}} # server kept serving + + +# ============================================================ results carry data twice +def test_a_result_carries_both_structured_and_text_json(tmp_path): + (tmp_path / "ledger.json").write_text(json.dumps( + {"f-1": {"finding_id": "f-1", "state": "found"}})) + out = _drive([_call("ledger", {"out": str(tmp_path)})]) + res = out[0]["result"] + assert res["structuredContent"]["count"] == 1 + assert json.loads(res["content"][0]["text"])["count"] == 1 # same payload, text form + + +def test_a_payload_with_an_unserialisable_value_degrades_instead_of_killing_the_frame(monkeypatch): + class Weird: + def __repr__(self): + return "" + + monkeypatch.setattr(mcp, "build_tools", lambda **kw: [ + mcp.Tool("weird", "Weird", "returns a non-JSON value", + {"type": "object", "properties": {}}, lambda **kw: {"v": Weird()})]) + out = _drive([_call("weird")]) + assert "" in out[0]["result"]["content"][0]["text"] + + +# ============================================================ the scan job +def test_scan_start_validates_the_input_combination_before_spending_anything(): + """Same rules the CLI and console enforce. Checked here so the answer is a clean decline rather + than a traceback out of run_pipeline's own guard.""" + out = _drive([_call("scan_start", {}), + _call("scan_start", {"repo": "./app", "cve": "CVE-2024-23334"}, mid=11)]) + assert out[0]["result"]["isError"] and "give a repo path" in out[0]["result"]["content"][0]["text"] + assert out[1]["result"]["isError"] + assert "cannot be combined" in out[1]["result"]["content"][0]["text"] + + +def test_scan_status_declines_on_an_unknown_job(): + out = _drive([_call("scan_status", {"job_id": "scan-nope"})]) + assert out[0]["result"]["isError"] + assert "no such scan job" in out[0]["result"]["content"][0]["text"] + + +def test_a_scan_runs_in_the_background_and_is_polled(monkeypatch, tmp_path): + """A scan takes minutes and an MCP call is request/response, so scan_start returns immediately + and scan_status tails the log — the start-then-poll shape the console already uses.""" + import time + + from vpcopilot import pipeline + started = {} + + def fake_pipeline(repo=None, **kw): + started["repo"] = repo + started["out"] = kw.get("out_dir") + log = kw["log"] + log("discovering…") + log("done") + return {"candidates": 1, "verified": 1, "out_dir": kw.get("out_dir")} + + monkeypatch.setattr(pipeline, "run_pipeline", fake_pipeline) + srv = mcp.Server() + start = mcp.handle(_call("scan_start", {"repo": str(tmp_path), "out": str(tmp_path)}), srv) + job_id = start["result"]["structuredContent"]["job_id"] + assert start["result"]["structuredContent"]["state"] == "running" + for _ in range(200): # the thread is real; wait for it + st = mcp.handle(_call("scan_status", {"job_id": job_id}), srv)["result"]["structuredContent"] + if st["state"] != "running": + break + time.sleep(0.01) + assert st["state"] == "done" + assert st["summary"]["verified"] == 1 + assert st["log"][:1] == ["discovering…"] + assert started["repo"] == str(tmp_path) + # `since` tails without re-reading what was already seen + tail = mcp.handle(_call("scan_status", {"job_id": job_id, "since": st["log_next"]}), + srv)["result"]["structuredContent"] + assert tail["log"] == [] + + +def test_a_failing_scan_is_a_reported_state_not_a_dead_server(monkeypatch, tmp_path): + import time + + from vpcopilot import pipeline + monkeypatch.setattr(pipeline, "run_pipeline", + lambda *a, **k: (_ for _ in ()).throw(RuntimeError("no API key"))) + srv = mcp.Server() + job_id = mcp.handle(_call("scan_start", {"repo": str(tmp_path), "out": str(tmp_path)}), + srv)["result"]["structuredContent"]["job_id"] + for _ in range(200): + st = mcp.handle(_call("scan_status", {"job_id": job_id}), srv)["result"]["structuredContent"] + if st["state"] != "running": + break + time.sleep(0.01) + assert st["state"] == "failed" and "no API key" in st["error"] + + +def test_the_scan_log_sink_is_never_stdout(monkeypatch, tmp_path, capsys): + """Independently of the sys.stdout swap: the sink handed to run_pipeline must be a list, so a + scan's progress lines are returned to the caller rather than written anywhere.""" + import time + + from vpcopilot import pipeline + monkeypatch.setattr(pipeline, "run_pipeline", + lambda *a, **k: (k["log"]("a line"), {"ok": True})[1]) + srv = mcp.Server() + job_id = mcp.handle(_call("scan_start", {"repo": str(tmp_path), "out": str(tmp_path)}), + srv)["result"]["structuredContent"]["job_id"] + for _ in range(200): + st = mcp.handle(_call("scan_status", {"job_id": job_id}), srv)["result"]["structuredContent"] + if st["state"] != "running": + break + time.sleep(0.01) + assert "a line" in st["log"] + assert "a line" not in capsys.readouterr().out + + +# ============================================================ writes are absent by default +def test_write_tools_are_absent_unless_enabled(): + """Acceptance: apply/pr/retire/reconcile are ABSENT from the tool list unless explicitly + enabled — not present-and-refusing. A tool an agent can see is a tool it will try.""" + read_only = {t.name for t in mcp.build_tools()} + for forbidden in ("apply", "pr", "retire", "reconcile", "simulate"): + assert forbidden not in read_only + + +def test_calling_a_write_tool_that_is_not_enabled_is_an_unknown_tool(): + out = _drive([_call("apply", {"lb": "x"})]) + assert out[0]["error"]["code"] == mcp.INVALID_PARAMS + assert "apply" not in out[0]["error"]["data"]["available"] + + +@pytest.mark.parametrize("name", ["scan_result", "patches_list", "ledger", "impact"]) +def test_a_read_tool_never_writes_into_an_empty_run_dir(tmp_path, name): + """Polling a read tool must not create artifacts — an MCP client may call these constantly, and + a read that writes would corrupt the run dir just by being observed (the same reason I2's drift + endpoint had to stay read-only).""" + before = set(p.name for p in tmp_path.iterdir()) + _drive([_call(name, {"out": str(tmp_path)})]) + assert set(p.name for p in tmp_path.iterdir()) == before + + +# ============================================================ the write surface +WRITE_TOOLS = ("apply", "pr", "retire", "reconcile", "simulate") + + +def test_the_write_tools_appear_only_when_enabled(): + names = {t.name for t in mcp.build_tools(enable_writes=True)} + for n in WRITE_TOOLS: + assert n in names + assert {t.name for t in mcp.build_tools()}.isdisjoint(WRITE_TOOLS) + + +def test_write_tools_are_marked_destructive(): + """`destructiveHint` defaults to true in the schema, but these assert it, because a client that + surfaces a confirmation prompt keys off exactly this.""" + by = {t.name: t for t in mcp.build_tools(enable_writes=True)} + for n in WRITE_TOOLS: + ann = by[n].definition()["annotations"] + assert ann["readOnlyHint"] is False and ann["destructiveHint"] is True, n + + +def test_simulate_is_classified_as_mutating_not_read_only(): + """The roadmap listed `simulate` among the read-only tools. G2's own implementation creates a + throwaway `-vpcsim` policy, ATTACHES it to the load balancer, replays through it and + deletes it. Cleaning up after itself makes it safe, not read-only — so it sits behind the same + opt-in as apply, and `simulation_result` is the ungated way to read the numbers.""" + by = {t.name: t for t in mcp.build_tools(enable_writes=True)} + assert by["simulate"].access is mcp.Access.MUTATES + assert by["simulation_result"].access is mcp.Access.READ + + +def test_apply_pr_and_retire_default_to_a_dry_run(): + """Every module function defaults `dry_run=False`; the CLI and console each pass a choice a human + made at a keyboard. An MCP tool call is issued by a MODEL, so the default here has to be the one + that changes nothing, and applying for real has to be a second explicit call.""" + by = {t.name: t for t in mcp.build_tools(enable_writes=True)} + for n in ("apply", "pr", "retire"): + assert by[n].definition()["inputSchema"]["properties"]["dry_run"]["default"] is True, n + # reconcile's equivalent is `apply`, report-only by default (I1) + assert by["reconcile"].definition()["inputSchema"]["properties"]["apply"]["default"] is False + + +def test_every_schema_default_matches_the_function_default(): + """A documented default that the code does not honour is a lie an agent acts on. The schema is + the only documentation an MCP client sees, so the two are compared rather than trusted.""" + import inspect + for t in mcp.build_tools(enable_writes=True): + sig = inspect.signature(t.fn) + for arg, spec in t.definition()["inputSchema"]["properties"].items(): + if "default" not in spec: + continue + assert arg in sig.parameters, f"{t.name}.{arg} is documented but not a parameter" + actual = sig.parameters[arg].default + assert actual == spec["default"], \ + f"{t.name}.{arg}: schema says {spec['default']!r}, code says {actual!r}" + + +def test_every_required_argument_is_a_real_parameter(): + import inspect + for t in mcp.build_tools(enable_writes=True): + sig = inspect.signature(t.fn) + for arg in t.definition()["inputSchema"].get("required", []): + assert arg in sig.parameters, f"{t.name} requires {arg} but does not accept it" + + +def test_apply_declines_when_the_named_policy_has_no_artifact(tmp_path): + """It takes a policy NAME and derives the artifact from the run directory — the J2 precedent: an + endpoint taking a caller-supplied filesystem path is an arbitrary-file reader, and a tool a model + invokes is a worse place for one than an endpoint a human drives.""" + out = _drive([_call("apply", {"policy_name": "nope", "lb": "lab", "url": "http://x"})], + enable_writes=True) + assert out[0]["result"]["isError"] + assert "no generated artifact" in out[0]["result"]["content"][0]["text"] + + +def test_apply_takes_no_filesystem_path_at_all(): + by = {t.name: t for t in mcp.build_tools(enable_writes=True)} + props = by["apply"].definition()["inputSchema"]["properties"] + assert "artifact" not in props and "artifact_path" not in props and "path" not in props + + +def test_pr_declines_for_a_dependency_advisory(tmp_path): + """H1/H2: the cure is a version bump in someone else's package. Saying so plainly means the tool + does not merely look like it failed.""" + (tmp_path / "remediations.json").write_text(json.dumps([ + {"finding_id": "GHSA-x", "kind": "dependency_upgrade", + "summary": "upgrade aiohttp 3.9.1 -> 3.9.2", "pr_title": "t", "pr_body": "b"}])) + out = _drive([_call("pr", {"finding_id": "GHSA-x", "repo": "o/n", "out": str(tmp_path)})], + enable_writes=True) + txt = out[0]["result"]["content"][0]["text"] + assert out[0]["result"]["isError"] and "dependency advisory" in txt and "3.9.2" in txt + + +def test_pr_declines_when_there_is_no_remediation(tmp_path): + out = _drive([_call("pr", {"finding_id": "absent", "repo": "o/n", "out": str(tmp_path)})], + enable_writes=True) + assert out[0]["result"]["isError"] + assert "no remediation" in out[0]["result"]["content"][0]["text"] + + +def test_reconcile_does_not_expose_force_probe(): + """I1's review found `--force-probe`'s guard living only in the CLI, leaving the console able to + mass-replay every destructive exploit. Its guard needs a single `--finding`, and a model deciding + to pass it is exactly the accident the guard exists for — so the argument is absent.""" + by = {t.name: t for t in mcp.build_tools(enable_writes=True)} + assert "force_probe" not in by["reconcile"].definition()["inputSchema"]["properties"] + + +def test_reconcile_records_mcp_as_the_trigger(monkeypatch, tmp_path): + """An audit trail that cannot say an action came from an agent session is missing the fact a + reviewer most wants.""" + from vpcopilot import reconcile as R + seen = {} + monkeypatch.setattr(R, "reconcile", + lambda out, **kw: (seen.update(kw), {"checked": 0, "pass_id": "p"})[1]) + _drive([_call("reconcile", {"out": str(tmp_path)})], enable_writes=True) + assert seen["trigger"] == "mcp" + assert seen["apply"] is False # report-only by default + + +def test_simulate_declines_without_a_traffic_sample(tmp_path): + out = _drive([_call("simulate", {"lb": "lab", "url": "http://x", "logs": ""}, + )], enable_writes=True) + assert out[0]["result"]["isError"] + assert "logs is required" in out[0]["result"]["content"][0]["text"] + + +def test_the_write_tools_call_the_same_module_functions_as_the_cli(): + """Acceptance: "the server calls the same module functions as the CLI and console". Checked by + reading the module source, so a future rewrite that inlines its own tenant call fails here.""" + import inspect + src = inspect.getsource(mcp) + for fn in ("refine_apply_service_policy", "apply_from_scan", "open_pr", "retire_finding", + "reconcile", "simulate_policies"): + assert fn in src, fn + # and it must not reach the tenant itself + assert "from .xc import" not in src and "XC()" not in src + + +# ============================================================ the launcher +def test_the_cli_exposes_mcp_and_defaults_writes_off(monkeypatch): + import inspect + + from vpcopilot import cli + assert "write" in inspect.signature(cli.mcp).parameters + src = inspect.getsource(cli.mcp) + assert "VPCOPILOT_MCP_WRITE" in src + assert "stderr" in src # the banner must not touch stdout + assert "rprint" not in src + + +def test_a_policy_name_cannot_become_a_path(tmp_path): + """`apply` derives the artifact from `out` + `policy_name`, so a name carrying path separators + would let the derived path leave the run directory and push an attacker-chosen JSON to the tenant + as a policy spec. Traversal happens to fail today because the mandatory `service_policy.` prefix + makes the first segment a directory that must exist — luck, not design. Both the name and the + resolved path are checked.""" + (tmp_path / "policies").mkdir() + for bad in ("../../etc/passwd", "a/b", "..", "x/../../../y", "a\\b", "name with spaces"): + out = _drive([_call("apply", {"policy_name": bad, "lb": "lab", "url": "http://x", + "out": str(tmp_path)})], enable_writes=True) + res = out[0]["result"] + assert res["isError"], bad + assert "not a policy name" in res["content"][0]["text"], bad + + +def test_a_legitimate_generated_slug_is_accepted(tmp_path): + """The names `generate` actually emits must still work — the guard must not be so tight that it + rejects the real thing.""" + d = tmp_path / "policies" + d.mkdir() + name = "deny-jndi-log4shell-headers" + (d / f"service_policy.{name}.json").write_text(json.dumps({"spec": {"rule_list": {"rules": []}}})) + out = _drive([_call("apply", {"policy_name": name, "lb": "lab", "url": "http://x", + "out": str(tmp_path), "dry_run": True})], enable_writes=True) + txt = out[0]["result"]["content"][0]["text"] + assert "not a policy name" not in txt and "no generated artifact" not in txt + + +# ============================================================ protocol edge cases +@pytest.mark.parametrize("raw,want_id,want_kind", [ + # A falsy-but-valid id is the classic JSON-RPC trap: `msg.get("id")` cannot distinguish 0 or + # false from absent, so notification detection has to test membership, not truthiness. + ('{"jsonrpc":"2.0","id":0,"method":"ping"}', 0, "result"), + ('{"jsonrpc":"2.0","id":false,"method":"ping"}', False, "result"), + ('{"jsonrpc":"2.0","id":"abc","method":"ping"}', "abc", "result"), + ('{"jsonrpc":"2.0","id":null,"method":"ping"}', None, "result"), + # MCP 2025-06-18 has no JSON-RPC batching; an array is not a valid message. + ('[{"jsonrpc":"2.0","id":1,"method":"ping"}]', None, "error"), + # `params` absent entirely on a tools/call + ('{"jsonrpc":"2.0","id":2,"method":"tools/call"}', 2, "error"), +]) +def test_protocol_edge_cases(raw, want_id, want_kind): + out = io.StringIO() + mcp.serve(io.StringIO(raw + "\n"), out, swap_stdout=False) + (frame,) = [json.loads(ln) for ln in out.getvalue().splitlines()] + assert frame["id"] == want_id and want_kind in frame + + +def test_crlf_and_a_final_line_without_a_newline_still_parse(): + """A client on Windows, or one that does not terminate its last frame, must not desynchronise.""" + for raw in ('{"jsonrpc":"2.0","id":1,"method":"ping"}\r\n', + '{"jsonrpc":"2.0","id":1,"method":"ping"}'): + out = io.StringIO() + mcp.serve(io.StringIO(raw), out, swap_stdout=False) + assert json.loads(out.getvalue())["id"] == 1 + + +def test_a_blank_or_whitespace_only_line_is_skipped_not_answered(): + out = io.StringIO() + mcp.serve(io.StringIO('\n \n{"jsonrpc":"2.0","id":4,"method":"ping"}\n'), out, + swap_stdout=False) + assert len(out.getvalue().splitlines()) == 1 + + +def test_a_tools_call_with_no_arguments_key_uses_defaults(tmp_path): + out = io.StringIO() + mcp.serve(io.StringIO(json.dumps({"jsonrpc": "2.0", "id": 3, "method": "tools/call", + "params": {"name": "ledger"}}) + "\n"), + out, swap_stdout=False) + assert "result" in json.loads(out.getvalue()) + + +def test_a_malformed_params_object_never_kills_the_server(): + """Found by adversarial review. JSON-RPC permits positional (array) `params`, and a list is + truthy — so `params.get(...)` raised straight out of `serve()`'s loop, terminating the process + with ZERO frames written. The client waits forever on a request that will never be answered and + every later request is lost with it. Dying silently is the worst available failure for a + transport. A non-string tool name did the same via an unhashable dict lookup.""" + for raw in ('{"jsonrpc":"2.0","id":1,"method":"tools/call","params":[{"name":"impact"}]}', + '{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":["impact"]}}', + '{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":{}}}', + '{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":7}}'): + out = io.StringIO() + mcp.serve(io.StringIO(raw + '\n{"jsonrpc":"2.0","id":2,"method":"ping"}\n'), out, + swap_stdout=False) + fr = [json.loads(ln) for ln in out.getvalue().splitlines()] + assert [f.get("id") for f in fr] == [1, 2], raw # answered, and still serving + assert fr[0]["error"]["code"] == mcp.INVALID_PARAMS, raw + + +def test_an_unexpected_error_inside_handle_is_answered_not_fatal(monkeypatch): + """The structural net, one level above the per-tool guard: it covers bugs not yet written. The + loop must never end without answering the request in flight.""" + class Broken(mcp.Tool): + def definition(self): + raise RuntimeError("definition blew up") + + monkeypatch.setattr(mcp, "build_tools", lambda **kw: [ + Broken("broken", "B", "d", {"type": "object", "properties": {}}, lambda **k: {})]) + out = io.StringIO() + mcp.serve(io.StringIO('{"jsonrpc":"2.0","id":1,"method":"tools/list"}\n' + '{"jsonrpc":"2.0","id":2,"method":"ping"}\n'), out, swap_stdout=False) + fr = [json.loads(ln) for ln in out.getvalue().splitlines()] + assert fr[0]["error"]["code"] == mcp.INTERNAL_ERROR + assert "definition blew up" in fr[0]["error"]["message"] + assert fr[1] == {"jsonrpc": "2.0", "id": 2, "result": {}} # kept serving + + +def test_every_documented_property_is_a_real_parameter(): + """This is what makes an argument-binding TypeError unreachable, which is in turn why + `tools/call` has no `except TypeError` pretending a bug inside a tool is a bad argument.""" + import inspect + for t in mcp.build_tools(enable_writes=True): + sig = inspect.signature(t.fn) + for arg in t.definition()["inputSchema"]["properties"]: + assert arg in sig.parameters, f"{t.name} documents {arg} but does not accept it" + + +def test_a_typeerror_inside_a_tool_is_a_tool_error_not_invalid_arguments(monkeypatch): + def boom(**kw): + return None + 1 # a genuine bug inside the tool + + monkeypatch.setattr(mcp, "build_tools", lambda **kw: [ + mcp.Tool("boom", "B", "raises TypeError", {"type": "object", "properties": {}}, boom)]) + out = _drive([_call("boom")]) + assert "error" not in out[0] # not a protocol error… + assert out[0]["result"]["isError"] is True # …a tool-execution error + assert "TypeError" in out[0]["result"]["content"][0]["text"] + + +# ============================================================ review findings, pinned +def test_a_notification_is_never_answered_whatever_its_method(): + """Found by adversarial review. The notification check was per-branch, so a request method + arriving WITHOUT an id — `{"method":"initialize"}` — was answered with an unsolicited `id: null` + frame. A client not expecting a response desynchronises on the next one it reads.""" + for method in ("initialize", "tools/list", "tools/call", "ping", "notifications/initialized"): + out = io.StringIO() + mcp.serve(io.StringIO(json.dumps({"jsonrpc": "2.0", "method": method}) + "\n"), out, + swap_stdout=False) + assert out.getvalue() == "", f"{method} answered a notification" + + +def test_scan_status_never_advances_the_cursor_past_what_it_returned(monkeypatch, tmp_path): + """Found by adversarial review, proven there with 565 of 20000 polls losing 27622 lines. Reading + `job["log"]` twice — slicing it, then taking its length for `log_next` — let the worker append + between the two, so `log_next` counted lines the slice never returned and a client tailing with + `since=log_next` could never ask for them again.""" + import threading + import time + + from vpcopilot import pipeline + stop = threading.Event() + + def chatty(repo=None, **kw): + log = kw["log"] + for i in range(4000): + if stop.is_set(): + break + log(f"line {i}") + return {"ok": True} + + monkeypatch.setattr(pipeline, "run_pipeline", chatty) + srv = mcp.Server() + jid = mcp.handle(_call("scan_start", {"repo": str(tmp_path), "out": str(tmp_path)}), + srv)["result"]["structuredContent"]["job_id"] + seen, since, polls = 0, 0, 0 + try: + while polls < 3000: + st = mcp.handle(_call("scan_status", {"job_id": jid, "since": since}), + srv)["result"]["structuredContent"] + # the invariant: the cursor may only advance by what we were actually handed + assert st["log_next"] == st["log_from"] + len(st["log"]), st + seen += len(st["log"]) + since = st["log_next"] + polls += 1 + if st["state"] != "running" and since >= st["log_next"]: + break + time.sleep(0) + finally: + stop.set() + time.sleep(0.05) + + +def test_a_second_scan_into_the_same_run_dir_is_refused(monkeypatch, tmp_path): + """Two pipelines writing one run directory interleave findings.json, triage.json, policies/ and + the ledger, and the loser's half-written state is indistinguishable from a complete run.""" + import threading + import time + + from vpcopilot import pipeline + release = threading.Event() + monkeypatch.setattr(pipeline, "run_pipeline", + lambda *a, **k: (release.wait(5), {"ok": True})[1]) + srv = mcp.Server() + first = mcp.handle(_call("scan_start", {"repo": str(tmp_path), "out": str(tmp_path)}), + srv)["result"]["structuredContent"]["job_id"] + try: + second = mcp.handle(_call("scan_start", {"repo": str(tmp_path), "out": str(tmp_path)}), + srv)["result"] + assert second["isError"] + assert "already running" in second["content"][0]["text"] + assert first in second["content"][0]["text"] # names the job to wait for + # …and a DIFFERENT out dir is fine + other = tmp_path / "other" + other.mkdir() + ok = mcp.handle(_call("scan_start", {"repo": str(tmp_path), "out": str(other)}), + srv)["result"] + assert not ok["isError"] + finally: + release.set() + time.sleep(0.05) + + +def test_a_run_dir_that_does_not_exist_declines_rather_than_reporting_zero(tmp_path): + """`list_patches`, `ledger.load` and `impact` all answer a missing directory with zeros, which + reads as "no live band-aids" — the most reassuring possible answer to a typo.""" + missing = str(tmp_path / "never-scanned") + for name in ("patches_list", "ledger", "impact"): + out = _drive([_call(name, {"out": missing})]) + assert out[0]["result"]["isError"], name + assert "no run directory" in out[0]["result"]["content"][0]["text"], name + + +def test_scan_start_refuses_a_path_that_does_not_exist(tmp_path): + """`run_pipeline`'s own docstring calls a scan of a nonexistent path "the failure mode not to + extend": it completes and writes a durable summary saying nothing was found.""" + for args in ({"repo": str(tmp_path / "nope")}, + {"spec": str(tmp_path / "nope.yaml")}, + {"manifest": [str(tmp_path / "nope.txt")]}): + out = _drive([_call("scan_start", args)]) + assert out[0]["result"]["isError"], args + assert "does not exist" in out[0]["result"]["content"][0]["text"], args + + +def test_the_drift_description_does_not_promise_a_check_it_may_not_run(): + """"Agents reason, code acts": a description is the only thing an agent has to go on, so it must + not claim a guard that needs an argument the agent was not told to pass.""" + by = {t.name: t for t in mcp.build_tools()} + desc = by["drift"].definition()["description"] + assert "Pass `policy`" in desc and "does not run" in desc diff --git a/tests/test_simulate.py b/tests/test_simulate.py index 9bf84da..1fa42a8 100644 --- a/tests/test_simulate.py +++ b/tests/test_simulate.py @@ -256,3 +256,169 @@ def test_zero_evaluated_is_reported_as_no_evidence_not_as_safe(monkeypatch, fake _recs("/a"), lb="lab", url="http://lab.test", out_dir=str(tmp_path), log=lambda m: None) p = res.policies[0] assert p.evaluated == 0 and "zero evidence" in p.reason + + +# ================= the G2 gate lives in the module, not in one surface (K1) ================= +def _flagged(out, name="deny-wide", blocked=True): + (out / "simulation.json").write_text(json.dumps({"policies": [ + {"policy_name": name, "blocked_promotion": blocked, "block_rate": 0.9, "threshold": 0.01, + "reason": "would block 180/200 recorded requests (90.0%), over the 1.0% threshold"}]})) + + +def _artifact(out, name="deny-wide"): + d = out / "policies" + d.mkdir(parents=True, exist_ok=True) + p = d / f"service_policy.{name}.json" + p.write_text(json.dumps({"metadata": {"name": name}, "spec": {"rule_list": {"rules": [ + {"spec": {"action": "DENY", "path": {"prefix_values": ["/"]}}}]}}})) + return str(p) + + +def test_the_gate_refuses_without_the_override(tmp_path): + _flagged(tmp_path) + with pytest.raises(RuntimeError, match="allow overbroad"): + simulate.promotion_gate(str(tmp_path), "deny-wide", log=lambda m: None) + + +def test_the_gate_with_the_override_proceeds_and_audits(tmp_path): + """Warn with an audited override, never a machine veto — the G2/I2 precedent. The record has to + carry the numbers a human would want to see justified.""" + from vpcopilot import audit + _flagged(tmp_path) + simulate.promotion_gate(str(tmp_path), "deny-wide", allow_overbroad=True, finding_id="f-1", + lb="lab", log=lambda m: None) + entries = [e for e in audit.load(str(tmp_path)) if e["action"] == "simulate_override"] + assert len(entries) == 1 + e = entries[0] + assert e["policy"] == "deny-wide" and e["finding_id"] == "f-1" and e["lb"] == "lab" + assert e["block_rate"] == 0.9 and e["threshold"] == 0.01 and e["reason"] + assert e["actor"] and e["run_id"] is not None # identity is stamped centrally by audit.py + + +def test_the_gate_is_silent_when_nothing_was_simulated(tmp_path): + """G2 adds a check, not a prerequisite: an operator who never runs `simulate` sees exactly the + behaviour they saw before it existed.""" + simulate.promotion_gate(str(tmp_path), "anything", log=lambda m: None) + _flagged(tmp_path, blocked=False) + simulate.promotion_gate(str(tmp_path), "deny-wide", log=lambda m: None) + + +def test_the_gate_is_silent_on_a_dry_run_and_without_a_policy_name(tmp_path): + _flagged(tmp_path) + simulate.promotion_gate(str(tmp_path), "deny-wide", dry_run=True, log=lambda m: None) + simulate.promotion_gate(str(tmp_path), None, log=lambda m: None) + + +def test_apply_from_scan_now_gets_the_gate_the_console_used_to_own(monkeypatch, fake_xc, tmp_path): + """The regression this fixes: `promotion_block` had exactly ONE production caller, + `console/app.py`, so `vpcopilot apply --from-scan` would attach an over-broad policy the console + refuses — and the `--allow-overbroad` flag ROADMAP.md:143 described did not exist. Same shape as + I1's `--force-probe`, whose guard lived only in the CLI. It raises before any XC call that + mutates, so nothing is created and nothing is attached.""" + from vpcopilot import apply as A + _flagged(tmp_path) + art = _artifact(tmp_path) + monkeypatch.setattr(A, "XC", lambda *a, **k: fake_xc) + with pytest.raises(RuntimeError, match="allow overbroad"): + A.apply_from_scan(art, "lab", "http://lab.test", out_dir=str(tmp_path), log=lambda m: None) + assert not fake_xc.service_policies # no policy object left behind + assert not fake_xc.put_lb_calls # nothing attached + + +def test_apply_from_scan_with_the_override_passes_the_gate(monkeypatch, fake_xc, tmp_path): + """The other direction: with the override the gate lets it through, so this is a warn and not a + veto. It fails later for want of a real tenant — reaching that point is the assertion.""" + from vpcopilot import apply as A + _flagged(tmp_path) + art = _artifact(tmp_path) + monkeypatch.setattr(A, "XC", lambda *a, **k: fake_xc) + try: + A.apply_from_scan(art, "lab", "http://lab.test", create_only=True, allow_overbroad=True, + out_dir=str(tmp_path), log=lambda m: None) + except RuntimeError as e: # anything but the gate + assert "allow overbroad" not in str(e) + from vpcopilot import audit + assert any(e["action"] == "simulate_override" for e in audit.load(str(tmp_path))) + + +def test_a_policy_with_no_simulation_applies_exactly_as_before(monkeypatch, fake_xc, tmp_path): + """No regression: with no simulation.json the new gate contributes nothing at all.""" + from vpcopilot import apply as A + art = _artifact(tmp_path, "unsimulated") + monkeypatch.setattr(A, "XC", lambda *a, **k: fake_xc) + res = A.apply_from_scan(art, "lab", "http://lab.test", create_only=True, + out_dir=str(tmp_path), log=lambda m: None) + assert res["mode"] == "create_only" + from vpcopilot import audit + assert not any(e["action"] == "simulate_override" for e in audit.load(str(tmp_path))) + + +def test_create_only_no_longer_skips_the_protected_lb_guard(monkeypatch, fake_xc, tmp_path): + """`create_only` returned before `apply_service_policy`, the only caller of `guard_lb`, so + `apply_from_scan(lb="nimbus-www", create_only=True)` wrote a policy object into the tenant with + neither that check nor the drift preflight. It attaches nothing, so no traffic changed — but a + persistent write against a protected target should not be the one path that skips the guard.""" + from vpcopilot import apply as A + art = _artifact(tmp_path, "p") + monkeypatch.setattr(A, "XC", lambda *a, **k: fake_xc) + monkeypatch.setenv("VPCOPILOT_PROTECTED_LBS", "nimbus-www") + with pytest.raises(RuntimeError, match="protected LB"): + A.apply_from_scan(art, "nimbus-www", "http://x.test", create_only=True, + out_dir=str(tmp_path), log=lambda m: None) + assert not fake_xc.service_policies + # …and the escape hatch still works, exactly as it does for every other apply path + A.apply_from_scan(art, "nimbus-www", "http://x.test", create_only=True, allow_protected=True, + out_dir=str(tmp_path), log=lambda m: None) + + +def test_the_cli_exposes_the_override_flag(): + """ROADMAP.md:143 described `--allow-overbroad` before it existed. Both surfaces now have it.""" + import inspect + + from vpcopilot import cli + assert "allow_overbroad" in inspect.signature(cli.apply).parameters + from vpcopilot.console.app import ActionReq + assert "allow_overbroad" in ActionReq.model_fields + + +def test_a_narrower_replay_does_not_erase_an_earlier_policys_flag(tmp_path): + """Found by adversarial review. `simulate --policy B` filters candidates to B, and + `write_result` overwrote the artifact wholesale — so a policy A an earlier run had flagged + `blocked_promotion` lost its flag and `promotion_block` went quiet for it. An operator who + simulated everything, saw A flagged, then re-simulated only B would find A applying with no + warning: a guard erased as a side effect of measuring something else.""" + from vpcopilot.schemas import PolicySimulation, SimulationResult + wide = SimulationResult(ts="2026-07-29T10:00:00Z", lb="lab", records=200, policies=[ + PolicySimulation(policy_name="A", blocked_promotion=True, block_rate=0.9, threshold=0.01, + reason="too broad"), + PolicySimulation(policy_name="B", blocked_promotion=False)]) + simulate.write_result(str(tmp_path), wide) + assert simulate.promotion_block(str(tmp_path), "A") # flagged + + narrow = SimulationResult(ts="2026-07-29T12:00:00Z", lb="lab", records=50, policies=[ + PolicySimulation(policy_name="B", blocked_promotion=False)]) + simulate.write_result(str(tmp_path), narrow) + + still = simulate.promotion_block(str(tmp_path), "A") + assert still, "A's blast-radius flag was erased by a replay that never measured A" + assert still["carried_from"] == "2026-07-29T10:00:00Z" # and it is not passed off as fresh + doc = json.loads((tmp_path / "simulation.json").read_text()) + assert doc["ts"] == "2026-07-29T12:00:00Z" # the head describes THIS run + assert any("carried forward" in c for c in doc["caveats"]) + b = next(p for p in doc["policies"] if p["policy_name"] == "B") + assert not b["carried_from"] # measured now, so not stamped + + +def test_a_full_replay_still_replaces_everything_it_measured(tmp_path): + """No regression: a run that measures a policy overwrites its old entry rather than keeping both, + and a first run is byte-identical to before.""" + from vpcopilot.schemas import PolicySimulation, SimulationResult + simulate.write_result(str(tmp_path), SimulationResult(ts="t1", policies=[ + PolicySimulation(policy_name="A", blocked_promotion=True, block_rate=0.9, threshold=0.01, + reason="too broad")])) + simulate.write_result(str(tmp_path), SimulationResult(ts="t2", policies=[ + PolicySimulation(policy_name="A", blocked_promotion=False)])) + doc = json.loads((tmp_path / "simulation.json").read_text()) + assert len(doc["policies"]) == 1 and doc["policies"][0]["blocked_promotion"] is False + assert simulate.promotion_block(str(tmp_path), "A") is None + assert not doc.get("caveats")