mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-16 03:49:26 +00:00
Marius CiepluchandClaude Opus 5.5
c5bf52f47d
0 of 12 positives; 12.6 false alarms per 100 ordinary pushes (2.1 on changed lines only). The misses come from what the rules and facts express (access control lives in obligations, which do not convert; sources cover request objects only), not from the engine. Comparison arm only: a Semgrep miss is never evidence against a hunter finding. Co-Authored-By: Claude Opus 5.5 <[email protected]>
635 lines
29 KiB
Python
635 lines
29 KiB
Python
"""Semgrep CE as a benchmark arm: our regex rules and taint facts converted, run per frozen change.
|
|
|
|
Two subcommands. ``rules`` converts every enabled quick rule to a ``pattern-regex`` rule and every
|
|
source/sink/sanitizer fact triple to a ``mode: taint`` rule, one YAML file per language, and writes a
|
|
conversion report naming what was skipped and why. ``run`` checks out each frozen change's head into a
|
|
temporary worktree, runs Semgrep over the changed files only, proves the targets were read (bytes per
|
|
file), and writes one scorer-shaped record per change, resumable exactly like ``hunter_arm``.
|
|
|
|
Nothing here ships: the rules are ours, the engine is a measurement instrument.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import tomllib
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
from openultrasast.hunter_tools import clamp_repo_path
|
|
from openultrasast.model.taxonomy import load_families
|
|
from openultrasast.ruleset import DEFAULT_RULESET_DIR, load_ruleset
|
|
from openultrasast.semantic.facts import DEFAULT_FACTS_DIR, load_facts
|
|
|
|
if __package__ in {None, ""}:
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
|
|
|
from benchmarks.unseen.extract import SHA, InstrumentFailure
|
|
from benchmarks.unseen.replay import canonical, load_slice
|
|
from benchmarks.valid.catalog import local_git
|
|
|
|
# Our language ids to Semgrep's. Groovy has no Semgrep parser: `generic` plus a path filter.
|
|
SEMGREP_LANGUAGES = {
|
|
"c": "c",
|
|
"cpp": "cpp",
|
|
"python": "python",
|
|
"javascript": "javascript",
|
|
"typescript": "typescript",
|
|
"java": "java",
|
|
"php": "php",
|
|
}
|
|
GENERIC_PATHS = {"groovy": ["*.groovy", "*.tpl"]}
|
|
# The rule file a rule lands in: one per language family.
|
|
FILE_OF_LANGUAGE = {"cpp": "c", "typescript": "javascript"}
|
|
# Fact languages whose Semgrep rules need a second dialect.
|
|
FACT_DIALECTS = {"c": ["c", "cpp"], "javascript": ["javascript", "typescript"]}
|
|
# PHP language constructs are not calls: `echo(...)` would parse as a call expression, and Semgrep's PHP
|
|
# pattern parser takes `echo` only as a whole statement.
|
|
PHP_CONSTRUCTS = {
|
|
"echo": "echo $VALUE;",
|
|
"print": "print $VALUE;",
|
|
"include": "include ...",
|
|
"include_once": "include_once ...",
|
|
"require": "require ...",
|
|
"require_once": "require_once ...",
|
|
}
|
|
# DOM properties assigned, not called.
|
|
ASSIGNED_SINKS = {"innerHTML", "outerHTML"}
|
|
SEMGREP_FLAGS = (
|
|
"--json",
|
|
"--metrics=off",
|
|
"--disable-version-check",
|
|
"--no-git-ignore",
|
|
"--timeout",
|
|
"60",
|
|
"--max-memory",
|
|
"1500",
|
|
"-j",
|
|
"1",
|
|
)
|
|
# A Semgrep launch that returns faster than its own startup never ran the engine.
|
|
MIN_PLAUSIBLE_SECONDS = 1.0
|
|
# PCRE2 compiles a bounded repeat of a group by copying it: `(?:...){0,2000}?` is "too large". The lazy
|
|
# unbounded form keeps the match and widens the window from 2000 characters to the statement.
|
|
BOUNDED_GROUP_REPEAT = re.compile(r"\{0,\d+\}\?")
|
|
|
|
|
|
def _family_of(cwe):
|
|
family = load_families().family_of_cwe(cwe or "")
|
|
return family.id if family else "unknown"
|
|
|
|
|
|
def _ident(value):
|
|
return re.sub(r"[^A-Za-z0-9_-]+", "-", value).strip("-")
|
|
|
|
|
|
# --- regex rules -------------------------------------------------------------------------------------------
|
|
|
|
|
|
def _raw_rules(ruleset_dir):
|
|
rows = []
|
|
for path in sorted(Path(ruleset_dir).rglob("*.toml")):
|
|
if path.parent.name in {"semantic", "obligations"} or path.parent == Path(ruleset_dir):
|
|
continue
|
|
for item in tomllib.loads(path.read_text()).get("rule", []):
|
|
rows.append(item)
|
|
return rows
|
|
|
|
|
|
def convert_regex_rules(ruleset_dir=DEFAULT_RULESET_DIR, *, min_precision=0.0):
|
|
"""Every enabled quick rule as a Semgrep ``pattern-regex`` rule; the rest named with a reason."""
|
|
active = {rule.rule_id: rule for rule in load_ruleset(Path(ruleset_dir))}
|
|
converted, skipped, rewrites = {}, [], []
|
|
for item in _raw_rules(ruleset_dir):
|
|
rule_id = str(item["rule_id"])
|
|
rule = active.get(rule_id)
|
|
if rule is None:
|
|
skipped.append({"origin": rule_id, "reason": f"fallback for {item.get('fallback_for')}: the tagged rule is active"})
|
|
continue
|
|
if rule.status == "disabled":
|
|
skipped.append({"origin": rule_id, "reason": "disabled"})
|
|
continue
|
|
if rule.status == "shadow":
|
|
skipped.append(
|
|
{"origin": rule_id, "reason": f"shadow status (precision_estimate {rule.precision_estimate}); not admitted by the hook"}
|
|
)
|
|
continue
|
|
if rule.precision_estimate < min_precision:
|
|
skipped.append({"origin": rule_id, "reason": f"emit floor: precision_estimate {rule.precision_estimate} < {min_precision}"})
|
|
continue
|
|
if not rule.languages:
|
|
skipped.append({"origin": rule_id, "reason": "non-code file targets: the rule applies to every file"})
|
|
continue
|
|
languages = [SEMGREP_LANGUAGES[language] for language in rule.languages if language in SEMGREP_LANGUAGES]
|
|
generic = [language for language in rule.languages if language in GENERIC_PATHS]
|
|
unsupported = [language for language in rule.languages if language not in SEMGREP_LANGUAGES and language not in GENERIC_PATHS]
|
|
if unsupported:
|
|
skipped.append({"origin": rule_id, "reason": f"no Semgrep language for {unsupported}"})
|
|
continue
|
|
family = _family_of(rule.cwe)
|
|
pattern = BOUNDED_GROUP_REPEAT.sub("*?", rule.pattern)
|
|
if pattern != rule.pattern:
|
|
rewrites.append({"origin": rule_id, "rewrite": "bounded lazy group repeat {0,N}? widened to *? (PCRE2 size limit)"})
|
|
entry = {
|
|
"id": f"ousast.{FILE_OF_LANGUAGE.get(rule.languages[0], rule.languages[0])}.regex.{_ident(rule_id)}",
|
|
"languages": languages or ["generic"],
|
|
"severity": "WARNING",
|
|
"message": rule.title,
|
|
"metadata": {
|
|
"family": family,
|
|
"cwe": rule.cwe,
|
|
"origin": rule_id,
|
|
"precision_estimate": str(rule.precision_estimate),
|
|
"kind": "regex",
|
|
"tags": list(rule.tags),
|
|
"status": rule.status,
|
|
"version": rule.version,
|
|
},
|
|
"pattern-regex": pattern,
|
|
}
|
|
if not languages:
|
|
entry["paths"] = {"include": [glob for language in generic for glob in GENERIC_PATHS[language]]}
|
|
key = FILE_OF_LANGUAGE.get(rule.languages[0], rule.languages[0])
|
|
converted.setdefault(key, []).append(entry)
|
|
return converted, skipped, rewrites
|
|
|
|
|
|
# --- taint facts -------------------------------------------------------------------------------------------
|
|
|
|
|
|
def _call_patterns(call, language):
|
|
"""The Semgrep call shapes of one fact token (a function, a method, a construct)."""
|
|
if language == "php":
|
|
if call in PHP_CONSTRUCTS:
|
|
return [PHP_CONSTRUCTS[call]]
|
|
return [f"{call}(...)"]
|
|
if "." in call or "->" in call:
|
|
return [f"{call}(...)"]
|
|
if language == "java" and call[:1].isupper():
|
|
return [f"new {call}(...)"]
|
|
return [f"{call}(...)", f"$OBJ.{call}(...)"]
|
|
|
|
|
|
def _source_patterns(token, language):
|
|
if language == "php":
|
|
if token == "php://input":
|
|
return ['file_get_contents("php://input")']
|
|
return [token, f"{token}[...]"]
|
|
if token.endswith("("):
|
|
return [f"{token}...)"]
|
|
if "." in token:
|
|
return [token, f"{token}(...)"]
|
|
if language == "c":
|
|
return [token, f"{token}(...)"]
|
|
return [token, f"$OBJ.{token}", f"{token}(...)", f"$OBJ.{token}(...)"]
|
|
|
|
|
|
def _sink_entries(call, language, *, format_arg=None, single_argument=False):
|
|
"""Sink patterns for one call; a focused argument where the fact names one."""
|
|
if call in ASSIGNED_SINKS:
|
|
return [{"patterns": [{"pattern": f"$OBJ.{call} = $VALUE"}, {"focus-metavariable": "$VALUE"}]}]
|
|
if call == "dangerouslySetInnerHTML":
|
|
return [{"patterns": [{"pattern": "{__html: $VALUE}"}, {"focus-metavariable": "$VALUE"}]}]
|
|
if format_arg is not None:
|
|
arguments = ", ".join([*(f"$ARG{index}" for index in range(format_arg)), "$FORMAT", "..."])
|
|
heads = [call] if "." in call or language == "c" else [call, f"$OBJ.{call}"]
|
|
return [{"patterns": [{"pattern": f"{head}({arguments})"}, {"focus-metavariable": "$FORMAT"}]} for head in heads]
|
|
if single_argument:
|
|
heads = [call] if "." in call or "->" in call else [call, f"$OBJ.{call}"]
|
|
return [{"patterns": [{"pattern": f"{head}($QUERY)"}, {"focus-metavariable": "$QUERY"}]} for head in heads]
|
|
return [{"pattern": pattern} for pattern in _call_patterns(call, language)]
|
|
|
|
|
|
def _weak_literal_rule(language, sink_id, cwe, calls, literals, dialects):
|
|
"""A `weak_literals` sink is a search, not a flow: the call with a named weak literal, or a call named after one."""
|
|
alternatives = []
|
|
lowered = {literal.lower() for literal in literals}
|
|
for call in sorted(calls):
|
|
for head in [call] if "." in call else [call, f"$OBJ.{call}"]:
|
|
if call.rsplit(".", 1)[-1].lower() in lowered:
|
|
alternatives.append({"pattern": f"{head}(...)"})
|
|
else:
|
|
alternatives.append(
|
|
{
|
|
"patterns": [
|
|
{"pattern": f"{head}($ALGORITHM, ...)"},
|
|
{
|
|
"metavariable-regex": {
|
|
"metavariable": "$ALGORITHM",
|
|
"regex": "^[\"']?(" + "|".join(re.escape(literal) for literal in sorted(literals)) + ")[\"']?$",
|
|
}
|
|
},
|
|
]
|
|
}
|
|
)
|
|
return {
|
|
"id": f"ousast.{language}.weak.{_ident(sink_id)}",
|
|
"languages": dialects,
|
|
"severity": "WARNING",
|
|
"message": f"{sink_id}: a weak literal reaches {', '.join(sorted(calls))}",
|
|
"metadata": {
|
|
"family": _family_of(cwe),
|
|
"cwe": cwe,
|
|
"origin": f"sink:{language}:{sink_id}",
|
|
"precision_estimate": "unmeasured",
|
|
"kind": "weak_literal",
|
|
"weak_literals": sorted(literals),
|
|
},
|
|
"pattern-either": alternatives,
|
|
}
|
|
|
|
|
|
def convert_facts(facts_dir=DEFAULT_FACTS_DIR):
|
|
"""One taint rule per (language, sink id, CWE); the facts Semgrep has no word for are named, not dropped silently."""
|
|
facts = load_facts(Path(facts_dir))
|
|
converted, skipped, lossy = {}, [], []
|
|
for language in sorted({fact.language for fact in facts.sinks}):
|
|
scoped = facts.for_language(language)
|
|
dialects = FACT_DIALECTS.get(language, [SEMGREP_LANGUAGES[language]])
|
|
sources = [
|
|
{"pattern": pattern} for fact in scoped.sources for token in fact.patterns for pattern in _source_patterns(token, language)
|
|
]
|
|
parameterized = set()
|
|
sanitizers = []
|
|
for fact in scoped.sanitizers:
|
|
origin = f"sanitizer:{language}:{fact.id}"
|
|
if fact.guard:
|
|
skipped.append({"origin": origin, "reason": "guard: a check whose safety depends on dominance; Semgrep has no equivalent"})
|
|
continue
|
|
if fact.parameterized:
|
|
parameterized.update(fact.calls)
|
|
skipped.append(
|
|
{"origin": origin, "reason": "parameterized: a call shape, not a cleansing; expressed as the single-argument sink form"}
|
|
)
|
|
continue
|
|
if fact.literal_format_arg is not None:
|
|
skipped.append(
|
|
{
|
|
"origin": origin,
|
|
"reason": "literal_format_arg: a shape; a literal format is never tainted at the focused sink argument",
|
|
}
|
|
)
|
|
continue
|
|
if fact.quoted_only:
|
|
loss = {"origin": origin, "loss": "quoted_only escape treated as a full sanitizer (silences the CVE-2024-1071 shape)"}
|
|
if loss not in lossy:
|
|
lossy.append(loss)
|
|
sanitizers.extend({"pattern": pattern} for call in fact.calls for pattern in _call_patterns(call, language))
|
|
for fact in scoped.dispatches:
|
|
skipped.append(
|
|
{
|
|
"origin": f"dispatch:{language}:{fact.id}",
|
|
"reason": "dispatch: string-keyed hook registry routing; no Semgrep equivalent",
|
|
}
|
|
)
|
|
for fact in scoped.layouts:
|
|
skipped.append({"origin": f"layout:{language}:{fact.id}", "reason": "layout: scope policy (vendored, tests), not a rule"})
|
|
grouped = {}
|
|
for fact in scoped.sinks:
|
|
grouped.setdefault((fact.id, fact.cwe), []).append(fact)
|
|
for (sink_id, cwe), rows in sorted(grouped.items()):
|
|
origin = f"sink:{language}:{sink_id}"
|
|
calls = sorted({call for row in rows for call in row.calls})
|
|
literals = {literal for row in rows for literal in row.weak_literals}
|
|
if literals:
|
|
converted.setdefault(language, []).append(_weak_literal_rule(language, sink_id, cwe, calls, literals, dialects))
|
|
continue
|
|
entries, extra_sanitizers = [], []
|
|
format_arg = next((row.format_arg for row in rows if row.format_arg is not None), None)
|
|
for call in calls:
|
|
if call == "v-html":
|
|
skipped.append({"origin": f"{origin}:{call}", "reason": "template directive, not a call"})
|
|
continue
|
|
entries.extend(_sink_entries(call, language, format_arg=format_arg, single_argument=call in parameterized))
|
|
for row in rows:
|
|
if row.prefix_fixes_origin:
|
|
lossy.append(
|
|
{"origin": origin, "loss": "prefix_fixes_origin: a value appended after a fixed origin still counts as tainted"}
|
|
)
|
|
break
|
|
for anchor in sorted({anchor for row in rows for anchor in row.origin_anchors}):
|
|
extra_sanitizers.extend({"pattern": pattern} for pattern in _call_patterns(anchor, language))
|
|
if not entries:
|
|
skipped.append({"origin": origin, "reason": "no call form survives conversion"})
|
|
continue
|
|
rule = {
|
|
"id": f"ousast.{language}.taint.{_ident(sink_id)}",
|
|
"mode": "taint",
|
|
"languages": dialects,
|
|
"severity": "ERROR",
|
|
"message": f"{sink_id}: untrusted input reaches {', '.join(calls)}",
|
|
"metadata": {
|
|
"family": _family_of(cwe),
|
|
"cwe": cwe,
|
|
"origin": origin,
|
|
"precision_estimate": "unmeasured",
|
|
"kind": "taint",
|
|
"calls": calls,
|
|
},
|
|
"pattern-sources": [{"pattern-either": sources}],
|
|
"pattern-sinks": [{"pattern-either": entries}],
|
|
}
|
|
if sanitizers or extra_sanitizers:
|
|
rule["pattern-sanitizers"] = [{"pattern-either": [*sanitizers, *extra_sanitizers]}]
|
|
converted.setdefault(language, []).append(rule)
|
|
return converted, skipped, lossy
|
|
|
|
|
|
def _obligation_tables(ruleset_dir):
|
|
count = 0
|
|
for path in sorted((Path(ruleset_dir) / "obligations").glob("*.toml")):
|
|
payload = tomllib.loads(path.read_text())
|
|
count += sum(len(value) for value in payload.values() if isinstance(value, list))
|
|
return count
|
|
|
|
|
|
def write_rules(out, *, ruleset_dir=DEFAULT_RULESET_DIR, facts_dir=DEFAULT_FACTS_DIR, min_precision=0.0, semgrep=None):
|
|
"""Write one YAML per language under ``out`` plus ``conversion_report.json``; validate with Semgrep when present."""
|
|
out = Path(out)
|
|
out.mkdir(parents=True, exist_ok=True)
|
|
regex_rules, regex_skipped, rewrites = convert_regex_rules(ruleset_dir, min_precision=min_precision)
|
|
taint_rules, fact_skipped, lossy = convert_facts(facts_dir)
|
|
files = {}
|
|
for language in sorted(set(regex_rules) | set(taint_rules)):
|
|
rules = [*regex_rules.get(language, []), *taint_rules.get(language, [])]
|
|
path = out / f"{language}.yaml"
|
|
path.write_text(yaml.safe_dump({"rules": rules}, sort_keys=False, width=200, allow_unicode=True))
|
|
files[path.name] = len(rules)
|
|
obligations = _obligation_tables(ruleset_dir)
|
|
report = {
|
|
"regex": {
|
|
"converted": sorted(rule["metadata"]["origin"] for rules in regex_rules.values() for rule in rules),
|
|
"skipped": regex_skipped,
|
|
"rewritten": rewrites,
|
|
},
|
|
"facts": {
|
|
"converted": sorted(rule["metadata"]["origin"] for rules in taint_rules.values() for rule in rules),
|
|
"skipped": fact_skipped,
|
|
"lossy": lossy,
|
|
},
|
|
"obligations": {
|
|
"skipped_tables": obligations,
|
|
"reason": "an obligation finds a missing guard (an absence), which no pattern expresses",
|
|
},
|
|
"not_carried": [
|
|
"comment skipping: the quick engine drops a regex match inside a comment; pattern-regex does not",
|
|
"framework priors and fallback rules: the tagged rule is emitted, its language-level fallback is not",
|
|
"min_evidence_level, ranking priority, the rung: post-match policy on our side",
|
|
],
|
|
"files": files,
|
|
"validation": None,
|
|
}
|
|
binary = semgrep or shutil.which("semgrep")
|
|
if binary:
|
|
try:
|
|
completed = subprocess.run(
|
|
[binary, "--validate", "--config", str(out), "--metrics=off", "--disable-version-check"], capture_output=True, text=True
|
|
)
|
|
except OSError as error:
|
|
report["validation"] = {"exit": None, "error": f"semgrep not found or not runnable: {type(error).__name__}"}
|
|
else:
|
|
report["validation"] = {"exit": completed.returncode, "stderr": completed.stderr[-4000:], "stdout": completed.stdout[-4000:]}
|
|
(out / "conversion_report.json").write_text(json.dumps(report, indent=2, sort_keys=True) + "\n")
|
|
return report
|
|
|
|
|
|
# --- run ---------------------------------------------------------------------------------------------------
|
|
|
|
|
|
def result_defect(result):
|
|
"""One Semgrep result as a scorer defect; the family and CWE come from the rule's own metadata."""
|
|
extra = result.get("extra", {})
|
|
metadata = extra.get("metadata", {})
|
|
path = result["path"]
|
|
line = result.get("start", {}).get("line", 0)
|
|
origin = metadata.get("origin", result["check_id"])
|
|
return {
|
|
"defect_id": f"{origin}:{path}:{line}",
|
|
"family": metadata.get("family", "unknown"),
|
|
"cwe": metadata.get("cwe"),
|
|
"locations": [[path, line]],
|
|
"source": "semgrep",
|
|
"rule_id": result["check_id"],
|
|
"origin": origin,
|
|
"title": extra.get("message", ""),
|
|
"rationale": extra.get("message", ""),
|
|
"confidence": extra.get("severity"),
|
|
"dataflow": bool(extra.get("dataflow_trace")),
|
|
}
|
|
|
|
|
|
def _write_json(path, value):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = path.with_suffix(".json.tmp")
|
|
temporary.write_text(json.dumps(value, sort_keys=True, indent=2) + "\n")
|
|
temporary.replace(path)
|
|
|
|
|
|
def _record_path(out, identity):
|
|
return out / (hashlib.sha256(identity.encode()).hexdigest() + ".json")
|
|
|
|
|
|
def _inputs(out):
|
|
path = out / "inputs.json"
|
|
entries = json.loads(path.read_text()) if path.exists() else []
|
|
for key in (lambda entry: entry["item_id"], lambda entry: entry["change"]["id"]):
|
|
if len({key(entry) for entry in entries}) != len(entries):
|
|
raise ValueError(f"duplicate ids in {path}")
|
|
return entries
|
|
|
|
|
|
def run_semgrep(binary, rules, root, targets, *, timeout=900):
|
|
"""Semgrep over explicit targets inside ``root``; paths in the output stay repository-relative."""
|
|
argv = [binary, "scan", "--config", str(rules), *SEMGREP_FLAGS, *targets]
|
|
started = time.monotonic()
|
|
completed = subprocess.run(argv, cwd=root, capture_output=True, text=True, timeout=timeout)
|
|
return completed, time.monotonic() - started
|
|
|
|
|
|
def _scan_change(change, repos, rules, *, binary):
|
|
instrument = {
|
|
"hook_exit": 0,
|
|
"hook_bytes_read": 0,
|
|
"changed_bytes": {},
|
|
"head_verified": False,
|
|
"base_verified": False,
|
|
"failure": None,
|
|
"suspect_fast": False,
|
|
"semgrep": {"exit": None, "wall_seconds": None, "scanned": [], "errors": [], "results": 0},
|
|
}
|
|
record = {
|
|
"id": change["id"],
|
|
"done": True,
|
|
"status": "semgrep_covered",
|
|
"instrument": instrument,
|
|
"push": {"admission": {"defects": [], "dispositions": []}, "quick_tier": []},
|
|
"skipped": [],
|
|
}
|
|
try:
|
|
if not all(SHA.fullmatch(change[side]) for side in ("base", "head")):
|
|
raise InstrumentFailure("invalid_revision")
|
|
git = local_git(Path(repos[change["repo"]]).expanduser().resolve())
|
|
git.run("cat-file", "-e", change["base"] + "^{commit}")
|
|
instrument["base_verified"] = True
|
|
with tempfile.TemporaryDirectory(prefix="ousast-semgrep-") as temporary:
|
|
root = Path(temporary) / "head"
|
|
added = False
|
|
try:
|
|
git.run("-c", "core.hooksPath=/dev/null", "worktree", "add", "--detach", str(root), change["head"])
|
|
added = True
|
|
if local_git(root).run("rev-parse", "HEAD").strip() != change["head"]:
|
|
raise InstrumentFailure("head_mismatch")
|
|
instrument["head_verified"] = True
|
|
targets = []
|
|
for relative in dict.fromkeys(file["path"] for file in change.get("files", [])):
|
|
path = clamp_repo_path(root, relative)
|
|
try:
|
|
path.stat()
|
|
except FileNotFoundError:
|
|
record["skipped"].append({"path": relative, "reason": "missing_at_head"})
|
|
print(f"{change['id']} {relative}: skipped (missing_at_head)", flush=True)
|
|
continue
|
|
try:
|
|
data = path.read_bytes()
|
|
except OSError:
|
|
print(f"{change['id']} {relative}: 0 bytes read", flush=True)
|
|
raise InstrumentFailure(f"unreadable_target:{relative}") from None
|
|
size = len(data)
|
|
print(f"{change['id']} {relative}: {size} bytes read", flush=True)
|
|
if not size:
|
|
record["skipped"].append({"path": relative, "reason": "empty_at_head"})
|
|
continue
|
|
instrument["changed_bytes"][relative] = size
|
|
instrument["hook_bytes_read"] += size
|
|
targets.append(relative)
|
|
if not targets:
|
|
raise InstrumentFailure("no_readable_targets")
|
|
try:
|
|
completed, wall = run_semgrep(binary, rules, root, targets)
|
|
except subprocess.TimeoutExpired:
|
|
raise InstrumentFailure("semgrep_timeout") from None
|
|
semgrep = instrument["semgrep"]
|
|
semgrep.update(exit=completed.returncode, wall_seconds=round(wall, 3))
|
|
instrument["hook_exit"] = completed.returncode
|
|
if wall < MIN_PLAUSIBLE_SECONDS:
|
|
instrument["suspect_fast"] = True
|
|
if completed.returncode not in (0, 1):
|
|
semgrep["stderr"] = completed.stderr[-2000:]
|
|
raise InstrumentFailure(f"semgrep_exit_{completed.returncode}")
|
|
try:
|
|
payload = json.loads(completed.stdout)
|
|
except json.JSONDecodeError:
|
|
semgrep["stderr"] = completed.stderr[-2000:]
|
|
raise InstrumentFailure("semgrep_output_unreadable") from None
|
|
semgrep["scanned"] = sorted(payload.get("paths", {}).get("scanned", []))
|
|
semgrep["errors"] = [
|
|
{"type": str(error.get("type")), "path": error.get("path"), "message": str(error.get("message", ""))[:300]}
|
|
for error in payload.get("errors", [])
|
|
]
|
|
results = payload.get("results", [])
|
|
semgrep["results"] = len(results)
|
|
record["push"]["admission"]["defects"] = [result_defect(result) for result in results]
|
|
print(f"{change['id']}: semgrep exit {completed.returncode}, {len(results)} results, {wall:.1f}s", flush=True)
|
|
finally:
|
|
if added:
|
|
git.run("worktree", "remove", "--force", str(root))
|
|
except Exception as error: # noqa: BLE001
|
|
reason = str(error) if isinstance(error, InstrumentFailure) else type(error).__name__
|
|
instrument.update(hook_exit=2, failure=reason)
|
|
record["status"] = "instrument_failure"
|
|
record["push"]["admission"]["defects"] = []
|
|
print(f"{change['id']}: instrument_failure ({reason})", file=sys.stderr, flush=True)
|
|
return record
|
|
|
|
|
|
def run(manifest, repos_path, rules, out, *, slice_id="1", only="all", semgrep=None):
|
|
binary = semgrep or shutil.which("semgrep")
|
|
if binary is None:
|
|
print("No semgrep binary found on PATH.", file=sys.stderr)
|
|
return 2
|
|
rules = Path(rules).expanduser().resolve()
|
|
if not rules.is_dir():
|
|
raise ValueError(f"rules directory missing: {rules}")
|
|
changes = load_slice(manifest, slice_id)
|
|
if len({change["id"] for change in changes}) != len(changes):
|
|
raise ValueError("duplicate change ids")
|
|
changes = [change for change in changes if only == "all" or change["kind"] == only]
|
|
entries = _inputs(out)
|
|
indexed = {entry["change"]["id"]: entry for entry in entries}
|
|
pending = []
|
|
for change in changes:
|
|
path = _record_path(out, change["id"])
|
|
previous = json.loads(path.read_text()) if path.exists() else {}
|
|
entry = indexed.get(change["id"])
|
|
if entry is None:
|
|
entry = {
|
|
"item_id": change["id"],
|
|
"change": change,
|
|
"input_digest": previous.get("input_digest") or hashlib.sha256(canonical(change)).hexdigest(),
|
|
}
|
|
entries.append(entry)
|
|
elif entry["change"] != change or entry["item_id"] != change["id"]:
|
|
raise ValueError("existing semgrep inputs do not match frozen change")
|
|
if previous.get("done") is True:
|
|
print(f"{change['id']}: skipped (done)", flush=True)
|
|
continue
|
|
pending.append(entry)
|
|
_write_json(out / "inputs.json", entries)
|
|
if not pending:
|
|
return 0
|
|
repos = json.loads(Path(repos_path).expanduser().read_text())
|
|
failed = False
|
|
for entry in pending:
|
|
record = _scan_change(entry["change"], repos, rules, binary=binary)
|
|
record["input_digest"] = entry["input_digest"]
|
|
_write_json(_record_path(out, entry["item_id"]), record)
|
|
failed |= record["status"] == "instrument_failure"
|
|
return 2 if failed else 0
|
|
|
|
|
|
def main(argv=None):
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
commands = parser.add_subparsers(dest="command", required=True)
|
|
rules = commands.add_parser("rules", help="convert the ruleset and the facts into Semgrep rules")
|
|
rules.add_argument("--out", type=Path, required=True)
|
|
rules.add_argument("--ruleset", type=Path, default=DEFAULT_RULESET_DIR)
|
|
rules.add_argument("--facts", type=Path, default=DEFAULT_FACTS_DIR)
|
|
rules.add_argument("--min-precision", type=float, default=0.0, help="the hook's min_emit_precision floor")
|
|
rules.add_argument("--semgrep", help="semgrep binary (default: PATH)")
|
|
scan = commands.add_parser("run", help="run Semgrep over each frozen change's changed files")
|
|
scan.add_argument("manifest", type=Path)
|
|
scan.add_argument("--slice", default="1")
|
|
scan.add_argument("--repos", type=Path, required=True)
|
|
scan.add_argument("--rules", type=Path, required=True)
|
|
scan.add_argument("--out", type=Path, required=True)
|
|
scan.add_argument("--only", choices=("vulnerability", "ordinary", "all"), default="all")
|
|
scan.add_argument("--semgrep", help="semgrep binary (default: PATH)")
|
|
args = parser.parse_args(argv)
|
|
if args.command == "rules":
|
|
report = write_rules(
|
|
args.out, ruleset_dir=args.ruleset, facts_dir=args.facts, min_precision=args.min_precision, semgrep=args.semgrep
|
|
)
|
|
validation = report["validation"]
|
|
print(
|
|
f"regex converted {len(report['regex']['converted'])}, skipped {len(report['regex']['skipped'])}; "
|
|
f"facts converted {len(report['facts']['converted'])}, skipped {len(report['facts']['skipped'])}, "
|
|
f"lossy {len(report['facts']['lossy'])}; "
|
|
f"obligations skipped {report['obligations']['skipped_tables']}; "
|
|
f"validation {'not run' if validation is None else 'exit ' + str(validation['exit'])}"
|
|
)
|
|
return 0 if validation is None or validation["exit"] == 0 else 2
|
|
return run(args.manifest, args.repos, args.rules, args.out.expanduser(), slice_id=args.slice, only=args.only, semgrep=args.semgrep)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|