From 3bbeb118ccc73b2881b9c08b2efc32cec25adb59 Mon Sep 17 00:00:00 2001 From: henleda Date: Wed, 29 Jul 2026 12:54:24 -0500 Subject: [PATCH] feat(inputs): H3 OpenAPI spec as a discovery input MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A4 goes one way: you hand XC a schema and it enforces it. This is the other direction — read the spec and find the flaws IN it. A spec is a security artifact whether anyone treats it as one, and it is often the only thing you have for a service you do not own. vpcopilot scan --spec ./openapi.yaml # the contract alone vpcopilot scan ./app --spec ./openapi.yaml # …and the drift between them `--spec` is additive, unlike `--cve`: alone it scans the contract, alongside a repo it also compares the two (which needs both). SPLIT BY WHAT NEEDS JUDGEMENT. A deterministic pass finds what the document leaves unstated — a number with no `minimum`, a string with no `maxLength`, an operation with no `security`, `additionalProperties` left open, `$ref` followed with a cycle bound. Those are facts, so recall does not depend on which model is configured. The agent decides which of them MATTER: an unbounded `page` is nothing, an unbounded `amount` on a transfer is a business-logic hole. The spec goes to the EXISTING discover agent as one more file — no fourth agent, so spec findings verify, triage and generate exactly like code findings with no parallel path to keep in step. Acceptance verified live: a spec declaring `amount` with no lower bound produced neg-amount-001 -> service_policy + api_schema; spec/code drift reports as `undocumented_or_orphaned`; findings flow into the same triage table. TWO REAL BUGS THE FIRST LIVE RUN SURFACED: * A spec is ONE file declaring MANY endpoints, so endpoint_of("pay-api.yaml") returned the same string for every finding — all six collapsed onto one service_policy coverage key and five were logged "already covered" and given no band-aid at all. Spec findings now carry `source` and leave `file` empty, and the coverage identity prefers `endpoint`: three endpoints, three keys, and only the two genuinely on /api/v1/transfer collapse. * The undocumented_or_orphaned finding was going through `verify`, whose whole method is reading the offending source. With no source to read it refuted the finding at 0.10 confidence. It is appended after verify instead — a comparison of two documents is not a claim about code. Drift is reported in both directions and they fail differently: declared-but- unserved is dead documentation or a shadow API; served-but-undeclared is what an api_schema band-aid built from this spec would start rejecting. run.json records input_kind as repo / spec / repo+spec / advisory. 502 tests, ruff clean. Fixture: bench/fixtures/specs/pay-api.yaml. Co-Authored-By: Claude Opus 5 (1M context) --- ROADMAP.md | 34 +++- bench/fixtures/specs/pay-api.yaml | 39 +++++ docs/USAGE.md | 31 ++++ src/vpcopilot/cli.py | 9 +- src/vpcopilot/console/app.py | 16 +- src/vpcopilot/console/static/index.html | 13 +- src/vpcopilot/inputs/openapi.py | 223 ++++++++++++++++++++++++ src/vpcopilot/pipeline.py | 94 +++++++++- tests/test_inputs_cve.py | 9 +- tests/test_inputs_openapi.py | 175 +++++++++++++++++++ 10 files changed, 609 insertions(+), 34 deletions(-) create mode 100644 bench/fixtures/specs/pay-api.yaml create mode 100644 src/vpcopilot/inputs/openapi.py create mode 100644 tests/test_inputs_openapi.py diff --git a/ROADMAP.md b/ROADMAP.md index a94becd..434453d 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -342,16 +342,32 @@ ship. That is the case virtual patching exists for, and the pipeline cannot see non-file identity (fall back to `Finding.endpoint` or the package coordinate) and `pipeline.py:40`'s dedup key needs the same treatment. -- [ ] **H3** OpenAPI as a discovery input. (M, P2) - `A4` applies a schema the operator supplies. This is the other direction: read a spec and - find the flaws in it, including endpoints absent from the code and code paths absent from - the spec. - - Acceptance: a spec declaring `amount` with no lower bound gets reported as a finding and - routes to `api_schema`; endpoints in the spec with no code match report as +- [x] **H3** OpenAPI as a discovery input. (M, P2) — **DONE:** `vpcopilot scan --spec `, + alone or alongside a repo. `inputs/openapi.py` does a deterministic structural pass (unbounded + numbers/strings/arrays, operations with no `security`, `additionalProperties` left open, `$ref` + followed with a cycle bound) and the spec is then handed to the **existing `discover` agent** as + one more file — no fourth agent, so spec findings verify, triage and generate exactly like code + findings. Verified live end to end. + - **Acceptance, as met:** a spec declaring `amount` with no lower bound produced + `neg-amount-001` → `service_policy` + **`api_schema`**; spec/code drift is reported as `undocumented_or_orphaned`; findings flow into the same triage table. - - Surfaces: `src/vpcopilot/inputs/openapi.py`, `vpcopilot scan --spec `. - - **Reconciled:** same `inputs/` package question as H1 — settle it there first. `--spec` - changes the same required `repo` positional H1 does; do both in one change. + - **Split by what needs judgement.** The structural pass is code, so recall does not depend on + which model is configured; the agent only decides which facts *matter* (an unbounded `page` is + nothing, an unbounded `amount` on a transfer is a hole). The orphan comparison is pure code — + it is a diff of two documents. + - **Two real bugs the first live run surfaced.** (a) A spec is ONE file declaring MANY endpoints, + so `endpoint_of("pay-api.yaml")` returned the same string for every finding and all six + collapsed onto one `service_policy` coverage key, with five logged "already covered" and given + no band-aid. Spec findings now carry `source` and leave `file` empty, and the coverage identity + prefers `endpoint` — three endpoints, three keys, and only the two genuinely on + `/api/v1/transfer` collapse. (b) The `undocumented_or_orphaned` finding was being sent through + `verify`, which reads the offending source; with no source to read it refuted the finding at + 0.10 confidence. It is appended after verify instead — a comparison of two documents is not a + claim about code. + - **`--spec` is additive**, unlike `--cve`: alone it scans the contract, with a repo it also + reports the drift (which needs both). `run.json` records `input_kind` as + `repo` / `spec` / `repo+spec` / `advisory`. + - Fixture: `bench/fixtures/specs/pay-api.yaml`. --- diff --git a/bench/fixtures/specs/pay-api.yaml b/bench/fixtures/specs/pay-api.yaml new file mode 100644 index 0000000..0de6979 --- /dev/null +++ b/bench/fixtures/specs/pay-api.yaml @@ -0,0 +1,39 @@ +openapi: 3.0.3 +info: + title: Nimbus Payments API + version: "1.0" +servers: [{url: https://lab.banknimbus.com}] +paths: + /api/v1/transfer: + post: + operationId: createTransfer + summary: Move funds between two accounts + security: [] + requestBody: + required: true + content: + application/json: + schema: + type: object + required: [from, to, amount] + additionalProperties: true + properties: + from: {type: string} + to: {type: string} + amount: + type: number + description: amount to transfer, in USD + memo: {type: string} + responses: {"200": {description: ok}} + /api/v1/accounts/{accountId}: + get: + operationId: getAccount + summary: Read an account + parameters: + - {name: accountId, in: path, required: true, schema: {type: string}} + responses: {"200": {description: ok}} + /api/v1/admin/reset: + post: + operationId: adminReset + summary: Reset the ledger (internal) + responses: {"204": {description: done}} diff --git a/docs/USAGE.md b/docs/USAGE.md index 25490d8..0ad60b2 100644 --- a/docs/USAGE.md +++ b/docs/USAGE.md @@ -85,6 +85,37 @@ Three things about OSV worth knowing, each found by querying it: Set `VPCOPILOT_ADVISORY_CACHE=` to cache advisories on disk so a demo does not depend on the network. +### Scan an OpenAPI spec (H3) + +A4 goes one way: you hand XC a schema and it enforces it. This is the other direction — read the +spec and find the flaws **in** it. A spec is a security artifact whether or not anyone treats it as +one, and it is often the only thing you have for a service you do not own. + +```sh +vpcopilot scan --spec ./openapi.yaml --out out # the contract alone +vpcopilot scan ./app --spec ./openapi.yaml --out out # …and the code, plus the drift between them +``` + +`--spec` is **additive**: alone it scans the contract; alongside a repo it also compares the two. +(`--cve` is the exception — an advisory scan cannot be combined with either.) + +**Split by what needs judgement.** A deterministic pass finds what the document leaves unstated — +a number with no `minimum`, a string with no `maxLength`, an operation with no `security`, an object +with `additionalProperties` open. Those are facts, so recall does not depend on which model is +configured. The agent then decides which of them *matter*: an unbounded `page` is nothing, an +unbounded `amount` on a transfer is a business-logic hole. + +**Spec/code drift is a finding in both directions**, reported as `undocumented_or_orphaned`: + +- **declared but unserved** — dead documentation, or a shadow API removed from one place and not + the other +- **served but undeclared** — the one that bites: applying an `api_schema` band-aid built from this + spec would start rejecting those routes + +That finding is a comparison of two documents, so it skips the verify agent entirely — asking an +adversarial code reviewer to confirm a vulnerability in source it cannot see got it refuted at 0.10 +confidence. + ## 4. Apply a band-aid (mutates XC — gated + reversible) ```sh vpcopilot apply --from-scan out/policies/.json --lb --url --dry-run # preview diff --git a/src/vpcopilot/cli.py b/src/vpcopilot/cli.py index a415348..ce7e25c 100644 --- a/src/vpcopilot/cli.py +++ b/src/vpcopilot/cli.py @@ -35,6 +35,7 @@ def _root( def scan( repo: str = typer.Argument(None, help="path to the target application repo (omit when using --cve)"), cve: str = typer.Option(None, "--cve", help="scan a security advisory instead of a repo: CVE-YYYY-NNNNN, GHSA-xxxx-xxxx-xxxx, PYSEC-YYYY-NN, GO-… or RUSTSEC-…"), + spec: str = typer.Option(None, "--spec", help="an OpenAPI/Swagger spec to scan for flaws IN THE CONTRACT — alone, or alongside a repo to also report spec/code drift"), out: str = typer.Option("out", help="output directory for findings/policies/PRs"), config: str = typer.Option(None, "--config", help="path to agents.yaml"), min_confidence: float = typer.Option(0.5, "--min-confidence", help="drop verified findings below this confidence"), @@ -51,13 +52,15 @@ def scan( With --cve the input is a security advisory instead of a repo: the advisory is resolved from OSV.dev, an agent derives its HTTP exploitation profile (or says there isn't one), and the result enters the same triage and generate stages.""" - if bool(repo) == bool(cve): - raise typer.BadParameter("pass a repo path or --cve, not " + ("both" if repo else "neither")) + if cve and (repo or spec): + raise typer.BadParameter("--cve scans one advisory; it cannot be combined with a repo or --spec") + if not (repo or cve or spec): + raise typer.BadParameter("pass a repo path, --cve, or --spec") if code_fixes is None: # match the console default (app.py /api/defaults) so headless == UI code_fixes = os.environ.get("VPCOPILOT_SCAN_REMEDIATE", "1").lower() not in ("0", "false", "no") summary = run_pipeline(repo, out_dir=out, config_path=config, min_confidence=min_confidence, concurrency=concurrency, max_files=max_files, max_bytes=max_bytes, - draft_code_fixes=code_fixes, advisory=cve, + draft_code_fixes=code_fixes, advisory=cve, spec_path=spec, log=lambda m: rprint(f"[dim]{m}[/dim]")) rprint(Panel.fit( "\n".join(f"[bold]{k}[/bold]: {v}" for k, v in summary.items()), diff --git a/src/vpcopilot/console/app.py b/src/vpcopilot/console/app.py index 92daba9..42f21ea 100644 --- a/src/vpcopilot/console/app.py +++ b/src/vpcopilot/console/app.py @@ -594,6 +594,7 @@ class ScanReq(BaseModel): # yields "". Every existing payload stays valid and nothing can 422. repo: str = "" cve: str = "" + spec: str = "" out: str = "out" min_confidence: float = 0.5 max_files: int = 200 @@ -603,14 +604,15 @@ class ScanReq(BaseModel): def _run_scan(repo: str, out: str, min_confidence: float = 0.5, max_files: int = 200, max_bytes: int = 60_000, draft_code_fixes: bool = True, - cve: str = ""): + cve: str = "", spec: str = ""): _scan.update(state="running", log=[], summary=None, error=None) try: from ..pipeline import run_pipeline summary = run_pipeline(repo or None, out_dir=out, config_path=_active_config, min_confidence=min_confidence, max_files=max_files, max_bytes=max_bytes, draft_code_fixes=draft_code_fixes, - advisory=cve or None, log=lambda m: _append(_scan["log"], m)) + advisory=cve or None, spec_path=spec or None, + log=lambda m: _append(_scan["log"], m)) _scan.update(state="done", summary=summary) except Exception as e: # noqa: BLE001 _scan.update(state="error", error=str(e)) @@ -624,9 +626,10 @@ def start_scan(body: ScanReq): # The console reads results from OUT — so point OUT at the dir this scan writes to, or Review / # Mitigate would read a different (empty) dir. Makes the Output-dir field authoritative even when # it differs from the model-switcher default (e.g. out-claude-vampi). - if bool(body.repo.strip()) == bool(body.cve.strip()): - raise HTTPException(400, "give a repo path or a CVE/GHSA id, not " - + ("both" if body.repo.strip() else "neither")) + if body.cve.strip() and (body.repo.strip() or body.spec.strip()): + raise HTTPException(400, "a CVE scan cannot be combined with a repo or a spec") + if not (body.repo.strip() or body.cve.strip() or body.spec.strip()): + raise HTTPException(400, "give a repo path, a CVE/GHSA id, or an OpenAPI spec") global OUT OUT = Path(body.out) # kwargs, not a positional tuple: H2/H3 add more inputs here and a positional args tuple is one @@ -634,7 +637,8 @@ def start_scan(body: ScanReq): threading.Thread(target=_run_scan, daemon=True, kwargs=dict(repo=body.repo, out=body.out, min_confidence=body.min_confidence, max_files=body.max_files, max_bytes=body.max_bytes, - draft_code_fixes=body.draft_code_fixes, cve=body.cve)).start() + draft_code_fixes=body.draft_code_fixes, cve=body.cve, + spec=body.spec)).start() return {"state": "running", "out": str(OUT)} diff --git a/src/vpcopilot/console/static/index.html b/src/vpcopilot/console/static/index.html index 2c6442f..2e19b0b 100644 --- a/src/vpcopilot/console/static/index.html +++ b/src/vpcopilot/console/static/index.html @@ -149,6 +149,8 @@ + +
@@ -866,12 +868,13 @@

Full matrix

${head} // ---- scan (auto-advances to Review on done) ---- let SCAN_N = 0; // lines already appended — the server streams the transcript from here scanLog.addEventListener("scroll",()=>{ scanFollow.style.display = atBottom(scanLog) ? "none" : "inline-block"; }); -async function runScan(){ const repo=scanRepo.value.trim(), cve=scanCve.value.trim(); - // Exactly one input — the server enforces this too; this is just a faster no. - if(!repo && !cve){alert("enter a repo path, or a CVE/GHSA id to scan an advisory instead");return;} - if(repo && cve){alert("give a repo path OR an advisory id, not both");return;} +async function runScan(){ const repo=scanRepo.value.trim(), cve=scanCve.value.trim(), spec=scanSpec.value.trim(); + // repo and --cve are alternatives; a spec is additive. The server enforces this too — this is + // just a faster no. + if(!repo && !cve && !spec){alert("enter a repo path, a CVE/GHSA id, or an OpenAPI spec");return;} + if(cve && (repo || spec)){alert("a CVE scan cannot be combined with a repo or a spec");return;} scanState.textContent="starting…"; - try { await jpost("/api/scan",{repo,cve,out:scanOut.value.trim()||"out",min_confidence:parseFloat(scanMinConf.value)||0.5,max_files:parseInt(scanMaxFiles.value)||200,max_bytes:parseInt(scanMaxBytes.value)||60000,draft_code_fixes:scanRemediate.checked}); } catch(e){ scanState.textContent="error: "+e.message; return; } + try { await jpost("/api/scan",{repo,cve,spec,out:scanOut.value.trim()||"out",min_confidence:parseFloat(scanMinConf.value)||0.5,max_files:parseInt(scanMaxFiles.value)||200,max_bytes:parseInt(scanMaxBytes.value)||60000,draft_code_fixes:scanRemediate.checked}); } catch(e){ scanState.textContent="error: "+e.message; return; } SCAN_N=0; scanLog.textContent=""; scanLogCount.textContent=""; scanLogWrap.style.display="block"; const poll=setInterval(async()=>{ let s; try { s=await jget("/api/scan?since="+SCAN_N); } catch(e){ return; } // transient fetch blip — keep polling diff --git a/src/vpcopilot/inputs/openapi.py b/src/vpcopilot/inputs/openapi.py new file mode 100644 index 0000000..a473abb --- /dev/null +++ b/src/vpcopilot/inputs/openapi.py @@ -0,0 +1,223 @@ +"""H3 — an OpenAPI spec as a discovery input. + +A4 goes one way: the operator hands over a schema and XC enforces it. This is the other direction — +read the spec and find the flaws *in* it. A spec is a security artifact whether anyone treats it as +one: it is where `amount` is declared with no lower bound, where a path takes an object id with no +authorization statement, where `additionalProperties` is left open so mass assignment is legal by +declaration. Those are findings before a line of code is read, and the spec is often the only thing +you have for a service you do not own. + +Two halves, deliberately split: + +* **Code** does the structural pass — a numeric field with no `minimum`, a string with no + `maxLength`, an operation with no `security`, an object that accepts extra properties. These are + facts about the document, they need no judgement, and deriving them deterministically means recall + does not depend on which model is configured. +* **The agent** decides which of those facts matter, and writes the exploit sketch. "No `minimum` on + `amount` in a transfer" is a business-logic hole; the same omission on `page` is nothing. That is + the judgement call, and it is the only part a model should be making. + +The **orphan** comparison is pure code and needs both inputs: an endpoint declared in the spec that +no route in the repo serves is either dead documentation or a shadow API, and a route the code +serves that the spec never mentions is what an api_schema band-aid would silently start blocking. +Neither is a vulnerability by itself; both are the kind of drift that produces one. +""" +from __future__ import annotations + +from pathlib import Path +from typing import Any + +HTTP_VERBS = ("get", "put", "post", "delete", "patch", "head", "options", "trace") + +# Fields whose name implies a value where "unbounded" is a real hazard rather than a nitpick. Used +# only to RANK the structural findings — the agent still decides what matters. +_MONEY_ISH = ("amount", "price", "total", "balance", "quantity", "qty", "count", "limit", "size") + + +def load_spec(path: str) -> dict: + """Parse an OpenAPI/Swagger document. YAML or JSON — `yaml.safe_load` handles both.""" + p = Path(path) + if not p.is_file(): + raise RuntimeError(f"no OpenAPI spec at {path}") + import yaml + try: + data = yaml.safe_load(p.read_text(errors="replace")) + except Exception as e: # noqa: BLE001 + raise RuntimeError(f"could not parse {path} as YAML or JSON: {e}") from None + if not isinstance(data, dict): + raise RuntimeError(f"{path} is not an OpenAPI document (top level is not a mapping)") + if not (data.get("openapi") or data.get("swagger")): + raise RuntimeError(f"{path} has no top-level `openapi`/`swagger` version — is it a spec?") + if not isinstance(data.get("paths"), dict) or not data["paths"]: + raise RuntimeError(f"{path} declares no `paths` — nothing to analyse") + return data + + +def _resolve(spec: dict, node: Any, _depth: int = 0) -> Any: + """Follow a local `$ref` one level at a time. Bounded: a spec with a cyclic ref is a real thing + and must not hang a scan.""" + if _depth > 8 or not isinstance(node, dict): + return node if isinstance(node, dict) else {} + ref = node.get("$ref") + if not isinstance(ref, str) or not ref.startswith("#/"): + return node + cur: Any = spec + for part in ref[2:].split("/"): + if not isinstance(cur, dict) or part not in cur: + return {} + cur = cur[part] + return _resolve(spec, cur, _depth + 1) + + +def operations(spec: dict) -> list[dict]: + """Every declared operation, flattened, with its parameters and request-body schema resolved.""" + base = (spec.get("basePath") or "").rstrip("/") + out = [] + for raw_path, item in (spec.get("paths") or {}).items(): + if not isinstance(item, dict): + continue + shared = item.get("parameters") or [] + for method, op in item.items(): + if method.lower() not in HTTP_VERBS or not isinstance(op, dict): + continue + body = {} + rb = _resolve(spec, op.get("requestBody") or {}) + for _ct, media in (rb.get("content") or {}).items(): + body = _resolve(spec, (media or {}).get("schema") or {}) + break + if not body and op.get("parameters") is None and "schema" in op: # swagger 2 style + body = _resolve(spec, op.get("schema") or {}) + out.append({ + "path": f"{base}{raw_path}" if base else raw_path, + "method": method.upper(), + "operation_id": op.get("operationId") or "", + "summary": op.get("summary") or op.get("description") or "", + "parameters": [_resolve(spec, p) for p in (list(shared) + list(op.get("parameters") or []))], + "body_schema": body, + "security": op.get("security", spec.get("security")), + }) + return out + + +def _walk_properties(spec: dict, schema: dict, prefix: str = "", _depth: int = 0): + """Yield `(dotted_name, subschema)` for every property, following $ref and arrays.""" + if _depth > 6: + return + schema = _resolve(spec, schema) + for name, sub in (schema.get("properties") or {}).items(): + sub = _resolve(spec, sub) + dotted = f"{prefix}{name}" + yield dotted, sub + if sub.get("type") == "object" or sub.get("properties"): + yield from _walk_properties(spec, sub, f"{dotted}.", _depth + 1) + if sub.get("type") == "array": + items = _resolve(spec, sub.get("items") or {}) + if items.get("properties"): + yield from _walk_properties(spec, items, f"{dotted}[].", _depth + 1) + + +def weak_constraints(spec: dict) -> list[dict]: + """The structural pass: what the document leaves unbounded or unstated. + + Deterministic on purpose. Recall here does not depend on which model is configured — the agent's + job is to say which of these is worth a band-aid, not to find them.""" + issues = [] + for op in operations(spec): + where = f"{op['method']} {op['path']}" + if op["security"] is not None and op["security"] == []: + issues.append({"kind": "no_auth", "where": where, "field": "", + "detail": "declares `security: []` — explicitly unauthenticated"}) + elif op["security"] is None: + issues.append({"kind": "no_auth", "where": where, "field": "", + "detail": "no `security` declared and no global default"}) + for p in op["parameters"]: + sch = _resolve(spec, p.get("schema") or p) + issues += _field_issues(where, p.get("name") or "", sch, "parameter") + body = op["body_schema"] + if body: + if body.get("additionalProperties") is True or ( + body.get("type") == "object" and "additionalProperties" not in body): + issues.append({"kind": "open_object", "where": where, "field": "(body)", + "detail": "the request body accepts undeclared properties — mass " + "assignment is legal by declaration"}) + for name, sub in _walk_properties(spec, body): + issues += _field_issues(where, name, sub, "body field") + return issues + + +def _field_issues(where: str, name: str, sch: dict, kind: str) -> list[dict]: + if not isinstance(sch, dict): + return [] + t, out = sch.get("type"), [] + money = any(w in (name or "").lower() for w in _MONEY_ISH) + if t in ("number", "integer"): + if "minimum" not in sch and "exclusiveMinimum" not in sch: + out.append({"kind": "unbounded_number", "where": where, "field": name, + "detail": f"{kind} `{name}` is a {t} with no `minimum`" + + (" — a negative value is legal by declaration" if money else ""), + "notable": money}) + if "maximum" not in sch and "exclusiveMaximum" not in sch: + out.append({"kind": "unbounded_number", "where": where, "field": name, + "detail": f"{kind} `{name}` is a {t} with no `maximum`", "notable": money}) + if t == "string" and "maxLength" not in sch and not sch.get("enum") and not sch.get("format"): + out.append({"kind": "unbounded_string", "where": where, "field": name, + "detail": f"{kind} `{name}` is a string with no `maxLength`, `enum` or `format`", + "notable": False}) + if t == "array" and "maxItems" not in sch: + out.append({"kind": "unbounded_array", "where": where, "field": name, + "detail": f"{kind} `{name}` is an array with no `maxItems`", "notable": False}) + return out + + +def _normalize(path: str) -> str: + """`/users/{id}/pay` and `/users/:id/pay` and `/users/[id]/pay` are the same endpoint.""" + import re + p = re.sub(r"\{[^}]*\}|\[[^\]]*\]|:[A-Za-z_][A-Za-z0-9_]*", "{}", path or "") + return "/" + p.strip("/").lower() + + +def orphans(spec: dict, repo_routes: list[str]) -> dict: + """Spec vs code, both directions. Pure comparison — no model, no network. + + `spec_only`: declared but nothing serves it — dead documentation, or a shadow API someone + forgot to remove. `code_only`: served but undeclared — and this is the one that bites, because + an `api_schema` band-aid built from this spec would start rejecting those routes the moment it + is applied.""" + declared = {_normalize(op["path"]): op["path"] for op in operations(spec)} + served = {} + for r in repo_routes or []: + # route context lines look like "GET POST /users/v1/register" or "/users/v1/register" + parts = r.split() + path = parts[-1] if parts else "" + if path.startswith("/"): + served[_normalize(path)] = path + return { + "spec_only": sorted(declared[k] for k in declared.keys() - served.keys()), + "code_only": sorted(served[k] for k in served.keys() - declared.keys()), + "matched": sorted(declared[k] for k in declared.keys() & served.keys()), + } + + +def as_scan_input(spec_path: str) -> tuple[str, str]: + """The spec rendered for the `discover` agent: `(relative_name, numbered_text)`. + + Feeding it through `discover` rather than adding a fourth agent is deliberate — the spec is a + file with content, discover's whole job is "read this and tell me what is wrong with it", and + routing it through the existing stage means H3's findings verify, triage and generate exactly + like a code finding with no new path to keep in step.""" + spec = load_spec(spec_path) + issues = weak_constraints(spec) + raw = Path(spec_path).read_text(errors="replace") + lines = raw.splitlines() + numbered = "\n".join(f"{i + 1:4}: {ln}" for i, ln in enumerate(lines[:1200])) + hints = "\n".join(f" - [{i['kind']}] {i['where']} {i['detail']}" + for i in sorted(issues, key=lambda x: (not x.get("notable"), x["kind"]))[:60]) + header = ( + "# This is an OpenAPI SPECIFICATION, not source code. Report flaws IN THE CONTRACT: " + "missing bounds on values that need them, operations with no authorization, bodies that " + "accept undeclared properties, ids in paths with no ownership check.\n" + "# A deterministic structural pass already found the following. Decide which are " + "SECURITY-RELEVANT — an unbounded `page` is nothing, an unbounded `amount` on a transfer " + "is a business-logic hole — and ignore the rest.\n" + f"{hints}\n\n") + return Path(spec_path).name, header + numbered diff --git a/src/vpcopilot/pipeline.py b/src/vpcopilot/pipeline.py index ede2402..cefc233 100644 --- a/src/vpcopilot/pipeline.py +++ b/src/vpcopilot/pipeline.py @@ -66,13 +66,17 @@ def run_pipeline( draft_code_fixes: bool = True, # off = skip remediation (band-aids only); saves ~half the tokens log: Callable[[str], None] = print, advisory: str | None = None, # H1: appended AFTER log so no positional call can shift + spec_path: str | None = None, # H3: an OpenAPI spec — alone, or alongside a repo ) -> dict: # H1 — exactly one input. Deliberately a hard error rather than a silent no-op: today # `run_pipeline("/does/not/exist")` completes and writes a full set of empty artifacts, and # that is the failure mode not to extend. - if bool(repo_path) == bool(advisory): - raise ValueError("pass a repo path or an advisory id (--cve), not " - + ("both" if repo_path else "neither")) + # H1/H3 — repo and advisory are alternatives; a spec is additive. `--spec` alone scans the + # contract; `--spec` with a repo also cross-checks the two for orphans, which needs both. + if advisory and (repo_path or spec_path): + raise ValueError("--cve scans one advisory; it cannot be combined with a repo or --spec") + if not (repo_path or advisory or spec_path): + raise ValueError("pass a repo path, an advisory id (--cve), or an OpenAPI spec (--spec)") h = Harness(config_path) h.warmup() # B6: warm instructor's mode registry before ANY fan-out — both inputs need it t0, started = time.perf_counter(), runmeta.utc_now() @@ -83,6 +87,8 @@ def run_pipeline( file_raw: dict[str, str] = {} route_ctx = None findings: list = [] + spec_code: dict[str, str] = {} # H3: spec name -> the text handed to discover + spec_orphans: dict | None = None forced_decision = forced_remediation = None advisory_meta: dict | None = None @@ -101,7 +107,8 @@ def run_pipeline( "fixed_version": res["remediation"].fixed_version} discover_s = time.perf_counter() - t0 else: - files, skipped = collect_files(repo_path, max_bytes=max_bytes, max_files=max_files) + if repo_path: + files, skipped = collect_files(repo_path, max_bytes=max_bytes, max_files=max_files) log(f"scanning {len(files)} files (caps: --max-files {max_files}, --max-bytes {max_bytes}; " f"{len(skipped)} skipped)") for reason in ("max-files-reached", "too-large"): @@ -111,7 +118,21 @@ def run_pipeline( # Ground endpoints in the app's DECLARED routes (OpenAPI spec / framework registrations) so a # weaker model looks a finding's path up instead of hallucinating it — and warn loudly if none. - route_ctx = collect_route_context(repo_path) + route_ctx = collect_route_context(repo_path) if repo_path else None + if spec_path: + # H3 — the spec is fed to `discover` as one more file rather than through a fourth + # agent. Its whole job is "read this and tell me what is wrong with it", and routing + # the spec through the existing stage means these findings verify, triage and generate + # exactly like a code finding, with no parallel path to keep in step. + from .inputs.openapi import as_scan_input, load_spec, orphans + spec_name, spec_text = as_scan_input(spec_path) + spec_code[spec_name] = spec_text + log(f"reading OpenAPI spec {spec_path} as a discovery input") + if repo_path and route_ctx: + o = orphans(load_spec(spec_path), route_ctx.splitlines()) + spec_orphans = o + log(f" spec vs code: {len(o['matched'])} matched, " + f"{len(o['spec_only'])} declared-but-unserved, {len(o['code_only'])} undeclared") if route_ctx: log("route context: found the app's declared routes — grounding finding endpoints (no guessing)") else: @@ -136,12 +157,29 @@ def _discover(p): if files: with ThreadPoolExecutor(max_workers=concurrency) as ex: disc_results.extend(ex.map(_discover, files)) + for spec_name, spec_text in spec_code.items(): + # The spec joins the same result list, so it flows through verify/triage/generate unchanged. + try: + disc_results.append((spec_name, spec_text, spec_text, + discover.run(h, spec_name, spec_text, route_context=route_ctx))) + except Exception as e: # noqa: BLE001 — a bad spec must not kill a repo scan alongside it + from .schemas import FindingList + log(f" ⚠ discover failed on {spec_name}: {e} — skipping the spec") + disc_results.append((spec_name, spec_text, spec_text, FindingList(findings=[]))) used_ids: set[str] = set() # A4: the pipeline owns finding ids — a model may reuse one across files for rel, code, raw, res in disc_results: file_code[rel] = code file_raw[rel] = raw for f in res.findings: - f.file = rel + if rel in spec_code: + # H3 — a spec is ONE file declaring MANY endpoints, so `file` is the wrong identity + # for it: `endpoint_of("pay-api.yaml")` is the same string for every finding, and + # all six would collapse onto one service_policy coverage key with five of them + # logged "already covered". Leave `file` empty and let the endpoint be the identity, + # exactly as an advisory does. + f.source, f.file, f.line = f"openapi:{rel}", "", 0 + else: + f.file = rel base, fid, n = (f.id or "finding"), (f.id or "finding"), 1 while fid in used_ids: n += 1 @@ -194,6 +232,43 @@ def _threshold(f): else: refuted += 1 log(f" verify {f.id}: refuted ({v.confidence:.2f})") + if spec_orphans and (spec_orphans["spec_only"] or spec_orphans["code_only"]): + # Deterministic, no agent: this is a comparison of two documents. Both directions matter and + # they fail differently — a declared-but-unserved endpoint is dead documentation or a shadow + # API, while a served-but-undeclared route is what an api_schema band-aid built from this + # spec would start rejecting the moment it is applied. + from .schemas import Finding + so, co = spec_orphans["spec_only"], spec_orphans["code_only"] + bits = [] + if so: + bits.append("declared in the spec but served by no route in the code: " + + ", ".join(so[:15]) + (f" (+{len(so) - 15} more)" if len(so) > 15 else "")) + if co: + bits.append("served by the code but absent from the spec: " + + ", ".join(co[:15]) + (f" (+{len(co) - 15} more)" if len(co) > 15 else "")) + orphan = Finding( + id="undocumented_or_orphaned", + title="Spec and code disagree about which endpoints exist", + vuln_class="other", severity="medium", source=f"openapi:{Path(spec_path).name}", + description="; ".join(bits) + ". " + + ("An endpoint nobody serves is dead documentation or a shadow API that was removed " + "from one place and not the other. " if so else "") + + ("An endpoint the spec does not declare is outside whatever a schema-based control " + "enforces — and applying an api_schema band-aid built from this spec would begin " + "rejecting it." if co else ""), + exploit_sketch=("Undeclared routes are the surface an attacker reaches that a " + "schema-based control never sees." if co + else "Declared-but-unserved endpoints indicate the spec and the " + "deployment have drifted; the schema may not describe what is " + "actually exposed.")) + # Appended AFTER verify, to both lists: this is a comparison of two documents, not a claim + # about code. Sending it through the verify agent asked a reviewer to confirm a + # vulnerability in source it cannot see, and it duly refuted it at 0.10 confidence. + findings.append(orphan) + verified.append(orphan) + log(f" undocumented_or_orphaned: {len(so)} declared-but-unserved, " + f"{len(co)} served-but-undeclared") + if advisory: verified = list(findings) verify_s = time.perf_counter() - t_verify @@ -275,7 +350,7 @@ def _probe(f): exploit, legit = pr.get("exploit"), pr.get("legit") for b in [b for b in d.bandaids if b.recommended] or d.bandaids: key = correlate.coverage_key(b.control.value, f.file, - identity=getattr(f, "source", "") or f.id) + identity=f.endpoint or getattr(f, "source", "") or f.id) if key in seen_keys: correlations.append({"finding_id": d.finding_id, "control": b.control.value, "covered_by": seen_keys[key], "coverage_key": key}) @@ -339,7 +414,10 @@ def _remediate(f): cfg = getattr(h, "cfg", None) runmeta.write_manifest( out_dir, repo=str(root.resolve()) if root else None, advisory=advisory_meta, - input_kind="advisory" if advisory else "repo", + input_kind=("advisory" if advisory else + ("repo+spec" if (repo_path and spec_path) else + ("spec" if spec_path else "repo"))), + spec=str(Path(spec_path).resolve()) if spec_path else None, config_path=config_path, started=started, models={a: cfg.for_agent(a).model for a in AGENT_NAMES} if cfg else None, caps={"min_confidence": min_confidence, "max_files": max_files, "max_bytes": max_bytes, diff --git a/tests/test_inputs_cve.py b/tests/test_inputs_cve.py index 096716f..329efb4 100644 --- a/tests/test_inputs_cve.py +++ b/tests/test_inputs_cve.py @@ -240,12 +240,15 @@ def run(self, *a, **k): return prof # ---- the pipeline entry, correlation identity, and the PR decline ---- -def test_the_pipeline_refuses_both_or_neither_input(): +def test_the_pipeline_refuses_a_missing_or_conflicting_input(): + """repo and --cve are alternatives; --spec is additive (H3) and may accompany a repo.""" from vpcopilot.pipeline import run_pipeline - with pytest.raises(ValueError, match="not neither"): + with pytest.raises(ValueError, match="pass a repo path"): run_pipeline() - with pytest.raises(ValueError, match="not both"): + with pytest.raises(ValueError, match="cannot be combined"): run_pipeline("/some/repo", advisory="CVE-2024-23334") + with pytest.raises(ValueError, match="cannot be combined"): + run_pipeline(advisory="CVE-2024-23334", spec_path="/some/spec.yaml") def test_a_file_less_finding_gets_its_own_coverage_key(): diff --git a/tests/test_inputs_openapi.py b/tests/test_inputs_openapi.py new file mode 100644 index 0000000..5208419 --- /dev/null +++ b/tests/test_inputs_openapi.py @@ -0,0 +1,175 @@ +"""H3 — an OpenAPI spec as a discovery input. Offline: no model, no network. + +The structural pass is deterministic on purpose, so it is tested directly; the agent's judgement +about which of those facts matter is not something a unit test can pin.""" +import pytest + +from vpcopilot.inputs import openapi as oa + +SPEC = """ +openapi: 3.0.3 +info: {title: t, version: "1"} +paths: + /api/pay: + post: + security: [] + requestBody: + content: + application/json: + schema: + type: object + additionalProperties: true + properties: + amount: {type: number} + memo: {type: string, maxLength: 100} + page: {type: integer, minimum: 0, maximum: 500} + /api/users/{id}: + get: + parameters: [{name: id, in: path, schema: {type: string, format: uuid}}] + security: [{bearer: []}] +""" + + +def _write(tmp_path, text=SPEC, name="spec.yaml"): + p = tmp_path / name + p.write_text(text) + return str(p) + + +def test_operations_are_flattened_with_their_methods(tmp_path): + ops = oa.operations(oa.load_spec(_write(tmp_path))) + assert {(o["method"], o["path"]) for o in ops} == { + ("POST", "/api/pay"), ("GET", "/api/users/{id}")} + + +def test_a_number_with_no_minimum_is_found_deterministically(tmp_path): + """The acceptance case. Recall here must not depend on which model is configured.""" + issues = oa.weak_constraints(oa.load_spec(_write(tmp_path))) + amt = [i for i in issues if i["field"] == "amount" and "minimum" in i["detail"]] + assert amt and amt[0]["notable"] is True # a money-ish name is ranked up + assert amt[0]["where"] == "POST /api/pay" + + +def test_a_bounded_field_is_not_flagged(tmp_path): + issues = oa.weak_constraints(oa.load_spec(_write(tmp_path))) + assert not [i for i in issues if i["field"] == "page"] + assert not [i for i in issues if i["field"] == "memo"] # maxLength present + assert not [i for i in issues if i["field"] == "id"] # format present + + +def test_an_open_object_and_a_missing_security_are_both_reported(tmp_path): + issues = oa.weak_constraints(oa.load_spec(_write(tmp_path))) + kinds = {(i["kind"], i["where"]) for i in issues} + assert ("open_object", "POST /api/pay") in kinds + assert ("no_auth", "POST /api/pay") in kinds + assert ("no_auth", "GET /api/users/{id}") not in kinds # it declares bearer + + +def test_refs_are_followed(tmp_path): + spec = """ +openapi: 3.0.3 +info: {title: t, version: "1"} +components: + schemas: + Xfer: {type: object, additionalProperties: false, properties: {amount: {type: number}}} +paths: + /p: + post: + security: [{b: []}] + requestBody: {content: {application/json: {schema: {$ref: '#/components/schemas/Xfer'}}}} +""" + issues = oa.weak_constraints(oa.load_spec(_write(tmp_path, spec))) + assert [i for i in issues if i["field"] == "amount"] + + +def test_a_cyclic_ref_does_not_hang(tmp_path): + spec = """ +openapi: 3.0.3 +info: {title: t, version: "1"} +components: {schemas: {A: {$ref: '#/components/schemas/A'}}} +paths: + /p: {post: {security: [{b: []}], requestBody: {content: {application/json: {schema: {$ref: '#/components/schemas/A'}}}}}} +""" + assert oa.weak_constraints(oa.load_spec(_write(tmp_path, spec))) is not None + + +@pytest.mark.parametrize("text,msg", [ + ("just: a mapping", "no top-level"), + ("openapi: 3.0.0\ninfo: {}", "declares no `paths`"), + ("- not\n- a mapping", "not an OpenAPI document"), +]) +def test_a_document_that_is_not_a_spec_says_so(tmp_path, text, msg): + with pytest.raises(RuntimeError, match=msg): + oa.load_spec(_write(tmp_path, text)) + + +def test_a_missing_file_says_so(): + with pytest.raises(RuntimeError, match="no OpenAPI spec at"): + oa.load_spec("/nope/nothing.yaml") + + +# ---- the orphan comparison ---- +def test_orphans_compare_both_directions(tmp_path): + o = oa.orphans(oa.load_spec(_write(tmp_path)), ["POST /api/pay", "GET /api/legacy"]) + assert o["matched"] == ["/api/pay"] + assert o["spec_only"] == ["/api/users/{id}"] + assert o["code_only"] == ["/api/legacy"] + + +@pytest.mark.parametrize("declared,served", [ + ("/users/{id}/pay", "/users/:id/pay"), + ("/users/{userId}", "/users/[id]"), + ("/API/Pay", "/api/pay"), + ("/api/pay/", "/api/pay"), +]) +def test_path_parameter_styles_are_treated_as_the_same_endpoint(tmp_path, declared, served): + spec = f"openapi: 3.0.3\ninfo: {{title: t, version: '1'}}\npaths:\n {declared}:\n get: {{}}\n" + o = oa.orphans(oa.load_spec(_write(tmp_path, spec)), [f"GET {served}"]) + assert o["matched"] and not o["spec_only"] and not o["code_only"] + + +def test_the_scan_input_carries_the_structural_findings_as_hints(tmp_path): + """The agent gets the deterministic pass as context, so it judges rather than hunts.""" + name, text = oa.as_scan_input(_write(tmp_path)) + assert name == "spec.yaml" + assert "OpenAPI SPECIFICATION, not source code" in text + assert "unbounded_number" in text and "amount" in text + assert " 1: " in text # and the spec itself, numbered + + +def test_the_spec_input_ranks_notable_fields_first(tmp_path): + _, text = oa.as_scan_input(_write(tmp_path)) + hints = [ln for ln in text.splitlines() if ln.startswith(" - [")] + assert "amount" in hints[0] + + +# ---- the pipeline wiring ---- +def test_spec_findings_key_on_their_endpoint_not_the_spec_filename(): + """A spec is ONE file declaring MANY endpoints. Keying on the file made + `endpoint_of("pay-api.yaml")` the same string for every finding, so all of them collapsed onto + one service_policy coverage key and all but the first were logged "already covered" — silently + generating no band-aid. Observed on a real spec scan.""" + from vpcopilot.correlate import coverage_key + a = coverage_key("service_policy", "", identity="/api/v1/transfer") + b = coverage_key("service_policy", "", identity="/api/v1/admin/reset") + assert a != b + # two findings on the SAME endpoint still share one band-aid, which is the point of B1 + assert coverage_key("service_policy", "", identity="/api/v1/transfer") == a + + +def test_a_spec_may_accompany_a_repo_but_not_an_advisory(): + from vpcopilot.pipeline import run_pipeline + with pytest.raises(ValueError, match="cannot be combined"): + run_pipeline(advisory="CVE-2024-23334", spec_path="/s.yaml") + with pytest.raises(ValueError, match="pass a repo path"): + run_pipeline() + + +def test_the_console_scan_surface_accepts_a_spec(): + from fastapi.testclient import TestClient + + from vpcopilot.console import app as A + c = TestClient(A.app) + assert c.post("/api/scan", json={"cve": "CVE-2024-23334", "spec": "/s.yaml"}).status_code == 400 + assert c.post("/api/scan", json={}).status_code == 400 + assert "spec" in A.ScanReq.model_fields