# Copyright (C) 2026 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information
from __future__ import annotations
from collections import defaultdict
import copy
import logging
import string
from typing import TYPE_CHECKING, Any, Dict, List, Optional
import requests
from swh.core.config import load_from_envvar
from swh.model.hashutil import hash_to_bytes
from swh.model.model import ReleaseTargetType, Snapshot
from swh.model.swhids import CoreSWHID, ObjectType
from swh.storage import get_storage
from swh.storage.algos.snapshot import snapshot_get_all_branches, snapshot_get_latest
from swh.storage.interface import StorageInterface
from swh.vulns.osv.model import OSVVulnerabilityEvent
from swh.vulns.storage import get_vulnerabilities_storage
from .utils import find_release_from_version, utc_now
if TYPE_CHECKING:
from swh.vulns.storage import VulnerabilitiesStorageInterface
logger = logging.getLogger(__name__)
[docs]
def is_sha1_hex_str(s):
return set(s).issubset(string.hexdigits) and len(s) == 40
[docs]
class OSVReportParser:
"""Class dedicated to the parsing of and OSV report so that vulnerability
events are extracted then persisted to a storage.
Args:
storage: instance of SWH archive main storage
vulnerabilities_storage: instance of OSV vulnerabilities storage
"""
OSV_REPORT_BASE_URL = "https://api.osv.dev/v1/vulns"
def __init__(
self,
storage: StorageInterface,
vulnerabilities_storage: VulnerabilitiesStorageInterface,
**kwargs,
):
self.storage = storage
self.vulnerabilities_storage = vulnerabilities_storage
self.session = requests.Session()
[docs]
@classmethod
def from_config(
cls,
storage: Dict[str, Any],
vulnerabilities_storage: Dict[str, Any],
**kwargs,
) -> OSVReportParser:
"""Instantiate an OSV report parser from configuration.
Args:
storage: Configuration for SWH main storage
vulnerabilities_storage: Configuration for OSV vulnerabilities storage
Returns:
an instance of the OSV report parser
"""
return cls(
storage=get_storage(**storage),
vulnerabilities_storage=get_vulnerabilities_storage(
**vulnerabilities_storage
),
**kwargs,
)
[docs]
@classmethod
def from_configfile(cls, **kwargs: Any) -> OSVReportParser:
"""Instantiate a parser from the configuration loaded from the
SWH_CONFIG_FILENAME envvar, with potential extra keyword arguments if their
value is not None.
Args:
kwargs: kwargs passed to the parser instantiation
Returns:
an instance of the OSV report parser
"""
config = dict(load_from_envvar())
config.update({k: v for k, v in kwargs.items() if v is not None})
return cls.from_config(**config)
[docs]
def process_osv_report(
self, osv_vulnerability_id: str, store_vulnerability_events: bool = True
) -> List[OSVVulnerabilityEvent]:
"""Process an OSV report by parsing it and persisting vulnerability events
in storage.
Args:
vulnerability_id: Identifier of vulnerability in OSV database
store_vulnerability_events: Whether to persist or not extracted vulnerability
events in storage
Returns:
list of extracted vulnerability events
"""
osv_report_url = f"{self.OSV_REPORT_BASE_URL}/{osv_vulnerability_id}"
response = self.session.get(osv_report_url)
response.raise_for_status()
osv_report = response.json()
vuln_events = parse_osv_report(osv_report, self.storage)
if store_vulnerability_events:
self.vulnerabilities_storage.osv_vulnerability_event_add(vuln_events)
self.vulnerabilities_storage.osv_vulnerability_report_add(
osv_vulnerability_id, utc_now(), osv_report
)
return vuln_events
[docs]
def parse_osv_report(
osv_report: Dict[str, Any], storage: StorageInterface
) -> List[OSVVulnerabilityEvent]:
"""Parse an OSV report: extract vulnerability events and find related SWHIDs
into the SWH archive.
Args:
osv_report: OSV report parsed from JSON
storage: instance of SWH main storage
Returns:
list of extracted vulnerability events
"""
if "affected" not in osv_report:
return []
vuln_id = osv_report["id"]
vuln_type = osv_report.get("severity", [{}])[0].get("type")
vulnerability_events: List[OSVVulnerabilityEvent] = []
snapshots = {}
snapshot_branches: Dict[str, Dict[bytes, Optional[Snapshot]]] = defaultdict(dict)
for aff in osv_report["affected"]:
if "ranges" not in aff:
continue
for range_ in aff["ranges"]:
if not range_["type"] == "GIT":
logger.warning(
"[%s] Skipping range of unknown type: %s",
vuln_id,
range_["type"],
)
continue
origin_url = range_["repo"]
origin_info = storage.origin_get([origin_url])
if not origin_info:
logger.warning(
"[%s] Could not find origin %s in the archive",
vuln_id,
origin_url,
)
continue
if origin_url not in snapshots:
snapshots[origin_url] = snapshot_get_latest(
storage, origin_url, branches_count=1, visit_type="git"
)
snapshot = snapshots[origin_url]
if snapshot is not None:
database_specific = range_.get("database_specific", {})
version_events = copy.deepcopy(
database_specific.get("versions", [])
if "versions" in database_specific
else database_specific.get("extracted_events", [])
)
for branch_prefix in (
b"refs/tags/",
b"refs/heads/",
b"refs/remotes/tags/",
):
release_swhid = None
if (
origin_url not in snapshot_branches
or branch_prefix not in snapshot_branches[origin_url]
):
snapshot_branches[origin_url][branch_prefix] = (
snapshot_get_all_branches(
storage,
snapshot.id,
branch_name_include_substring=branch_prefix,
)
)
snapshot_content = snapshot_branches[origin_url][branch_prefix]
assert snapshot_content
for version_event in version_events:
for k in version_event:
if version_event[k] == "0":
vulnerability_events.append(
OSVVulnerabilityEvent(
vulnerability_id=vuln_id,
vulnerability_severity=vuln_type,
event_type=k,
origin_url=origin_url,
)
)
continue
elif isinstance(version_event[k], CoreSWHID):
continue
version = (
version_event[k].lstrip("=").strip().replace("\\", "")
)
release_swhid = find_release_from_version(
version,
snapshot_content,
branch_prefix=branch_prefix,
)
if release_swhid:
vulnerability_events.append(
OSVVulnerabilityEvent(
vulnerability_id=vuln_id,
vulnerability_severity=vuln_type,
event_type=k,
origin_url=origin_url,
swhid=release_swhid,
version=version,
)
)
if release_swhid.object_type == ObjectType.RELEASE:
release = storage.release_get(
[release_swhid.object_id]
)
if (
release
and release[0]
and release[0].target
and release[0].target_type
== ReleaseTargetType.REVISION
):
vulnerability_events.append(
OSVVulnerabilityEvent(
vulnerability_id=vuln_id,
vulnerability_severity=vuln_type,
event_type=k,
origin_url=origin_url,
swhid=CoreSWHID(
object_type=ObjectType.REVISION,
object_id=release[0].target,
),
version=version,
)
)
version_event[k] = release_swhid
break
if all(
isinstance(v, CoreSWHID) or v == "0"
for version_event in version_events
for v in version_event.values()
):
break
for version_event in version_events:
for k, v in version_event.items():
if not isinstance(v, CoreSWHID) and v != "0":
logger.warning(
"[%s] Could not find release or revision SWHID for release named %s in origin %s",
vuln_id,
v,
origin_url,
)
vuln_events = range_["events"]
for vuln_event in vuln_events:
for k in vuln_event:
if is_sha1_hex_str(vuln_event[k]):
swhid = CoreSWHID(
object_type=ObjectType.REVISION,
object_id=hash_to_bytes(vuln_event[k]),
)
if not any(
vuln.swhid == swhid for vuln in vulnerability_events
):
vulnerability_events.append(
OSVVulnerabilityEvent(
vulnerability_id=vuln_id,
vulnerability_severity=vuln_type,
event_type=k,
origin_url=origin_url,
swhid=swhid,
)
)
return vulnerability_events
