marius/Open_Ultra_SAST
1
0
Fork 0
mirror of https://github.com/norandom/OpenUltraSAST.git synced 2026-10-16 03:49:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore(p0): retire regress, the skill router, the module audit tool and vocabulary.sc

Browse Source 
Off every active line (PATHFINDER-2026-10-13/03-unified-proposal.md, cuts).
Deleted: src/openultrasast/regress/ and the deep-mode REGRESS branch (deep
now records a regress_retired degradation and equals standard; proof is the
search-with-proof arm's job), the sandbox docker runner and probe that only
regress used, --fail-on worth-fixing, RegressConfig, the two ledger helpers
only regress called; skills.py and index.py with the hunter pool's skill
routing; model/audit.py; cpg/queries/vocabulary.sc with
benchmarks/push/ontology_gap.py. Docs and the scan skill describe the
retirement. Net: 3,895 lines deleted, 86 added. Targeted tests: 180 passed,
28 skipped; 3,331 tests collect.

Co-Authored-By: Claude Fable 5.1 <[email protected]>
This commit is contained in:
Marius Ciepluch
2026-10-13 10:50:01 +02:00
co-authored by Claude Fable 5.1
parent 94e96c6885
commit f5a904b7f2
50 changed files with 86 additions and 3895 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_agent_skills.py
+1 -1
View File
@@ -41,7 +41,7 @@ def test_scan_skill_states_the_evidence_ladder_as_the_code_defines_it() -> None:
"--mode quick|standard|deep",
"--fail-on",
"cpg_unavailable",
"sandbox_unavailable",
"regress_retired",
"report.sarif",
"harness.json",
"ousast mcp",
tests/test_complexity_ledger.py
+1 -68
View File
@@ -1,6 +1,5 @@
import json
from pathlib import Path
from types import SimpleNamespace
from openultrasast.complexity.ledger import (
NOT_TRIGGERABLE_DELTA,
@@ -9,15 +8,11 @@ from openultrasast.complexity.ledger import (
apply_overlay,
hotspot_key,
load_ledger,
must_keep_as_candidate,
persist_verdicts,
record_verdict,
select_forced_candidates,
write_ledger,
)
from openultrasast.complexity.map import Hotspot, build_complexity_map
from openultrasast.findings import StaticFinding
from openultrasast.policy import CwePolicy
from openultrasast.preprocess import FileTarget
@@ -129,7 +124,7 @@ def test_not_triggerable_overlay_drops_score_and_keeps_inventory_hit() -> None:
assert demoted.function_name == hotspot.function_name
def test_not_triggerable_overlay_drops_map_score_and_sev5_reachable_stays_candidate(tmp_path: Path) -> None:
def test_not_triggerable_overlay_drops_map_score_and_keeps_inventory(tmp_path: Path) -> None:
auth = _target(
"src/auth.py",
loc=80,
@@ -168,49 +163,6 @@ def test_not_triggerable_overlay_drops_map_score_and_sev5_reachable_stays_candid
assert sev5.severity == "critical"
assert sev5.rationale == ("Static pattern cmd (CWE-78) matched os.system on line 12; manual verification still required.")
policy = {
sev5.finding_id: 5,
noise.finding_id: 2,
}
forced = select_forced_candidates([sev5, noise], policy)
assert forced == (sev5,)
assert must_keep_as_candidate(sev5, 5, "reachable") is True
assert must_keep_as_candidate(noise, 2, "unknown") is False
def test_must_keep_as_candidate_requires_sev5_and_reachable_surface() -> None:
reachable = _hit("a.py", "a:1", reachability="reachable")
inferred = _hit("b.py", "b:1", reachability="inferred-file-surface")
unknown = _hit("c.py", "c:1", reachability="unknown")
assert must_keep_as_candidate(reachable, 5, "reachable") is True
assert must_keep_as_candidate(inferred, 5, "inferred-file-surface") is True
assert must_keep_as_candidate(unknown, 5, "unknown") is False
assert must_keep_as_candidate(reachable, 4, "reachable") is False
assert must_keep_as_candidate(reachable, 5, None) is True
def test_select_forced_candidates_resolves_policy_severity() -> None:
sev5 = _hit("src/auth.py", "cmd:src/auth.py:12", reachability="reachable")
inferred = _hit("src/upload.py", "sql:src/upload.py:4", reachability="inferred-file-surface")
mild = _hit("src/helper.py", "print:src/helper.py:1", reachability="reachable")
policy = {
"CWE-78": CwePolicy("Command Injection", 5, True, False),
"CWE-89": CwePolicy("SQL Injection", 5, True, False),
"CWE-116": CwePolicy("Encode Output", 2, True, False),
}
forced = select_forced_candidates(
[mild, sev5, inferred],
policy,
rule_cwe={
sev5.finding_id: "CWE-78",
inferred.finding_id: "CWE-89",
mild.finding_id: "CWE-116",
},
)
assert forced == (sev5, inferred)
def test_ledger_round_trip_matches_design_schema(tmp_path: Path) -> None:
@@ -250,25 +202,6 @@ def test_overlay_reorders_and_rebands_without_dropping_hits() -> None:
assert overlaid[0].inventory_finding_ids == ("low:1",)
def test_persist_verdicts_records_triggerable_and_skips_inconclusive(tmp_path: Path) -> None:
path = tmp_path / ".openultrasast" / "calibration" / "complexity_ledger.json"
persist_verdicts(
path,
[
SimpleNamespace(path="app.py", function_name="admin", verdict="triggerable"),
SimpleNamespace(path="safe.py", function_name=None, verdict="not_triggerable"),
SimpleNamespace(path="other.py", function_name="skip", verdict="inconclusive"),
],
)
loaded = load_ledger(path)
assert loaded[hotspot_key("app.py", "admin")].last_verdict == "triggerable"
assert loaded[hotspot_key("app.py", "admin")].score_delta == TRIGGERABLE_DELTA
assert loaded[hotspot_key("safe.py", None)].last_verdict == "not_triggerable"
assert loaded[hotspot_key("safe.py", None)].score_delta == NOT_TRIGGERABLE_DELTA
assert hotspot_key("other.py", "skip") not in loaded
def test_build_complexity_map_without_ledger_path_skips_overlay(tmp_path: Path) -> None:
target = _target("src/app.py", loc=20)
finding = _hit("src/app.py", "rule:src/app.py:1")
tests/test_config.py
+2 -30
View File
@@ -147,26 +147,22 @@ def test_write_resolved_config_creates_json(tmp_path: Path) -> None:
assert not {"dynamic", "evidence", "variants", "embeddings"} & set(payload) and "patcher" not in payload["models"]
def test_complexity_and_regress_defaults_when_section_absent() -> None:
def test_complexity_defaults_when_section_absent() -> None:
config = load_config(None)
assert config.complexity.max_hunter_hotspots == 8
assert config.regress.max_candidates == 5
assert config.regress.images == ()
assert config.sandbox.memory_mb == 2048
assert config.sandbox.timeout_seconds == 300
assert config.sandbox.pids_limit == 512
def test_override_changes_max_hunter_hotspots_and_max_candidates_only(tmp_path: Path) -> None:
def test_override_changes_max_hunter_hotspots_only(tmp_path: Path) -> None:
config_path = tmp_path / "openultrasast.toml"
config_path.write_text(
"\n".join(
[
"[complexity]",
"max_hunter_hotspots = 3",
"[regress]",
"max_candidates = 2",
]
)
)
@@ -174,35 +170,11 @@ def test_override_changes_max_hunter_hotspots_and_max_candidates_only(tmp_path:
config = load_config(config_path)
assert config.complexity.max_hunter_hotspots == 3
assert config.regress.max_candidates == 2
assert config.regress.images == ()
assert config.sandbox.memory_mb == 2048
assert config.sandbox.timeout_seconds == 300
assert config.sandbox.pids_limit == 512
def test_load_config_reads_regress_image_pins(tmp_path: Path) -> None:
config_path = tmp_path / "openultrasast.toml"
config_path.write_text(
"\n".join(
[
"[regress]",
"max_candidates = 4",
"[regress.images]",
'python = "python:3.12-slim"',
'javascript = "node:22-slim"',
]
)
)
config = load_config(config_path)
assert config.regress.max_candidates == 4
assert dict(config.regress.images) == {"python": "python:3.12-slim", "javascript": "node:22-slim"}
assert config.complexity.max_hunter_hotspots == 8
assert config.sandbox.memory_mb == 2048
def test_load_dotenv_sets_missing_keys_only(tmp_path: Path, monkeypatch) -> None: # type: ignore[no-untyped-def]
env_file = tmp_path / ".env"
env_file.write_text("OPENROUTER_API_KEY=from-file\nS3_REGION=us-east-1\n")
tests/test_cpg_queries_compile.py
+1 -1
View File
@@ -29,7 +29,7 @@ import pytest
QUERIES = Path("src/openultrasast/cpg/queries")
@pytest.mark.parametrize("name", ["taint", "dominance", "config", "census", "overlay", "vocabulary"])
@pytest.mark.parametrize("name", ["taint", "dominance", "config", "census", "overlay"])
def test_query_scope_checks_guard_nullable_methods(name: str) -> None:
"""PHP class constants can have no method; one unsafe scope check aborts the batch."""
script = QUERIES / f"{name}.sc"
tests/test_degradation_matrix.py
+2 -4
View File
@@ -64,16 +64,14 @@ def test_without_the_semantic_extra_an_unsupported_language_is_named_not_guessed
assert ir.reason == "language_unsupported" and ir.parse_ok is False
def test_without_a_sandbox_the_regress_stage_is_skipped_by_name(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""A claim that could not be proved because nothing could run it is not the same as a claim that failed."""
def test_retired_regress_stage_is_skipped_by_name(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
import json
from openultrasast.cli import main
monkeypatch.setenv("OPENULTRASAST_SANDBOX_PROBE", "0")
(tmp_path / "app.py").write_text("import os\n\n\ndef run(cmd):\n os.system(cmd)\n")
assert main(["scan", str(tmp_path), "--mode", "deep"]) in {0, 1}
runs = sorted((tmp_path / ".openultrasast" / "runs").iterdir())
manifest = json.loads((runs[-1] / "manifest.json").read_text())
reasons = {item["reason"] for item in manifest.get("degradations", [])}
assert "sandbox_unavailable" in reasons
assert "regress_retired" in reasons
tests/test_gate.py
-1
View File
@@ -133,7 +133,6 @@ GOVERNANCE_SCORING_BENCHMARK_PLANES = (
"openultrasast.ruleset",
"openultrasast.ruleset.store",
"openultrasast.fusion",
"openultrasast.skills",
"openultrasast.redaction",
)
tests/test_hunter.py
+2 -20
View File
@@ -2,7 +2,7 @@ import json
from pathlib import Path
from openultrasast.cli import main
from openultrasast.hunter import assign_tier, run_hunter_pool, schedule_hunter_tasks, select_skill_snippets, write_hunter_trajectories
from openultrasast.hunter import assign_tier, run_hunter_pool, schedule_hunter_tasks, write_hunter_trajectories
from openultrasast.preprocess import FileTarget
from openultrasast.rank import RankingScore
@@ -13,7 +13,7 @@ def test_assign_tier_uses_priority_bands() -> None:
assert assign_tier(1.99) == "C"
def test_schedule_hunter_tasks_assigns_budget_and_skill_metadata() -> None:
def test_schedule_hunter_tasks_assigns_budget() -> None:
target = FileTarget(
path="parser.c",
absolute_path="/repo/parser.c",
@@ -39,8 +39,6 @@ def test_schedule_hunter_tasks_assigns_budget_and_skill_metadata() -> None:
assert tasks[0].tier == "A"
assert tasks[0].budget.max_findings_per_target == 20
assert len(tasks[0].retrieval_context) == tasks[0].budget.retrieval_budget_chars
assert "address-sanitizer" in tasks[0].selected_skills
assert "harness-writing" in tasks[0].selected_skills
def test_run_hunter_pool_persists_trajectory_linked_to_finding(tmp_path: Path) -> None:
@@ -78,22 +76,6 @@ def test_run_hunter_pool_persists_trajectory_linked_to_finding(tmp_path: Path) -
assert line["status"] == "completed"
def test_select_skill_snippets_routes_crypto_and_auth_targets() -> None:
target = FileTarget(
path="auth/crypto.py",
absolute_path="/repo/auth/crypto.py",
language="python",
loc=20,
tags=["crypto", "auth_boundary"],
has_fuzz_entry_point=False,
)
snippets = select_skill_snippets(target)
assert "constant-time-analysis" in snippets
assert "sharp-edges" in snippets
def test_standard_mode_writes_hunter_trajectory_and_manifest_ref(tmp_path: Path, monkeypatch) -> None: # type: ignore[no-untyped-def]
repo = tmp_path / "repo"
repo.mkdir()
tests/test_index.py
-33
View File
@@ -1,33 +0,0 @@
import pytest
from openultrasast.index import chunk_text_namespace
def test_namespace_chunks_are_metadata_scoped() -> None:
chunks = chunk_text_namespace(
namespace="skills",
path="semgrep.md",
text="Use Semgrep for variant mapping.\nKeep provenance.",
metadata={"language": "markdown", "vulnerability_class": "mapping"},
max_lines=1,
)
assert len(chunks) == 2
assert chunks[0].namespace == "skills"
assert chunks[0].language == "markdown"
assert chunks[0].metadata["vulnerability_class"] == "mapping"
assert (chunks[0].start_line, chunks[0].end_line) == (1, 1)
assert chunks[0].chunk_id != chunks[1].chunk_id
def test_blank_blocks_are_skipped_and_the_language_defaults_to_text() -> None:
chunks = chunk_text_namespace(namespace="docs", path="README.md", text="alpha\n\n\nbeta\n", max_lines=1)
assert [chunk.text for chunk in chunks] == ["alpha", "beta"]
assert chunks[0].language == "text"
@pytest.mark.parametrize(("namespace", "max_lines"), [("unknown", 80), ("docs", 0)])
def test_invalid_namespace_or_window_is_rejected(namespace: str, max_lines: int) -> None:
with pytest.raises(ValueError):
chunk_text_namespace(namespace=namespace, path="x", text="y", max_lines=max_lines)
tests/test_module_audit.py
-93
View File
@@ -1,93 +0,0 @@
"""model-grounded-detection Req 3: outgrowth is visible and checkable, not noticed two specs later."""
from __future__ import annotations
import json
from pathlib import Path
import pytest
SRC = Path("src/openultrasast")
MANIFEST = Path("benchmarks/measurements/2026-09-08-module-audit.json")
def test_no_module_in_the_tree_is_orphaned() -> None:
from openultrasast.model.audit import audit
orphans = [row.module for row in audit(SRC) if row.classification == "orphaned"]
assert not orphans, f"orphaned modules must be removed, not carried: {orphans}"
def test_every_module_is_classified() -> None:
from openultrasast.model.audit import audit, module_name
classified = {row.module for row in audit(SRC)}
on_disk = {module_name(path, SRC) for path in SRC.rglob("*.py") if "__pycache__" not in path.parts}
assert on_disk - classified <= {""}, f"unclassified: {sorted(on_disk - classified - {''})}"
def test_the_named_subsystems_each_carry_an_explicit_decision() -> None:
"""Req 3.3: the older-phase subsystems still in the tree get a decision, not silence."""
from openultrasast.model.audit import audit
rows = {row.module: row for row in audit(SRC)}
for name in ("fusion", "mcp", "skills"):
assert name in rows, f"{name} vanished without a recorded decision"
assert rows[name].reason, f"{name} has no stated reason"
# The removed agentic plane's modules are gone from the tree, not carried as rows (harnessx-removal Req 2.1).
assert not {"harness_ext", "hunter_harness", "verify_judge"} & set(rows)
# The five the reachability-first audit found orphaned are gone too (legacy cleanup 2026-10-02).
assert not {"model.judge", "model.execution", "model.calibrate", "stage_processors", "slot_contract"} & set(rows)
def test_the_manifest_matches_the_tree() -> None:
"""A later reader can check the tree against the committed manifest (Req 3.4)."""
from openultrasast.model.audit import audit, to_dict
assert MANIFEST.is_file(), "the audit manifest is committed evidence, not a local artifact"
committed = json.loads(MANIFEST.read_text())
current = to_dict(audit(SRC))
assert committed["counts"] == current["counts"], "the tree drifted from its committed audit"
assert committed["orphaned"] == current["orphaned"] == []
def _tree(root: Path, files: dict[str, str]) -> Path:
for name, text in files.items():
path = root / name
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(text)
return root
def test_a_spec_listed_module_nothing_imports_is_still_orphaned(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""Reachability decides; SPEC_OWNED and STANDALONE only supply the reason (the list hid five dead modules)."""
from openultrasast.model import audit as module
src = _tree(tmp_path, {"cli.py": "from . import used\n", "used.py": "", "ghost.py": "", "lonely.py": ""})
monkeypatch.setitem(module.SPEC_OWNED, "ghost", "some spec Req 1: still listed")
monkeypatch.setitem(module.STANDALONE, "lonely", "its own entry point, supposedly")
rows = {row.module: row for row in module.audit(src)}
assert rows["used"].classification == "load_bearing"
assert rows["ghost"].classification == "orphaned"
assert "still listed" in rows["ghost"].reason
assert rows["lonely"].classification == "orphaned"
def test_submodule_imports_and_python_m_modules_count_as_reachable(tmp_path: Path) -> None:
from openultrasast.model.audit import audit
src = _tree(
tmp_path,
{
"cli.py": "def run():\n from .pkg import sub\n",
"pkg/__init__.py": "",
"pkg/sub.py": "",
"tool.py": 'from . import helper\n\nif __name__ == "__main__":\n pass\n',
"helper.py": "",
"dead.py": "from . import deader\n",
"deader.py": "",
},
)
rows = {row.module: row.classification for row in audit(src)}
assert rows["pkg.sub"] == rows["tool"] == rows["helper"] == "load_bearing"
assert rows["dead"] == rows["deader"] == "orphaned"
tests/test_obligation_scan.py
-16
View File
@@ -92,19 +92,3 @@ def test_quick_scan_and_disabled_config_skip_obligations(tmp_path: Path, monkeyp
assert not any(f["finding_id"].startswith("obligation:") for f in findings)
def test_deep_scan_never_offers_obligation_findings_to_the_sandbox(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
from openultrasast import cli
from openultrasast.sandbox import FakeSandboxRunner, SandboxResult
repo = _repo(tmp_path, monkeypatch)
(repo / "openultrasast.toml").write_text("[obligations]\nmin_siblings = 2\n")
monkeypatch.setenv("OPENULTRASAST_SANDBOX_PROBE", "1")
fake = FakeSandboxRunner(SandboxResult(exit_code=0, stdout="", stderr="", timed_out=False))
monkeypatch.setattr(cli, "resolve_sandbox_runner", lambda: fake)
assert main(["scan", str(repo), "--mode", "deep", "--config", str(repo / "openultrasast.toml")]) == 0
run_dir = _run_dir(repo)
verdicts = json.loads((run_dir / "verdicts.json").read_text())["verdicts"]
offered = {finding_id for item in verdicts for finding_id in item["inventory_finding_ids"]}
assert not any(finding_id.startswith("obligation:") for finding_id in offered)
findings = json.loads((run_dir / "findings.json").read_text())["findings"]
assert any(f["finding_id"].startswith("obligation:") for f in findings) # still reported
tests/test_regress_candidates.py
-159
View File
@@ -1,159 +0,0 @@
"""Bounded regression candidate selection from hotspots and severe inventory (task 5.1)."""
from openultrasast.complexity.ledger import NOT_TRIGGERABLE_DELTA, apply_overlay, record_verdict
from openultrasast.complexity.map import Hotspot
from openultrasast.findings import StaticFinding
from openultrasast.regress.candidate import select_candidates
def _finding(
path: str,
finding_id: str,
*,
function_name: str | None = None,
reachability: str = "unknown",
severity: str = "low",
) -> StaticFinding:
return StaticFinding(
finding_id=finding_id,
path=path,
title="pattern hit",
severity=severity,
confidence="medium",
evidence_level="static_corroboration",
rationale="inventory",
line=1,
function_name=function_name,
reachability_status=reachability,
reachability_evidence=[],
reachability_conditions=[],
tags=[],
ranking_priority=1.0,
)
def _hotspot(
path: str,
*,
function_name: str | None = None,
score: float = 4.0,
band: str = "medium",
inventory_finding_ids: tuple[str, ...] = (),
) -> Hotspot:
return Hotspot(
path=path,
function_name=function_name,
score=score,
band=band,
signals={"loc": 10, "inventory_hit_count": len(inventory_finding_ids)},
rationale=f"{band} band: score {score}",
test_hint=None,
inventory_finding_ids=inventory_finding_ids,
)
def _six_hotspots(*, sev5_id: str | None = None, sev5_score: float = 4.0) -> tuple[Hotspot, ...]:
return (
_hotspot("a.py", function_name="a", score=5.0),
_hotspot("b.py", function_name="b", score=4.8),
_hotspot("c.py", function_name="c", score=4.6),
_hotspot("d.py", function_name="d", score=4.4),
_hotspot("e.py", function_name="e", score=4.2),
_hotspot(
"f.py",
function_name="f",
score=sev5_score,
inventory_finding_ids=(sev5_id,) if sev5_id else (),
),
)
def test_six_hotspots_max_candidates_three_returns_three() -> None:
noise = _finding("f.py", "print:f.py:1", function_name="f", reachability="reachable")
selected = select_candidates(
_six_hotspots(),
[noise],
max_candidates=3,
policy_severity_by_id={noise.finding_id: 2},
)
assert len(selected) == 3
assert [item.path for item in selected] == ["a.py", "b.py", "c.py"]
def test_sev5_reachable_extra_included_when_ledger_demoted_hotspot() -> None:
sev5_id = "cmd:f.py:1"
hotspots = _six_hotspots(sev5_id=sev5_id, sev5_score=5.2)
ledger = record_verdict({}, path="f.py", function_name="f", verdict="not_triggerable")
demoted = apply_overlay(hotspots, ledger)
finding = _finding(
"f.py",
sev5_id,
function_name="f",
reachability="reachable",
severity="critical",
)
selected = select_candidates(
demoted,
[finding],
max_candidates=3,
policy_severity_by_id={sev5_id: 5},
)
demoted_f = next(item for item in demoted if item.path == "f.py")
assert demoted_f.score == round(5.2 + NOT_TRIGGERABLE_DELTA, 4)
assert [item.path for item in demoted[:3]] == ["a.py", "b.py", "c.py"]
assert demoted_f not in demoted[:3]
assert len(selected) == 4
assert [item.path for item in selected[:3]] == ["a.py", "b.py", "c.py"]
assert selected[3] is demoted_f
assert sev5_id in selected[3].inventory_finding_ids
def test_sev5_already_in_cap_is_not_duplicated() -> None:
sev5_id = "cmd:a.py:1"
hotspots = (
_hotspot("a.py", function_name="a", score=5.0, inventory_finding_ids=(sev5_id,)),
_hotspot("b.py", function_name="b", score=4.8),
_hotspot("c.py", function_name="c", score=4.6),
_hotspot("d.py", function_name="d", score=4.4),
_hotspot("e.py", function_name="e", score=4.2),
_hotspot("f.py", function_name="f", score=4.0),
)
finding = _finding("a.py", sev5_id, function_name="a", reachability="reachable", severity="critical")
selected = select_candidates(
hotspots,
[finding],
max_candidates=3,
policy_severity_by_id={sev5_id: 5},
)
assert len(selected) == 3
assert [item.path for item in selected] == ["a.py", "b.py", "c.py"]
assert sev5_id in selected[0].inventory_finding_ids
def test_sev5_reachable_inventory_without_hotspot_is_still_selected() -> None:
finding = _finding(
"z.py",
"cmd:z.py:1",
function_name="z",
reachability="reachable",
severity="critical",
)
selected = select_candidates(
_six_hotspots(),
[finding],
max_candidates=3,
policy_severity_by_id={finding.finding_id: 5},
)
assert len(selected) == 4
assert [item.path for item in selected[:3]] == ["a.py", "b.py", "c.py"]
extra = selected[3]
assert extra.path == "z.py"
assert extra.function_name == "z"
assert extra.inventory_finding_ids == (finding.finding_id,)
tests/test_regress_recipes.py
-103
View File
@@ -1,103 +0,0 @@
"""Language recipes for isolated regression snippets (task 5.2)."""
from pathlib import Path
from openultrasast.config import SandboxConfig
from openultrasast.regress import INCONCLUSIVE, RegressionRunner
from openultrasast.regress.recipes import recipe_for
from openultrasast.sandbox import FakeSandboxRunner, SandboxJob
LIMITS = SandboxConfig(memory_mb=256, timeout_seconds=30, pids_limit=128)
PYTHON_SNIPPET = "assert 1 + 1 == 2\n"
JS_SNIPPET = "console.log('ok')\n"
C_SNIPPET = "int main(void) { return 0; }\n"
JAVA_SNIPPET = "class Case { public static void main(String[] args) {} }\n"
def _repo(tmp_path: Path) -> Path:
repo = tmp_path / "repo"
repo.mkdir()
return repo
def test_python_recipe_runs_scratch_snippet_with_repo_on_pythonpath(tmp_path: Path) -> None:
repo = _repo(tmp_path)
job = recipe_for("python", PYTHON_SNIPPET, "python:3.12-alpine", LIMITS, repo_root=repo)
assert isinstance(job, SandboxJob)
joined = " ".join(job.command)
assert job.image == "python:3.12-alpine"
assert job.repo_root == repo
assert job.scratch_files == {"case.py": PYTHON_SNIPPET}
assert "python" in job.command
assert "/scratch/case.py" in job.command
assert "PYTHONPATH=/workspace" in joined
assert job.timeout_seconds == LIMITS.timeout_seconds
assert job.memory_mb == LIMITS.memory_mb
assert job.pids_limit == LIMITS.pids_limit
def test_javascript_recipe_runs_scratch_snippet_with_repo_on_node_path(tmp_path: Path) -> None:
repo = _repo(tmp_path)
job = recipe_for("javascript", JS_SNIPPET, "node:22-alpine", LIMITS, repo_root=repo)
assert isinstance(job, SandboxJob)
joined = " ".join(job.command)
assert job.image == "node:22-alpine"
assert job.repo_root == repo
assert job.scratch_files == {"case.js": JS_SNIPPET}
assert "node" in job.command
assert "/scratch/case.js" in job.command
assert "NODE_PATH=/workspace" in joined
assert job.timeout_seconds == LIMITS.timeout_seconds
assert job.memory_mb == LIMITS.memory_mb
assert job.pids_limit == LIMITS.pids_limit
def test_c_recipe_compiles_and_runs_when_image_has_compiler(tmp_path: Path) -> None:
repo = _repo(tmp_path)
job = recipe_for("c", C_SNIPPET, "gcc:13", LIMITS, repo_root=repo)
assert isinstance(job, SandboxJob)
joined = " ".join(job.command)
assert job.scratch_files == {"case.c": C_SNIPPET}
assert "cc" in joined
assert "/scratch/case.c" in joined
assert "-o" in joined
assert "/scratch/case" in joined
def test_unknown_language_recipe_is_none(tmp_path: Path) -> None:
repo = _repo(tmp_path)
assert recipe_for("java", JAVA_SNIPPET, "eclipse-temurin:21", LIMITS, repo_root=repo) is None
assert recipe_for("unknown", "print(1)\n", "python:3.12-alpine", LIMITS, repo_root=repo) is None
def test_python_and_javascript_recipes_produce_runner_jobs(tmp_path: Path) -> None:
repo = _repo(tmp_path)
python_job = recipe_for("python", PYTHON_SNIPPET, "python:3.12-alpine", LIMITS, repo_root=repo)
js_job = recipe_for("javascript", JS_SNIPPET, "node:22-alpine", LIMITS, repo_root=repo)
sandbox = FakeSandboxRunner()
runner = RegressionRunner(sandbox)
python_verdict = runner.run_recipe("python", PYTHON_SNIPPET, "python:3.12-alpine", LIMITS, repo_root=repo)
js_verdict = runner.run_recipe("javascript", JS_SNIPPET, "node:22-alpine", LIMITS, repo_root=repo)
assert python_job is not None
assert js_job is not None
assert python_verdict.reason != "missing_recipe"
assert js_verdict.reason != "missing_recipe"
assert sandbox.jobs == [python_job, js_job]
def test_unknown_language_returns_inconclusive_without_calling_runner(tmp_path: Path) -> None:
repo = _repo(tmp_path)
sandbox = FakeSandboxRunner()
runner = RegressionRunner(sandbox)
assert recipe_for("java", JAVA_SNIPPET, "eclipse-temurin:21", LIMITS, repo_root=repo) is None
verdict = runner.run_recipe("java", JAVA_SNIPPET, "eclipse-temurin:21", LIMITS, repo_root=repo)
assert verdict.verdict == INCONCLUSIVE
assert verdict.reason == "missing_recipe"
assert sandbox.jobs == []
tests/test_regress_safety.py
-86
View File
@@ -1,86 +0,0 @@
"""Reject unsafe regression snippets before the sandbox runner sees them (task 4.3)."""
from pathlib import Path
import pytest
from openultrasast.regress import RegressionRunner, UnsafeSnippetError, check_snippet_safety
from openultrasast.sandbox import FakeSandboxRunner, SandboxJob
BENIGN_ASSERT = "assert 1 + 1 == 2\n"
def _job(tmp_path: Path, snippet: str) -> SandboxJob:
repo = tmp_path / "repo"
repo.mkdir()
return SandboxJob(
image="python:3.12-alpine",
command=("python", "/scratch/case.py"),
repo_root=repo,
scratch_files={"case.py": snippet},
timeout_seconds=30,
memory_mb=256,
pids_limit=128,
)
def test_docker_sock_rejected_does_not_create_runner_job(tmp_path: Path) -> None:
snippet = 'client = docker.DockerClient(base_url="unix://var/run/docker.sock")\n'
sandbox = FakeSandboxRunner()
runner = RegressionRunner(sandbox)
with pytest.raises(UnsafeSnippetError, match="Docker socket"):
check_snippet_safety(snippet)
verdict = runner.run_snippet(snippet, _job(tmp_path, snippet))
assert verdict.verdict == "inconclusive"
assert verdict.reason == "safety_rejected"
assert sandbox.jobs == []
def test_host_network_rejected_does_not_create_runner_job(tmp_path: Path) -> None:
snippet = "subprocess.run(['docker', 'run', '--network', 'host', 'alpine'])\n"
sandbox = FakeSandboxRunner()
runner = RegressionRunner(sandbox)
with pytest.raises(UnsafeSnippetError, match="host network"):
check_snippet_safety(snippet)
verdict = runner.run_snippet(snippet, _job(tmp_path, snippet))
assert verdict.verdict == "inconclusive"
assert verdict.reason == "safety_rejected"
assert sandbox.jobs == []
def test_write_under_workspace_rejected_does_not_create_runner_job(tmp_path: Path) -> None:
snippet = 'open("/workspace/pwned.py", "w").write("owned")\n'
sandbox = FakeSandboxRunner()
runner = RegressionRunner(sandbox)
with pytest.raises(UnsafeSnippetError, match="/workspace"):
check_snippet_safety(snippet)
verdict = runner.run_snippet(snippet, _job(tmp_path, snippet))
assert verdict.verdict == "inconclusive"
assert verdict.reason == "safety_rejected"
assert sandbox.jobs == []
def test_benign_python_assert_snippet_is_allowed(tmp_path: Path) -> None:
assert check_snippet_safety(BENIGN_ASSERT) is None
sandbox = FakeSandboxRunner()
job = _job(tmp_path, BENIGN_ASSERT)
RegressionRunner(sandbox).run_snippet(BENIGN_ASSERT, job)
assert sandbox.jobs == [job]
@pytest.mark.parametrize(
"snippet",
[
"import socket\nsocket.create_connection(('127.0.0.1', 80))\n",
"curl https://example.invalid/pwn\n",
],
)
def test_socket_and_curl_mentions_are_rejected(snippet: str) -> None:
with pytest.raises(UnsafeSnippetError):
check_snippet_safety(snippet)
tests/test_regress_verdicts.py
-201
View File
@@ -1,201 +0,0 @@
"""Map sandbox outcomes to the four verdicts and the worth-fixing gate (task 5.3)."""
from __future__ import annotations
from pathlib import Path
import pytest
from openultrasast.regress import RegressionRunner
from openultrasast.regress.verdict import (
ALREADY_COVERED,
INCONCLUSIVE,
NOT_TRIGGERABLE,
TRIGGERABLE,
is_worth_fixing,
verdict_from_result,
)
from openultrasast.sandbox import FakeSandboxRunner, SandboxJob, SandboxResult
_REACHABLE = "reachable"
_INFERRED = "inferred-file-surface"
_UNKNOWN = "unknown"
def _result(*, exit_code: int = 0, stdout: str = "", stderr: str = "", timed_out: bool = False) -> SandboxResult:
return SandboxResult(exit_code=exit_code, stdout=stdout, stderr=stderr, timed_out=timed_out)
def _job(tmp_path: Path, snippet: str) -> SandboxJob:
repo = tmp_path / "repo"
repo.mkdir()
return SandboxJob(
image="python:3.12-alpine",
command=("python", "/scratch/case.py"),
repo_root=repo,
scratch_files={"case.py": snippet},
timeout_seconds=30,
memory_mb=256,
pids_limit=128,
)
# observation kwargs, sandbox result, reachability, expected verdict, expected reason, worth-fixing
_CASES = (
pytest.param(
{"recipe_missing": True},
None,
_REACHABLE,
INCONCLUSIVE,
"missing_recipe",
False,
id="recipe-missing-inconclusive",
),
pytest.param(
{"sandbox_missing": True},
None,
_REACHABLE,
INCONCLUSIVE,
"sandbox_unavailable",
False,
id="sandbox-missing-inconclusive",
),
pytest.param(
{},
_result(exit_code=-1, timed_out=True),
_REACHABLE,
INCONCLUSIVE,
"timeout",
False,
id="timeout-inconclusive",
),
pytest.param(
{"safety_rejected": True},
None,
_INFERRED,
INCONCLUSIVE,
"safety_rejected",
False,
id="safety-reject-inconclusive",
),
pytest.param(
{"covering_test": True},
_result(exit_code=0, stdout="ok\n"),
_REACHABLE,
ALREADY_COVERED,
"covered_test_passed",
False,
id="covering-test-passed-already-covered",
),
pytest.param(
{},
_result(exit_code=1, stderr="AssertionError: boom\n"),
_REACHABLE,
TRIGGERABLE,
"assertion",
True,
id="assertion-triggerable-reachable-worth-fixing",
),
pytest.param(
{},
_result(exit_code=1, stderr="ERROR: AddressSanitizer: heap-buffer-overflow\n"),
_INFERRED,
TRIGGERABLE,
"sanitizer_abort",
True,
id="sanitizer-triggerable-inferred-worth-fixing",
),
pytest.param(
{},
_result(exit_code=139, stderr="Segmentation fault\n"),
_UNKNOWN,
TRIGGERABLE,
"crash",
False,
id="crash-triggerable-unknown-not-worth-fixing",
),
pytest.param(
{},
_result(exit_code=0, stdout="ok\n"),
_REACHABLE,
NOT_TRIGGERABLE,
"exit_zero",
False,
id="exit-zero-not-triggerable",
),
pytest.param(
{"covering_test": True},
_result(exit_code=0),
_UNKNOWN,
ALREADY_COVERED,
"covered_test_passed",
False,
id="already-covered-unknown-does-not-fail-build",
),
pytest.param(
{},
_result(exit_code=0),
_INFERRED,
NOT_TRIGGERABLE,
"exit_zero",
False,
id="not-triggerable-inferred-does-not-fail-build",
),
pytest.param(
{},
_result(exit_code=-1, timed_out=True),
_INFERRED,
INCONCLUSIVE,
"timeout",
False,
id="inconclusive-inferred-does-not-fail-build",
),
)
@pytest.mark.parametrize(
("flags", "result", "reachability", "expected_verdict", "expected_reason", "expected_worth_fixing"),
_CASES,
)
def test_sandbox_outcomes_map_to_verdicts_and_worth_fixing(
flags: dict[str, bool],
result: SandboxResult | None,
reachability: str,
expected_verdict: str,
expected_reason: str,
expected_worth_fixing: bool,
) -> None:
mapped = verdict_from_result(result, **flags)
assert mapped.verdict == expected_verdict
assert mapped.reason == expected_reason
assert mapped.reason
assert is_worth_fixing(mapped.verdict, reachability) is expected_worth_fixing
def test_table_covers_four_verdicts_and_worth_fixing_boolean() -> None:
verdicts = {case.values[3] for case in _CASES}
worth_fixing_values = {case.values[5] for case in _CASES}
assert verdicts == {TRIGGERABLE, NOT_TRIGGERABLE, ALREADY_COVERED, INCONCLUSIVE}
assert worth_fixing_values == {True, False}
def test_runner_maps_timeout_crash_and_exit_zero(tmp_path: Path) -> None:
snippet = "assert 1 + 1 == 2\n"
job = _job(tmp_path, snippet)
timeout = FakeSandboxRunner(_result(exit_code=-1, timed_out=True))
crash = FakeSandboxRunner(_result(exit_code=1, stderr="AssertionError\n"))
ok = FakeSandboxRunner(_result(exit_code=0))
timed_out = RegressionRunner(timeout).run_snippet(snippet, job)
triggered = RegressionRunner(crash).run_snippet(snippet, job)
not_triggered = RegressionRunner(ok).run_snippet(snippet, job)
assert timed_out.verdict == INCONCLUSIVE
assert timed_out.reason == "timeout"
assert triggered.verdict == TRIGGERABLE
assert not_triggered.verdict == NOT_TRIGGERABLE
assert is_worth_fixing(timed_out.verdict, _REACHABLE) is False
assert is_worth_fixing(triggered.verdict, _REACHABLE) is True
assert is_worth_fixing(not_triggered.verdict, _REACHABLE) is False
tests/test_reports.py
+33 -127
View File
@@ -1,7 +1,6 @@
import json
import subprocess
from pathlib import Path
from types import SimpleNamespace
import pytest
@@ -9,7 +8,6 @@ from openultrasast.cli import main
from openultrasast.complexity.hints import TestHint
from openultrasast.complexity.map import ComplexityMap, Hotspot
from openultrasast.findings import StaticFinding
from openultrasast.regress import CandidateVerdict
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
@@ -99,7 +97,7 @@ def test_markdown_report_includes_verification_status(tmp_path: Path) -> None:
assert "worth_fixing" not in text.lower()
def test_markdown_report_includes_complexity_map_and_worth_fixing_sections(tmp_path: Path) -> None:
def test_markdown_report_includes_complexity_map(tmp_path: Path) -> None:
finding = _finding()
output = tmp_path / "report.md"
hint = TestHint(
@@ -119,47 +117,21 @@ def test_markdown_report_includes_complexity_map_and_worth_fixing_sections(tmp_p
test_hint=hint,
inventory_finding_ids=(finding.finding_id,),
)
verdict = CandidateVerdict(
path="app.py",
function_name="admin",
language="python",
verdict="triggerable",
reason="crash",
inventory_finding_ids=(finding.finding_id,),
worth_fixing=True,
)
skipped = CandidateVerdict(
path="safe.py",
function_name=None,
language="python",
verdict="not_triggerable",
reason="exit_zero",
inventory_finding_ids=(),
worth_fixing=False,
)
write_markdown_report(
[finding],
output,
[verify_finding(finding)],
complexity_map=ComplexityMap(hotspots=(hotspot,), heuristic_only=True),
verdicts=(verdict, skipped),
)
text = output.read_text()
inventory_at = text.index("## Inventory")
map_at = text.index("## Complexity map")
worth_at = text.index("## Worth fixing")
assert inventory_at < map_at < worth_at
map_section = text[map_at:worth_at]
worth_section = text[worth_at:]
assert inventory_at < map_at
map_section = text[map_at:]
assert "app.py::admin" in map_section
assert "Gap: `no_adjacent_test`" in map_section
assert "Recommended test: `http-contract`" in map_section
assert "triggerable" in worth_section
assert "crash" in worth_section
assert "safe.py" not in worth_section
assert "## Inventory" in text
def test_markdown_report_labels_hits_as_inventory_not_worth_fixing(tmp_path: Path) -> None:
@@ -178,30 +150,12 @@ def test_markdown_report_labels_hits_as_inventory_not_worth_fixing(tmp_path: Pat
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:
@@ -209,16 +163,6 @@ def test_scan_exit_code_unknown_policy() -> None:
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()
@@ -236,20 +180,9 @@ def test_cli_scan_writes_sarif_manifest_and_honors_fail_policy(tmp_path: Path, m
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()
@@ -273,8 +206,6 @@ def test_quick_scan_report_is_inventory_without_map_or_docker(tmp_path: Path, mo
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"))
@@ -297,8 +228,6 @@ def test_quick_scan_report_is_inventory_without_map_or_docker(tmp_path: Path, mo
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()
@@ -315,15 +244,7 @@ def test_standard_scan_writes_complexity_map_without_docker(tmp_path: Path, monk
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
@@ -351,23 +272,12 @@ def test_standard_scan_writes_complexity_map_without_docker(tmp_path: Path, monk
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]
@@ -408,38 +318,6 @@ def test_manifest_includes_complexity_summary(tmp_path: Path) -> None:
assert payload["artifacts"]["complexity_map"] == "complexity_map.json"
def test_manifest_includes_worth_fixing_count(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",
"verdicts": run.root / "verdicts.json",
}
output = run.root / "manifest.json"
write_manifest(
run=run,
findings=[finding],
verifications=[verify_finding(finding)],
artifact_paths=artifacts,
path=output,
stages={"requested": ["static", "map", "regress"], "completed": ["static", "map", "regress"], "skipped": []},
worth_fixing={
"count": 1,
"verdicts": [{"path": "app.py", "function_name": "admin", "verdict": "triggerable", "reason": "crash"}],
},
)
payload = json.loads(output.read_text())
assert payload["stages"]["completed"] == ["static", "map", "regress"]
assert payload["worth_fixing"]["count"] == 1
assert payload["worth_fixing"]["verdicts"][0]["verdict"] == "triggerable"
def _finding() -> StaticFinding:
return StaticFinding(
finding_id="python-unsafe-eval:app.py:3",
@@ -669,7 +547,7 @@ def test_every_recorded_scan_degradation_is_rendered(tmp_path) -> None: # type:
{"stage": "model", "reason": "model_withheld_execution_budget"},
{"stage": "budget", "reason": "max_findings_exceeded", "requested": 5, "actual": 9},
{"stage": "map", "reason": "hunter_model_unavailable"},
{"stage": "regress", "reason": "sandbox_unavailable"},
{"stage": "regress", "reason": "regress_retired"},
{"stage": "x", "reason": "not_yet_described"},
]
path = tmp_path / "report.md"
@@ -684,7 +562,8 @@ def test_every_recorded_scan_degradation_is_rendered(tmp_path) -> None: # type:
"time budget leaves no room",
"first 5 of 9 findings",
"needs a model and none is available",
"Docker is not available",
"the sandboxed snippet stage was retired on 2026-10-13",
"execution proof comes from the search-with-proof arm",
"`not_yet_described`",
):
assert needle in text, needle
@@ -697,3 +576,30 @@ def test_quick_scan_names_a_language_nothing_covers(tmp_path, capsys) -> None:
assert "notice: language not covered: go (1 files)" in capsys.readouterr().out
report = next((tmp_path / ".openultrasast" / "runs").iterdir()) / "report.md"
assert "Not a clean result for go (1 files)" in report.read_text()
@pytest.mark.parametrize("fail_on, expected", [("never", 0), ("findings", 1), ("verified", 1)])
def test_deep_scan_records_retirement_without_running_snippets(tmp_path, monkeypatch, fail_on, expected):
(tmp_path / "app.py").write_text("@app.route('/admin')\ndef admin():\n return eval(request.data)\n")
def guard(command, *args, **kwargs):
argv = list(command) if isinstance(command, list | tuple) else [command]
assert not argv or Path(str(argv[0])).name != "docker", "deep scan must not start Docker"
return original_run(command, *args, **kwargs)
original_run = subprocess.run
monkeypatch.setattr(subprocess, "run", guard)
assert main(["scan", str(tmp_path), "--mode", "deep", "--fail-on", fail_on]) == expected
run_dir = next((tmp_path / ".openultrasast" / "runs").iterdir())
manifest = json.loads((run_dir / "manifest.json").read_text())
retired = {"stage": "regress", "reason": "regress_retired"}
assert manifest["stages"] == {
"requested": ["static", "map", "regress"],
"completed": ["static", "map"],
"skipped": [retired],
}
assert manifest["degradations"].count(retired) == 1
assert "verdicts" not in manifest["artifacts"]
assert "worth_fixing" not in manifest
assert not (run_dir / "verdicts.json").exists()
assert "execution proof comes from the search-with-proof arm" in (run_dir / "report.md").read_text()
tests/test_sandbox_probe.py
-117
View File
@@ -1,117 +0,0 @@
import subprocess
from typing import Any
import pytest
from openultrasast.sandbox import FakeSandboxProbe, SandboxProbe, resolve_sandbox_probe
from openultrasast.sandbox import probe as probe_mod
def _completed(returncode: int) -> subprocess.CompletedProcess[str]:
return subprocess.CompletedProcess(["docker", "info"], returncode, stdout="", stderr="")
def test_fake_probe_can_be_forced_off() -> None:
probe = FakeSandboxProbe(available=False)
assert probe.available() is False
def test_fake_probe_can_be_forced_on() -> None:
probe = FakeSandboxProbe(available=True)
assert probe.available() is True
def test_resolve_probe_forced_off_via_env(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv(probe_mod.PROBE_ENV, "0")
probe = resolve_sandbox_probe()
assert isinstance(probe, FakeSandboxProbe)
assert probe.available() is False
def test_resolve_probe_forced_on_via_env(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv(probe_mod.PROBE_ENV, "1")
probe = resolve_sandbox_probe()
assert isinstance(probe, FakeSandboxProbe)
assert probe.available() is True
def test_resolve_probe_defaults_to_real_probe(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv(probe_mod.PROBE_ENV, raising=False)
assert isinstance(resolve_sandbox_probe(), SandboxProbe)
def test_real_probe_is_available_when_docker_info_succeeds() -> None:
def runner(*_args: object, **_kwargs: object) -> subprocess.CompletedProcess[str]:
return _completed(0)
assert SandboxProbe(runner=runner).available() is True
def test_real_probe_is_unavailable_when_docker_info_fails() -> None:
def runner(*_args: object, **_kwargs: object) -> subprocess.CompletedProcess[str]:
return _completed(1)
assert SandboxProbe(runner=runner).available() is False
@pytest.mark.parametrize(
"error",
[
FileNotFoundError("docker"),
PermissionError("denied"),
OSError("cannot talk to daemon"),
subprocess.TimeoutExpired(cmd=["docker", "info"], timeout=5),
RuntimeError("unexpected probe failure"),
],
)
def test_real_probe_returns_unavailable_without_raising(error: Exception) -> None:
def runner(*_args: object, **_kwargs: object) -> subprocess.CompletedProcess[str]:
raise error
assert SandboxProbe(runner=runner).available() is False
def test_real_probe_runs_docker_info_with_short_timeout() -> None:
seen: dict[str, Any] = {}
def runner(command: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]:
seen["command"] = list(command)
seen["timeout"] = kwargs.get("timeout")
seen["check"] = kwargs.get("check")
return _completed(0)
assert SandboxProbe(runner=runner).available() is True
assert seen["command"] == ["docker", "info"]
assert seen["timeout"] == probe_mod.PROBE_TIMEOUT_SECONDS
assert probe_mod.PROBE_TIMEOUT_SECONDS <= 10
assert seen["check"] is False
def test_real_probe_is_skippable_in_unit_tests(monkeypatch: pytest.MonkeyPatch) -> None:
def boom(*_args: object, **_kwargs: object) -> subprocess.CompletedProcess[str]:
raise AssertionError("unit tests must not invoke the docker daemon")
monkeypatch.setattr(probe_mod.subprocess, "run", boom)
def runner(*_args: object, **_kwargs: object) -> subprocess.CompletedProcess[str]:
return _completed(0)
assert SandboxProbe(runner=runner).available() is True
def test_real_probe_default_path_uses_subprocess_run(monkeypatch: pytest.MonkeyPatch) -> None:
seen: dict[str, Any] = {}
def fake_run(command: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]:
seen["command"] = list(command)
seen["timeout"] = kwargs.get("timeout")
seen["capture_output"] = kwargs.get("capture_output")
seen["text"] = kwargs.get("text")
return _completed(0)
monkeypatch.setattr(probe_mod.subprocess, "run", fake_run)
assert SandboxProbe().available() is True
assert seen["command"] == ["docker", "info"]
assert seen["timeout"] == probe_mod.PROBE_TIMEOUT_SECONDS
assert seen["capture_output"] is True
assert seen["text"] is True
tests/test_sandbox_runner.py
-364
View File
@@ -1,364 +0,0 @@
from __future__ import annotations
import os
import stat
import subprocess
from pathlib import Path
from typing import Any
import pytest
from openultrasast.sandbox import (
DockerCliRunner,
FakeSandboxRunner,
SandboxJob,
SandboxResult,
SandboxRunner,
build_docker_argv,
resolve_sandbox_runner,
)
from openultrasast.sandbox import runner as runner_mod
def _job(tmp_path: Path, **overrides: object) -> SandboxJob:
repo = tmp_path / "repo"
repo.mkdir(exist_ok=True)
(repo / "app.py").write_text("print(1)\n")
values: dict[str, object] = {
"image": "python:3.12-alpine",
"command": ("python", "/scratch/case.py"),
"repo_root": repo,
"scratch_files": {"case.py": "print('ok')\n"},
"timeout_seconds": 30,
"memory_mb": 256,
"pids_limit": 128,
}
values.update(overrides)
return SandboxJob(**values) # type: ignore[arg-type]
def _flag_value(argv: list[str], flag: str) -> str:
try:
return argv[argv.index(flag) + 1]
except (ValueError, IndexError) as exc:
raise AssertionError(f"{flag} missing from {argv}") from exc
def _mount_options(spec: str) -> dict[str, str]:
options: dict[str, str] = {}
for part in spec.split(","):
if "=" in part:
key, value = part.split("=", 1)
options[key] = value
else:
options[part] = "true"
return options
def _workspace_mount(argv: list[str]) -> dict[str, str]:
specs: list[str] = []
for index, item in enumerate(argv):
if item in {"--mount", "-v", "--volume"} and index + 1 < len(argv):
specs.append(argv[index + 1])
for spec in specs:
options = _mount_options(spec)
target = options.get("dst") or options.get("destination") or options.get("target")
if target == "/workspace" or spec.endswith(":/workspace") or ":/workspace:" in spec:
return options if target == "/workspace" else {"raw": spec}
if "/workspace" in spec:
return options if options else {"raw": spec}
raise AssertionError(f"no /workspace mount in {argv}")
def _seed_source(argv: list[str]) -> Path:
mounts = [argv[index + 1] for index, item in enumerate(argv) if item == "--mount" and index + 1 < len(argv)]
seed_mounts = [spec for spec in mounts if "/scratch-in" in spec]
assert seed_mounts, argv
options = _mount_options(seed_mounts[0])
source = Path(options.get("src") or options.get("source") or "")
assert source.is_dir(), argv
return source
def _copy_script(argv: list[str]) -> str:
try:
return argv[argv.index("-c") + 1]
except (ValueError, IndexError) as exc:
raise AssertionError(f"no /bin/sh -c wrapper in {argv}") from exc
def _posix_mode(path: Path) -> int:
return path.stat().st_mode & 0o777
def test_build_docker_argv_starts_with_docker_run_rm_init(tmp_path: Path) -> None:
argv = build_docker_argv(_job(tmp_path))
assert argv[0] == "docker"
assert argv[1] == "run"
assert "--rm" in argv
assert "--init" in argv
def test_build_docker_argv_includes_network_none_and_read_only(tmp_path: Path) -> None:
argv = build_docker_argv(_job(tmp_path))
joined = " ".join(argv)
assert "--read-only" in argv
assert "--network none" in joined or "--network=none" in argv
assert "--network host" not in joined
assert "--network=host" not in argv
def test_build_docker_argv_excludes_docker_sock(tmp_path: Path) -> None:
argv = build_docker_argv(_job(tmp_path))
joined = " ".join(argv)
assert "docker.sock" not in joined
assert "/var/run/docker.sock" not in joined
assert "-v" not in argv
assert "--volume" not in argv
def test_build_docker_argv_forbids_privileged_and_host_network(tmp_path: Path) -> None:
argv = build_docker_argv(_job(tmp_path))
assert "--privileged" not in argv
assert "--pid" not in argv or _flag_value(argv, "--pid") != "host"
assert "--ipc" not in argv or _flag_value(argv, "--ipc") != "host"
assert "--userns" not in argv or "host" not in _flag_value(argv, "--userns")
def test_build_docker_argv_scratch_is_writable_tmpfs(tmp_path: Path) -> None:
argv = build_docker_argv(_job(tmp_path))
tmpfs = _flag_value(argv, "--tmpfs")
assert tmpfs.startswith("/scratch")
assert "rw" in tmpfs
assert "exec" in tmpfs
def test_build_docker_argv_applies_memory_and_pids_limits(tmp_path: Path) -> None:
job = _job(tmp_path, memory_mb=512, pids_limit=64)
argv = build_docker_argv(job)
memory = _flag_value(argv, "--memory")
assert "512" in memory
assert memory.endswith("m") or memory.endswith("M") or memory.endswith("b")
assert _flag_value(argv, "--pids-limit") == "64"
def test_build_docker_argv_drops_capabilities_and_new_privileges(tmp_path: Path) -> None:
argv = build_docker_argv(_job(tmp_path))
assert _flag_value(argv, "--cap-drop") == "ALL"
security = _flag_value(argv, "--security-opt")
assert "no-new-privileges" in security
def test_build_docker_argv_uses_non_root_user(tmp_path: Path) -> None:
argv = build_docker_argv(_job(tmp_path))
user = _flag_value(argv, "--user")
uid = user.split(":", 1)[0]
assert uid not in {"0", "root"}
assert int(uid) != 0
def test_build_docker_argv_bind_mounts_repo_read_only_at_workspace(tmp_path: Path) -> None:
job = _job(tmp_path)
argv = build_docker_argv(job)
mount = _workspace_mount(argv)
source = mount.get("src") or mount.get("source")
assert source == str(job.repo_root.resolve())
assert mount.get("dst") == "/workspace" or mount.get("destination") == "/workspace" or mount.get("target") == "/workspace"
assert mount.get("ro") == "true" or mount.get("readonly") == "true"
assert "rw" not in mount
def test_build_docker_argv_source_tree_is_not_writable(tmp_path: Path) -> None:
argv = build_docker_argv(_job(tmp_path))
mount = _workspace_mount(argv)
assert mount.get("ro") == "true" or mount.get("readonly") == "true"
assert "--read-only" in argv
joined = " ".join(argv)
assert f"{_job(tmp_path).repo_root.resolve()}:/workspace:rw" not in joined
assert f"{_job(tmp_path).repo_root.resolve()}:/workspace " not in joined + " "
def test_build_docker_argv_ends_with_image_and_command(tmp_path: Path) -> None:
job = _job(tmp_path)
argv = build_docker_argv(job)
assert argv[-len(job.command) :] == list(job.command)
assert argv[-len(job.command) - 1] == job.image
def test_fake_sandbox_runner_records_jobs_and_returns_programmed_result(tmp_path: Path) -> None:
programmed = SandboxResult(exit_code=7, stdout="out", stderr="err", timed_out=False)
runner: SandboxRunner = FakeSandboxRunner(programmed)
job = _job(tmp_path)
result = runner.run(job)
assert result == programmed
assert isinstance(runner, FakeSandboxRunner)
assert runner.jobs == [job]
def test_resolve_runner_fake_via_env(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv(runner_mod.RUNNER_ENV, "fake")
assert isinstance(resolve_sandbox_runner(), FakeSandboxRunner)
def test_resolve_runner_defaults_to_docker_cli(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv(runner_mod.RUNNER_ENV, raising=False)
assert isinstance(resolve_sandbox_runner(), DockerCliRunner)
def test_fake_sandbox_runner_can_be_reprogrammed(tmp_path: Path) -> None:
runner = FakeSandboxRunner()
first = _job(tmp_path, command=("echo", "one"))
second = _job(tmp_path, command=("echo", "two"))
timeout_result = SandboxResult(exit_code=-1, stdout="", stderr="deadline", timed_out=True)
runner.result = timeout_result
assert runner.run(first).timed_out is True
runner.result = SandboxResult(exit_code=0, stdout="ok", stderr="", timed_out=False)
assert runner.run(second).stdout == "ok"
assert runner.jobs == [first, second]
def test_fake_sandbox_runner_does_not_need_docker_daemon(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
def boom(*_args: object, **_kwargs: object) -> subprocess.CompletedProcess[str]:
raise AssertionError("unit tests must not invoke the docker daemon")
monkeypatch.setattr(subprocess, "run", boom)
monkeypatch.setattr(runner_mod.subprocess, "run", boom)
programmed = SandboxResult(exit_code=0, stdout="fake", stderr="", timed_out=False)
runner = FakeSandboxRunner(programmed)
assert runner.run(_job(tmp_path)) == programmed
def test_docker_cli_runner_uses_constructed_argv_and_timeout(tmp_path: Path) -> None:
seen: dict[str, Any] = {}
job = _job(tmp_path, scratch_files={}, timeout_seconds=19)
def fake_run(command: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]:
seen["command"] = list(command)
seen["timeout"] = kwargs.get("timeout")
seen["check"] = kwargs.get("check")
seen["capture_output"] = kwargs.get("capture_output")
seen["text"] = kwargs.get("text")
return subprocess.CompletedProcess(command, 3, stdout="hello", stderr="warn")
result = DockerCliRunner(runner=fake_run).run(job)
assert seen["command"] == build_docker_argv(job)
assert seen["timeout"] == 19
assert seen["check"] is False
assert seen["capture_output"] is True
assert seen["text"] is True
assert result == SandboxResult(exit_code=3, stdout="hello", stderr="warn", timed_out=False)
assert "--network" in seen["command"]
assert "--read-only" in seen["command"]
assert "docker.sock" not in " ".join(seen["command"])
def test_docker_cli_runner_maps_timeout_expired_to_timed_out(tmp_path: Path) -> None:
job = _job(tmp_path, scratch_files={}, timeout_seconds=5)
def fake_run(command: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[str]:
raise subprocess.TimeoutExpired(cmd=command, timeout=5, output="partial", stderr="slow")
result = DockerCliRunner(runner=fake_run).run(job)
assert result.timed_out is True
assert result.stdout == "partial"
assert result.stderr == "slow"
def test_docker_cli_runner_does_not_invoke_daemon_when_runner_injected(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
def boom(*_args: object, **_kwargs: object) -> subprocess.CompletedProcess[str]:
raise AssertionError("unit tests must not invoke the docker daemon")
monkeypatch.setattr(subprocess, "run", boom)
monkeypatch.setattr(runner_mod.subprocess, "run", boom)
def fake_run(command: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[str]:
return subprocess.CompletedProcess(command, 0, stdout="", stderr="")
result = DockerCliRunner(runner=fake_run).run(_job(tmp_path, scratch_files={}))
assert result.timed_out is False
assert result.exit_code == 0
def test_docker_cli_runner_does_not_write_the_source_tree(tmp_path: Path) -> None:
job = _job(tmp_path)
original = (job.repo_root / "app.py").read_text()
before = {path.relative_to(job.repo_root): path.read_text() for path in job.repo_root.rglob("*") if path.is_file()}
def fake_run(command: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[str]:
joined = " ".join(command)
assert "docker.sock" not in joined
assert "--read-only" in command
return subprocess.CompletedProcess(command, 0, stdout="", stderr="")
DockerCliRunner(runner=fake_run).run(job)
assert (job.repo_root / "app.py").read_text() == original
after = {path.relative_to(job.repo_root): path.read_text() for path in job.repo_root.rglob("*") if path.is_file()}
assert after == before
assert "case.py" not in {path.name for path in job.repo_root.iterdir()}
def test_docker_cli_runner_seeds_scratch_files_outside_the_repo(tmp_path: Path) -> None:
job = _job(tmp_path, scratch_files={"nested/case.py": "print('seeded')\n"})
seen: dict[str, Any] = {}
def fake_run(command: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[str]:
seen["command"] = list(command)
mounts = [command[index + 1] for index, item in enumerate(command) if item == "--mount" and index + 1 < len(command)]
seed_mounts = [spec for spec in mounts if "/scratch-in" in spec]
assert seed_mounts, command
options = _mount_options(seed_mounts[0])
source = Path(options.get("src") or options.get("source") or "")
assert source.is_dir()
assert job.repo_root.resolve() not in source.resolve().parents
assert source.resolve() != job.repo_root.resolve()
assert (source / "nested" / "case.py").read_text() == "print('seeded')\n"
assert options.get("ro") == "true" or options.get("readonly") == "true"
assert "--tmpfs" in command
assert _flag_value(command, "--tmpfs").startswith("/scratch")
return subprocess.CompletedProcess(command, 0, stdout="", stderr="")
result = DockerCliRunner(runner=fake_run).run(job)
assert result.exit_code == 0
argv = seen["command"]
assert "docker.sock" not in " ".join(argv)
assert "--read-only" in argv
assert job.image in argv
def test_docker_cli_runner_scratch_seed_is_world_readable_for_container_user(tmp_path: Path) -> None:
job = _job(tmp_path, scratch_files={"nested/dir/case.py": "print('seeded')\n"})
seen: dict[str, Path] = {}
def fake_run(command: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[str]:
source = _seed_source(command)
seen["source"] = source
assert _posix_mode(source) == 0o755
for dirpath, _dirnames, filenames in os.walk(source):
directory = Path(dirpath)
assert stat.S_ISDIR(directory.stat().st_mode)
assert _posix_mode(directory) == 0o755, directory
for name in filenames:
file_path = directory / name
assert _posix_mode(file_path) == 0o644, file_path
return subprocess.CompletedProcess(command, 0, stdout="", stderr="")
DockerCliRunner(runner=fake_run).run(job)
assert "source" in seen
def test_docker_cli_runner_scratch_copy_uses_cp_r_not_cp_a(tmp_path: Path) -> None:
job = _job(tmp_path, scratch_files={"nested/case.py": "print('seeded')\n"})
seen: dict[str, str] = {}
def fake_run(command: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[str]:
script = _copy_script(command)
seen["script"] = script
assert command[command.index(job.image) + 1] == "/bin/sh"
assert "cp -a" not in script
assert "cp -a" not in " ".join(command)
assert script.startswith("cp -r /scratch-in/. /scratch/")
assert 'exec "$@"' in script
return subprocess.CompletedProcess(command, 0, stdout="", stderr="")
DockerCliRunner(runner=fake_run).run(job)
assert seen["script"] == 'cp -r /scratch-in/. /scratch/ && exec "$@"'
tests/test_scan_deep.py
-197
View File
@@ -1,197 +0,0 @@
"""Deep mode: skip regression when the sandbox is missing; record verdicts with a fake runner (task 6.3)."""
from __future__ import annotations
import json
import subprocess
from pathlib import Path
import pytest
from openultrasast.cli import main
from openultrasast.sandbox import FakeSandboxRunner, SandboxResult
from openultrasast.sandbox import runner as sandbox_runner
_VULN = "@app.route('/admin')\ndef admin():\n return eval(request.data)\n"
def _repo(tmp_path: Path) -> Path:
repo = tmp_path / "repo"
repo.mkdir()
(repo / "app.py").write_text(_VULN)
return repo
def _fail(message: str): # type: ignore[no-untyped-def]
def _inner(*args: object, **kwargs: object) -> object:
raise AssertionError(message)
return _inner
def _guard_docker(monkeypatch: pytest.MonkeyPatch) -> list[list[str]]:
docker_argv: list[list[str]] = []
original_run = subprocess.run
def guarded_run(command: object, *args: object, **kwargs: object) -> object:
argv = list(command) if isinstance(command, list | tuple) else [command]
if argv and Path(str(argv[0])).name == "docker":
docker_argv.append([str(item) for item in argv])
raise AssertionError(f"deep scan test must not start docker: {argv}")
return original_run(command, *args, **kwargs)
monkeypatch.setattr(subprocess, "run", guarded_run)
monkeypatch.setattr(sandbox_runner.DockerCliRunner, "run", _fail("deep scan test must not use DockerCliRunner"))
return docker_argv
def _latest_run(repo: Path) -> Path:
return sorted((repo / ".runs").iterdir())[-1]
def _crash_runner() -> FakeSandboxRunner:
return FakeSandboxRunner(SandboxResult(exit_code=139, stdout="", stderr="Segmentation fault\n", timed_out=False))
def _enable_crash_sandbox(monkeypatch: pytest.MonkeyPatch) -> FakeSandboxRunner:
from openultrasast import cli
fake = _crash_runner()
monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs")
monkeypatch.setenv("OPENULTRASAST_SANDBOX_PROBE", "1")
monkeypatch.setattr(cli, "resolve_sandbox_runner", lambda: fake)
return fake
def test_deep_scan_skips_regress_when_sandbox_unavailable(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
repo = _repo(tmp_path)
monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs")
monkeypatch.setenv("OPENULTRASAST_SANDBOX_PROBE", "0")
docker_argv = _guard_docker(monkeypatch)
assert main(["scan", str(repo), "--mode", "deep"]) == 0
run_dir = _latest_run(repo)
manifest = json.loads((run_dir / "manifest.json").read_text())
stages = manifest["stages"]
degradations = manifest.get("degradations", [])
assert (run_dir / "complexity_map.json").is_file()
assert not (run_dir / "verdicts.json").exists()
assert docker_argv == []
assert stages["requested"] == ["static", "map", "regress"]
assert stages["completed"] == ["static", "map"]
assert {"stage": "regress", "reason": "sandbox_unavailable"} in stages["skipped"]
assert any(entry.get("stage") == "regress" and entry.get("reason") == "sandbox_unavailable" for entry in degradations)
def test_deep_scan_with_fake_runner_writes_verdicts(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
repo = _repo(tmp_path)
monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs")
monkeypatch.setenv("OPENULTRASAST_SANDBOX_PROBE", "1")
monkeypatch.setenv("OPENULTRASAST_SANDBOX_RUNNER", "fake")
docker_argv = _guard_docker(monkeypatch)
assert main(["scan", str(repo), "--mode", "deep"]) == 0
run_dir = _latest_run(repo)
verdicts_path = run_dir / "verdicts.json"
manifest = json.loads((run_dir / "manifest.json").read_text())
payload = json.loads(verdicts_path.read_text())
stages = manifest["stages"]
assert docker_argv == []
assert verdicts_path.is_file()
assert (run_dir / "complexity_map.json").is_file()
assert stages["requested"] == ["static", "map", "regress"]
assert stages["completed"] == ["static", "map", "regress"]
assert not any(entry.get("reason") == "sandbox_unavailable" for entry in stages["skipped"])
assert isinstance(payload["verdicts"], list)
assert payload["verdicts"]
for item in payload["verdicts"]:
assert item["verdict"] in {"triggerable", "not_triggerable", "already_covered", "inconclusive"}
assert "reason" in item
assert "path" in item
assert manifest["artifacts"]["verdicts"] == "verdicts.json"
assert manifest["artifacts"]["complexity_map"] == "complexity_map.json"
assert manifest["worth_fixing"]["count"] == sum(1 for item in payload["verdicts"] if item.get("worth_fixing"))
report = (run_dir / "report.md").read_text()
inventory_at = report.index("## Inventory")
map_at = report.index("## Complexity map")
worth_at = report.index("## Worth fixing")
assert inventory_at < map_at < worth_at
map_section = report[map_at:worth_at]
assert "app.py" in map_section
assert "Gap:" in map_section
ledger_path = repo / ".openultrasast" / "calibration" / "complexity_ledger.json"
assert ledger_path.is_file()
ledger = json.loads(ledger_path.read_text())
assert ledger
assert any(entry.get("last_verdict") in {"triggerable", "not_triggerable"} for entry in ledger.values())
def test_deep_scan_unsafe_hunter_snippet_is_safety_rejected_without_docker_job(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
from openultrasast import cli
from openultrasast.sandbox import FakeSandboxRunner
repo = _repo(tmp_path)
config = tmp_path / "openultrasast.toml"
config.write_text('[models]\nhunter = "test-hunter"\n')
fake = FakeSandboxRunner()
monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs")
monkeypatch.setenv("OPENULTRASAST_SANDBOX_PROBE", "1")
monkeypatch.setenv("OPENULTRASAST_HUNTER_CLIENT", "unsafe-snippet")
monkeypatch.setattr(cli, "resolve_sandbox_runner", lambda: fake)
docker_argv = _guard_docker(monkeypatch)
assert main(["scan", str(repo), "--mode", "deep", "--config", str(config)]) == 0
run_dir = _latest_run(repo)
payload = json.loads((run_dir / "verdicts.json").read_text())
findings = json.loads((run_dir / "findings.json").read_text())["findings"]
hunter_findings = [item for item in findings if str(item["finding_id"]).startswith("tool-hunter:")]
assert docker_argv == []
assert hunter_findings
assert all(item["evidence_level"] == "suspicion" for item in hunter_findings)
hunter_ids = {item["finding_id"] for item in hunter_findings}
# Overlay promotions may still be proven; hunter suspicion never becomes a sandbox candidate.
for item in payload["verdicts"]:
assert hunter_ids.isdisjoint(item["inventory_finding_ids"])
def test_deep_scan_fail_on_worth_fixing_exits_one_when_triggerable(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
repo = _repo(tmp_path)
_enable_crash_sandbox(monkeypatch)
docker_argv = _guard_docker(monkeypatch)
assert main(["scan", str(repo), "--mode", "deep", "--fail-on", "worth-fixing"]) == 1
assert docker_argv == []
run_dir = _latest_run(repo)
payload = json.loads((run_dir / "verdicts.json").read_text())
manifest = json.loads((run_dir / "manifest.json").read_text())
report = (run_dir / "report.md").read_text()
worth_fixing = [item for item in payload["verdicts"] if item.get("worth_fixing")]
assert worth_fixing
assert len(worth_fixing) >= 1
assert any(item["verdict"] == "triggerable" for item in worth_fixing)
assert manifest["worth_fixing"]["count"] == len(worth_fixing)
assert "## Complexity map" in report
assert "## Worth fixing" in report
assert "triggerable" in report
ledger = json.loads((repo / ".openultrasast" / "calibration" / "complexity_ledger.json").read_text())
assert any(entry.get("last_verdict") == "triggerable" for entry in ledger.values())
def test_deep_scan_fail_on_never_exits_zero_with_worth_fixing_verdict(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
repo = _repo(tmp_path)
_enable_crash_sandbox(monkeypatch)
docker_argv = _guard_docker(monkeypatch)
assert main(["scan", str(repo), "--mode", "deep", "--fail-on", "never"]) == 0
assert docker_argv == []
payload = json.loads((_latest_run(repo) / "verdicts.json").read_text())
assert any(item.get("worth_fixing") for item in payload["verdicts"])
tests/test_search_demo_replay.py
-7
View File
@@ -196,13 +196,6 @@ def test_lane_failure_is_could_not_run(tmp_path, monkeypatch):
assert "verifier unavailable" in record["reason"]
def test_replay_registered_as_standalone():
from openultrasast.model.audit import audit
rows = {r.module: r for r in audit(Path("benchmarks"))}
assert rows["search.demo_replay"].classification == "standalone_capability"
def test_docker_selection_uses_same_preparation_and_fresh_runs(tmp_path, monkeypatch):
args, root, fake_lane, events = harness(tmp_path, monkeypatch)
args.lane = "docker"
tests/test_semantic_pipeline.py
-40
View File
@@ -125,46 +125,6 @@ def test_prove_filter_drops_demoted_keeps_promoted() -> None:
assert [item.finding_id for item in promoted_findings(findings, records)] == [promoted_id]
def test_deep_scan_offers_only_promotions(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
from openultrasast import cli
from openultrasast.sandbox import FakeSandboxRunner, SandboxResult
repo = tmp_path / "repo"
repo.mkdir()
(repo / "app.py").write_text("def safe():\n exec('ls')\n\ndef bad():\n return eval(request.args)\n")
monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs")
monkeypatch.setenv("OPENULTRASAST_SANDBOX_PROBE", "1")
fake = FakeSandboxRunner(SandboxResult(exit_code=0, stdout="", stderr="", timed_out=False))
monkeypatch.setattr(cli, "resolve_sandbox_runner", lambda: fake)
assert main(["scan", str(repo), "--mode", "deep", "--fail-on", "worth-fixing"]) == 0
run_dir = _latest_run(repo)
overlay = json.loads((run_dir / "overlay.json").read_text())["records"]
dispositions = {item["proposal_id"]: item["disposition"] for item in overlay}
demoted = [key for key, value in dispositions.items() if value == "demote"]
promoted = [key for key, value in dispositions.items() if value == "promote"]
assert demoted
assert promoted
verdicts = json.loads((run_dir / "verdicts.json").read_text())["verdicts"]
offered = {finding_id for item in verdicts for finding_id in item["inventory_finding_ids"]}
assert offered <= set(promoted)
assert not (offered & set(demoted))
def test_fail_on_worth_fixing_ignores_unadjudicated(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
repo = tmp_path / "repo"
repo.mkdir()
(repo / "broken.py").write_text("eval(request.args)\ndef oops(\n")
monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs")
monkeypatch.setenv("OPENULTRASAST_SANDBOX_PROBE", "0")
assert main(["scan", str(repo), "--mode", "standard", "--fail-on", "worth-fixing"]) == 0
run_dir = _latest_run(repo)
overlay = json.loads((run_dir / "overlay.json").read_text())["records"]
assert overlay
assert all(item["disposition"] == "unadjudicated" for item in overlay)
report = (run_dir / "report.md").read_text()
assert "unadjudicated" in report
def test_unlabeled_messy_tree_has_both_dispositions(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
repo = tmp_path / "repo"
repo.mkdir()
tests/test_skills.py
-85
View File
@@ -1,85 +0,0 @@
"""Trail of Bits skill index + router (Phase 14)."""
from openultrasast.preprocess import FileTarget
from openultrasast.skills import (
SKILL_INVENTORY,
route_skill_ids,
route_skills,
route_skills_for_target,
select_skill_context,
skill_chunks,
)
def _target(language: str, tags: list[str]) -> FileTarget:
return FileTarget(path="t", absolute_path="/repo/t", language=language, loc=10, tags=tags, has_fuzz_entry_point=False)
# ---- routing by language / tags / vulnerability class -----------------------
def test_c_parser_target_routes_to_memory_and_fuzzing_skills() -> None:
ids = route_skill_ids(_target("c", ["memory_unsafe", "parser"]))
assert {"address-sanitizer", "harness-writing"} <= set(ids) # memory + fuzzing
assert "libfuzzer" in ids
def test_crypto_target_routes_to_constant_time_and_vector_skills() -> None:
ids = route_skill_ids(_target("python", ["crypto"]))
assert "constant-time-analysis" in ids
assert "vector-forge" in ids
def test_language_mismatch_excludes_other_language_skills() -> None:
# A C target must not pull Python-only (atheris) or Rust-only (cargo-fuzz) skills.
ids = route_skill_ids(_target("c", ["memory_unsafe", "parser", "fuzzable"]))
assert "atheris" not in ids
assert "cargo-fuzz" not in ids
def test_language_only_match_does_not_select() -> None:
# python + crypto must not surface atheris (python but no crypto signal).
ids = route_skills_for_target(_target("python", ["crypto"]), max_snippets=10)
assert "atheris" not in [skill.id for skill in ids]
# ---- routing by stage -------------------------------------------------------
def test_stage_filter_restricts_to_stage_skills() -> None:
# fuzzing-dictionary is verify-only; at the map stage it must not appear, while a
# map-stage skill (semgrep) does for an injection signal.
verify = route_skills(tags=("parser",), vulnerability_classes=("parsing",), stage="verify", max_snippets=10)
mapped = route_skills(tags=("parser",), vulnerability_classes=("parsing",), stage="map", max_snippets=10)
assert "fuzzing-dictionary" in {s.id for s in verify}
assert "fuzzing-dictionary" not in {s.id for s in mapped}
# ---- prompt budget caps -----------------------------------------------------
def test_select_skill_context_respects_budget() -> None:
target = _target("c", ["memory_unsafe", "parser"])
full = select_skill_context(target, budget_chars=10_000)
capped = select_skill_context(target, budget_chars=80)
assert len(capped) <= 80
assert len(full) > len(capped)
def test_zero_budget_yields_no_context() -> None:
assert select_skill_context(_target("c", ["memory_unsafe"]), budget_chars=0) == ""
# ---- retrieval-namespace chunking + inventory integrity ---------------------
def test_skill_chunks_land_in_the_skills_namespace() -> None:
chunks = skill_chunks()
assert chunks
assert all(chunk.namespace == "skills" for chunk in chunks)
assert any(chunk.metadata.get("skill_id") == "address-sanitizer" for chunk in chunks)
def test_inventory_ids_are_unique() -> None:
ids = [skill.id for skill in SKILL_INVENTORY]
assert len(ids) == len(set(ids))
tests/test_split_sink_regress.py
-40
View File
@@ -1,40 +0,0 @@
"""Stage-3 regression on split-sink: fake runner always; live Docker skips without a daemon."""
from pathlib import Path
import pytest
from openultrasast.config import SandboxConfig
from openultrasast.regress.recipes import recipe_for
from openultrasast.regress.safety import check_snippet_safety
from openultrasast.regress.verdict import TRIGGERABLE, verdict_from_result
from openultrasast.sandbox import DockerCliRunner, FakeSandboxRunner, SandboxProbe, SandboxResult
_SPLIT_SINK_PYTHON = Path("benchmarks/fixtures/split-sink-python")
_LIMITS = SandboxConfig(memory_mb=256, timeout_seconds=30, pids_limit=128)
_PYTHON_IMAGE = "python:3.12-alpine"
_SNIPPET = 'term = "x\'; drop"\nquery = "select * from items where title like \'%" + term + "%\'"\nassert "\'" not in query\n'
def _split_sink_job():
check_snippet_safety(_SNIPPET)
job = recipe_for("python", _SNIPPET, _PYTHON_IMAGE, _LIMITS, repo_root=_SPLIT_SINK_PYTHON.resolve())
assert job is not None
return job
def test_fake_runner_marks_split_sink_snippet_triggerable() -> None:
fake = FakeSandboxRunner(SandboxResult(exit_code=1, stdout="", stderr="AssertionError", timed_out=False))
result = fake.run(_split_sink_job())
verdict = verdict_from_result(result)
assert verdict.verdict == TRIGGERABLE
assert fake.jobs
@pytest.mark.docker
def test_docker_marks_split_sink_python_triggerable() -> None:
if not SandboxProbe().available():
pytest.skip("docker daemon unavailable")
result = DockerCliRunner().run(_split_sink_job())
verdict = verdict_from_result(result)
assert verdict.verdict == TRIGGERABLE, (result.exit_code, result.stdout, result.stderr)
tests/test_stages.py
+4 -4
View File
@@ -36,15 +36,15 @@ def test_plan_for_mode_requests_ordered_stages() -> None:
def test_skipped_stage_is_structured_degradation_not_exception() -> None:
plan = plan_for_mode("deep")
skipped = record_skip(plan, Stage.REGRESS, "sandbox_unavailable")
skipped = record_skip(plan, Stage.REGRESS, "regress_retired")
assert skipped.requested == (Stage.STATIC, Stage.MAP, Stage.REGRESS)
assert skipped.completed == ()
assert skipped.skipped == ({"stage": "regress", "reason": "sandbox_unavailable"},)
assert skipped.skipped == ({"stage": "regress", "reason": "regress_retired"},)
degradations = list(skipped.skipped)
assert degradations[0]["stage"] == "regress"
assert degradations[0]["reason"] == "sandbox_unavailable"
assert json.dumps(degradations) == json.dumps([{"stage": "regress", "reason": "sandbox_unavailable"}])
assert degradations[0]["reason"] == "regress_retired"
assert json.dumps(degradations) == json.dumps([{"stage": "regress", "reason": "regress_retired"}])
def test_record_completed_static_for_quick() -> None:
tests/test_tool_hunter.py
-6
View File
@@ -198,8 +198,6 @@ def _latest_run(repo: Path) -> Path:
def _guard_docker(monkeypatch: pytest.MonkeyPatch) -> list[list[str]]:
import subprocess
from openultrasast.sandbox import runner as sandbox_runner
docker_argv: list[list[str]] = []
original_run = subprocess.run
@@ -210,11 +208,7 @@ def _guard_docker(monkeypatch: pytest.MonkeyPatch) -> list[list[str]]:
raise AssertionError(f"hunter scan test must not start docker: {argv}")
return original_run(command, *args, **kwargs)
def fail(*args: object, **kwargs: object) -> object:
raise AssertionError("hunter scan test must not use DockerCliRunner")
monkeypatch.setattr(subprocess, "run", guarded_run)
monkeypatch.setattr(sandbox_runner.DockerCliRunner, "run", fail)
return docker_argv