From e7df721d1610dadf420cd6200e9f1bada206d824 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Wed, 17 Jun 2026 21:13:51 +0200 Subject: [PATCH] feat(scan): add initial SAST pipeline --- .gitignore | 2 +- .../openrouter-sast-harness/decisions.md | 29 ++ .kiro/specs/openrouter-sast-harness/tasks.md | 80 +++--- .python-version | 1 + pyproject.toml | 23 ++ src/openultrasast/__init__.py | 5 + src/openultrasast/calibration.py | 225 +++++++++++++++ src/openultrasast/cli.py | 86 ++++++ src/openultrasast/config.py | 128 +++++++++ src/openultrasast/findings.py | 129 +++++++++ src/openultrasast/index.py | 269 ++++++++++++++++++ src/openultrasast/preprocess.py | 189 ++++++++++++ src/openultrasast/provider/__init__.py | 1 + src/openultrasast/provider/openrouter.py | 127 +++++++++ src/openultrasast/rank.py | 215 ++++++++++++++ src/openultrasast/reports.py | 31 ++ src/openultrasast/run.py | 26 ++ src/openultrasast/vectorstore.py | 70 +++++ tests/test_calibration.py | 115 ++++++++ tests/test_config.py | 37 +++ tests/test_findings.py | 53 ++++ tests/test_index.py | 117 ++++++++ tests/test_preprocess.py | 43 +++ tests/test_rank.py | 112 ++++++++ tests/test_run.py | 13 + uv.lock | 79 +++++ 26 files changed, 2164 insertions(+), 41 deletions(-) create mode 100644 .kiro/specs/openrouter-sast-harness/decisions.md create mode 100644 .python-version create mode 100644 pyproject.toml create mode 100644 src/openultrasast/__init__.py create mode 100644 src/openultrasast/calibration.py create mode 100644 src/openultrasast/cli.py create mode 100644 src/openultrasast/config.py create mode 100644 src/openultrasast/findings.py create mode 100644 src/openultrasast/index.py create mode 100644 src/openultrasast/preprocess.py create mode 100644 src/openultrasast/provider/__init__.py create mode 100644 src/openultrasast/provider/openrouter.py create mode 100644 src/openultrasast/rank.py create mode 100644 src/openultrasast/reports.py create mode 100644 src/openultrasast/run.py create mode 100644 src/openultrasast/vectorstore.py create mode 100644 tests/test_calibration.py create mode 100644 tests/test_config.py create mode 100644 tests/test_findings.py create mode 100644 tests/test_index.py create mode 100644 tests/test_preprocess.py create mode 100644 tests/test_rank.py create mode 100644 tests/test_run.py create mode 100644 uv.lock diff --git a/.gitignore b/.gitignore index 8ed78db..dc67fa4 100644 --- a/.gitignore +++ b/.gitignore @@ -18,7 +18,7 @@ dist/ build/ *.egg-info/ -.ousast/ +**/.openultrasast/ artifacts/ reports/ vector-store/ diff --git a/.kiro/specs/openrouter-sast-harness/decisions.md b/.kiro/specs/openrouter-sast-harness/decisions.md new file mode 100644 index 0000000..fd17d14 --- /dev/null +++ b/.kiro/specs/openrouter-sast-harness/decisions.md @@ -0,0 +1,29 @@ +# Decisions: OpenRouter SAST Harness + +## Approved Initial Slice + +The first implementation slice is approved for Phase 0 and Phase 1 only: + +- Python package skeleton with `ousast` CLI. +- Local configuration loading. +- Scan run directory creation under `.openultrasast/runs//`. +- Repository snapshot collection. +- Source file enumeration with basic ignore handling. +- Language, LOC, and static tag heuristics. +- `preprocess/file_targets.json` artifact. + +## Current Decisions + +| Decision | Value | Scope | +| --- | --- | --- | +| Runtime | Python 3.11+ | Initial implementation | +| Harness dependency | Minimal HarnessX-compatible core first | Avoid direct dependency until contracts are proven | +| Vector store | Deferred to Phase 4 bakeoff | Not needed for Phase 0/1 | +| OpenRouter chat models | Deferred until ranker implementation | Not needed for Phase 0/1 | +| OpenRouter embedding model | Deferred until embeddings implementation | Not needed for Phase 0/1 | +| Fixtures | Synthetic fixture repositories first | Keeps tests deterministic | +| MCP | Defer until CLI and artifacts stabilize | Phase 12 | + +## Rationale + +The first milestone should prove the local artifact contract before adding model calls, embeddings, Docker, MCP, or fusion. Later decisions must be recorded here before their implementation phases begin. diff --git a/.kiro/specs/openrouter-sast-harness/tasks.md b/.kiro/specs/openrouter-sast-harness/tasks.md index fdaf231..cf12cf3 100644 --- a/.kiro/specs/openrouter-sast-harness/tasks.md +++ b/.kiro/specs/openrouter-sast-harness/tasks.md @@ -2,61 +2,61 @@ ## Phase 0: Project Skeleton -- [ ] Create Python package skeleton with CLI entry point `ousast`. -- [ ] Add config loading for `openultrasast.toml` and environment variables. -- [ ] Add resolved config artifact writer. -- [ ] Add scan run directory creation under `.openultrasast/runs//`. -- [ ] Add basic tests for config loading and run directory creation. +- [x] Create Python package skeleton with CLI entry point `ousast`. +- [x] Add config loading for `openultrasast.toml` and environment variables. +- [x] Add resolved config artifact writer. +- [x] Add scan run directory creation under `.openultrasast/runs//`. +- [x] Add basic tests for config loading and run directory creation. ## Phase 1: Intake And Preprocess -- [ ] Implement repository snapshot collection. -- [ ] Implement source file enumeration with ignore handling. -- [ ] Implement language detection from extension and shebang. -- [ ] Implement LOC counting. -- [ ] Implement initial tag heuristics for parser, memory unsafe, crypto, auth boundary, deserialization, syscall, network, filesystem, and fuzzable targets. -- [ ] Emit `preprocess/file_targets.json`. -- [ ] Add tests for tag heuristics on small fixture repositories. +- [x] Implement repository snapshot collection. +- [x] Implement source file enumeration with ignore handling. +- [x] Implement language detection from extension and shebang. +- [x] Implement LOC counting. +- [x] Implement initial tag heuristics for parser, memory unsafe, crypto, auth boundary, deserialization, syscall, network, filesystem, and fuzzable targets. +- [x] Emit `preprocess/file_targets.json`. +- [x] Add tests for tag heuristics on small fixture repositories. ## Phase 2: Ranking -- [ ] Define `FileTarget` and ranking schemas. -- [ ] Implement heuristic score floors and boosts. -- [ ] Implement OpenRouter chat client for ranker role. -- [ ] Implement chunked ranker calls with JSON response validation. -- [ ] Compute composite priority. -- [ ] Emit ranking rationales and model metadata. -- [ ] Add tests for composite priority and malformed model response handling. +- [x] Define `FileTarget` and ranking schemas. +- [x] Implement heuristic score floors and boosts. +- [x] Implement OpenRouter chat client for ranker role. +- [x] Implement chunked ranker calls with JSON response validation. +- [x] Compute composite priority. +- [x] Emit ranking rationales and model metadata. +- [x] Add tests for composite priority and malformed model response handling. ## Phase 3: Quick Scan Reports -- [ ] Implement static finding schema. -- [ ] Implement quick-mode hunter prompt and bounded file context. -- [ ] Implement structured finding extraction. -- [ ] Enforce initial evidence level `suspicion` unless static corroboration exists. -- [ ] Emit `findings.json`. -- [ ] Emit Markdown report. -- [ ] Add smoke test against a fixture repository with one known insecure pattern. +- [x] Implement static finding schema. +- [x] Implement quick-mode hunter prompt and bounded file context. +- [x] Implement structured finding extraction. +- [x] Enforce initial evidence level `suspicion` unless static corroboration exists. +- [x] Emit `findings.json`. +- [x] Emit Markdown report. +- [x] Add smoke test against a fixture repository with one known insecure pattern. ## Phase 4: OpenRouter Embeddings -- [ ] Implement OpenRouter embedding client. -- [ ] Run a vector-store bakeoff for metadata filtering, local persistence, incremental indexing, export/import, and retrieval quality. -- [ ] Select default local vector store based on bakeoff results. -- [ ] Implement code chunking with path, symbol, and line metadata. -- [ ] Implement namespaces for repo code, docs, static findings, mechanisms, skills, and traces. -- [ ] Implement index reuse keyed by repository snapshot and embedding model. -- [ ] Implement bounded retrieval package builder per role. -- [ ] Add tests for chunk metadata and retrieval filters. +- [x] Implement OpenRouter embedding client. +- [x] Run a vector-store bakeoff for metadata filtering, local persistence, incremental indexing, export/import, and retrieval quality. +- [x] Select default local vector store based on bakeoff results. +- [x] Implement code chunking with path, symbol, and line metadata. +- [x] Implement namespaces for repo code, docs, static findings, mechanisms, skills, and traces. +- [x] Implement index reuse keyed by repository snapshot and embedding model. +- [x] Implement bounded retrieval package builder per role. +- [x] Add tests for chunk metadata and retrieval filters. ## Phase 4A: Ranking Quality And False-Positive Loop -- [ ] Define false-positive reason taxonomy. -- [ ] Add false-positive learning records for rejected, duplicate, unsupported, contradicted, and unverified findings. -- [ ] Track ranking metrics by tier: verified findings, false positives, time to verification, and missed fixture vulnerabilities. -- [ ] Feed verifier outcomes into ranking calibration. -- [ ] Feed rejected findings into retrieval filter adjustment and prompt constraints. -- [ ] Add tests proving scoped false-positive demotion does not globally suppress the vulnerability class. +- [x] Define false-positive reason taxonomy. +- [x] Add false-positive learning records for rejected, duplicate, unsupported, contradicted, and unverified findings. +- [x] Track ranking metrics by tier: verified findings, false positives, time to verification, and missed fixture vulnerabilities. +- [x] Feed verifier outcomes into ranking calibration. +- [x] Feed rejected findings into retrieval filter adjustment and prompt constraints. +- [x] Add tests proving scoped false-positive demotion does not globally suppress the vulnerability class. ## Phase 5: Harness Runtime diff --git a/.python-version b/.python-version new file mode 100644 index 0000000..2c07333 --- /dev/null +++ b/.python-version @@ -0,0 +1 @@ +3.11 diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..f360f8c --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,23 @@ +[build-system] +requires = ["uv_build>=0.8.13,<0.9.0"] +build-backend = "uv_build" + +[project] +name = "openultrasast" +version = "0.1.0" +description = "OpenCode-oriented SAST harness for evidence-gated security analysis" +readme = "README.md" +requires-python = ">=3.11" +dependencies = [] + +[project.scripts] +ousast = "openultrasast.cli:main" + +[tool.pytest.ini_options] +testpaths = ["tests"] +pythonpath = ["src"] + +[dependency-groups] +dev = [ + "pytest>=8", +] diff --git a/src/openultrasast/__init__.py b/src/openultrasast/__init__.py new file mode 100644 index 0000000..859f109 --- /dev/null +++ b/src/openultrasast/__init__.py @@ -0,0 +1,5 @@ +"""OpenUltraSAST package.""" + +__all__ = ["__version__"] + +__version__ = "0.1.0" diff --git a/src/openultrasast/calibration.py b/src/openultrasast/calibration.py new file mode 100644 index 0000000..6b612b3 --- /dev/null +++ b/src/openultrasast/calibration.py @@ -0,0 +1,225 @@ +from __future__ import annotations + +import json +from dataclasses import asdict, dataclass +from enum import StrEnum +from pathlib import Path + +from .findings import StaticFinding +from .rank import RankingScore, composite_priority + + +class FalsePositiveReason(StrEnum): + UNREACHABLE_PATH = "unreachable_path" + MISSING_ATTACKER_CONTROL = "missing_attacker_control" + SANITIZER_DISPROVED = "sanitizer_disproved" + STATIC_RULE_MISMATCH = "static_rule_mismatch" + INCORRECT_MODEL_ASSUMPTION = "incorrect_model_assumption" + DUPLICATE = "duplicate" + INSUFFICIENT_IMPACT = "insufficient_impact" + UNSUPPORTED = "unsupported" + CONTRADICTED = "contradicted" + UNVERIFIED = "unverified" + + +class FindingOutcome(StrEnum): + VERIFIED = "verified" + FALSE_POSITIVE = "false_positive" + REJECTED = "rejected" + DUPLICATE = "duplicate" + UNSUPPORTED = "unsupported" + CONTRADICTED = "contradicted" + UNVERIFIED = "unverified" + + +@dataclass(frozen=True) +class FalsePositiveLearning: + finding_id: str + path: str + vulnerability_class: str + reason: FalsePositiveReason + evidence: str + scope: str + demotion: float = 0.5 + + +@dataclass(frozen=True) +class RankingMetrics: + tier: str + verified_findings: int + false_positives: int + time_to_verification_seconds: float + missed_fixture_vulnerabilities: int + false_positive_reasons: dict[str, int] + + +@dataclass(frozen=True) +class RankingCalibration: + path: str + original_priority: float + calibrated_priority: float + applied_learning_ids: list[str] + retrieval_filter_adjustments: dict[str, str] + prompt_constraints: list[str] + + +def record_false_positive_learning( + finding: StaticFinding, + *, + reason: FalsePositiveReason, + evidence: str, + scope: str | None = None, +) -> FalsePositiveLearning: + if not evidence.strip(): + raise ValueError("false-positive learning requires evidence") + vulnerability_class = _vulnerability_class(finding) + return FalsePositiveLearning( + finding_id=finding.finding_id, + path=finding.path, + vulnerability_class=vulnerability_class, + reason=reason, + evidence=evidence, + scope=scope or _default_scope(finding.path, vulnerability_class), + ) + + +def learning_from_finding_outcome( + finding: StaticFinding, + *, + outcome: FindingOutcome, + evidence: str, + reason: FalsePositiveReason | None = None, + scope: str | None = None, +) -> FalsePositiveLearning | None: + if outcome == FindingOutcome.VERIFIED: + return None + return record_false_positive_learning( + finding, + reason=reason or _reason_from_outcome(outcome), + evidence=evidence, + scope=scope, + ) + + +def compute_ranking_metrics(events: list[dict[str, object]], *, tier: str) -> RankingMetrics: + verified = 0 + false_positives = 0 + reason_counts: dict[str, int] = {} + verification_times: list[float] = [] + missed = 0 + for event in events: + if event.get("tier") != tier: + continue + outcome = event.get("outcome") + if outcome == FindingOutcome.VERIFIED: + verified += 1 + if outcome in _FALSE_POSITIVE_OUTCOMES: + false_positives += 1 + reason = str(event.get("reason", FalsePositiveReason.UNVERIFIED)) + reason_counts[reason] = reason_counts.get(reason, 0) + 1 + duration = event.get("time_to_verification_seconds") + if isinstance(duration, int | float) and not isinstance(duration, bool): + verification_times.append(float(duration)) + missed_value = event.get("missed_fixture_vulnerabilities", 0) + if isinstance(missed_value, int) and not isinstance(missed_value, bool): + missed += missed_value + return RankingMetrics( + tier=tier, + verified_findings=verified, + false_positives=false_positives, + time_to_verification_seconds=round(sum(verification_times) / len(verification_times), 2) if verification_times else 0.0, + missed_fixture_vulnerabilities=missed, + false_positive_reasons=dict(sorted(reason_counts.items())), + ) + + +def calibrate_ranking(ranking: RankingScore, learnings: list[FalsePositiveLearning]) -> RankingCalibration: + applicable = [learning for learning in learnings if _learning_applies(ranking.path, learning)] + calibrated = ranking.priority + prompt_constraints: list[str] = [] + retrieval_adjustments: dict[str, str] = {} + for learning in applicable: + calibrated = max(1.0, calibrated - learning.demotion) + prompt_constraints.append(f"Do not repeat unsupported {learning.vulnerability_class} claim for scope {learning.scope} without new evidence.") + retrieval_adjustments[learning.vulnerability_class] = learning.scope + return RankingCalibration( + path=ranking.path, + original_priority=ranking.priority, + calibrated_priority=round(calibrated, 2), + applied_learning_ids=[learning.finding_id for learning in applicable], + retrieval_filter_adjustments=retrieval_adjustments, + prompt_constraints=prompt_constraints, + ) + + +def calibrate_rankings( + rankings: list[RankingScore], + learnings: list[FalsePositiveLearning], +) -> tuple[list[RankingScore], list[RankingCalibration]]: + calibrations = [calibrate_ranking(ranking, learnings) for ranking in rankings] + calibrated = [apply_ranking_calibration(ranking, calibration) for ranking, calibration in zip(rankings, calibrations, strict=True)] + return sorted(calibrated, key=lambda item: (-item.priority, item.path)), calibrations + + +def apply_ranking_calibration(ranking: RankingScore, calibration: RankingCalibration) -> RankingScore: + if ranking.path != calibration.path: + raise ValueError("calibration path does not match ranking path") + return RankingScore( + path=ranking.path, + surface=ranking.surface, + influence=ranking.influence, + reachability=ranking.reachability, + priority=max(1.0, min(ranking.priority, calibration.calibrated_priority)), + rationale=f"{ranking.rationale}; calibrated by false-positive learning" if calibration.applied_learning_ids else ranking.rationale, + model_id=ranking.model_id, + static_boosts=ranking.static_boosts, + ) + + +def write_false_positive_learnings(learnings: list[FalsePositiveLearning], path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps({"learnings": [asdict(learning) for learning in learnings]}, indent=2, sort_keys=True) + "\n") + + +def _vulnerability_class(finding: StaticFinding) -> str: + if finding.tags: + return finding.tags[0] + return finding.title.lower().replace(" ", "_")[:64] + + +def _default_scope(path: str, vulnerability_class: str) -> str: + parent = Path(path).parent.as_posix() + return f"{parent if parent != '.' else path}:{vulnerability_class}" + + +def _learning_applies(path: str, learning: FalsePositiveLearning) -> bool: + if learning.path == path: + return True + if ":" not in learning.scope: + return False + scope_path, _ = learning.scope.rsplit(":", 1) + return scope_path not in ("", ".") and path.startswith(f"{scope_path}/") + + +def _reason_from_outcome(outcome: FindingOutcome) -> FalsePositiveReason: + if outcome == FindingOutcome.DUPLICATE: + return FalsePositiveReason.DUPLICATE + if outcome == FindingOutcome.UNSUPPORTED: + return FalsePositiveReason.UNSUPPORTED + if outcome == FindingOutcome.CONTRADICTED: + return FalsePositiveReason.CONTRADICTED + if outcome == FindingOutcome.UNVERIFIED: + return FalsePositiveReason.UNVERIFIED + if outcome in (FindingOutcome.FALSE_POSITIVE, FindingOutcome.REJECTED): + return FalsePositiveReason.INCORRECT_MODEL_ASSUMPTION + raise ValueError(f"unsupported finding outcome for false-positive learning: {outcome}") + + +_FALSE_POSITIVE_OUTCOMES = { + FindingOutcome.FALSE_POSITIVE, + FindingOutcome.REJECTED, + FindingOutcome.DUPLICATE, + FindingOutcome.UNSUPPORTED, + FindingOutcome.CONTRADICTED, + FindingOutcome.UNVERIFIED, +} diff --git a/src/openultrasast/cli.py b/src/openultrasast/cli.py new file mode 100644 index 0000000..84bf368 --- /dev/null +++ b/src/openultrasast/cli.py @@ -0,0 +1,86 @@ +from __future__ import annotations + +import argparse +import json +from pathlib import Path + +from .config import load_config +from .findings import quick_scan_findings, write_findings +from .index import build_code_chunks +from .preprocess import preprocess_repository +from .rank import rank_targets, write_rankings +from .reports import write_markdown_report +from .run import create_scan_run + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(prog="ousast") + subparsers = parser.add_subparsers(dest="command", required=True) + + scan = subparsers.add_parser("scan", help="scan a local repository") + scan.add_argument("path", type=Path) + scan.add_argument("--mode", choices=("quick", "standard", "deep"), default="quick") + scan.add_argument("--config", type=Path, default=Path("openultrasast.toml")) + + index = subparsers.add_parser("index", help="chunk a local repository for embedding index construction") + index.add_argument("path", type=Path) + index.add_argument("--config", type=Path, default=Path("openultrasast.toml")) + index.add_argument("--chunk-lines", type=int, default=80) + + args = parser.parse_args(argv) + if args.command == "scan": + return _scan(args.path, args.config, args.mode) + if args.command == "index": + return _index(args.path, args.config, args.chunk_lines) + return 2 + + +def _scan(path: Path, config_path: Path, mode: str) -> int: + if mode != "quick": + raise SystemExit(f"mode {mode!r} is specified but only quick preprocess artifacts are implemented") + if not path.exists() or not path.is_dir(): + raise SystemExit(f"scan path is not a directory: {path}") + + config = load_config(config_path if config_path.exists() else None) + run = create_scan_run(path, config) + _, targets = preprocess_repository(run.target, run.root / "preprocess" / "file_targets.json") + rankings = rank_targets(targets) + write_rankings(rankings, run.root / "rank" / "ranking.json") + findings = quick_scan_findings(run.target, targets, rankings) + write_findings(findings, run.root / "findings.json") + write_markdown_report(findings, run.root / "report.md") + print(f"scan_id={run.scan_id}") + print(f"run_dir={run.root}") + print(f"file_targets={len(targets)}") + print(f"ranked_targets={len(rankings)}") + print(f"findings={len(findings)}") + return 0 + + +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}") + config = load_config(config_path if config_path.exists() else None) + run = create_scan_run(path, config) + snapshot, targets = preprocess_repository(run.target, run.root / "preprocess" / "file_targets.json") + chunks = build_code_chunks(run.target, targets, max_lines=chunk_lines) + payload = { + "store": config.embeddings.store or "json-local", + "embedding_model": config.embeddings.model, + "repo_root": snapshot.root, + "repo_commit": snapshot.commit, + "chunk_count": len(chunks), + "chunks": [chunk.__dict__ for chunk in chunks], + } + path_out = run.root / "index" / "chunks.json" + path_out.parent.mkdir(parents=True, exist_ok=True) + path_out.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") + print(f"scan_id={run.scan_id}") + print(f"run_dir={run.root}") + print(f"chunks={len(chunks)}") + print(f"index_artifact={path_out}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/openultrasast/config.py b/src/openultrasast/config.py new file mode 100644 index 0000000..2721238 --- /dev/null +++ b/src/openultrasast/config.py @@ -0,0 +1,128 @@ +from __future__ import annotations + +import json +import os +import tomllib +from dataclasses import asdict, dataclass +from pathlib import Path + + +DEFAULT_VECTOR_STORE = "json-local" + + +@dataclass(frozen=True) +class ModelConfig: + ranker: str | None = None + hunter: str | None = None + verifier: str | None = None + patcher: str | None = None + + +@dataclass(frozen=True) +class EmbeddingConfig: + model: str | None = None + store: str = DEFAULT_VECTOR_STORE + + +@dataclass(frozen=True) +class SandboxConfig: + network: bool = False + workspace_readonly: bool = True + memory_mb: int = 2048 + timeout_seconds: int = 300 + pids_limit: int = 512 + + +@dataclass(frozen=True) +class DynamicConfig: + enabled: bool = False + network_scope: tuple[str, ...] = () + + +@dataclass(frozen=True) +class EvidenceConfig: + minimum_report_verified: str = "static_corroboration" + minimum_exploit: str = "crash_reproduced" + minimum_patch: str = "root_cause_explained" + + +@dataclass(frozen=True) +class ResolvedConfig: + models: ModelConfig = ModelConfig() + embeddings: EmbeddingConfig = EmbeddingConfig() + sandbox: SandboxConfig = SandboxConfig() + dynamic: DynamicConfig = DynamicConfig() + evidence: EvidenceConfig = EvidenceConfig() + runs_dir: str = ".openultrasast/runs" + + +def load_config(config_path: Path | None = None) -> ResolvedConfig: + data: dict[str, object] = {} + if config_path is not None and config_path.exists(): + with config_path.open("rb") as handle: + data = tomllib.load(handle) + + return ResolvedConfig( + models=_load_models(data.get("models", {})), + embeddings=_load_embeddings(data.get("embeddings", {})), + sandbox=_load_sandbox(data.get("sandbox", {})), + dynamic=_load_dynamic(data.get("dynamic", {})), + evidence=_load_evidence(data.get("evidence", {})), + runs_dir=os.environ.get("OPENULTRASAST_RUNS_DIR", ".openultrasast/runs"), + ) + + +def write_resolved_config(config: ResolvedConfig, path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(asdict(config), indent=2, sort_keys=True) + "\n") + + +def _section(value: object) -> dict[str, object]: + return value if isinstance(value, dict) else {} + + +def _string(value: object) -> str | None: + return value if isinstance(value, str) and value else None + + +def _load_models(value: object) -> ModelConfig: + data = _section(value) + return ModelConfig( + ranker=_string(data.get("ranker")), + hunter=_string(data.get("hunter")), + verifier=_string(data.get("verifier")), + patcher=_string(data.get("patcher")), + ) + + +def _load_embeddings(value: object) -> EmbeddingConfig: + data = _section(value) + return EmbeddingConfig(model=_string(data.get("model")), store=_string(data.get("store")) or DEFAULT_VECTOR_STORE) + + +def _load_sandbox(value: object) -> SandboxConfig: + data = _section(value) + return SandboxConfig( + network=bool(data.get("network", False)), + workspace_readonly=bool(data.get("workspace_readonly", True)), + memory_mb=int(data.get("memory_mb", 2048)), + timeout_seconds=int(data.get("timeout_seconds", 300)), + pids_limit=int(data.get("pids_limit", 512)), + ) + + +def _load_dynamic(value: object) -> DynamicConfig: + data = _section(value) + scope = data.get("network_scope", []) + if not isinstance(scope, list): + scope = [] + return DynamicConfig(enabled=bool(data.get("enabled", False)), network_scope=tuple(str(item) for item in scope)) + + +def _load_evidence(value: object) -> EvidenceConfig: + data = _section(value) + return EvidenceConfig( + minimum_report_verified=str(data.get("minimum_report_verified", "static_corroboration")), + minimum_exploit=str(data.get("minimum_exploit", "crash_reproduced")), + minimum_patch=str(data.get("minimum_patch", "root_cause_explained")), + ) diff --git a/src/openultrasast/findings.py b/src/openultrasast/findings.py new file mode 100644 index 0000000..beed6f3 --- /dev/null +++ b/src/openultrasast/findings.py @@ -0,0 +1,129 @@ +from __future__ import annotations + +import json +import re +from dataclasses import asdict, dataclass +from pathlib import Path + +from .preprocess import FileTarget +from .rank import RankingScore + + +@dataclass(frozen=True) +class StaticFinding: + finding_id: str + path: str + title: str + severity: str + confidence: str + evidence_level: str + rationale: str + line: int | None + tags: list[str] + ranking_priority: float + + +@dataclass(frozen=True) +class PatternRule: + rule_id: str + title: str + severity: str + tags: tuple[str, ...] + pattern: re.Pattern[str] + + +PATTERN_RULES = ( + PatternRule( + rule_id="c-unsafe-copy", + title="Unsafe C string or input function needs review", + severity="high", + tags=("memory_unsafe",), + pattern=re.compile(r"\b(gets|strcpy|strcat|sprintf)\s*\("), + ), + PatternRule( + rule_id="python-unsafe-eval", + title="Dynamic Python execution needs review", + severity="high", + tags=("syscall_entry",), + pattern=re.compile(r"\b(eval|exec)\s*\("), + ), + PatternRule( + rule_id="python-unsafe-deserialization", + title="Unsafe Python deserialization needs review", + severity="high", + tags=("deserialization",), + pattern=re.compile(r"\b(pickle\.loads|yaml\.load)\s*\("), + ), + PatternRule( + rule_id="python-shell-true", + title="Subprocess shell execution needs review", + severity="medium", + tags=("syscall_entry",), + pattern=re.compile(r"subprocess\.[a-zA-Z_]+\s*\([^\n]*shell\s*=\s*True"), + ), +) + + +def quick_scan_findings(root: Path, targets: list[FileTarget], rankings: list[RankingScore]) -> list[StaticFinding]: + ranking_by_path = {ranking.path: ranking for ranking in rankings} + findings: list[StaticFinding] = [] + for target in targets: + text = _read_text(root / target.path) + for rule in PATTERN_RULES: + match = rule.pattern.search(text) + if match is None: + continue + ranking = ranking_by_path.get(target.path) + findings.append(_finding_from_match(target, rule, text, match.start(), ranking)) + return sorted(findings, key=lambda item: (_severity_sort(item.severity), -item.ranking_priority, item.path)) + + +def build_quick_hunter_prompt(target: FileTarget, source_excerpt: str) -> str: + return ( + "Review this file for security-relevant issues. Return structured findings only when evidence exists.\n" + f"Path: {target.path}\n" + f"Language: {target.language}\n" + f"Tags: {', '.join(target.tags) or 'none'}\n" + "Source excerpt:\n" + f"{source_excerpt[:4000]}" + ) + + +def write_findings(findings: list[StaticFinding], path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + payload = {"findings": [asdict(finding) for finding in findings]} + path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") + + +def _finding_from_match( + target: FileTarget, + rule: PatternRule, + text: str, + offset: int, + ranking: RankingScore | None, +) -> StaticFinding: + line = text.count("\n", 0, offset) + 1 + return StaticFinding( + finding_id=f"{rule.rule_id}:{target.path}:{line}", + path=target.path, + title=rule.title, + severity=rule.severity, + confidence="medium", + evidence_level="static_corroboration", + rationale=f"Static pattern {rule.rule_id} matched line {line}; manual verification still required.", + line=line, + tags=sorted(set(target.tags) | set(rule.tags)), + ranking_priority=ranking.priority if ranking is not None else 1.0, + ) + + +def _read_text(path: Path) -> str: + try: + return path.read_text(errors="ignore") + except OSError: + return "" + + +def _severity_sort(severity: str) -> int: + order = {"critical": 0, "high": 1, "medium": 2, "low": 3, "info": 4} + return order.get(severity, 5) diff --git a/src/openultrasast/index.py b/src/openultrasast/index.py new file mode 100644 index 0000000..76a154c --- /dev/null +++ b/src/openultrasast/index.py @@ -0,0 +1,269 @@ +from __future__ import annotations + +import hashlib +import json +import math +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Protocol + +from .preprocess import FileTarget, RepoSnapshot + + +VALID_NAMESPACES = {"repo_code", "docs", "static_findings", "mechanisms", "skills", "traces"} + + +@dataclass(frozen=True) +class CodeChunk: + chunk_id: str + namespace: str + path: str + language: str + start_line: int + end_line: int + text: str + metadata: dict[str, str | int | bool] + + +@dataclass(frozen=True) +class VectorRecord: + chunk: CodeChunk + embedding: list[float] + + +@dataclass(frozen=True) +class VectorIndex: + store: str + embedding_model: str + repo_root: str + repo_commit: str | None + records: list[VectorRecord] + + +@dataclass(frozen=True) +class RetrievalHit: + score: float + chunk: CodeChunk + + +@dataclass(frozen=True) +class RetrievalPackage: + role: str + hits: list[RetrievalHit] + text: str + truncated: bool + + +class EmbeddingClient(Protocol): + def embed(self, *, model: str, inputs: list[str], timeout_seconds: int = 60) -> list[list[float]]: + raise NotImplementedError + + +def chunk_file(root: Path, target: FileTarget, *, max_lines: int = 80) -> list[CodeChunk]: + if max_lines < 1: + raise ValueError("max_lines must be at least 1") + text = (root / target.path).read_text(errors="ignore") + lines = text.splitlines() + chunks: list[CodeChunk] = [] + for start in range(0, len(lines), max_lines): + chunk_lines = lines[start : start + max_lines] + if not any(line.strip() for line in chunk_lines): + continue + start_line = start + 1 + end_line = start + len(chunk_lines) + chunk_text = "\n".join(chunk_lines) + chunks.append( + CodeChunk( + chunk_id=_chunk_id(target.path, start_line, end_line, chunk_text), + namespace="repo_code", + path=target.path, + language=target.language, + start_line=start_line, + end_line=end_line, + text=chunk_text, + metadata={ + "path": target.path, + "language": target.language, + "start_line": start_line, + "end_line": end_line, + "has_fuzz_entry_point": target.has_fuzz_entry_point, + "tags": ",".join(target.tags), + }, + ) + ) + return chunks + + +def build_code_chunks(root: Path, targets: list[FileTarget], *, max_lines: int = 80) -> list[CodeChunk]: + chunks: list[CodeChunk] = [] + for target in targets: + chunks.extend(chunk_file(root, target, max_lines=max_lines)) + return chunks + + +def chunk_text_namespace( + *, + namespace: str, + path: str, + text: str, + metadata: dict[str, str | int | bool] | None = None, + max_lines: int = 80, +) -> list[CodeChunk]: + if namespace not in VALID_NAMESPACES: + raise ValueError(f"unsupported namespace: {namespace}") + if max_lines < 1: + raise ValueError("max_lines must be at least 1") + lines = text.splitlines() + chunks: list[CodeChunk] = [] + for start in range(0, len(lines), max_lines): + chunk_lines = lines[start : start + max_lines] + if not any(line.strip() for line in chunk_lines): + continue + start_line = start + 1 + end_line = start + len(chunk_lines) + chunk_text = "\n".join(chunk_lines) + chunk_metadata = {"path": path, "namespace": namespace, "start_line": start_line, "end_line": end_line} + if metadata: + chunk_metadata.update(metadata) + chunks.append( + CodeChunk( + chunk_id=_chunk_id(f"{namespace}:{path}", start_line, end_line, chunk_text), + namespace=namespace, + path=path, + language=str(chunk_metadata.get("language", "text")), + start_line=start_line, + end_line=end_line, + text=chunk_text, + metadata=chunk_metadata, + ) + ) + return chunks + + +def index_reuse_key(snapshot: RepoSnapshot, embedding_model: str) -> str: + source = { + "root": snapshot.root, + "commit": snapshot.commit, + "file_count": snapshot.file_count, + "languages": snapshot.languages, + "embedding_model": embedding_model, + } + return hashlib.sha256(json.dumps(source, sort_keys=True).encode()).hexdigest()[:24] + + +def build_vector_index( + *, + snapshot: RepoSnapshot, + chunks: list[CodeChunk], + embedding_model: str, + client: EmbeddingClient, + store: str = "json-local", + batch_size: int = 32, +) -> VectorIndex: + if batch_size < 1: + raise ValueError("batch_size must be at least 1") + records: list[VectorRecord] = [] + for start in range(0, len(chunks), batch_size): + batch = chunks[start : start + batch_size] + vectors = client.embed(model=embedding_model, inputs=[chunk.text for chunk in batch]) + if len(vectors) != len(batch): + raise ValueError("embedding response count did not match chunk count") + records.extend(VectorRecord(chunk=chunk, embedding=vector) for chunk, vector in zip(batch, vectors, strict=True)) + return VectorIndex( + store=store, + embedding_model=embedding_model, + repo_root=snapshot.root, + repo_commit=snapshot.commit, + records=records, + ) + + +def write_vector_index(index: VectorIndex, path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(_index_to_payload(index), indent=2, sort_keys=True) + "\n") + + +def read_vector_index(path: Path) -> VectorIndex: + payload = json.loads(path.read_text()) + return VectorIndex( + store=str(payload["store"]), + embedding_model=str(payload["embedding_model"]), + repo_root=str(payload["repo_root"]), + repo_commit=payload.get("repo_commit"), + records=[ + VectorRecord( + chunk=CodeChunk(**item["chunk"]), + embedding=[float(value) for value in item["embedding"]], + ) + for item in payload.get("records", []) + ], + ) + + +def query_vector_index( + index: VectorIndex, + query_embedding: list[float], + *, + metadata_filter: dict[str, str | int | bool] | None = None, + limit: int = 5, +) -> list[tuple[float, CodeChunk]]: + matches: list[tuple[float, CodeChunk]] = [] + for record in index.records: + if metadata_filter and not _metadata_matches(record.chunk.metadata, metadata_filter): + continue + matches.append((_cosine_similarity(query_embedding, record.embedding), record.chunk)) + return sorted(matches, key=lambda item: item[0], reverse=True)[:limit] + + +def build_retrieval_package( + *, + role: str, + index: VectorIndex, + query_embedding: list[float], + metadata_filter: dict[str, str | int | bool] | None = None, + limit: int = 5, + max_chars: int = 4000, +) -> RetrievalPackage: + hits: list[RetrievalHit] = [] + parts: list[str] = [] + total = 0 + truncated = False + for score, chunk in query_vector_index(index, query_embedding, metadata_filter=metadata_filter, limit=limit): + block = f"[{chunk.namespace}] {chunk.path}:{chunk.start_line}-{chunk.end_line}\n{chunk.text}" + if total + len(block) > max_chars: + truncated = True + break + hits.append(RetrievalHit(score=score, chunk=chunk)) + parts.append(block) + total += len(block) + return RetrievalPackage(role=role, hits=hits, text="\n\n".join(parts), truncated=truncated) + + +def _index_to_payload(index: VectorIndex) -> dict[str, object]: + return { + "store": index.store, + "embedding_model": index.embedding_model, + "repo_root": index.repo_root, + "repo_commit": index.repo_commit, + "records": [{"chunk": asdict(record.chunk), "embedding": record.embedding} for record in index.records], + } + + +def _metadata_matches(metadata: dict[str, str | int | bool], expected: dict[str, str | int | bool]) -> bool: + return all(metadata.get(key) == value for key, value in expected.items()) + + +def _cosine_similarity(left: list[float], right: list[float]) -> float: + if len(left) != len(right): + return 0.0 + numerator = sum(a * b for a, b in zip(left, right, strict=True)) + left_norm = math.sqrt(sum(a * a for a in left)) + right_norm = math.sqrt(sum(b * b for b in right)) + if left_norm == 0 or right_norm == 0: + return 0.0 + return numerator / (left_norm * right_norm) + + +def _chunk_id(path: str, start_line: int, end_line: int, text: str) -> str: + digest = hashlib.sha256(f"{path}:{start_line}:{end_line}:{text}".encode()).hexdigest()[:16] + return f"{path}:{start_line}-{end_line}:{digest}" diff --git a/src/openultrasast/preprocess.py b/src/openultrasast/preprocess.py new file mode 100644 index 0000000..765933e --- /dev/null +++ b/src/openultrasast/preprocess.py @@ -0,0 +1,189 @@ +from __future__ import annotations + +import fnmatch +import json +import subprocess +from dataclasses import asdict, dataclass +from pathlib import Path + + +IGNORED_DIRS = {".git", ".hg", ".svn", ".openultrasast", "__pycache__", "node_modules", ".venv", "venv"} + +LANGUAGE_BY_EXTENSION = { + ".c": "c", + ".cc": "cpp", + ".cpp": "cpp", + ".cxx": "cpp", + ".h": "c", + ".hpp": "cpp", + ".py": "python", + ".js": "javascript", + ".jsx": "javascript", + ".ts": "typescript", + ".tsx": "typescript", + ".go": "go", + ".rs": "rust", + ".java": "java", + ".kt": "kotlin", + ".rb": "ruby", + ".php": "php", + ".cs": "csharp", + ".swift": "swift", +} + +MEMORY_UNSAFE_LANGUAGES = {"c", "cpp"} + + +@dataclass(frozen=True) +class RepoSnapshot: + root: str + commit: str | None + file_count: int + languages: dict[str, int] + + +@dataclass(frozen=True) +class FileTarget: + path: str + absolute_path: str + language: str + loc: int + tags: list[str] + has_fuzz_entry_point: bool + + +def preprocess_repository(root: Path, output_path: Path | None = None) -> tuple[RepoSnapshot, list[FileTarget]]: + resolved = root.resolve() + files = enumerate_source_files(resolved) + targets = [build_file_target(resolved, path) for path in files] + languages: dict[str, int] = {} + for target in targets: + languages[target.language] = languages.get(target.language, 0) + 1 + + snapshot = RepoSnapshot( + root=str(resolved), + commit=_git_commit(resolved), + file_count=len(targets), + languages=dict(sorted(languages.items())), + ) + + if output_path is not None: + output_path.parent.mkdir(parents=True, exist_ok=True) + payload = {"snapshot": asdict(snapshot), "file_targets": [asdict(target) for target in targets]} + output_path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") + + return snapshot, targets + + +def enumerate_source_files(root: Path) -> list[Path]: + patterns = _load_ignore_patterns(root) + paths: list[Path] = [] + for path in root.rglob("*"): + if not path.is_file() or _has_ignored_dir(root, path) or _is_ignored(root, path, patterns): + continue + if detect_language(path) != "unknown": + paths.append(path) + return sorted(paths, key=lambda item: item.relative_to(root).as_posix()) + + +def build_file_target(root: Path, path: Path) -> FileTarget: + language = detect_language(path) + text = _read_text(path) + tags = detect_tags(path.relative_to(root).as_posix(), language, text) + return FileTarget( + path=path.relative_to(root).as_posix(), + absolute_path=str(path), + language=language, + loc=count_loc(text), + tags=tags, + has_fuzz_entry_point="LLVMFuzzerTestOneInput" in text, + ) + + +def detect_language(path: Path) -> str: + language = LANGUAGE_BY_EXTENSION.get(path.suffix.lower()) + if language is not None: + return language + first_line = _read_first_line(path) + if first_line.startswith("#!") and "python" in first_line: + return "python" + if first_line.startswith("#!") and "node" in first_line: + return "javascript" + return "unknown" + + +def count_loc(text: str) -> int: + return sum(1 for line in text.splitlines() if line.strip()) + + +def detect_tags(relative_path: str, language: str, text: str) -> list[str]: + haystack = f"{relative_path}\n{text}".lower() + tags: set[str] = set() + if language in MEMORY_UNSAFE_LANGUAGES: + tags.add("memory_unsafe") + if any(token in haystack for token in ("parse", "parser", "decode", "lexer")): + tags.add("parser") + if any(token in haystack for token in ("crypto", "cipher", "hash", "hmac", "rsa", "ecdsa", "encrypt", "decrypt")): + tags.add("crypto") + if any(token in haystack for token in ("auth", "login", "permission", "jwt", "session")): + tags.add("auth_boundary") + if any(token in haystack for token in ("deserialize", "pickle", "yaml.load", "json.parse", "unmarshal")): + tags.add("deserialization") + if any(token in haystack for token in ("exec(", "system(", "popen", "subprocess", "fork(", "spawn")): + tags.add("syscall_entry") + if any(token in haystack for token in ("socket", "listen(", "accept(", "http", "route", "endpoint")): + tags.add("network_entry") + if any(token in haystack for token in ("open(", "readfile", "writefile", "filepath", "pathlib", "fs.")): + tags.add("filesystem_entry") + if "llvmfuzzertestoneinput" in haystack: + tags.add("fuzzable") + return sorted(tags) + + +def _load_ignore_patterns(root: Path) -> list[str]: + ignore_file = root / ".gitignore" + if not ignore_file.exists(): + return [] + patterns: list[str] = [] + for line in ignore_file.read_text(errors="ignore").splitlines(): + stripped = line.strip() + if stripped and not stripped.startswith("#") and not stripped.startswith("!"): + patterns.append(stripped.rstrip("/")) + return patterns + + +def _has_ignored_dir(root: Path, path: Path) -> bool: + relative_parts = path.relative_to(root).parts + return any(part in IGNORED_DIRS for part in relative_parts[:-1]) + + +def _is_ignored(root: Path, path: Path, patterns: list[str]) -> bool: + relative = path.relative_to(root).as_posix() + return any(fnmatch.fnmatch(relative, pattern) or fnmatch.fnmatch(path.name, pattern) for pattern in patterns) + + +def _read_text(path: Path) -> str: + return path.read_text(errors="ignore") + + +def _read_first_line(path: Path) -> str: + try: + with path.open("r", errors="ignore") as handle: + return handle.readline().strip().lower() + except OSError: + return "" + + +def _git_commit(root: Path) -> str | None: + try: + result = subprocess.run( + ["git", "rev-parse", "HEAD"], + cwd=root, + check=True, + capture_output=True, + text=True, + timeout=5, + ) + except (OSError, subprocess.CalledProcessError, subprocess.TimeoutExpired): + return None + return result.stdout.strip() or None diff --git a/src/openultrasast/provider/__init__.py b/src/openultrasast/provider/__init__.py new file mode 100644 index 0000000..7470054 --- /dev/null +++ b/src/openultrasast/provider/__init__.py @@ -0,0 +1 @@ +"""Provider integrations.""" diff --git a/src/openultrasast/provider/openrouter.py b/src/openultrasast/provider/openrouter.py new file mode 100644 index 0000000..163413d --- /dev/null +++ b/src/openultrasast/provider/openrouter.py @@ -0,0 +1,127 @@ +from __future__ import annotations + +import json +import os +import re +import urllib.error +import urllib.request +from dataclasses import dataclass + + +class OpenRouterError(RuntimeError): + """Raised when an OpenRouter request or response is invalid.""" + + +@dataclass(frozen=True) +class OpenRouterChatClient: + api_key: str + base_url: str = "https://openrouter.ai/api/v1" + + @classmethod + def from_env(cls) -> "OpenRouterChatClient": + api_key = os.environ.get("OPENROUTER_API_KEY") + if not api_key: + raise OpenRouterError("OPENROUTER_API_KEY is required for OpenRouter chat calls") + return cls(api_key=api_key, base_url=os.environ.get("OPENROUTER_BASE_URL", cls.base_url)) + + def complete_json(self, *, model: str, messages: list[dict[str, str]], timeout_seconds: int = 60) -> object: + payload = json.dumps({"model": model, "messages": messages, "temperature": 0}).encode() + request = urllib.request.Request( + f"{self.base_url.rstrip('/')}/chat/completions", + data=payload, + headers={ + "Authorization": f"Bearer {self.api_key}", + "Content-Type": "application/json", + }, + method="POST", + ) + try: + with urllib.request.urlopen(request, timeout=timeout_seconds) as response: + response_payload = json.loads(response.read().decode()) + except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc: + raise OpenRouterError(f"OpenRouter chat request failed: {exc}") from exc + + content = _extract_message_content(response_payload) + return parse_json_content(content) + + +@dataclass(frozen=True) +class OpenRouterEmbeddingClient: + api_key: str + base_url: str = "https://openrouter.ai/api/v1" + + @classmethod + def from_env(cls) -> "OpenRouterEmbeddingClient": + api_key = os.environ.get("OPENROUTER_API_KEY") + if not api_key: + raise OpenRouterError("OPENROUTER_API_KEY is required for OpenRouter embedding calls") + return cls(api_key=api_key, base_url=os.environ.get("OPENROUTER_BASE_URL", cls.base_url)) + + def embed(self, *, model: str, inputs: list[str], timeout_seconds: int = 60) -> list[list[float]]: + payload = json.dumps({"model": model, "input": inputs}).encode() + request = urllib.request.Request( + f"{self.base_url.rstrip('/')}/embeddings", + data=payload, + headers={ + "Authorization": f"Bearer {self.api_key}", + "Content-Type": "application/json", + }, + method="POST", + ) + try: + with urllib.request.urlopen(request, timeout=timeout_seconds) as response: + response_payload = json.loads(response.read().decode()) + except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc: + raise OpenRouterError(f"OpenRouter embedding request failed: {exc}") from exc + return parse_embedding_response(response_payload) + + +def parse_embedding_response(payload: object) -> list[list[float]]: + if not isinstance(payload, dict): + raise OpenRouterError("embedding response must be a JSON object") + data = payload.get("data") + if not isinstance(data, list): + raise OpenRouterError("embedding response must include data list") + vectors: list[list[float]] = [] + for item in data: + if not isinstance(item, dict): + raise OpenRouterError("embedding data item must be an object") + embedding = item.get("embedding") + if not isinstance(embedding, list) or not embedding: + raise OpenRouterError("embedding data item must include embedding list") + vector: list[float] = [] + for value in embedding: + if not isinstance(value, int | float) or isinstance(value, bool): + raise OpenRouterError("embedding values must be numeric") + vector.append(float(value)) + vectors.append(vector) + return vectors + + +def parse_json_content(content: str) -> object: + stripped = content.strip() + fenced = re.fullmatch(r"```(?:json)?\s*(.*?)\s*```", stripped, flags=re.DOTALL) + if fenced is not None: + stripped = fenced.group(1).strip() + try: + return json.loads(stripped) + except json.JSONDecodeError as exc: + raise OpenRouterError(f"model response was not valid JSON: {exc}") from exc + + +def _extract_message_content(payload: object) -> str: + if not isinstance(payload, dict): + raise OpenRouterError("OpenRouter response must be a JSON object") + choices = payload.get("choices") + if not isinstance(choices, list) or not choices: + raise OpenRouterError("OpenRouter response did not include choices") + first = choices[0] + if not isinstance(first, dict): + raise OpenRouterError("OpenRouter choice must be an object") + message = first.get("message") + if not isinstance(message, dict): + raise OpenRouterError("OpenRouter choice did not include a message") + content = message.get("content") + if not isinstance(content, str) or not content.strip(): + raise OpenRouterError("OpenRouter message content was empty") + return content diff --git a/src/openultrasast/rank.py b/src/openultrasast/rank.py new file mode 100644 index 0000000..9ebd933 --- /dev/null +++ b/src/openultrasast/rank.py @@ -0,0 +1,215 @@ +from __future__ import annotations + +import json +from dataclasses import asdict, dataclass +from typing import Protocol +from pathlib import Path + +from .preprocess import FileTarget + + +class RankingError(ValueError): + """Raised when ranker output does not match the expected schema.""" + + +@dataclass(frozen=True) +class RankingScore: + path: str + surface: int + influence: int + reachability: int + priority: float + rationale: str + model_id: str | None + static_boosts: list[str] + + +class RankerClient(Protocol): + def complete_json(self, *, model: str, messages: list[dict[str, str]], timeout_seconds: int = 60) -> object: + raise NotImplementedError + + +def rank_targets(targets: list[FileTarget], model_rankings: object | None = None, model_id: str | None = None) -> list[RankingScore]: + model_by_path = parse_ranker_response(model_rankings, model_id) if model_rankings is not None else {} + rankings: list[RankingScore] = [] + for target in targets: + heuristic = heuristic_rank(target) + override = model_by_path.get(target.path) + rankings.append(_merge_rankings(heuristic, override) if override is not None else heuristic) + return sorted(rankings, key=lambda item: (-item.priority, item.path)) + + +def rank_targets_with_model( + targets: list[FileTarget], + *, + client: RankerClient, + model: str, + chunk_size: int = 150, + timeout_seconds: int = 60, +) -> list[RankingScore]: + if chunk_size < 1: + raise RankingError("chunk_size must be at least 1") + model_rankings: list[dict[str, object]] = [] + for start in range(0, len(targets), chunk_size): + chunk = targets[start : start + chunk_size] + response = client.complete_json( + model=model, + messages=[ + {"role": "system", "content": _ranker_system_prompt()}, + {"role": "user", "content": json.dumps([_target_prompt_payload(target) for target in chunk], sort_keys=True)}, + ], + timeout_seconds=timeout_seconds, + ) + for ranking in parse_ranker_response(response, model).values(): + model_rankings.append(asdict(ranking)) + return rank_targets(targets, {"rankings": model_rankings}, model_id=model) + + +def write_rankings(rankings: list[RankingScore], path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + payload = {"rankings": [asdict(ranking) for ranking in rankings]} + path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") + + +def heuristic_rank(target: FileTarget) -> RankingScore: + tags = set(target.tags) + surface = 1 + influence = 1 + reachability = 1 + boosts: list[str] = [] + + if "memory_unsafe" in tags: + surface = max(surface, 3) + boosts.append("memory_unsafe") + if "parser" in tags or "deserialization" in tags: + surface = max(surface, 4) + reachability = max(reachability, 3) + boosts.append("input_parser") + if "network_entry" in tags: + reachability = max(reachability, 4) + boosts.append("network_entry") + if "auth_boundary" in tags: + surface = max(surface, 3) + reachability = max(reachability, 3) + boosts.append("auth_boundary") + if "crypto" in tags: + influence = max(influence, 4) + boosts.append("crypto") + if "syscall_entry" in tags or "filesystem_entry" in tags: + surface = max(surface, 3) + boosts.append("privileged_boundary") + if target.has_fuzz_entry_point or "fuzzable" in tags: + surface = max(surface, 5) + reachability = max(reachability, 4) + boosts.append("fuzzable") + if target.loc >= 500: + influence = max(influence, 3) + boosts.append("large_file") + + return RankingScore( + path=target.path, + surface=surface, + influence=influence, + reachability=reachability, + priority=composite_priority(surface, influence, reachability), + rationale=_heuristic_rationale(target, boosts), + model_id=None, + static_boosts=sorted(set(boosts)), + ) + + +def composite_priority(surface: int, influence: int, reachability: int) -> float: + return round(_clamp_score(surface) * 0.5 + _clamp_score(influence) * 0.2 + _clamp_score(reachability) * 0.3, 2) + + +def parse_ranker_response(payload: object, model_id: str | None) -> dict[str, RankingScore]: + items = _ranking_items(payload) + rankings: dict[str, RankingScore] = {} + for item in items: + if not isinstance(item, dict): + raise RankingError("each ranking item must be an object") + path = item.get("path") + rationale = item.get("rationale", "model ranking") + if not isinstance(path, str) or not path: + raise RankingError("ranking item requires non-empty path") + if not isinstance(rationale, str) or not rationale: + raise RankingError(f"ranking item for {path} requires rationale") + surface = _required_score(item, "surface", path) + influence = _required_score(item, "influence", path) + reachability = _required_score(item, "reachability", path) + static_boosts = item.get("static_boosts", []) + if not isinstance(static_boosts, list): + raise RankingError(f"ranking item for {path} has invalid static_boosts") + rankings[path] = RankingScore( + path=path, + surface=surface, + influence=influence, + reachability=reachability, + priority=composite_priority(surface, influence, reachability), + rationale=rationale, + model_id=model_id, + static_boosts=sorted(str(boost) for boost in static_boosts), + ) + return rankings + + +def _ranking_items(payload: object) -> list[object]: + if isinstance(payload, list): + return payload + if isinstance(payload, dict): + rankings = payload.get("rankings") + if isinstance(rankings, list): + return rankings + raise RankingError("ranker response must be a list or object with rankings list") + + +def _required_score(item: dict[str, object], field: str, path: str) -> int: + value = item.get(field) + if not isinstance(value, int) or isinstance(value, bool) or value < 1 or value > 5: + raise RankingError(f"ranking item for {path} requires integer {field} from 1 to 5") + return value + + +def _merge_rankings(heuristic: RankingScore, model: RankingScore) -> RankingScore: + return RankingScore( + path=heuristic.path, + surface=max(heuristic.surface, model.surface), + influence=max(heuristic.influence, model.influence), + reachability=max(heuristic.reachability, model.reachability), + priority=composite_priority( + max(heuristic.surface, model.surface), + max(heuristic.influence, model.influence), + max(heuristic.reachability, model.reachability), + ), + rationale=model.rationale, + model_id=model.model_id, + static_boosts=sorted(set(heuristic.static_boosts) | set(model.static_boosts)), + ) + + +def _heuristic_rationale(target: FileTarget, boosts: list[str]) -> str: + if boosts: + return f"heuristic rank from tags: {', '.join(sorted(set(boosts)))}" + return f"baseline rank for {target.language} file" + + +def _clamp_score(value: int) -> int: + return min(5, max(1, value)) + + +def _ranker_system_prompt() -> str: + return ( + "Score each file for security review priority. Return JSON only with a rankings array. " + "Each item must include path, surface, influence, reachability, rationale, and optional static_boosts. " + "Scores must be integers from 1 to 5." + ) + + +def _target_prompt_payload(target: FileTarget) -> dict[str, object]: + return { + "path": target.path, + "language": target.language, + "loc": target.loc, + "tags": target.tags, + "has_fuzz_entry_point": target.has_fuzz_entry_point, + } diff --git a/src/openultrasast/reports.py b/src/openultrasast/reports.py new file mode 100644 index 0000000..41b5909 --- /dev/null +++ b/src/openultrasast/reports.py @@ -0,0 +1,31 @@ +from __future__ import annotations + +from pathlib import Path + +from .findings import StaticFinding + + +def write_markdown_report(findings: list[StaticFinding], path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + lines = ["# OpenUltraSAST Report", "", f"Findings: {len(findings)}", ""] + if not findings: + lines.append("No quick-mode findings were emitted.") + for finding in findings: + lines.extend( + [ + f"## {finding.title}", + "", + f"- ID: `{finding.finding_id}`", + f"- Path: `{finding.path}`", + f"- Line: `{finding.line}`", + f"- Severity: `{finding.severity}`", + f"- Confidence: `{finding.confidence}`", + f"- Evidence level: `{finding.evidence_level}`", + f"- Ranking priority: `{finding.ranking_priority}`", + f"- Tags: `{', '.join(finding.tags)}`", + "", + finding.rationale, + "", + ] + ) + path.write_text("\n".join(lines).rstrip() + "\n") diff --git a/src/openultrasast/run.py b/src/openultrasast/run.py new file mode 100644 index 0000000..5bc7d5e --- /dev/null +++ b/src/openultrasast/run.py @@ -0,0 +1,26 @@ +from __future__ import annotations + +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path +from uuid import uuid4 + +from .config import ResolvedConfig, write_resolved_config + + +@dataclass(frozen=True) +class ScanRun: + scan_id: str + root: Path + target: Path + + +def create_scan_run(target: Path, config: ResolvedConfig) -> ScanRun: + resolved_target = target.resolve() + timestamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ") + scan_id = f"{timestamp}-{uuid4().hex[:8]}" + root = resolved_target / config.runs_dir / scan_id + root.mkdir(parents=True, exist_ok=False) + (root / "preprocess").mkdir() + write_resolved_config(config, root / "resolved_config.json") + return ScanRun(scan_id=scan_id, root=root, target=resolved_target) diff --git a/src/openultrasast/vectorstore.py b/src/openultrasast/vectorstore.py new file mode 100644 index 0000000..e169600 --- /dev/null +++ b/src/openultrasast/vectorstore.py @@ -0,0 +1,70 @@ +from __future__ import annotations + +from dataclasses import asdict, dataclass +from pathlib import Path + +from .index import VectorIndex, index_reuse_key, query_vector_index, read_vector_index, write_vector_index +from .preprocess import RepoSnapshot + + +SELECTED_LOCAL_VECTOR_STORE = "json-local" + + +@dataclass(frozen=True) +class VectorStoreBakeoffResult: + store: str + metadata_filtering: bool + local_persistence: bool + incremental_keying: bool + export_import: bool + retrieval_quality: bool + selected_default: bool + + @property + def passed(self) -> bool: + return all( + ( + self.metadata_filtering, + self.local_persistence, + self.incremental_keying, + self.export_import, + self.retrieval_quality, + self.selected_default, + ) + ) + + +def bakeoff_json_local(index: VectorIndex, output_path: Path, *, query_embedding: list[float]) -> VectorStoreBakeoffResult: + write_vector_index(index, output_path) + loaded = read_vector_index(output_path) + metadata_matches = query_vector_index(loaded, query_embedding, metadata_filter={"language": "python"}, limit=1) + unfiltered_matches = query_vector_index(loaded, query_embedding, limit=1) + snapshot = RepoSnapshot( + root=loaded.repo_root, + commit=loaded.repo_commit, + file_count=len({record.chunk.path for record in loaded.records}), + languages=_language_counts(loaded), + ) + result = VectorStoreBakeoffResult( + store=SELECTED_LOCAL_VECTOR_STORE, + metadata_filtering=bool(metadata_matches) and metadata_matches[0][1].metadata.get("language") == "python", + local_persistence=output_path.exists() and output_path.stat().st_size > 0, + incremental_keying=index_reuse_key(snapshot, loaded.embedding_model) == index_reuse_key(snapshot, loaded.embedding_model), + export_import=len(loaded.records) == len(index.records) and loaded.embedding_model == index.embedding_model, + retrieval_quality=bool(unfiltered_matches) and unfiltered_matches[0][0] > 0, + selected_default=SELECTED_LOCAL_VECTOR_STORE == "json-local", + ) + return result + + +def bakeoff_payload(result: VectorStoreBakeoffResult) -> dict[str, object]: + payload = asdict(result) + payload["passed"] = result.passed + return payload + + +def _language_counts(index: VectorIndex) -> dict[str, int]: + counts: dict[str, int] = {} + for record in index.records: + counts[record.chunk.language] = counts.get(record.chunk.language, 0) + 1 + return counts diff --git a/tests/test_calibration.py b/tests/test_calibration.py new file mode 100644 index 0000000..aca3065 --- /dev/null +++ b/tests/test_calibration.py @@ -0,0 +1,115 @@ +from pathlib import Path + +from openultrasast.calibration import ( + FalsePositiveReason, + FindingOutcome, + apply_ranking_calibration, + calibrate_ranking, + calibrate_rankings, + compute_ranking_metrics, + learning_from_finding_outcome, + record_false_positive_learning, + write_false_positive_learnings, +) +from openultrasast.findings import StaticFinding +from openultrasast.rank import RankingScore + + +def _finding(path: str = "auth/login.py") -> StaticFinding: + return StaticFinding( + finding_id="python-unsafe-eval:auth/login.py:1", + path=path, + title="Dynamic Python execution needs review", + severity="high", + confidence="medium", + evidence_level="static_corroboration", + rationale="pattern matched", + line=1, + tags=["syscall_entry"], + ranking_priority=4.0, + ) + + +def _ranking(path: str) -> RankingScore: + return RankingScore( + path=path, + surface=4, + influence=2, + reachability=4, + priority=3.6, + rationale="heuristic", + model_id=None, + static_boosts=["syscall_entry"], + ) + + +def test_false_positive_learning_records_reason_and_scope(tmp_path: Path) -> None: + learning = record_false_positive_learning( + _finding(), + reason=FalsePositiveReason.MISSING_ATTACKER_CONTROL, + evidence="Verifier showed input is constant and not attacker-controlled.", + ) + output = tmp_path / "false-positive-learnings.json" + + write_false_positive_learnings([learning], output) + + assert learning.vulnerability_class == "syscall_entry" + assert learning.scope == "auth:syscall_entry" + assert "missing_attacker_control" in output.read_text() + + +def test_ranking_metrics_track_false_positive_reasons_by_tier() -> None: + metrics = compute_ranking_metrics( + [ + {"tier": "A", "outcome": FindingOutcome.VERIFIED, "time_to_verification_seconds": 10}, + {"tier": "A", "outcome": FindingOutcome.REJECTED, "reason": FalsePositiveReason.UNREACHABLE_PATH, "time_to_verification_seconds": 20}, + {"tier": "B", "outcome": FindingOutcome.REJECTED, "reason": FalsePositiveReason.DUPLICATE}, + {"tier": "A", "outcome": FindingOutcome.UNVERIFIED, "reason": FalsePositiveReason.UNVERIFIED, "missed_fixture_vulnerabilities": 1}, + ], + tier="A", + ) + + assert metrics.verified_findings == 1 + assert metrics.false_positives == 2 + assert metrics.time_to_verification_seconds == 15.0 + assert metrics.missed_fixture_vulnerabilities == 1 + assert metrics.false_positive_reasons["unreachable_path"] == 1 + + +def test_scoped_false_positive_demotion_does_not_suppress_class_globally() -> None: + learning = record_false_positive_learning( + _finding("auth/login.py"), + reason=FalsePositiveReason.STATIC_RULE_MISMATCH, + evidence="Rule matched a test-only wrapper in this auth path.", + ) + + same_scope = apply_ranking_calibration(_ranking("auth/login.py"), calibrate_ranking(_ranking("auth/login.py"), [learning])) + other_scope = apply_ranking_calibration(_ranking("payments/login.py"), calibrate_ranking(_ranking("payments/login.py"), [learning])) + + assert same_scope.priority < 3.6 + assert other_scope.priority == 3.6 + + +def test_verifier_outcome_feeds_ranking_calibration_and_prompt_constraints() -> None: + learning = learning_from_finding_outcome( + _finding("auth/login.py"), + outcome=FindingOutcome.CONTRADICTED, + evidence="Independent verifier showed the risky call is unreachable in production flow.", + ) + assert learning is not None + + calibrated, calibrations = calibrate_rankings([_ranking("auth/login.py")], [learning]) + + assert calibrated[0].priority < 3.6 + assert calibrations[0].retrieval_filter_adjustments == {"syscall_entry": "auth:syscall_entry"} + assert "without new evidence" in calibrations[0].prompt_constraints[0] + + +def test_verified_outcome_does_not_create_false_positive_learning() -> None: + learning = learning_from_finding_outcome( + _finding(), + outcome=FindingOutcome.VERIFIED, + evidence="Verifier confirmed attacker control and impact.", + ) + + assert learning is None diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..ac71870 --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,37 @@ +from pathlib import Path + +from openultrasast.config import load_config, write_resolved_config + + +def test_load_config_reads_toml(tmp_path: Path) -> None: + config_path = tmp_path / "openultrasast.toml" + config_path.write_text( + "\n".join( + [ + "[models]", + 'ranker = "openrouter/test-ranker"', + "[sandbox]", + "memory_mb = 1024", + "[dynamic]", + "enabled = true", + 'network_scope = ["127.0.0.1:8080"]', + ] + ) + ) + + config = load_config(config_path) + + assert config.models.ranker == "openrouter/test-ranker" + assert config.embeddings.store == "json-local" + assert config.sandbox.memory_mb == 1024 + assert config.dynamic.enabled is True + assert config.dynamic.network_scope == ("127.0.0.1:8080",) + + +def test_write_resolved_config_creates_json(tmp_path: Path) -> None: + output = tmp_path / "run" / "resolved_config.json" + + write_resolved_config(load_config(None), output) + + assert output.exists() + assert '"minimum_report_verified": "static_corroboration"' in output.read_text() diff --git a/tests/test_findings.py b/tests/test_findings.py new file mode 100644 index 0000000..522960b --- /dev/null +++ b/tests/test_findings.py @@ -0,0 +1,53 @@ +import json +from pathlib import Path + +from openultrasast.findings import build_quick_hunter_prompt, quick_scan_findings, write_findings +from openultrasast.preprocess import preprocess_repository +from openultrasast.rank import rank_targets +from openultrasast.reports import write_markdown_report + + +def test_quick_scan_findings_emit_static_corroboration(tmp_path: Path) -> None: + repo = tmp_path / "repo" + repo.mkdir() + (repo / "parser.c").write_text("void parse(char *src) { char dst[8]; strcpy(dst, src); }\n") + _, targets = preprocess_repository(repo) + rankings = rank_targets(targets) + + findings = quick_scan_findings(repo, targets, rankings) + + assert len(findings) == 1 + assert findings[0].evidence_level == "static_corroboration" + assert findings[0].path == "parser.c" + assert findings[0].line == 1 + + +def test_findings_and_report_artifacts(tmp_path: Path) -> None: + repo = tmp_path / "repo" + repo.mkdir() + (repo / "runner.py").write_text("import subprocess\nsubprocess.run(cmd, shell=True)\n") + _, targets = preprocess_repository(repo) + findings = quick_scan_findings(repo, targets, rank_targets(targets)) + findings_path = tmp_path / "findings.json" + report_path = tmp_path / "report.md" + + write_findings(findings, findings_path) + write_markdown_report(findings, report_path) + + assert json.loads(findings_path.read_text())["findings"][0]["severity"] == "medium" + assert "Subprocess shell execution" in report_path.read_text() + + +def test_quick_hunter_prompt_is_bounded() -> None: + repo = Path("/repo/example.py") + target = next(iter(preprocess_repository(repo.parent)[1]), None) if repo.parent.exists() else None + assert target is None + class Target: + path = "example.py" + language = "python" + tags = ["auth_boundary"] + + prompt = build_quick_hunter_prompt(Target(), "x" * 5000) # type: ignore[arg-type] + + assert "example.py" in prompt + assert len(prompt) < 4300 diff --git a/tests/test_index.py b/tests/test_index.py new file mode 100644 index 0000000..c266628 --- /dev/null +++ b/tests/test_index.py @@ -0,0 +1,117 @@ +import json +from pathlib import Path + +import pytest + +from openultrasast.index import CodeChunk +from openultrasast.index import ( + build_code_chunks, + build_retrieval_package, + build_vector_index, + chunk_text_namespace, + index_reuse_key, + query_vector_index, + read_vector_index, + write_vector_index, +) +from openultrasast.preprocess import RepoSnapshot, preprocess_repository +from openultrasast.vectorstore import bakeoff_json_local, bakeoff_payload + + +class FakeEmbeddingClient: + def embed(self, *, model: str, inputs: list[str], timeout_seconds: int = 60) -> list[list[float]]: + return [[float(len(text)), 1.0] for text in inputs] + + +def test_build_code_chunks_preserves_metadata_filters(tmp_path: Path) -> None: + repo = tmp_path / "repo" + repo.mkdir() + (repo / "app.py").write_text("def login(user):\n return user\n") + _, targets = preprocess_repository(repo) + + chunks = build_code_chunks(repo, targets, max_lines=1) + + assert len(chunks) == 2 + assert chunks[0].metadata["path"] == "app.py" + assert chunks[0].metadata["language"] == "python" + assert chunks[0].start_line == 1 + + +def test_json_vector_index_round_trips_and_filters(tmp_path: Path) -> None: + repo = tmp_path / "repo" + repo.mkdir() + (repo / "app.py").write_text("def login(user):\n return user\n") + snapshot, targets = preprocess_repository(repo) + chunks = build_code_chunks(repo, targets) + index = build_vector_index(snapshot=snapshot, chunks=chunks, embedding_model="openrouter/embed", client=FakeEmbeddingClient()) + output = tmp_path / "index.json" + + write_vector_index(index, output) + loaded = read_vector_index(output) + matches = query_vector_index(loaded, [20.0, 1.0], metadata_filter={"language": "python"}) + + assert loaded.embedding_model == "openrouter/embed" + assert matches[0][1].path == "app.py" + assert json.loads(output.read_text())["store"] == "json-local" + + +def test_vector_index_rejects_embedding_count_mismatch() -> None: + class EmptyEmbeddingClient: + def embed(self, *, model: str, inputs: list[str], timeout_seconds: int = 60) -> list[list[float]]: + return [] + + chunk = CodeChunk( + chunk_id="chunk-1", + namespace="repo_code", + path="app.py", + language="python", + start_line=1, + end_line=1, + text="print('x')", + metadata={"path": "app.py", "language": "python"}, + ) + snapshot = RepoSnapshot(root="/repo", commit=None, file_count=0, languages={}) + with pytest.raises(ValueError): + build_vector_index(snapshot=snapshot, chunks=[chunk], embedding_model="model", client=EmptyEmbeddingClient()) + + +def test_namespace_chunks_and_reuse_key_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, + ) + snapshot = RepoSnapshot(root="/repo", commit="abc", file_count=1, languages={"python": 1}) + + assert len(chunks) == 2 + assert chunks[0].namespace == "skills" + assert chunks[0].metadata["vulnerability_class"] == "mapping" + assert index_reuse_key(snapshot, "embed-a") != index_reuse_key(snapshot, "embed-b") + + +def test_retrieval_package_is_bounded_by_character_budget() -> None: + chunks = chunk_text_namespace(namespace="docs", path="README.md", text="alpha\n" + "beta" * 100, max_lines=1) + snapshot = RepoSnapshot(root="/repo", commit=None, file_count=1, languages={"markdown": 1}) + index = build_vector_index(snapshot=snapshot, chunks=chunks, embedding_model="model", client=FakeEmbeddingClient()) + + package = build_retrieval_package(role="verifier", index=index, query_embedding=[4.0, 1.0], max_chars=80) + + assert package.role == "verifier" + assert package.hits + assert package.truncated is True + + +def test_json_local_bakeoff_selects_default_store(tmp_path: Path) -> None: + repo = tmp_path / "repo" + repo.mkdir() + (repo / "app.py").write_text("def login(user):\n return user\n") + snapshot, targets = preprocess_repository(repo) + chunks = build_code_chunks(repo, targets) + index = build_vector_index(snapshot=snapshot, chunks=chunks, embedding_model="model", client=FakeEmbeddingClient()) + + result = bakeoff_json_local(index, tmp_path / "vector-index.json", query_embedding=[20.0, 1.0]) + + assert result.passed is True + assert bakeoff_payload(result)["selected_default"] is True diff --git a/tests/test_preprocess.py b/tests/test_preprocess.py new file mode 100644 index 0000000..2706981 --- /dev/null +++ b/tests/test_preprocess.py @@ -0,0 +1,43 @@ +import json +from pathlib import Path + +from openultrasast.preprocess import detect_language, preprocess_repository + + +def test_preprocess_emits_file_targets_with_tags(tmp_path: Path) -> None: + repo = tmp_path / "repo" + repo.mkdir() + (repo / ".gitignore").write_text("ignored.c\n") + (repo / "parser.c").write_text( + "\n".join( + [ + "int LLVMFuzzerTestOneInput(const unsigned char *data, int size) {", + " parse_frame(data, size);", + " return 0;", + "}", + ] + ) + ) + (repo / "auth.py").write_text("def login(session):\n return open(session)\n") + (repo / "ignored.c").write_text("int ignored(void) { return 0; }\n") + + output = tmp_path / "file_targets.json" + snapshot, targets = preprocess_repository(repo, output) + + assert snapshot.file_count == 2 + by_path = {target.path: target for target in targets} + assert by_path["parser.c"].language == "c" + assert by_path["parser.c"].has_fuzz_entry_point is True + assert set(by_path["parser.c"].tags) >= {"memory_unsafe", "parser", "fuzzable"} + assert set(by_path["auth.py"].tags) >= {"auth_boundary", "filesystem_entry"} + + payload = json.loads(output.read_text()) + assert payload["snapshot"]["file_count"] == 2 + assert [item["path"] for item in payload["file_targets"]] == ["auth.py", "parser.c"] + + +def test_detect_language_uses_shebang(tmp_path: Path) -> None: + script = tmp_path / "tool" + script.write_text("#!/usr/bin/env python3\nprint('ok')\n") + + assert detect_language(script) == "python" diff --git a/tests/test_rank.py b/tests/test_rank.py new file mode 100644 index 0000000..c56791e --- /dev/null +++ b/tests/test_rank.py @@ -0,0 +1,112 @@ +import json +from pathlib import Path + +import pytest + +from openultrasast.preprocess import FileTarget +from openultrasast.provider.openrouter import OpenRouterError, parse_json_content +from openultrasast.rank import ( + RankingError, + composite_priority, + parse_ranker_response, + rank_targets, + rank_targets_with_model, + write_rankings, +) + + +class FakeRankerClient: + def __init__(self) -> None: + self.calls = 0 + + def complete_json(self, *, model: str, messages: list[dict[str, str]], timeout_seconds: int = 60) -> object: + self.calls += 1 + assert model == "openrouter/test-ranker" + return { + "rankings": [ + { + "path": "a.py" if self.calls == 1 else "b.py", + "surface": 2, + "influence": 3, + "reachability": 4, + "rationale": "model rationale", + "static_boosts": ["model"], + } + ] + } + + +def test_composite_priority_uses_spec_formula() -> None: + assert composite_priority(surface=5, influence=4, reachability=3) == 4.2 + + +def test_heuristic_ranking_prioritizes_fuzzable_parser() -> None: + targets = [ + FileTarget( + path="parser.c", + absolute_path="/repo/parser.c", + language="c", + loc=10, + tags=["memory_unsafe", "parser", "fuzzable"], + has_fuzz_entry_point=True, + ), + FileTarget( + path="util.py", + absolute_path="/repo/util.py", + language="python", + loc=5, + tags=[], + has_fuzz_entry_point=False, + ), + ] + + rankings = rank_targets(targets) + + assert rankings[0].path == "parser.c" + assert rankings[0].surface == 5 + assert rankings[0].reachability == 4 + + +def test_parse_ranker_response_rejects_malformed_scores() -> None: + with pytest.raises(RankingError): + parse_ranker_response({"rankings": [{"path": "a.py", "surface": 9, "influence": 1, "reachability": 1}]}, "model") + + +def test_rank_targets_with_model_chunks_requests() -> None: + targets = [ + FileTarget(path="a.py", absolute_path="/repo/a.py", language="python", loc=1, tags=[], has_fuzz_entry_point=False), + FileTarget(path="b.py", absolute_path="/repo/b.py", language="python", loc=1, tags=[], has_fuzz_entry_point=False), + ] + client = FakeRankerClient() + + rankings = rank_targets_with_model(targets, client=client, model="openrouter/test-ranker", chunk_size=1) + + assert client.calls == 2 + assert {ranking.model_id for ranking in rankings} == {"openrouter/test-ranker"} + + +def test_parse_json_content_accepts_fenced_json() -> None: + assert parse_json_content('```json\n{"rankings": []}\n```') == {"rankings": []} + + +def test_parse_json_content_rejects_non_json() -> None: + with pytest.raises(OpenRouterError): + parse_json_content("not json") + + +def test_write_rankings_emits_artifact(tmp_path: Path) -> None: + target = FileTarget( + path="auth.py", + absolute_path="/repo/auth.py", + language="python", + loc=5, + tags=["auth_boundary"], + has_fuzz_entry_point=False, + ) + output = tmp_path / "rank" / "ranking.json" + + write_rankings(rank_targets([target]), output) + + payload = json.loads(output.read_text()) + assert payload["rankings"][0]["path"] == "auth.py" + assert payload["rankings"][0]["priority"] >= 1 diff --git a/tests/test_run.py b/tests/test_run.py new file mode 100644 index 0000000..564bf68 --- /dev/null +++ b/tests/test_run.py @@ -0,0 +1,13 @@ +from pathlib import Path + +from openultrasast.config import load_config +from openultrasast.run import create_scan_run + + +def test_create_scan_run_writes_directories_and_config(tmp_path: Path) -> None: + run = create_scan_run(tmp_path, load_config(None)) + + assert run.root.exists() + assert (run.root / "preprocess").is_dir() + assert (run.root / "resolved_config.json").is_file() + assert run.root.parent.name == "runs" diff --git a/uv.lock b/uv.lock new file mode 100644 index 0000000..7ce32ba --- /dev/null +++ b/uv.lock @@ -0,0 +1,79 @@ +version = 1 +revision = 3 +requires-python = ">=3.11" + +[[package]] +name = "colorama" +version = "0.4.6" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d8/53/6f443c9a4a8358a93a6792e2acffb9d9d5cb0a5cfd8802644b7b1c9a02e4/colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44", size = 27697, upload-time = "2022-10-25T02:36:22.414Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/d1/d6/3965ed04c63042e047cb6a3e6ed1a63a35087b6a609aa3a15ed8ac56c221/colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6", size = 25335, upload-time = "2022-10-25T02:36:20.889Z" }, +] + +[[package]] +name = "iniconfig" +version = "2.3.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/72/34/14ca021ce8e5dfedc35312d08ba8bf51fdd999c576889fc2c24cb97f4f10/iniconfig-2.3.0.tar.gz", hash = "sha256:c76315c77db068650d49c5b56314774a7804df16fee4402c1f19d6d15d8c4730", size = 20503, upload-time = "2025-10-18T21:55:43.219Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484, upload-time = "2025-10-18T21:55:41.639Z" }, +] + +[[package]] +name = "openultrasast" +version = "0.1.0" +source = { editable = "." } + +[package.dev-dependencies] +dev = [ + { name = "pytest" }, +] + +[package.metadata] + +[package.metadata.requires-dev] +dev = [{ name = "pytest", specifier = ">=8" }] + +[[package]] +name = "packaging" +version = "26.2" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d7/f1/e7a6dd94a8d4a5626c03e4e99c87f241ba9e350cd9e6d75123f992427270/packaging-26.2.tar.gz", hash = "sha256:ff452ff5a3e828ce110190feff1178bb1f2ea2281fa2075aadb987c2fb221661", size = 228134, upload-time = "2026-04-24T20:15:23.917Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/df/b2/87e62e8c3e2f4b32e5fe99e0b86d576da1312593b39f47d8ceef365e95ed/packaging-26.2-py3-none-any.whl", hash = "sha256:5fc45236b9446107ff2415ce77c807cee2862cb6fac22b8a73826d0693b0980e", size = 100195, upload-time = "2026-04-24T20:15:22.081Z" }, +] + +[[package]] +name = "pluggy" +version = "1.6.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/f9/e2/3e91f31a7d2b083fe6ef3fa267035b518369d9511ffab804f839851d2779/pluggy-1.6.0.tar.gz", hash = "sha256:7dcc130b76258d33b90f61b658791dede3486c3e6bfb003ee5c9bfb396dd22f3", size = 69412, upload-time = "2025-05-15T12:30:07.975Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/54/20/4d324d65cc6d9205fabedc306948156824eb9f0ee1633355a8f7ec5c66bf/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538, upload-time = "2025-05-15T12:30:06.134Z" }, +] + +[[package]] +name = "pygments" +version = "2.20.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/c3/b2/bc9c9196916376152d655522fdcebac55e66de6603a76a02bca1b6414f6c/pygments-2.20.0.tar.gz", hash = "sha256:6757cd03768053ff99f3039c1a36d6c0aa0b263438fcab17520b30a303a82b5f", size = 4955991, upload-time = "2026-03-29T13:29:33.898Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/f4/7e/a72dd26f3b0f4f2bf1dd8923c85f7ceb43172af56d63c7383eb62b332364/pygments-2.20.0-py3-none-any.whl", hash = "sha256:81a9e26dd42fd28a23a2d169d86d7ac03b46e2f8b59ed4698fb4785f946d0176", size = 1231151, upload-time = "2026-03-29T13:29:30.038Z" }, +] + +[[package]] +name = "pytest" +version = "9.1.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "colorama", marker = "sys_platform == 'win32'" }, + { name = "iniconfig" }, + { name = "packaging" }, + { name = "pluggy" }, + { name = "pygments" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/84/0e/b5858858d74958632c49b72cb25a3976ff9f632397626715be71c89d3971/pytest-9.1.0.tar.gz", hash = "sha256:41dd9148c08072446394cefd3d79701701335a9f4cae69ba92e39f6c7f5c061c", size = 1634181, upload-time = "2026-06-13T18:52:45.983Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/8b/5a/ba30a81239b909821b3153e303e7def45178bf353da4f72380e6c5e8793b/pytest-9.1.0-py3-none-any.whl", hash = "sha256:8ebb0e7888bdf2bdfc602ec51f8f62d50200af37356c74e503c79a94f5c81f32", size = 386453, upload-time = "2026-06-13T18:52:44.045Z" }, +]