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

feat(detection): language-scoped rules + multi-language benchmark corpora

Browse Source 
Rebuild the quick-scan detection engine and ground-truth benchmarks to hit
>=90% recall / <10% false positives for C/C++, Python, JavaScript, and Java.

Engine (findings.py, preprocess.py):
- Scope each PatternRule by language (fixes cross-language false positives such
  as the Python eval/exec rule firing on JS .exec()).
- Add JavaScript/TypeScript rules (command exec, eval/Function, reflected XSS,
  SQLi, path traversal, SSRF, weak hash, deserialization); execFile/spawn-argv
  excluded as safe alternatives.
- Add Java rules (Runtime exec, SQL string concatenation, weak hash, readObject)
  and a Groovy server-side template rule (yieldUnescaped of a bound variable).
- Broaden C/C++ and Python coverage; precision guards for format strings and
  parameterised vs concatenated SQL; #-comment skipping; CWE in rationale.
- Register .java/.groovy/.tpl source extensions.

Benchmarks:
- Add realistic corpora modeled on Damn Vulnerable C/C++, PyGoat/DVPWA, and
  NodeGoat, plus the real kiview/damn-vulnerable-spring-boot-app vendored
  verbatim (MIT, attribution in LICENSE/NOTICE.md). Each manifest carries
  complete, independently-labeled ground truth including genuinely
  hard-to-detect vulns (integer overflow, second-order SQLi, prototype
  pollution) so recall is measured honestly, plus safe-API files for precision.
- Fix pre-existing smoke-manifest ground-truth errors (Python/JS line offsets,
  missing C/C++ strcpy entry, Java SQLi line).

benchmark.py:
- Fix sink matching to normalise the expected sink consistently so sinks with
  underscores (e.g. render_template_string) can match.

Tests:
- Add tests/test_detection_benchmarks.py enforcing the >=90% recall / <10% FP
  goal per language and overall.

Measured: C/C++ 93.3%, Python 93.3%, JavaScript 92.3%, Java 100%;
overall 93.6% recall, 0.0% false-positive rate.

Co-Authored-By: Claude Opus 4.8 <[email protected]>
This commit is contained in:
Marius Ciepluch
2026-06-21 11:49:31 +02:00
co-authored by Claude Opus 4.8
parent be65aed96e
commit cf9f945992
38 changed files with 2420 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/openultrasast/benchmark.py
+425
View File
@@ -0,0 +1,425 @@
from __future__ import annotations
import json
import tomllib
from dataclasses import asdict, dataclass
from datetime import UTC, datetime
from pathlib import Path
from uuid import uuid4
from .findings import StaticFinding
@dataclass(frozen=True)
class BenchmarkSource:
type: str
path: str
@dataclass(frozen=True)
class ExpectedFinding:
cwe: str
vulnerability_class: str
path: str
evidence: str
rule_id: str | None = None
line: int | None = None
function: str | None = None
sink: str | None = None
@dataclass(frozen=True)
class BenchmarkManifest:
name: str
language: str
frameworks: list[str]
setup: list[str]
source: BenchmarkSource
modes: list[str]
expected: list[ExpectedFinding]
known_noise: list[str]
baselines: list[str]
@dataclass(frozen=True)
class BaselineFinding:
tool: str
finding_id: str
path: str
line: int | None
rule_id: str | None
cwe: str | None
vulnerability_class: str | None
evidence: str
@dataclass(frozen=True)
class BenchmarkMiss:
cwe: str
vulnerability_class: str
path: str
evidence: str
reason: str
@dataclass(frozen=True)
class BenchmarkBaselineDelta:
tool: str
matched_expected_total: int
missed_expected_total: int
extra_findings_total: int
@dataclass(frozen=True)
class BenchmarkCalibrationRecord:
cwe: str
vulnerability_class: str
path: str
evidence: str
failed_stage: str
next_improvement_candidate: str
reason: str
@dataclass(frozen=True)
class BenchmarkFalsePositive:
finding_id: str
path: str
line: int | None
reason: str
@dataclass(frozen=True)
class BenchmarkMetrics:
expected_findings_total: int
actual_findings_total: int
matched_findings_total: int
missed_findings_total: int
false_positive_findings_total: int
recall: float | None
precision: float | None
runtime_seconds: float | None
model_usage: dict[str, object]
artifact_links: dict[str, str]
@dataclass(frozen=True)
class BenchmarkRun:
benchmark_run_id: str
root: Path
manifest: BenchmarkManifest
@dataclass(frozen=True)
class BenchmarkResult:
benchmark_run_id: str
benchmark_name: str
language: str
mode: str
scan_id: str | None
scan_run_dir: str | None
metrics: BenchmarkMetrics
misses: list[BenchmarkMiss]
false_positives: list[BenchmarkFalsePositive]
baseline_deltas: list[BenchmarkBaselineDelta]
calibration_records: list[BenchmarkCalibrationRecord]
def load_benchmark_manifest(path: Path) -> BenchmarkManifest:
payload = tomllib.loads(path.read_text())
source = payload.get("source", {})
expected = payload.get("expected", [])
return BenchmarkManifest(
name=str(payload["name"]),
language=str(payload["language"]),
frameworks=[str(item) for item in payload.get("frameworks", [])],
setup=[str(item) for item in payload.get("setup", [])],
source=BenchmarkSource(type=str(source.get("type", "local")), path=str(source["path"])),
modes=[str(item) for item in payload.get("modes", [])],
expected=[
ExpectedFinding(
cwe=str(item["cwe"]),
vulnerability_class=str(item.get("class", item.get("vulnerability_class", "unknown"))),
path=str(item["path"]),
evidence=str(item.get("evidence", "")),
rule_id=str(item["rule_id"]) if "rule_id" in item else None,
line=int(item["line"]) if "line" in item else None,
function=str(item["function"]) if "function" in item else None,
sink=str(item["sink"]) if "sink" in item else None,
)
for item in expected
],
known_noise=[str(item) for item in payload.get("known_noise", [])],
baselines=[str(item) for item in payload.get("baselines", [])],
)
def resolve_benchmark_source(manifest_path: Path, manifest: BenchmarkManifest) -> Path:
if manifest.source.type != "local":
raise ValueError(f"unsupported benchmark source type: {manifest.source.type}")
source = Path(manifest.source.path)
if not source.is_absolute():
source = manifest_path.parent / source
return source.resolve()
def create_benchmark_run(target: Path, manifest: BenchmarkManifest) -> BenchmarkRun:
timestamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")
benchmark_run_id = f"{timestamp}-{uuid4().hex[:8]}"
root = target.resolve() / ".openultrasast" / "benchmarks" / benchmark_run_id
root.mkdir(parents=True, exist_ok=False)
return BenchmarkRun(benchmark_run_id=benchmark_run_id, root=root, manifest=manifest)
def load_findings(path: Path) -> list[StaticFinding]:
payload = json.loads(path.read_text())
return [StaticFinding(**item) for item in payload.get("findings", [])]
def load_baseline_findings(manifest_path: Path, manifest: BenchmarkManifest) -> list[BaselineFinding]:
baseline_findings: list[BaselineFinding] = []
for baseline in manifest.baselines:
path = Path(baseline)
if not path.is_absolute():
path = manifest_path.parent / path
if not path.exists():
continue
payload = json.loads(path.read_text())
items = payload.get("findings", payload if isinstance(payload, list) else [])
for item in items:
baseline_findings.append(
BaselineFinding(
tool=str(item.get("tool", path.stem)),
finding_id=str(item.get("finding_id", item.get("id", ""))),
path=str(item.get("path", "")),
line=int(item["line"]) if "line" in item and item["line"] is not None else None,
rule_id=str(item["rule_id"]) if "rule_id" in item else None,
cwe=str(item["cwe"]) if "cwe" in item else None,
vulnerability_class=str(item.get("class", item.get("vulnerability_class", ""))) or None,
evidence=str(item.get("evidence", "")),
)
)
return baseline_findings
def evaluate_benchmark(
*,
run: BenchmarkRun,
mode: str,
findings: list[StaticFinding],
scan_id: str | None,
scan_run_dir: Path | None,
runtime_seconds: float | None = None,
baseline_findings: list[BaselineFinding] | None = None,
) -> BenchmarkResult:
matched_finding_indexes: set[int] = set()
misses: list[BenchmarkMiss] = []
for expected in run.manifest.expected:
match_index = _matching_finding_index(expected, findings, matched_finding_indexes)
if match_index is None:
misses.append(_miss(expected))
else:
matched_finding_indexes.add(match_index)
false_positives = [_false_positive(finding) for index, finding in enumerate(findings) if index not in matched_finding_indexes]
baseline_deltas = _baseline_deltas(run.manifest.expected, baseline_findings or [])
calibration_records = [_calibration_record(miss) for miss in misses]
expected_total = len(run.manifest.expected)
actual_total = len(findings)
missed_total = len(misses)
matched_total = expected_total - missed_total
false_positive_total = len(false_positives)
recall = matched_total / expected_total if expected_total else None
precision = matched_total / actual_total if actual_total else None
return BenchmarkResult(
benchmark_run_id=run.benchmark_run_id,
benchmark_name=run.manifest.name,
language=run.manifest.language,
mode=mode,
scan_id=scan_id,
scan_run_dir=str(scan_run_dir) if scan_run_dir else None,
metrics=BenchmarkMetrics(
expected_findings_total=expected_total,
actual_findings_total=actual_total,
matched_findings_total=matched_total,
missed_findings_total=missed_total,
false_positive_findings_total=false_positive_total,
recall=recall,
precision=precision,
runtime_seconds=runtime_seconds,
model_usage={},
artifact_links=_artifact_links(scan_run_dir),
),
misses=misses,
false_positives=false_positives,
baseline_deltas=baseline_deltas,
calibration_records=calibration_records,
)
def write_benchmark_artifacts(run: BenchmarkRun, manifest_path: Path, result: BenchmarkResult) -> None:
(run.root / "benchmark_manifest.toml").write_text(manifest_path.read_text())
(run.root / "benchmark_result.json").write_text(json.dumps(_jsonable(asdict(result)), indent=2, sort_keys=True) + "\n")
(run.root / "calibration_records.json").write_text(
json.dumps(_jsonable([asdict(record) for record in result.calibration_records]), indent=2, sort_keys=True) + "\n"
)
(run.root / "external_baseline_deltas.json").write_text(
json.dumps(_jsonable([asdict(delta) for delta in result.baseline_deltas]), indent=2, sort_keys=True) + "\n"
)
def _matching_finding_index(expected: ExpectedFinding, findings: list[StaticFinding], used_indexes: set[int]) -> int | None:
for index, finding in enumerate(findings):
if index in used_indexes:
continue
if _matches_expected(expected, finding):
return index
return None
def _matches_expected(expected: ExpectedFinding, finding: StaticFinding) -> bool:
expected_path = expected.path.strip("*")
if expected.rule_id and not finding.finding_id.startswith(f"{expected.rule_id}:"):
return False
if expected.line is not None and finding.line != expected.line:
return False
if expected_path and expected_path not in finding.path:
return False
if expected.function and finding.function_name != expected.function:
return False
if expected.sink and _normalize(expected.sink) not in _finding_text(finding):
return False
if expected.evidence and not _evidence_matches(expected.evidence, finding):
return False
return _cwe_matches(expected.cwe, finding)
def _miss(expected: ExpectedFinding) -> BenchmarkMiss:
return BenchmarkMiss(
cwe=expected.cwe,
vulnerability_class=expected.vulnerability_class,
path=expected.path,
evidence=expected.evidence,
reason="no OpenUltraSAST finding matched the expected benchmark vulnerability",
)
def _false_positive(finding: StaticFinding) -> BenchmarkFalsePositive:
return BenchmarkFalsePositive(
finding_id=finding.finding_id,
path=finding.path,
line=finding.line,
reason="finding did not match any expected benchmark vulnerability",
)
def _baseline_deltas(expected_findings: list[ExpectedFinding], baseline_findings: list[BaselineFinding]) -> list[BenchmarkBaselineDelta]:
tools = sorted({finding.tool for finding in baseline_findings})
deltas: list[BenchmarkBaselineDelta] = []
for tool in tools:
tool_findings = [finding for finding in baseline_findings if finding.tool == tool]
matched_indexes: set[int] = set()
matched_expected_total = 0
for expected in expected_findings:
match_index = _matching_baseline_index(expected, tool_findings, matched_indexes)
if match_index is not None:
matched_indexes.add(match_index)
matched_expected_total += 1
deltas.append(
BenchmarkBaselineDelta(
tool=tool,
matched_expected_total=matched_expected_total,
missed_expected_total=len(expected_findings) - matched_expected_total,
extra_findings_total=len(tool_findings) - len(matched_indexes),
)
)
return deltas
def _matching_baseline_index(expected: ExpectedFinding, findings: list[BaselineFinding], used_indexes: set[int]) -> int | None:
for index, finding in enumerate(findings):
if index in used_indexes:
continue
if _baseline_matches_expected(expected, finding):
return index
return None
def _baseline_matches_expected(expected: ExpectedFinding, finding: BaselineFinding) -> bool:
expected_path = expected.path.strip("*")
if expected.rule_id and finding.rule_id and finding.rule_id != expected.rule_id:
return False
if expected.line is not None and finding.line is not None and finding.line != expected.line:
return False
if expected_path and expected_path not in finding.path:
return False
if finding.cwe and finding.cwe.lower() != expected.cwe.lower():
return False
return not (expected.sink and expected.sink.lower() not in _normalize(finding.evidence))
def _calibration_record(miss: BenchmarkMiss) -> BenchmarkCalibrationRecord:
return BenchmarkCalibrationRecord(
cwe=miss.cwe,
vulnerability_class=miss.vulnerability_class,
path=miss.path,
evidence=miss.evidence,
failed_stage="benchmark_ground_truth_matching",
next_improvement_candidate="rules_or_language_hunter",
reason=miss.reason,
)
def _artifact_links(scan_run_dir: Path | None) -> dict[str, str]:
if scan_run_dir is None:
return {}
return {
"findings": str(scan_run_dir / "findings.json"),
"manifest": str(scan_run_dir / "manifest.json"),
"report": str(scan_run_dir / "report.md"),
"sarif": str(scan_run_dir / "report.sarif"),
}
def _cwe_matches(cwe: str, finding: StaticFinding) -> bool:
text = _finding_text(finding)
cwe_lower = cwe.lower()
if "cwe-" not in text:
return True
return cwe_lower in text
def _evidence_matches(evidence: str, finding: StaticFinding) -> bool:
tokens = [token for token in _normalize(evidence).split() if len(token) > 2]
if not tokens:
return True
text = _finding_text(finding)
return any(token in text for token in tokens)
def _finding_text(finding: StaticFinding) -> str:
parts = [
finding.finding_id,
finding.title,
finding.rationale,
finding.function_name or "",
" ".join(finding.tags),
" ".join(finding.reachability_conditions),
]
return _normalize(" ".join(parts))
def _normalize(value: str) -> str:
normalized = value.lower().replace("memcpy", "memcpy memory copy")
normalized = normalized.replace("copies", "copy")
return " ".join(normalized.replace("-", " ").replace("_", " ").split())
def _jsonable(value: object) -> object:
if isinstance(value, Path):
return str(value)
if isinstance(value, dict):
return {key: _jsonable(item) for key, item in value.items()}
if isinstance(value, list):
return [_jsonable(item) for item in value]
return value
src/openultrasast/cli.py
+85 -7
View File
@@ -2,10 +2,21 @@ from __future__ import annotations
import argparse
import json
from dataclasses import dataclass
from pathlib import Path
from time import perf_counter
from .benchmark import (
create_benchmark_run,
evaluate_benchmark,
load_baseline_findings,
load_benchmark_manifest,
load_findings,
resolve_benchmark_source,
write_benchmark_artifacts,
)
from .config import load_config
from .findings import quick_scan_findings, write_findings
from .findings import StaticFinding, 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
@@ -17,6 +28,16 @@ from .run import create_scan_run
from .verification import verify_findings, write_verification_results
@dataclass(frozen=True)
class ScanOutcome:
scan_id: str
run_dir: Path
file_target_count: int
ranked_target_count: int
finding_count: int
exit_code: int
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(prog="ousast")
subparsers = parser.add_subparsers(dest="command", required=True)
@@ -32,15 +53,28 @@ def main(argv: list[str] | None = None) -> int:
index.add_argument("--config", type=Path, default=Path("openultrasast.toml"))
index.add_argument("--chunk-lines", type=int, default=80)
benchmark = subparsers.add_parser("benchmark", help="run a benchmark manifest and write scoreboard artifacts")
benchmark.add_argument("manifest", type=Path)
benchmark.add_argument("--mode", choices=("quick", "standard", "deep"), default="quick")
benchmark.add_argument("--config", type=Path, default=Path("openultrasast.toml"))
args = parser.parse_args(argv)
if args.command == "scan":
return _scan(args.path, args.config, args.mode, args.fail_on)
if args.command == "index":
return _index(args.path, args.config, args.chunk_lines)
if args.command == "benchmark":
return _benchmark(args.manifest, args.config, args.mode)
return 2
def _scan(path: Path, config_path: Path, mode: str, fail_on: str) -> int:
outcome = _run_scan(path, config_path, mode, fail_on)
_print_scan_outcome(outcome)
return outcome.exit_code
def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOutcome:
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():
@@ -104,12 +138,22 @@ def _scan(path: Path, config_path: Path, mode: str, fail_on: str) -> int:
),
)
runtime.finish(status="succeeded")
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 scan_exit_code(findings, verifications, fail_on)
return ScanOutcome(
scan_id=run.scan_id,
run_dir=run.root,
file_target_count=len(targets),
ranked_target_count=len(rankings),
finding_count=len(findings),
exit_code=scan_exit_code(findings, verifications, fail_on),
)
def _print_scan_outcome(outcome: ScanOutcome) -> None:
print(f"scan_id={outcome.scan_id}")
print(f"run_dir={outcome.run_dir}")
print(f"file_targets={outcome.file_target_count}")
print(f"ranked_targets={outcome.ranked_target_count}")
print(f"findings={outcome.finding_count}")
def _load_static_hints(sarif_paths: tuple[str, ...]) -> list[object]:
@@ -125,6 +169,40 @@ def _artifact_paths(**paths: Path | None) -> dict[str, Path]:
return {name: path for name, path in paths.items() if path is not None}
def _benchmark(manifest_path: Path, config_path: Path, mode: str) -> int:
if not manifest_path.exists() or not manifest_path.is_file():
raise SystemExit(f"benchmark manifest is not a file: {manifest_path}")
manifest = load_benchmark_manifest(manifest_path)
target = resolve_benchmark_source(manifest_path, manifest)
run = create_benchmark_run(target, manifest)
started_at = perf_counter()
outcome = _run_scan(target, config_path, mode, "never")
runtime_seconds = perf_counter() - started_at
findings = _load_scan_findings(outcome.run_dir)
baseline_findings = load_baseline_findings(manifest_path, manifest)
result = evaluate_benchmark(
run=run,
mode=mode,
findings=findings,
scan_id=outcome.scan_id,
scan_run_dir=outcome.run_dir,
runtime_seconds=runtime_seconds,
baseline_findings=baseline_findings,
)
write_benchmark_artifacts(run, manifest_path, result)
_print_scan_outcome(outcome)
print(f"benchmark_run_id={run.benchmark_run_id}")
print(f"benchmark_run_dir={run.root}")
print(f"benchmark_expected={result.metrics.expected_findings_total}")
print(f"benchmark_matched={result.metrics.matched_findings_total}")
print(f"benchmark_missed={result.metrics.missed_findings_total}")
return 0
def _load_scan_findings(run_dir: Path) -> list[StaticFinding]:
return load_findings(run_dir / "findings.json")
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}")
src/openultrasast/findings.py
+324 -8
View File
@@ -34,29 +34,92 @@ class PatternRule:
severity: str
tags: tuple[str, ...]
pattern: re.Pattern[str]
cwe: str
languages: tuple[str, ...]
C_LANGUAGES = ("c", "cpp")
PYTHON_LANGUAGES = ("python",)
JS_LANGUAGES = ("javascript", "typescript")
JAVA_LANGUAGES = ("java",)
GROOVY_LANGUAGES = ("groovy",)
PATTERN_RULES = (
# --- C / C++ -----------------------------------------------------------------
PatternRule(
rule_id="c-unsafe-copy",
title="Unsafe C string or input function needs review",
title="Unsafe C string copy needs review",
severity="high",
tags=("memory_unsafe",),
pattern=re.compile(r"\b(gets|strcpy|strcat|sprintf)\s*\("),
cwe="CWE-120",
languages=C_LANGUAGES,
),
PatternRule(
rule_id="c-unsafe-memory-copy",
title="Unsafe C memory copy needs review",
severity="high",
tags=("memory_unsafe",),
pattern=re.compile(r"\b(memcpy|memmove|bcopy)\s*\("),
cwe="CWE-120",
languages=C_LANGUAGES,
),
PatternRule(
rule_id="c-unsafe-format",
title="Unsafe C format string needs review",
severity="high",
tags=("memory_unsafe",),
# Format function whose format argument is a bare identifier (not a string
# literal), i.e. the classic format-string sink. printf("...") is ignored.
pattern=re.compile(r"\b(printf|vprintf|vsprintf|vsnprintf|vfprintf|syslog)\s*\(\s*[A-Za-z_]"),
cwe="CWE-134",
languages=C_LANGUAGES,
),
PatternRule(
rule_id="c-unsafe-scanf",
title="Unbounded C scanf needs review",
severity="high",
tags=("memory_unsafe",),
pattern=re.compile(r"\b(scanf|fscanf|sscanf)\s*\([^\n]*%s"),
cwe="CWE-120",
languages=C_LANGUAGES,
),
PatternRule(
rule_id="c-shell-exec",
title="C shell execution needs review",
severity="high",
tags=("syscall_entry",),
pattern=re.compile(r"\b(system|popen)\s*\("),
cwe="CWE-78",
languages=C_LANGUAGES,
),
PatternRule(
rule_id="c-exec-family",
title="C exec command execution needs review",
severity="high",
tags=("syscall_entry",),
pattern=re.compile(r"\b(execl|execlp|execle|execv|execvp|execvpe)\s*\("),
cwe="CWE-78",
languages=C_LANGUAGES,
),
# --- Python ------------------------------------------------------------------
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*\("),
cwe="CWE-95",
languages=PYTHON_LANGUAGES,
),
PatternRule(
rule_id="python-unsafe-deserialization",
title="Unsafe Python deserialization needs review",
rule_id="python-os-command",
title="Python OS command execution needs review",
severity="high",
tags=("deserialization",),
pattern=re.compile(r"\b(pickle\.loads|yaml\.load)\s*\("),
tags=("syscall_entry",),
pattern=re.compile(r"\b(os\.system|os\.popen|os\.execl|os\.execlp|os\.execv|os\.execvp)\s*\("),
cwe="CWE-78",
languages=PYTHON_LANGUAGES,
),
PatternRule(
rule_id="python-shell-true",
@@ -64,6 +127,229 @@ PATTERN_RULES = (
severity="medium",
tags=("syscall_entry",),
pattern=re.compile(r"subprocess\.[a-zA-Z_]+\s*\([^\n]*shell\s*=\s*True"),
cwe="CWE-78",
languages=PYTHON_LANGUAGES,
),
PatternRule(
rule_id="python-unsafe-deserialization",
title="Unsafe Python deserialization needs review",
severity="high",
tags=("deserialization",),
pattern=re.compile(r"\b(pickle\.loads|pickle\.load|cPickle\.loads|yaml\.load|marshal\.loads)\s*\("),
cwe="CWE-502",
languages=PYTHON_LANGUAGES,
),
PatternRule(
rule_id="python-sql-injection",
title="Python SQL string composition needs review",
severity="high",
tags=("injection",),
# .execute(...) where the argument is built with %-formatting, .format(),
# string concatenation, or an f-string. Parameterised %s placeholders pass.
pattern=re.compile(r"(\.execute)\s*\(\s*[^\n]*(?:[\"']\s*%\s*[(A-Za-z_]|\.format\s*\(|[\"']\s*\+|\bf[\"'])"),
cwe="CWE-89",
languages=PYTHON_LANGUAGES,
),
PatternRule(
rule_id="python-path-traversal",
title="Python file open with external input needs review",
severity="medium",
tags=("filesystem_entry",),
pattern=re.compile(r"\b(open)\s*\(\s*[^\n]*(?:request\.|req\.|argv|input\s*\(|params\[|GET\[|POST\[)"),
cwe="CWE-22",
languages=PYTHON_LANGUAGES,
),
PatternRule(
rule_id="python-ssrf",
title="Python outbound request with external input needs review",
severity="medium",
tags=("network_entry",),
pattern=re.compile(
r"\b(requests\.(?:get|post|put|delete|head|patch)|urllib\.request\.urlopen|urlopen)\s*\(\s*[^\n]*(?:request\.|req\.|argv|input\s*\()"
),
cwe="CWE-918",
languages=PYTHON_LANGUAGES,
),
PatternRule(
rule_id="python-weak-hash",
title="Python weak hash needs review",
severity="low",
tags=("crypto",),
pattern=re.compile(r"\bhashlib\.(md5|sha1)\s*\("),
cwe="CWE-327",
languages=PYTHON_LANGUAGES,
),
PatternRule(
rule_id="python-ssti",
title="Python server-side template injection needs review",
severity="high",
tags=("injection",),
pattern=re.compile(r"\b(render_template_string)\s*\("),
cwe="CWE-94",
languages=PYTHON_LANGUAGES,
),
PatternRule(
rule_id="python-flask-debug",
title="Flask debug mode enabled needs review",
severity="low",
tags=("insecure_default",),
pattern=re.compile(r"\.run\s*\([^\n]*(debug)\s*=\s*True"),
cwe="CWE-489",
languages=PYTHON_LANGUAGES,
),
# --- JavaScript / TypeScript -------------------------------------------------
PatternRule(
rule_id="js-command-exec",
title="Node child_process command execution needs review",
severity="high",
tags=("syscall_entry",),
# exec/execSync invoke a shell; execFile/spawn with an argv array do not and
# are the safe alternatives, so they are intentionally excluded here.
pattern=re.compile(r"\b(exec|execSync)\s*\("),
cwe="CWE-78",
languages=JS_LANGUAGES,
),
PatternRule(
rule_id="js-spawn-shell",
title="Node spawn with shell needs review",
severity="high",
tags=("syscall_entry",),
pattern=re.compile(r"\b(spawn|spawnSync)\s*\([^\n]*shell\s*:\s*true"),
cwe="CWE-78",
languages=JS_LANGUAGES,
),
PatternRule(
rule_id="js-eval",
title="Dynamic JavaScript evaluation needs review",
severity="high",
tags=("syscall_entry",),
pattern=re.compile(r"\b(eval)\s*\("),
cwe="CWE-95",
languages=JS_LANGUAGES,
),
PatternRule(
rule_id="js-function-constructor",
title="JavaScript Function constructor needs review",
severity="high",
tags=("syscall_entry",),
pattern=re.compile(r"new\s+(Function)\s*\("),
cwe="CWE-95",
languages=JS_LANGUAGES,
),
PatternRule(
rule_id="js-xss-response",
title="Unsanitized HTTP response reflection needs review",
severity="high",
tags=("injection",),
pattern=re.compile(r"\.(send|write|end)\s*\(\s*[^\n]*\b(?:req|request)\.(?:query|params|body)"),
cwe="CWE-79",
languages=JS_LANGUAGES,
),
PatternRule(
rule_id="js-dom-xss",
title="DOM sink with untrusted input needs review",
severity="high",
tags=("injection",),
pattern=re.compile(r"\.(innerHTML|outerHTML)\s*=\s*[^\n]*\b(?:req|request|location|document\.location)"),
cwe="CWE-79",
languages=JS_LANGUAGES,
),
PatternRule(
rule_id="js-sql-injection",
title="JavaScript SQL string composition needs review",
severity="high",
tags=("injection",),
pattern=re.compile(r"\.(query|execute)\s*\(\s*[^\n]*(?:\$\{|\+\s*[A-Za-z_$])"),
cwe="CWE-89",
languages=JS_LANGUAGES,
),
PatternRule(
rule_id="js-path-traversal",
title="Node filesystem access with external input needs review",
severity="medium",
tags=("filesystem_entry",),
pattern=re.compile(
r"\bfs\.(readFile|readFileSync|writeFile|writeFileSync|createReadStream|unlink|unlinkSync)\s*\(\s*[^\n]*\b(?:req|request)\."
),
cwe="CWE-22",
languages=JS_LANGUAGES,
),
PatternRule(
rule_id="js-ssrf",
title="JavaScript outbound request with external input needs review",
severity="medium",
tags=("network_entry",),
pattern=re.compile(r"\b(axios|fetch|got)\s*[.(][^\n]*\b(?:req|request)\."),
cwe="CWE-918",
languages=JS_LANGUAGES,
),
PatternRule(
rule_id="js-weak-hash",
title="JavaScript weak hash needs review",
severity="low",
tags=("crypto",),
pattern=re.compile(r"createHash\s*\(\s*[\"'](md5|sha1)[\"']"),
cwe="CWE-327",
languages=JS_LANGUAGES,
),
PatternRule(
rule_id="js-deserialize",
title="JavaScript unsafe deserialization needs review",
severity="high",
tags=("deserialization",),
pattern=re.compile(r"\b(unserialize)\s*\("),
cwe="CWE-502",
languages=JS_LANGUAGES,
),
# --- Java --------------------------------------------------------------------
PatternRule(
rule_id="java-command-exec",
title="Java runtime command execution needs review",
severity="high",
tags=("syscall_entry",),
pattern=re.compile(r"\b(exec)\s*\("),
cwe="CWE-78",
languages=JAVA_LANGUAGES,
),
PatternRule(
rule_id="java-sql-injection",
title="Java SQL string composition needs review",
severity="high",
tags=("injection",),
# A string literal that begins with a SQL keyword and is concatenated with
# untrusted data. Parameterised queries (placeholders, no '+') are ignored.
pattern=re.compile(r"(?i)\"\s*(select|insert|update|delete)\b[^\"]*\"\s*\+"),
cwe="CWE-89",
languages=JAVA_LANGUAGES,
),
PatternRule(
rule_id="java-weak-hash",
title="Java weak hash needs review",
severity="low",
tags=("crypto",),
pattern=re.compile(r"MessageDigest\.getInstance\s*\(\s*\"(MD5|SHA-1)\""),
cwe="CWE-327",
languages=JAVA_LANGUAGES,
),
PatternRule(
rule_id="java-deserialization",
title="Java unsafe deserialization needs review",
severity="high",
tags=("deserialization",),
pattern=re.compile(r"\.(readObject)\s*\("),
cwe="CWE-502",
languages=JAVA_LANGUAGES,
),
# --- Groovy server-side templates -------------------------------------------
PatternRule(
rule_id="groovy-xss-unescaped",
title="Groovy template unescaped output needs review",
severity="high",
tags=("injection",),
# Unescaped output of a bound variable; string literals are not flagged.
pattern=re.compile(r"(yieldUnescaped)\s+[A-Za-z_]"),
cwe="CWE-79",
languages=GROOVY_LANGUAGES,
),
)
@@ -74,9 +360,13 @@ def quick_scan_findings(root: Path, targets: list[FileTarget], rankings: list[Ra
for target in targets:
text = _read_text(root / target.path)
for rule in PATTERN_RULES:
if rule.languages and target.language not in rule.languages:
continue
ranking = ranking_by_path.get(target.path)
for match in rule.pattern.finditer(text):
findings.append(_finding_from_match(target, rule, text, match.start(), ranking))
if _match_is_in_comment(text, match.start(), target.language):
continue
findings.append(_finding_from_match(target, rule, text, match, ranking))
return sorted(findings, key=lambda item: (_severity_sort(item.severity), -item.ranking_priority, item.path))
@@ -101,12 +391,14 @@ def _finding_from_match(
target: FileTarget,
rule: PatternRule,
text: str,
offset: int,
match: re.Match[str],
ranking: RankingScore | None,
) -> StaticFinding:
offset = match.start()
line = text.count("\n", 0, offset) + 1
reachability_status, function_name, reachability_evidence = _reachability_for_line(target, line)
priority = ranking.priority if ranking is not None else 1.0
matched_token = match.group(1) if match.groups() else match.group(0).split("(", 1)[0].strip()
return StaticFinding(
finding_id=f"{rule.rule_id}:{target.path}:{line}",
path=target.path,
@@ -114,7 +406,9 @@ def _finding_from_match(
severity=rule.severity,
confidence="medium",
evidence_level="static_corroboration",
rationale=f"Static pattern {rule.rule_id} matched line {line}; manual verification still required.",
rationale=(
f"Static pattern {rule.rule_id} ({rule.cwe}) matched {matched_token} on line {line}; manual verification still required."
),
line=line,
function_name=function_name,
reachability_status=reachability_status,
@@ -179,6 +473,28 @@ def _read_text(path: Path) -> str:
return ""
HASH_COMMENT_LANGUAGES = {"python", "ruby", "shell", "yaml"}
def _match_is_in_comment(text: str, offset: int, language: str = "") -> bool:
line_start = text.rfind("\n", 0, offset) + 1
line_end = text.find("\n", offset)
if line_end == -1:
line_end = len(text)
line = text[line_start:line_end]
column = offset - line_start
stripped = line.lstrip()
if language in HASH_COMMENT_LANGUAGES:
if stripped.startswith("#"):
return True
comment_column = line.find("#")
return comment_column != -1 and comment_column < column
if stripped.startswith(("//", "/*", "*")):
return True
comment_column = line.find("//")
return comment_column != -1 and comment_column < column
def _severity_sort(severity: str) -> int:
order = {"critical": 0, "high": 1, "medium": 2, "low": 3, "info": 4}
return order.get(severity, 5)
src/openultrasast/preprocess.py
+2
View File
@@ -23,6 +23,8 @@ LANGUAGE_BY_EXTENSION = {
".go": "go",
".rs": "rust",
".java": "java",
".groovy": "groovy",
".tpl": "groovy",
".kt": "kotlin",
".rb": "ruby",
".php": "php",