From 69697a12e62af616f94beee7e8da5329376cc59c Mon Sep 17 00:00:00 2001 From: Wolfvin Date: Thu, 2 Jul 2026 18:02:18 +0000 Subject: [PATCH 1/2] feat(deps-audit): dependency vulnerability scan via OSV.dev API (closes #158) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add new `deps-audit` command — pure-Python dependency audit that queries the OSV.dev batch API (`/v1/querybatch`) for known CVEs across three ecosystems: PyPI, npm, crates.io. Files added: - scripts/commands/deps_audit.py — thin CLI wrapper (follows secrets.py pattern) - scripts/dep_audit_engine.py — engine: manifest/lock parsing, OSV client, CVSS v3 severity bucketing, graph persistence, recommendations - tests/test_deps_audit.py — 47 tests covering parsing, offline mode, ecosystem filter, lock-file preference, mocked OSV queries, CVSS vector parsing, severity filter, graph persistence (idempotent), edge cases Schema additions (graph_model.py): - NODE_TYPE_DEPENDENCY_VULN = 'dependency_vuln' - EDGE_TYPE_HAS_VULN = 'HAS_VULN' - No DDL change — node_type column is TEXT, accepts new value as-is Features per issue #158 DoD: - Auto-detects requirements.txt / pyproject.toml / Pipfile (PyPI), package.json + package-lock.json/yarn.lock/pnpm-lock.yaml (npm), Cargo.toml + Cargo.lock (crates.io) - Lock file preferred over manifest when both present (pinned versions) - OSV batch API with 1000-query chunks, retry+backoff on 429/5xx - Vuln detail cache (1 GET per unique vuln ID across the run) - Severity bucketed from database_specific.severity OR CVSS v3 vector (pure-Python CVSS base score computation, no external library) - --severity filter (includes higher severities — high returns high+critical) - --ecosystem filter (PyPI / npm / crates.io) - --offline mode (skip API, return packages-only with status='offline') - Findings persisted as dependency_vuln graph nodes + HAS_VULN edges; idempotent re-scan (deletes previous findings for the same source file) - Standard --format support (json / markdown / ai / sarif / compact) - Zero external binary deps — pure stdlib (urllib, json, sqlite3, tomllib) - Network failures graceful: status stays 'ok', api_errors field surfaces partial failures, NEVER raises sync_command_count.py --apply run: command count 70 -> 71, MCP tools 68 -> 69. README/SKILL/SKILL-QUICK updated with new command entry. Tests: 47 passed (deps_audit), 235 passed + 39 skipped (broader regression sweep across command_registry, command_count, graph_model, persistent_registry, registry, secrets_engine, vuln_staleness, config_secret_redaction). --- scripts/codelens.py | 3 +- scripts/commands/deps_audit.py | 52 ++ scripts/dep_audit_engine.py | 1350 ++++++++++++++++++++++++++++++++ tests/test_deps_audit.py | 812 +++++++++++++++++++ 4 files changed, 2216 insertions(+), 1 deletion(-) create mode 100644 scripts/commands/deps_audit.py create mode 100644 scripts/dep_audit_engine.py create mode 100644 tests/test_deps_audit.py diff --git a/scripts/codelens.py b/scripts/codelens.py index 9e0933e8..0eb3235a 100755 --- a/scripts/codelens.py +++ b/scripts/codelens.py @@ -395,7 +395,7 @@ def _auto_setup(workspace: str) -> Dict[str, Any]: "secrets", "a11y", "css-deep", "regex-audit", "vuln-scan", "side-effect", "missing-refs", "circular", "list", "env-check", "test-map", "ownership", "entrypoints", "api-map", "state-map", - "dataflow", "search", "symbols", "summary", "taint", + "dataflow", "search", "symbols", "summary", "taint", "deps-audit", } # Sort strategies per command for --top (sort by relevance before truncating) @@ -406,6 +406,7 @@ def _auto_setup(workspace: str) -> Dict[str, Any]: "perf-hint": ("severity", True), # sort by severity desc "secrets": ("severity", True), # sort by severity desc "vuln-scan": ("severity", True), # sort by severity desc + "deps-audit": ("severity", True), # sort by severity desc (issue #158) "a11y": ("severity", True), # sort by severity desc "css-deep": ("severity", True), # sort by severity desc "regex-audit": ("severity", True), # sort by severity desc diff --git a/scripts/commands/deps_audit.py b/scripts/commands/deps_audit.py new file mode 100644 index 00000000..97d87605 --- /dev/null +++ b/scripts/commands/deps_audit.py @@ -0,0 +1,52 @@ +# @WHO: scripts/commands/deps_audit.py +# @WHAT: deps-audit CLI command — dependency vulnerability scan via OSV.dev +# @PART: commands +# @ENTRY: execute() +"""Deps-audit command — Scan dependencies for known CVEs via OSV.dev API.""" + +from dep_audit_engine import audit_dependencies +from commands import register_command + + +def add_args(parser): + parser.add_argument( + "workspace", + nargs="?", + default=None, + help="Path to workspace root (auto-detected if omitted)", + ) + parser.add_argument( + "--severity", + choices=["critical", "high", "medium", "low"], + default=None, + help="Filter by severity (includes higher severities)", + ) + parser.add_argument( + "--ecosystem", + choices=["PyPI", "npm", "crates.io"], + default=None, + help="Limit scan to one package ecosystem", + ) + parser.add_argument( + "--offline", + action="store_true", + default=False, + help="Skip OSV API queries (report packages only, no findings)", + ) + + +def execute(args, workspace): + return audit_dependencies( + workspace, + severity=args.severity, + ecosystem=args.ecosystem, + offline=args.offline, + ) + + +register_command( + "deps-audit", + "Scan dependencies for known CVEs via OSV.dev (PyPI/npm/crates.io)", + add_args, + execute, +) diff --git a/scripts/dep_audit_engine.py b/scripts/dep_audit_engine.py new file mode 100644 index 00000000..0f7d08e9 --- /dev/null +++ b/scripts/dep_audit_engine.py @@ -0,0 +1,1350 @@ +# @WHO: scripts/dep_audit_engine.py +# @WHAT: Dependency audit engine — scans manifests/lockfiles for known CVEs via OSV.dev +# @PART: engine +# @ENTRY: audit_dependencies() +""" +Dependency Audit Engine for CodeLens. + +Scans dependency manifests (requirements.txt, package.json, Cargo.toml) and +lock files (package-lock.json, yarn.lock, Cargo.lock) for packages with known +vulnerabilities, using the OSV.dev batch API. + +Answers: "Do any of my dependencies have known CVEs?" + +Architecture (issue #158): +- Manifest / lock-file parsing: pure Python, no external binary deps +- OSV batch API: POST /v1/querybatch (up to 1000 queries per request) +- Per-vuln details: GET /v1/vulns/{id} (cached in-memory for the run) +- Severity mapping: CVSS v3 vector → critical/high/medium/low +- Persistence: findings stored in SQLite graph as `dependency_vuln` nodes + linked to the source file via `HAS_VULN` edges (graph_model.py constants) + +Design invariants: +- Zero external binary dependency (pure-stdlib: urllib, json, sqlite3, tomllib) +- Network failures are graceful: return status "offline" with empty findings, + NEVER raise. Caller (MCP / CLI) decides what to do with status. +- No telemetry, no PII. Only package names + versions are sent to OSV.dev + (which is the minimum required to query vulnerabilities). + +Issue: #158 +""" + +import json +import os +import sqlite3 +import sys +import time +import urllib.error +import urllib.request +from typing import Any, Dict, List, Optional, Tuple + +from graph_model import ( + EDGE_TYPE_HAS_VULN, + GRAPH_EDGES_TABLE, + GRAPH_NODES_TABLE, + NODE_TYPE_DEPENDENCY_VULN, + NODE_TYPE_FILE, + init_graph_schema, +) +from utils import default_db_path, logger + +# ─── Constants ───────────────────────────────────────────────── + +OSV_BATCH_URL = "https://api.osv.dev/v1/querybatch" +OSV_VULN_URL = "https://api.osv.dev/v1/vulns/{vuln_id}" + +# OSV batch endpoint accepts up to 1000 queries per request. +OSV_BATCH_SIZE = 1000 + +# Network timeouts and retry behavior. +HTTP_TIMEOUT_SECONDS = 30 +MAX_RETRIES = 3 +RETRY_BACKOFF_BASE = 1.5 # seconds; doubled each retry + +# Severity ranking — CVSS v3 base score → bucket. +# Per CVSS v3 spec: 0.1-3.9 low, 4.0-6.9 medium, 7.0-8.9 high, 9.0-10.0 critical. +SEVERITY_ORDER = {"critical": 0, "high": 1, "medium": 2, "low": 3, "unknown": 4} + +# Ecosystems supported by deps-audit (issue #158 scope). +SUPPORTED_ECOSYSTEMS = ("PyPI", "npm", "crates.io") + +# Map of ecosystem → (manifest files, lock files). +# Lock files are preferred when present because they pin exact versions. +_ECOSYSTEM_FILES: Dict[str, Tuple[Tuple[str, ...], Tuple[str, ...]]] = { + "PyPI": ( + ("requirements.txt", "pyproject.toml", "Pipfile"), + ("requirements.txt",), # requirements.txt is itself the lock file + ), + "npm": ( + ("package.json",), + ("package-lock.json", "yarn.lock", "pnpm-lock.yaml"), + ), + "crates.io": ( + ("Cargo.toml",), + ("Cargo.lock",), + ), +} + + +# ─── Public API ──────────────────────────────────────────────── + + +def audit_dependencies( + workspace: str, + severity: Optional[str] = None, + ecosystem: Optional[str] = None, + offline: bool = False, + db_path: Optional[str] = None, +) -> Dict[str, Any]: + """Scan workspace dependencies for known vulnerabilities via OSV.dev. + + Args: + workspace: Absolute path to workspace root. + severity: Optional filter — only return findings at this severity or + higher. One of "critical", "high", "medium", "low". + ecosystem: Optional filter — only scan one ecosystem. One of + "PyPI", "npm", "crates.io". None = scan all. + offline: If True, skip OSV API queries. Returns the set of + packages that *would* be checked plus a status of "offline" + and an empty findings list. + db_path: Optional SQLite db path for graph persistence. Defaults + to `/.codelens/codelens.db`. + + Returns: + Dict with keys: + - status: "ok" | "offline" | "error" + - workspace: the resolved workspace path + - severity_filter: the severity filter (or None) + - ecosystem_filter: the ecosystem filter (or None) + - stats: {total, critical, high, medium, low, unknown, + packages_scanned, ecosystems_scanned} + - findings: list of finding dicts (see _build_finding) + - recommendations: list of actionable strings + - files_scanned: list of manifest/lock file paths read + - packages_scanned: list of {name, version, ecosystem, source_file} + """ + workspace = os.path.abspath(workspace) + + # ─── Discover dependency files ───────────────────────────── + ecosystems_to_scan = ( + (ecosystem,) if ecosystem else SUPPORTED_ECOSYSTEMS + ) + files_scanned: List[str] = [] + packages: List[Dict[str, str]] = [] + + for eco in ecosystems_to_scan: + manifests, lockfiles = _ECOSYSTEM_FILES[eco] + eco_packages, eco_files = _collect_packages_for_ecosystem( + workspace, eco, manifests, lockfiles + ) + packages.extend(eco_packages) + files_scanned.extend(eco_files) + + # Deduplicate packages by (name, version, ecosystem) — a package may + # appear in both manifest and lock file. + seen = set() + deduped_packages: List[Dict[str, str]] = [] + for pkg in packages: + key = (pkg["name"], pkg["version"], pkg["ecosystem"]) + if key in seen: + continue + seen.add(key) + deduped_packages.append(pkg) + packages = deduped_packages + + # ─── Offline short-circuit ───────────────────────────────── + if offline: + return { + "status": "offline", + "workspace": workspace, + "severity_filter": severity, + "ecosystem_filter": ecosystem, + "stats": { + "total": 0, + "critical": 0, + "high": 0, + "medium": 0, + "low": 0, + "unknown": 0, + "packages_scanned": len(packages), + "ecosystems_scanned": len(ecosystems_to_scan), + }, + "findings": [], + "recommendations": [ + "Offline mode: skipped OSV API queries. Re-run without " + "--offline to scan for known vulnerabilities." + ], + "files_scanned": sorted(set(files_scanned)), + "packages_scanned": packages, + } + + if not packages: + return { + "status": "ok", + "workspace": workspace, + "severity_filter": severity, + "ecosystem_filter": ecosystem, + "stats": { + "total": 0, + "critical": 0, + "high": 0, + "medium": 0, + "low": 0, + "unknown": 0, + "packages_scanned": 0, + "ecosystems_scanned": len(ecosystems_to_scan), + }, + "findings": [], + "recommendations": [ + "No dependency manifests found. Supported files: " + "requirements.txt, pyproject.toml, Pipfile, package.json, " + "package-lock.json, yarn.lock, pnpm-lock.yaml, Cargo.toml, " + "Cargo.lock." + ], + "files_scanned": [], + "packages_scanned": [], + } + + # ─── Query OSV.dev ───────────────────────────────────────── + raw_findings, api_errors = _query_osv_batch(packages) + + # ─── Apply severity filter ───────────────────────────────── + if severity: + threshold = SEVERITY_ORDER.get(severity, 99) + filtered: List[Dict[str, Any]] = [] + for f in raw_findings: + f_sev = SEVERITY_ORDER.get(f.get("severity", "unknown"), 4) + if f_sev <= threshold: + filtered.append(f) + findings = filtered + else: + findings = raw_findings + + # ─── Compute stats ───────────────────────────────────────── + stats = _compute_stats(findings, packages, ecosystems_to_scan) + + # ─── Persist to SQLite graph ─────────────────────────────── + persistence_info = _persist_findings_to_graph( + workspace, db_path, findings, packages + ) + + # ─── Recommendations ─────────────────────────────────────── + recommendations = _generate_recommendations(findings, api_errors) + + result: Dict[str, Any] = { + "status": "ok", + "workspace": workspace, + "severity_filter": severity, + "ecosystem_filter": ecosystem, + "stats": stats, + "findings": findings, + "recommendations": recommendations, + "files_scanned": sorted(set(files_scanned)), + "packages_scanned": packages, + } + if persistence_info: + result["persistence"] = persistence_info + if api_errors: + # Surface non-fatal API errors (e.g. partial batch failure) so the + # caller knows the result may be incomplete. status stays "ok" because + # we did return *some* findings. + result["api_errors"] = api_errors + return result + + +# ─── Manifest / Lock Parsing ─────────────────────────────────── + + +def _collect_packages_for_ecosystem( + workspace: str, + ecosystem: str, + manifests: Tuple[str, ...], + lockfiles: Tuple[str, ...], +) -> Tuple[List[Dict[str, str]], List[str]]: + """Find and parse dependency files for one ecosystem. + + Lock files are preferred when present (pinned versions). Falls back to + manifest files otherwise. Files that don't exist are silently skipped. + + Returns: + (packages, files_scanned) — packages is a list of dicts with keys + {name, version, ecosystem, source_file}. files_scanned is the list + of file paths that were actually read. + """ + packages: List[Dict[str, str]] = [] + files_scanned: List[str] = [] + + # Try lock files first — they have exact pinned versions. + for lock_name in lockfiles: + lock_path = os.path.join(workspace, lock_name) + if not os.path.isfile(lock_path): + continue + try: + parsed = _parse_lock_file(lock_path, ecosystem) + except Exception as e: + logger.warning( + f"[dep_audit] failed to parse lock file {lock_path}: {e}" + ) + continue + for name, version in parsed: + packages.append( + { + "name": name, + "version": version, + "ecosystem": ecosystem, + "source_file": lock_name, + } + ) + files_scanned.append(lock_name) + + # If we got packages from a lock file, don't also parse the manifest — + # lock-file versions are more accurate. + if packages: + return packages, files_scanned + + # No lock file (or lock file empty) — fall back to manifests. + for manifest_name in manifests: + manifest_path = os.path.join(workspace, manifest_name) + if not os.path.isfile(manifest_path): + continue + try: + parsed = _parse_manifest_file(manifest_path, ecosystem) + except Exception as e: + logger.warning( + f"[dep_audit] failed to parse manifest {manifest_path}: {e}" + ) + continue + for name, version in parsed: + packages.append( + { + "name": name, + "version": version, + "ecosystem": ecosystem, + "source_file": manifest_name, + } + ) + files_scanned.append(manifest_name) + + return packages, files_scanned + + +def _parse_lock_file(path: str, ecosystem: str) -> List[Tuple[str, str]]: + """Dispatch to the right parser for a lock file. + + Args: + path: Absolute path to the lock file. + ecosystem: One of "PyPI", "npm", "crates.io". + + Returns: + List of (name, version) tuples. Empty list on parse failure. + """ + name = os.path.basename(path) + with open(path, "r", encoding="utf-8", errors="replace") as f: + content = f.read() + + if ecosystem == "PyPI" and name == "requirements.txt": + return _parse_requirements_txt(content) + if ecosystem == "npm": + if name == "package-lock.json": + return _parse_package_lock_json(content) + if name == "yarn.lock": + return _parse_yarn_lock(content) + if name == "pnpm-lock.yaml": + return _parse_pnpm_lock_yaml(content) + if ecosystem == "crates.io" and name == "Cargo.lock": + return _parse_cargo_lock(content) + + logger.warning( + f"[dep_audit] no parser for lock file {name} (ecosystem={ecosystem})" + ) + return [] + + +def _parse_manifest_file(path: str, ecosystem: str) -> List[Tuple[str, str]]: + """Dispatch to the right parser for a manifest file.""" + name = os.path.basename(path) + with open(path, "r", encoding="utf-8", errors="replace") as f: + content = f.read() + + if ecosystem == "PyPI": + if name == "requirements.txt": + return _parse_requirements_txt(content) + if name == "pyproject.toml": + return _parse_pyproject_toml(content) + if name == "Pipfile": + return _parse_pipfile(content) + if ecosystem == "npm" and name == "package.json": + return _parse_package_json(content) + if ecosystem == "crates.io" and name == "Cargo.toml": + return _parse_cargo_toml(content) + + logger.warning( + f"[dep_audit] no parser for manifest {name} (ecosystem={ecosystem})" + ) + return [] + + +# ─── Per-format Parsers ──────────────────────────────────────── + + +_REQUIREMENT_LINE_RE = None # compiled lazily to keep import side-effects minimal + + +def _parse_requirements_txt(content: str) -> List[Tuple[str, str]]: + """Parse requirements.txt — supports `name==version`, `name>=version`, + `name~=version`, and bare `name` (version becomes ""). + Skips comments, blank lines, options (-r, --, -e), and VCS URLs. + """ + import re + + global _REQUIREMENT_LINE_RE + if _REQUIREMENT_LINE_RE is None: + _REQUIREMENT_LINE_RE = re.compile( + r"^\s*([A-Za-z0-9_.\-]+)\s*(?:==|>=|~=|<=|>|<|=)?\s*" + r"([A-Za-z0-9_.\-*+!]+)?" + ) + + out: List[Tuple[str, str]] = [] + for raw_line in content.splitlines(): + line = raw_line.split("#", 1)[0].strip() + if not line: + continue + if line.startswith(("-", "git+", "http://", "https://", "svn+", "hg+")): + continue + # Strip environment markers (e.g. "; python_version >= '3.8'") + line = line.split(";", 1)[0].strip() + # Strip extras (e.g. "package[extra1,extra2]") + line = re.sub(r"\[[^\]]+\]", "", line) + match = _REQUIREMENT_LINE_RE.match(line) + if not match: + continue + name = match.group(1) + version = (match.group(2) or "").strip() + if not name: + continue + out.append((name, version)) + return out + + +def _parse_package_json(content: str) -> List[Tuple[str, str]]: + """Parse package.json — extracts dependencies + devDependencies. + Versions are npm semver specs (e.g. "^1.2.3", "~2.0.0", ">=3.0.0"). + The leading operator is stripped to give a best-effort version. + """ + try: + data = json.loads(content) + except json.JSONDecodeError as e: + logger.warning(f"[dep_audit] invalid package.json: {e}") + return [] + out: List[Tuple[str, str]] = [] + for section in ("dependencies", "devDependencies", "optionalDependencies"): + section_deps = data.get(section) or {} + if not isinstance(section_deps, dict): + continue + for name, spec in section_deps.items(): + if not isinstance(spec, str): + continue + # Skip git/file/link specs — only registry versions are checkable. + if spec.startswith(("file:", "git+", "link:", "http:", "https:")): + continue + version = _strip_npm_operator(spec) + out.append((name, version)) + return out + + +def _parse_package_lock_json(content: str) -> List[Tuple[str, str]]: + """Parse package-lock.json (npm v1/v2/v3) — extracts `packages` and + `dependencies` sections, returning pinned (name, version) pairs. + """ + try: + data = json.loads(content) + except json.JSONDecodeError as e: + logger.warning(f"[dep_audit] invalid package-lock.json: {e}") + return [] + out: List[Tuple[str, str]] = [] + + # Lockfile v2/v3: "packages" dict keyed by "node_modules/..." paths + packages_section = data.get("packages") or {} + if isinstance(packages_section, dict): + for path_key, info in packages_section.items(): + if not isinstance(info, dict): + continue + # Skip the root package itself (key is ""). + if not path_key: + continue + # Skip bundled / optional / workspace packages without a real version. + version = info.get("version") + if not version: + continue + # Extract the package name from the path key (last segment after + # the last "node_modules/"). + name = path_key.split("node_modules/")[-1] + if not name: + continue + out.append((name, version)) + + # Lockfile v1: "dependencies" dict (flat). + deps_section = data.get("dependencies") or {} + if isinstance(deps_section, dict): + for name, info in deps_section.items(): + if not isinstance(info, dict): + continue + version = info.get("version") + if not version: + continue + out.append((name, version)) + + # Deduplicate — v2 lockfiles have both sections pointing at the same packages. + seen = set() + deduped: List[Tuple[str, str]] = [] + for name, version in out: + if (name, version) in seen: + continue + seen.add((name, version)) + deduped.append((name, version)) + return deduped + + +def _parse_yarn_lock(content: str) -> List[Tuple[str, str]]: + """Parse yarn.lock (v1 format). Best-effort — handles the common case + of `name@version:` blocks with a `version "X.Y.Z"` line below. + """ + out: List[Tuple[str, str]] = [] + current_names: List[str] = [] + for raw_line in content.splitlines(): + line = raw_line.rstrip() + if not line: + current_names = [] + continue + # Header line: 'name@spec, name@spec:' or '"name@spec":' + if line and (line.endswith(":") or line.rstrip().endswith(":")) and "@" in line: + header = line.rstrip().rstrip(":") + # Split multiple aliases: "name@^1.0.0, name@^1.0.1" + current_names = [] + for alias in header.split(","): + alias = alias.strip().strip('"') + if "@" in alias: + # name is everything before the LAST @ (handles scoped @org/pkg) + name = alias.rsplit("@", 1)[0] + if name: + current_names.append(name) + continue + # Version line: ' version "1.2.3"' + stripped = line.strip() + if stripped.startswith("version ") and current_names: + version = stripped[len("version "):].strip().strip('"').strip("'") + if version: + for name in current_names: + out.append((name, version)) + current_names = [] + return out + + +def _parse_pnpm_lock_yaml(content: str) -> List[Tuple[str, str]]: + """Parse pnpm-lock.yaml (v9+ format). pnpm v9 uses a `packages:` section + with entries like `'@scope/pkg@1.2.3':` followed by metadata. + Falls back to a regex scrape if the structure is unusual. + + Pure-Python parser — no PyYAML dependency. Handles the common v9 layout. + For v5/v6 lockfiles, the `dependencies:` section is parsed instead. + """ + import re + + out: List[Tuple[str, str]] = [] + # v9+: scan for `packages:` section entries. + in_packages = False + pkg_re = re.compile(r"^\s*'([^']+)@([^@']+)':\s*$") + for raw_line in content.splitlines(): + line = raw_line.rstrip() + if line.startswith("packages:") or line == "packages:": + in_packages = True + continue + if in_packages: + # Next top-level key (no leading whitespace, ends with `:`) ends + # the packages section. + if line and not line[0].isspace() and line.endswith(":"): + in_packages = False + continue + match = pkg_re.match(line) + if match: + name, version = match.group(1), match.group(2) + # Strip any path / peer-dep suffix after the version. + version = version.split("(")[0].strip() + if name and version: + out.append((name, version)) + if out: + return out + + # Fallback: v5/v6 `dependencies:` section with `version: X.Y.Z` lines. + in_deps = False + current_name = "" + for raw_line in content.splitlines(): + line = raw_line.rstrip() + if line.startswith("dependencies:") or line == "dependencies:": + in_deps = True + continue + if in_deps: + if line and not line[0].isspace() and line.endswith(":"): + in_deps = False + continue + stripped = line.strip() + if stripped.endswith(":") and not stripped.startswith("version"): + # 'name:' header + current_name = stripped.rstrip(":").strip("'\"") + elif stripped.startswith("version:") and current_name: + version = stripped[len("version:"):].strip().strip("'\"") + if version: + out.append((current_name, version)) + current_name = "" + return out + + +def _parse_cargo_lock(content: str) -> List[Tuple[str, str]]: + """Parse Cargo.lock — TOML-ish format with `[[package]]` sections + containing `name = "..."` and `version = "..."` lines. + """ + out: List[Tuple[str, str]] = [] + current_name: Optional[str] = None + current_version: Optional[str] = None + in_package = False + for raw_line in content.splitlines(): + line = raw_line.strip() + if line == "[[package]]": + if current_name and current_version: + out.append((current_name, current_version)) + current_name = None + current_version = None + in_package = True + continue + if not in_package: + continue + if line.startswith("name = "): + current_name = _unquote(line[len("name = "):]) + elif line.startswith("version = "): + current_version = _unquote(line[len("version = "):]) + elif line.startswith("[") and line != "[[package]]": + # Entering a different section inside the package (e.g. [package.dependencies]) + # — but a new [[package]] was already handled above, so just ignore. + pass + # Flush the last package + if current_name and current_version: + out.append((current_name, current_version)) + return out + + +def _parse_pyproject_toml(content: str) -> List[Tuple[str, str]]: + """Parse pyproject.toml — extracts `dependencies` array under + `[project]` and `[tool.poetry.dependencies]`. Uses tomllib (Python 3.11+). + """ + import tomllib + + try: + data = tomllib.loads(content) + except tomllib.TOMLDecodeError as e: + logger.warning(f"[dep_audit] invalid pyproject.toml: {e}") + return [] + + out: List[Tuple[str, str]] = [] + + # PEP 621: [project] dependencies = ["name>=1.0", ...] + project = data.get("project") or {} + project_deps = project.get("dependencies") or [] + if isinstance(project_deps, list): + for spec in project_deps: + name, version = _split_pep508_spec(spec) + if name: + out.append((name, version)) + + # Poetry: [tool.poetry.dependencies] — dict of name → version spec + poetry = (data.get("tool") or {}).get("poetry") or {} + poetry_deps = poetry.get("dependencies") or {} + if isinstance(poetry_deps, dict): + for name, spec in poetry_deps.items(): + if name == "python": + continue + if isinstance(spec, str): + version = _strip_python_operator(spec) + out.append((name, version)) + elif isinstance(spec, dict) and "version" in spec: + version = _strip_python_operator(str(spec["version"])) + out.append((name, version)) + + return out + + +def _parse_pipfile(content: str) -> List[Tuple[str, str]]: + """Parse Pipfile (TOML) — `[packages]` and `[dev-packages]` sections.""" + import tomllib + + try: + data = tomllib.loads(content) + except tomllib.TOMLDecodeError as e: + logger.warning(f"[dep_audit] invalid Pipfile: {e}") + return [] + out: List[Tuple[str, str]] = [] + for section in ("packages", "dev-packages"): + section_deps = data.get(section) or {} + if not isinstance(section_deps, dict): + continue + for name, spec in section_deps.items(): + if isinstance(spec, str): + out.append((name, _strip_python_operator(spec))) + elif isinstance(spec, dict) and "version" in spec: + out.append((name, _strip_python_operator(str(spec["version"])))) + return out + + +def _parse_cargo_toml(content: str) -> List[Tuple[str, str]]: + """Parse Cargo.toml — `[dependencies]` and `[dev-dependencies]` sections. + Supports both `name = "version"` and `name = { version = "X", ... }` forms. + """ + import tomllib + + try: + data = tomllib.loads(content) + except tomllib.TOMLDecodeError as e: + logger.warning(f"[dep_audit] invalid Cargo.toml: {e}") + return [] + out: List[Tuple[str, str]] = [] + for section in ("dependencies", "dev-dependencies", "build-dependencies"): + section_deps = data.get(section) or {} + if not isinstance(section_deps, dict): + continue + for name, spec in section_deps.items(): + if isinstance(spec, str): + out.append((name, _strip_cargo_operator(spec))) + elif isinstance(spec, dict) and "version" in spec: + out.append( + (name, _strip_cargo_operator(str(spec["version"]))) + ) + return out + + +# ─── Spec-string Helpers ─────────────────────────────────────── + + +def _strip_npm_operator(spec: str) -> str: + """Strip npm semver operators (^, ~, >=, >, <=, <, =) and any range + whitespace. Returns the bare version like '1.2.3' or '' for wildcards. + """ + spec = spec.strip() + if not spec: + return "" + if spec.startswith(("^", "~", ">", "<", "=")): + spec = spec.lstrip("^~><= ") + # Handle '1.2.x' / '*' / 'latest' — these can't be queried. + if spec in ("*", "latest", "") or "x" in spec.lower(): + return "" + # Take only the first version if it's a range (e.g. "1.2.3 || 2.0.0"). + spec = spec.split(" ", 1)[0].split("||", 1)[0].strip() + return spec + + +def _strip_python_operator(spec: str) -> str: + """Strip PEP 440 operators from a version spec. Returns '' for wildcards.""" + import re + + spec = spec.strip() + if not spec: + return "" + spec = re.sub(r"^[=<>!~^]+", "", spec).strip() + if spec in ("*", "any", ""): + return "" + return spec + + +def _strip_cargo_operator(spec: str) -> str: + """Strip Cargo version operators (^, ~, *, >=). Returns '' for wildcards.""" + spec = spec.strip() + if not spec or spec == "*": + return "" + if spec.startswith(("^", "~", ">", "<", "=")): + spec = spec.lstrip("^~><= ") + if "*" in spec: + return "" + return spec + + +def _split_pep508_spec(spec: str) -> Tuple[str, str]: + """Split a PEP 508 requirement spec like 'name>=1.0,<2.0; python_version>"3.8"' + into (name, version) where version is the first pinned version, or '' if + only ranges/wildcards are given. + """ + import re + + if not isinstance(spec, str): + return ("", "") + # Strip environment markers. + spec = spec.split(";", 1)[0].strip() + # Strip extras. + spec = re.sub(r"\[[^\]]+\]", "", spec) + match = re.match( + r"^\s*([A-Za-z0-9_.\-]+)\s*(?:\s*(==|>=|~=|<=|>|<|=)\s*([A-Za-z0-9_.\-*+!]+))?", + spec, + ) + if not match: + return ("", "") + name = match.group(1) + op = match.group(2) + version = match.group(3) or "" + # Only == and ~= give a checkable version. >= / > / <= / < are ranges. + # We still send the bound version to OSV — it'll just return vulns that + # affect that version, which is the conservative behavior. + if version in ("*", "any"): + version = "" + return (name, version) + + +def _unquote(s: str) -> str: + """Strip surrounding double or single quotes from a TOML value string.""" + s = s.strip() + if len(s) >= 2 and s[0] == s[-1] and s[0] in ('"', "'"): + return s[1:-1] + return s + + +# ─── OSV API Client ──────────────────────────────────────────── + + +def _query_osv_batch( + packages: List[Dict[str, str]], +) -> Tuple[List[Dict[str, Any]], List[str]]: + """Query OSV.dev batch endpoint for all packages. + + Splits `packages` into chunks of OSV_BATCH_SIZE, sends each chunk as a + POST to /v1/querybatch, then for each returned vuln ID fetches full + details via GET /v1/vulns/{id}. Vuln details are cached per-run to avoid + refetching across packages that share vulns. + + Args: + packages: List of {name, version, ecosystem, source_file} dicts. + + Returns: + (findings, api_errors) — findings is a list of dicts with keys + {package, version, ecosystem, source_file, vuln_id, severity, + fixed_in, summary, osv_url}. api_errors is a list of human-readable + error strings (empty if all requests succeeded). + """ + findings: List[Dict[str, Any]] = [] + api_errors: List[str] = [] + vuln_cache: Dict[str, Dict[str, Any]] = {} + + # Skip packages with empty version — OSV requires a version to query. + queryable = [p for p in packages if p.get("version")] + skipped = len(packages) - len(queryable) + if skipped: + api_errors.append( + f"[dep_audit] {skipped} package(s) skipped — missing or " + f"unparseable version (wildcards/ranges not supported by OSV)." + ) + + for chunk_start in range(0, len(queryable), OSV_BATCH_SIZE): + chunk = queryable[chunk_start : chunk_start + OSV_BATCH_SIZE] + batch_payload = {"queries": [_build_osv_query(p) for p in chunk]} + + try: + batch_resp = _http_post_json(OSV_BATCH_URL, batch_payload) + except Exception as e: + msg = ( + f"[dep_audit] OSV batch query failed for chunk " + f"{chunk_start}-{chunk_start + len(chunk)}: {e}" + ) + logger.error(msg) + api_errors.append(msg) + continue + + results = batch_resp.get("results") or [] + for pkg, result in zip(chunk, results): + vulns = result.get("vulns") or [] + for vuln_entry in vulns: + vuln_id = vuln_entry.get("id") + if not vuln_id: + continue + # Fetch full vuln details (cached). + if vuln_id not in vuln_cache: + try: + vuln_cache[vuln_id] = _fetch_osv_vuln(vuln_id) + except Exception as e: + msg = ( + f"[dep_audit] OSV vuln detail fetch failed for " + f"{vuln_id}: {e}" + ) + logger.warning(msg) + api_errors.append(msg) + vuln_cache[vuln_id] = {} + details = vuln_cache[vuln_id] + findings.append( + _build_finding(pkg, vuln_id, details) + ) + + return findings, api_errors + + +def _build_osv_query(pkg: Dict[str, str]) -> Dict[str, Any]: + """Build a single OSV query object for one package.""" + return { + "package": {"name": pkg["name"], "ecosystem": pkg["ecosystem"]}, + "version": pkg["version"], + } + + +def _fetch_osv_vuln(vuln_id: str) -> Dict[str, Any]: + """GET /v1/vulns/{id} — returns full vuln record with severity, + affected ranges, and summaries. Retries with exponential backoff on + transient errors (5xx, network). + """ + url = OSV_VULN_URL.format(vuln_id=vuln_id) + return _http_get_json(url) + + +def _build_finding( + pkg: Dict[str, str], vuln_id: str, details: Dict[str, Any] +) -> Dict[str, Any]: + """Assemble a finding dict from package + OSV vuln details. + + Extracts: + - severity: from database_specific.severity OR CVSS v3 base score + - fixed_in: first fixed version from affected[].ranges[].events + - summary: vuln summary field + - cve_ids: list of CVE aliases + """ + severity = _extract_severity(details) + fixed_in = _extract_fixed_version(details, pkg) + summary = details.get("summary") or "" + aliases = details.get("aliases") or [] + cve_ids = [a for a in aliases if a.startswith("CVE-")] + + return { + "package": pkg["name"], + "version": pkg["version"], + "ecosystem": pkg["ecosystem"], + "source_file": pkg["source_file"], + "vuln_id": vuln_id, + "severity": severity, + "fixed_in": fixed_in, + "summary": summary, + "cve_ids": cve_ids, + "osv_url": f"https://osv.dev/vulnerability/{vuln_id}", + } + + +def _extract_severity(details: Dict[str, Any]) -> str: + """Extract severity bucket from an OSV vuln record. + + Priority: + 1. database_specific.severity (e.g. "CRITICAL", "HIGH") if it matches + a known bucket. + 2. severity[].score string — parse CVSS v3 base score from the vector. + 3. "unknown" if neither is present. + """ + db_specific = details.get("database_specific") or {} + if isinstance(db_specific, dict): + raw_sev = db_specific.get("severity") + if isinstance(raw_sev, str): + bucket = _normalize_severity_string(raw_sev) + if bucket != "unknown": + return bucket + # Some ecosystems put severity under severity[].type + # Try severity[] array (CVSS vectors) + for sev_entry in details.get("severity") or []: + if not isinstance(sev_entry, dict): + continue + score_str = sev_entry.get("score") or "" + # CVSS v3 vector string like + # "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H" — base score not + # directly embedded. We need to compute it. Or look for a numeric + # string in database_specific. + bucket = _cvss_vector_to_severity(score_str) + if bucket != "unknown": + return bucket + return "unknown" + + +def _normalize_severity_string(raw: str) -> str: + """Normalize 'CRITICAL' / 'High' / 'MODERATE' / etc. to canonical bucket.""" + raw = raw.strip().upper() + if raw in ("CRITICAL",): + return "critical" + if raw in ("HIGH",): + return "high" + if raw in ("MEDIUM", "MODERATE"): + return "medium" + if raw in ("LOW",): + return "low" + return "unknown" + + +def _cvss_vector_to_severity(vector: str) -> str: + """Compute CVSS v3 base score from a vector string and return the + severity bucket. Returns 'unknown' if the vector can't be parsed. + + Pure-Python implementation — no external CVSS library. Uses the standard + CVSS v3.1 base score formula from the official spec. + """ + if not isinstance(vector, str) or not vector.startswith("CVSS:3."): + return "unknown" + parts = vector.split("/") + metrics: Dict[str, str] = {} + for part in parts: + if ":" in part: + k, v = part.split(":", 1) + metrics[k] = v + try: + # Metric values per CVSS v3.1 spec + av = {"N": 0.85, "A": 0.62, "L": 0.55, "P": 0.2}[metrics["AV"]] + ac = {"L": 0.77, "H": 0.44}[metrics["AC"]] + pr = ( + {"N": 0.85, "L": 0.62, "H": 0.27}[metrics["PR"]] + if metrics.get("S") == "U" + else {"N": 0.85, "L": 0.68, "H": 0.5}[metrics["PR"]] + ) + ui = {"N": 0.85, "R": 0.62}[metrics["UI"]] + c = {"H": 0.56, "L": 0.22, "N": 0.0}[metrics["C"]] + i = {"H": 0.56, "L": 0.22, "N": 0.0}[metrics["I"]] + a = {"H": 0.56, "L": 0.22, "N": 0.0}[metrics["A"]] + except (KeyError, TypeError): + return "unknown" + + iss = 1 - ((1 - c) * (1 - i) * (1 - a)) + if metrics.get("S") == "U": + impact = 6.42 * iss + else: + impact = 7.52 * (iss - 0.029) - 3.25 * ((iss - 0.02) ** 15) + exploitability = 8.22 * av * ac * pr * ui + if impact <= 0: + return "unknown" + if metrics.get("S") == "U": + base = min(impact + exploitability, 10) + else: + base = min(1.08 * (impact + exploitability), 10) + base = round(base, 1) + # Round up to one decimal per CVSS v3.1 + if base >= 9.0: + return "critical" + if base >= 7.0: + return "high" + if base >= 4.0: + return "medium" + return "low" + + +def _extract_fixed_version( + details: Dict[str, Any], pkg: Dict[str, str] +) -> str: + """Extract the first fixed version from an OSV vuln record. + + Scans `affected[]` for an entry matching our package, then walks + `ranges[].events[]` looking for a `fixed` event. Returns "" if no fixed + version is recorded. + """ + for affected in details.get("affected") or []: + if not isinstance(affected, dict): + continue + affected_pkg = affected.get("package") or {} + if affected_pkg.get("name") != pkg["name"]: + continue + if affected_pkg.get("ecosystem") != pkg["ecosystem"]: + continue + for rng in affected.get("ranges") or []: + if not isinstance(rng, dict): + continue + for event in rng.get("events") or []: + if not isinstance(event, dict): + continue + if "fixed" in event: + return str(event["fixed"]) + return "" + + +# ─── HTTP Helpers ────────────────────────────────────────────── + + +def _http_post_json(url: str, payload: Dict[str, Any]) -> Dict[str, Any]: + """POST JSON to OSV API with retry+backoff. Raises on persistent failure.""" + body = json.dumps(payload).encode("utf-8") + last_exc: Optional[Exception] = None + for attempt in range(MAX_RETRIES): + try: + req = urllib.request.Request( + url, + data=body, + method="POST", + headers={ + "Content-Type": "application/json", + "Accept": "application/json", + "User-Agent": "CodeLens-deps-audit/1.0", + }, + ) + with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT_SECONDS) as resp: + return json.loads(resp.read().decode("utf-8")) + except urllib.error.HTTPError as e: + last_exc = e + # 429 = rate limit; 5xx = transient. Retry with backoff. + if e.code in (429, 500, 502, 503, 504): + retry_after = e.headers.get("Retry-After") + if retry_after: + try: + time.sleep(min(int(retry_after), 30)) + except ValueError: + time.sleep(RETRY_BACKOFF_BASE * (2 ** attempt)) + else: + time.sleep(RETRY_BACKOFF_BASE * (2 ** attempt)) + continue + # 4xx other than 429 — don't retry, fail fast. + raise + except urllib.error.URLError as e: + last_exc = e + time.sleep(RETRY_BACKOFF_BASE * (2 ** attempt)) + continue + raise last_exc # type: ignore[misc] + + +def _http_get_json(url: str) -> Dict[str, Any]: + """GET JSON from OSV API with retry+backoff. Raises on persistent failure.""" + last_exc: Optional[Exception] = None + for attempt in range(MAX_RETRIES): + try: + req = urllib.request.Request( + url, + method="GET", + headers={ + "Accept": "application/json", + "User-Agent": "CodeLens-deps-audit/1.0", + }, + ) + with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT_SECONDS) as resp: + return json.loads(resp.read().decode("utf-8")) + except urllib.error.HTTPError as e: + last_exc = e + if e.code in (429, 500, 502, 503, 504): + time.sleep(RETRY_BACKOFF_BASE * (2 ** attempt)) + continue + raise + except urllib.error.URLError as e: + last_exc = e + time.sleep(RETRY_BACKOFF_BASE * (2 ** attempt)) + continue + raise last_exc # type: ignore[misc] + + +# ─── Stats + Persistence ─────────────────────────────────────── + + +def _compute_stats( + findings: List[Dict[str, Any]], + packages: List[Dict[str, str]], + ecosystems_scanned: Tuple[str, ...], +) -> Dict[str, Any]: + """Compute summary stats for the audit run.""" + by_sev = {"critical": 0, "high": 0, "medium": 0, "low": 0, "unknown": 0} + for f in findings: + sev = f.get("severity", "unknown") + if sev in by_sev: + by_sev[sev] += 1 + else: + by_sev["unknown"] += 1 + return { + "total": len(findings), + "critical": by_sev["critical"], + "high": by_sev["high"], + "medium": by_sev["medium"], + "low": by_sev["low"], + "unknown": by_sev["unknown"], + "packages_scanned": len(packages), + "ecosystems_scanned": len(ecosystems_scanned), + } + + +def _persist_findings_to_graph( + workspace: str, + db_path: Optional[str], + findings: List[Dict[str, Any]], + packages: List[Dict[str, str]], +) -> Optional[Dict[str, Any]]: + """Persist findings as `dependency_vuln` graph nodes + `HAS_VULN` edges. + + Each finding becomes a node with: + node_id = "dep_vuln:{vuln_id}:{package}:{ecosystem}" + node_type = "dependency_vuln" + name = "{package}@{version} ({vuln_id})" + file = source_file (the lock/manifest path) + extra_json = full finding dict + + For each finding, an edge is created from the source file (as a `file` + node) to the vuln node with edge_type = "HAS_VULN". + + Returns: + Dict with {nodes_inserted, edges_inserted, db_path} or None if + persistence was skipped (e.g. no db_path, no findings). + """ + if not findings: + return None + + workspace = os.path.abspath(workspace) + db_path = db_path or default_db_path(workspace) + os.makedirs(os.path.dirname(db_path), exist_ok=True) + + conn = sqlite3.connect(db_path) + nodes_inserted = 0 + edges_inserted = 0 + try: + init_graph_schema(conn) + + # Idempotent: delete previous deps-audit findings for these files + # before re-inserting, so re-scans don't accumulate duplicates. + source_files = sorted({f.get("source_file", "") for f in findings if f.get("source_file")}) + if source_files: + placeholders = ",".join("?" * len(source_files)) + conn.execute( + f"DELETE FROM {GRAPH_EDGES_TABLE} WHERE edge_type = ? " + f"AND file IN ({placeholders})", + [EDGE_TYPE_HAS_VULN] + source_files, + ) + # Delete vuln nodes that are no longer referenced by any edge. + conn.execute( + f"DELETE FROM {GRAPH_NODES_TABLE} WHERE node_type = ? " + f"AND node_id NOT IN (SELECT target_id FROM {GRAPH_EDGES_TABLE})", + [NODE_TYPE_DEPENDENCY_VULN], + ) + + for finding in findings: + vuln_id = finding.get("vuln_id", "") + pkg_name = finding.get("package", "") + version = finding.get("version", "") + ecosystem = finding.get("ecosystem", "") + source_file = finding.get("source_file", "") + + node_id = f"dep_vuln:{vuln_id}:{pkg_name}:{ecosystem}" + node_name = f"{pkg_name}@{version} ({vuln_id})" + extra_json = json.dumps(finding, sort_keys=True) + + try: + cur = conn.execute( + f"INSERT OR IGNORE INTO {GRAPH_NODES_TABLE} " + f"(node_id, node_type, name, file, line, extra_json) " + f"VALUES (?, ?, ?, ?, ?, ?)", + (node_id, NODE_TYPE_DEPENDENCY_VULN, node_name, source_file, 0, extra_json), + ) + if cur.rowcount > 0: + nodes_inserted += 1 + except sqlite3.Error as e: + logger.warning( + f"[dep_audit] failed to insert vuln node {node_id}: {e}" + ) + continue + + # Insert the file node (if not already present) and link with HAS_VULN + if source_file: + file_node_id = f"file:{source_file}" + try: + conn.execute( + f"INSERT OR IGNORE INTO {GRAPH_NODES_TABLE} " + f"(node_id, node_type, name, file, line, extra_json) " + f"VALUES (?, ?, ?, ?, ?, ?)", + (file_node_id, NODE_TYPE_FILE, source_file, source_file, 0, None), + ) + except sqlite3.Error as e: + logger.warning( + f"[dep_audit] failed to insert file node {file_node_id}: {e}" + ) + + try: + cur = conn.execute( + f"INSERT INTO {GRAPH_EDGES_TABLE} " + f"(source_id, target_id, edge_type, file, line, confidence, extra_json) " + f"VALUES (?, ?, ?, ?, ?, ?, ?)", + (file_node_id, node_id, EDGE_TYPE_HAS_VULN, source_file, 0, 1.0, None), + ) + if cur.rowcount > 0: + edges_inserted += 1 + except sqlite3.Error as e: + logger.warning( + f"[dep_audit] failed to insert HAS_VULN edge: {e}" + ) + + conn.commit() + except sqlite3.Error as e: + logger.error(f"[dep_audit] graph persistence failed: {e}") + conn.rollback() + return None + finally: + conn.close() + + return { + "nodes_inserted": nodes_inserted, + "edges_inserted": edges_inserted, + "db_path": db_path, + } + + +# ─── Recommendations ─────────────────────────────────────────── + + +def _generate_recommendations( + findings: List[Dict[str, Any]], api_errors: List[str] +) -> List[str]: + """Generate actionable recommendations based on the audit findings.""" + recs: List[str] = [] + + if not findings and not api_errors: + recs.append( + "No known vulnerabilities found in scanned dependencies. " + "Re-run periodically — new CVEs are published daily." + ) + return recs + + if api_errors: + recs.append( + "WARNING: API errors occurred during the scan — results may be " + "incomplete. See the `api_errors` field for details." + ) + + if not findings: + return recs + + critical = [f for f in findings if f.get("severity") == "critical"] + high = [f for f in findings if f.get("severity") == "high"] + + if critical: + crit_packages = sorted({f["package"] for f in critical}) + recs.append( + f"CRITICAL: {len(critical)} critical vulnerability(ies) found in " + f"{len(crit_packages)} package(s): " + f"{', '.join(crit_packages[:10])}. " + f"Upgrade immediately — these are likely exploitable." + ) + + if high: + high_packages = sorted({f["package"] for f in high}) + recs.append( + f"HIGH: {len(high)} high-severity vulnerability(ies) in " + f"{len(high_packages)} package(s): " + f"{', '.join(high_packages[:10])}. " + f"Upgrade before the next release." + ) + + # Group by package to suggest consolidated upgrades + by_pkg: Dict[str, List[Dict[str, Any]]] = {} + for f in findings: + by_pkg.setdefault(f["package"], []).append(f) + + upgradable = [ + pkg for pkg, flist in by_pkg.items() + if any(f.get("fixed_in") for f in flist) + ] + if upgradable: + recs.append( + f"UPGRADE: {len(upgradable)} package(s) have a known fixed " + f"version. Update your manifest/lock file to the version listed " + f"in the `fixed_in` field of each finding." + ) + + no_fix = [pkg for pkg, flist in by_pkg.items() if not any(f.get("fixed_in") for f in flist)] + if no_fix: + recs.append( + f"NO FIX AVAILABLE: {len(no_fix)} package(s) have vulnerabilities " + f"with no recorded fix. Consider replacing or pinning a " + f"workaround: {', '.join(sorted(no_fix)[:10])}." + ) + + return recs diff --git a/tests/test_deps_audit.py b/tests/test_deps_audit.py new file mode 100644 index 00000000..9888d287 --- /dev/null +++ b/tests/test_deps_audit.py @@ -0,0 +1,812 @@ +""" +Tests for the deps-audit engine (issue #158). + +Covers: +- Manifest parsing (requirements.txt, pyproject.toml, Pipfile, package.json, + Cargo.toml) — operator stripping, comments, extras, environment markers. +- Lock-file parsing (package-lock.json v1+v2, yarn.lock, pnpm-lock.yaml, + Cargo.lock) — pinned version extraction. +- Offline mode — packages discovered, no findings returned. +- OSV batch query (mocked) — single + multiple packages, vuln detail fetch, + severity extraction (CVSS vector + database_specific). +- Severity filter — "high" returns high + critical. +- Ecosystem filter — only one ecosystem scanned. +- Graph persistence — dependency_vuln nodes + HAS_VULN edges, idempotent + re-scan. +- Edge cases — empty workspace, no manifests, missing version. +""" + +import json +import os +import shutil +import sqlite3 +import sys +import tempfile +from unittest.mock import patch, MagicMock + +import pytest + +SCRIPT_DIR = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "scripts" +) +sys.path.insert(0, SCRIPT_DIR) + +import dep_audit_engine as engine +from graph_model import ( + EDGE_TYPE_HAS_VULN, + GRAPH_EDGES_TABLE, + GRAPH_NODES_TABLE, + NODE_TYPE_DEPENDENCY_VULN, +) + + +# ─── Test Helpers ────────────────────────────────────────────── + + +def _make_workspace(files: dict) -> str: + """Create a temp workspace with the given {filename: content} mapping.""" + ws = tempfile.mkdtemp() + for name, content in files.items(): + path = os.path.join(ws, name) + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as f: + f.write(content) + return ws + + +def _cleanup(ws: str) -> None: + shutil.rmtree(ws, ignore_errors=True) + + +def _mock_osv_response(vulns_by_package: dict) -> dict: + """Build a fake /v1/querybatch response. + + Args: + vulns_by_package: {"package_name:version": [{"id": "GHSA-xxx", ...}]} + """ + return vulns_by_package # the test will patch _query_osv_batch directly + + +# ─── Manifest Parsing ────────────────────────────────────────── + + +class TestRequirementsTxtParsing: + def test_basic_pinned(self): + content = "requests==2.27.0\nurllib3==1.26.0\n" + result = engine._parse_requirements_txt(content) + assert ("requests", "2.27.0") in result + assert ("urllib3", "1.26.0") in result + + def test_operators(self): + content = "pkg1>=1.0.0\npkg2~=2.0\npkg3>3.0\npkg4<=4.0\n" + result = engine._parse_requirements_txt(content) + names = [r[0] for r in result] + assert "pkg1" in names + assert "pkg2" in names + assert "pkg3" in names + assert "pkg4" in names + + def test_comments_and_blanks(self): + content = """ +# This is a comment +requests==2.27.0 # inline comment + +# Another comment +flask==2.0.0 +""" + result = engine._parse_requirements_txt(content) + names = [r[0] for r in result] + assert "requests" in names + assert "flask" in names + assert len(result) == 2 + + def test_extras_and_markers(self): + content = 'package[extra1,extra2]==1.0.0; python_version >= "3.8"\n' + result = engine._parse_requirements_txt(content) + assert len(result) == 1 + assert result[0][0] == "package" + assert result[0][1] == "1.0.0" + + def test_skip_options_and_vcs(self): + content = """ +-r other-requirements.txt +--index-url https://example.com +git+https://github.com/foo/bar.git#egg=bar +https://example.com/pkg.tar.gz +-e . +normal-pkg==1.0.0 +""" + result = engine._parse_requirements_txt(content) + names = [r[0] for r in result] + assert "normal-pkg" in names + assert len(result) == 1 + + def test_empty(self): + assert engine._parse_requirements_txt("") == [] + assert engine._parse_requirements_txt("# just a comment\n") == [] + + +class TestPyprojectTomlParsing: + def test_pep621_dependencies(self): + content = """ +[project] +name = "myproj" +dependencies = [ + "requests>=2.27.0", + "flask==2.0.0", +] +""" + result = engine._parse_pyproject_toml(content) + names = {r[0] for r in result} + assert "requests" in names + assert "flask" in names + + def test_poetry_dependencies(self): + content = """ +[tool.poetry.dependencies] +python = "^3.10" +requests = "^2.27.0" +flask = "2.0.0" +""" + result = engine._parse_pyproject_toml(content) + names = {r[0] for r in result} + assert "requests" in names + assert "flask" in names + # python constraint should be skipped + assert "python" not in names + + def test_invalid_toml(self): + # Should return empty list, not raise + result = engine._parse_pyproject_toml("not valid toml = = =") + assert result == [] + + +class TestPackageJsonParsing: + def test_deps_and_dev_deps(self): + content = json.dumps({ + "name": "test", + "dependencies": {"express": "^4.18.0", "lodash": "~4.17.21"}, + "devDependencies": {"jest": ">=29.0.0"}, + "optionalDependencies": {"fsevents": "2.3.2"}, + }) + result = engine._parse_package_json(content) + names = {r[0] for r in result} + assert "express" in names + assert "lodash" in names + assert "jest" in names + assert "fsevents" in names + # Operators stripped + express_version = next(v for n, v in result if n == "express") + assert express_version == "4.18.0" + + def test_scoped_package(self): + content = json.dumps({ + "dependencies": {"@babel/core": "^7.20.0"}, + }) + result = engine._parse_package_json(content) + assert len(result) == 1 + assert result[0][0] == "@babel/core" + assert result[0][1] == "7.20.0" + + def test_skip_git_and_file_specs(self): + content = json.dumps({ + "dependencies": { + "normal-pkg": "1.0.0", + "git-pkg": "git+https://github.com/foo/bar.git", + "file-pkg": "file:./local", + "link-pkg": "link:../other", + }, + }) + result = engine._parse_package_json(content) + names = {r[0] for r in result} + assert "normal-pkg" in names + assert "git-pkg" not in names + assert "file-pkg" not in names + assert "link-pkg" not in names + + def test_wildcards_return_empty_version(self): + content = json.dumps({ + "dependencies": {"wildcard-pkg": "*", "latest-pkg": "latest"}, + }) + result = engine._parse_package_json(content) + for name, version in result: + assert version == "", f"{name} should have empty version" + + def test_invalid_json(self): + result = engine._parse_package_json("{not valid json") + assert result == [] + + +class TestCargoTomlParsing: + def test_string_and_table_forms(self): + content = """ +[dependencies] +serde = "1.0" +tokio = { version = "1.0", features = ["full"] } + +[dev-dependencies] +pytest = "7.0" + +[build-dependencies] +cc = "1.0" +""" + result = engine._parse_cargo_toml(content) + names = {r[0] for r in result} + assert "serde" in names + assert "tokio" in names + assert "pytest" in names + assert "cc" in names + + +# ─── Lock File Parsing ───────────────────────────────────────── + + +class TestPackageLockJsonParsing: + def test_v2_packages_section(self): + content = json.dumps({ + "lockfileVersion": 2, + "packages": { + "": {"name": "root", "version": "1.0.0"}, + "node_modules/express": {"version": "4.18.0"}, + "node_modules/lodash": {"version": "4.17.21"}, + "node_modules/@babel/core": {"version": "7.20.0"}, + }, + }) + result = engine._parse_package_lock_json(content) + names = {r[0] for r in result} + assert "express" in names + assert "lodash" in names + assert "@babel/core" in names + # Root package (empty key) should be skipped + assert "root" not in names + + def test_v1_dependencies_section(self): + content = json.dumps({ + "lockfileVersion": 1, + "dependencies": { + "express": {"version": "4.18.0"}, + "lodash": {"version": "4.17.21"}, + }, + }) + result = engine._parse_package_lock_json(content) + assert ("express", "4.18.0") in result + assert ("lodash", "4.17.21") in result + + def test_dedup_across_sections(self): + # v2 lockfile with both sections pointing at the same package + content = json.dumps({ + "lockfileVersion": 2, + "packages": {"node_modules/express": {"version": "4.18.0"}}, + "dependencies": {"express": {"version": "4.18.0"}}, + }) + result = engine._parse_package_lock_json(content) + assert result == [("express", "4.18.0")] + + +class TestCargoLockParsing: + def test_multiple_packages(self): + content = """ +# This file is automatically @generated by Cargo. +version = 3 + +[[package]] +name = "serde" +version = "1.0.150" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "abc123" + +[[package]] +name = "tokio" +version = "1.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" + +[[package]] +name = "my-crate" +version = "0.1.0" +""" + result = engine._parse_cargo_lock(content) + assert ("serde", "1.0.150") in result + assert ("tokio", "1.23.0") in result + assert ("my-crate", "0.1.0") in result + + +# ─── Offline Mode ────────────────────────────────────────────── + + +class TestOfflineMode: + def test_offline_with_no_manifests(self): + ws = _make_workspace({"README.md": "no deps here"}) + try: + result = engine.audit_dependencies(ws, offline=True) + assert result["status"] == "offline" + assert result["findings"] == [] + assert result["stats"]["packages_scanned"] == 0 + finally: + _cleanup(ws) + + def test_offline_with_pyproject(self): + ws = _make_workspace({ + "pyproject.toml": """ +[project] +name = "test" +dependencies = ["requests==2.27.0", "flask==2.0.0"] +""" + }) + try: + result = engine.audit_dependencies(ws, offline=True) + assert result["status"] == "offline" + assert result["stats"]["packages_scanned"] == 2 + packages = {p["name"] for p in result["packages_scanned"]} + assert "requests" in packages + assert "flask" in packages + finally: + _cleanup(ws) + + def test_offline_with_package_json(self): + ws = _make_workspace({ + "package.json": json.dumps({ + "dependencies": {"express": "^4.18.0"} + }) + }) + try: + result = engine.audit_dependencies(ws, offline=True) + assert result["status"] == "offline" + # Only npm ecosystem scanned (auto-detect finds package.json) + assert "package.json" in result["files_scanned"] + finally: + _cleanup(ws) + + def test_offline_with_cargo_toml(self): + ws = _make_workspace({ + "Cargo.toml": """ +[dependencies] +serde = "1.0" +""" + }) + try: + result = engine.audit_dependencies(ws, offline=True) + assert result["status"] == "offline" + assert "Cargo.toml" in result["files_scanned"] + assert result["stats"]["packages_scanned"] >= 1 + finally: + _cleanup(ws) + + +# ─── Ecosystem Filter ────────────────────────────────────────── + + +class TestEcosystemFilter: + def test_only_pypi_scanned(self): + ws = _make_workspace({ + "pyproject.toml": '[project]\nname="x"\ndependencies=["flask==1.0"]\n', + "package.json": '{"dependencies": {"express": "4.0"}}', + }) + try: + result = engine.audit_dependencies(ws, ecosystem="PyPI", offline=True) + # Only PyPI files in the scanned list + assert "pyproject.toml" in result["files_scanned"] + assert "package.json" not in result["files_scanned"] + finally: + _cleanup(ws) + + def test_only_npm_scanned(self): + ws = _make_workspace({ + "pyproject.toml": '[project]\nname="x"\ndependencies=["flask==1.0"]\n', + "package.json": '{"dependencies": {"express": "4.0"}}', + }) + try: + result = engine.audit_dependencies(ws, ecosystem="npm", offline=True) + assert "package.json" in result["files_scanned"] + assert "pyproject.toml" not in result["files_scanned"] + finally: + _cleanup(ws) + + +# ─── Lock File Preference ────────────────────────────────────── + + +class TestLockFilePreference: + def test_lock_file_preferred_over_manifest(self): + ws = _make_workspace({ + "package.json": json.dumps({ + "dependencies": {"express": "^4.18.0"} # manifest spec + }), + "package-lock.json": json.dumps({ + "lockfileVersion": 2, + "packages": {"node_modules/express": {"version": "4.18.2"}}, + }), + }) + try: + result = engine.audit_dependencies(ws, ecosystem="npm", offline=True) + # Should use the pinned version from lock file + express = next( + p for p in result["packages_scanned"] if p["name"] == "express" + ) + assert express["version"] == "4.18.2" + assert express["source_file"] == "package-lock.json" + # Manifest should NOT be scanned when lock file is present + assert "package.json" not in result["files_scanned"] + finally: + _cleanup(ws) + + +# ─── OSV API (Mocked) ────────────────────────────────────────── + + +class TestOsvBatchQuery: + def test_single_package_with_vuln(self): + """Mock _http_post_json + _http_get_json to return canned responses.""" + packages = [ + {"name": "requests", "version": "2.27.0", "ecosystem": "PyPI", "source_file": "requirements.txt"} + ] + batch_response = { + "results": [ + {"vulns": [{"id": "GHSA-j8r2-6x86-q33q"}]} + ] + } + vuln_detail = { + "id": "GHSA-j8r2-6x86-q33q", + "summary": "Unintended leak of Proxy-Authorization header", + "aliases": ["CVE-2023-32681"], + "database_specific": {"severity": "HIGH"}, + "affected": [ + { + "package": {"name": "requests", "ecosystem": "PyPI"}, + "ranges": [{"events": [{"introduced": "0"}, {"fixed": "2.31.0"}]}], + } + ], + } + with patch.object(engine, "_http_post_json", return_value=batch_response), \ + patch.object(engine, "_http_get_json", return_value=vuln_detail): + findings, errors = engine._query_osv_batch(packages) + + assert len(findings) == 1 + f = findings[0] + assert f["package"] == "requests" + assert f["version"] == "2.27.0" + assert f["vuln_id"] == "GHSA-j8r2-6x86-q33q" + assert f["severity"] == "high" + assert f["fixed_in"] == "2.31.0" + assert f["summary"].startswith("Unintended leak") + assert "CVE-2023-32681" in f["cve_ids"] + assert f["osv_url"] == "https://osv.dev/vulnerability/GHSA-j8r2-6x86-q33q" + assert errors == [] + + def test_multiple_packages_batched(self): + packages = [ + {"name": f"pkg{i}", "version": f"1.0.{i}", "ecosystem": "PyPI", "source_file": "requirements.txt"} + for i in range(5) + ] + # Only pkg0 and pkg2 have vulns + batch_response = { + "results": [ + {"vulns": [{"id": "VULN-0"}]} if i == 0 else + {"vulns": []} if i in (1, 3, 4) else + {"vulns": [{"id": "VULN-2"}]} + for i in range(5) + ] + } + vuln_details = { + "VULN-0": {"id": "VULN-0", "summary": "vuln 0", "database_specific": {"severity": "CRITICAL"}}, + "VULN-2": {"id": "VULN-2", "summary": "vuln 2", "database_specific": {"severity": "LOW"}}, + } + + def fake_get(url): + vuln_id = url.rsplit("/", 1)[-1] + return vuln_details.get(vuln_id, {}) + + with patch.object(engine, "_http_post_json", return_value=batch_response) as mock_post, \ + patch.object(engine, "_http_get_json", side_effect=fake_get) as mock_get: + findings, errors = engine._query_osv_batch(packages) + + assert len(findings) == 2 + vuln_ids = {f["vuln_id"] for f in findings} + assert vuln_ids == {"VULN-0", "VULN-2"} + # Verify vuln detail cache (only 2 GETs, not 4) + assert mock_get.call_count == 2 + + def test_skips_packages_without_version(self): + packages = [ + {"name": "no-version", "version": "", "ecosystem": "PyPI", "source_file": "requirements.txt"}, + {"name": "has-version", "version": "1.0.0", "ecosystem": "PyPI", "source_file": "requirements.txt"}, + ] + batch_response = {"results": [{"vulns": []}, {"vulns": []}]} + with patch.object(engine, "_http_post_json", return_value=batch_response) as mock_post: + findings, errors = engine._query_osv_batch(packages) + # Only one batch POST (the empty-version pkg is skipped) + assert mock_post.call_count == 1 + # Error message should mention skipped packages + assert any("skipped" in e for e in errors) + + def test_batch_api_error_recorded(self): + packages = [ + {"name": "pkg", "version": "1.0", "ecosystem": "PyPI", "source_file": "requirements.txt"} + ] + with patch.object(engine, "_http_post_json", side_effect=Exception("network down")): + findings, errors = engine._query_osv_batch(packages) + assert findings == [] + assert len(errors) == 1 + assert "network down" in errors[0] + + def test_vuln_detail_fetch_failure_does_not_break_batch(self): + packages = [ + {"name": "pkg", "version": "1.0", "ecosystem": "PyPI", "source_file": "requirements.txt"} + ] + batch_response = {"results": [{"vulns": [{"id": "BAD-VULN"}]}]} + with patch.object(engine, "_http_post_json", return_value=batch_response), \ + patch.object(engine, "_http_get_json", side_effect=Exception("detail fetch failed")): + findings, errors = engine._query_osv_batch(packages) + # Finding still recorded, but with empty detail + assert len(findings) == 1 + assert findings[0]["vuln_id"] == "BAD-VULN" + assert findings[0]["severity"] == "unknown" + # Error recorded + assert any("detail fetch failed" in e for e in errors) + + +# ─── CVSS Vector Parsing ─────────────────────────────────────── + + +class TestCvssVectorParsing: + def test_critical_vector(self): + # CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H → ~10.0 (critical) + vector = "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H" + assert engine._cvss_vector_to_severity(vector) == "critical" + + def test_high_vector(self): + # CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:N/A:N → ~7.5 (high) + vector = "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:N/A:N" + assert engine._cvss_vector_to_severity(vector) == "high" + + def test_medium_vector(self): + # CVSS:3.1/AV:N/AC:H/PR:N/UI:N/S:U/C:L/I:N/A:N → ~3.1 (low) + # Use a definite medium: 4.0-6.9 + # AV:N(0.85) AC:L(0.77) PR:N(0.85) UI:N(0.85) C:L(0.22) I:N(0) A:N(0) S:U + # ISS = 1 - (1-0.22)*(1)*(1) = 0.22 + # impact = 6.42 * 0.22 = 1.4124 + # exploitability = 8.22 * 0.85 * 0.77 * 0.85 * 0.85 = 3.8915 + # base = min(1.4124 + 3.8915, 10) = 5.3 → medium + vector = "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:L/I:N/A:N" + assert engine._cvss_vector_to_severity(vector) == "medium" + + def test_low_vector(self): + # Lower impact → low + vector = "CVSS:3.1/AV:N/AC:H/PR:H/UI:R/S:U/C:L/I:N/A:N" + result = engine._cvss_vector_to_severity(vector) + assert result in ("low", "medium") # depends on exact computation + + def test_invalid_vector(self): + assert engine._cvss_vector_to_severity("") == "unknown" + assert engine._cvss_vector_to_severity("not a vector") == "unknown" + assert engine._cvss_vector_to_severity("CVSS:3.1/AV:N") == "unknown" # missing metrics + + +# ─── Severity Filter ─────────────────────────────────────────── + + +class TestSeverityFilter: + def test_high_filter_includes_critical(self): + ws = _make_workspace({ + "requirements.txt": "requests==2.27.0\n" + }) + try: + # Mock OSV to return 3 findings: 1 critical, 1 high, 1 low + batch_response = { + "results": [{"vulns": [{"id": "V1"}, {"id": "V2"}, {"id": "V3"}]}] + } + vuln_details = { + "V1": {"id": "V1", "database_specific": {"severity": "CRITICAL"}}, + "V2": {"id": "V2", "database_specific": {"severity": "HIGH"}}, + "V3": {"id": "V3", "database_specific": {"severity": "LOW"}}, + } + + def fake_get(url): + return vuln_details.get(url.rsplit("/", 1)[-1], {}) + + with patch.object(engine, "_http_post_json", return_value=batch_response), \ + patch.object(engine, "_http_get_json", side_effect=fake_get): + result = engine.audit_dependencies(ws, severity="high") + + assert result["status"] == "ok" + assert result["severity_filter"] == "high" + # Should include critical + high, exclude low + severities = {f["severity"] for f in result["findings"]} + assert "critical" in severities + assert "high" in severities + assert "low" not in severities + finally: + _cleanup(ws) + + +# ─── Graph Persistence ───────────────────────────────────────── + + +class TestGraphPersistence: + def test_findings_persisted_as_nodes_and_edges(self): + ws = _make_workspace({"requirements.txt": "requests==2.27.0\n"}) + try: + batch_response = { + "results": [{"vulns": [{"id": "GHSA-xxx"}]}] + } + vuln_detail = { + "id": "GHSA-xxx", + "database_specific": {"severity": "HIGH"}, + "affected": [ + { + "package": {"name": "requests", "ecosystem": "PyPI"}, + "ranges": [{"events": [{"fixed": "2.31.0"}]}], + } + ], + } + with patch.object(engine, "_http_post_json", return_value=batch_response), \ + patch.object(engine, "_http_get_json", return_value=vuln_detail): + result = engine.audit_dependencies(ws) + + assert result["status"] == "ok" + assert "persistence" in result + assert result["persistence"]["nodes_inserted"] >= 1 + assert result["persistence"]["edges_inserted"] >= 1 + + # Verify in SQLite directly + db_path = result["persistence"]["db_path"] + conn = sqlite3.connect(db_path) + try: + cur = conn.execute( + f"SELECT node_id, node_type, name, file FROM {GRAPH_NODES_TABLE} " + f"WHERE node_type = ?", + [NODE_TYPE_DEPENDENCY_VULN], + ) + vuln_nodes = cur.fetchall() + assert len(vuln_nodes) == 1 + node_id, node_type, name, file = vuln_nodes[0] + assert node_type == NODE_TYPE_DEPENDENCY_VULN + assert "GHSA-xxx" in node_id + assert "requests" in name + assert file == "requirements.txt" + + cur = conn.execute( + f"SELECT COUNT(*) FROM {GRAPH_EDGES_TABLE} WHERE edge_type = ?", + [EDGE_TYPE_HAS_VULN], + ) + edge_count = cur.fetchone()[0] + assert edge_count == 1 + finally: + conn.close() + finally: + _cleanup(ws) + + def test_idempotent_rescan_no_duplicates(self): + ws = _make_workspace({"requirements.txt": "requests==2.27.0\n"}) + try: + batch_response = { + "results": [{"vulns": [{"id": "GHSA-xxx"}]}] + } + vuln_detail = { + "id": "GHSA-xxx", + "database_specific": {"severity": "HIGH"}, + "affected": [ + { + "package": {"name": "requests", "ecosystem": "PyPI"}, + "ranges": [{"events": [{"fixed": "2.31.0"}]}], + } + ], + } + with patch.object(engine, "_http_post_json", return_value=batch_response), \ + patch.object(engine, "_http_get_json", return_value=vuln_detail): + # First scan + engine.audit_dependencies(ws) + # Second scan (same data) + result2 = engine.audit_dependencies(ws) + + # Verify only one vuln node + one edge in the DB after re-scan + db_path = result2["persistence"]["db_path"] + conn = sqlite3.connect(db_path) + try: + cur = conn.execute( + f"SELECT COUNT(*) FROM {GRAPH_NODES_TABLE} WHERE node_type = ?", + [NODE_TYPE_DEPENDENCY_VULN], + ) + assert cur.fetchone()[0] == 1, "should not duplicate vuln nodes" + + cur = conn.execute( + f"SELECT COUNT(*) FROM {GRAPH_EDGES_TABLE} WHERE edge_type = ?", + [EDGE_TYPE_HAS_VULN], + ) + assert cur.fetchone()[0] == 1, "should not duplicate HAS_VULN edges" + finally: + conn.close() + finally: + _cleanup(ws) + + +# ─── Edge Cases ──────────────────────────────────────────────── + + +class TestEdgeCases: + def test_empty_workspace(self): + ws = _make_workspace({"README.md": "no deps"}) + try: + result = engine.audit_dependencies(ws) + assert result["status"] == "ok" + assert result["findings"] == [] + assert result["stats"]["total"] == 0 + assert "No dependency manifests found" in result["recommendations"][0] + finally: + _cleanup(ws) + + def test_no_dependencies_in_manifest(self): + ws = _make_workspace({ + "package.json": json.dumps({"name": "empty", "dependencies": {}}) + }) + try: + result = engine.audit_dependencies(ws, offline=True) + # Even with no packages, the file should be scanned + assert result["status"] == "offline" + assert result["stats"]["packages_scanned"] == 0 + finally: + _cleanup(ws) + + def test_return_structure(self): + ws = _make_workspace({"requirements.txt": "flask==2.0.0\n"}) + try: + result = engine.audit_dependencies(ws, offline=True) + # Required keys per issue #158 spec + assert "status" in result + assert "stats" in result + assert "findings" in result + assert "recommendations" in result + assert "files_scanned" in result + assert "packages_scanned" in result + # stats structure + stats = result["stats"] + for key in ("total", "critical", "high", "medium", "low", + "packages_scanned", "ecosystems_scanned"): + assert key in stats, f"missing stat: {key}" + finally: + _cleanup(ws) + + +# ─── Spec String Helpers ─────────────────────────────────────── + + +class TestSpecStringHelpers: + def test_strip_npm_operator(self): + assert engine._strip_npm_operator("^1.2.3") == "1.2.3" + assert engine._strip_npm_operator("~1.2.3") == "1.2.3" + assert engine._strip_npm_operator(">=1.2.3") == "1.2.3" + assert engine._strip_npm_operator("1.2.3") == "1.2.3" + assert engine._strip_npm_operator("*") == "" + assert engine._strip_npm_operator("latest") == "" + assert engine._strip_npm_operator("1.2.x") == "" + assert engine._strip_npm_operator("1.2.3 || 2.0.0") == "1.2.3" + + def test_strip_python_operator(self): + assert engine._strip_python_operator(">=1.0") == "1.0" + assert engine._strip_python_operator("==1.0") == "1.0" + assert engine._strip_python_operator("~=1.0") == "1.0" + assert engine._strip_python_operator("1.0") == "1.0" + assert engine._strip_python_operator("*") == "" + + def test_strip_cargo_operator(self): + assert engine._strip_cargo_operator("^1.0") == "1.0" + assert engine._strip_cargo_operator("1.0") == "1.0" + assert engine._strip_cargo_operator("*") == "" + assert engine._strip_cargo_operator(">=1.0") == "1.0" + + def test_split_pep508_spec(self): + assert engine._split_pep508_spec("requests==2.27.0") == ("requests", "2.27.0") + assert engine._split_pep508_spec("requests>=2.0,<3.0") == ("requests", "2.0") + # With extras and markers + name, version = engine._split_pep508_spec('package[extra]==1.0; python_version>="3.8"') + assert name == "package" + assert version == "1.0" + # Wildcards + assert engine._split_pep508_spec("pkg==*") == ("pkg", "") + + def test_normalize_severity_string(self): + assert engine._normalize_severity_string("CRITICAL") == "critical" + assert engine._normalize_severity_string("High") == "high" + assert engine._normalize_severity_string("MODERATE") == "medium" + assert engine._normalize_severity_string("medium") == "medium" + assert engine._normalize_severity_string("low") == "low" + assert engine._normalize_severity_string("bogus") == "unknown" From b0ba5e7266ac8ac54836b139947522b6ac63e961 Mon Sep 17 00:00:00 2001 From: Wolfvin Date: Fri, 3 Jul 2026 02:58:49 +0000 Subject: [PATCH 2/2] fix(deps-audit): rebase on main + run sync_command_count.py (closes #172) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Issue #172 requested verification that sync_command_count.py was run on PR #162 (deps-audit), not manual count bump. What was done: - Rebased feat/issue-158-deps-audit-osv on latest main (39 commits behind) - Resolved conflicts in 6 doc files (all were command-count numbers): * README.md, SKILL.md, SKILL-QUICK.md, pyproject.toml, skill.json, scripts/graph_model.py * Took main's version for count numbers, then re-applied my changes (graph_model NODE_TYPE_DEPENDENCY_VULN/EDGE_TYPE_HAS_VULN constants, README deps-audit row, SKILL-QUICK deps-audit entry) - Ran `python3 scripts/sync_command_count.py --apply`: * Command count: 76 (was 71 in original PR — main has moved +5 commands from other merged PRs: orient #165, large-file-threshold #163, etc.) * MCP tools: 74 (56 static + 18 dynamic) - Verified `sync_command_count.py --check` reports 'All documentation files are in sync with COMMAND_REGISTRY' - Re-ran deps_audit tests: 47 passed, 0 failed Definition of Done (issue #172): - [x] PR #162 rebased on latest main - [x] sync_command_count.py --apply run — output counts match docs files - [x] Counts corrected: 71 → 76 commands, MCP tools updated accordingly - [x] PR #162 is mergeable (no conflicts, tests pass) Note: sync_command_count.py works fine when invoked directly (it imports COMMAND_REGISTRY without going through codelens.py main). However, `codelens.py --command-count` currently crashes with an argparse 'conflicting option string: -f' error — this is a pre-existing bug in main unrelated to this PR (the `affected` command defines `-f --filter` which conflicts with the global `--format -f` alias). Flagged in PR description for BOS awareness but NOT fixed here (out of scope for #172). --- README.md | 11 ++++++----- SKILL-QUICK.md | 14 +++++++------- SKILL.md | 4 ++-- pyproject.toml | 2 +- scripts/graph_model.py | 8 +++++++- skill.json | 2 +- 6 files changed, 24 insertions(+), 17 deletions(-) diff --git a/README.md b/README.md index 504033d1..761b8daf 100644 --- a/README.md +++ b/README.md @@ -2,12 +2,12 @@ > **Before an AI writes a new class/id/function, CodeLens must be checked. This is not optional.** -CodeLens is an AI-native code intelligence platform that gives AI agents **full visibility** into a codebase before they write any code. It prevents collision, overwrite of existing logic, security vulnerabilities, and dead code through 71 CLI commands, an MCP server with 69 tools (54 static + 15 dynamic), AST-based taint analysis, live CVE/OSV scanning, a plugin system with OWASP Top 10 + Compliance rule packs, a true graph data model (nodes + edges) for structural code queries, and token-efficient `--format compact` output for high-volume agent workflows (issue #17). +CodeLens is an AI-native code intelligence platform that gives AI agents **full visibility** into a codebase before they write any code. It prevents collision, overwrite of existing logic, security vulnerabilities, and dead code through 76 CLI commands, an MCP server with 74 tools (56 static + 18 dynamic), AST-based taint analysis, live CVE/OSV scanning, a plugin system with OWASP Top 10 + Compliance rule packs, a true graph data model (nodes + edges) for structural code queries, and token-efficient `--format compact` output for high-volume agent workflows (issue #17). ## Features -- **71 CLI Commands** — From basic scan/query to AST taint analysis, CVE scanning, plugin management, auto-fix, dashboards, CI/CD quality gates, and `graph-schema` for cheap graph-shape introspection -- **MCP Server (69 Tools)** — Native AI agent integration via Model Context Protocol (JSON-RPC over stdio), 54 statically-defined tools + 15 dynamically discovered, every tool accepts a `format` parameter (`json`/`markdown`/`ai`/`sarif`/`compact`) +- **76 CLI Commands** — From basic scan/query to AST taint analysis, CVE scanning, plugin management, auto-fix, dashboards, CI/CD quality gates, and `graph-schema` for cheap graph-shape introspection +- **MCP Server (74 Tools)** — Native AI agent integration via Model Context Protocol (JSON-RPC over stdio), 56 statically-defined tools + 18 dynamically discovered, every tool accepts a `format` parameter (`json`/`markdown`/`ai`/`sarif`/`compact`) - **Token-Efficient Compact Output (v8.2, issue #17)** — `--format compact` produces single-char-key JSON with abbreviated types, omitted null fields, and relative paths — ~50% smaller than `json` on real trace output. Combined with `--limit`/`--offset` pagination, 5 structural queries now cost <5k tokens (down from 30-80k) - **AST Taint Engine** — Tree-sitter based taint analysis with return-value propagation, scope hierarchy, and branch condition refinement - **Live CVE/OSV Scanning** — Real-time vulnerability data from OSV.dev API with SQLite cache, 9 ecosystems (PyPI, npm, crates.io, Go, Maven, NuGet, RubyGems, Pub, Hex) @@ -122,6 +122,7 @@ python3 scripts/codelens.py query "myFunction" --lite |---------|-------------| | `secrets [workspace] [--severity ...]` | Detect hardcoded API keys, passwords, tokens | | `vuln-scan [workspace] [--severity ...] [--offline] [--osv-ttl N] [--refresh] [--max-age Nh]` | Scan dependencies for known CVEs (OSV.dev + native audit). `--refresh` bypasses the OSV cache and forces fresh API calls; `--max-age Nh` treats cache entries older than N hours as stale for this run only (issue #30). Output includes a `cache_info` block (`last_refresh`, `age_hours`, `ttl_hours`, `is_stale`, `stale_packages`) so agents can decide whether to trust the cached CVE data. | +| `deps-audit [workspace] [--severity ...] [--ecosystem PyPI\|npm\|crates.io] [--offline]` | Pure-Python dependency audit via OSV.dev batch API. Auto-detects `requirements.txt` / `pyproject.toml` / `Pipfile` (PyPI), `package.json` + lock files (npm), `Cargo.toml` + `Cargo.lock` (crates.io). Stores findings as `dependency_vuln` graph nodes linked via `HAS_VULN` edges (issue #158). | | `taint [workspace]` | Run AST-based taint analysis for vulnerability detection | | `dataflow [workspace] [--source] [--sink]` | Data flow taint analysis with cross-file call graph | | `env-check [workspace] [--var NAME]` | Audit environment variables | @@ -225,8 +226,8 @@ codelens/ │ ├── changelog.md # Older changelog (per-version highlights) │ └── agent-integration.md # AI agent integration guide ├── scripts/ -│ ├── codelens.py # CLI entry point (71 commands registered) -│ ├── mcp_server.py # MCP JSON-RPC server (69 tools) +│ ├── codelens.py # CLI entry point (76 commands registered) +│ ├── mcp_server.py # MCP JSON-RPC server (74 tools) │ ├── registry.py # Registry read/write/build │ ├── persistent_registry.py # SQLite persistent storage (WAL mode) │ ├── base_parser.py # Base tree-sitter parser diff --git a/SKILL-QUICK.md b/SKILL-QUICK.md index fc8a5a2e..266b75b9 100755 --- a/SKILL-QUICK.md +++ b/SKILL-QUICK.md @@ -116,7 +116,7 @@ $CLI list --limit 5 --offset 10 --format compact # → paginated + co | "Cross-file taint" | `dataflow` | `taint` (taint is single-file, AST-deep) | | "Auto-fix issues" | `fix` | `check` (check just gates, doesn't fix) | -## All 71 Commands +## All 76 Commands ### Setup & Lifecycle (8+) `init` · `scan [--incremental] [--max-files N] [--full]` · `registry-validate` · `detect` · `watch [--debounce SECS] [--git-mode] [--interval SECS]` · `git-status` · `migrate` · `serve` · `lsp-status` (issue #33: `codelens --lsp-status` top-level flag is an alias of `codelens lsp-status` — both delegate to `hybrid_engine.get_lsp_status()` and return the identical payload) @@ -130,8 +130,8 @@ $CLI list --limit 5 --offset 10 --format compact # → paginated + co ### Architecture (10) `entrypoints` · `api-map` · `state-map` · `detect` · `handbook` · `diff [--git-aware]` · `dashboard` · `history` · `graph-schema` · `resolve-types` -### Security (5) -`secrets [--severity ...]` · `taint` (AST-based) · `dataflow [--source ...] [--sink ...]` (cross-file) · `vuln-scan [--offline] [--osv-ttl N] [--refresh] [--max-age Nh]` (OSV.dev + native audit; `--refresh` bypasses cache, `--max-age Nh` overrides per-run TTL, `cache_info` in output signals staleness — issue #30) · `env-check [--var NAME]` +### Security (6) +`secrets [--severity ...]` · `taint` (AST-based) · `dataflow [--source ...] [--sink ...]` (cross-file) · `vuln-scan [--offline] [--osv-ttl N] [--refresh] [--max-age Nh]` (OSV.dev + native audit; `--refresh` bypasses cache, `--max-age Nh` overrides per-run TTL, `cache_info` in output signals staleness — issue #30) · `deps-audit [--severity ...] [--ecosystem PyPI|npm|crates.io] [--offline]` (pure-Python OSV.dev dependency audit, stores findings as `dependency_vuln` graph nodes — issue #158) · `env-check [--var NAME]` ### Quality (9) `smell [--categories ...] [--severity ...]` · `complexity [--name FN] [--threshold N] [--sort ...]` · `dead-code [--categories ...]` · `debug-leak [--category ...]` · `circular [--domain ...]` · `missing-refs` · `side-effect [--name FN]` · `perf-hint [--severity ...] [--category ...]` · `fix [--apply]` @@ -148,9 +148,9 @@ $CLI list --limit 5 --offset 10 --format compact # → paginated + co ### Tooling (1) `plugin ` -**Total: 71 commands** (auto-registered via `commands/__init__.py`; rerun `python3 scripts/sync_command_count.py --apply` after adding/removing a command) +**Total: 76 commands** (auto-registered via `commands/__init__.py`; rerun `python3 scripts/sync_command_count.py --apply` after adding/removing a command) -## MCP Server (69 Tools) +## MCP Server (74 Tools) Start the MCP server for AI agent integration: @@ -158,9 +158,9 @@ Start the MCP server for AI agent integration: python3 scripts/codelens.py serve ``` -Exposes 69 tools as `codelens_` (e.g., `codelens_query`, `codelens_taint`, `codelens_graph_schema`, `codelens_architecture`, `codelens_resolve_types`, `codelens_git_status`): +Exposes 74 tools as `codelens_` (e.g., `codelens_query`, `codelens_taint`, `codelens_graph_schema`, `codelens_architecture`, `codelens_resolve_types`, `codelens_git_status`): - 50 statically-defined tools (full JSON schemas in `mcp_server.py`) -- 15 dynamically-discovered tools (auto-discovered from `COMMAND_REGISTRY`; long-running `watch` and `serve` are excluded) +- 18 dynamically-discovered tools (auto-discovered from `COMMAND_REGISTRY`; long-running `watch` and `serve` are excluded) - Every tool accepts a `format` parameter (`json`/`markdown`/`ai`/`sarif`/`compact`). Use `format: "compact"` for token-efficient responses (~50% smaller than `json`). - `watch` and `serve` itself are excluded (long-running) diff --git a/SKILL.md b/SKILL.md index 50c98307..bb2c551a 100755 --- a/SKILL.md +++ b/SKILL.md @@ -1,10 +1,10 @@ --- name: codelens description: > - CodeLens — AI-Native Code Intelligence. 71 commands for AI-powered code analysis, + CodeLens — AI-Native Code Intelligence. 76 commands for AI-powered code analysis, security auditing, quality scoring, AST-based taint analysis, live CVE scanning, and pre-write safety checks. Supports 28+ languages with tree-sitter + regex - fallback parsing. MCP server exposes 69 tools for AI agent integration. + fallback parsing. MCP server exposes 74 tools for AI agent integration. For quick command reference with validated output schemas, see SKILL-QUICK.md. For version history, see CHANGELOG.md. --- diff --git a/pyproject.toml b/pyproject.toml index 63dfd97d..97251660 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta" [project] name = "codelens" version = "8.2.0" -description = "Live Codebase Reference Intelligence — 71 commands for AI-powered code analysis, security auditing, and quality scoring" +description = "Live Codebase Reference Intelligence — 76 commands for AI-powered code analysis, security auditing, and quality scoring" readme = "README.md" license = {text = "MIT"} requires-python = ">=3.8" diff --git a/scripts/graph_model.py b/scripts/graph_model.py index 8b2e8cd5..2a6a4b4d 100644 --- a/scripts/graph_model.py +++ b/scripts/graph_model.py @@ -13,7 +13,7 @@ - New tables `graph_nodes` and `graph_edges` are additive (prefixed `graph_` to avoid colliding with any existing table name). - The flat registry tables and JSON files are untouched. -- All 71 existing CLI commands continue to work unchanged. +- All 76 existing CLI commands continue to work unchanged. Schema: graph_nodes( @@ -66,6 +66,9 @@ NODE_TYPE_ROUTE = "route" NODE_TYPE_TYPE = "type" NODE_TYPE_INTERFACE = "interface" +# Dependency vulnerability node (issue #158: deps-audit stores findings as +# graph nodes linked to the lock file that contained the vulnerable package). +NODE_TYPE_DEPENDENCY_VULN = "dependency_vuln" # Edge types (per issue #8 spec) EDGE_TYPE_CALLS = "CALLS" @@ -74,6 +77,9 @@ EDGE_TYPE_INHERITS = "INHERITS" EDGE_TYPE_IMPLEMENTS = "IMPLEMENTS" EDGE_TYPE_USES_TYPE = "USES_TYPE" +# Edge from a file node to a dependency_vuln node (issue #158). The source is +# the lock file / manifest path, the target is the vuln node_id. +EDGE_TYPE_HAS_VULN = "HAS_VULN" # Map flat-registry node "type" values to graph node_type. # Anything not in this map defaults to NODE_TYPE_FUNCTION. diff --git a/skill.json b/skill.json index d236d4ae..1bc1470b 100755 --- a/skill.json +++ b/skill.json @@ -1,7 +1,7 @@ { "name": "codelens", "version": "8.2.0", - "description": "Live Codebase Reference Intelligence. 71 commands for AI-powered code analysis, security auditing, quality scoring, and pre-write safety checks. Supports 28+ languages with regex+AST hybrid parsing. Must activate before writing/editing/deleting any class, id, or function.", + "description": "Live Codebase Reference Intelligence. 76 commands for AI-powered code analysis, security auditing, quality scoring, and pre-write safety checks. Supports 28+ languages with regex+AST hybrid parsing. Must activate before writing/editing/deleting any class, id, or function.", "author": "codelens", "command_categories": { "setup": [