diff --git a/README.md b/README.md index 2fc3f3af..5e750972 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,7 @@ ## Table of Contents +- [What's new (2026-06-21) — Dependency Vulnerability Scanning (OSV)](#whats-new-2026-06-21--dependency-vulnerability-scanning-osv) - [What's new (2026-06-21) — JSON Schema Validation](#whats-new-2026-06-21--json-schema-validation) - [What's new (2026-06-20) — SARIF 2.1.0 Findings Export](#whats-new-2026-06-20--sarif-210-findings-export) - [What's new (2026-06-20) — Text PII Detection & Redaction](#whats-new-2026-06-20--text-pii-detection--redaction) @@ -110,6 +111,12 @@ --- +## What's new (2026-06-21) — Dependency Vulnerability Scanning (OSV) + +Match the SBOM against known CVEs. Full reference: [`docs/source/Eng/doc/new_features/v58_features_doc.rst`](docs/source/Eng/doc/new_features/v58_features_doc.rst). + +- **`scan_components` / `match_package` / `is_affected` / `findings_to_sarif`** (`AC_scan_vulns`, `ac_scan_vulns`): `build_sbom` only *inventoried* dependencies and `to_sarif` only *exported* findings — nothing ever **produced** a vulnerability finding. This matches the SBOM's `(ecosystem, name, version)` components against an [OSV](https://osv.dev) advisory database (sweeping `introduced`/`fixed`/`last_affected` ranges, PEP-503 name normalization, severity→SARIF level) and bridges results into the existing SARIF exporter for GitHub/Azure DevOps code scanning. The advisory DB is **injected as data** (offline, deterministic); the live `osv.dev` query is an optional `fetcher` seam. Pure-stdlib `re`. + ## What's new (2026-06-21) — JSON Schema Validation Validate nested JSON against a real schema. Full reference: [`docs/source/Eng/doc/new_features/v57_features_doc.rst`](docs/source/Eng/doc/new_features/v57_features_doc.rst). diff --git a/README/README_zh-CN.md b/README/README_zh-CN.md index 03368989..d5da819a 100644 --- a/README/README_zh-CN.md +++ b/README/README_zh-CN.md @@ -12,6 +12,7 @@ ## 目录 +- [本次更新 (2026-06-21) — 依赖项漏洞扫描(OSV)](#本次更新-2026-06-21--依赖项漏洞扫描osv) - [本次更新 (2026-06-21) — JSON Schema 验证](#本次更新-2026-06-21--json-schema-验证) - [本次更新 (2026-06-20) — SARIF 2.1.0 发现项目导出](#本次更新-2026-06-20--sarif-210-发现项目导出) - [本次更新 (2026-06-20) — 文本 PII 检测与遮蔽](#本次更新-2026-06-20--文本-pii-检测与遮蔽) @@ -109,6 +110,12 @@ --- +## 本次更新 (2026-06-21) — 依赖项漏洞扫描(OSV) + +以 SBOM 比对已知 CVE。完整参考:[`docs/source/Zh/doc/new_features/v58_features_doc.rst`](../docs/source/Zh/doc/new_features/v58_features_doc.rst)。 + +- **`scan_components` / `match_package` / `is_affected` / `findings_to_sarif`**(`AC_scan_vulns`、`ac_scan_vulns`):`build_sbom` 只会*盘点*依赖项,`to_sarif` 只会*导出*发现项目 —— 从未真正**产生**漏洞发现项目。本功能以 SBOM 的 `(ecosystem, name, version)` 组件比对 [OSV](https://osv.dev) 咨询数据库(扫描 `introduced`/`fixed`/`last_affected` 范围、PEP-503 名称规范化、严重度对应 SARIF 等级),并把结果桥接到既有 SARIF 导出器供 GitHub/Azure DevOps 代码扫描。咨询数据库以**数据注入**(离线、确定性);线上 `osv.dev` 查询为可选的 `fetcher` 接缝。纯标准库 `re`。 + ## 本次更新 (2026-06-21) — JSON Schema 验证 以真正的 schema 验证嵌套 JSON。完整参考:[`docs/source/Zh/doc/new_features/v57_features_doc.rst`](../docs/source/Zh/doc/new_features/v57_features_doc.rst)。 diff --git a/README/README_zh-TW.md b/README/README_zh-TW.md index fc9d5705..dd443bf5 100644 --- a/README/README_zh-TW.md +++ b/README/README_zh-TW.md @@ -12,6 +12,7 @@ ## 目錄 +- [本次更新 (2026-06-21) — 相依套件漏洞掃描(OSV)](#本次更新-2026-06-21--相依套件漏洞掃描osv) - [本次更新 (2026-06-21) — JSON Schema 驗證](#本次更新-2026-06-21--json-schema-驗證) - [本次更新 (2026-06-20) — SARIF 2.1.0 發現項目匯出](#本次更新-2026-06-20--sarif-210-發現項目匯出) - [本次更新 (2026-06-20) — 文字 PII 偵測與遮蔽](#本次更新-2026-06-20--文字-pii-偵測與遮蔽) @@ -109,6 +110,12 @@ --- +## 本次更新 (2026-06-21) — 相依套件漏洞掃描(OSV) + +以 SBOM 比對已知 CVE。完整參考:[`docs/source/Zh/doc/new_features/v58_features_doc.rst`](../docs/source/Zh/doc/new_features/v58_features_doc.rst)。 + +- **`scan_components` / `match_package` / `is_affected` / `findings_to_sarif`**(`AC_scan_vulns`、`ac_scan_vulns`):`build_sbom` 只會*盤點*相依套件,`to_sarif` 只會*匯出*發現項目 —— 從未真正**產生**漏洞發現項目。本功能以 SBOM 的 `(ecosystem, name, version)` 元件比對 [OSV](https://osv.dev) 諮詢資料庫(掃描 `introduced`/`fixed`/`last_affected` 範圍、PEP-503 名稱正規化、嚴重度對應 SARIF 等級),並把結果橋接到既有 SARIF 匯出器供 GitHub/Azure DevOps 程式碼掃描。諮詢資料庫以**資料注入**(離線、具決定性);線上 `osv.dev` 查詢為選用的 `fetcher` 接縫。純標準函式庫 `re`。 + ## 本次更新 (2026-06-21) — JSON Schema 驗證 以真正的 schema 驗證巢狀 JSON。完整參考:[`docs/source/Zh/doc/new_features/v57_features_doc.rst`](../docs/source/Zh/doc/new_features/v57_features_doc.rst)。 diff --git a/docs/source/Eng/doc/new_features/v58_features_doc.rst b/docs/source/Eng/doc/new_features/v58_features_doc.rst new file mode 100644 index 00000000..85a2bb60 --- /dev/null +++ b/docs/source/Eng/doc/new_features/v58_features_doc.rst @@ -0,0 +1,72 @@ +Dependency Vulnerability Scanning (OSV) +======================================= + +``build_sbom`` *inventories* dependencies and ``to_sarif`` *exports* findings, +but nothing in between ever **produced** a vulnerability finding — there was no +advisory matching at all. This closes that loop: given the SBOM's +``(ecosystem, name, version)`` components and an `OSV `_ +advisory database, ``scan_components`` reports which packages are affected, and +``findings_to_sarif`` bridges the results straight into the existing SARIF +exporter for GitHub / Azure DevOps code scanning. + +The advisory database is **injected as plain data** (a list of OSV records), so +matching is fully offline and deterministic; the live ``osv.dev`` query is a +separate, optional ``fetcher`` seam. Pure standard library (``re``); imports no +``PySide6``. + +Headless API +------------ + +.. code-block:: python + + from je_auto_control import ( + build_sbom, scan_components, findings_to_sarif, write_sarif) + + advisories = [{ + "id": "GHSA-foo", "summary": "RCE in foo", + "database_specific": {"severity": "HIGH"}, + "affected": [{ + "package": {"ecosystem": "PyPI", "name": "foo"}, + "ranges": [{"type": "ECOSYSTEM", + "events": [{"introduced": "0"}, {"fixed": "1.2.0"}]}], + }], + }] + + sbom = build_sbom("je_auto_control") + findings = scan_components(sbom["components"], advisories) + # findings: [{id, package, version, summary, severity, fixed, aliases}] + write_sarif(findings_to_sarif(findings), "vulns.sarif", + tool_name="AutoControl-VulnScan") + +``is_affected(version, osv_range)`` evaluates one OSV range by sweeping its +``introduced`` / ``fixed`` / ``last_affected`` events; ``match_package`` checks +a single package against the database (explicit ``versions`` **or** ranges); +``scan_components`` runs the whole SBOM. Package names are compared with PEP-503 +normalization, and the OSV severity word (``CRITICAL`` / ``HIGH`` / +``MODERATE`` / ``LOW``) maps to a SARIF ``error`` / ``warning`` / ``note`` +level (defaulting to ``warning``). + +Version ordering is a pragmatic numeric comparison: release components compare +as integers and a pre-release suffix (``1.2.0-rc1``) sorts before the final +release. Git ranges and full CVSS-vector scoring are out of scope. + +Live advisories (optional) +-------------------------- + +Pass a ``fetcher`` callable to pull advisories at scan time (e.g. from the +``osv.dev`` API). Tests inject a fake fetcher, so the matching logic is verified +without any network: + +.. code-block:: python + + findings = scan_components(sbom["components"], None, + fetcher=my_osv_fetcher) + +Executor command +---------------- + +``AC_scan_vulns`` takes ``components`` (a component list, a full SBOM dict, or a +JSON string) and ``advisories`` (a list or JSON string), with an optional +``sarif_path`` to write a SARIF report; it returns ``{findings, count}`` (plus +``sarif_path`` when written). The same operation is exposed as the MCP tool +``ac_scan_vulns`` and as a Script Builder command under **Security**. diff --git a/docs/source/Eng/eng_index.rst b/docs/source/Eng/eng_index.rst index 42311254..ae96698f 100644 --- a/docs/source/Eng/eng_index.rst +++ b/docs/source/Eng/eng_index.rst @@ -80,6 +80,7 @@ Comprehensive guides for all AutoControl features. doc/new_features/v55_features_doc doc/new_features/v56_features_doc doc/new_features/v57_features_doc + doc/new_features/v58_features_doc doc/ocr_backends/ocr_backends_doc doc/observability/observability_doc doc/operations_layer/operations_layer_doc diff --git a/docs/source/Zh/doc/new_features/v58_features_doc.rst b/docs/source/Zh/doc/new_features/v58_features_doc.rst new file mode 100644 index 00000000..504ffc07 --- /dev/null +++ b/docs/source/Zh/doc/new_features/v58_features_doc.rst @@ -0,0 +1,64 @@ +相依套件漏洞掃描(OSV) +====================== + +``build_sbom`` 會*盤點*相依套件,``to_sarif`` 會*匯出*發現項目,但兩者之間從未真正**產生** +漏洞發現項目 —— 完全沒有諮詢比對。本功能補上這個環節:給定 SBOM 的 +``(ecosystem, name, version)`` 元件與一份 `OSV `_ 諮詢資料庫, +``scan_components`` 會回報哪些套件受影響,而 ``findings_to_sarif`` 把結果直接橋接到既有的 +SARIF 匯出器,供 GitHub / Azure DevOps 程式碼掃描使用。 + +諮詢資料庫以**純資料注入**(一份 OSV 紀錄清單),因此比對完全離線且具決定性;線上的 +``osv.dev`` 查詢則是另一個選用的 ``fetcher`` 接縫。純標準函式庫(``re``);不匯入 +``PySide6``。 + +無頭 API +-------- + +.. code-block:: python + + from je_auto_control import ( + build_sbom, scan_components, findings_to_sarif, write_sarif) + + advisories = [{ + "id": "GHSA-foo", "summary": "RCE in foo", + "database_specific": {"severity": "HIGH"}, + "affected": [{ + "package": {"ecosystem": "PyPI", "name": "foo"}, + "ranges": [{"type": "ECOSYSTEM", + "events": [{"introduced": "0"}, {"fixed": "1.2.0"}]}], + }], + }] + + sbom = build_sbom("je_auto_control") + findings = scan_components(sbom["components"], advisories) + # findings: [{id, package, version, summary, severity, fixed, aliases}] + write_sarif(findings_to_sarif(findings), "vulns.sarif", + tool_name="AutoControl-VulnScan") + +``is_affected(version, osv_range)`` 透過掃描 ``introduced`` / ``fixed`` / +``last_affected`` 事件來評估單一 OSV 範圍;``match_package`` 以資料庫比對單一套件(明確的 +``versions`` **或**範圍);``scan_components`` 跑完整份 SBOM。套件名稱以 PEP-503 正規化後比較, +OSV 嚴重度字詞(``CRITICAL`` / ``HIGH`` / ``MODERATE`` / ``LOW``)對應到 SARIF 的 +``error`` / ``warning`` / ``note`` 等級(預設為 ``warning``)。 + +版本排序採務實的數值比較:release 元件以整數比較,pre-release 後綴(``1.2.0-rc1``)排在最終 +release 之前。Git 範圍與完整 CVSS 向量評分不在範圍內。 + +線上諮詢(選用) +---------------- + +傳入 ``fetcher`` callable 即可在掃描當下抓取諮詢(例如從 ``osv.dev`` API)。測試注入假的 +fetcher,因此比對邏輯不需任何網路即可驗證: + +.. code-block:: python + + findings = scan_components(sbom["components"], None, + fetcher=my_osv_fetcher) + +執行器命令 +---------- + +``AC_scan_vulns`` 接受 ``components``(元件清單、完整 SBOM dict 或 JSON 字串)與 +``advisories``(清單或 JSON 字串),並可選用 ``sarif_path`` 寫出 SARIF 報告;回傳 +``{findings, count}``(寫檔時另含 ``sarif_path``)。同一操作亦以 MCP 工具 ``ac_scan_vulns`` +以及 Script Builder 中 **Security** 分類下的命令提供。 diff --git a/docs/source/Zh/zh_index.rst b/docs/source/Zh/zh_index.rst index 92cd092a..3172b515 100644 --- a/docs/source/Zh/zh_index.rst +++ b/docs/source/Zh/zh_index.rst @@ -80,6 +80,7 @@ AutoControl 所有功能的完整使用指南。 doc/new_features/v55_features_doc doc/new_features/v56_features_doc doc/new_features/v57_features_doc + doc/new_features/v58_features_doc doc/ocr_backends/ocr_backends_doc doc/observability/observability_doc doc/operations_layer/operations_layer_doc diff --git a/je_auto_control/__init__.py b/je_auto_control/__init__.py index c00c65b7..b1474cb4 100644 --- a/je_auto_control/__init__.py +++ b/je_auto_control/__init__.py @@ -302,6 +302,10 @@ from je_auto_control.utils.json_schema import ( SchemaValidationResult, assert_schema, is_valid, validate_json, ) +# OSV vulnerability matching for SBOM components +from je_auto_control.utils.vuln_scan import ( + findings_to_sarif, is_affected, match_package, scan_components, version_key, +) # Background popup/interrupt watchdog (unattended automation) from je_auto_control.utils.watchdog import ( PopupWatchdog, WatchdogRule, default_popup_watchdog, @@ -772,6 +776,8 @@ def start_autocontrol_gui(*args, **kwargs): "from_audit_findings", "from_lint_issues", "make_finding", "to_sarif", "write_sarif", "SchemaValidationResult", "assert_schema", "is_valid", "validate_json", + "findings_to_sarif", "is_affected", "match_package", "scan_components", + "version_key", # MCP server "AuditLogger", "HttpMCPServer", "MCPContent", "MCPPrompt", "MCPPromptArgument", "MCPResource", "MCPServer", "MCPTool", diff --git a/je_auto_control/gui/script_builder/command_schema.py b/je_auto_control/gui/script_builder/command_schema.py index c7b91b42..be5c4a55 100644 --- a/je_auto_control/gui/script_builder/command_schema.py +++ b/je_auto_control/gui/script_builder/command_schema.py @@ -1155,6 +1155,19 @@ def _add_misc_specs(specs: List[CommandSpec]) -> None: ), description="Validate JSON against a JSON Schema; returns {ok, errors}.", )) + specs.append(CommandSpec( + "AC_scan_vulns", "Security", "Scan Dependencies for Vulnerabilities", + fields=( + FieldSpec("components", FieldType.STRING, + placeholder='{"components": [{"name": "foo", ' + '"version": "1.0", "purl": "pkg:pypi/foo@1.0"}]}'), + FieldSpec("advisories", FieldType.STRING, + placeholder='[{"id": "GHSA-...", "affected": [...]}]'), + FieldSpec("sarif_path", FieldType.STRING, optional=True, + placeholder="vulns.sarif"), + ), + description="Match SBOM components against an OSV advisory database.", + )) specs.append(CommandSpec( "AC_run_saga", "Flow", "Run Saga (Compensating Rollback)", fields=( diff --git a/je_auto_control/utils/executor/action_executor.py b/je_auto_control/utils/executor/action_executor.py index 8b513316..fd9e0c47 100644 --- a/je_auto_control/utils/executor/action_executor.py +++ b/je_auto_control/utils/executor/action_executor.py @@ -2863,6 +2863,28 @@ def _scan_secrets(data: Any) -> Dict[str, Any]: return {"findings": scan_secrets(data)} +def _scan_vulns(components: Any, advisories: Any = None, + sarif_path: Optional[str] = None) -> Dict[str, Any]: + """Adapter: match SBOM components against an OSV advisory database.""" + import json + from je_auto_control.utils.vuln_scan import ( + findings_to_sarif, scan_components) + if isinstance(components, str): + components = json.loads(components) + if isinstance(components, dict): + components = components.get("components", []) + if isinstance(advisories, str): + advisories = json.loads(advisories) + findings = scan_components(components, advisories or []) + result: Dict[str, Any] = {"findings": findings, "count": len(findings)} + if sarif_path: + from je_auto_control.utils.sarif import write_sarif + result["sarif_path"] = write_sarif( + findings_to_sarif(findings), sarif_path, + tool_name="AutoControl-VulnScan") + return result + + def _generate_sop(actions: List[Any], title: str = "Automation Procedure", path: Optional[str] = None) -> Dict[str, Any]: """Adapter: build (or write) a step-by-step SOP from an action list.""" @@ -3642,6 +3664,7 @@ def __init__(self): "AC_clip_history_stop": _clip_history_stop, "AC_heal_stats": _heal_stats, "AC_scan_secrets": _scan_secrets, + "AC_scan_vulns": _scan_vulns, "AC_generate_sop": _generate_sop, "AC_tween_drag": _tween_drag, "AC_list_plugins": _list_plugins, diff --git a/je_auto_control/utils/mcp_server/tools/_factories.py b/je_auto_control/utils/mcp_server/tools/_factories.py index ce03e453..0a5e717a 100644 --- a/je_auto_control/utils/mcp_server/tools/_factories.py +++ b/je_auto_control/utils/mcp_server/tools/_factories.py @@ -3244,6 +3244,24 @@ def json_schema_tools() -> List[MCPTool]: ] +def vuln_scan_tools() -> List[MCPTool]: + return [ + MCPTool( + name="ac_scan_vulns", + description=("Match SBOM 'components' (or a full SBOM dict) against " + "an OSV 'advisories' database. Returns {findings:" + "[{id, package, version, severity, fixed, aliases}], " + "count}. Advisories are supplied as data (offline)."), + input_schema=schema( + {"components": {"type": "object"}, + "advisories": {"type": "array"}}, + ["components"]), + handler=h.scan_vulns, + annotations=READ_ONLY, + ), + ] + + def saga_tools() -> List[MCPTool]: return [ MCPTool( @@ -4438,8 +4456,8 @@ def media_assert_tools() -> List[MCPTool]: video_report_tools, fuzzy_tools, artifact_store_tools, image_dedup_tools, locale_tools, voice_tools, coordinate_space_tools, loop_guard_tools, process_mining_tools, asset_tools, events_tools, notify_channel_tools, - jsonpath_tools, json_schema_tools, saga_tools, decision_table_tools, - locator_repair_tools, + jsonpath_tools, json_schema_tools, vuln_scan_tools, saga_tools, + decision_table_tools, locator_repair_tools, pii_text_tools, sarif_tools, screen_record_tools, process_and_shell_tools, remote_desktop_tools, gamepad_tools, diff --git a/je_auto_control/utils/mcp_server/tools/_handlers.py b/je_auto_control/utils/mcp_server/tools/_handlers.py index f73fe9ef..8281f8eb 100644 --- a/je_auto_control/utils/mcp_server/tools/_handlers.py +++ b/je_auto_control/utils/mcp_server/tools/_handlers.py @@ -1561,6 +1561,14 @@ def validate_json(data, schema): return _v(data, schema).to_dict() +def scan_vulns(components, advisories=None): + from je_auto_control.utils.vuln_scan import scan_components + if isinstance(components, dict): + components = components.get("components", []) + findings = scan_components(components, advisories or []) + return {"findings": findings, "count": len(findings)} + + def run_saga(steps): from je_auto_control.utils.saga import run_saga as _run result = _run(steps) diff --git a/je_auto_control/utils/vuln_scan/__init__.py b/je_auto_control/utils/vuln_scan/__init__.py new file mode 100644 index 00000000..693d32b0 --- /dev/null +++ b/je_auto_control/utils/vuln_scan/__init__.py @@ -0,0 +1,9 @@ +"""OSV vulnerability matching for SBOM components (pure standard library).""" +from je_auto_control.utils.vuln_scan.vuln_scan import ( + findings_to_sarif, is_affected, match_package, scan_components, version_key, +) + +__all__ = [ + "findings_to_sarif", "is_affected", "match_package", "scan_components", + "version_key", +] diff --git a/je_auto_control/utils/vuln_scan/vuln_scan.py b/je_auto_control/utils/vuln_scan/vuln_scan.py new file mode 100644 index 00000000..384d5e26 --- /dev/null +++ b/je_auto_control/utils/vuln_scan/vuln_scan.py @@ -0,0 +1,179 @@ +"""Match installed dependencies against OSV vulnerability advisories. + +``utils/sbom`` *inventories* dependencies and ``utils/sarif`` *exports* +findings, but nothing in between ever **produced** a vulnerability finding — +there was no advisory matching at all. This closes that loop: given the SBOM's +``(ecosystem, name, version)`` components and an OSV advisory database, it +reports which packages are affected and bridges the results into the existing +SARIF exporter for GitHub / Azure DevOps code scanning. + +The advisory database is *injected* as plain data (a list of OSV records), so +matching is fully offline and unit-testable; the live ``osv.dev`` query is a +separate, optional ``fetcher`` seam. + +Version ordering is a pragmatic numeric comparison (release components compared +as integers, a pre-release suffix sorting before the final release). Remote/ +git ranges and full CVSS-vector scoring are intentionally out of scope. + +Pure standard library (``re``); imports no ``PySide6``. +""" +import re +from typing import Any, Callable, Dict, List, Mapping, Optional, Sequence, Tuple + +from je_auto_control.utils.sarif import make_finding + +_NUMBERS_RE = re.compile(r"\d+") + +# OSV / GHSA severity word -> SARIF level. +_SEVERITY_LEVELS = { + "critical": "error", "high": "error", "moderate": "warning", + "medium": "warning", "low": "note", +} + +# purl type -> OSV ecosystem name. +_PURL_ECOSYSTEM = { + "pypi": "PyPI", "npm": "npm", "cargo": "crates.io", "golang": "Go", + "maven": "Maven", "gem": "RubyGems", "nuget": "NuGet", + "composer": "Packagist", +} + + +def version_key(version: str) -> Tuple[Tuple[int, ...], int, str]: + """Return a sortable key for a version string (release, pre-rank, pre).""" + text = str(version).strip().lstrip("vV") + parts = re.split(r"[-+]", text, maxsplit=1) + numbers = tuple(int(n) for n in _NUMBERS_RE.findall(parts[0])) + pre = parts[1] if len(parts) > 1 else "" + return (numbers, 0 if pre else 1, pre) + + +def _normalize_name(name: str) -> str: + return re.sub(r"[-_.]+", "-", str(name).strip().lower()) + + +def _ecosystem_from_purl(purl: str) -> str: + match = re.match(r"pkg:([^/]+)/", purl or "") + if not match: + return "" + kind = match.group(1).lower() + return _PURL_ECOSYSTEM.get(kind, match.group(1)) + + +def _sorted_events(events: Sequence[Mapping[str, Any]]) -> List[Tuple[str, str]]: + parsed: List[Tuple[str, str]] = [] + for event in events: + for kind in ("introduced", "fixed", "last_affected"): + if kind in event: + parsed.append((kind, str(event[kind]))) + break + return sorted(parsed, key=lambda item: ( + version_key(item[1]), 0 if item[0] == "introduced" else 1)) + + +def is_affected(version: str, osv_range: Mapping[str, Any]) -> bool: + """Return ``True`` if ``version`` falls inside one OSV range's events.""" + if osv_range.get("type") == "GIT": + return False + target = version_key(version) + affected = False + for kind, bound in _sorted_events(osv_range.get("events", [])): + if kind == "introduced": + affected = bound == "0" or target >= version_key(bound) + elif kind == "fixed" and target >= version_key(bound): + affected = False + elif kind == "last_affected" and target > version_key(bound): + affected = False + return affected + + +def _package_matches(package: Mapping[str, Any], ecosystem: str, name: str) -> bool: + if _normalize_name(package.get("name", "")) != _normalize_name(name): + return False + package_eco = str(package.get("ecosystem", "")) + return not (ecosystem and package_eco and package_eco.lower() != ecosystem.lower()) + + +def _affected_entry_hits(entry: Mapping[str, Any], ecosystem: str, + name: str, version: str) -> bool: + if not _package_matches(entry.get("package", {}), ecosystem, name): + return False + if version in [str(v) for v in entry.get("versions", [])]: + return True + return any(is_affected(version, rng) for rng in entry.get("ranges", [])) + + +def _advisory_hits(advisory: Mapping[str, Any], ecosystem: str, + name: str, version: str) -> bool: + return any(_affected_entry_hits(entry, ecosystem, name, version) + for entry in advisory.get("affected", [])) + + +def _fixed_in_range(osv_range: Mapping[str, Any]) -> Optional[str]: + for event in osv_range.get("events", []): + if "fixed" in event: + return str(event["fixed"]) + return None + + +def _first_fixed(advisory: Mapping[str, Any]) -> Optional[str]: + for entry in advisory.get("affected", []): + for osv_range in entry.get("ranges", []): + fixed = _fixed_in_range(osv_range) + if fixed is not None: + return fixed + return None + + +def _severity_level(advisory: Mapping[str, Any]) -> str: + raw = str(advisory.get("database_specific", {}).get("severity", "")).lower() + return _SEVERITY_LEVELS.get(raw, "warning") + + +def _to_finding(advisory: Mapping[str, Any], name: str, version: str) -> Dict[str, Any]: + return { + "id": str(advisory.get("id", "OSV-UNKNOWN")), + "package": name, + "version": version, + "summary": str(advisory.get("summary", "")), + "severity": _severity_level(advisory), + "fixed": _first_fixed(advisory), + "aliases": list(advisory.get("aliases", [])), + } + + +def match_package(ecosystem: str, name: str, version: str, + advisories: Sequence[Mapping[str, Any]]) -> List[Dict[str, Any]]: + """Return a finding per advisory that affects ``name``@``version``.""" + return [_to_finding(advisory, name, version) for advisory in advisories + if _advisory_hits(advisory, ecosystem, name, version)] + + +def scan_components(components: Sequence[Mapping[str, Any]], + advisories: Optional[Sequence[Mapping[str, Any]]] = None, *, + fetcher: Optional[Callable[[str, str], Sequence]] = None + ) -> List[Dict[str, Any]]: + """Scan SBOM ``components`` against ``advisories`` (and an optional fetcher).""" + base = list(advisories or []) + findings: List[Dict[str, Any]] = [] + for component in components: + name = str(component.get("name", "")) + version = str(component.get("version", "")) + ecosystem = str(component.get("ecosystem", "")) or \ + _ecosystem_from_purl(component.get("purl", "")) + records = base + list(fetcher(ecosystem, name) or []) if fetcher else base + findings.extend(match_package(ecosystem, name, version, records)) + return findings + + +def findings_to_sarif(findings: Sequence[Mapping[str, Any]] + ) -> List[Dict[str, Any]]: + """Convert vulnerability findings into SARIF-ready normalized findings.""" + results = [] + for finding in findings: + summary = finding.get("summary") or finding["id"] + message = f"{finding['package']} {finding['version']}: {summary}" + if finding.get("fixed"): + message += f" (fixed in {finding['fixed']})" + results.append(make_finding(finding["id"], message, + level=finding.get("severity", "warning"))) + return results diff --git a/test/unit_test/headless/test_vuln_scan_batch.py b/test/unit_test/headless/test_vuln_scan_batch.py new file mode 100644 index 00000000..ead924fe --- /dev/null +++ b/test/unit_test/headless/test_vuln_scan_batch.py @@ -0,0 +1,140 @@ +"""Headless tests for the OSV vulnerability matcher. Pure stdlib, no Qt.""" +import json + +import je_auto_control as ac +from je_auto_control.utils.sarif import to_sarif +from je_auto_control.utils.vuln_scan import ( + findings_to_sarif, is_affected, match_package, scan_components, version_key) + +ADVISORIES = [ + { + "id": "GHSA-foo", "summary": "RCE in foo", "aliases": ["CVE-2024-1"], + "database_specific": {"severity": "HIGH"}, + "affected": [{ + "package": {"ecosystem": "PyPI", "name": "foo"}, + "ranges": [{"type": "ECOSYSTEM", + "events": [{"introduced": "0"}, {"fixed": "1.2.0"}]}], + }], + }, + { + "id": "GHSA-bar", "summary": "bug", "database_specific": {"severity": "LOW"}, + "affected": [{"package": {"ecosystem": "PyPI", "name": "bar"}, + "versions": ["2.0.0"]}], + }, +] + + +def test_version_key_orders_releases_and_prereleases(): + assert version_key("2.0.0") > version_key("1.9.9") + assert version_key("1.2.0") > version_key("1.2.0-rc1") + assert version_key("1.10.0") > version_key("1.9.0") + + +def test_is_affected_introduced_fixed_range(): + rng = {"type": "ECOSYSTEM", + "events": [{"introduced": "1.0.0"}, {"fixed": "2.0.0"}]} + assert is_affected("1.5.0", rng) is True + assert is_affected("2.0.0", rng) is False + assert is_affected("0.9.0", rng) is False + + +def test_is_affected_last_affected_and_git_skip(): + rng = {"type": "ECOSYSTEM", + "events": [{"introduced": "1.0.0"}, {"last_affected": "1.5.0"}]} + assert is_affected("1.5.0", rng) is True + assert is_affected("1.6.0", rng) is False + assert is_affected("abcdef", {"type": "GIT", "events": []}) is False + + +def test_match_package_range_and_boundary(): + assert match_package("PyPI", "foo", "1.0.0", ADVISORIES) + assert not match_package("PyPI", "foo", "1.2.0", ADVISORIES) + assert not match_package("PyPI", "foo", "1.3.0", ADVISORIES) + + +def test_match_package_explicit_versions(): + assert match_package("PyPI", "bar", "2.0.0", ADVISORIES) + assert not match_package("PyPI", "bar", "2.1.0", ADVISORIES) + + +def test_match_package_name_normalization_and_ecosystem(): + assert match_package("PyPI", "Foo", "1.0.0", ADVISORIES) + assert not match_package("npm", "foo", "1.0.0", ADVISORIES) + + +def test_finding_shape_severity_and_fixed(): + finding = match_package("PyPI", "foo", "1.0.0", ADVISORIES)[0] + assert finding["id"] == "GHSA-foo" + assert finding["severity"] == "error" + assert finding["fixed"] == "1.2.0" + assert finding["aliases"] == ["CVE-2024-1"] + + +def test_scan_components_from_purl_ecosystem(): + components = [ + {"name": "foo", "version": "1.1.0", "purl": "pkg:pypi/foo@1.1.0"}, + {"name": "safe", "version": "9.9.9", "purl": "pkg:pypi/safe@9.9.9"}, + ] + findings = scan_components(components, ADVISORIES) + assert [f["id"] for f in findings] == ["GHSA-foo"] + + +def test_scan_components_with_injected_fetcher(): + extra = ADVISORIES + components = [{"name": "foo", "version": "1.0.0", + "purl": "pkg:pypi/foo@1.0.0"}] + calls = [] + + def fetcher(ecosystem, name): + calls.append((ecosystem, name)) + return extra + + findings = scan_components(components, None, fetcher=fetcher) + assert calls == [("PyPI", "foo")] + assert findings and findings[0]["id"] == "GHSA-foo" + + +def test_findings_to_sarif_bridge(): + findings = scan_components( + [{"name": "foo", "version": "1.0.0", "purl": "pkg:pypi/foo@1.0.0"}], + ADVISORIES) + document = to_sarif(findings_to_sarif(findings)) + result = document["runs"][0]["results"][0] + assert result["ruleId"] == "GHSA-foo" + assert result["level"] == "error" + assert "fixed in 1.2.0" in result["message"]["text"] + + +def test_severity_defaults_to_warning_when_unknown(): + advisory = [{"id": "X", "affected": [{ + "package": {"ecosystem": "PyPI", "name": "q"}, "versions": ["1.0.0"]}]}] + assert match_package("PyPI", "q", "1.0.0", advisory)[0]["severity"] == "warning" + + +# --- wiring --------------------------------------------------------------- + +def test_executor_round_trip(): + sbom = {"components": [{"name": "foo", "version": "1.0.0", + "purl": "pkg:pypi/foo@1.0.0"}]} + rec = ac.execute_action([[ + "AC_scan_vulns", + {"components": json.dumps(sbom), "advisories": json.dumps(ADVISORIES)}, + ]]) + payload = next(v for v in rec.values() if isinstance(v, dict)) + assert payload["count"] == 1 + assert payload["findings"][0]["id"] == "GHSA-foo" + + +def test_wiring(): + assert "AC_scan_vulns" in ac.executor.known_commands() + from je_auto_control.utils.mcp_server.tools import build_default_tool_registry + assert "ac_scan_vulns" in {t.name for t in build_default_tool_registry()} + from je_auto_control.gui.script_builder.command_schema import _build_specs + assert "AC_scan_vulns" in {s.command for s in _build_specs()} + + +def test_facade_exports(): + for attr in ("scan_components", "match_package", "is_affected", + "version_key", "findings_to_sarif"): + assert hasattr(ac, attr) + assert attr in ac.__all__