82 lines
2.4 KiB
Python
82 lines
2.4 KiB
Python
|
"""Network retries must never turn findings or incomplete audits into success."""
|
||
|
|
|
||
|
|
import importlib.util
|
||
|
|
import json
|
||
|
|
from pathlib import Path
|
||
|
|
from types import SimpleNamespace
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def audit(monkeypatch, tmp_path):
|
||
|
|
spec = importlib.util.spec_from_file_location(
|
||
|
|
"ci_audit", Path(__file__).parents[1] / "scripts/ci-audit.py"
|
||
|
|
)
|
||
|
|
module = importlib.util.module_from_spec(spec)
|
||
|
|
spec.loader.exec_module(module)
|
||
|
|
monkeypatch.setattr(module.tempfile, "gettempdir", lambda: str(tmp_path))
|
||
|
|
monkeypatch.setattr(module.time, "sleep", lambda _: None)
|
||
|
|
monkeypatch.delenv("CI_AUDIT_TIMEOUT", raising=False)
|
||
|
|
monkeypatch.delenv("CI_AUDIT_ATTEMPTS", raising=False)
|
||
|
|
return module
|
||
|
|
|
||
|
|
|
||
|
|
def result(code=0, stderr="", vulns=None, dependencies=None):
|
||
|
|
report = {
|
||
|
|
"dependencies": dependencies
|
||
|
|
if dependencies is not None
|
||
|
|
else [{"name": "example", "vulns": vulns or []}]
|
||
|
|
}
|
||
|
|
return SimpleNamespace(returncode=code, stderr=stderr, stdout=json.dumps(report))
|
||
|
|
|
||
|
|
|
||
|
|
def test_transient_retry_then_complete(audit, monkeypatch):
|
||
|
|
replies = iter([result(1, "ReadTimeout"), result()])
|
||
|
|
calls = []
|
||
|
|
|
||
|
|
def run(command, **kwargs):
|
||
|
|
calls.append(command)
|
||
|
|
return next(replies)
|
||
|
|
|
||
|
|
monkeypatch.setattr(audit.subprocess, "run", run)
|
||
|
|
audit.main()
|
||
|
|
assert len(calls) == 2
|
||
|
|
assert calls[0] == calls[1]
|
||
|
|
assert "--cache-dir" in calls[0] and "--disable-pip" in calls[0]
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize(
|
||
|
|
"reply",
|
||
|
|
[
|
||
|
|
result(1, vulns=[{"id": "CVE-test"}]),
|
||
|
|
result(dependencies=[]),
|
||
|
|
result(dependencies=[{"name": "example", "skip_reason": "unavailable"}]),
|
||
|
|
result(2),
|
||
|
|
],
|
||
|
|
)
|
||
|
|
def test_no_retry_for_findings_or_incomplete(audit, monkeypatch, reply):
|
||
|
|
calls = []
|
||
|
|
monkeypatch.setattr(
|
||
|
|
audit.subprocess, "run", lambda *a, **k: calls.append(a) or reply
|
||
|
|
)
|
||
|
|
with pytest.raises(SystemExit):
|
||
|
|
audit.main()
|
||
|
|
assert len(calls) == 1
|
||
|
|
|
||
|
|
|
||
|
|
def test_network_exhaustion_blocks_and_redacts(audit, monkeypatch, capsys):
|
||
|
|
calls = []
|
||
|
|
monkeypatch.setattr(
|
||
|
|
audit.subprocess,
|
||
|
|
"run",
|
||
|
|
lambda *a, **k: (
|
||
|
|
calls.append(a)
|
||
|
|
or result(1, "ConnectionError https://user:secret@example.invalid")
|
||
|
|
),
|
||
|
|
)
|
||
|
|
with pytest.raises(SystemExit):
|
||
|
|
audit.main()
|
||
|
|
assert len(calls) == 3
|
||
|
|
assert "secret" not in capsys.readouterr().out
|