diff --git a/CHANGELOG.md b/CHANGELOG.md index 7fa51c7b..e8b903db 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/packages/pythinker-review/README.md b/packages/pythinker-review/README.md index 4c16b91a..9f1116f1 100644 --- a/packages/pythinker-review/README.md +++ b/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 diff --git a/packages/pythinker-review/src/pythinker_review/cli/security_scan.py b/packages/pythinker-review/src/pythinker_review/cli/security_scan.py index 099aa5f3..4d0d8cf6 100644 --- a/packages/pythinker-review/src/pythinker_review/cli/security_scan.py +++ b/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), diff --git a/packages/pythinker-review/src/pythinker_review/reviewers/prompts/security_review.system.md b/packages/pythinker-review/src/pythinker_review/reviewers/prompts/security_review.system.md index 877e2eed..9164e0f8 100644 --- a/packages/pythinker-review/src/pythinker_review/reviewers/prompts/security_review.system.md +++ b/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: diff --git a/packages/pythinker-review/src/pythinker_review/security_intel/__init__.py b/packages/pythinker-review/src/pythinker_review/security_intel/__init__.py new file mode 100644 index 00000000..15647c95 --- /dev/null +++ b/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"] diff --git a/packages/pythinker-review/src/pythinker_review/security_intel/cache.py b/packages/pythinker-review/src/pythinker_review/security_intel/cache.py new file mode 100644 index 00000000..3de18157 --- /dev/null +++ b/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 diff --git a/packages/pythinker-review/src/pythinker_review/security_intel/client.py b/packages/pythinker-review/src/pythinker_review/security_intel/client.py new file mode 100644 index 00000000..b111c7be --- /dev/null +++ b/packages/pythinker-review/src/pythinker_review/security_intel/client.py @@ -0,0 +1,148 @@ +"""Bounded HTTP client for public vulnerability-intelligence APIs. + +The client is deliberately narrow: it validates hosts, caps response bodies, redacts sensitive data +from errors, and never targets arbitrary project URLs. +""" + +from __future__ import annotations + +import asyncio +import json +import urllib.error +import urllib.parse +import urllib.request +from dataclasses import dataclass +from typing import Any + +from pythinker_review.security_intel.validators import sanitize_url_for_log, validate_intel_url + +DEFAULT_TIMEOUT_S = 30.0 +DEFAULT_MAX_RESPONSE_BYTES = 10 * 1024 * 1024 + + +class IntelClientError(RuntimeError): + pass + + +class _NoRedirectHandler(urllib.request.HTTPRedirectHandler): + """Disable implicit redirects so host allowlisting cannot be bypassed.""" + + def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: ANN001 + return None + + +@dataclass(frozen=True, slots=True) +class IntelResponse: + status: int + body: bytes + headers: dict[str, str] + + def json(self) -> Any: + return json.loads(self.body.decode("utf-8")) + + def text(self) -> str: + return self.body.decode("utf-8", errors="replace") + + +class IntelHttpClient: + def __init__( + self, + *, + timeout_s: float = DEFAULT_TIMEOUT_S, + max_response_bytes: int = DEFAULT_MAX_RESPONSE_BYTES, + ) -> None: + self.timeout_s = timeout_s + self.max_response_bytes = max_response_bytes + self._opener = urllib.request.build_opener(_NoRedirectHandler) + + async def get_json( + self, + url: str, + *, + params: dict[str, object] | None = None, + headers: dict[str, str] | None = None, + ) -> Any: + response = await self.get(url, params=params, headers=headers) + return response.json() + + async def post_json( + self, + url: str, + *, + payload: dict[str, object], + headers: dict[str, str] | None = None, + ) -> Any: + response = await self.post(url, payload=payload, headers=headers) + return response.json() + + async def get_text( + self, + url: str, + *, + params: dict[str, object] | None = None, + headers: dict[str, str] | None = None, + ) -> str: + response = await self.get(url, params=params, headers=headers) + return response.text() + + async def get( + self, + url: str, + *, + params: dict[str, object] | None = None, + headers: dict[str, str] | None = None, + ) -> IntelResponse: + return await asyncio.to_thread(self._request, "GET", url, params, None, headers) + + async def post( + self, + url: str, + *, + payload: dict[str, object], + headers: dict[str, str] | None = None, + ) -> IntelResponse: + return await asyncio.to_thread(self._request, "POST", url, None, payload, headers) + + def _request( + self, + method: str, + url: str, + params: dict[str, object] | None, + payload: dict[str, object] | None, + headers: dict[str, str] | None, + ) -> IntelResponse: + if params: + query = urllib.parse.urlencode(params) + separator = "&" if "?" in url else "?" + url = f"{url}{separator}{query}" + validate_intel_url(url) + body: bytes | None = None + request_headers = {"Accept": "application/json", **(headers or {})} + if payload is not None: + body = json.dumps(payload).encode("utf-8") + request_headers.setdefault("Content-Type", "application/json") + req = urllib.request.Request(url, data=body, headers=request_headers, method=method) + safe_url = sanitize_url_for_log(url) + try: + with self._opener.open(req, timeout=self.timeout_s) as resp: + content_length = resp.headers.get("content-length") + if content_length: + try: + if int(content_length) > self.max_response_bytes: + raise IntelClientError("response too large (Content-Length)") + except ValueError as exc: + raise IntelClientError("invalid Content-Length header") from exc + data = resp.read(self.max_response_bytes + 1) + if len(data) > self.max_response_bytes: + raise IntelClientError("response too large (actual body)") + return IntelResponse( + status=resp.status, + body=data, + headers={key.lower(): value for key, value in resp.headers.items()}, + ) + except urllib.error.HTTPError as exc: + raise IntelClientError(f"HTTP {exc.code} for {safe_url}") from exc + except urllib.error.URLError as exc: + raise IntelClientError(f"network error for {safe_url}: {exc.reason}") from exc + except TimeoutError as exc: + raise IntelClientError(f"timeout for {safe_url}") from exc diff --git a/packages/pythinker-review/src/pythinker_review/security_intel/models.py b/packages/pythinker-review/src/pythinker_review/security_intel/models.py new file mode 100644 index 00000000..0f4aacd6 --- /dev/null +++ b/packages/pythinker-review/src/pythinker_review/security_intel/models.py @@ -0,0 +1,122 @@ +"""Normalized vulnerability-intelligence models. + +External API models use ``extra='ignore'`` because public security feeds evolve. Internal models use +``extra='forbid'`` so stored Pythinker state remains stable. +""" + +from __future__ import annotations + +from typing import Any, Literal + +from pydantic import BaseModel, ConfigDict, Field + +RiskLabel = Literal["LOW", "MEDIUM", "HIGH", "CRITICAL"] +PoCConfidence = Literal[ + "WEAPONIZED", + "PUBLIC_EXPLOIT_REMOTE", + "PUBLIC_EXPLOIT", + "PUBLIC_POC_HIGH_QUALITY", + "PUBLIC_POC_LOW_QUALITY", + "NONE", +] + + +class ExternalIntelModel(BaseModel): + model_config = ConfigDict(extra="ignore") + + +class IntelModel(BaseModel): + model_config = ConfigDict(extra="forbid", populate_by_name=True) + + +class CVERecord(ExternalIntelModel): + id: str + published: str = "" + lastModified: str = "" + vulnStatus: str = "" + descriptions: list[dict[str, Any]] = Field(default_factory=list) + metrics: dict[str, Any] = Field(default_factory=dict) + weaknesses: list[dict[str, Any]] = Field(default_factory=list) + references: list[dict[str, Any]] = Field(default_factory=list) + + +class EPSSScore(ExternalIntelModel): + cve: str + epss: float = Field(ge=0.0, le=1.0) + percentile: float = Field(ge=0.0, le=1.0) + date: str = "" + + +class KEVEntry(ExternalIntelModel): + cveID: str + vendorProject: str = "" + product: str = "" + vulnerabilityName: str = "" + dateAdded: str = "" + dueDate: str = "" + knownRansomwareCampaignUse: str = "" + + +class PackageRef(IntelModel): + name: str + ecosystem: str + version: str = "" + manifest_path: str | None = None + line: int | None = Field(default=None, ge=1) + + +class PackageVulnerability(IntelModel): + id: str + summary: str = "" + aliases: list[str] = Field(default_factory=list) + severity: str = "UNKNOWN" + references: list[str] = Field(default_factory=list) + + +class DependencyIntel(IntelModel): + package: PackageRef + vuln_count: int = Field(ge=0) + vulns: list[PackageVulnerability] = Field(default_factory=list) + + +class ExploitIntel(IntelModel): + cve_id: str + has_public_exploit: bool = False + poc_count: int = 0 + confidence: PoCConfidence = "NONE" + references: list[str] = Field(default_factory=list) + + +class VendorAdvisoryIntel(IntelModel): + cve_id: str + microsoft: list[dict[str, Any]] = Field(default_factory=list) + redhat: list[dict[str, Any]] = Field(default_factory=list) + ubuntu: list[dict[str, Any]] = Field(default_factory=list) + + +class RiskScore(IntelModel): + cve_id: str + risk_score: float = Field(ge=0.0, le=100.0) + risk_label: RiskLabel + urgency: str + recommendation: str + components: dict[str, Any] + boosters_applied: list[str] = Field(default_factory=list) + days_since_published: int | None = None + + +class CVEIntelBundle(IntelModel): + cve_id: str + nvd: CVERecord | None = None + epss: EPSSScore | None = None + kev: KEVEntry | None = None + exploit: ExploitIntel | None = None + vendor: VendorAdvisoryIntel | None = None + risk: RiskScore | None = None + source_errors: list[str] = Field(default_factory=list) + + +class IntelSourceStatus(IntelModel): + source: str + ok: bool + message: str = "" diff --git a/packages/pythinker-review/src/pythinker_review/security_intel/risk.py b/packages/pythinker-review/src/pythinker_review/security_intel/risk.py new file mode 100644 index 00000000..b0859262 --- /dev/null +++ b/packages/pythinker-review/src/pythinker_review/security_intel/risk.py @@ -0,0 +1,134 @@ +from __future__ import annotations + +from datetime import UTC, datetime +from typing import Any + +from pythinker_review.security_intel.models import ( + CVERecord, + EPSSScore, + ExploitIntel, + KEVEntry, + RiskScore, +) + + +def _cvss_score(record: CVERecord | None) -> float: + if record is None: + return 0.0 + for key in ("cvssMetricV31", "cvssMetricV30", "cvssMetricV2"): + entries = record.metrics.get(key, []) + if entries and isinstance(entries, list): + score = entries[0].get("cvssData", {}).get("baseScore") + if score is not None: + return float(score) + return 0.0 + + +def _published_date(record: CVERecord | None) -> datetime | None: + if record is None or not record.published: + return None + raw = record.published + for fmt in ("%Y-%m-%dT%H:%M:%S.%f", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%d"): + try: + return datetime.strptime(raw, fmt).replace(tzinfo=UTC) + except ValueError: + continue + try: + return datetime.fromisoformat(raw.replace("Z", "+00:00")) + except ValueError: + return None + + +def score_cve( + *, + cve_id: str, + nvd: CVERecord | None, + epss: EPSSScore | None, + kev: KEVEntry | None, + exploit: ExploitIntel | None, +) -> RiskScore: + cvss = _cvss_score(nvd) + epss_probability = epss.epss if epss else 0.0 + in_kev = kev is not None + poc_confidence = exploit.confidence if exploit else "NONE" + poc_score = { + "WEAPONIZED": 15, + "PUBLIC_EXPLOIT_REMOTE": 12, + "PUBLIC_EXPLOIT": 10, + "PUBLIC_POC_HIGH_QUALITY": 7, + "PUBLIC_POC_LOW_QUALITY": 3, + "NONE": 0, + }[poc_confidence] + + base = (cvss / 10.0) * 20.0 + epss_probability * 35.0 + (30.0 if in_kev else 0.0) + poc_score + multiplier = 1.0 + boosters: list[str] = [] + if in_kev and poc_confidence != "NONE": + multiplier *= 1.15 + boosters.append("KEV+PoC") + if cvss >= 9.0 and epss_probability > 0.7: + multiplier *= 1.10 + boosters.append("CVSS>=9+EPSS>0.7") + days_since_published: int | None = None + if published := _published_date(nvd): + days_since_published = (datetime.now(UTC) - published).days + if days_since_published <= 7: + multiplier *= 1.05 + boosters.append("Published<7days") + risk_score = min(100.0, round(base * multiplier, 2)) + if risk_score <= 25: + label = "LOW" + elif risk_score <= 50: + label = "MEDIUM" + elif risk_score <= 75: + label = "HIGH" + else: + label = "CRITICAL" + if in_kev and epss_probability > 0.5: + urgency = "PATCH IMMEDIATELY" + elif in_kev: + urgency = "PATCH WITHIN 24 HOURS" + elif epss_probability > 0.5: + urgency = "PATCH WITHIN 72 HOURS" + elif cvss >= 9.0: + urgency = "PATCH THIS WEEK" + elif cvss >= 7.0: + urgency = "PATCH THIS MONTH" + else: + urgency = "SCHEDULE FOR NEXT CYCLE" + recommendation = _recommendation(cve_id, urgency, cvss, epss_probability, in_kev) + components: dict[str, Any] = { + "cvss_score": cvss, + "epss_probability": epss_probability, + "in_kev": in_kev, + "poc_confidence": poc_confidence, + } + return RiskScore( + cve_id=cve_id, + risk_score=risk_score, + risk_label=label, + urgency=urgency, + recommendation=recommendation, + components=components, + boosters_applied=boosters, + days_since_published=days_since_published, + ) + + +def _recommendation( + cve_id: str, urgency: str, cvss: float, epss_probability: float, in_kev: bool +) -> str: + if urgency == "PATCH IMMEDIATELY": + return f"{cve_id} is actively prioritized: KEV-listed with high EPSS; patch immediately." + if in_kev: + return f"{cve_id} is listed in CISA KEV; apply vendor patches within 24 hours." + if epss_probability > 0.5: + return ( + f"{cve_id} has high EPSS ({epss_probability:.1%}); " + "prioritize remediation within 72 hours." + ) + if cvss >= 9.0: + return f"{cve_id} has critical CVSS {cvss}; schedule remediation this week." + if cvss >= 7.0: + return f"{cve_id} has high CVSS {cvss}; include in the next patch cycle." + return f"{cve_id} has lower immediate risk; schedule remediation in normal maintenance." diff --git a/packages/pythinker-review/src/pythinker_review/security_intel/service.py b/packages/pythinker-review/src/pythinker_review/security_intel/service.py new file mode 100644 index 00000000..56a91d26 --- /dev/null +++ b/packages/pythinker-review/src/pythinker_review/security_intel/service.py @@ -0,0 +1,95 @@ +from __future__ import annotations + +import asyncio +from pathlib import Path + +from pythinker_review.security_intel.cache import IntelCache +from pythinker_review.security_intel.client import IntelHttpClient +from pythinker_review.security_intel.models import CVEIntelBundle, DependencyIntel, PackageRef +from pythinker_review.security_intel.risk import score_cve +from pythinker_review.security_intel.sources import epss, github, kev, nvd, osv, vendor +from pythinker_review.security_intel.validators import normalize_cve + + +def default_cache(data_root: Path) -> IntelCache: + return IntelCache(data_root.parent / "security-intel" / "cache") + + +async def lookup_cve_bundle( + cve_id: str, + *, + data_root: Path, + client: IntelHttpClient | None = None, + include_exploit: bool = True, + include_vendor: bool = True, +) -> CVEIntelBundle: + normalized = normalize_cve(cve_id) + if normalized is None: + raise ValueError("Invalid CVE ID") + http = client or IntelHttpClient() + cache = default_cache(data_root) + errors: list[str] = [] + + async def capture(name: str, coro): + try: + return await coro + except Exception as exc: # noqa: BLE001 - intel sources degrade independently + errors.append(f"{name}: {type(exc).__name__}: {exc}") + return None + + nvd_task = capture("nvd", nvd.fetch_cve(normalized, client=http, cache=cache)) + epss_task = capture("epss", epss.get_epss([normalized], client=http, cache=cache)) + kev_task = capture("kev", kev.lookup_kev(normalized, client=http, cache=cache)) + tasks = [nvd_task, epss_task, kev_task] + if include_exploit: + tasks.append( + capture( + "github-exploit", + github.check_exploit_availability(normalized, client=http, cache=cache), + ) + ) + if include_vendor: + tasks.append( + capture("vendor", vendor.get_vendor_advisory(normalized, client=http, cache=cache)) + ) + results = await asyncio.gather(*tasks) + nvd_record = results[0] + epss_scores = results[1] or [] + kev_entry = results[2] + exploit = results[3] if include_exploit and len(results) > 3 else None + vendor_result = results[-1] if include_vendor else None + epss_score = epss_scores[0] if epss_scores else None + if any(x is not None for x in (nvd_record, epss_score, kev_entry, exploit)): + risk = score_cve( + cve_id=normalized, + nvd=nvd_record, + epss=epss_score, + kev=kev_entry, + exploit=exploit, + ) + else: + risk = None + return CVEIntelBundle( + cve_id=normalized, + nvd=nvd_record, + epss=epss_score, + kev=kev_entry, + exploit=exploit, + vendor=vendor_result, + risk=risk, + source_errors=errors, + ) + + +async def scan_packages( + packages: list[PackageRef], *, data_root: Path, client: IntelHttpClient | None = None +) -> list[DependencyIntel]: + http = client or IntelHttpClient() + return await osv.query_packages(packages, client=http, cache=default_cache(data_root)) + + +async def lookup_package( + package: PackageRef, *, data_root: Path, client: IntelHttpClient | None = None +) -> DependencyIntel: + http = client or IntelHttpClient() + return await osv.query_package(package, client=http, cache=default_cache(data_root)) diff --git a/packages/pythinker-review/src/pythinker_review/security_intel/sources/__init__.py b/packages/pythinker-review/src/pythinker_review/security_intel/sources/__init__.py new file mode 100644 index 00000000..25fcd978 --- /dev/null +++ b/packages/pythinker-review/src/pythinker_review/security_intel/sources/__init__.py @@ -0,0 +1 @@ +"""Public intelligence source adapters.""" diff --git a/packages/pythinker-review/src/pythinker_review/security_intel/sources/epss.py b/packages/pythinker-review/src/pythinker_review/security_intel/sources/epss.py new file mode 100644 index 00000000..cd1bbeaf --- /dev/null +++ b/packages/pythinker-review/src/pythinker_review/security_intel/sources/epss.py @@ -0,0 +1,35 @@ +from __future__ import annotations + +from pythinker_review.security_intel.cache import TTL_EPSS, IntelCache +from pythinker_review.security_intel.client import IntelHttpClient +from pythinker_review.security_intel.models import EPSSScore +from pythinker_review.security_intel.validators import normalize_cve + +EPSS_BASE = "https://api.first.org/data/v1/epss" +_CHUNK_SIZE = 30 + + +async def get_epss( + cve_ids: list[str], *, client: IntelHttpClient, cache: IntelCache +) -> list[EPSSScore]: + out: list[EPSSScore] = [] + uncached: list[str] = [] + for cve in cve_ids: + normalized = normalize_cve(cve) + if normalized is None: + raise ValueError(f"Invalid CVE ID: {cve}") + cached = cache.get(f"epss:{normalized}") + if cached is None: + uncached.append(normalized) + else: + out.append(EPSSScore.model_validate(cached)) + for start in range(0, len(uncached), _CHUNK_SIZE): + chunk = uncached[start : start + _CHUNK_SIZE] + data = await client.get_json( + EPSS_BASE, params={"cve": ",".join(chunk), "limit": len(chunk)} + ) + for item in data.get("data", []) if isinstance(data, dict) else []: + score = EPSSScore.model_validate(item) + cache.set(f"epss:{score.cve}", score.model_dump(), TTL_EPSS) + out.append(score) + return out diff --git a/packages/pythinker-review/src/pythinker_review/security_intel/sources/github.py b/packages/pythinker-review/src/pythinker_review/security_intel/sources/github.py new file mode 100644 index 00000000..daf76963 --- /dev/null +++ b/packages/pythinker-review/src/pythinker_review/security_intel/sources/github.py @@ -0,0 +1,91 @@ +from __future__ import annotations + +import os + +from pythinker_review.security_intel.cache import TTL_EXPLOIT, IntelCache +from pythinker_review.security_intel.client import IntelHttpClient +from pythinker_review.security_intel.models import ExploitIntel, PoCConfidence +from pythinker_review.security_intel.validators import normalize_cve + +GITHUB_REPO_SEARCH_URL = "https://api.github.com/search/repositories" + + +def _headers() -> dict[str, str]: + headers = {"Accept": "application/vnd.github+json"} + if token := os.environ.get("GITHUB_TOKEN"): + headers["Authorization"] = f"Bearer {token}" + return headers + + +def _score_repo(repo: dict) -> int: + score = 0 + stars = int(repo.get("stargazers_count") or 0) + if stars > 100: + score += 3 + elif stars > 10: + score += 2 + if not repo.get("fork", True): + score += 2 + desc = (repo.get("description") or "").lower() + if "exploit" in desc: + score += 1 + if "poc" in desc: + score += 1 + return score + + +def _confidence(scores: list[int]) -> PoCConfidence: + if not scores: + return "NONE" + if max(scores) >= 5: + return "PUBLIC_EXPLOIT" + return "PUBLIC_POC_LOW_QUALITY" + + +async def check_exploit_availability( + cve_id: str, *, client: IntelHttpClient, cache: IntelCache +) -> ExploitIntel: + normalized = normalize_cve(cve_id) + if normalized is None: + raise ValueError("Invalid CVE ID") + key = f"github:exploit:{normalized}" + cached = cache.get(key) + if cached is not None: + return ExploitIntel.model_validate(cached) + data = await client.get_json( + GITHUB_REPO_SEARCH_URL, + params={"q": f"{normalized} poc exploit", "sort": "stars", "order": "desc", "per_page": 10}, + headers=_headers(), + ) + if not (isinstance(data, dict) and "items" in data and isinstance(data["items"], list)): + return ExploitIntel( + cve_id=normalized, + has_public_exploit=False, + poc_count=0, + confidence="NONE", + references=[], + ) + refs: list[str] = [] + scores: list[int] = [] + for repo in data["items"]: + if not isinstance(repo, dict): + continue + full_name = repo.get("full_name", "") + description = repo.get("description") or "" + if normalized.lower() not in f"{full_name} {description}".lower(): + continue + score = _score_repo(repo) + if score < 2: + continue + scores.append(score) + if url := repo.get("html_url"): + refs.append(str(url)) + result = ExploitIntel( + cve_id=normalized, + has_public_exploit=bool(refs), + poc_count=len(refs), + confidence=_confidence(scores), + references=refs[:10], + ) + cache.set(key, result.model_dump(), TTL_EXPLOIT) + return result diff --git a/packages/pythinker-review/src/pythinker_review/security_intel/sources/kev.py b/packages/pythinker-review/src/pythinker_review/security_intel/sources/kev.py new file mode 100644 index 00000000..27164183 --- /dev/null +++ b/packages/pythinker-review/src/pythinker_review/security_intel/sources/kev.py @@ -0,0 +1,37 @@ +from __future__ import annotations + +from pythinker_review.security_intel.cache import TTL_KEV, IntelCache +from pythinker_review.security_intel.client import IntelHttpClient +from pythinker_review.security_intel.models import KEVEntry +from pythinker_review.security_intel.validators import normalize_cve + +CISA_KEV_URL = "https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json" +KEV_FALLBACK = "https://raw.githubusercontent.com/cisagov/kev-data/main/data/known_exploited_vulnerabilities.json" + + +async def fetch_kev_catalog(*, client: IntelHttpClient, cache: IntelCache) -> list[KEVEntry]: + cached = cache.get("kev:catalog") + if cached is not None: + return [KEVEntry.model_validate(item) for item in cached] + last_error: Exception | None = None + for url in (CISA_KEV_URL, KEV_FALLBACK): + try: + data = await client.get_json(url) + if not isinstance(data, dict) or not isinstance(data.get("vulnerabilities"), list): + raise ValueError(f"Unexpected KEV catalog shape from {url}") + entries = [KEVEntry.model_validate(item) for item in data["vulnerabilities"]] + cache.set("kev:catalog", [entry.model_dump() for entry in entries], TTL_KEV) + return entries + except Exception as exc: # noqa: BLE001 - fallback source boundary + last_error = exc + if last_error is not None: + raise last_error + return [] + + +async def lookup_kev(cve_id: str, *, client: IntelHttpClient, cache: IntelCache) -> KEVEntry | None: + normalized = normalize_cve(cve_id) + if normalized is None: + raise ValueError("Invalid CVE ID") + catalog = await fetch_kev_catalog(client=client, cache=cache) + return next((entry for entry in catalog if entry.cveID == normalized), None) diff --git a/packages/pythinker-review/src/pythinker_review/security_intel/sources/nvd.py b/packages/pythinker-review/src/pythinker_review/security_intel/sources/nvd.py new file mode 100644 index 00000000..53adb33f --- /dev/null +++ b/packages/pythinker-review/src/pythinker_review/security_intel/sources/nvd.py @@ -0,0 +1,73 @@ +from __future__ import annotations + +import os + +from pythinker_review.security_intel.cache import TTL_CVE, TTL_SEARCH, IntelCache +from pythinker_review.security_intel.client import IntelHttpClient +from pythinker_review.security_intel.models import CVERecord +from pythinker_review.security_intel.validators import normalize_cve, sanitize_keyword + +NVD_BASE = "https://services.nvd.nist.gov/rest/json/cves/2.0" +MAX_SEARCH_LIMIT = 50 + + +def _headers() -> dict[str, str]: + headers = {"Accept": "application/json"} + if key := os.environ.get("NVD_API_KEY"): + headers["apiKey"] = key + return headers + + +async def fetch_cve(cve_id: str, *, client: IntelHttpClient, cache: IntelCache) -> CVERecord | None: + normalized = normalize_cve(cve_id) + if normalized is None: + raise ValueError("Invalid CVE ID") + key = f"nvd:cve:{normalized}" + cached = cache.get(key) + if cached is not None: + return CVERecord.model_validate(cached) + data = await client.get_json(NVD_BASE, params={"cveId": normalized}, headers=_headers()) + if not isinstance(data, dict) or data.get("totalResults", 0) == 0: + return None + vulnerabilities = data.get("vulnerabilities", []) + if not vulnerabilities: + return None + record_data = vulnerabilities[0].get("cve", {}) + record = CVERecord.model_validate(record_data) + cache.set(key, record.model_dump(), TTL_CVE) + return record + + +async def search_cves( + query: str, + *, + severity: str = "", + limit: int = 10, + client: IntelHttpClient, + cache: IntelCache, +) -> list[CVERecord]: + safe_query = sanitize_keyword(query) + if safe_query is None: + raise ValueError("Invalid search query") + safe_limit = max(1, min(limit, MAX_SEARCH_LIMIT)) + sev = severity.upper().strip() + key = f"nvd:search:{safe_query}:{sev}:{safe_limit}" + cached = cache.get(key) + if cached is not None: + return [CVERecord.model_validate(item) for item in cached] + params: dict[str, object] = {"keywordSearch": safe_query, "resultsPerPage": safe_limit} + if sev: + params["cvssV3Severity"] = sev + data = await client.get_json(NVD_BASE, params=params, headers=_headers()) + if not isinstance(data, dict): + raise ValueError("Unexpected NVD response shape: expected a dict") + vulnerabilities = data.get("vulnerabilities", []) + if not isinstance(vulnerabilities, list): + raise ValueError("Unexpected NVD response shape: 'vulnerabilities' is not a list") + records = [ + CVERecord.model_validate(item.get("cve", {})) + for item in vulnerabilities + if isinstance(item, dict) + ] + cache.set(key, [record.model_dump() for record in records], TTL_SEARCH) + return records diff --git a/packages/pythinker-review/src/pythinker_review/security_intel/sources/osv.py b/packages/pythinker-review/src/pythinker_review/security_intel/sources/osv.py new file mode 100644 index 00000000..45809d15 --- /dev/null +++ b/packages/pythinker-review/src/pythinker_review/security_intel/sources/osv.py @@ -0,0 +1,130 @@ +from __future__ import annotations + +from pythinker_review.security_intel.cache import TTL_OSV, IntelCache +from pythinker_review.security_intel.client import IntelHttpClient +from pythinker_review.security_intel.models import DependencyIntel, PackageRef, PackageVulnerability +from pythinker_review.security_intel.validators import validate_ecosystem, validate_package_name + +OSV_BASE = "https://api.osv.dev/v1" +_MAX_BATCH = 1_000 + + +def _extract_severity(vuln: dict) -> str: + db_specific = vuln.get("database_specific", {}) + if isinstance(db_specific, dict) and db_specific.get("severity"): + return str(db_specific["severity"]).upper() + for affected in vuln.get("affected", []): + if not isinstance(affected, dict): + continue + eco = affected.get("ecosystem_specific", {}) + if isinstance(eco, dict) and eco.get("severity"): + return str(eco["severity"]).upper() + if vuln.get("severity"): + return "MEDIUM" + return "UNKNOWN" + + +def _summarize_vulns(vulns: list[dict]) -> list[PackageVulnerability]: + severity_order = {"CRITICAL": 0, "HIGH": 1, "MODERATE": 2, "MEDIUM": 2, "LOW": 3} + sorted_vulns = sorted(vulns, key=lambda v: severity_order.get(_extract_severity(v), 4)) + out: list[PackageVulnerability] = [] + for vuln in sorted_vulns[:5]: + out.append( + PackageVulnerability.model_validate( + { + "id": vuln.get("id", ""), + "summary": (vuln.get("summary") or vuln.get("details") or "")[:240], + "aliases": vuln.get("aliases", [])[:10], + "severity": _extract_severity(vuln), + "references": [ + r.get("url", "") + for r in vuln.get("references", [])[:5] + if isinstance(r, dict) + ], + } + ) + ) + return out + + +async def query_package( + package: PackageRef, *, client: IntelHttpClient, cache: IntelCache +) -> DependencyIntel: + package = PackageRef.model_validate( + { + **package.model_dump(), + "name": validate_package_name(package.name), + "ecosystem": validate_ecosystem(package.ecosystem), + } + ) + key = f"osv:pkg:{package.ecosystem}:{package.name}:{package.version}" + cached = cache.get(key) + if cached is not None: + return DependencyIntel.model_validate(cached) + payload: dict[str, object] = {"package": {"name": package.name, "ecosystem": package.ecosystem}} + if package.version: + payload["version"] = package.version + data = await client.post_json(f"{OSV_BASE}/query", payload=payload) + raw_vulns = data.get("vulns", []) if isinstance(data, dict) else [] + result = DependencyIntel( + package=package, vuln_count=len(raw_vulns), vulns=_summarize_vulns(raw_vulns) + ) + cache.set(key, result.model_dump(), TTL_OSV) + return result + + +async def query_packages( + packages: list[PackageRef], *, client: IntelHttpClient, cache: IntelCache +) -> list[DependencyIntel]: + if not packages: + return [] + output: list[DependencyIntel] = [] + for start in range(0, len(packages), _MAX_BATCH): + batch = packages[start : start + _MAX_BATCH] + uncached: list[PackageRef] = [] + cached_results: list[DependencyIntel] = [] + for package in batch: + try: + package = PackageRef.model_validate( + { + **package.model_dump(), + "name": validate_package_name(package.name), + "ecosystem": validate_ecosystem(package.ecosystem), + } + ) + except ValueError: + continue + key = f"osv:pkg:{package.ecosystem}:{package.name}:{package.version}" + cached = cache.get(key) + if cached is None: + uncached.append(package) + else: + cached_results.append(DependencyIntel.model_validate(cached)) + output.extend(result for result in cached_results if result.vuln_count > 0) + if not uncached: + continue + queries = [] + for package in uncached: + query: dict[str, object] = { + "package": {"name": package.name, "ecosystem": package.ecosystem} + } + if package.version: + query["version"] = package.version + queries.append(query) + data = await client.post_json(f"{OSV_BASE}/querybatch", payload={"queries": queries}) + results = data.get("results", []) if isinstance(data, dict) else [] + for idx, package in enumerate(uncached): + raw_vulns = [] + if idx < len(results) and isinstance(results[idx], dict): + raw_vulns = results[idx].get("vulns", []) or [] + result = DependencyIntel( + package=package, vuln_count=len(raw_vulns), vulns=_summarize_vulns(raw_vulns) + ) + cache.set( + f"osv:pkg:{package.ecosystem}:{package.name}:{package.version}", + result.model_dump(), + TTL_OSV, + ) + if result.vuln_count > 0: + output.append(result) + return output diff --git a/packages/pythinker-review/src/pythinker_review/security_intel/sources/vendor.py b/packages/pythinker-review/src/pythinker_review/security_intel/sources/vendor.py new file mode 100644 index 00000000..68912c87 --- /dev/null +++ b/packages/pythinker-review/src/pythinker_review/security_intel/sources/vendor.py @@ -0,0 +1,91 @@ +from __future__ import annotations + +import asyncio + +from pythinker_review.security_intel.cache import TTL_VENDOR, IntelCache +from pythinker_review.security_intel.client import IntelHttpClient +from pythinker_review.security_intel.models import VendorAdvisoryIntel +from pythinker_review.security_intel.validators import normalize_cve + +MSRC_SUG_URL = "https://api.msrc.microsoft.com/sug/v2.0/en-US/vulnerability" +REDHAT_SECURITY_BASE = "https://access.redhat.com/hydra/rest/securitydata" +UBUNTU_SECURITY_BASE = "https://ubuntu.com/security/cves" + + +async def get_vendor_advisory( + cve_id: str, *, client: IntelHttpClient, cache: IntelCache +) -> VendorAdvisoryIntel: + normalized = normalize_cve(cve_id) + if normalized is None: + raise ValueError("Invalid CVE ID") + key = f"vendor:{normalized}" + cached = cache.get(key) + if cached is not None: + return VendorAdvisoryIntel.model_validate(cached) + microsoft, redhat, ubuntu = await asyncio.gather( + _msrc(normalized, client), + _redhat(normalized, client), + _ubuntu(normalized, client), + ) + result = VendorAdvisoryIntel( + cve_id=normalized, microsoft=microsoft, redhat=redhat, ubuntu=ubuntu + ) + cache.set(key, result.model_dump(), TTL_VENDOR) + return result + + +async def _msrc(cve_id: str, client: IntelHttpClient) -> list[dict]: + try: + data = await client.get_json(MSRC_SUG_URL, params={"$filter": f"cveNumber eq '{cve_id}'"}) + except Exception: # noqa: BLE001 - best-effort enrichment + return [] + out: list[dict] = [] + for entry in data.get("value", [])[:20] if isinstance(data, dict) else []: + if not isinstance(entry, dict): + continue + out.append( + { + "title": entry.get("cveTitle", ""), + "severity": entry.get("severity", ""), + "impact": entry.get("impact", ""), + "article_url": entry.get("articleUrl1", ""), + "release_date": entry.get("releaseDate", ""), + } + ) + return out + + +async def _redhat(cve_id: str, client: IntelHttpClient) -> list[dict]: + try: + data = await client.get_json(f"{REDHAT_SECURITY_BASE}/cve/{cve_id}.json") + except Exception: # noqa: BLE001 - best-effort enrichment + return [] + if not isinstance(data, dict): + return [] + out: list[dict] = [] + for release in data.get("affected_release", [])[:20]: + if not isinstance(release, dict): + continue + out.append( + { + "product_name": release.get("product_name", ""), + "advisory": release.get("advisory", ""), + "package": release.get("package", ""), + "release_date": release.get("release_date", ""), + "severity": data.get("threat_severity", ""), + } + ) + return out + + +async def _ubuntu(cve_id: str, client: IntelHttpClient) -> list[dict]: + try: + data = await client.get_json(f"{UBUNTU_SECURITY_BASE}/{cve_id}.json") + except Exception: # noqa: BLE001 - best-effort enrichment + return [] + if not isinstance(data, dict): + return [] + status = data.get("status", []) + if isinstance(status, list): + return [item for item in status[:20] if isinstance(item, dict)] + return [] diff --git a/packages/pythinker-review/src/pythinker_review/security_intel/validators.py b/packages/pythinker-review/src/pythinker_review/security_intel/validators.py new file mode 100644 index 00000000..8e7abe1b --- /dev/null +++ b/packages/pythinker-review/src/pythinker_review/security_intel/validators.py @@ -0,0 +1,84 @@ +"""Input validation and redaction for public security-intelligence lookups.""" + +from __future__ import annotations + +import ipaddress +import re +from urllib.parse import urlparse + +CVE_RE = re.compile(r"^CVE-\d{4}-\d{4,}$") +SAFE_KEYWORD_RE = re.compile(r"^[a-zA-Z0-9\s\-_.:/@+]{1,240}$") +PACKAGE_RE = re.compile(r"^[a-zA-Z0-9@][a-zA-Z0-9_.@/+:\-]{0,240}$") +ECOSYSTEM_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9 ._+\-/]{0,80}$") + +ALLOWED_HOSTS = frozenset( + { + "services.nvd.nist.gov", + "api.osv.dev", + "api.first.org", + "www.cisa.gov", + "raw.githubusercontent.com", + "api.github.com", + "gitlab.com", + "api.msrc.microsoft.com", + "access.redhat.com", + "ubuntu.com", + } +) + +_SENSITIVE_PARAMS = re.compile( + r"((?:apikey|api_key|key|token|access_token|secret|client_secret)=)[^&\s]+", + re.IGNORECASE, +) +_BEARER = re.compile(r"\b(Bearer|token)\s+[A-Za-z0-9._\-+/=]+", re.IGNORECASE) + + +def normalize_cve(cve_id: str) -> str | None: + value = cve_id.strip().upper() + return value if CVE_RE.fullmatch(value) else None + + +def sanitize_keyword(query: str) -> str | None: + value = query.strip() + return value if SAFE_KEYWORD_RE.fullmatch(value) else None + + +def validate_package_name(name: str) -> str: + value = name.strip() + if not PACKAGE_RE.fullmatch(value): + raise ValueError("Invalid package name") + return value + + +def validate_ecosystem(ecosystem: str) -> str: + value = ecosystem.strip() + if not ECOSYSTEM_RE.fullmatch(value): + raise ValueError("Invalid ecosystem") + return value + + +def validate_intel_url(url: str) -> str: + parsed = urlparse(url) + if parsed.scheme != "https": + raise ValueError("Security intelligence requests must use https") + if parsed.hostname not in ALLOWED_HOSTS: + raise ValueError(f"Blocked request to unauthorized host: {parsed.hostname}") + return url + + +def sanitize_url_for_log(url: str) -> str: + return _BEARER.sub(r"\1 ***REDACTED***", _SENSITIVE_PARAMS.sub(r"\1***REDACTED***", url)) + + +def validate_ip_address(ip: str) -> str: + addr = ipaddress.ip_address(ip.strip()) + if addr.is_private or addr.is_loopback or addr.is_link_local or addr.is_reserved: + raise ValueError(f"Private/reserved IP not allowed: {ip}") + return str(addr) + + +def validate_hash(hash_str: str) -> str | None: + value = hash_str.strip().lower() + if len(value) in (32, 40, 64) and all(c in "0123456789abcdef" for c in value): + return value + return None diff --git a/packages/pythinker-review/src/pythinker_review/security_scan/dependencies.py b/packages/pythinker-review/src/pythinker_review/security_scan/dependencies.py new file mode 100644 index 00000000..90b25e3a --- /dev/null +++ b/packages/pythinker-review/src/pythinker_review/security_scan/dependencies.py @@ -0,0 +1,289 @@ +"""Dependency-manifest parsing and OSV-backed enrichment for Pythinker Security Scan.""" + +from __future__ import annotations + +import json +import re +import tomllib +from pathlib import Path +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field + +from pythinker_review.security_intel.models import DependencyIntel, PackageRef +from pythinker_review.security_intel.service import scan_packages +from pythinker_review.security_scan.paths import data_dir + +_VERSION_PREFIX_RE = re.compile(r"^[\^~>=<\s=]+") +_REQUIREMENT_RE = re.compile( + r"^([A-Za-z0-9_.\-]+(?:\[[A-Za-z0-9_,]+\])?)\s*(==|>=|<=|~=|!=|>|<)\s*([A-Za-z0-9_.\-+*,<>=]+)" +) +_BARE_REQUIREMENT_RE = re.compile(r"^([A-Za-z0-9_.\-]+(?:\[[A-Za-z0-9_,]+\])?)\s*$") + + +class DependencyScanReport(BaseModel): + model_config = ConfigDict(extra="forbid") + + project_id: str = Field(alias="projectId") + package_count: int = Field(alias="packageCount", ge=0) + vulnerable_count: int = Field(alias="vulnerableCount", ge=0) + dependencies: list[DependencyIntel] + source_errors: list[str] = Field(default_factory=list, alias="sourceErrors") + + +def parse_dependency_manifests(root: Path) -> list[PackageRef]: + """Parse dependency manifests at the given root directory (non-recursive). + + Only manifest files directly under *root* are considered. Manifests in + subdirectories are not discovered; callers that need recursive discovery + should walk the tree and call this function (or the individual helpers) per + directory. + """ + packages: list[PackageRef] = [] + for rel in ("requirements.txt", "requirements-dev.txt"): + path = root / rel + if path.exists(): + packages.extend(_parse_requirements(path, rel)) + package_json = root / "package.json" + if package_json.exists(): + packages.extend(_parse_package_json(package_json, "package.json")) + pyproject = root / "pyproject.toml" + if pyproject.exists(): + packages.extend(_parse_pyproject(pyproject, "pyproject.toml")) + pom = root / "pom.xml" + if pom.exists(): + packages.extend(_parse_pom_xml(pom, "pom.xml")) + return _dedupe(packages) + + +async def scan_project_dependencies( + *, project_id: str, root: Path, data_root: Path +) -> DependencyScanReport: + packages = parse_dependency_manifests(root) + errors: list[str] = [] + try: + vulnerable = await scan_packages(packages, data_root=data_root) + except Exception as exc: # noqa: BLE001 - surfaced in report instead of crashing local scans + vulnerable = [] + errors.append(f"OSV lookup failed: {type(exc).__name__}: {exc}") + report = DependencyScanReport.model_validate( + { + "projectId": project_id, + "packageCount": len(packages), + "vulnerableCount": len(vulnerable), + "dependencies": [item.model_dump() for item in vulnerable], + "sourceErrors": errors, + } + ) + write_dependency_report(report, data_root=data_root) + return report + + +def write_dependency_report(report: DependencyScanReport, *, data_root: Path) -> Path: + path = dependency_report_path(report.project_id, data_root=data_root) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(report.model_dump_json(by_alias=True, indent=2) + "\n", encoding="utf-8") + return path + + +def read_dependency_report(project_id: str, *, data_root: Path) -> DependencyScanReport | None: + path = dependency_report_path(project_id, data_root=data_root) + if not path.exists(): + return None + try: + return DependencyScanReport.model_validate_json(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return None + + +def dependency_report_path(project_id: str, *, data_root: Path) -> Path: + return data_dir(project_id, data_root=data_root) / "dependencies.json" + + +def _parse_requirements(path: Path, rel: str) -> list[PackageRef]: + out: list[PackageRef] = [] + for lineno, line in enumerate(_read_text(path).splitlines(), start=1): + stripped = line.strip() + if not stripped or stripped.startswith(("#", "-")): + continue + match = _REQUIREMENT_RE.match(stripped) + if match: + version = _clean_version(match.group(3)) if match.group(2) == "==" else "" + out.append( + PackageRef( + name=match.group(1).split("[", 1)[0], + ecosystem="PyPI", + version=version, + manifest_path=rel, + line=lineno, + ) + ) + continue + bare = _BARE_REQUIREMENT_RE.match(stripped) + if bare: + out.append( + PackageRef( + name=bare.group(1).split("[", 1)[0], + ecosystem="PyPI", + manifest_path=rel, + line=lineno, + ) + ) + return out + + +def _parse_package_json(path: Path, rel: str) -> list[PackageRef]: + try: + package = json.loads(_read_text(path)) + except json.JSONDecodeError: + return [] + if not isinstance(package, dict): + return [] + line_by_name = _line_index(path) + out: list[PackageRef] = [] + for key in ("dependencies", "devDependencies", "peerDependencies", "optionalDependencies"): + deps = package.get(key, {}) + if not isinstance(deps, dict): + continue + for name, raw_version in deps.items(): + version = _clean_version(str(raw_version)) + out.append( + PackageRef( + name=name, + ecosystem="npm", + version=version, + manifest_path=rel, + line=line_by_name.get(name), + ) + ) + return out + + +def _parse_pyproject(path: Path, rel: str) -> list[PackageRef]: + try: + project = tomllib.loads(_read_text(path)) + except tomllib.TOMLDecodeError: + return [] + out: list[PackageRef] = [] + deps = project.get("project", {}).get("dependencies", []) + if isinstance(deps, list): + out.extend(_python_dep_refs(deps, rel, path)) + optional = project.get("project", {}).get("optional-dependencies", {}) + if isinstance(optional, dict): + for values in optional.values(): + if isinstance(values, list): + out.extend(_python_dep_refs(values, rel, path)) + poetry_deps = project.get("tool", {}).get("poetry", {}).get("dependencies", {}) + if isinstance(poetry_deps, dict): + for name, raw_version in poetry_deps.items(): + if name.lower() == "python": + continue + version = _clean_version(str(raw_version)) if isinstance(raw_version, str) else "" + out.append( + PackageRef( + name=name, + ecosystem="PyPI", + version=version, + manifest_path=rel, + line=_find_line(path, name), + ) + ) + return out + + +def _parse_pom_xml(path: Path, rel: str) -> list[PackageRef]: + # Minimal Maven parser without external XML dependencies. It intentionally ignores complex + # property resolution and only extracts direct dependency coordinates. + text = _read_text(path) + out: list[PackageRef] = [] + for match in re.finditer(r"(.*?)", text, flags=re.DOTALL): + block = match.group(1) + group = _xml_text(block, "groupId") + artifact = _xml_text(block, "artifactId") + version = _xml_text(block, "version") + if group and artifact: + if version.startswith("${"): + version = "" + line = text[: match.start()].count("\n") + 1 + out.append( + PackageRef( + name=f"{group}:{artifact}", + ecosystem="Maven", + version=version, + manifest_path=rel, + line=line, + ) + ) + return out + + +def _python_dep_refs(items: list[Any], rel: str, path: Path) -> list[PackageRef]: + out: list[PackageRef] = [] + for item in items: + if not isinstance(item, str): + continue + match = re.match(r"^([A-Za-z0-9_.\-]+)(?:\[[^]]+\])?\s*([><=!~]+)?\s*([^;\s]+)?", item) + if not match: + continue + name = match.group(1) + operator = match.group(2) or "" + version = _clean_version(match.group(3) or "") if operator == "==" else "" + out.append( + PackageRef( + name=name, + ecosystem="PyPI", + version=version, + manifest_path=rel, + line=_find_line(path, name), + ) + ) + return out + + +def _clean_version(raw: str) -> str: + value = _VERSION_PREFIX_RE.sub("", raw).strip().strip("\"'") + return "" if value in {"", "*", "latest"} else value + + +def _xml_text(block: str, tag: str) -> str: + match = re.search(rf"<{tag}>(.*?)", block, flags=re.DOTALL) + return match.group(1).strip() if match else "" + + +def _line_index(path: Path) -> dict[str, int]: + out: dict[str, int] = {} + for lineno, line in enumerate(_read_text(path).splitlines(), start=1): + match = re.search(r'"([^"\\]+)"\s*:', line) + if match: + out.setdefault(match.group(1), lineno) + return out + + +def _find_line(path: Path, needle: str) -> int | None: + pattern = re.compile(r"\b" + re.escape(needle) + r"\b") + for lineno, line in enumerate(_read_text(path).splitlines(), start=1): + stripped = line.lstrip() + if stripped.startswith(("#", "//")): + continue + if pattern.search(line): + return lineno + return None + + +def _read_text(path: Path) -> str: + try: + return path.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError): + return "" + + +def _dedupe(packages: list[PackageRef]) -> list[PackageRef]: + seen: set[tuple[str, str, str]] = set() + out: list[PackageRef] = [] + for package in packages: + key = (package.ecosystem.lower(), package.name.lower(), package.version) + if key in seen: + continue + seen.add(key) + out.append(package) + return out diff --git a/packages/pythinker-review/src/pythinker_review/security_scan/processor.py b/packages/pythinker-review/src/pythinker_review/security_scan/processor.py index b9a3c8b9..a23cc1eb 100644 --- a/packages/pythinker-review/src/pythinker_review/security_scan/processor.py +++ b/packages/pythinker-review/src/pythinker_review/security_scan/processor.py @@ -12,6 +12,7 @@ from pythinker_review.diagnostics.parser import redact_secrets from pythinker_review.engine.token_budget import clip_text from pythinker_review.llm.protocol import ReviewLLM +from pythinker_review.security_scan.dependencies import read_dependency_report from pythinker_review.security_scan.matchers import create_default_registry from pythinker_review.security_scan.models import ( AnalysisEntry, @@ -133,6 +134,11 @@ async def process_project( semaphore = asyncio.Semaphore(max(1, jobs)) counters = {"analysis": 0, "findings": 0, "errors": 0} error_messages: list[str] = [] + project_info = _with_dependency_context( + read_info(project_id, data_root=data_root), + project_id=project_id, + data_root=data_root, + ) async def worker(batch: list[FileRecord]) -> None: async with semaphore: @@ -144,7 +150,7 @@ async def worker(batch: list[FileRecord]) -> None: root=root, data_root=data_root, llm=llm, - project_info=read_info(project_id, data_root=data_root), + project_info=project_info, prompt_append=settings.prompt_append, timeout_s=timeout_s, ) @@ -185,6 +191,25 @@ def _safe_error_message(exc: Exception) -> str: return clip_text(redact_secrets(f"{type(exc).__name__}: {exc}"), 500) +def _with_dependency_context(project_info: str, *, project_id: str, data_root: Path) -> str: + report = read_dependency_report(project_id, data_root=data_root) + if report is None or not report.dependencies: + return project_info + lines = [project_info.rstrip(), "", "## Dependency vulnerability intelligence", ""] + for item in report.dependencies[:20]: + vulns = ", ".join(v.id for v in item.vulns[:3]) + severity = item.vulns[0].severity if item.vulns else "UNKNOWN" + lines.append( + f"- {item.package.ecosystem}/{item.package.name} " + f"{item.package.version or '(unversioned)'}: {severity} ({vulns})" + ) + lines.append( + "Treat this as supporting context only: report package issues only when the " + "manifest/lock evidence proves the vulnerable dependency is present." + ) + return "\n".join(lines).strip() + + async def _process_batch( *, batch: list[FileRecord], diff --git a/packages/pythinker-review/src/pythinker_review/security_scan/prompts/system.md b/packages/pythinker-review/src/pythinker_review/security_scan/prompts/system.md index 8ced9c76..a9e89857 100644 --- a/packages/pythinker-review/src/pythinker_review/security_scan/prompts/system.md +++ b/packages/pythinker-review/src/pythinker_review/security_scan/prompts/system.md @@ -3,7 +3,7 @@ You are Pythinker Security Scan, a production static-analysis security agent for ## Role and scope - Review source code statically for exploitable security issues and serious correctness bugs. -- Treat deterministic matcher hits as leads, not conclusions. +- Treat deterministic matcher hits and vulnerability-intelligence enrichment as leads, not conclusions. - Preserve signal quality: report only findings with a concrete source, sink, missing mitigation, and attacker path. - Prefer no finding over vague speculation. - Do not exploit, run target services, send network requests to the target, or execute proof-of-concept payloads. @@ -12,8 +12,9 @@ You are Pythinker Security Scan, a production static-analysis security agent for 1. Real exploitable vulnerabilities in production-reachable code. 2. Auth, authorization, tenant isolation, secret handling, code execution, injection, SSRF, path traversal, unsafe deserialization, XSS, webhook verification, supply-chain, IaC, and agent/tool trust-boundary flaws. -3. Major non-security bugs only when they can cause data loss, corruption, outages, or severely broken behavior. -4. Clear minimal remediation. +3. Dependency vulnerabilities only when package/version evidence proves the vulnerable dependency is present in a manifest or lockfile. +4. Major non-security bugs only when they can cause data loss, corruption, outages, or severely broken behavior. +5. Clear minimal remediation. ## Reasoning workflow @@ -24,7 +25,8 @@ For each target file: 3. Trace user-controlled or externally controlled inputs to sensitive sinks. 4. Check handler-local mitigations: auth middleware/guards/decorators, schema validation, permission checks, output escaping, allowlists, parameter binding, containment checks, signature verification, rate limits, and safe framework defaults. 5. Distinguish production code from tests, generated files, examples, vendored code, and docs. -6. Emit a finding only when the exploit path remains plausible after mitigation checks. +6. For CVE/OSV/EPSS/KEV/PoC context, verify the affected package, version, and reachability evidence; intelligence changes priority, not validity. +7. Emit a finding only when the exploit path remains plausible after mitigation checks. ## Severity guide diff --git a/packages/pythinker-review/src/pythinker_review/security_scan/reporting.py b/packages/pythinker-review/src/pythinker_review/security_scan/reporting.py index ccf6ec7e..996f48a7 100644 --- a/packages/pythinker-review/src/pythinker_review/security_scan/reporting.py +++ b/packages/pythinker-review/src/pythinker_review/security_scan/reporting.py @@ -8,6 +8,7 @@ from pathlib import Path from typing import Any, Literal +from pythinker_review.security_scan.dependencies import read_dependency_report from pythinker_review.security_scan.models import FileRecord, Finding from pythinker_review.security_scan.paths import reports_dir from pythinker_review.security_scan.store import load_all_file_records, read_project_config @@ -67,12 +68,16 @@ def metrics(project_id: str, *, data_root: Path) -> dict[str, Any]: for _record, finding in pairs if finding.revalidation is not None ) + dependency_report = read_dependency_report(project_id, data_root=data_root) return { "projectId": project_id, "findings": len(pairs), "bySeverity": dict(sorted(by_severity.items(), key=lambda item: -_SEVERITY_ORDER[item[0]])), "bySlug": dict(by_slug.most_common()), "revalidation": dict(revalidation), + "dependencyVulnerabilities": dependency_report.vulnerable_count + if dependency_report is not None + else 0, } @@ -92,8 +97,7 @@ def render_markdown_report(project_id: str, *, data_root: Path) -> str: "", ] if not pairs: - lines.append("No findings recorded.") - return "\n".join(lines) + "\n" + lines.append("No source findings recorded.") for record, finding in pairs: verdict = f" — {finding.revalidation.verdict}" if finding.revalidation else "" lines.extend( @@ -110,6 +114,33 @@ def render_markdown_report(project_id: str, *, data_root: Path) -> str: "", ] ) + dependency_report = read_dependency_report(project_id, data_root=data_root) + if dependency_report is not None: + lines.extend(["", "## Dependency vulnerabilities", ""]) + if not dependency_report.dependencies: + lines.append("No vulnerable dependencies recorded.") + else: + lines.append("| Package | Version | Vulnerabilities | Severity | Manifest |") + lines.append("| --- | --- | --- | --- | --- |") + for item in dependency_report.dependencies: + vulns = ", ".join(v.id for v in item.vulns[:3]) + severity = item.vulns[0].severity if item.vulns else "UNKNOWN" + manifest = item.package.manifest_path or "" + if item.package.line: + manifest += f":{item.package.line}" + lines.append( + f"| `{item.package.ecosystem}/{item.package.name}` | " + f"`{item.package.version or 'unversioned'}` | {vulns} | {severity} | " + f"`{manifest}` |" + ) + if dependency_report.source_errors: + lines.extend(["", "### Dependency intel errors", ""]) + _MAX_ERRORS = 10 + shown = dependency_report.source_errors[:_MAX_ERRORS] + lines.extend(f"- {error}" for error in shown) + remainder = len(dependency_report.source_errors) - len(shown) + if remainder > 0: + lines.append(f"- and {remainder} more errors") return "\n".join(lines).rstrip() + "\n" diff --git a/packages/pythinker-review/src/pythinker_review/signals/advisor.py b/packages/pythinker-review/src/pythinker_review/signals/advisor.py index 1759e91e..94ace0f2 100644 --- a/packages/pythinker-review/src/pythinker_review/signals/advisor.py +++ b/packages/pythinker-review/src/pythinker_review/signals/advisor.py @@ -21,6 +21,8 @@ "path-traversal-file-join-user-input": "Check path normalization and base containment.", "jwt-handling-algorithm-confusion": "Check signature verification and algorithm pinning.", "agentic-untrusted-prompt-input-prompt-injection": "Treat external text as data.", + "vulnerability-intel-cve-reference": "CVE mentions are context, not proof; require affected package/version evidence.", + "vulnerability-intel-dependency-change": "Dependency diffs may need OSV/NVD enrichment; validate actual vulnerable ranges before reporting.", } @@ -47,6 +49,8 @@ def build_advisor_context(*, repo: Path, signals_by_file: dict[str, list[Signal] sections.append(highlights) if slug_notes := _slug_notes(batch_slugs): sections.append(slug_notes) + if intel_notes := _intel_notes(signals_by_file): + sections.append(intel_notes) return "\n\n".join(sections) @@ -81,3 +85,34 @@ def _slug_notes(slugs: list[str]) -> str: if not lines: return "" return "## Slug-specific reviewer notes\n\n" + "\n".join(lines) + + +def _intel_notes(signals_by_file: dict[str, list[Signal]]) -> str: + cves = sorted( + { + signal.metadata.get("cve", "") + for signals in signals_by_file.values() + for signal in signals + if signal.metadata.get("cve") + } + ) + manifests = sorted( + { + signal.file + for signals in signals_by_file.values() + for signal in signals + if signal.rule_id == "sec.signal.vulnerability_intel.dependency_change" + } + ) + if not cves and not manifests: + return "" + lines = ["## Vulnerability intelligence leads", ""] + if cves: + lines.append("CVE IDs mentioned in the diff: " + ", ".join(cves[:20])) + if manifests: + lines.append("Dependency manifests changed: " + ", ".join(manifests[:20])) + lines.append( + "Use these as leads only. Emit dependency findings only when changed manifest/lockfile " + "evidence proves the vulnerable package and version are present." + ) + return "\n".join(lines) diff --git a/packages/pythinker-review/src/pythinker_review/signals/models.py b/packages/pythinker-review/src/pythinker_review/signals/models.py index a11f2193..2f6a3c5e 100644 --- a/packages/pythinker-review/src/pythinker_review/signals/models.py +++ b/packages/pythinker-review/src/pythinker_review/signals/models.py @@ -1,6 +1,6 @@ from __future__ import annotations -from dataclasses import dataclass +from dataclasses import dataclass, field @dataclass(frozen=True, slots=True) @@ -17,6 +17,7 @@ class Signal: mitigation_hint: str | None = None cwe: str | None = None severity_hint: str | None = None + metadata: dict[str, str] = field(default_factory=dict) @property def slug(self) -> str: diff --git a/packages/pythinker-review/src/pythinker_review/signals/scanner.py b/packages/pythinker-review/src/pythinker_review/signals/scanner.py index 52128bed..a08dbf1b 100644 --- a/packages/pythinker-review/src/pythinker_review/signals/scanner.py +++ b/packages/pythinker-review/src/pythinker_review/signals/scanner.py @@ -22,6 +22,13 @@ class _Rule: severity_hint: str | None = None +_CVE_RE = re.compile(r"\bCVE-\d{4}-\d{4,}\b", re.IGNORECASE) +_DEP_LINE_RE = re.compile( + r"(?i)\b(?:dependencies|devDependencies|requires|requirement|package)\b|" + r"(?:==|>=|<=|!=|~=|>|<|\^|~)\s*\d+\.\d+(?:\.\d+)?(?:[.\-][A-Za-z0-9]+)*|" + r'"[A-Za-z0-9_\-]+"\s*:\s*"\d[\d.]*"' +) + _RULES: tuple[_Rule, ...] = ( _Rule( "sec.signal.secrets_exposure.aws_access_key", @@ -301,6 +308,36 @@ class _Rule: def scan_signals(*, file_path: str, added_lines: list[tuple[int, str]]) -> list[Signal]: out: list[Signal] = [] for lineno, text in added_lines: + for cve in _CVE_RE.findall(text): + out.append( + Signal( + rule_id="sec.signal.vulnerability_intel.cve_reference", + file=file_path, + line=lineno, + snippet=text.strip(), + reason="CVE identifier added; check whether the vulnerable component/version is introduced or documented as fixed.", + confidence=0.6, + sink_kind="dependency_or_vulnerability_reference", + mitigation_hint="Verify affected package/version evidence before reporting a dependency finding.", + severity_hint="medium", + metadata={"cve": cve.upper()}, + ) + ) + if _is_dependency_file(file_path) and _DEP_LINE_RE.search(text): + out.append( + Signal( + rule_id="sec.signal.vulnerability_intel.dependency_change", + file=file_path, + line=lineno, + snippet=text.strip(), + reason="Dependency manifest line changed; check OSV/NVD intelligence when online enrichment is enabled.", + confidence=0.45, + sink_kind="dependency_manifest", + mitigation_hint="Pin safe versions and verify vulnerable ranges against advisories.", + severity_hint="medium", + metadata={"manifest": file_path}, + ) + ) for rule in _RULES: if rule.pattern.search(text): out.append( @@ -320,3 +357,19 @@ def scan_signals(*, file_path: str, added_lines: list[tuple[int, str]]) -> list[ ) ) return out + + +def _is_dependency_file(file_path: str) -> bool: + return file_path.endswith( + ( + "requirements.txt", + "pyproject.toml", + "package.json", + "package-lock.json", + "pnpm-lock.yaml", + "yarn.lock", + "pom.xml", + "go.mod", + "Cargo.lock", + ) + ) diff --git a/packages/pythinker-review/tests/unit/test_security_intel.py b/packages/pythinker-review/tests/unit/test_security_intel.py new file mode 100644 index 00000000..d745ab13 --- /dev/null +++ b/packages/pythinker-review/tests/unit/test_security_intel.py @@ -0,0 +1,84 @@ +from __future__ import annotations + +import urllib.request +from http.client import HTTPMessage +from io import BytesIO +from pathlib import Path + +import pytest + +from pythinker_review.security_intel.cache import IntelCache +from pythinker_review.security_intel.client import _NoRedirectHandler +from pythinker_review.security_intel.models import CVERecord, EPSSScore, ExploitIntel +from pythinker_review.security_intel.risk import score_cve +from pythinker_review.security_intel.validators import ( + normalize_cve, + sanitize_url_for_log, + validate_ip_address, +) + + +def test_normalize_cve_and_reject_private_ip() -> None: + assert normalize_cve("cve-2024-12345") == "CVE-2024-12345" + assert normalize_cve("not-a-cve") is None + + with pytest.raises(ValueError): + validate_ip_address("127.0.0.1") + + +def test_sanitize_url_for_log_redacts_tokens() -> None: + redacted = sanitize_url_for_log("https://example.test/?api_key=secret&access_token=tok") + assert "api_key=secret" not in redacted + assert "access_token=tok" not in redacted + assert redacted.count("***REDACTED***") == 2 + + +def test_intel_client_disables_implicit_redirects() -> None: + handler = _NoRedirectHandler() + + request = urllib.request.Request("https://services.nvd.nist.gov/") + + assert ( + handler.redirect_request( + request, BytesIO(), 302, "Found", HTTPMessage(), "https://example.test/" + ) + is None + ) + + from pythinker_review.security_intel.client import IntelHttpClient + + client = IntelHttpClient() + assert any(isinstance(h, _NoRedirectHandler) for h in client._opener.handlers) + + +def test_intel_cache_roundtrip(tmp_path: Path) -> None: + cache = IntelCache(tmp_path) + cache.set("k", {"v": 1}, ttl=60) + + assert cache.get("k") == {"v": 1} + + +def test_risk_score_combines_cvss_epss_kev_and_poc() -> None: + nvd = CVERecord.model_validate( + { + "id": "CVE-2024-0001", + "published": "2024-01-01T00:00:00.000", + "metrics": { + "cvssMetricV31": [{"cvssData": {"baseScore": 9.8, "baseSeverity": "CRITICAL"}}] + }, + } + ) + epss = EPSSScore(cve="CVE-2024-0001", epss=0.8, percentile=0.99, date="2024-01-02") + exploit = ExploitIntel( + cve_id="CVE-2024-0001", + has_public_exploit=True, + poc_count=1, + confidence="PUBLIC_EXPLOIT", + references=["https://github.com/example/poc"], + ) + + result = score_cve(cve_id="CVE-2024-0001", nvd=nvd, epss=epss, kev=None, exploit=exploit) + + assert result.risk_label in {"HIGH", "CRITICAL"} + assert result.components["epss_probability"] == 0.8 + assert "CVSS>=9+EPSS>0.7" in result.boosters_applied diff --git a/packages/pythinker-review/tests/unit/test_security_scan_dependencies.py b/packages/pythinker-review/tests/unit/test_security_scan_dependencies.py new file mode 100644 index 00000000..4691ac27 --- /dev/null +++ b/packages/pythinker-review/tests/unit/test_security_scan_dependencies.py @@ -0,0 +1,78 @@ +from __future__ import annotations + +from pathlib import Path + +from pythinker_review.security_intel.models import DependencyIntel, PackageRef, PackageVulnerability +from pythinker_review.security_scan.dependencies import ( + DependencyScanReport, + parse_dependency_manifests, + read_dependency_report, + write_dependency_report, +) + + +def test_parse_dependency_manifests_reads_python_and_node(tmp_path: Path) -> None: + (tmp_path / "requirements.txt").write_text("requests==2.28.0\nflask\n", encoding="utf-8") + (tmp_path / "package.json").write_text( + '{"dependencies":{"lodash":"^4.17.20","@scope/pkg":"1.0.0"}}', encoding="utf-8" + ) + (tmp_path / "pyproject.toml").write_text( + '[project]\ndependencies = ["pydantic>=2.0"]\n', encoding="utf-8" + ) + + packages = parse_dependency_manifests(tmp_path) + keys = {(pkg.ecosystem, pkg.name, pkg.version) for pkg in packages} + + assert ("PyPI", "requests", "2.28.0") in keys + assert ("PyPI", "flask", "") in keys + assert ("npm", "lodash", "4.17.20") in keys + assert ("npm", "@scope/pkg", "1.0.0") in keys + assert ("PyPI", "pydantic", "") in keys + assert any(pkg.manifest_path == "requirements.txt" and pkg.line == 1 for pkg in packages) + + +def test_parse_requirements_strips_extras(tmp_path: Path) -> None: + (tmp_path / "requirements.txt").write_text( + "requests[socks]==2.31.0\nurllib3[secure]\n", encoding="utf-8" + ) + packages = parse_dependency_manifests(tmp_path) + names = {pkg.name for pkg in packages} + assert "requests" in names + assert "urllib3" in names + assert not any("[" in pkg.name for pkg in packages) + + +def test_dependency_report_roundtrip(tmp_path: Path) -> None: + report = DependencyScanReport.model_validate( + { + "projectId": "repo", + "packageCount": 1, + "vulnerableCount": 1, + "dependencies": [ + DependencyIntel( + package=PackageRef( + name="lodash", + ecosystem="npm", + version="4.17.20", + manifest_path="package.json", + line=3, + ), + vuln_count=1, + vulns=[ + PackageVulnerability( + id="GHSA-xxxx", + summary="prototype pollution", + severity="HIGH", + ) + ], + ).model_dump() + ], + } + ) + + write_dependency_report(report, data_root=tmp_path) + loaded = read_dependency_report("repo", data_root=tmp_path) + + assert loaded is not None + assert loaded.vulnerable_count == 1 + assert loaded.dependencies[0].package.name == "lodash" diff --git a/packages/pythinker-review/tests/unit/test_signals.py b/packages/pythinker-review/tests/unit/test_signals.py index 4e5ba354..f7f971e6 100644 --- a/packages/pythinker-review/tests/unit/test_signals.py +++ b/packages/pythinker-review/tests/unit/test_signals.py @@ -56,6 +56,33 @@ def test_advisor_context_detects_python_stack(tmp_path) -> None: assert "Threat highlights" in context +def test_detects_cve_and_dependency_manifest_leads() -> None: + findings = scan_signals( + file_path="package.json", + added_lines=[(2, '"lodash": "^4.17.20", // CVE-2020-8203')], + ) + ids = {signal.rule_id for signal in findings} + + assert "sec.signal.vulnerability_intel.cve_reference" in ids + assert "sec.signal.vulnerability_intel.dependency_change" in ids + assert any(signal.metadata.get("cve") == "CVE-2020-8203" for signal in findings) + + +def test_advisor_context_includes_vulnerability_intel_leads(tmp_path) -> None: + signals = { + "requirements.txt": scan_signals( + file_path="requirements.txt", + added_lines=[(1, "django==1.2 # CVE-2019-19844")], + ) + } + + context = build_advisor_context(repo=tmp_path, signals_by_file=signals) + + assert "Vulnerability intelligence leads" in context + assert "CVE-2019-19844" in context + assert "requirements.txt" in context + + def test_advisor_context_uses_ported_framework_highlights(tmp_path) -> None: (tmp_path / "package.json").write_text('{"dependencies":{"koa":"^2.15.0"}}') signals = {