mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-16 03:49:26 +00:00
366 lines
15 KiB
Python
366 lines
15 KiB
Python
import json
|
|
import subprocess
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from openultrasast.cli import main
|
|
from openultrasast.findings import StaticFinding
|
|
from openultrasast.reports import scan_exit_code, write_manifest, write_markdown_report, write_sarif_report
|
|
from openultrasast.run import ScanRun
|
|
from openultrasast.stages import Stage, plan_for_mode, record_completed, stages_payload
|
|
from openultrasast.verification import verify_finding
|
|
|
|
|
|
def test_sarif_report_uses_finding_ids_and_evidence_properties(tmp_path: Path) -> None:
|
|
finding = _finding()
|
|
verification = verify_finding(finding)
|
|
output = tmp_path / "report.sarif"
|
|
|
|
write_sarif_report([finding], [verification], output)
|
|
|
|
payload = json.loads(output.read_text())
|
|
result = payload["runs"][0]["results"][0]
|
|
|
|
assert payload["version"] == "2.1.0"
|
|
assert result["partialFingerprints"]["openultrasastFindingId"] == finding.finding_id
|
|
assert result["properties"]["finding_id"] == finding.finding_id
|
|
assert result["properties"]["evidence_level"] == "static_corroboration"
|
|
assert result["properties"]["verification"]["status"] == "accepted"
|
|
|
|
|
|
def test_manifest_links_shared_artifacts_by_finding_id(tmp_path: Path) -> None:
|
|
run = ScanRun(scan_id="scan-1", root=tmp_path / "run", target=tmp_path / "repo")
|
|
run.root.mkdir()
|
|
finding = _finding()
|
|
verification = verify_finding(finding)
|
|
artifacts = {
|
|
"findings": run.root / "findings.json",
|
|
"verification": run.root / "verification.json",
|
|
"markdown": run.root / "report.md",
|
|
"sarif": run.root / "report.sarif",
|
|
}
|
|
output = run.root / "manifest.json"
|
|
|
|
write_manifest(run=run, findings=[finding], verifications=[verification], artifact_paths=artifacts, path=output)
|
|
|
|
payload = json.loads(output.read_text())
|
|
manifest_finding = payload["findings"][0]
|
|
|
|
assert payload["scan_id"] == "scan-1"
|
|
assert manifest_finding["finding_id"] == finding.finding_id
|
|
assert manifest_finding["verification_status"] == "accepted"
|
|
assert manifest_finding["artifact_refs"]["sarif"] == "report.sarif"
|
|
assert "stages" not in payload
|
|
|
|
|
|
def test_manifest_includes_requested_completed_skipped_stages(tmp_path: Path) -> None:
|
|
run = ScanRun(scan_id="scan-1", root=tmp_path / "run", target=tmp_path / "repo")
|
|
run.root.mkdir()
|
|
finding = _finding()
|
|
artifacts = {
|
|
"findings": run.root / "findings.json",
|
|
"verification": run.root / "verification.json",
|
|
"markdown": run.root / "report.md",
|
|
"sarif": run.root / "report.sarif",
|
|
}
|
|
output = run.root / "manifest.json"
|
|
plan = record_completed(plan_for_mode("quick"), Stage.STATIC)
|
|
|
|
write_manifest(
|
|
run=run,
|
|
findings=[finding],
|
|
verifications=[verify_finding(finding)],
|
|
artifact_paths=artifacts,
|
|
path=output,
|
|
stages=stages_payload(plan),
|
|
)
|
|
|
|
payload = json.loads(output.read_text())
|
|
assert payload["stages"] == {"requested": ["static"], "completed": ["static"], "skipped": []}
|
|
assert payload["findings"][0]["finding_id"] == finding.finding_id
|
|
|
|
|
|
def test_markdown_report_includes_verification_status(tmp_path: Path) -> None:
|
|
finding = _finding()
|
|
output = tmp_path / "report.md"
|
|
|
|
write_markdown_report([finding], output, [verify_finding(finding)])
|
|
|
|
text = output.read_text()
|
|
assert "## Inventory" in text
|
|
assert "Verification: `accepted`" in text
|
|
assert finding.finding_id in text
|
|
assert "worth-fixing" not in text.lower()
|
|
assert "worth_fixing" not in text.lower()
|
|
|
|
|
|
def test_markdown_report_labels_hits_as_inventory_not_worth_fixing(tmp_path: Path) -> None:
|
|
finding = _finding()
|
|
output = tmp_path / "report.md"
|
|
|
|
write_markdown_report([finding], output, [verify_finding(finding)])
|
|
|
|
text = output.read_text()
|
|
inventory_at = text.index("## Inventory")
|
|
finding_at = text.index(f"## {finding.title}")
|
|
assert inventory_at < finding_at
|
|
assert "worth-fixing" not in text.lower()
|
|
|
|
|
|
def test_scan_exit_code_policy() -> None:
|
|
finding = _finding()
|
|
verification = verify_finding(finding)
|
|
worth_fixing = SimpleNamespace(worth_fixing=True)
|
|
|
|
assert scan_exit_code([finding], [verification], "never") == 0
|
|
assert scan_exit_code([finding], [verification], "findings") == 1
|
|
assert scan_exit_code([], [], "findings") == 0
|
|
assert scan_exit_code([finding], [verification], "verified") == 1
|
|
assert scan_exit_code([], [], "verified") == 0
|
|
# findings/verified ignore worth-fixing verdicts
|
|
assert scan_exit_code([], [], "findings", worth_fixing_verdicts=[worth_fixing]) == 0
|
|
assert scan_exit_code([], [], "verified", worth_fixing_verdicts=[worth_fixing]) == 0
|
|
assert scan_exit_code([finding], [verification], "findings", worth_fixing_verdicts=[worth_fixing]) == 1
|
|
assert scan_exit_code([finding], [verification], "verified", worth_fixing_verdicts=[worth_fixing]) == 1
|
|
|
|
|
|
def test_scan_exit_code_worth_fixing() -> None:
|
|
finding = _finding()
|
|
verification = verify_finding(finding)
|
|
worth_fixing = SimpleNamespace(worth_fixing=True)
|
|
not_worth_fixing = SimpleNamespace(worth_fixing=False)
|
|
|
|
assert scan_exit_code([finding], [verification], "worth-fixing") == 0
|
|
assert scan_exit_code([finding], [verification], "worth-fixing", worth_fixing_verdicts=[]) == 0
|
|
assert scan_exit_code([finding], [verification], "worth-fixing", worth_fixing_verdicts=[not_worth_fixing]) == 0
|
|
assert scan_exit_code([finding], [verification], "worth-fixing", worth_fixing_verdicts=[worth_fixing]) == 1
|
|
|
|
|
|
def test_scan_exit_code_unknown_policy() -> None:
|
|
with pytest.raises(ValueError, match="unknown fail policy"):
|
|
scan_exit_code([], [], "maybe")
|
|
|
|
|
|
def test_scan_help_lists_worth_fixing(capsys: pytest.CaptureFixture[str]) -> None:
|
|
with pytest.raises(SystemExit) as exc:
|
|
main(["scan", "--help"])
|
|
assert exc.value.code == 0
|
|
help_text = capsys.readouterr().out
|
|
assert "worth-fixing" in help_text
|
|
assert "findings" in help_text
|
|
assert "verified" in help_text
|
|
|
|
|
|
def test_cli_scan_writes_sarif_manifest_and_honors_fail_policy(tmp_path: Path, monkeypatch) -> None: # type: ignore[no-untyped-def]
|
|
repo = tmp_path / "repo"
|
|
repo.mkdir()
|
|
(repo / "app.py").write_text("@app.route('/admin')\ndef admin():\n return eval(request.data)\n")
|
|
monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs")
|
|
|
|
assert main(["scan", str(repo), "--mode", "quick", "--fail-on", "verified"]) == 1
|
|
|
|
run_dir = sorted((repo / ".runs").iterdir())[-1]
|
|
sarif = json.loads((run_dir / "report.sarif").read_text())
|
|
manifest = json.loads((run_dir / "manifest.json").read_text())
|
|
|
|
finding_id = manifest["findings"][0]["finding_id"]
|
|
assert sarif["runs"][0]["results"][0]["properties"]["finding_id"] == finding_id
|
|
assert manifest["findings"][0]["artifact_refs"]["verification_json"] == "verification.json"
|
|
|
|
|
|
def test_cli_scan_fail_on_worth_fixing_exits_zero_without_verdicts(tmp_path: Path, monkeypatch) -> None: # type: ignore[no-untyped-def]
|
|
repo = tmp_path / "repo"
|
|
repo.mkdir()
|
|
(repo / "app.py").write_text("@app.route('/admin')\ndef admin():\n return eval(request.data)\n")
|
|
monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs")
|
|
|
|
assert main(["scan", str(repo), "--mode", "quick", "--fail-on", "worth-fixing"]) == 0
|
|
|
|
|
|
def test_quick_scan_report_is_inventory_without_map_or_docker(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from openultrasast import tool_hunter
|
|
from openultrasast.complexity import map as complexity_map
|
|
from openultrasast.sandbox import probe as sandbox_probe
|
|
from openultrasast.sandbox import runner as sandbox_runner
|
|
|
|
repo = tmp_path / "repo"
|
|
repo.mkdir()
|
|
(repo / "app.py").write_text("@app.route('/admin')\ndef admin():\n return eval(request.data)\n")
|
|
monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs")
|
|
|
|
docker_argv: list[list[str]] = []
|
|
original_run = subprocess.run
|
|
|
|
def guarded_run(command, *args, **kwargs): # type: ignore[no-untyped-def]
|
|
argv = list(command) if isinstance(command, list | tuple) else [command]
|
|
if argv and Path(str(argv[0])).name == "docker":
|
|
docker_argv.append(argv)
|
|
raise AssertionError(f"quick scan must not start docker: {argv}")
|
|
return original_run(command, *args, **kwargs)
|
|
|
|
def fail(message: str): # type: ignore[no-untyped-def]
|
|
def _fail(*args, **kwargs): # type: ignore[no-untyped-def]
|
|
raise AssertionError(message)
|
|
|
|
return _fail
|
|
|
|
monkeypatch.setattr(subprocess, "run", guarded_run)
|
|
monkeypatch.setattr(sandbox_runner, "build_docker_argv", fail("quick scan must not build docker argv"))
|
|
monkeypatch.setattr(sandbox_probe.SandboxProbe, "available", fail("quick scan must not probe docker"))
|
|
monkeypatch.setattr(complexity_map, "build_complexity_map", fail("quick scan must not build a complexity map"))
|
|
monkeypatch.setattr(tool_hunter, "run_tool_hunter", fail("quick scan must not start a hunter model"))
|
|
|
|
assert main(["scan", str(repo), "--mode", "quick"]) == 0
|
|
|
|
run_dir = sorted((repo / ".runs").iterdir())[-1]
|
|
report = (run_dir / "report.md").read_text()
|
|
manifest = json.loads((run_dir / "manifest.json").read_text())
|
|
|
|
assert "## Inventory" in report
|
|
assert "worth-fixing" not in report.lower()
|
|
assert not (run_dir / "complexity_map.json").exists()
|
|
assert docker_argv == []
|
|
assert manifest["stages"]["requested"] == ["static"]
|
|
assert manifest["stages"]["completed"] == ["static"]
|
|
assert manifest["stages"]["skipped"] == []
|
|
assert "complexity" not in manifest
|
|
assert "complexity_map" not in manifest.get("artifacts", {})
|
|
assert not any(entry.get("reason") == "hunter_model_unavailable" for entry in manifest.get("degradations", []))
|
|
|
|
|
|
def test_standard_scan_writes_complexity_map_without_docker(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from openultrasast.sandbox import probe as sandbox_probe
|
|
from openultrasast.sandbox import runner as sandbox_runner
|
|
|
|
repo = tmp_path / "repo"
|
|
repo.mkdir()
|
|
(repo / "app.py").write_text("@app.route('/admin')\ndef admin():\n return eval(request.data)\n")
|
|
monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs")
|
|
|
|
docker_argv: list[list[str]] = []
|
|
original_run = subprocess.run
|
|
|
|
def guarded_run(command, *args, **kwargs): # type: ignore[no-untyped-def]
|
|
argv = list(command) if isinstance(command, list | tuple) else [command]
|
|
if argv and Path(str(argv[0])).name == "docker":
|
|
docker_argv.append(argv)
|
|
raise AssertionError(f"standard scan must not start docker: {argv}")
|
|
return original_run(command, *args, **kwargs)
|
|
|
|
def fail(message: str): # type: ignore[no-untyped-def]
|
|
def _fail(*args, **kwargs): # type: ignore[no-untyped-def]
|
|
raise AssertionError(message)
|
|
|
|
return _fail
|
|
|
|
monkeypatch.setattr(subprocess, "run", guarded_run)
|
|
monkeypatch.setattr(sandbox_runner, "build_docker_argv", fail("standard scan must not build docker argv"))
|
|
monkeypatch.setattr(sandbox_probe.SandboxProbe, "available", fail("standard scan must not probe docker"))
|
|
|
|
assert main(["scan", str(repo), "--mode", "standard"]) == 0
|
|
|
|
run_dir = sorted((repo / ".runs").iterdir())[-1]
|
|
map_path = run_dir / "complexity_map.json"
|
|
manifest = json.loads((run_dir / "manifest.json").read_text())
|
|
complexity_map = json.loads(map_path.read_text())
|
|
|
|
assert map_path.is_file()
|
|
assert complexity_map["heuristic_only"] is True
|
|
assert isinstance(complexity_map["hotspots"], list)
|
|
assert complexity_map["hotspots"]
|
|
assert docker_argv == []
|
|
assert manifest["stages"]["requested"] == ["static", "map"]
|
|
assert manifest["stages"]["completed"] == ["static", "map"]
|
|
assert manifest["stages"]["skipped"] == []
|
|
assert manifest["complexity"]["hotspot_count"] == len(complexity_map["hotspots"])
|
|
assert manifest["complexity"]["heuristic_only"] is True
|
|
assert manifest["artifacts"]["complexity_map"] == "complexity_map.json"
|
|
|
|
|
|
def test_standard_scan_records_hunter_model_unavailable_when_model_unset(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from openultrasast.sandbox import probe as sandbox_probe
|
|
from openultrasast.sandbox import runner as sandbox_runner
|
|
|
|
repo = tmp_path / "repo"
|
|
repo.mkdir()
|
|
(repo / "app.py").write_text("@app.route('/admin')\ndef admin():\n return eval(request.data)\n")
|
|
monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs")
|
|
|
|
def fail(message: str): # type: ignore[no-untyped-def]
|
|
def _fail(*args, **kwargs): # type: ignore[no-untyped-def]
|
|
raise AssertionError(message)
|
|
|
|
return _fail
|
|
|
|
monkeypatch.setattr(sandbox_runner, "build_docker_argv", fail("standard scan must not build docker argv"))
|
|
monkeypatch.setattr(sandbox_probe.SandboxProbe, "available", fail("standard scan must not probe docker"))
|
|
|
|
assert main(["scan", str(repo), "--mode", "standard"]) == 0
|
|
|
|
run_dir = sorted((repo / ".runs").iterdir())[-1]
|
|
manifest = json.loads((run_dir / "manifest.json").read_text())
|
|
degradations = manifest.get("degradations", [])
|
|
|
|
assert any(entry.get("reason") == "hunter_model_unavailable" for entry in degradations)
|
|
hunter_skip = next(entry for entry in degradations if entry.get("reason") == "hunter_model_unavailable")
|
|
assert hunter_skip["stage"] in {"map", "hunter_pool"}
|
|
assert manifest["complexity"]["heuristic_only"] is True
|
|
assert (run_dir / "complexity_map.json").is_file()
|
|
|
|
|
|
def test_manifest_includes_complexity_summary(tmp_path: Path) -> None:
|
|
run = ScanRun(scan_id="scan-1", root=tmp_path / "run", target=tmp_path / "repo")
|
|
run.root.mkdir()
|
|
finding = _finding()
|
|
artifacts = {
|
|
"findings": run.root / "findings.json",
|
|
"verification": run.root / "verification.json",
|
|
"markdown": run.root / "report.md",
|
|
"sarif": run.root / "report.sarif",
|
|
"complexity_map": run.root / "complexity_map.json",
|
|
}
|
|
output = run.root / "manifest.json"
|
|
|
|
write_manifest(
|
|
run=run,
|
|
findings=[finding],
|
|
verifications=[verify_finding(finding)],
|
|
artifact_paths=artifacts,
|
|
path=output,
|
|
complexity={"hotspot_count": 3, "heuristic_only": True},
|
|
)
|
|
|
|
payload = json.loads(output.read_text())
|
|
assert payload["complexity"] == {"hotspot_count": 3, "heuristic_only": True}
|
|
assert payload["artifacts"]["complexity_map"] == "complexity_map.json"
|
|
|
|
|
|
def _finding() -> StaticFinding:
|
|
return StaticFinding(
|
|
finding_id="python-unsafe-eval:app.py:3",
|
|
path="app.py",
|
|
title="Dynamic Python execution needs review",
|
|
severity="high",
|
|
confidence="medium",
|
|
evidence_level="static_corroboration",
|
|
rationale="Static pattern matched.",
|
|
line=3,
|
|
function_name="admin",
|
|
reachability_status="reachable",
|
|
reachability_evidence=[
|
|
{
|
|
"kind": "route",
|
|
"access_level": "public",
|
|
"line": 1,
|
|
"end_line": 3,
|
|
"function_name": "admin",
|
|
"conditions": [],
|
|
}
|
|
],
|
|
reachability_conditions=[],
|
|
tags=["syscall_entry"],
|
|
ranking_priority=3.0,
|
|
)
|