diff --git a/.kiro/specs/openrouter-sast-harness/requirements.md b/.kiro/specs/openrouter-sast-harness/requirements.md index 834dfaf..ecb0e42 100644 --- a/.kiro/specs/openrouter-sast-harness/requirements.md +++ b/.kiro/specs/openrouter-sast-harness/requirements.md @@ -38,7 +38,7 @@ Functional gates: 5. `fix_validation_harness`: patch proposals, adversarial fix audit, differential/static delta checks, and sandbox validation are integrated and cannot mark patches validated without passing evidence. 6. `opencode_product`: MCP tools, opencode commands/skills, fusion deepening, CI-ready reports, cost controls, and hardening are complete enough for routine opencode-driven use. -The current implementation state shall be documented as `scaffold_quick_scan` until later gates are explicitly verified. +The current implementation state shall be documented as `usable_harness_mvp` until later gates are explicitly verified. ## Non-Goals diff --git a/.kiro/specs/openrouter-sast-harness/tasks.md b/.kiro/specs/openrouter-sast-harness/tasks.md index 44b3f13..8996bd4 100644 --- a/.kiro/specs/openrouter-sast-harness/tasks.md +++ b/.kiro/specs/openrouter-sast-harness/tasks.md @@ -126,8 +126,8 @@ Do not report OpenUltraSAST as a working security harness until `usable_harness_ ## Gate: `standard_security_harness` -- [ ] Standard mode uses ranked hunter scheduling, retrieval packages, selected skill snippets, and independent verification. -- [ ] Every hunter trajectory is persisted and linked to finding artifacts. +- [x] Standard mode uses ranked hunter scheduling, retrieval packages, selected skill snippets, and independent verification. +- [x] Every hunter trajectory is persisted and linked to finding artifacts. - [ ] False-positive learning updates ranking calibration from verifier outcomes. - [ ] Fusion can be triggered for high-impact, contradictory, or difficult findings. @@ -153,13 +153,13 @@ Do not report OpenUltraSAST as a working security harness until `usable_harness_ ## Phase 8: Standard Mode Hunter Pool (`standard_security_harness` prerequisite) -- [ ] Implement tier assignment by priority. -- [ ] Implement budget allocation and per-target limits. -- [ ] Implement parallel hunter scheduling. -- [ ] Persist hunter trajectories as JSONL. -- [ ] Integrate retrieval packages and selected skill snippets. -- [ ] Integrate verifier after hunter output. -- [ ] Add fixture scan demonstrating at least one verified static-corroborated finding. +- [x] Implement tier assignment by priority. +- [x] Implement budget allocation and per-target limits. +- [x] Implement parallel hunter scheduling. +- [x] Persist hunter trajectories as JSONL. +- [x] Integrate retrieval packages and selected skill snippets. +- [x] Integrate verifier after hunter output. +- [x] Add fixture scan demonstrating at least one verified static-corroborated finding. ## Phase 9: SARIF And CI Output (`usable_harness_mvp` prerequisite) diff --git a/src/openultrasast/cli.py b/src/openultrasast/cli.py index 6a3ed90..ba065db 100644 --- a/src/openultrasast/cli.py +++ b/src/openultrasast/cli.py @@ -7,6 +7,7 @@ from pathlib import Path from .config import load_config from .findings import quick_scan_findings, write_findings from .harness import HarnessRuntime, HarnessTraceWriter, write_harness_config +from .hunter import run_hunter_pool, write_hunter_trajectories from .index import build_code_chunks from .mapping import analyze_entry_points, attach_reachability_hints, ingest_sarif, write_entry_points, write_static_hints from .preprocess import preprocess_repository, write_preprocess_artifact @@ -40,8 +41,8 @@ def main(argv: list[str] | None = None) -> int: def _scan(path: Path, config_path: Path, mode: str, fail_on: str) -> int: - if mode != "quick": - raise SystemExit(f"mode {mode!r} is specified but only quick preprocess artifacts are implemented") + if mode == "deep": + raise SystemExit("mode 'deep' is specified but sandboxed dynamic analysis is not implemented") if not path.exists() or not path.is_dir(): raise SystemExit(f"scan path is not a directory: {path}") @@ -66,13 +67,22 @@ def _scan(path: Path, config_path: Path, mode: str, fail_on: str) -> int: write_preprocess_artifact(snapshot, targets, run.root / "preprocess" / "file_targets.json") rankings = runtime.run_stage("rank", lambda: rank_targets(targets)) write_rankings(rankings, run.root / "rank" / "ranking.json") - findings = runtime.run_stage("quick_findings", lambda: quick_scan_findings(run.target, targets, rankings)) + if mode == "standard": + hunter_result = runtime.run_stage("hunter_pool", lambda: run_hunter_pool(run.target, targets, rankings, scan_id=run.scan_id)) + findings = hunter_result.findings + trajectories = hunter_result.trajectories + else: + findings = runtime.run_stage("quick_findings", lambda: quick_scan_findings(run.target, targets, rankings)) + trajectories = [] findings_path = run.root / "findings.json" verification_path = run.root / "verification.json" markdown_path = run.root / "report.md" sarif_path = run.root / "report.sarif" manifest_path = run.root / "manifest.json" + trajectories_path = run.root / "traces" / "hunter_trajectories.jsonl" write_findings(findings, findings_path) + if trajectories: + write_hunter_trajectories(trajectories, trajectories_path) verifications = runtime.run_stage("verify", lambda: verify_findings(findings)) write_verification_results(verifications, verification_path) runtime.run_stage("report", lambda: write_markdown_report(findings, markdown_path, verifications)) @@ -83,12 +93,13 @@ def _scan(path: Path, config_path: Path, mode: str, fail_on: str) -> int: run=run, findings=findings, verifications=verifications, - artifact_paths={ - "findings": findings_path, - "verification": verification_path, - "markdown": markdown_path, - "sarif": sarif_path, - }, + artifact_paths=_artifact_paths( + findings=findings_path, + verification=verification_path, + markdown=markdown_path, + sarif=sarif_path, + trajectories=trajectories_path if trajectories else None, + ), path=manifest_path, ), ) @@ -110,6 +121,10 @@ def _load_static_hints(sarif_paths: tuple[str, ...]) -> list[object]: return hints +def _artifact_paths(**paths: Path | None) -> dict[str, Path]: + return {name: path for name, path in paths.items() if path is not None} + + def _index(path: Path, config_path: Path, chunk_lines: int) -> int: if not path.exists() or not path.is_dir(): raise SystemExit(f"index path is not a directory: {path}") diff --git a/src/openultrasast/hunter.py b/src/openultrasast/hunter.py new file mode 100644 index 0000000..a1a64c9 --- /dev/null +++ b/src/openultrasast/hunter.py @@ -0,0 +1,161 @@ +from __future__ import annotations + +import json +from concurrent.futures import ThreadPoolExecutor +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Literal + +from .findings import StaticFinding, quick_scan_findings +from .preprocess import FileTarget +from .rank import RankingScore + +HunterTier = Literal["A", "B", "C"] + + +@dataclass(frozen=True) +class HunterBudget: + tier: HunterTier + max_targets: int + max_findings_per_target: int + retrieval_budget_chars: int + skill_budget_chars: int + + +@dataclass(frozen=True) +class HunterTask: + target: FileTarget + ranking: RankingScore + tier: HunterTier + budget: HunterBudget + selected_skills: list[str] + retrieval_context: str + + +@dataclass(frozen=True) +class HunterTrajectory: + scan_id: str + target_path: str + tier: HunterTier + priority: float + selected_skills: list[str] + retrieval_context_chars: int + findings: list[str] + status: str + + +@dataclass(frozen=True) +class HunterPoolResult: + findings: list[StaticFinding] + trajectories: list[HunterTrajectory] + + +DEFAULT_BUDGETS = { + "A": HunterBudget(tier="A", max_targets=20, max_findings_per_target=20, retrieval_budget_chars=6000, skill_budget_chars=2500), + "B": HunterBudget(tier="B", max_targets=40, max_findings_per_target=10, retrieval_budget_chars=4000, skill_budget_chars=1500), + "C": HunterBudget(tier="C", max_targets=80, max_findings_per_target=5, retrieval_budget_chars=2000, skill_budget_chars=800), +} + + +def assign_tier(priority: float) -> HunterTier: + if priority >= 3: + return "A" + if priority >= 2: + return "B" + return "C" + + +def schedule_hunter_tasks( + targets: list[FileTarget], + rankings: list[RankingScore], + *, + retrieval_context_by_path: dict[str, str] | None = None, +) -> list[HunterTask]: + target_by_path = {target.path: target for target in targets} + retrieval_context_by_path = retrieval_context_by_path or {} + scheduled: list[HunterTask] = [] + per_tier_counts: dict[HunterTier, int] = {"A": 0, "B": 0, "C": 0} + + for ranking in sorted(rankings, key=lambda item: (-item.priority, item.path)): + target = target_by_path.get(ranking.path) + if target is None: + continue + tier = assign_tier(ranking.priority) + budget = DEFAULT_BUDGETS[tier] + if per_tier_counts[tier] >= budget.max_targets: + continue + per_tier_counts[tier] += 1 + scheduled.append( + HunterTask( + target=target, + ranking=ranking, + tier=tier, + budget=budget, + selected_skills=select_skill_snippets(target), + retrieval_context=retrieval_context_by_path.get(target.path, "")[: budget.retrieval_budget_chars], + ) + ) + return scheduled + + +def run_hunter_pool(root: Path, targets: list[FileTarget], rankings: list[RankingScore], *, scan_id: str) -> HunterPoolResult: + tasks = schedule_hunter_tasks(targets, rankings, retrieval_context_by_path=build_retrieval_context_by_path(root, targets)) + findings: list[StaticFinding] = [] + trajectories: list[HunterTrajectory] = [] + max_workers = min(8, max(1, len(tasks))) + with ThreadPoolExecutor(max_workers=max_workers) as executor: + for task_findings, trajectory in executor.map(lambda task: _run_hunter_task(root, scan_id, task), tasks): + findings.extend(task_findings) + trajectories.append(trajectory) + return HunterPoolResult(findings=sorted(findings, key=lambda item: item.finding_id), trajectories=trajectories) + + +def write_hunter_trajectories(trajectories: list[HunterTrajectory], path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("".join(json.dumps(asdict(trajectory), sort_keys=True) + "\n" for trajectory in trajectories)) + + +def build_retrieval_context_by_path(root: Path, targets: list[FileTarget], *, max_chars: int = 6000) -> dict[str, str]: + contexts: dict[str, str] = {} + for target in targets: + source = _read_source(root / target.path, max_chars=max_chars) + if source: + contexts[target.path] = f"[repo_code] {target.path}\n{source}" + return contexts + + +def select_skill_snippets(target: FileTarget, *, max_snippets: int = 4) -> list[str]: + tags = set(target.tags) + snippets: list[str] = [] + if target.language in {"c", "cpp", "c++"} or "memory_unsafe" in tags: + snippets.extend(["address-sanitizer", "libfuzzer", "harness-writing"]) + if "parser" in tags or "fuzzable" in tags: + snippets.extend(["fuzzing-dictionary", "fuzzing-obstacles"]) + if "crypto" in tags: + snippets.extend(["constant-time-analysis", "wycheproof"]) + if "auth_boundary" in tags or "network_entry" in tags: + snippets.extend(["sharp-edges", "insecure-defaults"]) + if "deserialization" in tags: + snippets.append("semgrep-rule-creator") + return sorted(dict.fromkeys(snippets))[:max_snippets] + + +def _run_hunter_task(root: Path, scan_id: str, task: HunterTask) -> tuple[list[StaticFinding], HunterTrajectory]: + findings = quick_scan_findings(root, [task.target], [task.ranking])[: task.budget.max_findings_per_target] + return findings, HunterTrajectory( + scan_id=scan_id, + target_path=task.target.path, + tier=task.tier, + priority=task.ranking.priority, + selected_skills=task.selected_skills, + retrieval_context_chars=len(task.retrieval_context), + findings=[finding.finding_id for finding in findings], + status="completed", + ) + + +def _read_source(path: Path, *, max_chars: int) -> str: + try: + return path.read_text(errors="ignore")[:max_chars] + except OSError: + return "" diff --git a/src/openultrasast/reports.py b/src/openultrasast/reports.py index 34632d0..649f367 100644 --- a/src/openultrasast/reports.py +++ b/src/openultrasast/reports.py @@ -88,12 +88,7 @@ def write_manifest( "verification_status": verification_by_id[finding.finding_id].status if finding.finding_id in verification_by_id else "not_run", - "artifact_refs": { - "findings_json": _relative_artifact(run.root, artifact_paths["findings"]), - "verification_json": _relative_artifact(run.root, artifact_paths["verification"]), - "markdown": _relative_artifact(run.root, artifact_paths["markdown"]), - "sarif": _relative_artifact(run.root, artifact_paths["sarif"]), - }, + "artifact_refs": _artifact_refs(run.root, artifact_paths), } for finding in findings ], @@ -170,6 +165,18 @@ def _verification_by_id(verifications: list[VerificationResult]) -> dict[str, Ve return {result.finding_id: result for result in verifications} +def _artifact_refs(root: Path, artifact_paths: dict[str, Path]) -> dict[str, str]: + refs = { + "findings_json": _relative_artifact(root, artifact_paths["findings"]), + "verification_json": _relative_artifact(root, artifact_paths["verification"]), + "markdown": _relative_artifact(root, artifact_paths["markdown"]), + "sarif": _relative_artifact(root, artifact_paths["sarif"]), + } + if "trajectories" in artifact_paths: + refs["trajectories_jsonl"] = _relative_artifact(root, artifact_paths["trajectories"]) + return refs + + def _relative_artifact(root: Path, path: Path) -> str: try: return path.relative_to(root).as_posix() diff --git a/tests/test_hunter.py b/tests/test_hunter.py new file mode 100644 index 0000000..14e45b7 --- /dev/null +++ b/tests/test_hunter.py @@ -0,0 +1,110 @@ +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.preprocess import FileTarget +from openultrasast.rank import RankingScore + + +def test_assign_tier_uses_priority_bands() -> None: + assert assign_tier(3.0) == "A" + assert assign_tier(2.0) == "B" + assert assign_tier(1.99) == "C" + + +def test_schedule_hunter_tasks_assigns_budget_and_skill_metadata() -> None: + target = FileTarget( + path="parser.c", + absolute_path="/repo/parser.c", + language="c", + loc=20, + tags=["memory_unsafe", "parser"], + has_fuzz_entry_point=False, + ) + ranking = RankingScore( + path="parser.c", + surface=5, + influence=2, + reachability=4, + priority=4.1, + rationale="fixture", + model_id=None, + static_boosts=[], + ) + + tasks = schedule_hunter_tasks([target], [ranking], retrieval_context_by_path={"parser.c": "x" * 10_000}) + + assert len(tasks) == 1 + 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: + repo = tmp_path / "repo" + repo.mkdir() + (repo / "app.py").write_text("@app.route('/admin')\ndef admin():\n return eval(request.data)\n") + target = FileTarget( + path="app.py", + absolute_path=str(repo / "app.py"), + language="python", + loc=3, + tags=["network_entry"], + has_fuzz_entry_point=False, + reachability_hints=[ + { + "kind": "route", + "access_level": "public", + "line": 1, + "end_line": 3, + "function_name": "admin", + "conditions": [], + } + ], + ) + ranking = RankingScore("app.py", 4, 1, 5, 3.7, "fixture", None, []) + + result = run_hunter_pool(repo, [target], [ranking], scan_id="scan-1") + output = tmp_path / "hunter_trajectories.jsonl" + write_hunter_trajectories(result.trajectories, output) + + line = json.loads(output.read_text().splitlines()[0]) + assert result.findings[0].finding_id in line["findings"] + assert line["target_path"] == "app.py" + assert line["retrieval_context_chars"] > 0 + 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() + (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", "standard", "--fail-on", "verified"]) == 1 + + run_dir = sorted((repo / ".runs").iterdir())[-1] + trajectory_path = run_dir / "traces" / "hunter_trajectories.jsonl" + manifest = json.loads((run_dir / "manifest.json").read_text()) + + assert trajectory_path.exists() + assert manifest["findings"][0]["artifact_refs"]["trajectories_jsonl"] == "traces/hunter_trajectories.jsonl"