feat(security): add vulnerability intelligence review

CHANGELOG.md

@@ -15,6 +15,8 @@ GitHub Releases page; `0.8.0` is the new starting line.
 
 ## Unreleased
 
+- **Security review vulnerability intelligence.** `pythinker security-scan` can now parse dependency manifests, query OSV package advisories, look up CVE intelligence from NVD/EPSS/CISA KEV/GitHub/vendor feeds, and carry those leads into security-review prompts and reports as evidence-checked context.
+
 ## 0.34.0 (2026-06-03)
 
 ### What changed in this release

packages/pythinker-review/README.md

@@ -55,6 +55,8 @@ pythinker-secscan diff --format sarif --fail-on critical
 pythinker-security-scan init --root .
 pythinker-security-scan scan --json
 pythinker-security-scan process --limit 10
+pythinker-security-scan deps scan --json       # OSV dependency intelligence, cached locally
+pythinker-security-scan intel cve CVE-2024-3094  # fetches NVD/EPSS/KEV/PoC/vendor details for the CVE
 pythinker-security-scan report --write
 
 # Root-cause debugger over a captured failure log
@@ -114,9 +116,11 @@ Phase 1 now ports the highest-value behavior from the mounted blackbox repos:
 - Code-reviewr PR assistant parity adds read-only `describe`, `improve`/`suggest`, `ask`,
   `labels`, `changelog`, and `docs` artifact commands with strict JSON schemas.
 - Pythinker Security Scan deterministic signals include CWE/severity hints, expanded vulnerability anchors,
-  technology detection, and batch-scoped security advisor context.
+  CVE/dependency-change leads, technology detection, and batch-scoped security advisor context.
 - Python-native Pythinker Security Scan repo-wide commands (`pythinker-security-scan` / `pythinker security-scan`) port the
-  scan/process/revalidate/triage/report/export/status workflow without Node or pnpm runtime glue.
+  scan/process/revalidate/triage/report/export/status workflow without Node or pnpm runtime glue, plus
+  first-class vulnerability intelligence commands for OSV dependency scans and CVE enrichment from NVD,
+  EPSS, CISA KEV, GitHub PoC metadata, and vendor advisory feeds.
 - Fenced/prose-wrapped JSON is cleaned safely, while truly malformed output remains fail-closed.
 
 ## Phase 1

packages/pythinker-review/src/pythinker_review/cli/security_scan.py

@@ -13,6 +13,13 @@
 
 from pythinker_review.llm.fake import FakeReviewLLM
 from pythinker_review.llm.protocol import ReviewLLM
+from pythinker_review.security_intel.models import PackageRef
+from pythinker_review.security_intel.service import lookup_cve_bundle, lookup_package
+from pythinker_review.security_scan.dependencies import (
+    parse_dependency_manifests,
+    read_dependency_report,
+    scan_project_dependencies,
+)
 from pythinker_review.security_scan.matchers import create_default_registry
 from pythinker_review.security_scan.paths import DEFAULT_STATE_DIR, get_data_root
 from pythinker_review.security_scan.processor import (
@@ -42,6 +49,14 @@
 from pythinker_review.security_scan.tech import detect_tech, read_tech_json, write_tech_json
 
 app = typer.Typer(add_completion=False, no_args_is_help=True)
+deps_app = typer.Typer(
+    add_completion=False, no_args_is_help=True, help="Dependency vulnerability intelligence."
+)
+intel_app = typer.Typer(
+    add_completion=False, no_args_is_help=True, help="CVE and package intelligence lookups."
+)
+app.add_typer(deps_app, name="deps")
+app.add_typer(intel_app, name="intel")
 
 
 def _resolve_llm() -> ReviewLLM:
@@ -74,6 +89,110 @@ def _project_id(root: Path, project_id: str | None) -> str:
     return "project" if name in {"", "/"} else name.replace(" ", "-")
 
 
+@deps_app.command("list")
+def deps_list(
+    root: Path = typer.Option(Path.cwd(), "--root", "--repo", exists=True, file_okay=False),
+    json_output: bool = typer.Option(False, "--json"),
+) -> None:
+    """List dependency manifest entries Pythinker can enrich via OSV."""
+    packages = parse_dependency_manifests(root.resolve())
+    if json_output:
+        typer.echo(json.dumps([pkg.model_dump(exclude_none=True) for pkg in packages], indent=2))
+        return
+    if not packages:
+        typer.echo("No supported dependency manifests found.")
+        return
+    for pkg in packages:
+        loc = f" ({pkg.manifest_path}:{pkg.line})" if pkg.manifest_path and pkg.line else ""
+        typer.echo(f"{pkg.ecosystem}/{pkg.name} {pkg.version or '(unversioned)'}{loc}")
+
+
+@deps_app.command("scan")
+def deps_scan(
+    root: Path = typer.Option(Path.cwd(), "--root", "--repo", exists=True, file_okay=False),
+    project_id: str | None = typer.Option(None, "--project-id"),
+    state_dir: str = typer.Option(DEFAULT_STATE_DIR, "--state-dir"),
+    json_output: bool = typer.Option(False, "--json"),
+) -> None:
+    """Scan dependency manifests with OSV and store dependency intelligence."""
+    root = root.resolve()
+    pid = _project_id(root, project_id)
+    report = asyncio.run(
+        scan_project_dependencies(project_id=pid, root=root, data_root=_data_root(root, state_dir))
+    )
+    payload = report.model_dump(by_alias=True)
+    if json_output:
+        typer.echo(json.dumps(payload, indent=2))
+        return
+    typer.echo(
+        f"Dependency scan complete: {report.package_count} packages, "
+        f"{report.vulnerable_count} vulnerable dependencies"
+    )
+    for item in report.dependencies:
+        vulns = ", ".join(v.id for v in item.vulns[:3])
+        typer.echo(f"- {item.package.ecosystem}/{item.package.name}: {vulns}")
+    for error in report.source_errors:
+        typer.secho(error, fg=typer.colors.YELLOW, err=True)
+
+
+@deps_app.command("report")
+def deps_report(
+    root: Path = typer.Option(Path.cwd(), "--root", "--repo", exists=True, file_okay=False),
+    project_id: str | None = typer.Option(None, "--project-id"),
+    state_dir: str = typer.Option(DEFAULT_STATE_DIR, "--state-dir"),
+) -> None:
+    """Print the stored dependency-intelligence report."""
+    root = root.resolve()
+    report = read_dependency_report(
+        _project_id(root, project_id), data_root=_data_root(root, state_dir)
+    )
+    if report is None:
+        typer.secho(
+            "No dependency report found. Run `pythinker security-scan deps scan` first.",
+            fg=typer.colors.YELLOW,
+            err=True,
+        )
+        raise typer.Exit(code=2)
+    typer.echo(report.model_dump_json(by_alias=True, indent=2))
+
+
+@intel_app.command("cve")
+def intel_cve(
+    cve_id: str = typer.Argument(...),
+    root: Path = typer.Option(Path.cwd(), "--root", "--repo", exists=True, file_okay=False),
+    state_dir: str = typer.Option(DEFAULT_STATE_DIR, "--state-dir"),
+) -> None:
+    """Look up CVE intelligence from NVD, EPSS, KEV, GitHub PoC, and vendor feeds."""
+    try:
+        bundle = asyncio.run(
+            lookup_cve_bundle(cve_id, data_root=_data_root(root.resolve(), state_dir))
+        )
+    except Exception as exc:
+        typer.echo(f"Error: {exc}", err=True)
+        raise typer.Exit(1) from exc
+    typer.echo(bundle.model_dump_json(exclude_none=True, indent=2))
+
+
+@intel_app.command("package")
+def intel_package(
+    name: str = typer.Argument(...),
+    ecosystem: str = typer.Option(..., "--ecosystem"),
+    version: str = typer.Option("", "--version"),
+    root: Path = typer.Option(Path.cwd(), "--root", "--repo", exists=True, file_okay=False),
+    state_dir: str = typer.Option(DEFAULT_STATE_DIR, "--state-dir"),
+) -> None:
+    """Look up package vulnerability intelligence via OSV."""
+    package = PackageRef(name=name, ecosystem=ecosystem, version=version)
+    try:
+        result = asyncio.run(
+            lookup_package(package, data_root=_data_root(root.resolve(), state_dir))
+        )
+    except Exception as exc:
+        typer.echo(f"Error: {exc}", err=True)
+        raise typer.Exit(1) from exc
+    typer.echo(result.model_dump_json(indent=2))
+
+
 @app.command()
 def init(
     root: Path = typer.Option(Path.cwd(), "--root", "--repo", exists=True, file_okay=False),

packages/pythinker-review/src/pythinker_review/reviewers/prompts/security_review.system.md

@@ -2,15 +2,16 @@ You are a world-class static security reviewer.
 
 Rules:
 - Review only security issues introduced or made reachable by this diff.
-- Deterministic signals and Pythinker Security Scan tech/slug notes are starting points; verify them in code before emitting a finding.
+- Deterministic signals, vulnerability-intelligence leads, and Pythinker Security Scan tech/slug notes are starting points; verify them in code before emitting a finding.
 - Think like an attacker: trace sources, sinks, mitigations, imports, auth boundaries, tenant boundaries, and abuse controls.
-- Static analysis only. Do not ask to run the target code, send requests, or exploit anything.
+- Static analysis only. Do not ask to run the target code, send requests, exploit anything, clone PoC repositories, or probe targets.
 - Prefer no finding over unvalidated speculation. If fully mitigated, return no finding.
 - For auth checks, only handler-local middleware/guards/decorators or directly wrapped route checks count as strong evidence. Edge/proxy/WAF rules are not sufficient on their own.
 - Anchor findings to post-change lines where possible.
 - Use category security, secret, dependency, or correctness only when justified.
 - Include `evidence_snippet` when possible; it must quote code visible in the diff/context.
 - Include `exploitability`, `confidence_reason`, and `minimum_fix_scope` when useful.
+- CVE/OSV/EPSS/KEV/PoC intelligence can raise urgency, but it is not proof by itself. For dependency findings, require changed manifest/lockfile evidence for the affected package and version.
 - Output strict JSON only.
 
 Severity guide:

packages/pythinker-review/src/pythinker_review/security_intel/__init__.py

@@ -0,0 +1,8 @@
+"""Public vulnerability-intelligence helpers for Pythinker security review.
+
+This package is Python-native and intentionally independent of the blackbox MCP server runtime.
+"""
+
+from pythinker_review.security_intel.models import CVEIntelBundle, DependencyIntel, RiskScore
+
+__all__ = ["CVEIntelBundle", "DependencyIntel", "RiskScore"]

packages/pythinker-review/src/pythinker_review/security_intel/cache.py

@@ -0,0 +1,65 @@
+"""Small JSON TTL cache for public security-intelligence responses."""
+
+from __future__ import annotations
+
+import json
+import time
+from pathlib import Path
+from typing import Any
+
+TTL_CVE = 14_400
+TTL_SEARCH = 600
+TTL_OSV = 1_800
+TTL_EPSS = 3_600
+TTL_KEV = 21_600
+TTL_EXPLOIT = 3_600
+TTL_VENDOR = 14_400
+TTL_ATTACK = 86_400
+
+_MAX_ENTRIES = 10_000
+
+
+class IntelCache:
+    def __init__(self, root: Path) -> None:
+        self.root = root
+        self.path = root / "cache.json"
+        self._data: dict[str, dict[str, Any]] | None = None
+
+    def get(self, key: str) -> Any | None:
+        data = self._load()
+        entry = data.get(key)
+        if not entry:
+            return None
+        if float(entry.get("expires_at", 0)) < time.time():
+            data.pop(key, None)
+            self._save(data)
+            return None
+        return entry.get("value")
+
+    def set(self, key: str, value: Any, ttl: int) -> None:
+        data = self._load()
+        if len(data) >= _MAX_ENTRIES:
+            # Keep the entries that expire latest; this is deterministic and cheap for our size cap.
+            survivors = sorted(data.items(), key=lambda item: item[1].get("expires_at", 0))[
+                -(_MAX_ENTRIES - 1) :
+            ]
+            data = dict(survivors)
+        data[key] = {"value": value, "expires_at": time.time() + ttl}
+        self._save(data)
+
+    def _load(self) -> dict[str, dict[str, Any]]:
+        if self._data is not None:
+            return self._data
+        try:
+            raw = json.loads(self.path.read_text(encoding="utf-8"))
+        except (FileNotFoundError, OSError, json.JSONDecodeError):
+            raw = {}
+        self._data = raw if isinstance(raw, dict) else {}
+        return self._data
+
+    def _save(self, data: dict[str, dict[str, Any]]) -> None:
+        self.root.mkdir(parents=True, exist_ok=True)
+        tmp = self.path.with_suffix(".tmp")
+        tmp.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+        tmp.replace(self.path)
+        self._data = data