Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 14 additions & 2 deletions src/skillspector/nodes/analyzers/osv_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -299,6 +299,16 @@ def _estimate_cvss_severity(vector: str) -> str | None:
return "LOW"


# GHSA's four severity levels are LOW/MODERATE/HIGH/CRITICAL (GitHub's own
# term is "moderate", not "medium" — see the Advisory Database UI and the
# `severity` enum in https://docs.github.com/en/rest/security-advisories).
# Every downstream table in static_patterns_supply_chain.py (_SEVERITY_ORDER,
# _osv_severity_to_app, _SEVERITY_CONFIDENCE) only recognises the app's own
# LOW/MEDIUM/HIGH/CRITICAL vocabulary, so an un-normalized "MODERATE" silently
# fell through every one of them to the LOW default.
_EXTERNAL_SEVERITY_ALIASES = {"MODERATE": "MEDIUM"}


def _severity_from_vuln(vuln: dict) -> str:
"""Extract the highest severity string from an OSV vulnerability object.

Expand All @@ -311,15 +321,17 @@ def _severity_from_vuln(vuln: dict) -> str:
db_specific = vuln.get("database_specific", {})
ghsa_severity = db_specific.get("severity", "") if isinstance(db_specific, dict) else ""
if isinstance(ghsa_severity, str) and ghsa_severity:
return ghsa_severity[:32].upper()
normalized = ghsa_severity[:32].upper()
return _EXTERNAL_SEVERITY_ALIASES.get(normalized, normalized)
raw_affected = vuln.get("affected", [])
for affected in raw_affected if isinstance(raw_affected, list) else []:
if not isinstance(affected, dict):
continue
eco_specific = affected.get("ecosystem_specific", {})
sev = eco_specific.get("severity", "") if isinstance(eco_specific, dict) else ""
if isinstance(sev, str) and sev:
return sev[:32].upper()
normalized = sev[:32].upper()
return _EXTERNAL_SEVERITY_ALIASES.get(normalized, normalized)
raw_severity = vuln.get("severity", [])
for severity_entry in raw_severity if isinstance(raw_severity, list) else []:
if not isinstance(severity_entry, dict):
Expand Down
51 changes: 51 additions & 0 deletions tests/unit/test_osv_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -120,6 +120,20 @@ def test_database_specific_case_insensitive(self) -> None:
def test_no_severity_defaults_high(self) -> None:
assert _severity_from_vuln({}) == "HIGH"

def test_ghsa_moderate_normalizes_to_medium(self) -> None:
# GHSA's own four levels are LOW/MODERATE/HIGH/CRITICAL, not
# LOW/MEDIUM/HIGH/CRITICAL — confirmed live against api.osv.dev
# (e.g. GHSA-29mw-wpgm-hmr9, a real lodash prototype-pollution
# advisory, reports `"database_specific": {"severity": "MODERATE"}`).
# Every downstream table keys on the app's own vocabulary, so this
# must normalize here rather than leak the external spelling.
vuln = {"database_specific": {"severity": "MODERATE"}}
assert _severity_from_vuln(vuln) == "MEDIUM"

def test_ecosystem_specific_moderate_normalizes_to_medium(self) -> None:
vuln = {"affected": [{"ecosystem_specific": {"severity": "moderate"}}]}
assert _severity_from_vuln(vuln) == "MEDIUM"


class TestQueryBatch:
def test_empty_packages_returns_empty(self) -> None:
Expand Down Expand Up @@ -170,6 +184,43 @@ def test_successful_batch_query(self) -> None:
assert "CVE-2024-22195" in results[0][0].aliases
assert len(results[1]) == 0

def test_batch_query_ghsa_moderate_normalizes_to_medium(self) -> None:
# Real OSV.dev record for lodash 4.17.15 (GHSA-29mw-wpgm-hmr9, a
# prototype-pollution advisory): `database_specific.severity` is the
# literal string "MODERATE", GHSA's own term, not "MEDIUM".
mock_batch_response = {
"results": [{"vulns": [{"id": "GHSA-29mw", "modified": "2024-01-01T00:00:00Z"}]}]
}
mock_detail_response = {
"id": "GHSA-29mw",
"summary": "Prototype pollution in lodash",
"database_specific": {"severity": "MODERATE"},
"aliases": ["CVE-2020-8203"],
}

mock_client = MagicMock()
mock_post_resp = MagicMock()
mock_post_resp.json.return_value = mock_batch_response
mock_post_resp.raise_for_status = MagicMock()

mock_get_resp = MagicMock()
mock_get_resp.json.return_value = mock_detail_response
mock_get_resp.raise_for_status = MagicMock()

mock_client.__enter__ = MagicMock(return_value=mock_client)
mock_client.__exit__ = MagicMock(return_value=False)
mock_client.post.return_value = mock_post_resp
mock_client.get.return_value = mock_get_resp

with patch(
"skillspector.nodes.analyzers.osv_client.httpx.Client", return_value=mock_client
):
results = query_batch([("lodash", "4.17.15")], ECOSYSTEM_NPM)

assert len(results) == 1
assert len(results[0]) == 1
assert results[0][0].severity == "MEDIUM"

def test_network_failure_returns_empty(self) -> None:
mock_client = MagicMock()
mock_client.__enter__ = MagicMock(return_value=mock_client)
Expand Down
Loading