You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tooling.apache.org by ar...@apache.org on 2025/12/17 18:15:46 UTC
(tooling-trusted-releases) 01/01: Saving first day's work
This is an automated email from the ASF dual-hosted git repository.
arm pushed a commit to branch vulnerabilities_sbom
in repository https://gitbox.apache.org/repos/asf/tooling-trusted-releases.git
commit 92b5e8b70b283e8cd506151fa41c2225b0f703e3
Author: Alastair McFarlane <ar...@apache.org>
AuthorDate: Wed Dec 17 18:15:27 2025 +0000
Saving first day's work
---
atr/get/sbom.py | 125 +++++++++++++++++++++++--------------
atr/models/results.py | 5 +-
atr/sbom/cli.py | 29 ++++++---
atr/sbom/models/osv.py | 35 +++++++++--
atr/sbom/osv.py | 164 +++++++++++++++++++++++++++++++++++++++++++------
atr/sbom/utilities.py | 68 +++++++++++++++++++-
atr/tasks/sbom.py | 31 +++++++++-
7 files changed, 377 insertions(+), 80 deletions(-)
diff --git a/atr/get/sbom.py b/atr/get/sbom.py
index fb570a7..8de0bb1 100644
--- a/atr/get/sbom.py
+++ b/atr/get/sbom.py
@@ -20,7 +20,10 @@ from __future__ import annotations
import json
from typing import TYPE_CHECKING, Any
+import aiofiles
import asfquart.base as base
+import cmarkgfm
+import markupsafe
import atr.blueprints.get as get
import atr.db as db
@@ -29,7 +32,9 @@ import atr.htm as htm
import atr.models.results as results
import atr.models.sql as sql
import atr.sbom as sbom
+import atr.sbom.models.osv as osv
import atr.shared as shared
+import atr.storage as storage
import atr.template as template
import atr.web as web
@@ -41,6 +46,7 @@ if TYPE_CHECKING:
async def report(session: web.Committer, project: str, version: str, file_path: str) -> str:
await session.check_access(project)
await session.release(project, version)
+
async with db.session() as data:
via = sql.validate_instrumented_attribute
# TODO: Abstract this code and the sbomtool.MissingAdapter validators
@@ -55,7 +61,7 @@ async def report(session: web.Committer, project: str, version: str, file_path:
.order_by(sql.sqlmodel.desc(via(sql.Task.completed)))
.all()
)
-
+ # Run or running scans for the current revision
osv_tasks = (
await data.task(
project_name=project,
@@ -80,13 +86,14 @@ async def report(session: web.Committer, project: str, version: str, file_path:
raise base.ASFQuartException("Invalid SBOM score result", errorcode=500)
warnings = [sbom.models.conformance.MissingAdapter.validate_python(json.loads(w)) for w in task_result.warnings]
errors = [sbom.models.conformance.MissingAdapter.validate_python(json.loads(e)) for e in task_result.errors]
+ vulnerabilities = [sbom.models.osv.CdxVulnAdapter.validate_python(json.loads(e)) for e in task_result.vulnerabilities]
block.p[
"""This is a report by the ATR SBOM tool, for debugging and
informational purposes. Please use it only as an approximate
guideline to the quality of your SBOM file."""
]
- block.p["This report is for revision ", htm.code[task_result.revision_number], "."]
+ block.p["This report is for revision ", htm.code[task_result.revision_number], "."] # TODO: Mark if a subsequent score has failed
# TODO: Show the status if the task to augment the SBOM is still running
# TODO: Add a field to the SBOM to show that it's been augmented
@@ -111,7 +118,7 @@ async def report(session: web.Committer, project: str, version: str, file_path:
block.p["No NTIA 2021 minimum data field conformance warnings or errors found."]
block.h2["Vulnerability scan"]
- _vulnerability_scan_section(block, project, version, file_path, task_result.revision_number, osv_tasks)
+ _vulnerability_scan_section(block, project, version, file_path, task_result.revision_number, vulnerabilities, osv_tasks)
block.h2["Outdated tool"]
outdated = None
@@ -142,13 +149,13 @@ async def report(session: web.Committer, project: str, version: str, file_path:
return await template.blank("SBOM report", content=block.collect())
-def _extract_vulnerability_severity(vuln: dict[str, Any]) -> str:
+def _extract_vulnerability_severity(vuln: osv.VulnerabilityDetails) -> str:
"""Extract severity information from vulnerability data."""
- db_specific = vuln.get("database_specific", {})
- if "severity" in db_specific:
- return db_specific["severity"]
+ data = vuln.database_specific or {}
+ if "severity" in data:
+ return data["severity"]
- severity_data = vuln.get("severity", [])
+ severity_data = vuln.severity
if severity_data and isinstance(severity_data, list):
first_severity = severity_data[0]
if isinstance(first_severity, dict) and ("type" in first_severity):
@@ -197,8 +204,7 @@ def _missing_tally(items: list[sbom.models.conformance.Missing]) -> list[tuple[s
key=lambda kv: (kv[0], kv[1]),
)
-
-def _vulnerability_component_details(block: htm.Block, component: results.OSVComponent) -> None:
+def _vulnerability_component_details_osv(block: htm.Block, component: results.OSVComponent) -> None:
details_content = []
summary_element = htm.summary[
htm.span(".badge.bg-danger.me-2.font-monospace")[str(len(component.vulnerabilities))],
@@ -207,17 +213,18 @@ def _vulnerability_component_details(block: htm.Block, component: results.OSVCom
details_content.append(summary_element)
for vuln in component.vulnerabilities:
- vuln_id = vuln.get("id", "Unknown")
- vuln_summary = vuln.get("summary", "No summary available")
- vuln_refs = [r for r in vuln.get("references", []) if r.get("type", "") == "WEB"]
+ vuln_id = vuln.id or "Unknown"
+ vuln_summary = vuln.summary
+ vuln_refs = [r for r in vuln.references if r.get("type", "") == "WEB"]
vuln_primary_ref = vuln_refs[0] if (len(vuln_refs) > 0) else {}
- vuln_modified = vuln.get("modified", "Unknown")
+ vuln_modified = vuln.modified or "Unknown"
vuln_severity = _extract_vulnerability_severity(vuln)
vuln_header = [htm.a(href=vuln_primary_ref.get("url", ""), target="_blank")[htm.strong(".me-2")[vuln_id]]]
if vuln_severity != "Unknown":
vuln_header.append(htm.span(".badge.bg-warning.text-dark")[vuln_severity])
+ details = markupsafe.Markup(cmarkgfm.github_flavored_markdown_to_html(vuln.details))
vuln_div = htm.div(".ms-3.mb-3.border-start.border-warning.border-3.ps-3")[
htm.div(".d-flex.align-items-center.mb-2")[*vuln_header],
htm.p(".mb-1")[vuln_summary],
@@ -225,15 +232,14 @@ def _vulnerability_component_details(block: htm.Block, component: results.OSVCom
"Last modified: ",
vuln_modified,
],
- htm.div(".mt-2.text-muted")[vuln.get("details", "No additional details available.")],
+ htm.div(".mt-2.text-muted")[details or "No additional details available."],
]
details_content.append(vuln_div)
block.append(htm.details(".mb-3.rounded")[*details_content])
-
-def _vulnerability_scan_button(block: htm.Block, project: str, version: str, file_path: str) -> None:
- block.p["No vulnerability scan has been performed for this revision."]
+def _vulnerability_scan_button(block: htm.Block) -> None:
+ block.p["No new vulnerability scan has been performed for this revision."]
form.render_block(
block,
@@ -266,34 +272,64 @@ def _vulnerability_scan_find_in_progress_task(
return None
-def _vulnerability_scan_results(block: htm.Block, task: sql.Task) -> None:
- task_result = task.result
- if not isinstance(task_result, results.SBOMOSVScan):
- block.p["Invalid scan result format."]
- return
+def _vulnerability_scan_results(block: htm.Block, vulns: list[osv.CdxVulnerabilityDetail], task: sql.Task | None) -> None:
+ if task is not None:
+ task_result = task.result
+ if not isinstance(task_result, results.SBOMOSVScan):
+ block.p["Invalid scan result format."]
+ return
- components = task_result.components
- ignored = task_result.ignored
- ignored_count = len(ignored)
+ components = task_result.components
+ ignored = task_result.ignored
+ ignored_count = len(ignored)
+
+ if not components:
+ block.p["No vulnerabilities found."]
+ if ignored_count > 0:
+ component_word = "component was" if (ignored_count == 1) else "components were"
+ block.p[f"{ignored_count} {component_word} ignored due to missing PURL or version information:"]
+ block.p[f"{','.join(ignored)}"]
+ return
+
+ block.p[f"Found vulnerabilities in {len(components)} components:"]
+
+ for component in components:
+ _vulnerability_component_details_osv(block, component)
- if not components:
- block.p["No vulnerabilities found."]
if ignored_count > 0:
component_word = "component was" if (ignored_count == 1) else "components were"
block.p[f"{ignored_count} {component_word} ignored due to missing PURL or version information:"]
block.p[f"{','.join(ignored)}"]
- return
-
- block.p[f"Found vulnerabilities in {len(components)} components:"]
-
- for component in components:
- _vulnerability_component_details(block, component)
-
- if ignored_count > 0:
- component_word = "component was" if (ignored_count == 1) else "components were"
- block.p[f"{ignored_count} {component_word} ignored due to missing PURL or version information:"]
- block.p[f"{','.join(ignored)}"]
+ else:
+ if len(vulns) == 0:
+ block.p["No vulnerabilities found."]
+ return
+ components = {a.get("ref") for v in vulns for a in v.affects}
+ block.p[f"Found vulnerabilities in {len(components)} components:"]
+
+ for component in components:
+ _vulnerability_component_details_osv(
+ block,
+ results.OSVComponent(
+ purl=component,
+ vulnerabilities=[_cdx_to_osv(v) for v in vulns if component in [a.get("ref") for a in v.affects]]
+ )
+ )
+def _cdx_to_osv(cdx: osv.CdxVulnerabilityDetail) -> osv.VulnerabilityDetails:
+ severity, score = sbom.utilities.cdx_severity_to_osv(cdx.ratings)
+ return osv.VulnerabilityDetails(
+ id=cdx.id,
+ summary=cdx.description,
+ details=cdx.detail,
+ modified=cdx.updated,
+ published=cdx.published,
+ severity=score,
+ database_specific={
+ "severity": severity
+ },
+ references=[{ "type": "WEB", "url": a.get("url","")} for a in cdx.advisories]
+ )
def _vulnerability_scan_section(
block: htm.Block,
@@ -301,21 +337,20 @@ def _vulnerability_scan_section(
version: str,
file_path: str,
revision_number: str,
+ vulnerabilities: list[osv.CdxVulnerabilityDetail],
osv_tasks: collections.abc.Sequence[sql.Task],
) -> None:
"""Display the vulnerability scan section based on task status."""
completed_task = _vulnerability_scan_find_completed_task(osv_tasks, revision_number)
- if completed_task is not None:
- _vulnerability_scan_results(block, completed_task)
- return
-
in_progress_task = _vulnerability_scan_find_in_progress_task(osv_tasks, revision_number)
if in_progress_task is not None:
_vulnerability_scan_status(block, in_progress_task, project, version, file_path)
else:
- _vulnerability_scan_button(block, project, version, file_path)
+ _vulnerability_scan_button(block)
+
+ _vulnerability_scan_results(block, vulnerabilities, completed_task)
def _vulnerability_scan_status(
@@ -334,4 +369,4 @@ def _vulnerability_scan_status(
htm.code[task.error],
". Additional details are unavailable from ATR.",
]
- _vulnerability_scan_button(block, project, version, file_path)
+ _vulnerability_scan_button(block)
diff --git a/atr/models/results.py b/atr/models/results.py
index 9db1b55..54ad43d 100644
--- a/atr/models/results.py
+++ b/atr/models/results.py
@@ -18,6 +18,7 @@
from typing import Annotated, Any, Literal
import pydantic
+import atr.sbom.models.osv as osv
from . import schema
@@ -48,7 +49,7 @@ class SBOMGenerateCycloneDX(schema.Strict):
class OSVComponent(schema.Strict):
purl: str = schema.description("Package URL")
- vulnerabilities: list[dict[str, Any]] = schema.description("Vulnerabilities found")
+ vulnerabilities: list[osv.VulnerabilityDetails] = schema.description("Vulnerabilities found")
class SBOMOSVScan(schema.Strict):
@@ -57,6 +58,7 @@ class SBOMOSVScan(schema.Strict):
version_name: str = schema.description("Version name")
revision_number: str = schema.description("Revision number")
file_path: str = schema.description("Relative path to the scanned SBOM file")
+ new_file_path: str = schema.description("Relative path to the updated SBOM file")
components: list[OSVComponent] = schema.description("Components with vulnerabilities")
ignored: list[str] = schema.description("Components ignored")
@@ -120,6 +122,7 @@ class SBOMToolScore(schema.Strict):
warnings: list[str] = schema.description("Warnings from the SBOM tool")
errors: list[str] = schema.description("Errors from the SBOM tool")
outdated: str | None = schema.description("Outdated tool from the SBOM tool")
+ vulnerabilities: list[str] | None = schema.description("Vulnerabilities stored in the SBOM")
cli_errors: list[str] | None = schema.description("Errors from the CycloneDX CLI")
diff --git a/atr/sbom/cli.py b/atr/sbom/cli.py
index 4ef3bec..e658423 100644
--- a/atr/sbom/cli.py
+++ b/atr/sbom/cli.py
@@ -27,7 +27,7 @@ from .cyclonedx import validate_cli, validate_py
from .licenses import check
from .maven import plugin_outdated_version
from .sbomqs import total_score
-from .utilities import bundle_to_patch, patch_to_data, path_to_bundle
+from .utilities import bundle_to_ntia_patch, bundle_to_vuln_patch, patch_to_data, path_to_bundle
def command_license(bundle: models.bundle.Bundle) -> None:
@@ -54,7 +54,7 @@ def command_license(bundle: models.bundle.Bundle) -> None:
def command_merge(bundle: models.bundle.Bundle) -> None:
- patch_ops = asyncio.run(bundle_to_patch(bundle))
+ patch_ops = asyncio.run(bundle_to_ntia_patch(bundle))
if patch_ops:
patch_data = patch_to_data(patch_ops)
merged = bundle.doc.patch(yyjson.Document(patch_data))
@@ -91,8 +91,18 @@ def command_outdated(bundle: models.bundle.Bundle) -> None:
print("no outdated tool found")
-def command_patch(bundle: models.bundle.Bundle) -> None:
- patch_ops = asyncio.run(bundle_to_patch(bundle))
+def command_patch_ntia(bundle: models.bundle.Bundle) -> None:
+ patch_ops = asyncio.run(bundle_to_ntia_patch(bundle))
+ if patch_ops:
+ patch_data = patch_to_data(patch_ops)
+ print(yyjson.Document(patch_data).dumps())
+ else:
+ print("no patch needed")
+
+
+def command_patch_vuln(bundle: models.bundle.Bundle) -> None:
+ results, ignored = asyncio.run(osv.scan_bundle(bundle))
+ patch_ops = asyncio.run(bundle_to_vuln_patch(bundle, results))
if patch_ops:
patch_data = patch_to_data(patch_ops)
print(yyjson.Document(patch_data).dumps())
@@ -101,7 +111,7 @@ def command_patch(bundle: models.bundle.Bundle) -> None:
def command_scores(bundle: models.bundle.Bundle) -> None:
- patch_ops = asyncio.run(bundle_to_patch(bundle))
+ patch_ops = asyncio.run(bundle_to_ntia_patch(bundle))
if patch_ops:
patch_data = patch_to_data(patch_ops)
merged = bundle.doc.patch(yyjson.Document(patch_data))
@@ -153,6 +163,9 @@ def command_where(bundle: models.bundle.Bundle) -> None:
def main() -> None: # noqa: C901
+ if len(sys.argv) < 3:
+ print("Usage: python -m atr.sbom <command> <sbom-path>")
+ sys.exit(1)
path = pathlib.Path(sys.argv[2])
bundle = path_to_bundle(path)
match sys.argv[1]:
@@ -166,8 +179,10 @@ def main() -> None: # noqa: C901
command_osv(bundle)
case "outdated":
command_outdated(bundle)
- case "patch":
- command_patch(bundle)
+ case "patch-ntia":
+ command_patch_ntia(bundle)
+ case "patch-vuln":
+ command_patch_vuln(bundle)
case "scores":
command_scores(bundle)
case "validate-cli":
diff --git a/atr/sbom/models/osv.py b/atr/sbom/models/osv.py
index 30cf8aa..23cf1f3 100644
--- a/atr/sbom/models/osv.py
+++ b/atr/sbom/models/osv.py
@@ -17,16 +17,43 @@
from __future__ import annotations
-from typing import Any
+from typing import Any, Optional, ClassVar
+
+import pydantic
from .base import Lax
class QueryResult(Lax):
- vulns: list[dict[str, Any]] | None = None
+ vulns: list[VulnerabilityDetails] | None = None
next_page_token: str | None = None
class ComponentVulnerabilities(Lax):
- purl: str
- vulnerabilities: list[dict[str, Any]]
+ ref: str
+ vulnerabilities: list[VulnerabilityDetails]
+
+class VulnerabilityDetails(Lax):
+ id: str
+ summary: str | None = None
+ details: str | None = None
+ references: list[dict[str, Any]] | None = None
+ severity: list[dict[str, Any]] | None = None
+ published: str | None = None
+ modified: str
+ database_specific: dict[str, Any] = pydantic.Field(default={})
+
+class CdxVulnerabilityDetail(Lax):
+ bom_ref: str | None = pydantic.Field(default=None, alias="bom-ref")
+ id: str | None = None
+ source: dict[str, str] | None = None
+ description: str | None = None
+ detail: str | None = None
+ advisories: list[dict[str,str]] | None = None
+ cwes: list[int] | None = None
+ published: str | None = None
+ updated: str | None = None
+ affects: list[dict[str,str]] | None = None
+ ratings: list[dict[str,str]] | None = None
+
+CdxVulnAdapter = pydantic.TypeAdapter(CdxVulnerabilityDetail)
diff --git a/atr/sbom/osv.py b/atr/sbom/osv.py
index fc84892..ee2d1a9 100644
--- a/atr/sbom/osv.py
+++ b/atr/sbom/osv.py
@@ -21,11 +21,61 @@ import os
from typing import Any
import aiohttp
+import yyjson
from . import models
+from .utilities import get_pointer, osv_severity_to_cdx
_DEBUG: bool = os.environ.get("DEBUG_SBOM_TOOL") == "1"
_OSV_API_BASE: str = "https://api.osv.dev/v1"
+_SOURCE_DATABASE_NAMES = {
+ "ASB": "Android Security Bulletin",
+ "PUB": "Android Security Bulletin",
+ "ALSA": "AlmaLinux Security Advisory",
+ "ALBA": "AlmaLinux Security Advisory",
+ "ALEA": "AlmaLinux Security Advisory",
+ "ALPINE": "Alpine Security Advisory",
+ "BELL": "BellSoft Security Advisory",
+ "BIT": "Bitnami Vulnerability Database",
+ "CGA": "Chainguard Security Notices",
+ "CURL": "Curl CVEs",
+ "CVE": "National Vulnerability Database",
+ "DEBIAN": "Debian Security Tracker",
+ "DSA": "Debian Security Advisory",
+ "DLA": "Debian Security Advisory",
+ "DTSA": "Debian Security Advisory",
+ "ECHO": "Echo Security Advisory",
+ "EEF": "Erlang Ecosystem Foundation CNA Vulnerabilities",
+ "ELA": "Debian Extended LTS Security Advisory",
+ "GHSA": "GitHub Security Advisory",
+ "GO": "Go Vulnerability Database",
+ "GSD": "Global Security Database",
+ "HSEC": "Haskell Security Advisory",
+ "JLSEC": "Julia Security Advisory",
+ "KUBE": "Kubernetes Official CVE Feed",
+ "LBSEC": "LoopBack Advisory Database",
+ "LSN": "Livepatch Security Notices",
+ "MGASA": "Mageia Security Advisory",
+ "MAL": "Malicious Packages Repository",
+ "MINI": "Minimus Security Notices",
+ "OESA": "openEuler Security Advisory",
+ "OSV": "OSV Advisory",
+ "PHSA": "VMWare Photon Security Advisory",
+ "PSF": "Python Software Foundation Vulnerability Database",
+ "PYSEC": "PyPI Vulnerability Database",
+ "RHSA": "Red Hat Security Data",
+ "RHBA": "Red Hat Security Data",
+ "RHEA": "Red Hat Security Data",
+ "RLSA": "Rocky Linux Security Advisory",
+ "RXSA": "Rocky Linux Security Advisory",
+ "RSEC": "RConsortium Advisory Database",
+ "RUSTSEC": "RustSec Advisory Database",
+ "SUSE": "SUSE Security Landing Page",
+ "openSUSE": "SUSE Security Landing Page",
+ "UBUNTU": "Ubuntu CVE Reports",
+ "USN": "Ubuntu Security Notices",
+ "V8": "V8/Chromium Time-Based Policy",
+}
async def scan_bundle(bundle: models.bundle.Bundle) -> tuple[list[models.osv.ComponentVulnerabilities], list[str]]:
@@ -42,10 +92,79 @@ async def scan_bundle(bundle: models.bundle.Bundle) -> tuple[list[models.osv.Com
print(f"[DEBUG] Total components with vulnerabilities: {len(component_vulns_map)}")
await _scan_bundle_populate_vulnerabilities(session, component_vulns_map)
result: list[models.osv.ComponentVulnerabilities] = []
- for purl, vulns in component_vulns_map.items():
- result.append(models.osv.ComponentVulnerabilities(purl=purl, vulnerabilities=vulns))
+ for ref, vulns in component_vulns_map.items():
+ result.append(models.osv.ComponentVulnerabilities(ref=ref, vulnerabilities=vulns))
return result, ignored
+def vulns_from_bundle(bundle: models.bundle.Bundle) -> list[models.osv.CdxVulnerabilityDetail]:
+ vulns = get_pointer(bundle.doc, "/vulnerabilities")
+ if vulns is None:
+ return []
+ print(vulns)
+ return [models.osv.CdxVulnerabilityDetail.model_validate(v) for v in vulns]
+
+async def vuln_patch(
+ session: aiohttp.ClientSession,
+ doc: yyjson.Document,
+ components: list[models.osv.ComponentVulnerabilities],
+) -> models.patch.Patch:
+ patch_ops: models.patch.Patch = []
+ _assemble_vulnerabilities(doc, patch_ops)
+ ix = 0
+ for c in components:
+ for vuln in c.vulnerabilities:
+ _assemble_component_vulnerability(doc, patch_ops, c.ref, vuln, ix)
+ ix += 1
+ return patch_ops
+
+def _assemble_vulnerabilities(doc: yyjson.Document, patch_ops: models.patch.Patch) -> None:
+ if get_pointer(doc, "/vulnerabilities") is not None:
+ patch_ops.append(
+ models.patch.RemoveOp(
+ op="remove",
+ path="/vulnerabilities"
+ )
+ )
+ patch_ops.append(
+ models.patch.AddOp(
+ op="add",
+ path="/vulnerabilities",
+ value=[],
+ )
+ )
+
+def _assemble_component_vulnerability(
+ doc: yyjson.Document,
+ patch_ops: models.patch.Patch,
+ ref: str,
+ vuln: models.osv.VulnerabilityDetails,
+ index: int
+)-> None:
+ vulnerability = {
+ "bom-ref": f"vuln:{ref}/{vuln.id}",
+ "id": vuln.id,
+ "source": _get_source(vuln),
+ "description": vuln.summary,
+ "detail": vuln.details,
+ "advisories": [
+ { "url": r["url"] }
+ for r in vuln.references
+ if (r.get("type", "") == "WEB" and "advisories" in r.get("url", "")) or r.get("type", "") == "ADVISORY"
+ ],
+ "cwes": [int(r.replace("CWE-", "")) for r in vuln.database_specific.get("cwe_ids", [])],
+ "published": vuln.published,
+ "updated": vuln.modified,
+ "affects": [{"ref": ref}],
+ "ratings": osv_severity_to_cdx(vuln.severity, vuln.database_specific.get("severity", "")),
+ }
+ patch_ops.append(
+ models.patch.AddOp(
+ op="add",
+ path=f"/vulnerabilities/{index!s}",
+ value=vulnerability,
+ )
+ )
+
def _component_purl_with_version(component: models.bom.Component) -> str | None:
if component.purl is None:
@@ -90,20 +209,27 @@ async def _fetch_vulnerabilities_for_batch(
async def _fetch_vulnerability_details(
session: aiohttp.ClientSession,
vuln_id: str,
-) -> dict[str, Any]:
+) -> models.osv.VulnerabilityDetails:
if _DEBUG:
print(f"[DEBUG] Fetching details for {vuln_id}")
async with session.get(f"{_OSV_API_BASE}/vulns/{vuln_id}") as response:
response.raise_for_status()
return await response.json()
+def _get_source(vuln: models.osv.VulnerabilityDetails) -> dict[str, str | None]:
+ db = vuln.id.split("-")[0]
+ web_refs = list(filter(lambda v: v.get("type", "") == "WEB", vuln.references)) if vuln.references else []
+ first_ref = web_refs[0] if len(web_refs) > 0 else None
+
+ name = _SOURCE_DATABASE_NAMES.get(db, "Unknown Database")
+ return {"name": name, "url": first_ref.get("url", "")}
async def _paginate_query(
session: aiohttp.ClientSession,
query: dict[str, Any],
page_token: str,
-) -> list[dict[str, Any]]:
- all_vulns: list[dict[str, Any]] = []
+) -> list[models.osv.VulnerabilityDetails]:
+ all_vulns: list[models.osv.VulnerabilityDetails] = []
current_query = query.copy()
current_query["page_token"] = page_token
page = 0
@@ -135,7 +261,7 @@ def _scan_bundle_build_queries(
ignored.append(component.name)
continue
query = {"package": {"purl": purl_with_version}}
- queries.append((purl_with_version, query))
+ queries.append((component.bom_ref, query))
return queries, ignored
@@ -143,31 +269,31 @@ async def _scan_bundle_fetch_vulnerabilities(
session: aiohttp.ClientSession,
queries: list[tuple[str, dict[str, Any]]],
batch_size: int,
-) -> dict[str, list[dict[str, Any]]]:
- component_vulns_map: dict[str, list[dict[str, Any]]] = {}
+) -> dict[str, list[models.osv.VulnerabilityDetails]]:
+ component_vulns_map: dict[str, list[models.osv.VulnerabilityDetails]] = {}
for batch_start in range(0, len(queries), batch_size):
batch_end = min(batch_start + batch_size, len(queries))
batch = queries[batch_start:batch_end]
if _DEBUG:
batch_num = batch_start // batch_size + 1
print(f"[DEBUG] Processing batch {batch_num} ({batch_start + 1}-{batch_end}/{len(queries)})")
- batch_queries = [query for _purl, query in batch]
+ batch_queries = [query for _ref, query in batch]
batch_results = await _fetch_vulnerabilities_for_batch(session, batch_queries)
if _DEBUG and (len(batch_results) != len(batch)):
print(f"[DEBUG] count mismatch (expected {len(batch)}, got {len(batch_results)})")
- for i, (purl, query) in enumerate(batch):
+ for i, (ref, query) in enumerate(batch):
if i >= len(batch_results):
break
query_result = batch_results[i]
if query_result.vulns:
- existing_vulns = component_vulns_map.setdefault(purl, [])
+ existing_vulns = component_vulns_map.setdefault(ref, [])
existing_vulns.extend(query_result.vulns)
if _DEBUG:
- print(f"[DEBUG] {purl}: {len(query_result.vulns)} vulnerabilities")
+ print(f"[DEBUG] {ref}: {len(query_result.vulns)} vulnerabilities")
if query_result.next_page_token:
if _DEBUG:
- print(f"[DEBUG] {purl}: has pagination, fetching remaining pages")
- existing_vulns = component_vulns_map.setdefault(purl, [])
+ print(f"[DEBUG] {ref}: has pagination, fetching remaining pages")
+ existing_vulns = component_vulns_map.setdefault(ref, [])
paginated = await _paginate_query(session, query, query_result.next_page_token)
existing_vulns.extend(paginated)
return component_vulns_map
@@ -175,19 +301,19 @@ async def _scan_bundle_fetch_vulnerabilities(
async def _scan_bundle_populate_vulnerabilities(
session: aiohttp.ClientSession,
- component_vulns_map: dict[str, list[dict[str, Any]]],
+ component_vulns_map: dict[str, list[models.osv.VulnerabilityDetails]],
) -> None:
- details_cache: dict[str, dict[str, Any]] = {}
+ details_cache: dict[str, models.osv.VulnerabilityDetails] = {}
for vulns in component_vulns_map.values():
for vuln in vulns:
- vuln_id = vuln.get("id")
+ vuln_id = vuln.id
if not vuln_id:
continue
details = details_cache.get(vuln_id)
if details is None:
details = await _fetch_vulnerability_details(session, vuln_id)
details_cache[vuln_id] = details
- vuln.clear()
- vuln.update(details)
+ vuln.__dict__.clear()
+ vuln.__dict__.update(details)
if _DEBUG:
print(f"[DEBUG] Fetched details for {len(details_cache)} unique vulnerabilities")
diff --git a/atr/sbom/utilities.py b/atr/sbom/utilities.py
index 496deb2..6d98695 100644
--- a/atr/sbom/utilities.py
+++ b/atr/sbom/utilities.py
@@ -19,16 +19,39 @@ from __future__ import annotations
from typing import TYPE_CHECKING, Any
+
if TYPE_CHECKING:
import pathlib
+ import atr.sbom.models.osv as osv
+
import aiohttp
import yyjson
from . import models
+_SCORING_METHODS_OSV = {
+ "CVSS_V2": "CVSSv2",
+ "CVSS_V3": "CVSSv3",
+ "CVSS_V4": "CVSSv4"
+}
+_SCORING_METHODS_CDX = {
+ "CVSSv2": "CVSS_V2",
+ "CVSSv3": "CVSS_V3",
+ "CVSSv4": "CVSS_V4",
+ "other": "Other"
+}
+_CDX_SEVERITIES = [
+ "critical",
+ "high",
+ "medium",
+ "low",
+ "info",
+ "none",
+ "unknown"
+]
-async def bundle_to_patch(bundle_value: models.bundle.Bundle) -> models.patch.Patch:
+async def bundle_to_ntia_patch(bundle_value: models.bundle.Bundle) -> models.patch.Patch:
from .conformance import ntia_2021_issues, ntia_2021_patch
_warnings, errors = ntia_2021_issues(bundle_value.bom)
@@ -36,6 +59,18 @@ async def bundle_to_patch(bundle_value: models.bundle.Bundle) -> models.patch.Pa
patch_ops = await ntia_2021_patch(session, bundle_value.doc, errors)
return patch_ops
+async def bundle_to_vuln_patch(
+ bundle_value: models.bundle.Bundle,
+ vulnerabilities: list[osv.ComponentVulnerabilities]
+) -> models.patch.Patch:
+ from .osv import vuln_patch
+
+ # TODO: May not need session (copied from ntia patch)
+ async with aiohttp.ClientSession() as session:
+ patch_ops = await vuln_patch(session, bundle_value.doc, vulnerabilities)
+ return patch_ops
+
+
def get_pointer(doc: yyjson.Document, path: str) -> Any | None:
try:
@@ -55,3 +90,34 @@ def path_to_bundle(path: pathlib.Path) -> models.bundle.Bundle:
text = path.read_text(encoding="utf-8")
bom = models.bom.Bom.model_validate_json(text)
return models.bundle.Bundle(doc=yyjson.Document(text), bom=bom, path=path, text=text)
+
+def osv_severity_to_cdx(severity: list[dict[str, Any]] | None, textual: str) -> list[dict[str, str]] | None:
+ if severity is not None:
+ return [
+ {
+ "severity": textual.lower() if textual.lower() in _CDX_SEVERITIES else "unknown", # TODO: constrain to valid enum
+ "method": _SCORING_METHODS_OSV.get(s.get("type"), "other"),
+ **_extract_cdx_score(s.get("score"))
+ }
+ for s in severity
+ ]
+ return None
+
+def cdx_severity_to_osv(severity: list[dict[str, str]]) -> tuple[str | None, list[dict[str, str]]]:
+ severities = [
+ {
+ "score": str(s.get("score", s.get("vector", ""))),
+ "type": _SCORING_METHODS_CDX.get(s.get("method")),
+ } for s in severity
+ ]
+ textual = severity[0].get("severity")
+ return textual, severities
+
+def _extract_cdx_score(score: str):
+ if "CVSS" in score:
+ return { "vector": score }
+ else:
+ try:
+ return { "score": float(score) }
+ except ValueError:
+ return { "vector": score }
diff --git a/atr/tasks/sbom.py b/atr/tasks/sbom.py
index e837a87..ad6c9db 100644
--- a/atr/tasks/sbom.py
+++ b/atr/tasks/sbom.py
@@ -85,7 +85,7 @@ async def augment(args: FileArgs) -> results.Results | None:
raise SBOMScoringError("SBOM file does not exist", {"file_path": args.file_path})
# Read from the old revision
bundle = sbom.utilities.path_to_bundle(pathlib.Path(full_path))
- patch_ops = await sbom.utilities.bundle_to_patch(bundle)
+ patch_ops = await sbom.utilities.bundle_to_ntia_patch(bundle)
new_full_path: str | None = None
if patch_ops:
patch_data = sbom.utilities.patch_to_data(patch_ops)
@@ -140,13 +140,36 @@ async def osv_scan(args: FileArgs) -> results.Results | None:
raise SBOMScanningError("SBOM file does not exist", {"file_path": args.file_path})
bundle = sbom.utilities.path_to_bundle(pathlib.Path(full_path))
vulnerabilities, ignored = await sbom.osv.scan_bundle(bundle)
- components = [results.OSVComponent(purl=v.purl, vulnerabilities=v.vulnerabilities) for v in vulnerabilities]
+ patch_ops = await sbom.utilities.bundle_to_vuln_patch(bundle, vulnerabilities)
+ components = [results.OSVComponent(purl=v.ref, vulnerabilities=v.vulnerabilities) for v in vulnerabilities]
+
+ new_full_path: str | None = None
+ if patch_ops:
+ patch_data = sbom.utilities.patch_to_data(patch_ops)
+ merged = bundle.doc.patch(yyjson.Document(patch_data))
+ description = "SBOM vulnerability scan through web interface"
+ async with storage.write(args.asf_uid) as write:
+ wacp = await write.as_project_committee_participant(args.project_name)
+ async with wacp.revision.create_and_manage(
+ args.project_name, args.version_name, args.asf_uid or "unknown", description=description
+ ) as creating:
+ new_full_path = os.path.join(str(creating.interim_path), args.file_path)
+ # Write to the new revision
+ log.info(f"Writing updated SBOM to {new_full_path}")
+ await aiofiles.os.remove(new_full_path)
+ async with aiofiles.open(new_full_path, "w", encoding="utf-8") as f:
+ await f.write(merged.dumps())
+
+ if creating.new is None:
+ raise RuntimeError("Internal error: New revision not found")
+
return results.SBOMOSVScan(
kind="sbom_osv_scan",
project_name=args.project_name,
version_name=args.version_name,
revision_number=args.revision_number,
- file_path=args.file_path,
+ file_path=full_path,
+ new_file_path=new_full_path,
components=components,
ignored=ignored,
)
@@ -198,6 +221,7 @@ async def score_tool(args: FileArgs) -> results.Results | None:
bundle = sbom.utilities.path_to_bundle(pathlib.Path(full_path))
warnings, errors = sbom.conformance.ntia_2021_issues(bundle.bom)
outdated = sbom.maven.plugin_outdated_version(bundle.bom)
+ vulnerabilities = sbom.osv.vulns_from_bundle(bundle)
cli_errors = sbom.cyclonedx.validate_cli(bundle)
return results.SBOMToolScore(
kind="sbom_tool_score",
@@ -208,6 +232,7 @@ async def score_tool(args: FileArgs) -> results.Results | None:
warnings=[w.model_dump_json() for w in warnings],
errors=[e.model_dump_json() for e in errors],
outdated=outdated.model_dump_json() if outdated else None,
+ vulnerabilities=[v.model_dump_json() for v in vulnerabilities],
cli_errors=cli_errors,
)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@tooling.apache.org
For additional commands, e-mail: commits-help@tooling.apache.org