diff --git a/ROADMAP.md b/ROADMAP.md index efd6a84..a94becd 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -282,42 +282,53 @@ Input today is a source repo. The vulnerabilities most customers lose sleep over dependencies they do not own, where the code cure is a version bump someone else has to ship. That is the case virtual patching exists for, and the pipeline cannot see it. -- [ ] **H1** CVE and advisory input path. (M, P1) - Accept a CVE ID or GHSA identifier instead of a repo. An agent resolves the advisory into - an exploitation profile of affected paths, parameters, headers, and request shapes, and - that profile enters the existing triage and generate stages unchanged. - - Acceptance: a known path-traversal CVE in a web framework produces a `waf` or - `service_policy` band-aid; an advisory with no network-observable exploitation pattern - routes to `no_bandaid` with residual risk stated; `remediate` recommends the fixed - version rather than drafting a patch to vendor code; the ledger seeds `found` the same - way a repo finding does. - - Surfaces: `src/vpcopilot/inputs/cve.py`, a `resolve` agent in `agents/`, - `vpcopilot scan --cve CVE-YYYY-NNNNN`. - - **Advisory source (decided 2026-07-27): OSV.dev primary, GHSA for enrichment.** OSV needs no - auth, spans ecosystems on one schema, and returns affected ranges **and the fixed version** — - which is exactly what the acceptance needs for "recommend the fixed version rather than - drafting a patch to vendor code". It also keeps H1 runnable with no credentials, matching - `scan`'s "safe to run anywhere". GHSA (reusing the existing `GITHUB_TOKEN`) only for advisory - prose an agent reasons over; NVD is rejected — slow, rate-limited, imprecise version data. - Note what OSV does **not** give: the network-observable exploitation pattern (paths, params, - request shapes). Deriving that is the agent's job, and it is what makes the `no_bandaid` - branch of the acceptance meaningful. - - **Reconciled:** `src/vpcopilot/inputs/` does not exist — every module is flat under - `src/vpcopilot/` except `agents/` and `console/`. Creating a package is a new convention; - decide it deliberately or use `src/vpcopilot/input_cve.py`. A new `resolve` agent must also - be added to `config.AGENT_NAMES`, or it will be absent from `run.json` provenance, the - console's agent list, and the report's model chips. **The agent name is duplicated in three - places** — `config.AGENT_NAMES` (`config.py:16`, feeds `run.json` only), `AGENT_ROLES` - (`console/app.py:198`, drives `GET /api/agents`) and a hardcoded list in `report.py:251` - (drives the report's model chips) — all three need the same change. - - **Also touches an existing signature:** `scan`'s `repo` is a required positional - (`cli.py:36`) flowing into `run_pipeline(repo_path)` which does `Path(repo_path)` - (`pipeline.py:49-61`). `--cve` means making `repo` optional with mutual exclusion, an - alternate `run_pipeline` entry that does not walk a filesystem root, and the same optionality - on `ScanReq` / `POST /api/scan`. `RemediationPlan` (`schemas.py:122-131`) also *requires* - `file`, `diff` and `patched_content`, so "recommend the fixed version" needs either an - advisory-shaped remediation artifact or optional fields plus a `pr.py` branch that skips - `update_file`. +- [x] **H1** CVE and advisory input path. (M, P1) — **DONE:** `vpcopilot scan --cve CVE-2024-23334`. + `inputs/osv.py` fetches the advisory, the new `resolve` agent derives its HTTP exploitation + profile (or declines), and the result enters triage and generate unchanged. Verified live against + api.osv.dev and a real model. + - **Acceptance, as met (all four checked live):** CVE-2024-23334 (aiohttp path traversal) → + **both** a `waf` and a `service_policy` band-aid; GHSA-8r6j-v8pm-fqw3 (fsevents supply-chain) → + `no_bandaid` with the residual risk naming why a load balancer cannot see it; the cure reads + `upgrade aiohttp to 3.9.2` with `patched_content` empty; the ledger seeds `found` with severity, + band-aids and `has_cure` exactly as a repo finding does. + - **What querying OSV for real changed.** Three behaviours are not visible from the schema and + each silently degrades the answer: (a) asking for a **CVE id usually returns the GIT-range + record** — no package, and `fixed` values that are commit SHAs; the installable `PyPI/aiohttp + 3.9.2` only exists on the `GHSA-5h86-8mv2-jq9f` alias, so the client follows aliases (2 of 4 + advisories tested needed the hop); (b) a commit SHA is never offered as an upgrade target — + "upgrade to 24a6d649…" is not a recommendation; (c) `summary` is frequently empty and OS-level + CVEs have no package at all, only a CPE, with the human versions hidden in + `database_specific.extracted_events`. + - **Declining is the load-bearing behaviour.** An agent that invents a plausible path for every + CVE would make this input path worse than useless — confident band-aids that block nothing + while a real vulnerability hides behind a green check. So `network_observable=false` is a + first-class answer with a required, min-length `reason`, the agent is forbidden from choosing a + control or guessing a version, and its paths are cleared in code if it declines and lists them + anyway. The `no_bandaid` routing is **deterministic**, not delegated to triage. + - **The fixed version is never model-generated.** `remediate` is not called on this path at all; + `inputs/cve.py` builds the `RemediationPlan` from OSV. Drafting a patch against vendor code is + structurally impossible rather than merely discouraged. + - **Decisions:** `inputs/` **is** a package (H1/H2/H3 are three siblings of one shape and share + the OSV client — the same criterion that justifies `agents/`), with a one-directional rule that + nothing under it imports `pipeline`. `VulnClass` is **not** widened — a CWE→class table covers + the common cases and `other` plus a concrete `exploit_sketch` is honest; widening ripples into + every agent prompt and golden. No sentinel in `file`. + - **Identity, not a fake path.** New optional `Finding.source` (`osv:CVE-…`) carries what `file` + carries for a code finding. Two real bugs it fixes: `coverage_key` returned the plausible- + looking `service_policy:` for every file-less finding, so all but the first were logged + "already covered" and got **no band-aid at all**; and the dedup key `("", class, "L0")` merged + distinct advisories of the same class. Both take a defaulted `identity` fallback, so the repo + path is structurally unreachable and byte-identical. + - **Registration is four places, not the three the roadmap said** — `config.AGENT_NAMES`, + `console.AGENT_ROLES`, `report.py`, and **`bench_model.AGENTS`**, plus a `resolve:` block in + all four `config/agents*.yaml` (an unlisted agent silently falls back to the default model and + `run.json` records that as fact). + - **Fixed en route (pre-existing `pr.py` bugs):** the no-patch check sat *above* the dry-run + branch, so `--dry-run` raised identically to a live run and could preview nothing; and an empty + `file` would have reached `repo.get_contents("")` as a directory listing rather than erroring. + - **Deliberate consequence:** an advisory finding has no cure PR, so `reconcile` holds its + band-aid and escalates at TTL. Someone still has to ship the upgrade — documented, not a + surprise. - [ ] **H2** Dependency manifest input. (M, P2) Depends on H1. Parse `requirements.txt`, `package-lock.json`, and `pom.xml`, resolve advisories, and run diff --git a/config/agents.dgx.yaml b/config/agents.dgx.yaml index 7ac73c8..d72a493 100644 --- a/config/agents.dgx.yaml +++ b/config/agents.dgx.yaml @@ -26,6 +26,7 @@ defaults: mode: json agents: + resolve: {} discover: { model: openai/qwen3-coder:30b-a3b-q8_0 } # high-signal reading verify: { model: openai/qwen3-coder:30b-a3b-q8_0 } # adversarial refute triage: { model: openai/qwen3-coder:30b-a3b-q8_0 } # route to the strongest control diff --git a/config/agents.gemini.yaml b/config/agents.gemini.yaml index 93b1959..f3a306b 100644 --- a/config/agents.gemini.yaml +++ b/config/agents.gemini.yaml @@ -21,6 +21,7 @@ defaults: timeout: 180 agents: + resolve: {} discover: { model: gemini/gemini-3.1-pro-preview } verify: { model: gemini/gemini-3.1-pro-preview } triage: { model: gemini/gemini-3.1-pro-preview } diff --git a/config/agents.openai.yaml b/config/agents.openai.yaml index c63bc65..5ad6450 100644 --- a/config/agents.openai.yaml +++ b/config/agents.openai.yaml @@ -10,6 +10,7 @@ defaults: timeout: 180 agents: + resolve: {} discover: { model: openai/gpt-4.1 } verify: { model: openai/gpt-4.1 } triage: { model: openai/gpt-4.1 } diff --git a/config/agents.yaml b/config/agents.yaml index 27ce797..89adee1 100644 --- a/config/agents.yaml +++ b/config/agents.yaml @@ -16,6 +16,8 @@ defaults: max_retries: 3 agents: + resolve: + model: anthropic/claude-opus-4-8 # high-signal reading — keep this strong discover: model: anthropic/claude-opus-4-8 # high-signal reading — keep this strong verify: diff --git a/docs/USAGE.md b/docs/USAGE.md index bd37f19..25490d8 100644 --- a/docs/USAGE.md +++ b/docs/USAGE.md @@ -40,6 +40,51 @@ Runs `discover → verify → triage → generate → remediate` and writes to ` `findings.json`, `triage.json`, `policies/*.json` (XC specs), `remediations/*.patch|.pr.md` (code fixes), `correlations.json`, `ledger.json`, `summary.json`. No XC/GitHub writes. +### Scan a CVE instead of a repo (H1) + +The vulnerabilities most people lose sleep over live in dependencies they do not own, where the +code cure is a version bump someone else has to ship and they then have to deploy. That gap is what +virtual patching is for. + +```sh +vpcopilot scan --cve CVE-2024-23334 --out out # or GHSA-…, PYSEC-…, GO-…, RUSTSEC-… +``` + +The advisory is resolved from **OSV.dev** (no credentials — `scan` stays safe to run anywhere), the +`resolve` agent derives its HTTP exploitation profile, and the result enters the same triage and +generate stages as a code finding. `--cve` and a repo path are mutually exclusive. + +**The agent is expected to decline.** Many advisories cannot be virtually patched at a load +balancer — a malicious build-time dependency, a bug reachable only from a local file, memory +corruption with no request signature. Those route to `no_bandaid` with the residual risk stated, +and that routing is decided **in code**, not asked of the model: a hard requirement should not +depend on a prompt being honoured. An agent that obligingly invented a plausible path for every CVE +would be worse than no advisory input at all — it would produce confident band-aids that block +nothing while hiding a real vulnerability behind a green check. + +**The cure is a version bump, never a patch.** `remediate` is not called on this path. The fixed +version is copied from OSV by code — it is the one string an operator acts on directly, so no model +goes near it — and `vpcopilot pr` reports the upgrade and opens nothing: + +``` +advisory: upgrade aiohttp to 3.9.2 — no PR to open (the fix is upstream, not in this repo) +``` + +Because no cure PR exists, the band-aid is never auto-retired and **`reconcile` escalates it at TTL +expiry**. That is deliberate: someone still has to ship the upgrade. + +Three things about OSV worth knowing, each found by querying it: + +- Asking for a **CVE id often returns the git-range record** — no package, and `fixed` values that + are commit SHAs. The installable version lives on the GHSA/PYSEC alias, so the client follows + aliases. Without that, `CVE-2024-23334` recommends "upgrade to 24a6d649…". +- When there genuinely is no released fix, it says so rather than offering a commit. +- `summary` is often empty and OS-level CVEs have no package at all; the prose in `details` is the + real payload, and the CPE is the fallback identity. + +Set `VPCOPILOT_ADVISORY_CACHE=` to cache advisories on disk so a demo does not depend on the +network. + ## 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/agents/resolve.py b/src/vpcopilot/agents/resolve.py new file mode 100644 index 0000000..5a66455 --- /dev/null +++ b/src/vpcopilot/agents/resolve.py @@ -0,0 +1,76 @@ +"""Resolve agent — turn a security advisory into an exploitation profile, or decline. + +The facts (package, affected range, fixed version, CVSS, CWE) are fetched from OSV by code and +handed to this agent already assembled; it never touches the network and never produces a version +number. What it contributes is the one thing OSV does not carry: **what this vulnerability looks +like in an HTTP request** — which paths, which parameters, which headers. That is the input the +existing triage and generate stages need in order to propose a virtual patch. + +The most important thing this agent does is refuse. A great many advisories cannot be virtually +patched at a load balancer — a deserialization bug reachable only from a local file, a malicious +build-time dependency, a memory-corruption issue with no request signature. For those the honest +answer is `network_observable=false`, and the pipeline routes them to `no_bandaid` with the +residual risk stated. An agent that obligingly invents a plausible path for every CVE would make +the whole input path worse than useless: it would produce confident band-aids that block nothing +and hide vulnerabilities behind a green check.""" +from __future__ import annotations + +import json + +from ..harness import Harness +from ..schemas import ExploitationProfile + +SYSTEM = """You are a vulnerability analyst. Given a security advisory, describe how the +vulnerability manifests IN AN HTTP REQUEST — or state plainly that it cannot be seen in one. + +Set network_observable=true ONLY when the advisory describes something a proxy could identify +in a request: a specific URL path or pattern, a parameter, a header, a body shape, a method. +Then fill paths / http_methods / parameters / headers / example_requests with what the +advisory ACTUALLY SUPPORTS. + +Set network_observable=false — and this is a correct, expected, frequent answer — when: +- the advisory describes the flaw only in terms of internal functions, classes or config +- exploitation needs local file access, a malicious package at build time, or a crafted file +- it is memory corruption, a crypto weakness, or a denial of service with no request signature +- the text is too vague to name a path, parameter or header +- you are simply not sure + +HARD PROHIBITIONS. Violating any of these makes the output actively harmful: +- Do NOT invent a path, parameter or header the advisory does not name or clearly imply. +- Do NOT infer an exploitation pattern from what the package generally does. "It is a web + framework, so probably /admin" is exactly the reasoning that produces a band-aid which + blocks nothing while a real vulnerability stays open. +- Do NOT state or guess version numbers. Those are supplied to you and are not yours to edit. +- Do NOT choose a mitigating control. Never say WAF, service policy, rate limit or schema — + a later stage decides that. Describe the exploit only. +- Do NOT fabricate a file path, line number or code snippet. There is no source code here. + +`reason` is required and must justify your network_observable verdict by pointing at what the +advisory does or does not say. "Not enough information" with nothing further is not acceptable; +say WHAT is missing. + +`confidence` is your calibrated confidence in the profile: 0.9+ = the advisory names the request +shape explicitly; ~0.5 = you inferred it from a clear description; <0.3 = weak. When +network_observable is false, confidence expresses how sure you are that it CANNOT be observed. + +Paths are app-relative and contain no scheme or host (e.g. /static/../../etc/passwd).""" + + +def run(h: Harness, advisory: dict) -> ExploitationProfile: + """`advisory` is the normalized OSV record from `inputs.osv.resolve` — already fetched, so the + agent reasons over facts rather than retrieving them.""" + facts = {k: advisory.get(k) for k in + ("id", "aliases", "summary", "details", "cwe_ids", "cvss", "affected", "references")} + user = ( + f"ADVISORY (from OSV.dev, verbatim):\n{json.dumps(facts, indent=2)}\n\n" + "Describe how this is exploited over HTTP, or state that it cannot be observed in a " + "request. Remember that declining is a correct answer." + ) + prof = h.run("resolve", SYSTEM, user, ExploitationProfile) + prof.advisory_id = advisory.get("id") or prof.advisory_id # authoritative, as probe.py does + if not prof.network_observable: + # Belt and braces: the prompt forbids it, but a model that says "cannot be observed" and + # then lists paths anyway must not have those paths reach `generate`. + prof.paths, prof.http_methods = [], [] + prof.parameters, prof.headers, prof.example_requests = [], [], [] + return prof diff --git a/src/vpcopilot/bench_model.py b/src/vpcopilot/bench_model.py index 91722b8..537e1ff 100644 --- a/src/vpcopilot/bench_model.py +++ b/src/vpcopilot/bench_model.py @@ -13,7 +13,7 @@ from pathlib import Path SEV = ("critical", "high", "medium", "low") -AGENTS = ("discover", "verify", "triage", "generate", "remediate", "probe", "refine") +AGENTS = ("resolve", "discover", "verify", "triage", "generate", "remediate", "probe", "refine") def _rj(out_dir: str, name: str, default): diff --git a/src/vpcopilot/cli.py b/src/vpcopilot/cli.py index 2c2fe7c..a415348 100644 --- a/src/vpcopilot/cli.py +++ b/src/vpcopilot/cli.py @@ -33,7 +33,8 @@ def _root( @app.command() def scan( - repo: str = typer.Argument(..., help="path to the target application repo"), + 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-…"), 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"), @@ -45,12 +46,18 @@ def scan( help="also draft the code-fix PRs (default: on; env VPCOPILOT_SCAN_REMEDIATE=0 to default off). " "--no-code-fixes = band-aids only, saves ~half the tokens (use for band-aid benchmarks)"), ): - """Discover -> verify -> triage -> generate policies + code-fix PRs (read-only).""" + """Discover -> verify -> triage -> generate policies + code-fix PRs (read-only). + + 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 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, + draft_code_fixes=code_fixes, advisory=cve, 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/config.py b/src/vpcopilot/config.py index 1921fcc..988b4cc 100644 --- a/src/vpcopilot/config.py +++ b/src/vpcopilot/config.py @@ -13,7 +13,7 @@ # Every agent in the pipeline, in lifecycle order. Recorded per run so an audit export can say which # model produced each finding, band-aid and cure. -AGENT_NAMES = ("discover", "verify", "triage", "generate", "remediate", "probe", "refine") +AGENT_NAMES = ("resolve", "discover", "verify", "triage", "generate", "remediate", "probe", "refine") @dataclass diff --git a/src/vpcopilot/console/app.py b/src/vpcopilot/console/app.py index 8670364..92daba9 100644 --- a/src/vpcopilot/console/app.py +++ b/src/vpcopilot/console/app.py @@ -337,6 +337,7 @@ def audit_export(scope: str = "run"): AGENT_ROLES = { + "resolve": "read a security advisory → an HTTP exploitation profile (or decline)", "discover": "read source → candidate findings", "verify": "adversarially confirm or refute each finding", "triage": "route each finding to the strongest XC band-aid (or code-only)", @@ -589,7 +590,10 @@ def list_repos(): # ---------------- scan (background) ---------------- class ScanReq(BaseModel): - repo: str + # H1 — `str = ""` rather than `str | None`: index.html always sends `repo`, and an empty input + # yields "". Every existing payload stays valid and nothing can 422. + repo: str = "" + cve: str = "" out: str = "out" min_confidence: float = 0.5 max_files: int = 200 @@ -598,13 +602,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): + max_files: int = 200, max_bytes: int = 60_000, draft_code_fixes: bool = True, + cve: str = ""): _scan.update(state="running", log=[], summary=None, error=None) try: from ..pipeline import run_pipeline - summary = run_pipeline(repo, 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, - log=lambda m: _append(_scan["log"], m)) + 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)) _scan.update(state="done", summary=summary) except Exception as e: # noqa: BLE001 _scan.update(state="error", error=str(e)) @@ -618,12 +624,17 @@ 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")) global OUT OUT = Path(body.out) - threading.Thread(target=_run_scan, - args=(body.repo, body.out, body.min_confidence, body.max_files, body.max_bytes, - body.draft_code_fixes), - daemon=True).start() + # kwargs, not a positional tuple: H2/H3 add more inputs here and a positional args tuple is one + # reordering away from scanning the wrong thing. + 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() return {"state": "running", "out": str(OUT)} diff --git a/src/vpcopilot/console/static/index.html b/src/vpcopilot/console/static/index.html index 1c3524b..2c6442f 100644 --- a/src/vpcopilot/console/static/index.html +++ b/src/vpcopilot/console/static/index.html @@ -147,6 +147,8 @@

① Scan a repository

Read-only and safe — no XC or GitHub writes happen here. Point it at a vulnerable app (try crapi/VAmPI first — see docs/TRY_IT.md).

+ +
@@ -864,9 +866,12 @@

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(); if(!repo){alert("enter a repo path");return;} +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;} scanState.textContent="starting…"; - try { await jpost("/api/scan",{repo,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,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/correlate.py b/src/vpcopilot/correlate.py index 49adf7a..755a3f3 100644 --- a/src/vpcopilot/correlate.py +++ b/src/vpcopilot/correlate.py @@ -15,8 +15,18 @@ def endpoint_of(file: str) -> str: return parts[-2] if len(parts) >= 2 else (parts[0] if parts else file) -def coverage_key(control: str, file: str) -> str: - """Identity of the band-aid instance. Same key => one band-aid covers both findings.""" +def coverage_key(control: str, file: str, identity: str = "") -> str: + """Identity of the band-aid instance. Same key => one band-aid covers both findings. + + `identity` is the fallback for a finding with no repo file — an advisory (H1), a manifest entry + (H2), a spec path (H3). It is used verbatim, deliberately NOT through `endpoint_of`: that + returns the second-to-last path segment, which is right for a repo path (the last segment is a + filename) and wrong for a URL, where `/api/pay` and `/api/login` would both collapse to `api`. + + Without this, every file-less finding produced the same plausible-looking key `service_policy:` + and all but the first were logged "already covered" — silently generating no band-aid at all.""" if control in LB_WIDE: return control + if not file and identity: + return f"{control}:{identity}" return f"{control}:{endpoint_of(file)}" diff --git a/src/vpcopilot/inputs/__init__.py b/src/vpcopilot/inputs/__init__.py new file mode 100644 index 0000000..cbd7e18 --- /dev/null +++ b/src/vpcopilot/inputs/__init__.py @@ -0,0 +1,11 @@ +"""Input adapters: an external artifact in, `Finding` objects out. + +Every module here answers the same question in a different dialect — *what is wrong with this +thing, expressed the way the rest of the pipeline already understands?* H1 reads a security +advisory, H2 a dependency manifest, H3 an OpenAPI spec. They share the OSV client and the +advisory→`Finding` mapping, which is why this is a package rather than three `input_*.py` files +importing each other sideways. + +**One-directional rule:** modules here may import `schemas`, `harness` and `agents.*`. Nothing here +may import `pipeline` — `pipeline` imports *us*, lazily, inside the branch that needs it. +""" diff --git a/src/vpcopilot/inputs/cve.py b/src/vpcopilot/inputs/cve.py new file mode 100644 index 0000000..2487722 --- /dev/null +++ b/src/vpcopilot/inputs/cve.py @@ -0,0 +1,141 @@ +"""H1 — an advisory id in, a pipeline-shaped finding out. + +`vpcopilot scan --cve CVE-2024-23334` instead of a repo path. The vulnerabilities most people +actually lose sleep over live in dependencies they do not own, where the code cure is a version +bump someone else has to ship and then they have to deploy. That gap — days or weeks between +"we know" and "it is fixed" — is the case virtual patching exists for, and until now the pipeline +could not see it. + +The division of labour is deliberate and strict: + +* **Code** fetches the advisory and owns every fact with a number in it — the affected package, the + vulnerable range, the fixed version, the CVSS-derived severity. The fixed version is the one + string an operator acts on directly, so no model is allowed near it. +* **The agent** contributes the only thing OSV does not carry: what the vulnerability looks like in + an HTTP request. And it is expected to decline when there is no such thing. +* **Code** turns a declined profile into `no_bandaid` deterministically, rather than asking triage + nicely. The acceptance criterion says a non-observable advisory must route to `no_bandaid` with + residual risk stated; a hard requirement should not depend on a model honouring a prompt. + +The remediation is built here too, and never by the `remediate` agent — for a dependency CVE the +cure is "upgrade to 3.9.2", and asking a model to draft a patch against vendor code it cannot see +is how you get a confident, wrong diff. +""" +from __future__ import annotations + +from typing import Callable + +from ..schemas import Finding, RemediationPlan, TriageDecision +from . import osv + +# CWE → the project's existing VulnClass. Deterministic where the advisory tells us, so the agent +# only has to guess when it does not. VulnClass is deliberately NOT widened: `other` plus a +# concrete exploit_sketch is honest, and adding members ripples into every agent prompt and golden. +CWE_CLASS = { + "CWE-89": "sqli", "CWE-564": "sqli", + "CWE-79": "xss", "CWE-80": "xss", + "CWE-77": "command_injection", "CWE-78": "command_injection", "CWE-94": "command_injection", + "CWE-918": "ssrf", + "CWE-639": "broken_object_authz", "CWE-566": "broken_object_authz", + "CWE-287": "broken_auth", "CWE-306": "broken_auth", "CWE-798": "broken_auth", + "CWE-915": "mass_assignment", "CWE-1321": "mass_assignment", + "CWE-200": "sensitive_data", "CWE-209": "sensitive_data", "CWE-532": "sensitive_data", + "CWE-770": "rate_abuse", "CWE-307": "rate_abuse", "CWE-400": "rate_abuse", + "CWE-840": "business_logic", +} + + +def _vuln_class(advisory: dict, fallback: str) -> str: + for cwe in advisory.get("cwe_ids") or []: + if cwe.upper() in CWE_CLASS: + return CWE_CLASS[cwe.upper()] + return fallback + + +def _title(advisory: dict, profile_title: str) -> str: + base = advisory.get("summary") or profile_title or advisory["id"] + return f"{advisory['id']}: {base}" if not base.startswith(advisory["id"]) else base + + +def resolve_advisory(h, advisory_id: str, *, log: Callable = print) -> dict: + """Fetch, reason, and assemble. Returns everything the pipeline's advisory branch needs: + `{advisory, profile, finding, decision, remediation}` — `decision` is None when the profile is + network-observable and triage should run normally.""" + from ..agents import resolve as resolve_agent + + log(f"resolving {advisory_id} from OSV.dev…") + advisory = osv.resolve(advisory_id, log=log) + target = osv.upgrade_target(advisory) + log(f" {advisory['id']}: {advisory['summary'][:90]}") + if target["fixed_version"]: + log(f" fixed in {target['ecosystem']}/{target['package']} {target['fixed_version']}") + else: + log(f" ⚠ no installable fixed version — {target['note']}") + + profile = resolve_agent.run(h, advisory) + log(f" exploitation profile: network_observable={profile.network_observable} " + f"(confidence {profile.confidence:.2f}) — {profile.reason[:120]}") + if profile.paths: + log(f" paths: {', '.join(profile.paths[:4])}") + + severity = osv.severity_from_cvss(advisory.get("cvss", "")) or profile.severity.value + finding = Finding( + id=advisory["id"], + title=_title(advisory, profile.title), + vuln_class=_vuln_class(advisory, profile.vuln_class.value), + severity=severity, + # file/line/code_snippet stay empty: there is no source. `source` carries the identity that + # `file` carries for a code finding, so correlation and dedup have something to key on. + source=f"osv:{advisory['id']}", + endpoint=(profile.paths[0] if profile.paths else ""), + http_method=(profile.http_methods[0] if profile.http_methods else ""), + description=(advisory.get("details") or profile.description)[:4000], + exploit_sketch=(profile.exploit_sketch if profile.network_observable + else f"no network-observable exploitation pattern: {profile.reason}"), + ) + + decision = None + if not profile.network_observable: + # Deterministic, not delegated. A load balancer cannot mitigate what it cannot see in a + # request, and the honest output is to say so and point at the upgrade. + fix = (f"fixed in {target['package']} {target['fixed_version']} — ship the upgrade" + if target["fixed_version"] else f"no fixed version published ({target['note']})") + decision = TriageDecision( + finding_id=finding.id, bandaids=[], no_bandaid=True, + residual_risk=f"{profile.reason} No band-aid can mitigate this at the load balancer; " + f"{fix}.") + log(" → no_bandaid: nothing observable in a request to block") + + remediation = RemediationPlan( + finding_id=finding.id, + kind="dependency_upgrade", + summary=(f"upgrade {target['package']} to {target['fixed_version']} " + f"(fixes {advisory['id']})" if target["fixed_version"] + else f"{advisory['id']}: {target['note']}"), + package=target["package"], ecosystem=target["ecosystem"], + vulnerable_range=target["vulnerable_range"], fixed_version=target["fixed_version"], + pr_title=(f"fix({target['package'] or 'deps'}): upgrade to {target['fixed_version']} " + f"for {advisory['id']}" if target["fixed_version"] + else f"chore: mitigate {advisory['id']}"), + pr_body=_pr_body(advisory, target, profile), + ) + return {"advisory": advisory, "profile": profile, "finding": finding, + "decision": decision, "remediation": remediation} + + +def _pr_body(advisory: dict, target: dict, profile) -> str: + lines = [f"## {advisory['id']}", "", advisory.get("summary") or "", ""] + if target["fixed_version"]: + lines += [f"**Upgrade** `{target['package']}` " + f"({target['ecosystem']}) to **{target['fixed_version']}**.", + f"Vulnerable: {target['vulnerable_range']}", ""] + else: + lines += [f"**No installable fixed version.** {target['note']}", ""] + lines += ["### Exploitation", "", + profile.exploit_sketch if profile.network_observable + else f"Not observable in an HTTP request. {profile.reason}", ""] + if advisory.get("references"): + lines += ["### References", ""] + [f"- {u}" for u in advisory["references"][:6]] + lines += ["", "_This is a dependency upgrade recommendation, not a patch — the fix lives in " + "the upstream package._"] + return "\n".join(lines) diff --git a/src/vpcopilot/inputs/osv.py b/src/vpcopilot/inputs/osv.py new file mode 100644 index 0000000..9708911 --- /dev/null +++ b/src/vpcopilot/inputs/osv.py @@ -0,0 +1,230 @@ +"""OSV.dev client — the advisory facts, fetched by code and never invented by a model. + +OSV was chosen over NVD and bare GHSA because it needs no credentials (so `scan --cve` stays "safe +to run anywhere", like the rest of scan), spans ecosystems on one schema, and carries the **fixed +version** — which is the whole of H1's "recommend the fixed version rather than drafting a patch to +vendor code". + +Three things the real API does that a reading of the schema does not prepare you for. All three +were found by querying it, and each one silently degrades the answer if unhandled: + +1. **Querying by CVE id often returns the GIT-range record.** `CVE-2024-23334` resolves to an entry + whose only `affected` block is a `GIT` range: no package, no ecosystem, and `fixed` values that + are 40-character commit SHAs. The clean `PyPI/aiohttp fixed=3.9.2` lives on its **aliases** + (`GHSA-5h86-8mv2-jq9f`, `PYSEC-2024-24`). So this follows aliases and merges. Without that, + "upgrade to 24a6d64966d99182e95f5d3a29541ef2fec397ad" is what the operator gets told. +2. **`fixed` is not always a version.** Same cause. A value that looks like a commit is not offered + as an upgrade target — it is recorded as evidence and the recommendation says so. +3. **`summary` is frequently empty** (`CVE-2021-41773`, `CVE-2022-22965`), and for OS-level CVEs + there is no package at all — only `database_specific.cpe`. The prose in `details` is the real + payload, and it is what the resolve agent reasons over. + +Read-only and cached on disk: an advisory is immutable enough for a run, and a demo should not +depend on the network being up. +""" +from __future__ import annotations + +import json +import os +import re +from pathlib import Path +from typing import Callable + +API = "https://api.osv.dev/v1/vulns" +CACHE_ENV = "VPCOPILOT_ADVISORY_CACHE" +TIMEOUT = 20 + +# A 40- (or 7+) character hex string in a `fixed` event is a commit, not something anyone can +# `pip install`. Offering it as an upgrade target is worse than admitting there isn't one. +_COMMITISH = re.compile(r"^[0-9a-f]{7,40}$") +_ADVISORY_ID = re.compile(r"^(CVE-\d{4}-\d{4,}|GHSA-[a-z0-9]{4}-[a-z0-9]{4}-[a-z0-9]{4}|" + r"PYSEC-\d{4}-\d+|GO-\d{4}-\d+|RUSTSEC-\d{4}-\d+)$", re.I) + + +def valid_id(advisory_id: str) -> bool: + return bool(_ADVISORY_ID.match((advisory_id or "").strip())) + + +def _cache_dir() -> Path | None: + raw = (os.environ.get(CACHE_ENV) or "").strip() + return Path(raw) if raw else None + + +def _cached(advisory_id: str) -> dict | None: + d = _cache_dir() + if not d: + return None + p = d / f"{advisory_id}.json" + try: + return json.loads(p.read_text()) if p.is_file() else None + except (OSError, json.JSONDecodeError): + return None + + +def _store(advisory_id: str, obj: dict) -> None: + d = _cache_dir() + if not d: + return + try: + d.mkdir(parents=True, exist_ok=True) + (d / f"{advisory_id}.json").write_text(json.dumps(obj, indent=2)) + except OSError: + pass + + +def fetch(advisory_id: str, *, log: Callable = print) -> dict: + """One raw OSV record. Cache first, then the API.""" + advisory_id = (advisory_id or "").strip() + hit = _cached(advisory_id) + if hit is not None: + return hit + import httpx + r = httpx.get(f"{API}/{advisory_id}", timeout=TIMEOUT) + if r.status_code == 404: + raise RuntimeError(f"no advisory '{advisory_id}' in OSV.dev — check the id " + "(CVE-YYYY-NNNNN, GHSA-xxxx-xxxx-xxxx, PYSEC-YYYY-NN)") + if r.status_code != 200: + raise RuntimeError(f"OSV.dev returned {r.status_code} for '{advisory_id}': {r.text[:200]}") + obj = r.json() + _store(advisory_id, obj) + return obj + + +def _affected_rows(obj: dict) -> list[dict]: + """Flatten `affected[].ranges[]` into rows that say what is broken and what fixes it. + + `versioned` marks a row whose `fixed` is something a human can install, as opposed to a commit + SHA from a GIT range. `database_specific.extracted_events` carries real version strings for some + GIT ranges (`CVE-2021-41773` → `introduced=2.4.49`), so it is read as a fallback identity even + when the events themselves are commits.""" + rows = [] + for a in obj.get("affected") or []: + pkg = a.get("package") or {} + cpe = ((a.get("database_specific") or {}).get("cpe") + or ((a.get("ranges") or [{}])[0].get("database_specific") or {}).get("cpe") or "") + for rng in a.get("ranges") or []: + events = rng.get("events") or [] + extracted = (rng.get("database_specific") or {}).get("extracted_events") or [] + fixed = [e["fixed"] for e in events if e.get("fixed")] + intro = [e["introduced"] for e in events if e.get("introduced")] + last = [e["last_affected"] for e in events if e.get("last_affected")] + versioned = [f for f in fixed if not _COMMITISH.match(str(f))] + rows.append({ + "ecosystem": pkg.get("ecosystem") or "", + "package": pkg.get("name") or "", + "cpe": cpe, + "range_type": rng.get("type") or "", + "introduced": intro, + "fixed": fixed, + "last_affected": last, + "fixed_versions": versioned, + "versioned": bool(versioned), + "extracted": [f"{k}={v}" for e in extracted for k, v in e.items()], + }) + if not (a.get("ranges") or []): # some records list bare versions with no range + vs = a.get("versions") or [] + rows.append({"ecosystem": pkg.get("ecosystem") or "", "package": pkg.get("name") or "", + "cpe": cpe, "range_type": "", "introduced": [], "fixed": [], + "last_affected": vs[-1:], "fixed_versions": [], "versioned": False, + "extracted": []}) + return rows + + +def _first_sentence(text: str) -> str: + t = " ".join((text or "").split()) + m = re.search(r"^(.{20,180}?[.!?])(\s|$)", t) + return (m.group(1) if m else t[:160]).strip() + + +def resolve(advisory_id: str, *, follow_aliases: bool = True, log: Callable = print) -> dict: + """A normalized advisory: the requested record, enriched from its aliases when the requested one + has no installable fixed version. + + This is the function that makes "recommend the fixed version" true rather than nearly true — + see the module docstring for what querying a CVE id alone actually returns.""" + if not valid_id(advisory_id): + raise RuntimeError(f"'{advisory_id}' is not an advisory id — expected CVE-YYYY-NNNNN, " + "GHSA-xxxx-xxxx-xxxx, PYSEC-YYYY-NN, GO-YYYY-NNNN or RUSTSEC-YYYY-NNNN") + obj = fetch(advisory_id, log=log) + rows = _affected_rows(obj) + consulted = [obj.get("id") or advisory_id] + + if follow_aliases and not any(r["versioned"] for r in rows): + for alias in (obj.get("aliases") or [])[:4]: + if not valid_id(alias): + continue + try: + alt = fetch(alias, log=log) + except Exception as e: # noqa: BLE001 — enrichment is best-effort, never fatal + log(f" ⚠ could not read alias {alias}: {e}") + continue + alt_rows = _affected_rows(alt) + consulted.append(alias) + if any(r["versioned"] for r in alt_rows): + log(f" {advisory_id} carries no installable fixed version; {alias} does " + f"({', '.join(sorted({v for r in alt_rows for v in r['fixed_versions']}))})") + rows = alt_rows + rows + break + + sev = next((s.get("score") for s in obj.get("severity") or [] if s.get("score")), "") + details = obj.get("details") or "" + return { + "id": obj.get("id") or advisory_id, + "aliases": obj.get("aliases") or [], + "consulted": consulted, + "summary": obj.get("summary") or _first_sentence(details), + "details": details[:4000], + "cwe_ids": (obj.get("database_specific") or {}).get("cwe_ids") or [], + "cvss": sev, + "published": obj.get("published") or "", + "affected": rows, + "references": [r.get("url") for r in (obj.get("references") or [])][:12], + } + + +def severity_from_cvss(cvss: str) -> str: + """CVSS vector → the project's four-level Severity. Absent scores are `medium`, never + `critical` — an unknown severity must not jump the queue ahead of a measured one.""" + m = re.search(r"CVSS:3\.[01]/(.+)", cvss or "") + if not m: + return "medium" + parts = dict(p.split(":", 1) for p in m.group(1).split("/") if ":" in p) + # Approximate the base score from the impact/exploitability metrics that dominate it. This is a + # bucketing, not a CVSS implementation — OSV rarely publishes the numeric score. + high_impact = sum(1 for k in ("C", "I", "A") if parts.get(k) == "H") + net = parts.get("AV") == "N" + easy = parts.get("AC") == "L" and parts.get("PR") == "N" and parts.get("UI") == "N" + if net and easy and high_impact >= 2: + return "critical" + if net and (high_impact >= 1 or easy): + return "high" + if high_impact >= 1: + return "medium" + return "low" + + +def upgrade_target(advisory: dict) -> dict: + """The single best "upgrade to X" recommendation, or an honest statement that OSV has none. + + Prefers a row with an installable version. A GIT-only advisory yields `fixed_version: ""` and a + note carrying the commit — because telling an operator to "upgrade to + 24a6d64966d99182e95f5d3a29541ef2fec397ad" is not a recommendation.""" + rows = advisory.get("affected") or [] + best = next((r for r in rows if r["versioned"]), None) + if best: + return {"package": best["package"], "ecosystem": best["ecosystem"], + "fixed_version": sorted(best["fixed_versions"])[0], + "vulnerable_range": ", ".join(best["introduced"]) or "see advisory", "note": ""} + any_row = rows[0] if rows else {} + commits = [f for r in rows for f in r.get("fixed", [])] + note = ("OSV records the fix as a source commit, not a released version" + if commits else "OSV lists no fixed version for this advisory") + # A GIT range's `introduced` is a commit SHA, which tells an operator nothing about which + # release they are running. `database_specific.extracted_events` carries the human versions + # for exactly that case (CVE-2021-41773 → introduced=2.4.49), so prefer it. + introduced = [v for v in (any_row.get("introduced") or []) if not _COMMITISH.match(str(v))] + return {"package": any_row.get("package") or any_row.get("cpe", ""), + "ecosystem": any_row.get("ecosystem", ""), "fixed_version": "", + "vulnerable_range": ", ".join(introduced) or ", ".join(any_row.get("extracted") or []) + or "see advisory", + "note": f"{note}: {', '.join(commits[:2])}" if commits else note} diff --git a/src/vpcopilot/pipeline.py b/src/vpcopilot/pipeline.py index d3a3ef5..ede2402 100644 --- a/src/vpcopilot/pipeline.py +++ b/src/vpcopilot/pipeline.py @@ -40,9 +40,13 @@ def _dedup_findings(findings, log, counter: dict | None = None): log — the residue the old `BACKLOG.md` per-stage-metrics item left behind.""" kept, seen = [], {} for f in sorted(findings, key=lambda f: _SEV_RANK.get(_sev(f), 9)): - key = (f.file, _vclass(f), (getattr(f, "endpoint", "") or f"L{f.line}")) + # `f.file` is always set on the repo path (pipeline sets it after discover), so the + # fallback is structurally unreachable there — advisory findings, which have no file, + # would otherwise all key on ("", class, "L0") and silently collapse into one. + ident = f.file or getattr(f, "source", "") or f.id + key = (ident, _vclass(f), (getattr(f, "endpoint", "") or f"L{f.line}")) if key in seen: - log(f" dedup: {f.id} duplicates {seen[key]} ({f.file} {key[1]} {key[2]}) — dropped") + log(f" dedup: {f.id} duplicates {seen[key]} ({ident} {key[1]} {key[2]}) — dropped") continue seen[key] = f.id kept.append(f) @@ -52,7 +56,7 @@ def _dedup_findings(findings, log, counter: dict | None = None): def run_pipeline( - repo_path: str, + repo_path: str | None = None, out_dir: str = "out", config_path: str | None = None, min_confidence: float = 0.5, @@ -61,33 +65,60 @@ def run_pipeline( max_bytes: int = 60_000, 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 ) -> 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")) h = Harness(config_path) - root = Path(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"): - n = sum(1 for _, r in skipped if r == reason) - if n: - log(f" ⚠ {n} file(s) skipped ({reason}) — raise --max-files/--max-bytes to include them") + h.warmup() # B6: warm instructor's mode registry before ANY fan-out — both inputs need it t0, started = time.perf_counter(), runmeta.utc_now() dedup_counter: dict = {} - - # 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) - if route_ctx: - log("route context: found the app's declared routes — grounding finding endpoints (no guessing)") - else: - log(" ⚠ NO app route context found (no OpenAPI/Swagger spec or route registrations detected) " - "— finding endpoints are INFERRED and may be inaccurate") - - # 1) discover (per file, parallel) -------------------------------------- - findings = [] + root = Path(repo_path) if repo_path else None + files, skipped = [], [] file_code: dict[str, str] = {} file_raw: dict[str, str] = {} + route_ctx = None + findings: list = [] + forced_decision = forced_remediation = None + advisory_meta: dict | None = None + + if advisory: + # H1 — an advisory produces ONE finding and then joins the ordinary stages. There is no + # repo to walk, nothing to verify a second time (OSV already asserts the vulnerability is + # real; re-litigating it against source we do not have would only invent doubt), so the + # branch supplies `findings` directly and everything from triage down runs unchanged. + from .inputs.cve import resolve_advisory + res = resolve_advisory(h, advisory, log=log) + findings = [res["finding"]] + forced_decision, forced_remediation = res["decision"], res["remediation"] + advisory_meta = {"id": res["advisory"]["id"], "source": "osv", + "consulted": res["advisory"]["consulted"], + "network_observable": res["profile"].network_observable, + "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) + 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"): + n = sum(1 for _, r in skipped if r == reason) + if n: + log(f" ⚠ {n} file(s) skipped ({reason}) — raise --max-files/--max-bytes to include them") + + # 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) + if route_ctx: + log("route context: found the app's declared routes — grounding finding endpoints (no guessing)") + else: + log(" ⚠ NO app route context found (no OpenAPI/Swagger spec or route registrations detected) " + "— finding endpoints are INFERRED and may be inaccurate") + # 1) discover (per file, parallel) -------------------------------------- def _discover(p): rel = str(p.relative_to(root)) try: @@ -98,9 +129,9 @@ def _discover(p): from .schemas import FindingList return rel, "", "", FindingList(findings=[]) - # B6: warm instructor's mode-registry once (its lazy init isn't thread-safe) before the fan-out, - # then discover every file in parallel with per-file error isolation. ex.map preserves order. - h.warmup() + # B6: instructor's mode registry is warmed at the top of run_pipeline (its lazy init isn't + # thread-safe) — discover every file in parallel with per-file error isolation. ex.map + # preserves order. disc_results = [] if files: with ThreadPoolExecutor(max_workers=concurrency) as ex: @@ -120,8 +151,9 @@ def _discover(p): findings.append(f) if res.findings: log(f" {rel}: {len(res.findings)} candidate finding(s)") - discover_s = time.perf_counter() - t0 - log(f"discovered {len(findings)} candidate finding(s)") + if not advisory: + discover_s = time.perf_counter() - t0 + log(f"discovered {len(findings)} candidate finding(s)") # 2) verify (adversarial, per finding, parallel) ------------------------ t_verify = time.perf_counter() @@ -143,7 +175,11 @@ def _threshold(f): return max(0.0, min(1.0, min_confidence + shift)) with ThreadPoolExecutor(max_workers=concurrency) as ex: - for f, v in ex.map(_verify, findings): + # H1 — an advisory run has no source to read, and verify's entire method is reading the + # offending code. OSV already asserts the vulnerability is real; re-litigating it against + # code we do not have would only manufacture doubt. The resolve agent's own confidence is + # the gate instead, applied in inputs/cve.py. + for f, v in (ex.map(_verify, findings) if not advisory else []): if v is None: # B6: verify errored — count as dropped, keep going dropped += 1 continue @@ -158,8 +194,11 @@ def _threshold(f): else: refuted += 1 log(f" verify {f.id}: refuted ({v.confidence:.2f})") + if advisory: + verified = list(findings) verify_s = time.perf_counter() - t_verify - log(f"{len(verified)} finding(s) verified real (min-confidence {min_confidence})") + if not advisory: + log(f"{len(verified)} finding(s) verified real (min-confidence {min_confidence})") # 3-5) triage -> generate band-aids -> remediate (code cure) ------------ t_synth = time.perf_counter() @@ -177,7 +216,13 @@ def _threshold(f): # findings) never sends one giant call that blows the per-call timeout; chunks run in # parallel and their decisions are concatenated. TRIAGE_CHUNK = 12 - if len(verified) <= TRIAGE_CHUNK: + if forced_decision is not None: + # H1 — the advisory has no network-observable exploitation pattern, so no control at a + # load balancer can mitigate it. That is a fact about the advisory, not a judgement + # call, and the acceptance requires it: routing it to `no_bandaid` in code rather than + # asking triage nicely is what makes it a guarantee instead of a hope. + decisions = [forced_decision] + elif len(verified) <= TRIAGE_CHUNK: decisions = triage.run(h, verified).decisions else: chunks = [verified[i:i + TRIAGE_CHUNK] for i in range(0, len(verified), TRIAGE_CHUNK)] @@ -229,7 +274,8 @@ def _probe(f): pr = probe_by_id.get(d.finding_id) or {} 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) + key = correlate.coverage_key(b.control.value, f.file, + identity=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}) @@ -253,7 +299,14 @@ def _probe(f): # 5) every verified finding gets a real code fix (band-aid != cure) — A5: over ALL # verified findings, in parallel, not only those triage handed a band-aid. Skippable # (draft_code_fixes) to save the biggest chunk of tokens when only band-aids are wanted. - if draft_code_fixes: + if forced_remediation is not None: + # H1 — the cure for a dependency CVE is a version bump in someone else's package. + # There is no file of ours to patch, and asking a model to draft a diff against vendor + # code it cannot see is how you get a confident, wrong patch. The version comes from + # OSV; `remediate` is never called on this path. + remediations = [forced_remediation] + log(f"cure: {forced_remediation.summary}") + elif draft_code_fixes: def _remediate(f): return remediate.run(h, f, file_raw.get(f.file, "")) @@ -285,13 +338,15 @@ def _remediate(f): try: cfg = getattr(h, "cfg", None) runmeta.write_manifest( - out_dir, repo=str(root.resolve()), config_path=config_path, started=started, + out_dir, repo=str(root.resolve()) if root else None, advisory=advisory_meta, + input_kind="advisory" if advisory else "repo", + 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, "draft_code_fixes": draft_code_fixes}, counts={"candidates": len(findings), "verified": len(verified), "policies": len(artifacts), "code_fix_prs": len(remediations)}, - finished=runmeta.utc_now(), **runmeta.git_provenance(root)) + finished=runmeta.utc_now(), **(runmeta.git_provenance(root) if root else {})) except Exception as e: # noqa: BLE001 log(f" ⚠ could not write the run manifest (run.json): {e} — an audit export will lack provenance") from . import report # E3: drop a standalone shareable HTML dashboard of the results diff --git a/src/vpcopilot/pr.py b/src/vpcopilot/pr.py index 3974b70..e3cb0a6 100644 --- a/src/vpcopilot/pr.py +++ b/src/vpcopilot/pr.py @@ -25,18 +25,39 @@ def open_pr(remediation: dict, repo_slug: str, *, base: str = "main", path_prefi token: str | None = None, dry_run: bool = False, out_dir: str = "out", log: Callable = print) -> dict: fid = remediation["finding_id"] - rel = remediation["file"] + rel = remediation.get("file") or "" path = f"{path_prefix.rstrip('/')}/{rel}" if path_prefix else rel branch = f"vpcopilot/fix-{fid}" content = remediation.get("patched_content") plan = {"repo": repo_slug, "base": base, "branch": branch, "path": path, "title": remediation.get("pr_title", "")} - if not content: - raise RuntimeError(f"remediation {fid} has no patched_content — re-run the scan") + # H1 — a dependency CVE's cure is a version bump in someone else's package. There is nothing of + # ours to patch, so this reports the upgrade and touches GitHub not at all: no token needed, no + # branch created, nothing to review. Checked FIRST, so it is reachable in a dry run too. + if remediation.get("kind") == "dependency_upgrade" and not content: + target = remediation.get("fixed_version") + rec = (f"upgrade {remediation.get('package') or 'the affected package'} to {target}" + if target else remediation.get("summary", "no fixed version published")) + log(f"advisory: {rec} — no PR to open (the fix is upstream, not in this repo)") + return {"mode": "advisory", "recommendation": rec, "finding_id": fid, + "package": remediation.get("package", ""), + "ecosystem": remediation.get("ecosystem", ""), + "fixed_version": target or "", + "vulnerable_range": remediation.get("vulnerable_range", "")} + + # Order matters: the dry run must be able to REPORT a missing patch rather than raising + # identically to a live run. This check used to sit above it, so `--dry-run` could not preview + # anything that lacked content. if dry_run: log(f"[dry-run] would open PR against {repo_slug}@{base}: branch {branch}, file {path}") - return {"mode": "dry_run", **plan} + return {"mode": "dry_run", "has_patch": bool(content), **plan} + if not content: + raise RuntimeError(f"remediation {fid} has no patched_content — re-run the scan") + if not rel: + # model_dump() always emits defaulted keys, so an absent file is "" rather than a KeyError — + # and "" would reach repo.get_contents() as a directory listing and AttributeError on .sha. + raise RuntimeError(f"remediation {fid} names no file to patch") from github import Github, GithubException diff --git a/src/vpcopilot/report.py b/src/vpcopilot/report.py index 0f94078..de35e54 100644 --- a/src/vpcopilot/report.py +++ b/src/vpcopilot/report.py @@ -248,7 +248,7 @@ def _models_html() -> str: from .config import load_config import os cfg = load_config(os.environ.get("VPCOPILOT_CONFIG", "config/agents.yaml")) - agents = ["discover", "verify", "triage", "generate", "remediate", "probe", "refine"] + agents = ["resolve", "discover", "verify", "triage", "generate", "remediate", "probe", "refine"] chips = "".join(f'{a} · {_e(cfg.for_agent(a).model)}' for a in agents) except Exception: # noqa: BLE001 diff --git a/src/vpcopilot/schemas.py b/src/vpcopilot/schemas.py index 063c8c4..60aee57 100644 --- a/src/vpcopilot/schemas.py +++ b/src/vpcopilot/schemas.py @@ -4,6 +4,8 @@ from __future__ import annotations from enum import Enum +from typing import Literal + from pydantic import BaseModel, Field @@ -55,6 +57,10 @@ class Finding(BaseModel): "", description="the EFFECTIVE HTTP request path a client calls, INCLUDING every router/" "blueprint/mount/file-route prefix (e.g. /users/v1/register) — not just the local handler string") http_method: str = Field("", description="the HTTP method(s) for that endpoint, e.g. POST") + source: str = Field( + "", description="H1: where this finding came from when it did NOT come from a repo file — " + "e.g. 'osv:CVE-2024-23334'. Carries the identity that `file` carries for a code finding, " + "so correlation and dedup have something to key on.") description: str exploit_sketch: str = Field(..., description="how an attacker would exploit it") code_snippet: str = Field("", description="the offending code") @@ -120,15 +126,59 @@ class GeneratedArtifacts(BaseModel): class RemediationPlan(BaseModel): + """The cure. Two kinds, and the distinction is load-bearing. + + A `code_fix` rewrites a file you own — `pr.py` writes `patched_content` to a branch. A + `dependency_upgrade` (H1) is a version bump in someone else's package: there is nothing to + patch, and drafting a diff against vendor code would be worse than useless. The file/diff + fields are therefore optional, and `kind` defaults to `code_fix` so an unlabelled plan that + forgot its patch still fails loudly rather than silently reading as an advisory.""" finding_id: str summary: str - file: str - diff: str = Field(..., description="unified diff (for the PR description / human review)") + file: str = "" + diff: str = Field("", description="unified diff (for the PR description / human review)") patched_content: str = Field( - ..., description="the COMPLETE corrected file, written verbatim to a branch to open the PR" + "", description="the COMPLETE corrected file, written verbatim to a branch to open the PR" ) pr_title: str pr_body: str + kind: Literal["code_fix", "dependency_upgrade"] = "code_fix" + # Filled from OSV by code, never by a model — the version number is the one string in this + # object an operator will act on directly. + package: str = "" + ecosystem: str = "" + vulnerable_range: str = "" + fixed_version: str = "" + + +class ExploitationProfile(BaseModel): + """H1: what a security advisory means in terms of HTTP requests — or an honest statement that + it cannot be expressed that way. + + `network_observable=False` is a first-class answer, not a failure. A deserialization bug + reachable only from a local file, a malicious build-time dependency, a memory-corruption issue + with no request signature — none of those can be virtually patched at a load balancer, and the + acceptance for H1 requires the pipeline to say so and route to `no_bandaid` rather than invent + a plausible-looking path to block.""" + advisory_id: str + network_observable: bool = Field( + ..., description="True only if the advisory describes something identifiable in an HTTP " + "request. When unsure, False.") + reason: str = Field(..., min_length=1, description="why a request-level pattern can or cannot " + "be derived, citing the advisory text") + vuln_class: VulnClass + severity: Severity + title: str + description: str + exploit_sketch: str = Field("", description="how an attacker exploits it over HTTP; empty when " + "network_observable is False") + paths: list[str] = Field(default_factory=list, description="app-relative URL patterns, no host") + http_methods: list[str] = Field(default_factory=list) + parameters: list[str] = Field(default_factory=list, description="query/body params carrying the payload") + headers: list[str] = Field(default_factory=list, description="request headers carrying the payload") + example_requests: list["ProbeRequest"] = Field(default_factory=list) + confidence: float = Field(..., ge=0, le=1) + caveats: list[str] = Field(default_factory=list) class ProbeRequest(BaseModel): @@ -220,3 +270,7 @@ class SimulationResult(BaseModel): True, description="False when records came from XC logs, which capture no request body") policies: list[PolicySimulation] = Field(default_factory=list) caveats: list[str] = Field(default_factory=list) + + +# ExploitationProfile references ProbeRequest, which is defined below it. +ExploitationProfile.model_rebuild() diff --git a/tests/test_inputs_cve.py b/tests/test_inputs_cve.py new file mode 100644 index 0000000..096716f --- /dev/null +++ b/tests/test_inputs_cve.py @@ -0,0 +1,343 @@ +"""H1 — the CVE/advisory input path. Offline: OSV is faked, the resolve agent is faked. + +The live behaviours these fakes encode were all observed against the real api.osv.dev — see the +docstrings. Three of them are not guessable from the OSV schema and each one silently degrades the +answer if unhandled.""" +import json + +import pytest + +from vpcopilot.inputs import osv + +_REAL_FETCH = osv.fetch # captured at import, before the autouse fake replaces it + +# Shapes taken verbatim from real api.osv.dev responses. +LODASH = {"id": "GHSA-jf85-cpcp-j695", "aliases": ["CVE-2019-10744"], + "summary": "Prototype Pollution in lodash", "details": "lodash before 4.17.12 …", + "severity": [{"type": "CVSS_V3", "score": "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:N/I:H/A:H"}], + "database_specific": {"cwe_ids": ["CWE-1321"]}, + "affected": [{"package": {"ecosystem": "npm", "name": "lodash"}, + "ranges": [{"type": "SEMVER", "events": [{"introduced": "0"}, + {"fixed": "4.17.12"}]}]}], + "references": [{"url": "https://example.test/a", "type": "ADVISORY"}]} + +# CVE-2024-23334: querying by CVE id returns a GIT-range record with NO package and commit-SHA +# "fixed" values. The installable 3.9.2 only exists on the GHSA alias. +AIOHTTP_CVE = {"id": "CVE-2024-23334", "aliases": ["GHSA-5h86-8mv2-jq9f", "PYSEC-2024-24"], + "summary": "aiohttp.web.static(follow_symlinks=True) is vulnerable to directory traversal", + "details": "When 'follow_symlinks' is set to True there is no validation …", + "severity": [{"type": "CVSS_V3", "score": "CVSS:3.1/AV:N/AC:H/PR:N/UI:N/S:U/C:H/I:N/A:N"}], + "database_specific": {"cwe_ids": ["CWE-22"]}, + "affected": [{"ranges": [{"type": "GIT", "repo": "https://github.com/aio-libs/aiohttp", + "events": [{"introduced": "0"}, + {"fixed": "24a6d64966d99182e95f5d3a29541ef2fec397ad"}]}]}], + "references": []} +AIOHTTP_GHSA = {"id": "GHSA-5h86-8mv2-jq9f", "aliases": ["CVE-2024-23334"], "summary": "aiohttp traversal", + "details": "…", "severity": [], "database_specific": {"cwe_ids": ["CWE-22"]}, + "affected": [{"package": {"ecosystem": "PyPI", "name": "aiohttp"}, + "ranges": [{"type": "ECOSYSTEM", + "events": [{"introduced": "1.0.5"}, {"fixed": "3.9.2"}]}]}], + "references": []} +# CVE-2021-41773: an OS-level CVE — no package at all, no fixed version, versions only in +# database_specific.extracted_events. +HTTPD = {"id": "CVE-2021-41773", "aliases": ["BIT-apache-2021-41773"], "summary": "", + "details": "A flaw was found in a change made to path normalization in Apache HTTP " + "Server 2.4.49. An attacker could use a path traversal attack.", + "severity": [{"type": "CVSS_V3", "score": "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H"}], + "database_specific": {}, + "affected": [{"ranges": [{"type": "GIT", "events": [{"introduced": "bbacd798"}, + {"last_affected": "bbacd798"}], + "database_specific": {"cpe": "cpe:2.3:a:apache:http_server:2.4.49", + "extracted_events": [{"introduced": "2.4.49"}]}}]}], + "references": []} + +RECORDS = {r["id"]: r for r in (LODASH, AIOHTTP_CVE, AIOHTTP_GHSA, HTTPD)} + + +@pytest.fixture(autouse=True) +def _fake_osv(monkeypatch): + def fetch(advisory_id, *, log=print): + if advisory_id not in RECORDS: + raise RuntimeError(f"no advisory '{advisory_id}' in OSV.dev") + return RECORDS[advisory_id] + monkeypatch.setattr(osv, "fetch", fetch) + + +# ---- the three things the real API does that the schema does not prepare you for ---- +def test_a_cve_id_resolves_through_its_alias_to_an_installable_version(): + """Observed live: `CVE-2024-23334` alone yields a GIT range whose only `fixed` value is + 24a6d64966d99182e95f5d3a29541ef2fec397ad. Telling an operator to "upgrade to" that is not a + recommendation. The real 3.9.2 lives on the GHSA alias, so resolve follows aliases.""" + a = osv.resolve("CVE-2024-23334", log=lambda m: None) + t = osv.upgrade_target(a) + assert t["fixed_version"] == "3.9.2" + assert t["package"] == "aiohttp" and t["ecosystem"] == "PyPI" + assert a["consulted"] == ["CVE-2024-23334", "GHSA-5h86-8mv2-jq9f"] + + +def test_a_commit_sha_is_never_offered_as_an_upgrade_target(): + """The mirror of the above: with no alias to rescue it, the honest answer is that OSV has no + installable version — not a 40-character hex string.""" + a = osv.resolve("CVE-2024-23334", follow_aliases=False, log=lambda m: None) + t = osv.upgrade_target(a) + assert t["fixed_version"] == "" + assert "source commit" in t["note"] and "24a6d649" in t["note"] + + +def test_an_advisory_with_no_package_still_resolves(): + """OS-level CVEs have no ecosystem package at all — identity falls back to the CPE, and the + affected version lives in database_specific.extracted_events.""" + a = osv.resolve("CVE-2021-41773", log=lambda m: None) + t = osv.upgrade_target(a) + assert t["fixed_version"] == "" and "no fixed version" in t["note"] + assert "apache" in t["package"] + assert "2.4.49" in t["vulnerable_range"] + + +def test_an_empty_summary_falls_back_to_the_first_sentence_of_details(): + """`summary` is empty on many CVE records; `details` is where the prose actually lives.""" + a = osv.resolve("CVE-2021-41773", log=lambda m: None) + assert a["summary"].startswith("A flaw was found") and a["summary"].endswith(".") + + +def test_the_clean_case_needs_no_alias_hop(): + a = osv.resolve("GHSA-jf85-cpcp-j695", log=lambda m: None) + assert osv.upgrade_target(a)["fixed_version"] == "4.17.12" + assert a["consulted"] == ["GHSA-jf85-cpcp-j695"] + + +# ---- ids and severity ---- +@pytest.mark.parametrize("vid,ok", [ + ("CVE-2024-23334", True), ("GHSA-5h86-8mv2-jq9f", True), ("PYSEC-2024-24", True), + ("GO-2024-1234", True), ("RUSTSEC-2021-0001", True), + ("", False), ("not-an-id", False), ("CVE-24-1", False), ("../../etc/passwd", False), +]) +def test_advisory_ids_are_validated(vid, ok): + assert osv.valid_id(vid) is ok + + +def test_a_bad_id_is_refused_before_any_request(monkeypatch): + monkeypatch.setattr(osv, "fetch", lambda *a, **k: pytest.fail("must not fetch")) + with pytest.raises(RuntimeError, match="not an advisory id"): + osv.resolve("'; DROP TABLE", log=lambda m: None) + + +def test_an_unknown_advisory_says_so(): + with pytest.raises(RuntimeError, match="no advisory"): + osv.resolve("CVE-1999-0001", log=lambda m: None) + + +@pytest.mark.parametrize("vector,expect", [ + ("CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H", "critical"), + ("CVSS:3.1/AV:N/AC:H/PR:N/UI:N/S:U/C:H/I:N/A:N", "high"), + ("CVSS:3.1/AV:L/AC:L/PR:L/UI:N/S:U/C:H/I:N/A:N", "medium"), + ("CVSS:3.1/AV:L/AC:H/PR:H/UI:R/S:U/C:L/I:N/A:N", "low"), + ("", "medium"), # unknown is never critical — it must not jump the queue + ("garbage", "medium"), +]) +def test_severity_buckets(vector, expect): + assert osv.severity_from_cvss(vector) == expect + + +# ---- the cache ---- +def test_the_cache_is_used_and_written(tmp_path, monkeypatch): + monkeypatch.setenv(osv.CACHE_ENV, str(tmp_path)) + monkeypatch.setattr(osv, "fetch", _REAL_FETCH) # undo the autouse fake for this one + calls = [] + + class R: + status_code = 200 + def json(self): return LODASH + import httpx + monkeypatch.setattr(httpx, "get", lambda url, **k: calls.append(url) or R()) + osv.fetch("GHSA-jf85-cpcp-j695") + osv.fetch("GHSA-jf85-cpcp-j695") + assert len(calls) == 1 # second call served from disk + assert json.loads((tmp_path / "GHSA-jf85-cpcp-j695.json").read_text())["id"] == LODASH["id"] + + +# ---- the finding + decision + remediation the pipeline consumes ---- +class FakeProfile: + def __init__(self, observable=True, **kw): + from vpcopilot.schemas import Severity, VulnClass + self.network_observable = observable + self.reason = kw.get("reason", "the advisory names the request path") + self.vuln_class = VulnClass("other") + self.severity = Severity("high") + self.title = kw.get("title", "t") + self.description = "d" + self.exploit_sketch = kw.get("sketch", "GET /static/../../etc/passwd") + self.paths = kw.get("paths", ["/static/../../etc/passwd"]) + self.http_methods = ["GET"] + self.parameters, self.headers, self.example_requests, self.caveats = [], [], [], [] + self.confidence = 0.8 + + +def _resolve(monkeypatch, advisory_id, profile): + from vpcopilot.agents import resolve as ra + from vpcopilot.inputs.cve import resolve_advisory + monkeypatch.setattr(ra, "run", lambda h, a: profile) + return resolve_advisory(None, advisory_id, log=lambda m: None) + + +def test_an_observable_advisory_becomes_a_finding_that_triage_can_route(monkeypatch): + r = _resolve(monkeypatch, "CVE-2024-23334", FakeProfile()) + f = r["finding"] + assert f.id == "CVE-2024-23334" and f.source == "osv:CVE-2024-23334" + assert f.file == "" and f.line == 0 and f.code_snippet == "" # there is no source + assert f.endpoint == "/static/../../etc/passwd" and f.http_method == "GET" + assert f.severity.value == "high" + assert r["decision"] is None # triage runs normally + + +def test_the_cwe_picks_the_vuln_class_before_the_agent_does(monkeypatch): + """CWE-1321 is prototype pollution → mass_assignment. Deterministic where the advisory tells + us; the agent only guesses when it does not.""" + r = _resolve(monkeypatch, "GHSA-jf85-cpcp-j695", FakeProfile()) + assert r["finding"].vuln_class.value == "mass_assignment" + + +def test_a_non_observable_advisory_routes_to_no_bandaid_in_code(monkeypatch): + """The acceptance requires this, so it is decided in code rather than asked of triage — a hard + requirement must not depend on a model honouring a prompt.""" + r = _resolve(monkeypatch, "CVE-2024-23334", + FakeProfile(observable=False, reason="exploitation needs a local file.")) + d = r["decision"] + assert d is not None and d.no_bandaid is True and d.bandaids == [] + assert "local file" in d.residual_risk + assert "3.9.2" in d.residual_risk # and it points at the real fix + assert r["finding"].exploit_sketch.startswith("no network-observable exploitation pattern") + + +def test_the_cure_is_a_version_bump_never_a_patch(monkeypatch): + r = _resolve(monkeypatch, "CVE-2024-23334", FakeProfile()) + rem = r["remediation"] + assert rem.kind == "dependency_upgrade" + assert rem.fixed_version == "3.9.2" and rem.package == "aiohttp" + assert rem.patched_content == "" and rem.diff == "" and rem.file == "" + assert "3.9.2" in rem.summary and "upstream" in rem.pr_body + + +def test_an_advisory_with_no_fix_says_so_rather_than_inventing_one(monkeypatch): + r = _resolve(monkeypatch, "CVE-2021-41773", FakeProfile()) + rem = r["remediation"] + assert rem.fixed_version == "" and "no fixed version" in rem.summary.lower() + assert "No installable fixed version" in rem.pr_body + + +def test_a_declining_agent_cannot_smuggle_paths_through(monkeypatch): + """The prompt forbids it, but a model that says "cannot be observed" and lists paths anyway + must not have those paths reach generate.""" + from vpcopilot.agents import resolve as ra + from vpcopilot.harness import Harness # noqa: F401 — only for the type + prof = FakeProfile(observable=False, paths=["/admin"]) + monkeypatch.setattr(ra.Harness, "run", lambda self, *a, **k: prof, raising=False) + + class H: + def run(self, *a, **k): return prof + out = ra.run(H(), {"id": "CVE-2024-23334"}) + assert out.paths == [] and out.http_methods == [] + + +# ---- the pipeline entry, correlation identity, and the PR decline ---- +def test_the_pipeline_refuses_both_or_neither_input(): + from vpcopilot.pipeline import run_pipeline + with pytest.raises(ValueError, match="not neither"): + run_pipeline() + with pytest.raises(ValueError, match="not both"): + run_pipeline("/some/repo", advisory="CVE-2024-23334") + + +def test_a_file_less_finding_gets_its_own_coverage_key(): + """Before this, `endpoint_of("")` returned "" so every advisory's service_policy collapsed onto + the single key `service_policy:` and all but the first were logged "already covered" — silently + generating no band-aid at all.""" + from vpcopilot.correlate import coverage_key + a = coverage_key("service_policy", "", identity="osv:CVE-2024-23334") + b = coverage_key("service_policy", "", identity="osv:CVE-2019-10744") + assert a != b + # and the repo path is untouched + assert coverage_key("service_policy", "app/api/pay/route.js") == "service_policy:pay" + assert coverage_key("waf", "", identity="x") == "waf" # LB-wide is still LB-wide + + +def test_identity_is_used_verbatim_not_through_endpoint_of(): + """`endpoint_of` returns the second-to-last segment — right for a repo path, wrong for a URL, + where /api/pay and /api/login would both collapse to `api`.""" + from vpcopilot.correlate import coverage_key + assert coverage_key("service_policy", "", identity="/api/pay") != \ + coverage_key("service_policy", "", identity="/api/login") + + +def test_two_advisories_of_one_class_do_not_dedup_into_each_other(): + """The dedup key was (file, class, endpoint or Lline) — ("", cls, "L0") for every advisory.""" + from vpcopilot.pipeline import _dedup_findings + from vpcopilot.schemas import Finding + + def f(fid): + return Finding(id=fid, title=fid, vuln_class="other", severity="high", description="d", + exploit_sketch="e", source=f"osv:{fid}") + kept = _dedup_findings([f("CVE-1"), f("CVE-2")], lambda m: None, {}) + assert {k.id for k in kept} == {"CVE-1", "CVE-2"} + + +def test_pr_declines_a_dependency_upgrade_without_touching_github(monkeypatch): + from vpcopilot import pr + monkeypatch.setattr(pr, "_resolve_token", lambda *a: pytest.fail("must not need a token")) + res = pr.open_pr({"finding_id": "CVE-2024-23334", "kind": "dependency_upgrade", + "package": "aiohttp", "ecosystem": "PyPI", "fixed_version": "3.9.2", + "vulnerable_range": "1.0.5", "summary": "s", "pr_title": "t", "pr_body": "b"}, + "acme/app", log=lambda m: None) + assert res["mode"] == "advisory" and res["fixed_version"] == "3.9.2" + assert "upgrade aiohttp to 3.9.2" in res["recommendation"] + + +def test_a_code_fix_with_no_patch_still_raises(): + """The advisory branch must not become a way for a broken code fix to pass silently.""" + from vpcopilot import pr + with pytest.raises(RuntimeError, match="no patched_content"): + pr.open_pr({"finding_id": "f1", "file": "a.js", "patched_content": "", + "pr_title": "t", "pr_body": "b"}, "acme/app", log=lambda m: None) + + +def test_a_dry_run_can_now_report_a_missing_patch_instead_of_raising(): + """The check used to sit above the dry-run branch, so --dry-run raised identically to a live + run and could preview nothing.""" + from vpcopilot import pr + res = pr.open_pr({"finding_id": "f1", "file": "a.js", "patched_content": "", + "pr_title": "t", "pr_body": "b"}, "acme/app", dry_run=True, + log=lambda m: None) + assert res["mode"] == "dry_run" and res["has_patch"] is False + + +def test_the_resolve_agent_is_registered_everywhere_it_has_to_be(): + """Four sites, not the three the roadmap names — bench_model.py is the fourth, and an agent + missing from it is absent from every benchmark's model map.""" + from vpcopilot.bench_model import AGENTS + from vpcopilot.config import AGENT_NAMES + from vpcopilot.console.app import AGENT_ROLES + assert "resolve" in AGENT_NAMES and "resolve" in AGENTS and "resolve" in AGENT_ROLES + assert "resolve" in (__import__("pathlib").Path("src/vpcopilot/report.py").read_text()) + + +def test_every_shipped_config_names_the_resolve_agent(): + """An agent absent from agents.yaml silently falls back to the defaults model, and run.json + then records that as fact.""" + import pathlib + + from vpcopilot.config import load_config + for cfg in sorted(pathlib.Path("config").glob("agents*.yaml")): + assert load_config(str(cfg)).for_agent("resolve").model, cfg + + +def test_the_console_scan_surface_accepts_an_advisory(): + """One module function, two surfaces — a CVE has to be reachable from the console too, and the + both/neither rule has to hold across the wire.""" + from fastapi.testclient import TestClient + + from vpcopilot.console import app as A + c = TestClient(A.app) + assert c.post("/api/scan", json={"repo": "", "cve": ""}).status_code == 400 + assert c.post("/api/scan", json={"repo": "/x", "cve": "CVE-2024-23334"}).status_code == 400 + html = (__import__("pathlib").Path(A.__file__).parent / "static" / "index.html").read_text() + assert 'id="scanCve"' in html and "cve," in html