feat(three-stage-scan): add fail-on worth-fixing

This commit is contained in:
Marius Ciepluch
2026-09-03 19:44:04 +02:00
parent 926c8c273e
commit c5a80c2817
5 changed files with 82 additions and 5 deletions
+8 -2
View File
@@ -4,7 +4,7 @@ This plan ships the three-stage scan without rebuilding CWE policy or the regex
## Phase 1 — Foundation: Stage contract ## Phase 1 — Foundation: Stage contract
- [ ] 1. Introduce the stage plan and mode mapping - [x] 1. Introduce the stage plan and mode mapping
- [x] 1.1 Encode the three stages and the mode-to-stage table - [x] 1.1 Encode the three stages and the mode-to-stage table
- Depth `quick` requests only static, `standard` requests static then map, `deep` requests static then map then regress. - Depth `quick` requests only static, `standard` requests static then map, `deep` requests static then map then regress.
- A skipped stage is a structured reason, not an exception, and can be serialized into the existing degradations list. - A skipped stage is a structured reason, not an exception, and can be serialized into the existing degradations list.
@@ -20,7 +20,7 @@ This plan ships the three-stage scan without rebuilding CWE policy or the regex
- _Boundary: Packaging Configuration_ - _Boundary: Packaging Configuration_
- _Depends: 1.1_ - _Depends: 1.1_
- [ ] 1.3 Expand fail-on vocabulary with worth-fixing - [x] 1.3 Expand fail-on vocabulary with worth-fixing
- The scan command accepts `worth-fixing` alongside the existing fail-on values and does not change the meaning of `findings` or `verified`. - The scan command accepts `worth-fixing` alongside the existing fail-on values and does not change the meaning of `findings` or `verified`.
- The existing report exit helper must recognize the new value; unknown values still fail loud. - The existing report exit helper must recognize the new value; unknown values still fail loud.
- MCP continues to reject `deep` and must not gain docker or shell tools. - MCP continues to reject `deep` and must not gain docker or shell tools.
@@ -191,3 +191,9 @@ This plan ships the three-stage scan without rebuilding CWE policy or the regex
- Observable completion: fake-runner tests always run; docker tests skip-or-pass; the default CI workflow still invokes quick-only detection plus the new split-sink map gate. - Observable completion: fake-runner tests always run; docker tests skip-or-pass; the default CI workflow still invokes quick-only detection plus the new split-sink map gate.
- _Requirements: 8.5, 9.1, 9.2_ - _Requirements: 8.5, 9.1, 9.2_
- _Depends: 6.3, 7.1_ - _Depends: 6.3, 7.1_
## Implementation Notes
- `RegressConfig.images` is a sorted tuple of `(language, image)` pairs so frozen config stays hashable.
- `scan_exit_code` duck-types `worth_fixing` on verdict objects; `_run_scan` still passes no verdicts until task 6.5. Dict-shaped verdicts would not fail the gate.
- Map `top_k` default is 20 (not named in design.md); `max_hunter_hotspots=8` and `max_candidates=5` follow design.
+1 -1
View File
@@ -69,7 +69,7 @@ def main(argv: list[str] | None = None) -> int:
scan.add_argument("path", type=Path) scan.add_argument("path", type=Path)
scan.add_argument("--mode", choices=("quick", "standard", "deep"), default="quick") scan.add_argument("--mode", choices=("quick", "standard", "deep"), default="quick")
scan.add_argument("--config", type=Path, default=Path("openultrasast.toml")) scan.add_argument("--config", type=Path, default=Path("openultrasast.toml"))
scan.add_argument("--fail-on", choices=("never", "findings", "verified"), default="never") scan.add_argument("--fail-on", choices=("never", "findings", "verified", "worth-fixing"), default="never")
index = subparsers.add_parser("index", help="chunk a local repository for embedding index construction") index = subparsers.add_parser("index", help="chunk a local repository for embedding index construction")
index.add_argument("path", type=Path) index.add_argument("path", type=Path)
+9 -1
View File
@@ -1,6 +1,7 @@
from __future__ import annotations from __future__ import annotations
import json import json
from collections.abc import Sequence
from dataclasses import asdict from dataclasses import asdict
from pathlib import Path from pathlib import Path
@@ -112,13 +113,20 @@ def write_manifest(
path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
def scan_exit_code(findings: list[StaticFinding], verifications: list[VerificationResult], policy: str) -> int: def scan_exit_code(
findings: list[StaticFinding],
verifications: list[VerificationResult],
policy: str,
worth_fixing_verdicts: Sequence[object] | None = None,
) -> int:
if policy == "never": if policy == "never":
return 0 return 0
if policy == "findings": if policy == "findings":
return 1 if findings else 0 return 1 if findings else 0
if policy == "verified": if policy == "verified":
return 1 if any(result.verified for result in verifications) else 0 return 1 if any(result.verified for result in verifications) else 0
if policy == "worth-fixing":
return 1 if any(getattr(verdict, "worth_fixing", False) for verdict in worth_fixing_verdicts or ()) else 0
raise ValueError(f"unknown fail policy: {policy}") raise ValueError(f"unknown fail policy: {policy}")
+18 -1
View File
@@ -54,7 +54,24 @@ def test_surface_exposes_no_shell_or_exec_tool() -> None:
for name in server.tool_names: for name in server.tool_names:
assert not any(token in name.lower() for token in forbidden), name assert not any(token in name.lower() for token in forbidden), name
for tool in McpServer()._tools.values(): # no tool accepts a free-form command argument for tool in McpServer()._tools.values(): # no tool accepts a free-form command argument
assert "command" not in tool.input_schema.get("properties", {}) properties = tool.input_schema.get("properties", {})
assert "command" not in properties
assert "shell" not in properties
assert "docker" not in properties
def test_scan_rejects_deep_and_keeps_cli_only_modes() -> None:
server = McpServer()
listed = server.handle({"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}})
assert listed is not None
scan_tool = next(tool for tool in listed["result"]["tools"] if tool["name"] == "openultrasast.scan")
mode_enum = scan_tool["inputSchema"]["properties"]["mode"]["enum"]
assert mode_enum == ["quick", "standard"]
assert "deep" not in mode_enum
result = _call(server, "openultrasast.scan", {"path": ".", "mode": "deep"})
assert result["isError"]
assert "deep" in result["error"]
def test_notifications_get_no_response() -> None: def test_notifications_get_no_response() -> None:
+46
View File
@@ -1,5 +1,8 @@
import json import json
from pathlib import Path from pathlib import Path
from types import SimpleNamespace
import pytest
from openultrasast.cli import main from openultrasast.cli import main
from openultrasast.findings import StaticFinding from openultrasast.findings import StaticFinding
@@ -63,11 +66,45 @@ def test_markdown_report_includes_verification_status(tmp_path: Path) -> None:
def test_scan_exit_code_policy() -> None: def test_scan_exit_code_policy() -> None:
finding = _finding() finding = _finding()
verification = verify_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], "never") == 0
assert scan_exit_code([finding], [verification], "findings") == 1 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([finding], [verification], "verified") == 1
assert scan_exit_code([], [], "verified") == 0 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] def test_cli_scan_writes_sarif_manifest_and_honors_fail_policy(tmp_path: Path, monkeypatch) -> None: # type: ignore[no-untyped-def]
@@ -87,6 +124,15 @@ 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" 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 _finding() -> StaticFinding: def _finding() -> StaticFinding:
return StaticFinding( return StaticFinding(
finding_id="python-unsafe-eval:app.py:3", finding_id="python-unsafe-eval:app.py:3",