From 31483987e9d71d9224dd51c69baee36d4f29be43 Mon Sep 17 00:00:00 2001 From: ernestprovo23 Date: Sun, 23 Aug 2026 19:55:11 -0400 Subject: [PATCH 1/4] =?UTF-8?q?feat:=20agent=20gates=20=E2=80=94=20deploy-?= =?UTF-8?q?gate=20+=20auth=20audit=20(DSE-1257,=20DSE-1258)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two fail-closed CI gates extending warden past MCP-surface integrity into the adjacent controls agent deployments lack. deploy-gate (DSE-1257) verifies a deploy's evidence against a declared gate policy: required eval suites met their thresholds, required guardrails are active, a budget/quota is declared, and a human-approval receipt is present when required. The gate adjudicates evidence rather than running evals, which keeps verdicts reproducible from two JSON files, free of any eval framework's dependency tree, and makes missing or malformed evidence an unambiguous failure instead of a silent skip. Nine WRD-GATE-* rules. auth audit (DSE-1258) audits MCP client/server config for remote endpoints declaring no authentication, cleartext http:// transport, and credential literals committed into config, reusing the existing vendor secret patterns. Static only — no server spawn, no DNS, no network — which keeps it safe in CI and immune to churn in the MCP auth spec. Runtime capability brokering stays out of scope (DSE-725). Deliberately does not flag loopback servers, ${VAR} secret references, or local stdio servers; credential literals are redacted in findings, snippets, and SARIF. Both reuse the check exit-code contract (0 clean / 1 finding / 2 fail closed) and the shared SARIF + JSONL emitters, so an existing code-scanning pipeline needs no new plumbing. Verified on the build server under CI conditions (dev+sigstore extras, COVERAGE_PROCESS_START): 834 passed, coverage 86.14% (floor 80), ruff clean. All four new modules at 100% coverage. --- CHANGELOG.md | 24 ++++ DOCUMENTATION_INDEX.md | 17 +++ README.md | 2 + SYSTEM_CONTEXT_DIAGRAM.md | 11 ++ docs/AGENT_GATES.md | 135 ++++++++++++++++++++ src/mcp_warden/auth_audit.py | 203 ++++++++++++++++++++++++++++++ src/mcp_warden/cli.py | 4 + src/mcp_warden/cli_auth.py | 76 +++++++++++ src/mcp_warden/cli_deploy_gate.py | 73 +++++++++++ src/mcp_warden/deploy_gate.py | 158 +++++++++++++++++++++++ tests/test_auth_audit.py | 155 +++++++++++++++++++++++ tests/test_deploy_gate.py | 179 ++++++++++++++++++++++++++ 12 files changed, 1037 insertions(+) create mode 100644 docs/AGENT_GATES.md create mode 100644 src/mcp_warden/auth_audit.py create mode 100644 src/mcp_warden/cli_auth.py create mode 100644 src/mcp_warden/cli_deploy_gate.py create mode 100644 src/mcp_warden/deploy_gate.py create mode 100644 tests/test_auth_audit.py create mode 100644 tests/test_deploy_gate.py diff --git a/CHANGELOG.md b/CHANGELOG.md index f49622c..757a82c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,30 @@ Streamable HTTP; the v0.3 `guard` proxy adds deterministic runtime *result* insp ## [Unreleased] +### Added + +- **`deploy-gate` — fail-closed CI gate for agent deployments (DSE-1257).** Verifies a + deploy's evidence against a declared gate policy: required eval suites met their + thresholds, required guardrails are active, a budget/quota is declared, and a human + approval receipt is present when required. The gate **adjudicates evidence rather than + running evals** — keeping verdicts reproducible from two JSON files, free of any eval + framework's dependency tree, and making missing/malformed evidence an unambiguous + failure instead of a silent skip. Nine `WRD-GATE-*` rules; exit `0` only when every + control is satisfied, `1` on any finding, `2` on unreadable input (fail closed). + See [`docs/AGENT_GATES.md`](docs/AGENT_GATES.md). +- **`auth audit` — static MCP auth-posture audit (DSE-1258).** Audits MCP client/server + config for remote endpoints declaring no authentication, cleartext `http://` transport, + and credential literals committed into config (`WRD-AUTH-*`), reusing the existing + vendor secret patterns from `check`. **Static only** — no server spawn, no DNS, no + network — which keeps it safe to run against any config in CI and immune to churn in + the MCP auth specification. Deliberately does not flag loopback servers, `${VAR}` + secret references, or local stdio servers; every credential literal is redacted in + findings, snippets, and SARIF. Runtime capability brokering stays out of scope + (DSE-725). See [`docs/AGENT_GATES.md`](docs/AGENT_GATES.md). + +Both commands reuse the `check` exit-code contract and the shared SARIF/JSONL emitters, +so an existing code-scanning pipeline needs no changes. + ## [1.1.0] — 2026-07-14 ### Added diff --git a/DOCUMENTATION_INDEX.md b/DOCUMENTATION_INDEX.md index a02bd86..051e97f 100644 --- a/DOCUMENTATION_INDEX.md +++ b/DOCUMENTATION_INDEX.md @@ -14,6 +14,23 @@ describe and visualize the implementation that satisfies that contract. | 2 | [`SYSTEM_CONTEXT_DIAGRAM.md`](SYSTEM_CONTEXT_DIAGRAM.md) | System context + pin/check sequence (mermaid); trust boundary; `conclave` as dev-time reviewer only; composite GitHub Action + **pre-commit hook** as consumer delivery vehicles | | 3 | [`DOCUMENTATION_INDEX.md`](DOCUMENTATION_INDEX.md) | This file | +## Agent gates (`deploy-gate` + `auth audit` — DSE-1257 / DSE-1258) + +Two fail-closed gates extending warden past MCP-surface integrity into the adjacent +CI controls agent deployments lack: did the deploy meet its declared safety bar, and +is the MCP auth posture sound. Both reuse the `check` exit-code contract (0 clean / +1 finding / 2 fail-closed) and the shared SARIF + JSONL emitters. + +| Artifact | Purpose | +|----------|---------| +| [`docs/AGENT_GATES.md`](docs/AGENT_GATES.md) | Security contract for both gates — policy/evidence schemas, full rule tables, scope honesty, and the two load-bearing design decisions (evidence-adjudication, static-only) | +| [`src/mcp_warden/deploy_gate.py`](src/mcp_warden/deploy_gate.py) | `WRD-GATE-*` engine: eval thresholds, guardrail presence, budget, approval receipt | +| [`src/mcp_warden/cli_deploy_gate.py`](src/mcp_warden/cli_deploy_gate.py) | `deploy-gate` command body (register idiom) | +| [`src/mcp_warden/auth_audit.py`](src/mcp_warden/auth_audit.py) | `WRD-AUTH-*` static config audit; reuses `checks_secret.scan_field` for vendor patterns | +| [`src/mcp_warden/cli_auth.py`](src/mcp_warden/cli_auth.py) | `auth audit` sub-app command body | +| [`tests/test_deploy_gate.py`](tests/test_deploy_gate.py) | Engine per-control pass/fail + CLI exit codes + fail-closed on malformed evidence | +| [`tests/test_auth_audit.py`](tests/test_auth_audit.py) | Every rule, the deliberate non-flags (loopback, `${VAR}` refs, stdio), redaction, CLI/JSON/SARIF | + ## GitHub Action (`action.yml` — Issue #18) The composite reusable action is the primary delivery vehicle for the `check` gate. diff --git a/README.md b/README.md index 3812f2e..3250b27 100644 --- a/README.md +++ b/README.md @@ -391,6 +391,8 @@ run the gate only on push: | `mcp-warden inspect [--lock F] [--sarif F]` | **(v0.2)** Offline analyzer over a recorded JSON-RPC session — same `WRD-RES-*` catalog as `guard` (always report-only) | non-zero on any BLOCK-tier finding; 2 on read error | | `mcp-warden lock rotate [--approver ID] [--actor ID] [--note T] [--json]` | **(v0.3)** Re-attest provenance on an existing baseline without re-capturing the surface; `overall_digest` stays **byte-identical** (WARDEN_LOCK_SCHEMA §8.2). Fails closed on a tampered/inconsistent lock | 0 on success, 2 on missing/invalid/tampered lock | | `mcp-warden diff [--json] [--sarif F] [--no-provenance] [--exit-code]` | **(v0.3)** Offline, **redacted** viewer over the drift engine: renders integrity drift between two existing locks (A=baseline, B=current) + a separate informational provenance section. Never re-captures and never prints raw `server.command`/`args` (secret-safe) | 0 (viewer); with `--exit-code`, 1 on **integrity** drift only; 2 on missing/invalid lock | +| `mcp-warden deploy-gate --policy F --evidence F [--json] [--sarif F]` | **(v1.2)** Fail-closed CI gate for agent deployments: verifies declared eval thresholds, required guardrails, a budget/quota, and a human-approval receipt. Adjudicates evidence — it does **not** run evals. See [`docs/AGENT_GATES.md`](docs/AGENT_GATES.md) | 0 only when every control is satisfied; 1 on any gate finding; 2 on unreadable/malformed input (fail closed) | +| `mcp-warden auth audit [--json] [--sarif F]` | **(v1.2)** Static MCP auth-posture audit over client/server config: remote endpoints without auth, cleartext `http://`, credential literals committed into config. No server spawn, no network. See [`docs/AGENT_GATES.md`](docs/AGENT_GATES.md) | 0 clean; 1 on any finding; 2 on read/parse error (fail closed) | | `mcp-warden-precommit [--lock F] [--timeout N] [--strict] -- ` | **(v0.3)** pre-commit hook entry point (see [pre-commit hook](#pre-commit-hook--the-local-pre-ci-gate)). Runs the same check verdict path; check-only (never pins, never writes the lock) | 0 clean / **1 drift** / 2 config error; server-unavailable → 0+warning (non-strict) or 2 (`--strict`) | For stdio, `` is passed to the OS as an **argv array, never through a diff --git a/SYSTEM_CONTEXT_DIAGRAM.md b/SYSTEM_CONTEXT_DIAGRAM.md index a34aec8..9440739 100644 --- a/SYSTEM_CONTEXT_DIAGRAM.md +++ b/SYSTEM_CONTEXT_DIAGRAM.md @@ -32,6 +32,17 @@ logic) plus a separate informational provenance section. It never prints raw > is a **dev-time design reviewer** that shaped this contract. It is **NOT** a > runtime dependency and is never invoked by `pin`/`check`/`policy`. +> **Agent gates (DSE-1257 / DSE-1258)** add two CI-only verbs that sit alongside `check` +> and never touch a running server. `deploy-gate` reads two JSON documents (a gate policy +> and pipeline-produced deploy evidence) and fail-closes a deploy whose declared eval +> thresholds, guardrails, budget, or human-approval receipt are unmet — it adjudicates +> evidence and deliberately does **not** execute evals. `auth audit` reads MCP client +> config files and flags weak auth posture statically: no server spawn, no DNS, no +> network, which is what keeps it immune to MCP auth-spec churn. Both reuse the `check` +> exit-code contract (0/1/2, fail closed) and the shared SARIF + JSONL emitters, so they +> enter an existing code-scanning pipeline with no new plumbing. Runtime capability +> brokering remains out of scope (DSE-725). See [`docs/AGENT_GATES.md`](docs/AGENT_GATES.md). +> > **`action.yml` (Issue #18)** is the primary consumer delivery vehicle for the `check` > gate. Consumers pin `DataScience-EngineeringExperts/mcp-warden@` in their workflow; the composite > action wraps the C2 sequence (steps 1–5 of the pin/check sequence above) behind a diff --git a/docs/AGENT_GATES.md b/docs/AGENT_GATES.md new file mode 100644 index 0000000..8847fb7 --- /dev/null +++ b/docs/AGENT_GATES.md @@ -0,0 +1,135 @@ +# Agent Gates — `deploy-gate` and `auth audit` + +Two fail-closed gates that extend mcp-warden past MCP-surface integrity into the +two adjacent controls that agent deployments actually lack in CI: **did the +deploy meet its declared safety bar** (`deploy-gate`, DSE-1257) and **is the MCP +auth posture sound** (`auth audit`, DSE-1258). + +Both follow the same contract as `check`: + +| Exit | Meaning | +|------|---------| +| `0` | Every declared control satisfied | +| `1` | At least one finding — the gate blocks | +| `2` | Unreadable/malformed input — **fail closed**, never a pass | + +Both emit `--json` (JSONL findings) and `--sarif` (code-scanning upload), reusing +the same emitters as `check`, so an existing SARIF pipeline needs no changes. + +--- + +## 1. `deploy-gate` — release control for agent deployments + +**The gap.** Agent frameworks ship evals, guardrails, and budgets as libraries, +but nothing *blocks a deploy* when the evals regress or a guardrail is switched +off. Teams write bespoke shell in CI, or skip the check. + +**The design decision that matters: the gate does not run evals.** It verifies +*evidence* that they ran and passed. Running evals is the pipeline's job and is +framework-specific; adjudicating them is a deterministic, portable control. This +keeps the gate free of every eval framework's dependency tree, makes verdicts +reproducible from two JSON files, and means a missing or malformed evidence file +is unambiguously a **failure** rather than a silent skip. + +### Policy schema + +```json +{ + "required_evals": [{ "suite": "safety", "min_score": 0.9 }], + "required_guardrails": ["prompt-injection", "pii-redaction"], + "require_budget": true, + "require_approval": true +} +``` + +### Evidence schema + +Produced by the deploy pipeline: + +```json +{ + "evals": { "safety": { "score": 0.95 } }, + "guardrails": ["prompt-injection", "pii-redaction"], + "budget": { "limit": 100 }, + "approval": { "approved": true, "approver": "release-manager@example.com" } +} +``` + +### Rules + +| Rule ID | Severity | Fires when | +|---------|----------|-----------| +| `WRD-GATE-EVAL-MISSING` | high | A required suite has no result in evidence | +| `WRD-GATE-EVAL-MALFORMED` | high | A suite reported no numeric score | +| `WRD-GATE-EVAL-THRESHOLD` | high | A suite scored below its `min_score` | +| `WRD-GATE-EVAL-EVIDENCE` | high | The `evals` block is not an object | +| `WRD-GATE-GUARDRAIL-MISSING` | high | A required guardrail is not active | +| `WRD-GATE-BUDGET-MISSING` | medium | `require_budget` set, no budget declared | +| `WRD-GATE-BUDGET-INVALID` | medium | Budget has no positive limit | +| `WRD-GATE-APPROVAL-MISSING` | critical | `require_approval` set, no receipt | +| `WRD-GATE-APPROVAL-INVALID` | critical | Receipt is not affirmative and attributed | + +### Usage + +```bash +mcp-warden deploy-gate --policy gate-policy.json --evidence deploy-evidence.json +``` + +```yaml +- name: Agent deploy gate + run: mcp-warden deploy-gate --policy gate-policy.json --evidence evidence.json --sarif gate.sarif +``` + +### Scope honesty + +`deploy-gate` adjudicates **declared evidence**. It does not verify that the +evidence is truthful — a pipeline that fabricates a score passes. Bind evidence +to a trusted producer (signed CI artifact, restricted branch) when that matters. +It is a release control, not an attestation system; the signed-decision path +lives in [`POLICY_ENFORCEMENT.md`](POLICY_ENFORCEMENT.md). + +--- + +## 2. `auth audit` — static MCP auth-posture audit + +**The gap.** MCP configs routinely point at remote endpoints with no +authentication, over cleartext `http://`, with bearer tokens pasted directly +into the committed config. None of that requires exploitation to find — it is +declared in the file. + +**The design decision that matters: static only.** No server is spawned, no DNS +is resolved, no network is touched. The audit reasons purely about what the +config declares. That makes it safe to run against any config in CI, and — this +is the load-bearing part — it keeps the feature **immune to churn in the MCP +auth specification**. Runtime capability brokering is deliberately out of scope +and tracked separately (DSE-725). + +### Rules + +| Rule ID | Severity | Fires when | +|---------|----------|-----------| +| `WRD-AUTH-NOAUTH` | medium | A remote endpoint declares no auth material | +| `WRD-AUTH-PLAINTEXT-HTTP` | high | A remote endpoint uses `http://` | +| `WRD-AUTH-TOKEN-IN-CONFIG` | high | An auth-bearing key holds a literal credential | +| `WRD-SEC-*` | varies | Vendor secret patterns found in any config value (shared with `check`) | + +### What it deliberately does not flag + +Precision matters more than recall for a gate that blocks CI: + +- **Loopback servers** (`localhost`, `127.0.0.1`, `::1`) — not remotely + reachable, so missing auth is not an exposure. +- **Secret references** — `${TOKEN}`, `$TOKEN`, `{{ secret }}` are the correct + pattern and are never flagged as literals. +- **Local stdio servers** — a `command`/`args` entry with no URL and no remote + transport has no auth posture to audit. + +### Usage + +```bash +mcp-warden auth audit ~/.claude/claude_desktop_config.json .mcp.json +mcp-warden auth audit .mcp.json --sarif auth.sarif +``` + +Every credential literal is redacted in findings, snippets, and SARIF output — +the audit never widens exposure of the thing it is reporting. diff --git a/src/mcp_warden/auth_audit.py b/src/mcp_warden/auth_audit.py new file mode 100644 index 0000000..e6a9d83 --- /dev/null +++ b/src/mcp_warden/auth_audit.py @@ -0,0 +1,203 @@ +"""Static MCP auth-posture audit (WRD-AUTH-*) — DSE-1258. + +Audits MCP client/server configuration files for weak authentication posture +without running anything: no server spawn, no DNS, no network. It parses the +declarative config (Claude Desktop ``claude_desktop_config.json``, ``.mcp.json``, +``mcp.json`` — all sharing the ``{"mcpServers": {...}}`` shape) and flags +remote servers reachable without auth, credential literals committed into +config, cleartext transport, and inline secrets that should reference a +manager instead. + +Deliberately static and conservative: it reasons only about what the config +declares. Runtime capability brokering is a separate concern (DSE-725); this +module stays in warden's fail-closed static lane so it is immune to the churn +in the MCP auth spec. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +from .checks_secret import scan_field +from .models import Finding + +#: Header keys whose values carry auth material worth scanning for literals. +_AUTH_HEADER_KEYS = {"authorization", "x-api-key", "api-key", "apikey", "token"} + +#: Substrings that mark ANY config key (env or header) as auth-bearing. +_AUTH_KEY_SUBSTRINGS = ("token", "api_key", "apikey", "secret", "auth", "password", "passwd") + + +def _is_auth_key(key: str) -> bool: + """Whether a config key name signals it holds authentication material.""" + k = key.lower() + return k in _AUTH_HEADER_KEYS or any(s in k for s in _AUTH_KEY_SUBSTRINGS) + +#: Hosts that are not remotely reachable, so a missing auth header is not an +#: exposure on its own. +_LOCAL_HOSTS = {"localhost", "127.0.0.1", "::1", "0.0.0.0"} + + +def _is_local_host(host: str) -> bool: + return host.split(":", 1)[0].lower() in _LOCAL_HOSTS + + +def _host_of(url: str) -> str: + """Extract the bare host[:port] from an http(s) URL without a full parse.""" + rest = url.split("://", 1)[-1] + return rest.split("/", 1)[0] + + +def _looks_like_secret_ref(value: str) -> bool: + """True when the value is an env/secret-manager reference, not a literal. + + ``${TOKEN}``, ``$TOKEN``, and ``{{ secret }}`` are references — the operator + is doing the right thing and we must not flag them. + """ + v = value.strip() + return ( + (v.startswith("${") and v.endswith("}")) + or (v.startswith("$") and v[1:].isidentifier()) + or (v.startswith("{{") and v.endswith("}}")) + ) + + +def _server_has_auth(server: dict[str, Any]) -> bool: + """Whether the server declares ANY authentication material.""" + headers = server.get("headers") or {} + if isinstance(headers, dict): + for key in headers: + if key.lower() in _AUTH_HEADER_KEYS: + return True + env = server.get("env") or {} + if isinstance(env, dict): + for key in env: + k = key.lower() + if "token" in k or "api_key" in k or "apikey" in k or "auth" in k or "secret" in k: + return True + return False + + +def _scan_mapping_for_literals(mapping: dict[str, Any], target: str) -> list[Finding]: + """Flag secret literals sitting directly in an env/headers mapping. + + An auth-bearing key whose value is a literal (not a ``${VAR}`` reference) + is a credential committed into config: high severity. Values are also run + through the vendor secret scanner so a stray ``sk-``/``ghp_``/AKIA literal + anywhere in the mapping is caught even under a non-obvious key. + """ + findings: list[Finding] = [] + if not isinstance(mapping, dict): + return findings + for key, value in mapping.items(): + if not isinstance(value, str) or not value: + continue + if _is_auth_key(key) and not _looks_like_secret_ref(value): + findings.append( + Finding( + rule_id="WRD-AUTH-TOKEN-IN-CONFIG", + severity="high", + target=target, + message=( + f"auth key '{key}' holds a literal credential in config; " + "reference a secret manager (${VAR}) instead" + ), + snippet=_redact(value), + ) + ) + # Value-level secret scan (already redacts). + findings.extend(scan_field(value, target)) + return findings + + +def _redact(value: str) -> str: + """Redact a credential literal to a short, non-recoverable hint.""" + v = value.strip() + if len(v) <= 8: + return "***" + return f"{v[:4]}...{v[-2:]}" + + +def audit_server(name: str, server: dict[str, Any]) -> list[Finding]: + """Audit one MCP server entry; return sorted WRD-AUTH-* findings.""" + findings: list[Finding] = [] + target = f"mcpServers/{name}" + + url = server.get("url") + transport = str(server.get("type") or server.get("transport") or "").lower() + is_remote = bool(url) or transport in {"http", "sse", "streamable-http", "websocket"} + + if isinstance(url, str) and url: + host = _host_of(url) + remote = not _is_local_host(host) + if url.lower().startswith("http://") and remote: + findings.append( + Finding( + rule_id="WRD-AUTH-PLAINTEXT-HTTP", + severity="high", + target=target, + message=f"remote MCP endpoint '{host}' uses cleartext http://; use https://", + snippet=url, + ) + ) + if remote and not _server_has_auth(server): + findings.append( + Finding( + rule_id="WRD-AUTH-NOAUTH", + severity="medium", + target=target, + message=( + f"remote MCP endpoint '{host}' declares no authentication " + "(no Authorization/token header or env)" + ), + snippet=host, + ) + ) + elif is_remote and not _server_has_auth(server): + findings.append( + Finding( + rule_id="WRD-AUTH-NOAUTH", + severity="medium", + target=target, + message=f"{transport or 'remote'} MCP server declares no authentication material", + snippet=name, + ) + ) + + findings.extend(_scan_mapping_for_literals(server.get("headers") or {}, target)) + findings.extend(_scan_mapping_for_literals(server.get("env") or {}, target)) + + return sorted(findings, key=lambda f: (f.target, f.rule_id, f.snippet)) + + +def audit_config(doc: dict[str, Any]) -> list[Finding]: + """Audit a parsed MCP config document (``{"mcpServers": {...}}``).""" + servers = doc.get("mcpServers") + if not isinstance(servers, dict): + return [] + findings: list[Finding] = [] + for name, server in servers.items(): + if isinstance(server, dict): + findings.extend(audit_server(str(name), server)) + return sorted(findings, key=lambda f: (f.target, f.rule_id, f.snippet)) + + +class AuthAuditError(ValueError): + """Raised on an unreadable or malformed config file (fail closed).""" + + +def audit_path(path: Path) -> list[Finding]: + """Read + audit one config file. Raises AuthAuditError on parse failure.""" + try: + raw = path.read_text(encoding="utf-8") + except OSError as exc: + raise AuthAuditError(f"cannot read {path}: {exc}") from exc + try: + doc = json.loads(raw) + except json.JSONDecodeError as exc: + raise AuthAuditError(f"invalid JSON in {path}: {exc}") from exc + if not isinstance(doc, dict): + raise AuthAuditError(f"{path}: top-level config must be a JSON object") + return audit_config(doc) diff --git a/src/mcp_warden/cli.py b/src/mcp_warden/cli.py index 9b1a78e..92f8bfd 100644 --- a/src/mcp_warden/cli.py +++ b/src/mcp_warden/cli.py @@ -28,6 +28,8 @@ from .capture import CaptureError, capture_surface_http_sync, capture_surface_sync from .check_core import run_check_full from .checks import run_checks +from .cli_auth import register as register_auth_commands +from .cli_deploy_gate import register as register_deploy_gate_command from .cli_diff import register as register_diff_command from .cli_guard import register as register_guard_commands from .cli_lock import register as register_lock_commands @@ -82,6 +84,8 @@ def _root( register_guard_commands(app, console, err_console) register_lock_commands(app, console, err_console) register_diff_command(app, console, err_console) +register_auth_commands(app, console, err_console) +register_deploy_gate_command(app, console, err_console) def _split_server_cmd(server_cmd: list[str]) -> tuple[str, list[str]]: diff --git a/src/mcp_warden/cli_auth.py b/src/mcp_warden/cli_auth.py new file mode 100644 index 0000000..84b7f62 --- /dev/null +++ b/src/mcp_warden/cli_auth.py @@ -0,0 +1,76 @@ +"""CLI command body for ``auth audit`` (WRD-AUTH-*) — DSE-1258. + +Split from ``cli.py`` to keep each module under the LOC budget. +``register(app, console, err_console)`` attaches an ``auth`` sub-app with a +single ``audit`` command, matching the ``policy`` sub-app idiom. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Optional + +import typer +from rich.console import Console +from rich.table import Table + +from .auth_audit import AuthAuditError, audit_path +from .emitters import build_sarif, findings_to_jsonl, sarif_to_json +from .models import Finding + + +def _print_summary(console: Console, findings: list[Finding], scanned: int) -> None: + if not findings: + console.print(f"[green]auth audit clean[/green] ({scanned} config file(s), no findings)") + return + table = Table(title=f"MCP auth-posture findings ({scanned} config file(s))") + table.add_column("severity", no_wrap=True) + table.add_column("rule", no_wrap=True) + table.add_column("target") + table.add_column("message") + for f in findings: + color = {"critical": "red", "high": "red", "medium": "yellow"}.get(f.severity, "white") + table.add_row(f"[{color}]{f.severity}[/{color}]", f.rule_id, f.target, f.message) + console.print(table) + + +def register(app: typer.Typer, console: Console, err_console: Console) -> None: + """Attach the ``auth audit`` command tree to ``app``.""" + auth_app = typer.Typer(add_completion=False, help="Static MCP auth-posture audit.") + app.add_typer(auth_app, name="auth") + + @auth_app.command("audit") + def audit( + configs: list[Path] = typer.Argument( + ..., help="MCP config file(s) to audit (claude_desktop_config.json / .mcp.json / mcp.json)" + ), + json_out: bool = typer.Option(False, "--json", help="Emit findings as JSONL to stdout"), + sarif: Optional[Path] = typer.Option(None, "--sarif", help="Write a SARIF report to this path"), + ) -> None: + """Audit MCP client/server config for weak auth posture; fail closed. + + Static only: no server is spawned, no network is touched. Flags remote + endpoints reachable without auth, credential literals in config, + cleartext http:// transport, and inline secrets that should reference a + secret manager. Exits 1 on any finding, 2 on a read/parse error. + """ + all_findings: list[Finding] = [] + for path in configs: + try: + all_findings.extend(audit_path(path)) + except AuthAuditError as exc: + err_console.print(f"[red]error:[/red] {exc}") + raise typer.Exit(code=2) from exc + + all_findings.sort(key=lambda f: (f.target, f.rule_id, f.snippet)) + + if sarif is not None: + sarif.write_text(sarif_to_json(build_sarif(all_findings)), encoding="utf-8") + + if json_out: + console.print(findings_to_jsonl(all_findings), end="") + else: + _print_summary(console, all_findings, len(configs)) + + if all_findings: + raise typer.Exit(code=1) diff --git a/src/mcp_warden/cli_deploy_gate.py b/src/mcp_warden/cli_deploy_gate.py new file mode 100644 index 0000000..8070ee2 --- /dev/null +++ b/src/mcp_warden/cli_deploy_gate.py @@ -0,0 +1,73 @@ +"""CLI command body for ``deploy-gate`` (WRD-GATE-*) — DSE-1257. + +Split from ``cli.py`` to keep each module under the LOC budget. +``register(app, console, err_console)`` attaches the ``deploy-gate`` command. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Optional + +import typer +from rich.console import Console +from rich.table import Table + +from .deploy_gate import DeployGateError, GateOutcome, run_deploy_gate +from .emitters import build_sarif, findings_to_jsonl, sarif_to_json + + +def _print_summary(console: Console, outcome: GateOutcome) -> None: + if outcome.passed: + console.print( + f"[green]deploy gate PASS[/green] ({outcome.controls_checked} control(s) satisfied)" + ) + return + table = Table(title=f"Deploy gate FAILED ({outcome.controls_checked} control(s) checked)") + table.add_column("severity", no_wrap=True) + table.add_column("rule", no_wrap=True) + table.add_column("control") + table.add_column("reason") + for f in outcome.findings: + color = {"critical": "red", "high": "red", "medium": "yellow"}.get(f.severity, "white") + table.add_row(f"[{color}]{f.severity}[/{color}]", f.rule_id, f.target, f.message) + console.print(table) + + +def register(app: typer.Typer, console: Console, err_console: Console) -> None: + """Attach the ``deploy-gate`` command to ``app``.""" + + @app.command("deploy-gate") + def deploy_gate( + policy: Path = typer.Option(..., "--policy", help="Gate policy JSON (required controls)"), + evidence: Path = typer.Option( + ..., "--evidence", help="Deploy evidence JSON produced by the pipeline" + ), + json_out: bool = typer.Option(False, "--json", help="Emit findings as JSONL to stdout"), + sarif: Optional[Path] = typer.Option(None, "--sarif", help="Write a SARIF report to this path"), + ) -> None: + """Fail-closed CI gate for agent deployments ("release_control for agents"). + + Verifies a deploy's evidence against a declared gate policy: required + eval suites met their thresholds, required guardrails are active, a + budget/quota is declared, and a human-approval receipt is present when + required. Any unmet or missing/malformed control fails the gate. + Exits 0 only on a fully satisfied gate; 1 on any gate finding; 2 on a + read/parse error (fail closed). + """ + try: + outcome = run_deploy_gate(policy, evidence) + except DeployGateError as exc: + err_console.print(f"[red]error:[/red] {exc}") + raise typer.Exit(code=2) from exc + + if sarif is not None: + sarif.write_text(sarif_to_json(build_sarif(outcome.findings)), encoding="utf-8") + + if json_out: + console.print(findings_to_jsonl(outcome.findings), end="") + else: + _print_summary(console, outcome) + + if outcome.findings: + raise typer.Exit(code=1) diff --git a/src/mcp_warden/deploy_gate.py b/src/mcp_warden/deploy_gate.py new file mode 100644 index 0000000..0ee63c6 --- /dev/null +++ b/src/mcp_warden/deploy_gate.py @@ -0,0 +1,158 @@ +"""Fail-closed CI deploy gate for agent deployments (WRD-GATE-*) — DSE-1257. + +"release_control for agents": a deterministic gate that reads a declared gate +policy plus an evidence bundle produced by a deploy pipeline, and fail-closes +the deploy unless every required control is present and passing. It runs in CI +and exits non-zero on any unmet requirement. + +The gate is intentionally evidence-driven and deterministic. It does not run +evals itself — it verifies that declared eval suites ran, met their thresholds, +that required guardrails are present, that a budget/quota is declared, and that +a human-approval receipt is present when the policy requires one. Missing or +malformed evidence is a failure, never a pass (fail closed). + +Policy and evidence are JSON. See DEPLOY_GATE.md for the schema. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from .models import Finding + + +class DeployGateError(ValueError): + """Raised on unreadable/malformed policy or evidence (fail closed).""" + + +@dataclass(frozen=True) +class GateOutcome: + """Result of a deploy-gate evaluation.""" + + findings: list[Finding] + controls_checked: int + + @property + def passed(self) -> bool: + return not self.findings + + +def _load_json(path: Path, kind: str) -> dict[str, Any]: + try: + raw = path.read_text(encoding="utf-8") + except OSError as exc: + raise DeployGateError(f"cannot read {kind} {path}: {exc}") from exc + try: + doc = json.loads(raw) + except json.JSONDecodeError as exc: + raise DeployGateError(f"invalid JSON in {kind} {path}: {exc}") from exc + if not isinstance(doc, dict): + raise DeployGateError(f"{kind} {path}: top-level must be a JSON object") + return doc + + +def _fail(rule: str, severity: str, target: str, message: str, snippet: str = "") -> Finding: + return Finding(rule_id=rule, severity=severity, target=target, message=message, snippet=snippet) + + +def _check_evals(policy: dict[str, Any], evidence: dict[str, Any]) -> list[Finding]: + """Each required eval suite must be present and meet its min score.""" + findings: list[Finding] = [] + required = policy.get("required_evals") or [] + reported = evidence.get("evals") or {} + if not isinstance(reported, dict): + return [_fail("WRD-GATE-EVAL-EVIDENCE", "high", "evals", + "evidence 'evals' must be an object of suite -> {score}")] + for spec in required: + if not isinstance(spec, dict): + continue + name = str(spec.get("suite", "")) + threshold = spec.get("min_score") + target = f"evals/{name}" + result = reported.get(name) + if result is None: + findings.append(_fail("WRD-GATE-EVAL-MISSING", "high", target, + f"required eval suite '{name}' has no result in evidence")) + continue + score = result.get("score") if isinstance(result, dict) else None + if not isinstance(score, (int, float)): + findings.append(_fail("WRD-GATE-EVAL-MALFORMED", "high", target, + f"eval suite '{name}' reported no numeric score")) + continue + if isinstance(threshold, (int, float)) and score < threshold: + findings.append(_fail("WRD-GATE-EVAL-THRESHOLD", "high", target, + f"eval '{name}' scored {score} < required {threshold}", + snippet=f"{score}<{threshold}")) + return findings + + +def _check_guardrails(policy: dict[str, Any], evidence: dict[str, Any]) -> list[Finding]: + """Every required guardrail must be declared active in the evidence.""" + findings: list[Finding] = [] + required = policy.get("required_guardrails") or [] + active = evidence.get("guardrails") or [] + active_set = {str(g) for g in active} if isinstance(active, list) else set() + for name in required: + if str(name) not in active_set: + findings.append(_fail("WRD-GATE-GUARDRAIL-MISSING", "high", f"guardrails/{name}", + f"required guardrail '{name}' is not active in this deploy")) + return findings + + +def _check_budget(policy: dict[str, Any], evidence: dict[str, Any]) -> list[Finding]: + """When the policy requires a budget, evidence must declare a positive one.""" + if not policy.get("require_budget"): + return [] + budget = evidence.get("budget") + if not isinstance(budget, dict): + return [_fail("WRD-GATE-BUDGET-MISSING", "medium", "budget", + "policy requires a declared budget/quota; none present in evidence")] + limit = budget.get("limit") + if not isinstance(limit, (int, float)) or limit <= 0: + return [_fail("WRD-GATE-BUDGET-INVALID", "medium", "budget", + "declared budget has no positive limit", snippet=str(limit))] + return [] + + +def _check_approval(policy: dict[str, Any], evidence: dict[str, Any]) -> list[Finding]: + """When the policy requires human approval, a valid receipt must be present.""" + if not policy.get("require_approval"): + return [] + receipt = evidence.get("approval") + if not isinstance(receipt, dict): + return [_fail("WRD-GATE-APPROVAL-MISSING", "critical", "approval", + "policy requires human approval; no approval receipt in evidence")] + approver = receipt.get("approver") + approved = receipt.get("approved") + if approved is not True or not isinstance(approver, str) or not approver.strip(): + return [_fail("WRD-GATE-APPROVAL-INVALID", "critical", "approval", + "approval receipt is not an affirmative, attributed approval", + snippet=str(approver))] + return [] + + +def evaluate_gate(policy: dict[str, Any], evidence: dict[str, Any]) -> GateOutcome: + """Evaluate an agent deploy against the gate policy; fail closed on any gap.""" + findings: list[Finding] = [] + findings.extend(_check_evals(policy, evidence)) + findings.extend(_check_guardrails(policy, evidence)) + findings.extend(_check_budget(policy, evidence)) + findings.extend(_check_approval(policy, evidence)) + controls = ( + len(policy.get("required_evals") or []) + + len(policy.get("required_guardrails") or []) + + (1 if policy.get("require_budget") else 0) + + (1 if policy.get("require_approval") else 0) + ) + findings.sort(key=lambda f: (f.target, f.rule_id)) + return GateOutcome(findings=findings, controls_checked=controls) + + +def run_deploy_gate(policy_path: Path, evidence_path: Path) -> GateOutcome: + """Load policy + evidence from disk and evaluate the gate.""" + policy = _load_json(policy_path, "policy") + evidence = _load_json(evidence_path, "evidence") + return evaluate_gate(policy, evidence) diff --git a/tests/test_auth_audit.py b/tests/test_auth_audit.py new file mode 100644 index 0000000..db325e5 --- /dev/null +++ b/tests/test_auth_audit.py @@ -0,0 +1,155 @@ +"""Static MCP auth-posture audit tests (WRD-AUTH-*) — DSE-1258. + +Covers the pure ``audit_config``/``audit_server`` helpers (each rule, and the +clean paths that must NOT flag) plus the ``auth audit`` CLI (exit codes, +JSON/SARIF, fail-closed on malformed config). +""" + +from __future__ import annotations + +import json + +from typer.testing import CliRunner + +from mcp_warden.auth_audit import AuthAuditError, audit_config, audit_server +from mcp_warden.cli import app + +runner = CliRunner() + + +def _rules(findings): + return {f.rule_id for f in findings} + + +def test_remote_http_without_auth_flags_noauth_and_plaintext(): + findings = audit_server("remote", {"url": "http://mcp.example.com/sse"}) + rules = _rules(findings) + assert "WRD-AUTH-NOAUTH" in rules + assert "WRD-AUTH-PLAINTEXT-HTTP" in rules + + +def test_https_with_auth_header_is_clean(): + server = {"url": "https://mcp.example.com/sse", "headers": {"Authorization": "${MCP_TOKEN}"}} + assert audit_server("ok", server) == [] + + +def test_localhost_without_auth_is_not_flagged(): + # A stdio-style local server reachable only on loopback is not an exposure. + assert audit_server("local", {"url": "http://127.0.0.1:8080/mcp"}) == [] + + +def test_literal_token_in_headers_flagged_high(): + server = {"url": "https://mcp.example.com", "headers": {"Authorization": "Bearer sk-abcdefghijklmnopqrstuvwx"}} + findings = audit_server("lit", server) + assert "WRD-AUTH-TOKEN-IN-CONFIG" in _rules(findings) + # The credential literal must be redacted in every snippet. + assert all("sk-abcdefghijklmnopqrstuvwx" not in f.snippet for f in findings) + + +def test_env_secret_reference_is_clean_literal_is_not(): + ref = {"url": "https://x.example.com", "env": {"API_TOKEN": "${API_TOKEN}"}} + assert audit_server("ref", ref) == [] + lit = {"url": "https://x.example.com", "env": {"API_TOKEN": "ghp_" + "a" * 36}} + assert "WRD-AUTH-TOKEN-IN-CONFIG" in _rules(audit_server("lit", lit)) + + +def test_stdio_local_command_server_no_findings(): + # No url + no remote transport => a local stdio server, nothing to flag. + assert audit_server("fs", {"command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem"]}) == [] + + +def test_audit_config_iterates_all_servers_sorted(): + doc = { + "mcpServers": { + "a": {"url": "http://a.example.com/sse"}, + "b": {"url": "https://b.example.com", "headers": {"Authorization": "${T}"}}, + } + } + findings = audit_config(doc) + assert all(f.target.startswith("mcpServers/") for f in findings) + assert findings == sorted(findings, key=lambda f: (f.target, f.rule_id, f.snippet)) + + +def test_audit_config_ignores_non_object_shapes(): + assert audit_config({"mcpServers": "nope"}) == [] + assert audit_config({}) == [] + + +def test_cli_audit_clean_exit_zero(tmp_path): + cfg = tmp_path / "mcp.json" + cfg.write_text(json.dumps({"mcpServers": {"ok": {"url": "https://x.example.com", "headers": {"Authorization": "${T}"}}}})) + result = runner.invoke(app, ["auth", "audit", str(cfg)]) + assert result.exit_code == 0, result.output + + +def test_cli_audit_findings_exit_one(tmp_path): + cfg = tmp_path / "mcp.json" + cfg.write_text(json.dumps({"mcpServers": {"bad": {"url": "http://x.example.com/sse"}}})) + result = runner.invoke(app, ["auth", "audit", str(cfg)]) + assert result.exit_code == 1, result.output + + +def test_cli_audit_json_output(tmp_path): + cfg = tmp_path / "mcp.json" + cfg.write_text(json.dumps({"mcpServers": {"bad": {"url": "http://x.example.com/sse"}}})) + result = runner.invoke(app, ["auth", "audit", str(cfg), "--json"]) + assert result.exit_code == 1 + lines = [line for line in result.stdout.splitlines() if line.strip()] + assert any("WRD-AUTH-" in line for line in lines) + + +def test_cli_audit_malformed_config_exit_two(tmp_path): + cfg = tmp_path / "mcp.json" + cfg.write_text("{ not json") + result = runner.invoke(app, ["auth", "audit", str(cfg)]) + assert result.exit_code == 2 + + +def test_non_dict_headers_and_non_string_values_are_ignored(): + # Malformed shapes must be skipped, never crash the audit. + from mcp_warden.auth_audit import _scan_mapping_for_literals + + assert _scan_mapping_for_literals("not-a-dict", "t") == [] # type: ignore[arg-type] + assert _scan_mapping_for_literals({"token": 12345, "empty": ""}, "t") == [] + + +def test_short_credential_literal_is_fully_masked(): + server = {"url": "https://x.example.com", "headers": {"Authorization": "abc"}} + findings = audit_server("short", server) + assert any(f.snippet == "***" for f in findings) + + +def test_unreadable_config_path_raises(tmp_path): + import pytest + + from mcp_warden.auth_audit import audit_path + + with pytest.raises(AuthAuditError): + audit_path(tmp_path / "does-not-exist.json") + + +def test_transport_only_remote_without_url_flags_noauth(): + # A streamable-http server declared by transport type, no url, no auth. + findings = audit_server("ws", {"type": "streamable-http"}) + assert "WRD-AUTH-NOAUTH" in _rules(findings) + + +def test_cli_audit_sarif_output(tmp_path): + cfg = tmp_path / "mcp.json" + cfg.write_text(json.dumps({"mcpServers": {"bad": {"url": "http://x.example.com/sse"}}})) + sarif = tmp_path / "out.sarif" + result = runner.invoke(app, ["auth", "audit", str(cfg), "--sarif", str(sarif)]) + assert result.exit_code == 1 + doc = json.loads(sarif.read_text()) + assert doc["runs"][0]["results"], "SARIF must carry the findings" + + +def test_audit_path_raises_on_non_object(tmp_path): + cfg = tmp_path / "mcp.json" + cfg.write_text("[1, 2, 3]") + import pytest + + from mcp_warden.auth_audit import audit_path + + with pytest.raises(AuthAuditError): + audit_path(cfg) diff --git a/tests/test_deploy_gate.py b/tests/test_deploy_gate.py new file mode 100644 index 0000000..efab76b --- /dev/null +++ b/tests/test_deploy_gate.py @@ -0,0 +1,179 @@ +"""Fail-closed deploy-gate tests (WRD-GATE-*) — DSE-1257. + +Covers the pure ``evaluate_gate`` engine (each control passing + each failing, +plus fail-closed on missing/malformed evidence) and the ``deploy-gate`` CLI +(exit codes, JSON, fail-closed on unreadable inputs). +""" + +from __future__ import annotations + +import json + +from typer.testing import CliRunner + +from mcp_warden.cli import app +from mcp_warden.deploy_gate import DeployGateError, evaluate_gate + +runner = CliRunner() + + +def _rules(outcome): + return {f.rule_id for f in outcome.findings} + + +FULL_POLICY = { + "required_evals": [{"suite": "safety", "min_score": 0.9}], + "required_guardrails": ["prompt-injection", "pii-redaction"], + "require_budget": True, + "require_approval": True, +} + +FULL_EVIDENCE = { + "evals": {"safety": {"score": 0.95}}, + "guardrails": ["prompt-injection", "pii-redaction"], + "budget": {"limit": 100}, + "approval": {"approved": True, "approver": "ernest@thedataexperts.us"}, +} + + +def test_fully_satisfied_gate_passes(): + outcome = evaluate_gate(FULL_POLICY, FULL_EVIDENCE) + assert outcome.passed + assert outcome.controls_checked == 5 + + +def test_eval_below_threshold_fails(): + ev = {**FULL_EVIDENCE, "evals": {"safety": {"score": 0.5}}} + outcome = evaluate_gate(FULL_POLICY, ev) + assert not outcome.passed + assert "WRD-GATE-EVAL-THRESHOLD" in _rules(outcome) + + +def test_missing_eval_suite_fails_closed(): + ev = {**FULL_EVIDENCE, "evals": {}} + assert "WRD-GATE-EVAL-MISSING" in _rules(evaluate_gate(FULL_POLICY, ev)) + + +def test_non_numeric_eval_score_fails_closed(): + ev = {**FULL_EVIDENCE, "evals": {"safety": {"score": "great"}}} + assert "WRD-GATE-EVAL-MALFORMED" in _rules(evaluate_gate(FULL_POLICY, ev)) + + +def test_missing_guardrail_fails(): + ev = {**FULL_EVIDENCE, "guardrails": ["prompt-injection"]} + outcome = evaluate_gate(FULL_POLICY, ev) + assert "WRD-GATE-GUARDRAIL-MISSING" in _rules(outcome) + + +def test_missing_budget_fails_when_required(): + ev = {k: v for k, v in FULL_EVIDENCE.items() if k != "budget"} + assert "WRD-GATE-BUDGET-MISSING" in _rules(evaluate_gate(FULL_POLICY, ev)) + + +def test_non_positive_budget_fails(): + ev = {**FULL_EVIDENCE, "budget": {"limit": 0}} + assert "WRD-GATE-BUDGET-INVALID" in _rules(evaluate_gate(FULL_POLICY, ev)) + + +def test_missing_approval_fails_when_required(): + ev = {k: v for k, v in FULL_EVIDENCE.items() if k != "approval"} + assert "WRD-GATE-APPROVAL-MISSING" in _rules(evaluate_gate(FULL_POLICY, ev)) + + +def test_unapproved_receipt_fails(): + ev = {**FULL_EVIDENCE, "approval": {"approved": False, "approver": "x"}} + assert "WRD-GATE-APPROVAL-INVALID" in _rules(evaluate_gate(FULL_POLICY, ev)) + + +def test_unattributed_approval_fails(): + ev = {**FULL_EVIDENCE, "approval": {"approved": True, "approver": ""}} + assert "WRD-GATE-APPROVAL-INVALID" in _rules(evaluate_gate(FULL_POLICY, ev)) + + +def test_empty_policy_passes_vacuously_but_reports_zero_controls(): + outcome = evaluate_gate({}, {}) + assert outcome.passed + assert outcome.controls_checked == 0 + + +def test_malformed_evals_evidence_block_fails_closed(): + outcome = evaluate_gate(FULL_POLICY, {**FULL_EVIDENCE, "evals": ["not", "a", "dict"]}) + assert "WRD-GATE-EVAL-EVIDENCE" in _rules(outcome) + + +def _write(tmp_path, name, doc): + p = tmp_path / name + p.write_text(json.dumps(doc)) + return p + + +def test_cli_pass_exit_zero(tmp_path): + policy = _write(tmp_path, "policy.json", FULL_POLICY) + evidence = _write(tmp_path, "evidence.json", FULL_EVIDENCE) + result = runner.invoke(app, ["deploy-gate", "--policy", str(policy), "--evidence", str(evidence)]) + assert result.exit_code == 0, result.output + + +def test_cli_fail_exit_one(tmp_path): + policy = _write(tmp_path, "policy.json", FULL_POLICY) + bad = {**FULL_EVIDENCE, "approval": {"approved": False, "approver": "x"}} + evidence = _write(tmp_path, "evidence.json", bad) + result = runner.invoke(app, ["deploy-gate", "--policy", str(policy), "--evidence", str(evidence)]) + assert result.exit_code == 1, result.output + + +def test_cli_json_output(tmp_path): + policy = _write(tmp_path, "policy.json", FULL_POLICY) + bad = {**FULL_EVIDENCE, "guardrails": []} + evidence = _write(tmp_path, "evidence.json", bad) + result = runner.invoke(app, ["deploy-gate", "--policy", str(policy), "--evidence", str(evidence), "--json"]) + assert result.exit_code == 1 + assert any("WRD-GATE-" in line for line in result.stdout.splitlines() if line.strip()) + + +def test_cli_unreadable_evidence_exit_two(tmp_path): + policy = _write(tmp_path, "policy.json", FULL_POLICY) + evidence = tmp_path / "evidence.json" + evidence.write_text("{ broken") + result = runner.invoke(app, ["deploy-gate", "--policy", str(policy), "--evidence", str(evidence)]) + assert result.exit_code == 2 + + +def test_non_object_policy_document_fails_closed(tmp_path): + import pytest + + from mcp_warden.deploy_gate import run_deploy_gate + + policy = tmp_path / "policy.json" + policy.write_text("[1, 2, 3]") + evidence = _write(tmp_path, "evidence.json", FULL_EVIDENCE) + with pytest.raises(DeployGateError): + run_deploy_gate(policy, evidence) + + +def test_malformed_eval_spec_entries_are_skipped(): + policy = {"required_evals": ["not-a-dict", {"suite": "safety", "min_score": 0.9}]} + outcome = evaluate_gate(policy, {"evals": {"safety": {"score": 0.99}}}) + assert outcome.passed + + +def test_cli_sarif_output(tmp_path): + policy = _write(tmp_path, "policy.json", FULL_POLICY) + bad = {**FULL_EVIDENCE, "budget": {"limit": 0}} + evidence = _write(tmp_path, "evidence.json", bad) + sarif = tmp_path / "gate.sarif" + result = runner.invoke( + app, ["deploy-gate", "--policy", str(policy), "--evidence", str(evidence), "--sarif", str(sarif)] + ) + assert result.exit_code == 1 + doc = json.loads(sarif.read_text()) + assert doc["runs"][0]["results"] + + +def test_run_deploy_gate_raises_on_missing_file(tmp_path): + import pytest + + from mcp_warden.deploy_gate import run_deploy_gate + + with pytest.raises(DeployGateError): + run_deploy_gate(tmp_path / "nope.json", tmp_path / "also-nope.json") From 93986bc4e907091209dcebcb246e6f2170defe4f Mon Sep 17 00:00:00 2001 From: ernestprovo23 Date: Sun, 23 Aug 2026 20:02:31 -0400 Subject: [PATCH 2/4] fix(auth-audit): strip userinfo credentials from findings and flag them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Self-review caught a real leak in the module's own reporting path: the WRD-AUTH-PLAINTEXT-HTTP finding echoed the raw URL, so an endpoint configured as https://user:token@host would have written that credential into the finding snippet and the SARIF report — the audit widening exposure of the very thing it reports. - _host_of now drops any user:pass@ userinfo, so authority parsing cannot mistake the userinfo for the host (which also fixed a locality-detection hole: user@127.0.0.1-style values no longer confuse the loopback check). - _safe_url renders scheme://host for snippets, never the credential. - New WRD-AUTH-URL-CREDENTIAL (high): a URL-embedded credential is itself a config finding, not just something to redact. 836 passed, auth_audit.py at 100% coverage, ruff clean. --- docs/AGENT_GATES.md | 1 + src/mcp_warden/auth_audit.py | 37 +++++++++++++++++++++++++++++++++--- tests/test_auth_audit.py | 19 ++++++++++++++++++ 3 files changed, 54 insertions(+), 3 deletions(-) diff --git a/docs/AGENT_GATES.md b/docs/AGENT_GATES.md index 8847fb7..eccf693 100644 --- a/docs/AGENT_GATES.md +++ b/docs/AGENT_GATES.md @@ -111,6 +111,7 @@ and tracked separately (DSE-725). | `WRD-AUTH-NOAUTH` | medium | A remote endpoint declares no auth material | | `WRD-AUTH-PLAINTEXT-HTTP` | high | A remote endpoint uses `http://` | | `WRD-AUTH-TOKEN-IN-CONFIG` | high | An auth-bearing key holds a literal credential | +| `WRD-AUTH-URL-CREDENTIAL` | high | The endpoint URL embeds a `user:pass@` userinfo credential | | `WRD-SEC-*` | varies | Vendor secret patterns found in any config value (shared with `check`) | ### What it deliberately does not flag diff --git a/src/mcp_warden/auth_audit.py b/src/mcp_warden/auth_audit.py index e6a9d83..ce73447 100644 --- a/src/mcp_warden/auth_audit.py +++ b/src/mcp_warden/auth_audit.py @@ -45,9 +45,26 @@ def _is_local_host(host: str) -> bool: def _host_of(url: str) -> str: - """Extract the bare host[:port] from an http(s) URL without a full parse.""" + """Extract the bare host[:port] from an http(s) URL without a full parse. + + Any ``user:password@`` userinfo is dropped: a credential embedded in the URL + must never reach a finding snippet (see :func:`_safe_url`). + """ rest = url.split("://", 1)[-1] - return rest.split("/", 1)[0] + authority = rest.split("/", 1)[0] + return authority.rsplit("@", 1)[-1] + + +def _safe_url(url: str) -> str: + """Render a URL for a finding snippet with any userinfo credential stripped. + + ``https://user:tok@host/path`` -> ``https://host`` — the audit must not widen + exposure of the very credential it is reporting. + """ + scheme, sep, rest = url.partition("://") + if not sep: + return _host_of(url) + return f"{scheme}://{_host_of(url)}" def _looks_like_secret_ref(value: str) -> bool: @@ -132,6 +149,20 @@ def audit_server(name: str, server: dict[str, Any]) -> list[Finding]: if isinstance(url, str) and url: host = _host_of(url) remote = not _is_local_host(host) + authority = url.partition("://")[2].split("/", 1)[0] + if "@" in authority: + findings.append( + Finding( + rule_id="WRD-AUTH-URL-CREDENTIAL", + severity="high", + target=target, + message=( + "MCP endpoint URL embeds a userinfo credential; move it to a " + "header referencing a secret manager" + ), + snippet=_safe_url(url), + ) + ) if url.lower().startswith("http://") and remote: findings.append( Finding( @@ -139,7 +170,7 @@ def audit_server(name: str, server: dict[str, Any]) -> list[Finding]: severity="high", target=target, message=f"remote MCP endpoint '{host}' uses cleartext http://; use https://", - snippet=url, + snippet=_safe_url(url), ) ) if remote and not _server_has_auth(server): diff --git a/tests/test_auth_audit.py b/tests/test_auth_audit.py index db325e5..3209fbc 100644 --- a/tests/test_auth_audit.py +++ b/tests/test_auth_audit.py @@ -105,6 +105,25 @@ def test_cli_audit_malformed_config_exit_two(tmp_path): assert result.exit_code == 2 +def test_url_userinfo_credential_flagged_and_never_echoed(): + # A credential in the URL authority must be flagged AND stripped from every + # snippet — the audit must not widen exposure of what it reports. + server = {"url": "https://svcuser:s3cr3t-token@mcp.example.com/sse"} + findings = audit_server("userinfo", server) + assert "WRD-AUTH-URL-CREDENTIAL" in _rules(findings) + assert all("s3cr3t-token" not in f.snippet for f in findings) + assert all("s3cr3t-token" not in f.message for f in findings) + + +def test_userinfo_host_parsing_does_not_confuse_locality(): + # '@' in the authority must not make host detection read the userinfo as host. + from mcp_warden.auth_audit import _host_of, _safe_url + + assert _host_of("https://user:pw@real.example.com/x") == "real.example.com" + assert _safe_url("https://user:pw@real.example.com/x") == "https://real.example.com" + assert _safe_url("real.example.com") == "real.example.com" + + def test_non_dict_headers_and_non_string_values_are_ignored(): # Malformed shapes must be skipped, never crash the audit. from mcp_warden.auth_audit import _scan_mapping_for_literals From 25890026059f27b30b3aabee5bce33932d2ce1be Mon Sep 17 00:00:00 2001 From: ernestprovo23 Date: Sun, 23 Aug 2026 20:44:20 -0400 Subject: [PATCH 3/4] docs: runnable agent-gate examples, pinned to their documented verdicts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds examples/agent-gates/ — a six-server MCP config and pass/fail deploy evidence — plus a regression suite that asserts the examples produce exactly the verdicts their README claims. The demo config deliberately includes three servers that must NOT be flagged (local stdio, loopback, and a ${VAR} secret reference). A gate that cries wolf on correct configuration gets switched off, so the non-flags are pinned as tightly as the findings. test_agent_gates_examples.py also asserts the two planted fake credentials never appear in any finding field, keeping the redaction guarantee honest against the real shipped artifact rather than a synthetic fixture. 843 passed, coverage 86.13%, ruff clean. --- DOCUMENTATION_INDEX.md | 1 + examples/README.md | 1 + examples/agent-gates/README.md | 61 ++++++++++++ examples/agent-gates/evidence-fail.json | 10 ++ examples/agent-gates/evidence-pass.json | 13 +++ examples/agent-gates/gate-policy.json | 9 ++ .../agent-gates/mcp-config-audit-demo.json | 26 +++++ tests/test_agent_gates_examples.py | 97 +++++++++++++++++++ 8 files changed, 218 insertions(+) create mode 100644 examples/agent-gates/README.md create mode 100644 examples/agent-gates/evidence-fail.json create mode 100644 examples/agent-gates/evidence-pass.json create mode 100644 examples/agent-gates/gate-policy.json create mode 100644 examples/agent-gates/mcp-config-audit-demo.json create mode 100644 tests/test_agent_gates_examples.py diff --git a/DOCUMENTATION_INDEX.md b/DOCUMENTATION_INDEX.md index 051e97f..98af702 100644 --- a/DOCUMENTATION_INDEX.md +++ b/DOCUMENTATION_INDEX.md @@ -30,6 +30,7 @@ is the MCP auth posture sound. Both reuse the `check` exit-code contract (0 clea | [`src/mcp_warden/cli_auth.py`](src/mcp_warden/cli_auth.py) | `auth audit` sub-app command body | | [`tests/test_deploy_gate.py`](tests/test_deploy_gate.py) | Engine per-control pass/fail + CLI exit codes + fail-closed on malformed evidence | | [`tests/test_auth_audit.py`](tests/test_auth_audit.py) | Every rule, the deliberate non-flags (loopback, `${VAR}` refs, stdio), redaction, CLI/JSON/SARIF | +| [`examples/agent-gates/`](examples/agent-gates/) | Runnable demos — a 6-server config (4 flagged / 2 deliberately clean) and pass/fail deploy evidence, with expected verdicts documented and verified | ## GitHub Action (`action.yml` — Issue #18) diff --git a/examples/README.md b/examples/README.md index 4d5b34b..e885936 100644 --- a/examples/README.md +++ b/examples/README.md @@ -17,6 +17,7 @@ are re-checked on every run so these examples stay green. | [`gitlab-ci/.gitlab-ci.yml`](gitlab-ci/.gitlab-ci.yml) | the same check gate on GitLab CI | | [`pre-commit/.pre-commit-config.yaml`](pre-commit/.pre-commit-config.yaml) | local pre-commit + pre-push hook variants | | [`pinned-servers/`](pinned-servers/) | real MCP servers pinned to a committed `warden.lock` each | +| [`agent-gates/`](agent-gates/) | runnable `deploy-gate` + `auth audit` demos — a config with four flagged servers and two that must NOT be flagged, plus pass/fail deploy evidence | ## Pinned-server examples diff --git a/examples/agent-gates/README.md b/examples/agent-gates/README.md new file mode 100644 index 0000000..dfc657c --- /dev/null +++ b/examples/agent-gates/README.md @@ -0,0 +1,61 @@ +# Agent gates — runnable examples + +Two fail-closed CI gates. Full contract: [`docs/AGENT_GATES.md`](../../docs/AGENT_GATES.md). + +## `auth audit` — static MCP auth posture + +```bash +mcp-warden auth audit examples/agent-gates/mcp-config-audit-demo.json +``` + +The demo config has six servers. **Four are flagged, two must not be** — the +non-flags matter as much as the findings, because a gate that cries wolf on +correct configuration gets disabled. + +| Server | Verdict | +|--------|---------| +| `filesystem-local` | clean — local stdio server, no auth posture to audit | +| `loopback-dev` | clean — loopback is not remotely reachable | +| `good-citizen` | clean — `${VENDOR_TOKEN}` is a reference, not a literal | +| `internal-http` | `WRD-AUTH-PLAINTEXT-HTTP` (high) + `WRD-AUTH-NOAUTH` (medium) | +| `vendor-api` | `WRD-AUTH-TOKEN-IN-CONFIG` (high) + `WRD-SEC-ENTROPY` (high) | +| `legacy` | `WRD-AUTH-URL-CREDENTIAL` (high) + `WRD-AUTH-NOAUTH` (medium) | + +Exits `1` with six findings. Every credential is redacted in output and SARIF. + +Static only: no server is spawned, no DNS is resolved, no network is touched. + +## `deploy-gate` — release control for agent deploys + +```bash +mcp-warden deploy-gate \ + --policy examples/agent-gates/gate-policy.json \ + --evidence examples/agent-gates/evidence-pass.json # exit 0 + +mcp-warden deploy-gate \ + --policy examples/agent-gates/gate-policy.json \ + --evidence examples/agent-gates/evidence-fail.json # exit 1 +``` + +`evidence-fail.json` trips four controls at once: the safety eval regressed +below threshold, a required guardrail was switched off, the budget has no +positive limit, and the approval receipt is unattributed. + +The gate **adjudicates evidence — it does not run evals.** Your pipeline runs +them and writes the evidence file; the gate decides deterministically whether +the deploy may proceed. Missing or malformed evidence is a failure, never a +pass. + +## In CI + +```yaml +- name: MCP auth posture + run: mcp-warden auth audit .mcp.json --sarif auth.sarif + +- name: Agent deploy gate + run: mcp-warden deploy-gate --policy gate-policy.json --evidence evidence.json --sarif gate.sarif + +- uses: github/codeql-action/upload-sarif@v3 + with: + sarif_file: auth.sarif +``` diff --git a/examples/agent-gates/evidence-fail.json b/examples/agent-gates/evidence-fail.json new file mode 100644 index 0000000..0ed5fb6 --- /dev/null +++ b/examples/agent-gates/evidence-fail.json @@ -0,0 +1,10 @@ +{ + "_comment": "A deploy that must be blocked: safety eval regressed below threshold, the pii-redaction guardrail was switched off, the budget has no positive limit, and the approval receipt is unattributed.", + "evals": { + "safety": { "score": 0.62 }, + "task-success": { "score": 0.91 } + }, + "guardrails": ["prompt-injection"], + "budget": { "limit": 0 }, + "approval": { "approved": true, "approver": "" } +} diff --git a/examples/agent-gates/evidence-pass.json b/examples/agent-gates/evidence-pass.json new file mode 100644 index 0000000..4c8d63f --- /dev/null +++ b/examples/agent-gates/evidence-pass.json @@ -0,0 +1,13 @@ +{ + "_comment": "Evidence a deploy pipeline emits after running its eval suites and recording active guardrails. Satisfies every control in gate-policy.json.", + "evals": { + "safety": { "score": 0.96 }, + "task-success": { "score": 0.88 } + }, + "guardrails": ["prompt-injection", "pii-redaction"], + "budget": { "limit": 250, "unit": "usd-per-day" }, + "approval": { + "approved": true, + "approver": "release-manager@example.com" + } +} diff --git a/examples/agent-gates/gate-policy.json b/examples/agent-gates/gate-policy.json new file mode 100644 index 0000000..82828bc --- /dev/null +++ b/examples/agent-gates/gate-policy.json @@ -0,0 +1,9 @@ +{ + "required_evals": [ + { "suite": "safety", "min_score": 0.9 }, + { "suite": "task-success", "min_score": 0.8 } + ], + "required_guardrails": ["prompt-injection", "pii-redaction"], + "require_budget": true, + "require_approval": true +} diff --git a/examples/agent-gates/mcp-config-audit-demo.json b/examples/agent-gates/mcp-config-audit-demo.json new file mode 100644 index 0000000..42cf8fa --- /dev/null +++ b/examples/agent-gates/mcp-config-audit-demo.json @@ -0,0 +1,26 @@ +{ + "_comment": "Demo MCP config for `mcp-warden auth audit`. Six servers: four with real posture problems, two that are correct and MUST NOT be flagged. All hosts are RFC 2606 example domains and every credential here is fake.", + "mcpServers": { + "filesystem-local": { + "command": "npx", + "args": ["-y", "@modelcontextprotocol/server-filesystem", "/data"] + }, + "loopback-dev": { + "url": "http://localhost:3000/mcp" + }, + "good-citizen": { + "url": "https://api.ok.example.com/mcp", + "headers": { "Authorization": "${VENDOR_TOKEN}" } + }, + "internal-http": { + "url": "http://mcp.internal.example.com:8080/sse" + }, + "vendor-api": { + "url": "https://api.vendor.example.com/mcp", + "headers": { "Authorization": "Bearer sk-live-9f2c8a7b6d5e4f3a2b1c0d9e" } + }, + "legacy": { + "url": "https://svc:hunter2pass@legacy.example.com/mcp" + } + } +} diff --git a/tests/test_agent_gates_examples.py b/tests/test_agent_gates_examples.py new file mode 100644 index 0000000..9369865 --- /dev/null +++ b/tests/test_agent_gates_examples.py @@ -0,0 +1,97 @@ +"""The shipped agent-gate examples must behave exactly as documented. + +`examples/agent-gates/README.md` states per-server verdicts and exit codes. A +documented claim nobody verifies rots; these tests pin the examples to their +README so a rule change that alters the demo output fails CI instead of quietly +making the docs wrong. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +from typer.testing import CliRunner + +from mcp_warden.auth_audit import audit_path +from mcp_warden.cli import app +from mcp_warden.deploy_gate import run_deploy_gate + +runner = CliRunner() + +EXAMPLES = Path(__file__).resolve().parents[1] / "examples" / "agent-gates" +CONFIG = EXAMPLES / "mcp-config-audit-demo.json" +POLICY = EXAMPLES / "gate-policy.json" +EVIDENCE_PASS = EXAMPLES / "evidence-pass.json" +EVIDENCE_FAIL = EXAMPLES / "evidence-fail.json" + +#: README table: which server yields which rules. +EXPECTED_AUTH = { + "mcpServers/internal-http": {"WRD-AUTH-PLAINTEXT-HTTP", "WRD-AUTH-NOAUTH"}, + "mcpServers/vendor-api": {"WRD-AUTH-TOKEN-IN-CONFIG", "WRD-SEC-ENTROPY"}, + "mcpServers/legacy": {"WRD-AUTH-URL-CREDENTIAL", "WRD-AUTH-NOAUTH"}, +} + +#: README: these three are correct configuration and must never be flagged. +CLEAN_SERVERS = {"mcpServers/filesystem-local", "mcpServers/loopback-dev", "mcpServers/good-citizen"} + + +def test_example_files_exist(): + for p in (CONFIG, POLICY, EVIDENCE_PASS, EVIDENCE_FAIL): + assert p.is_file(), f"shipped example missing: {p}" + + +def test_auth_demo_matches_documented_verdicts(): + findings = audit_path(CONFIG) + by_target: dict[str, set[str]] = {} + for f in findings: + by_target.setdefault(f.target, set()).add(f.rule_id) + assert by_target == EXPECTED_AUTH + assert len(findings) == 6, "README says six findings" + + +def test_auth_demo_never_flags_the_correct_servers(): + targets = {f.target for f in audit_path(CONFIG)} + assert not (targets & CLEAN_SERVERS), "a correct config was flagged — false positive" + + +def test_auth_demo_leaks_no_credential(): + # The two fake credentials planted in the demo config must never appear in + # any finding field. + planted = ("hunter2pass", "sk-live-9f2c8a7b6d5e4f3a2b1c0d9e") + blob = json.dumps([f.model_dump() for f in audit_path(CONFIG)]) + for secret in planted: + assert secret not in blob, f"credential leaked into findings: {secret[:6]}..." + + +def test_gate_pass_example_passes(): + outcome = run_deploy_gate(POLICY, EVIDENCE_PASS) + assert outcome.passed + assert outcome.controls_checked == 6 + + +def test_gate_fail_example_trips_exactly_four_controls(): + outcome = run_deploy_gate(POLICY, EVIDENCE_FAIL) + assert not outcome.passed + assert {f.rule_id for f in outcome.findings} == { + "WRD-GATE-EVAL-THRESHOLD", + "WRD-GATE-GUARDRAIL-MISSING", + "WRD-GATE-BUDGET-INVALID", + "WRD-GATE-APPROVAL-INVALID", + } + + +def test_documented_cli_exit_codes(): + assert runner.invoke(app, ["auth", "audit", str(CONFIG)]).exit_code == 1 + assert ( + runner.invoke( + app, ["deploy-gate", "--policy", str(POLICY), "--evidence", str(EVIDENCE_PASS)] + ).exit_code + == 0 + ) + assert ( + runner.invoke( + app, ["deploy-gate", "--policy", str(POLICY), "--evidence", str(EVIDENCE_FAIL)] + ).exit_code + == 1 + ) From d78a14c081ca0f6af9faedea52e90142a2453501 Mon Sep 17 00:00:00 2001 From: ernestprovo23 Date: Sun, 23 Aug 2026 20:45:23 -0400 Subject: [PATCH 4/4] docs: position the gates around what they block, not just what they report DSE-1256. The loudest complaint about agents in production is that they are insecure by default and nothing stops a bad config or a regressed deploy from shipping. Plenty of tools report; few return a non-zero exit a pipeline must answer for. Reframes the layer table and 'who it's for' around the common thread across all four commands: check blocks on surface drift, auth audit blocks on weak MCP auth posture, deploy-gate blocks a deploy below its declared safety bar, and all of them fail closed on missing or unreadable input. --- README.md | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index 3250b27..8683bd9 100644 --- a/README.md +++ b/README.md @@ -107,10 +107,23 @@ closes gaps none of them cover alone. | **Static tool-poisoning scanner** | [mcp-scan](https://github.com/invariantlabs-ai/mcp-scan) | pin-time / pre-flight | suspicious *content* in tool definitions (injection-style descriptions, known-bad patterns) | you want to catch a poisoned definition the first time you see it | | **Runtime gateway / proxy** | ContextForge, Lunar MCPX, TrueFoundry, Docker MCP Gateway | every live request | runtime mediation — auth, rate limits, request/response policy on calls in flight | you need to mediate or police live traffic between agent and server | | **Lockfile + CI gate** | **mcp-warden** | CI / pre-commit | *drift* — the declared surface changing after a human approved it (rug-pull / silent redefinition) | you want a reproducible, human-approved baseline that fails the build when the surface changes | +| **Config + deploy gates** | **mcp-warden** (`auth audit`, `deploy-gate`) | CI / pre-commit | *posture* — remote MCP endpoints with no auth or credentials pasted into config; agent deploys whose evals regressed or whose guardrails were switched off | you want the deploy blocked, not just reported, when the declared safety bar isn't met | mcp-warden does not replace a scanner or a gateway — it adds the missing **drift gate**: a signed baseline plus a deterministic CI check that the surface you -approved is the surface you still run. For the full, sourced breakdown of how +approved is the surface you still run. + +**The common thread across all four commands is that they *block*.** The loudest +complaint about agents in production is that they are insecure by default and +nothing stops a bad configuration or a regressed deploy from shipping — plenty of +tools *report*, very few return a non-zero exit code that a pipeline must answer +for. `check` blocks on surface drift, `auth audit` blocks on weak MCP auth +posture, and `deploy-gate` blocks an agent deploy whose evals, guardrails, +budget, or human approval don't meet the declared bar. All three fail **closed**: +unreadable or missing input is a failure, never a silent pass. See +[`docs/AGENT_GATES.md`](docs/AGENT_GATES.md). + +For the full, sourced breakdown of how these layers complement each other and when to use which, see the [**comparison page**](https://datascience-engineeringexperts.github.io/mcp-warden/comparison/) on the docs site. @@ -129,6 +142,9 @@ automatically — so the use cases are sequenced by leverage: pre-commit hook) fails when upstream silently redefines its surface — the core rug-pull defense. - **Security / platform engineer.** Run the [Action](#github-action-one-step-drop-in) across a fleet; SARIF → code scanning; signed locks = auditable human-approval evidence. + Add `auth audit` to catch MCP endpoints configured without auth or with credentials + committed into config, and `deploy-gate` to make the agent safety bar a build failure + rather than a dashboard nobody reads. - **Incident responder / auditor.** `inspect` an offline trace and `warden diff` a suspect lock against a known-good baseline — no live server required. - **Agent-framework integrator** *(post-launch).* Enforce that only warden-locked servers