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..98af702 100644 --- a/DOCUMENTATION_INDEX.md +++ b/DOCUMENTATION_INDEX.md @@ -14,6 +14,24 @@ 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 | +| [`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) The composite reusable action is the primary delivery vehicle for the `check` gate. diff --git a/README.md b/README.md index 3812f2e..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 @@ -391,6 +407,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..eccf693 --- /dev/null +++ b/docs/AGENT_GATES.md @@ -0,0 +1,136 @@ +# 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-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 + +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/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/src/mcp_warden/auth_audit.py b/src/mcp_warden/auth_audit.py new file mode 100644 index 0000000..ce73447 --- /dev/null +++ b/src/mcp_warden/auth_audit.py @@ -0,0 +1,234 @@ +"""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. + + 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] + 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: + """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) + 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( + rule_id="WRD-AUTH-PLAINTEXT-HTTP", + severity="high", + target=target, + message=f"remote MCP endpoint '{host}' uses cleartext http://; use https://", + snippet=_safe_url(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_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 + ) diff --git a/tests/test_auth_audit.py b/tests/test_auth_audit.py new file mode 100644 index 0000000..3209fbc --- /dev/null +++ b/tests/test_auth_audit.py @@ -0,0 +1,174 @@ +"""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_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 + + 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")