cb530c25d9
Bundled commit across unrelated threads: perf collector/meter/logs/export
updates, SSO bootstrap and traefik/nginx/gitea infra, dev backend scripts
(gitea/github/woodpecker), diag test updates, and misc config.
Also fix test pollution: five tests were doing raw sys.modules.pop("conf")
to simulate ImportError on `import conf`, but raw dict mutation isn't
tracked by monkeypatch and leaked the eviction into subsequent tests.
Downstream tests that did `from conf import secret` at module level held
references to the pre-eviction conf, while re-imports inside tests got a
fresh conf, so patches to conf.cfg._data didn't land on the secret()
closure's cfg — causing tests/conf/test_conf.py::TestSecret and
tests/conf/test_connect.py::TestDuckdb::test_custom_db_name to fail.
Fix: use monkeypatch.delitem(sys.modules, "conf", raising=False) so the
eviction is reverted on teardown. Applied to test_diag_ci.py, test_init.py,
test_tracer.py, and test_resource.py (two sites).
389 lines
13 KiB
Python
389 lines
13 KiB
Python
"""Tests for api.diag.vuln — trivy vulnerability issue reporter."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from api.diag.vuln import (
|
|
ScanReport,
|
|
Vuln,
|
|
build_vuln_issue_body,
|
|
build_vuln_issue_title,
|
|
close_vuln_issue,
|
|
file_vuln_issue,
|
|
parse_trivy_json,
|
|
)
|
|
|
|
SAMPLE_TRIVY = {
|
|
"ArtifactName": "git.fhirworx.io/homelab/notebooks:sha-abc12345",
|
|
"Results": [
|
|
{
|
|
"Vulnerabilities": [
|
|
{
|
|
"VulnerabilityID": "CVE-2026-1111",
|
|
"Severity": "CRITICAL",
|
|
"PkgName": "openssl",
|
|
"InstalledVersion": "3.0.1",
|
|
"FixedVersion": "3.0.2",
|
|
"Title": "Buffer overflow in SSL handshake",
|
|
},
|
|
{
|
|
"VulnerabilityID": "CVE-2026-2222",
|
|
"Severity": "HIGH",
|
|
"PkgName": "curl",
|
|
"InstalledVersion": "7.88.0",
|
|
"FixedVersion": "7.88.1",
|
|
"Title": "HSTS bypass via redirect",
|
|
},
|
|
{
|
|
"VulnerabilityID": "CVE-2026-3333",
|
|
"Severity": "MEDIUM",
|
|
"PkgName": "zlib",
|
|
"InstalledVersion": "1.2.13",
|
|
"FixedVersion": "1.2.14",
|
|
"Title": "Denial of service",
|
|
},
|
|
{
|
|
"VulnerabilityID": "CVE-2026-4444",
|
|
"Severity": "HIGH",
|
|
"PkgName": "libxml2",
|
|
"InstalledVersion": "2.10.3",
|
|
"FixedVersion": "",
|
|
"Title": "No fix available",
|
|
},
|
|
]
|
|
}
|
|
],
|
|
}
|
|
|
|
EMPTY_TRIVY = {"ArtifactName": "empty-image", "Results": []}
|
|
|
|
LOW_ONLY_TRIVY = {
|
|
"ArtifactName": "low-image",
|
|
"Results": [
|
|
{
|
|
"Vulnerabilities": [
|
|
{
|
|
"VulnerabilityID": "CVE-2026-9999",
|
|
"Severity": "LOW",
|
|
"PkgName": "bash",
|
|
"InstalledVersion": "5.2",
|
|
"FixedVersion": "5.3",
|
|
"Title": "Minor issue",
|
|
},
|
|
]
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
class TestParseTrivy:
|
|
def test_extracts_image_name(self) -> None:
|
|
report = parse_trivy_json(SAMPLE_TRIVY)
|
|
assert "notebooks" in report.image
|
|
|
|
def test_filters_high_critical_only(self) -> None:
|
|
report = parse_trivy_json(SAMPLE_TRIVY)
|
|
severities = {v.severity for v in report.vulns}
|
|
assert severities <= {"HIGH", "CRITICAL"}
|
|
assert "MEDIUM" not in severities
|
|
|
|
def test_excludes_no_fix(self) -> None:
|
|
report = parse_trivy_json(SAMPLE_TRIVY)
|
|
cves = {v.cve for v in report.vulns}
|
|
assert "CVE-2026-4444" not in cves
|
|
|
|
def test_includes_fixable_vulns(self) -> None:
|
|
report = parse_trivy_json(SAMPLE_TRIVY)
|
|
cves = {v.cve for v in report.vulns}
|
|
assert "CVE-2026-1111" in cves
|
|
assert "CVE-2026-2222" in cves
|
|
|
|
def test_sorted_critical_first(self) -> None:
|
|
report = parse_trivy_json(SAMPLE_TRIVY)
|
|
assert report.vulns[0].severity == "CRITICAL"
|
|
assert report.vulns[1].severity == "HIGH"
|
|
|
|
def test_empty_results(self) -> None:
|
|
report = parse_trivy_json(EMPTY_TRIVY)
|
|
assert report.vulns == []
|
|
|
|
def test_low_only_excluded(self) -> None:
|
|
report = parse_trivy_json(LOW_ONLY_TRIVY)
|
|
assert report.vulns == []
|
|
|
|
def test_handles_lowercase_keys(self) -> None:
|
|
data = {
|
|
"artifactName": "test",
|
|
"results": [
|
|
{
|
|
"vulnerabilities": [
|
|
{
|
|
"vulnerabilityID": "CVE-2026-5555",
|
|
"severity": "critical",
|
|
"pkgName": "pkg",
|
|
"installedVersion": "1.0",
|
|
"fixedVersion": "1.1",
|
|
"title": "Test",
|
|
}
|
|
]
|
|
}
|
|
],
|
|
}
|
|
report = parse_trivy_json(data)
|
|
assert len(report.vulns) == 1
|
|
|
|
|
|
class TestScanReport:
|
|
def test_critical_count(self) -> None:
|
|
report = parse_trivy_json(SAMPLE_TRIVY)
|
|
assert report.critical_count == 1
|
|
|
|
def test_high_count(self) -> None:
|
|
report = parse_trivy_json(SAMPLE_TRIVY)
|
|
assert report.high_count == 1
|
|
|
|
def test_summary_format(self) -> None:
|
|
report = parse_trivy_json(SAMPLE_TRIVY)
|
|
assert "1 CRITICAL" in report.summary
|
|
assert "1 HIGH" in report.summary
|
|
|
|
def test_empty_summary(self) -> None:
|
|
report = parse_trivy_json(EMPTY_TRIVY)
|
|
assert report.summary == "no actionable vulns"
|
|
|
|
|
|
class TestBuildIssue:
|
|
def test_title_contains_image_short_name(self) -> None:
|
|
report = parse_trivy_json(SAMPLE_TRIVY)
|
|
title = build_vuln_issue_title(report)
|
|
assert "notebooks" in title
|
|
assert "vuln:" in title
|
|
|
|
def test_title_contains_counts(self) -> None:
|
|
report = parse_trivy_json(SAMPLE_TRIVY)
|
|
title = build_vuln_issue_title(report)
|
|
assert "CRITICAL" in title
|
|
|
|
def test_body_has_table(self) -> None:
|
|
report = parse_trivy_json(SAMPLE_TRIVY)
|
|
body = build_vuln_issue_body(report, pipeline="42")
|
|
assert "| CVE |" in body
|
|
assert "CVE-2026-1111" in body
|
|
assert "openssl" in body
|
|
|
|
def test_body_contains_pipeline(self) -> None:
|
|
report = parse_trivy_json(SAMPLE_TRIVY)
|
|
body = build_vuln_issue_body(report, pipeline="42")
|
|
assert "#42" in body
|
|
|
|
def test_body_contains_image(self) -> None:
|
|
report = parse_trivy_json(SAMPLE_TRIVY)
|
|
body = build_vuln_issue_body(report)
|
|
assert "notebooks" in body
|
|
|
|
def test_body_truncates_long_title(self) -> None:
|
|
report = ScanReport(
|
|
image="test",
|
|
vulns=[
|
|
Vuln(
|
|
cve="CVE-X",
|
|
severity="HIGH",
|
|
pkg_name="p",
|
|
installed="1",
|
|
fixed="2",
|
|
title="A" * 100,
|
|
)
|
|
],
|
|
)
|
|
body = build_vuln_issue_body(report)
|
|
assert "..." in body
|
|
|
|
|
|
class TestFileVulnIssue:
|
|
def test_skips_if_no_file(self, tmp_path: Path) -> None:
|
|
result = file_vuln_issue(tmp_path / "missing.json")
|
|
assert result is None
|
|
|
|
def test_skips_if_no_vulns(self, tmp_path: Path) -> None:
|
|
scan = tmp_path / "empty.json"
|
|
scan.write_text(json.dumps(EMPTY_TRIVY))
|
|
result = file_vuln_issue(scan)
|
|
assert result is None
|
|
|
|
def test_skips_if_no_token(self, tmp_path: Path) -> None:
|
|
scan = tmp_path / "scan.json"
|
|
scan.write_text(json.dumps(SAMPLE_TRIVY))
|
|
with patch.dict("os.environ", {"GITEA_TOKEN": ""}, clear=False):
|
|
result = file_vuln_issue(scan)
|
|
assert result is None
|
|
|
|
def test_deduplicates(self, tmp_path: Path) -> None:
|
|
scan = tmp_path / "scan.json"
|
|
scan.write_text(json.dumps(SAMPLE_TRIVY))
|
|
mock_client = MagicMock()
|
|
mock_client.get.return_value.json.return_value = [
|
|
{"number": 99, "title": "vuln: notebooks — old issue"}
|
|
]
|
|
with (
|
|
patch.dict("os.environ", {"GITEA_TOKEN": "tok"}, clear=False),
|
|
patch("api.diag.vuln.GiteaClient", return_value=mock_client),
|
|
):
|
|
result = file_vuln_issue(scan)
|
|
assert result is None
|
|
mock_client.create_issue.assert_not_called()
|
|
|
|
def test_creates_issue(self, tmp_path: Path) -> None:
|
|
scan = tmp_path / "scan.json"
|
|
scan.write_text(json.dumps(SAMPLE_TRIVY))
|
|
mock_client = MagicMock()
|
|
mock_client.get.return_value.json.return_value = []
|
|
mock_client.create_issue.return_value = {"number": 200}
|
|
with (
|
|
patch.dict("os.environ", {"GITEA_TOKEN": "tok"}, clear=False),
|
|
patch("api.diag.vuln.GiteaClient", return_value=mock_client),
|
|
):
|
|
result = file_vuln_issue(scan, label_ids=[11])
|
|
assert result == {"number": 200}
|
|
call_args = mock_client.create_issue.call_args[0]
|
|
assert call_args[0] == "homelab"
|
|
assert call_args[1] == "stack"
|
|
issue = call_args[2]
|
|
assert "vuln:" in issue["title"]
|
|
assert issue["labels"] == [11]
|
|
|
|
|
|
class TestCloseVulnIssue:
|
|
def test_skips_if_vulns_remain(self, tmp_path: Path) -> None:
|
|
scan = tmp_path / "scan.json"
|
|
scan.write_text(json.dumps(SAMPLE_TRIVY))
|
|
with patch.dict("os.environ", {"GITEA_TOKEN": "tok"}, clear=False):
|
|
result = close_vuln_issue(scan)
|
|
assert result is False
|
|
|
|
def test_closes_matching_issue(self, tmp_path: Path) -> None:
|
|
scan = tmp_path / "scan.json"
|
|
scan.write_text(json.dumps(EMPTY_TRIVY))
|
|
mock_client = MagicMock()
|
|
mock_client.get.return_value.json.return_value = [
|
|
{"number": 114, "title": "vuln: empty-image — old issue"}
|
|
]
|
|
with (
|
|
patch.dict("os.environ", {"GITEA_TOKEN": "tok"}, clear=False),
|
|
patch("api.diag.vuln.GiteaClient", return_value=mock_client),
|
|
):
|
|
result = close_vuln_issue(scan)
|
|
assert result is True
|
|
mock_client.patch.assert_called_once()
|
|
mock_client.post.assert_called_once()
|
|
|
|
def test_no_issue_to_close(self, tmp_path: Path) -> None:
|
|
scan = tmp_path / "scan.json"
|
|
scan.write_text(json.dumps(EMPTY_TRIVY))
|
|
mock_client = MagicMock()
|
|
mock_client.get.return_value.json.return_value = []
|
|
with (
|
|
patch.dict("os.environ", {"GITEA_TOKEN": "tok"}, clear=False),
|
|
patch("api.diag.vuln.GiteaClient", return_value=mock_client),
|
|
):
|
|
result = close_vuln_issue(scan)
|
|
assert result is False
|
|
|
|
def test_skips_if_no_file(self, tmp_path: Path) -> None:
|
|
result = close_vuln_issue(tmp_path / "missing.json")
|
|
assert result is False
|
|
|
|
def test_skips_if_no_token(self, tmp_path: Path) -> None:
|
|
scan = tmp_path / "scan.json"
|
|
scan.write_text(json.dumps(EMPTY_TRIVY))
|
|
with patch.dict("os.environ", {"GITEA_TOKEN": ""}, clear=False):
|
|
result = close_vuln_issue(scan)
|
|
assert result is False
|
|
|
|
def test_handles_exception_during_search(self, tmp_path: Path) -> None:
|
|
scan = tmp_path / "scan.json"
|
|
scan.write_text(json.dumps(EMPTY_TRIVY))
|
|
mock_client = MagicMock()
|
|
mock_client.get.side_effect = Exception("API error")
|
|
with (
|
|
patch.dict("os.environ", {"GITEA_TOKEN": "tok"}, clear=False),
|
|
patch("api.diag.vuln.GiteaClient", return_value=mock_client),
|
|
):
|
|
result = close_vuln_issue(scan)
|
|
assert result is False
|
|
|
|
|
|
class TestFileVulnIssueExceptions:
|
|
def test_handles_exception_in_dedup_check(self, tmp_path: Path) -> None:
|
|
scan = tmp_path / "scan.json"
|
|
scan.write_text(json.dumps(SAMPLE_TRIVY))
|
|
mock_client = MagicMock()
|
|
mock_client.get.side_effect = Exception("network error")
|
|
mock_client.create_issue.return_value = {"number": 123}
|
|
with (
|
|
patch.dict("os.environ", {"GITEA_TOKEN": "tok"}, clear=False),
|
|
patch("api.diag.vuln.GiteaClient", return_value=mock_client),
|
|
):
|
|
# Should proceed to file issue when dedup check fails
|
|
file_vuln_issue(scan)
|
|
|
|
def test_handles_exception_in_create_issue(self, tmp_path: Path) -> None:
|
|
scan = tmp_path / "scan.json"
|
|
scan.write_text(json.dumps(SAMPLE_TRIVY))
|
|
mock_client = MagicMock()
|
|
mock_client.get.return_value.json.return_value = []
|
|
mock_client.create_issue.side_effect = Exception("create failed")
|
|
with (
|
|
patch.dict("os.environ", {"GITEA_TOKEN": "tok"}, clear=False),
|
|
patch("api.diag.vuln.GiteaClient", return_value=mock_client),
|
|
):
|
|
result = file_vuln_issue(scan)
|
|
assert result is None
|
|
|
|
|
|
class TestVulnMain:
|
|
def test_main_no_args_returns_1(self) -> None:
|
|
from api.diag.vuln import main
|
|
|
|
old_argv = __import__("sys").argv
|
|
__import__("sys").argv = ["vuln"]
|
|
try:
|
|
result = main()
|
|
finally:
|
|
__import__("sys").argv = old_argv
|
|
assert result == 1
|
|
|
|
def test_main_files_issues(self, tmp_path: Path) -> None:
|
|
scan = tmp_path / "scan.json"
|
|
scan.write_text(json.dumps(SAMPLE_TRIVY))
|
|
import sys
|
|
|
|
from api.diag.vuln import main
|
|
|
|
old_argv = sys.argv
|
|
sys.argv = ["vuln", str(scan)]
|
|
try:
|
|
with patch("api.diag.vuln.file_vuln_issue", return_value={"number": 1}):
|
|
result = main()
|
|
finally:
|
|
sys.argv = old_argv
|
|
assert result == 0
|
|
|
|
def test_main_close_mode(self, tmp_path: Path) -> None:
|
|
scan = tmp_path / "scan.json"
|
|
scan.write_text(json.dumps(EMPTY_TRIVY))
|
|
import sys
|
|
|
|
from api.diag.vuln import main
|
|
|
|
old_argv = sys.argv
|
|
sys.argv = ["vuln", "--close", str(scan)]
|
|
try:
|
|
with patch("api.diag.vuln.close_vuln_issue", return_value=True):
|
|
result = main()
|
|
finally:
|
|
sys.argv = old_argv
|
|
assert result == 0
|