mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-17 20:19:29 +00:00
feat(pair-corpus-honesty): shared function naming, harvest library, catalog generator
Tasks 3.1, 3.2 (harvest library) and the Scorer-owned naming helper. - semantic/functions.py: one source of function names for scorer, overlay and catalog generator (declarator wins over parser name; JS/TS bare-param arrows and undeclared methods are <anon>; C++ Class::method resolves to its unqualified tail; Perl sub; statements never name). - benchmarks/pairs/harvest.py: shared library with name / line_range / hunk / enclosing modes, comment-blind masking (triple-quote aware), upstream_start headers, code-safe quoted-only redaction. - benchmarks/pairs/catalog_gen.py: derive function labels and sinks from the declarator at the mapped sink location, CWE-family sink tokens, --prune with refusal when nothing derived, fails loud without the package; vfc/harvest.py and vfc/generate_catalog.py become shims. - tests/test_pair_harvest.py replaces test_vfc_harvest.py. Co-Authored-By: Claude Fable 5.1 <[email protected]> Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG
This commit is contained in:
Marius Ciepluch
co-authored by
Claude Fable 5.1
Claude Fable 5.1
parent
0116f9bea6
commit
071dea4712
@@ -0,0 +1,395 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Rebuild <slice>/catalog.toml and <slice>/training/manifest.jsonl from <slice>/recipes.toml + excerpts.
|
||||
|
||||
Shared by every harvested slice (vfc, vibe-py, vfc-js, agent-vfc). Recipes carry the
|
||||
labels; the catalog is generated data. Do not hand-edit generated catalogs.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import re
|
||||
import shutil
|
||||
import tomllib
|
||||
from pathlib import Path
|
||||
|
||||
try:
|
||||
from openultrasast.semantic.functions import ANON_FUNCTION, named_function_ranges
|
||||
except ImportError as exc: # never degrade into "no named function": that path deletes corpora under --prune
|
||||
raise SystemExit(f"catalog_gen needs the openultrasast package (run it with the project venv): {exc}") from exc
|
||||
|
||||
ROOT = Path(__file__).resolve().parent
|
||||
LANG_TO_PAIR = {
|
||||
"c": "c_cpp",
|
||||
"cpp": "c_cpp",
|
||||
"perl": "unknown",
|
||||
"javascript": "javascript",
|
||||
"typescript": "typescript",
|
||||
"python": "python",
|
||||
"java": "java",
|
||||
}
|
||||
|
||||
|
||||
def excerpt_rel(recipe: dict[str, object]) -> tuple[str, str]:
|
||||
name = str(recipe["name"])
|
||||
ext = Path(str(recipe.get("path", "a.c"))).suffix or ".c"
|
||||
return f"{name}/vuln{ext}", f"{name}/fixed{ext}"
|
||||
|
||||
|
||||
def _q(value: object) -> str:
|
||||
return str(value).replace("\\", "\\\\").replace('"', '\\"')
|
||||
|
||||
|
||||
def catalog_text(slice_name: str, recipes: list[dict[str, object]]) -> str:
|
||||
lines = [
|
||||
f"# {slice_name} pair slice. Reviewed isolated functions with provenance.",
|
||||
"# Honesty dashboard — not the local 100% CI gate.",
|
||||
"# Do not add these names to LANGUAGE_MANIFESTS.",
|
||||
"# Regenerated by benchmarks/pairs/catalog_gen.py from recipes.toml.",
|
||||
"",
|
||||
]
|
||||
for recipe in recipes:
|
||||
vuln, fixed = excerpt_rel(recipe)
|
||||
lang = LANG_TO_PAIR.get(str(recipe.get("language", "c")), "c_cpp")
|
||||
lines += [
|
||||
"[[pair]]",
|
||||
f'name = "{_q(recipe["name"])}"',
|
||||
f'slice = "{slice_name}"',
|
||||
f'language = "{lang}"',
|
||||
f'origin = "{_q(recipe.get("origin", ""))}"',
|
||||
f'license = "{_q(recipe.get("license", ""))}"',
|
||||
f'repo = "{_q(recipe.get("repo", ""))}"',
|
||||
f'cve = "{_q(recipe.get("cve", ""))}"',
|
||||
f'cwe = "{_q(recipe.get("cwe", ""))}"',
|
||||
f'commit = "{_q(recipe.get("commit", ""))}"',
|
||||
f'parent = "{_q(recipe.get("parent", ""))}"',
|
||||
f'commit_url = "{_q(recipe.get("commit_url", ""))}"',
|
||||
f'vuln = "{vuln}"',
|
||||
f'fixed = "{fixed}"',
|
||||
f'relpath = "{_q(recipe.get("relpath", recipe.get("path", "")))}"',
|
||||
"min_recall = 1.0",
|
||||
'fix_policy = "silent"',
|
||||
f'provenance = "{_q(recipe.get("provenance", "human"))}"',
|
||||
f'split = "{_q(recipe.get("split", "train"))}"',
|
||||
]
|
||||
if recipe.get("known_limit"):
|
||||
lines.append(f'known_limit = "{_q(recipe["known_limit"])}"')
|
||||
lines += [
|
||||
"",
|
||||
"[[pair.expected]]",
|
||||
f'cwe = "{_q(recipe.get("cwe", ""))}"',
|
||||
f'class = "{_q(recipe.get("class", ""))}"',
|
||||
f'path = "{_q(recipe.get("relpath", recipe.get("path", "")))}"',
|
||||
]
|
||||
if recipe.get("function"):
|
||||
lines.append(f'function = "{_q(recipe["function"])}"')
|
||||
lines.append(f'mechanism = "{_q(recipe.get("mechanism", "other"))}"')
|
||||
if recipe.get("rule_id"):
|
||||
lines.append(f'rule_id = "{_q(recipe["rule_id"])}"')
|
||||
if recipe.get("sink"):
|
||||
lines.append(f'sink = "{_q(recipe["sink"])}"')
|
||||
if recipe.get("evidence"):
|
||||
lines.append(f'evidence = "{_q(recipe["evidence"])}"')
|
||||
lines.append("")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def training_rows(recipes: list[dict[str, object]]) -> list[dict[str, object]]:
|
||||
rows: list[dict[str, object]] = []
|
||||
for recipe in recipes:
|
||||
vuln, fixed = excerpt_rel(recipe)
|
||||
base = {
|
||||
"pair": recipe["name"],
|
||||
"cve": recipe.get("cve", ""),
|
||||
"year": int(recipe.get("year", 0) or 0),
|
||||
"cwe": recipe.get("cwe", ""),
|
||||
"class": recipe.get("class", ""),
|
||||
"project": recipe.get("project", recipe.get("origin", "")),
|
||||
"repo": recipe.get("repo", ""),
|
||||
"license": recipe.get("license", ""),
|
||||
"function": recipe.get("function", ""),
|
||||
"mechanism": recipe.get("mechanism", "other"),
|
||||
"provenance": recipe.get("provenance", "human"),
|
||||
"split": recipe.get("split", "train"),
|
||||
"relpath": recipe.get("relpath", recipe.get("path", "")),
|
||||
"commit_url": recipe.get("commit_url", ""),
|
||||
}
|
||||
rows.append({**base, "id": f"{recipe['name']}:vuln", "label": "vuln", "file": vuln, "commit": recipe.get("parent", "")})
|
||||
rows.append({**base, "id": f"{recipe['name']}:fixed", "label": "fixed", "file": fixed, "commit": recipe.get("commit", "")})
|
||||
return rows
|
||||
|
||||
|
||||
# Sink call tokens per language, keyed by the CWE family the recipe declares, so a stray
|
||||
# `re.exec(` in a code-injection excerpt is never mistaken for the labeled sink.
|
||||
_SINK_TOKENS: dict[str, dict[str, dict[str, str]]] = {
|
||||
"javascript": {
|
||||
"CWE-94": {"eval": "eval", "Function": "eval"},
|
||||
"CWE-95": {"eval": "eval", "Function": "eval"},
|
||||
"CWE-78": {"exec": "exec", "execSync": "exec", "execFile": "exec", "execFileSync": "exec", "spawn": "exec", "spawnSync": "exec"},
|
||||
"CWE-22": {
|
||||
"join": "path",
|
||||
"resolve": "path",
|
||||
"readFile": "path",
|
||||
"readFileSync": "path",
|
||||
"createReadStream": "path",
|
||||
"sendFile": "path",
|
||||
"normalize": "path",
|
||||
},
|
||||
"CWE-89": {"query": "query", "execute": "query"},
|
||||
},
|
||||
"python": {
|
||||
"CWE-94": {"eval": "eval", "exec": "eval"},
|
||||
"CWE-95": {"eval": "eval", "exec": "eval"},
|
||||
"CWE-78": {
|
||||
"system": "os.system",
|
||||
"popen": "os.system",
|
||||
"run": "os.system",
|
||||
"Popen": "os.system",
|
||||
"call": "os.system",
|
||||
"check_output": "os.system",
|
||||
},
|
||||
"CWE-89": {"execute": "execute", "executemany": "execute"},
|
||||
"CWE-502": {"loads": "pickle.loads", "load": "pickle.loads"},
|
||||
"CWE-22": {"open": "open", "join": "path", "send_file": "path"},
|
||||
},
|
||||
"c": {
|
||||
"CWE-119": {"memcpy": "memcpy", "memmove": "memcpy", "strcpy": "strcpy", "strcat": "strcpy", "sprintf": "printf"},
|
||||
"CWE-120": {"strcpy": "strcpy", "strcat": "strcpy", "memcpy": "memcpy"},
|
||||
"CWE-78": {"system": "system", "popen": "system"},
|
||||
"CWE-134": {"printf": "printf", "sprintf": "printf", "fprintf": "printf"},
|
||||
},
|
||||
"java": {
|
||||
"CWE-78": {"exec": "exec"},
|
||||
"CWE-89": {
|
||||
"executeQuery": "executeQuery",
|
||||
"execute": "executeQuery",
|
||||
"prepareCall": "executeQuery",
|
||||
"prepareStatement": "executeQuery",
|
||||
},
|
||||
"CWE-502": {"readObject": "readObject"},
|
||||
"CWE-327": {"getInstance": "MessageDigest"},
|
||||
},
|
||||
}
|
||||
_SINK_TOKENS["typescript"] = _SINK_TOKENS["javascript"]
|
||||
_SINK_TOKENS["cpp"] = _SINK_TOKENS["c"]
|
||||
_REGEX_EXEC = re.compile(r"(?:/[^/\n]*/[gimsuy]*|\)|\b(?:re|regex|regexp|pattern|matcher)[A-Za-z_]*)\.exec\s*\(")
|
||||
_HEADER_LINE = re.compile(r"^\s*(//|#|/\*|\*|\*/)")
|
||||
_UPSTREAM_START = re.compile(r"^(?://|#| \*)\s*upstream_start:\s*(\d+)\s*$", re.M)
|
||||
|
||||
|
||||
def _body_start(lines: list[str]) -> int:
|
||||
"""0-based index of the first line after the provenance header block (leading comment lines plus blanks)."""
|
||||
index = 0
|
||||
while index < len(lines) and (_HEADER_LINE.match(lines[index]) or not lines[index].strip()):
|
||||
index += 1
|
||||
return index
|
||||
|
||||
|
||||
def _language(recipe: dict[str, object]) -> str:
|
||||
return {"python": "python", "typescript": "typescript", "javascript": "javascript", "java": "java", "cpp": "cpp"}.get(
|
||||
str(recipe.get("language", "c")), "c"
|
||||
)
|
||||
|
||||
|
||||
def _ranges(path: Path, language: str) -> tuple[tuple[str, int, int], ...] | None:
|
||||
ranges = named_function_ranges(str(path), path.read_text(errors="ignore"), language)
|
||||
if ranges is None:
|
||||
return None # parser rejected the excerpt; a regex label would not scope
|
||||
return tuple(r for r in ranges if r[0] != ANON_FUNCTION)
|
||||
|
||||
|
||||
def _innermost(ranges: tuple[tuple[str, int, int], ...], line: int) -> str | None:
|
||||
inside = [r for r in ranges if r[1] <= line <= r[2]]
|
||||
return min(inside, key=lambda r: r[2] - r[1])[0] if inside else None
|
||||
|
||||
|
||||
def _sink_tokens(recipe: dict[str, object]) -> dict[str, str]:
|
||||
by_cwe = _SINK_TOKENS.get(_language(recipe), {})
|
||||
tokens = by_cwe.get(str(recipe.get("cwe", "")), {})
|
||||
if not tokens: # unknown CWE family: accept any token of the language
|
||||
for family in by_cwe.values():
|
||||
tokens = {**tokens, **family}
|
||||
return tokens
|
||||
|
||||
|
||||
def _sink_on_line(line: str, tokens: dict[str, str]) -> str | None:
|
||||
if _REGEX_EXEC.search(line):
|
||||
return None
|
||||
for token, sink_id in tokens.items():
|
||||
if re.search(r"(?<![\w$])" + re.escape(token) + r"\s*\(", line):
|
||||
return sink_id
|
||||
return None
|
||||
|
||||
|
||||
def _sink_line_in_excerpt(recipe: dict[str, object], text: str, ranges: tuple[tuple[str, int, int], ...]) -> tuple[int | None, str | None]:
|
||||
"""Excerpt line of the labeled sink: mapped from ``sink_location`` + ``upstream_start`` when possible, else the
|
||||
unique named function that contains a sink token of the recipe's CWE family."""
|
||||
lines = text.splitlines()
|
||||
tokens = _sink_tokens(recipe)
|
||||
if not tokens:
|
||||
return None, None
|
||||
location = str(recipe.get("sink_location") or "")
|
||||
parts = location.rsplit(":", 2)
|
||||
relpath = str(recipe.get("relpath", recipe.get("path", "")))
|
||||
match = _UPSTREAM_START.search(text)
|
||||
if (
|
||||
len(parts) >= 2
|
||||
and parts[1].isdigit()
|
||||
and match
|
||||
and (parts[0] == relpath or relpath.endswith(parts[0]) or Path(parts[0]).name == Path(relpath).name)
|
||||
):
|
||||
mapped = int(parts[1]) - int(match.group(1)) + _body_start(lines) + 1
|
||||
for delta in sorted(range(-10, 11), key=abs): # tarball-vs-blob drift is usually a few lines
|
||||
line_no = mapped + delta
|
||||
if 1 <= line_no <= len(lines):
|
||||
sink_id = _sink_on_line(lines[line_no - 1], tokens)
|
||||
if sink_id:
|
||||
return line_no, sink_id
|
||||
mapped_fallback = mapped if 1 <= mapped <= len(lines) and re.search(r"[A-Za-z_$][\w$.]*\s*\(", lines[mapped - 1]) else None
|
||||
else:
|
||||
mapped_fallback = None
|
||||
# Drift (SecBench.js locations come from the npm tarball, not the commit blob): fall back to the
|
||||
# sink tokens of the CWE family anywhere in the excerpt, accepted only when they name one function.
|
||||
all_tokens: dict[str, str] = {}
|
||||
for family in _SINK_TOKENS.get(_language(recipe), {}).values():
|
||||
all_tokens.update(family)
|
||||
for candidate_tokens in (tokens, all_tokens):
|
||||
hits = [(index, sink) for index, line in enumerate(lines, start=1) for sink in [_sink_on_line(line, candidate_tokens)] if sink]
|
||||
owners = {name for index, _ in hits for name in [_innermost(ranges, index)] if name}
|
||||
if len(owners) == 1:
|
||||
line_no, sink_id = next((index, sink) for index, sink in hits if _innermost(ranges, index) in owners)
|
||||
return line_no, sink_id
|
||||
# The upstream location is ground truth even when the sink is a project wrapper (`proc(cmd)`):
|
||||
# label the function enclosing the mapped line and emit no sink id.
|
||||
if mapped_fallback is not None:
|
||||
return mapped_fallback, None
|
||||
return None, None
|
||||
|
||||
|
||||
def derive_function(slice_root: Path, recipe: dict[str, object]) -> str | None:
|
||||
"""Label = innermost named function enclosing the sink (from ``sink_location`` + ``upstream_start``).
|
||||
|
||||
Falls back to the function enclosing the first changed body line only when the
|
||||
recipe has no usable sink location, and says so. Never returns ``<anon>`` or a
|
||||
reserved word; returns None (skip the pair) when nothing names a function.
|
||||
"""
|
||||
vuln, fixed = excerpt_rel(recipe)
|
||||
vuln_path, fixed_path = slice_root / vuln, slice_root / fixed
|
||||
language = _language(recipe)
|
||||
ranges = _ranges(vuln_path, language)
|
||||
if not ranges:
|
||||
return None
|
||||
text = vuln_path.read_text(errors="ignore")
|
||||
sink_line, sink_id = _sink_line_in_excerpt(recipe, text, ranges)
|
||||
if sink_line is not None:
|
||||
name = _innermost(ranges, sink_line)
|
||||
if name:
|
||||
if sink_id and not recipe.get("sink"):
|
||||
recipe["sink"] = sink_id
|
||||
return name
|
||||
print(f"weak: {recipe.get('name')}: sink line {sink_line} is not inside a named function; falling back to the diff")
|
||||
elif recipe.get("sink_location"):
|
||||
print(
|
||||
f"weak: {recipe.get('name')}: no sink token of the CWE family near the mapped sink_location; label from diff, no sink emitted"
|
||||
)
|
||||
a = text.splitlines()
|
||||
b = fixed_path.read_text(errors="ignore").splitlines()
|
||||
a_body, b_body = a[_body_start(a) :], b[_body_start(b) :]
|
||||
offset = _body_start(a)
|
||||
|
||||
first_diff = _first_semantic_change(a_body, b_body)
|
||||
return _innermost(ranges, offset + first_diff + 1) or ranges[0][0]
|
||||
|
||||
|
||||
def _first_semantic_change(a: list[str], b: list[str]) -> int:
|
||||
"""0-based index in ``a`` of the first hunk that changes more than whitespace (formatting hunks never label a pair)."""
|
||||
import difflib
|
||||
|
||||
def norm(chunk: list[str]) -> str:
|
||||
return re.sub(r"\s+", "", "".join(chunk))
|
||||
|
||||
matcher = difflib.SequenceMatcher(a=a, b=b, autojunk=False)
|
||||
for tag, i1, i2, j1, j2 in matcher.get_opcodes():
|
||||
if tag == "equal":
|
||||
continue
|
||||
if norm(a[i1:i2]) == norm(b[j1:j2]):
|
||||
continue
|
||||
return i1 if i2 > i1 else max(i1 - 1, 0)
|
||||
return min(len(a), len(b))
|
||||
|
||||
|
||||
def _stamp_function(path: Path, function: str) -> None:
|
||||
"""Fill the empty ``function:`` header line of a hunk-harvested excerpt."""
|
||||
text = path.read_text(errors="ignore")
|
||||
updated = re.sub(r"^((?://|#| \*) function:).*$", rf"\1 {function}", text, count=1, flags=re.M)
|
||||
if updated != text:
|
||||
path.write_text(updated)
|
||||
|
||||
|
||||
def generate(slice_name: str, *, prune: bool = False) -> tuple[Path, int]:
|
||||
slice_root = ROOT / slice_name
|
||||
loaded = [dict(item) for item in tomllib.loads((slice_root / "recipes.toml").read_text()).get("recipe", [])]
|
||||
recipes: list[dict[str, object]] = []
|
||||
pending: list[dict[str, object]] = []
|
||||
for recipe in loaded:
|
||||
vuln, fixed = excerpt_rel(recipe)
|
||||
if (slice_root / vuln).is_file() and (slice_root / fixed).is_file():
|
||||
if not recipe.get("function"):
|
||||
derived = derive_function(slice_root, recipe)
|
||||
if not derived:
|
||||
print(f"skip {recipe['name']}: no named function in the excerpt (label would not scope)")
|
||||
continue
|
||||
recipe["function"] = derived
|
||||
_stamp_function(slice_root / vuln, derived)
|
||||
_stamp_function(slice_root / fixed, derived)
|
||||
if str(recipe.get("reviewer", "")).lower() == "pending":
|
||||
pending.append(recipe)
|
||||
continue
|
||||
recipes.append(recipe)
|
||||
else:
|
||||
print(f"skip {recipe['name']}: excerpt missing (harvest failed or not fetched)")
|
||||
catalog = slice_root / "catalog.toml"
|
||||
catalog.write_text(catalog_text(slice_name, recipes), encoding="utf-8")
|
||||
candidates = slice_root / "catalog-candidates.toml"
|
||||
if prune:
|
||||
if not recipes and not pending:
|
||||
raise SystemExit(f"refusing --prune for {slice_name}: no recipe derived a label, nothing would survive")
|
||||
keep = {str(r["name"]) for r in recipes} | {str(r["name"]) for r in pending}
|
||||
for entry in sorted(slice_root.iterdir()):
|
||||
if entry.is_dir() and entry.name not in keep and entry.name not in {"training", "__pycache__"}:
|
||||
shutil.rmtree(entry)
|
||||
print(f"pruned {entry.name}: excerpt not in the catalog or the candidates")
|
||||
if pending:
|
||||
# Unreviewed rows never enter the loaded catalog (Req 5.3); they wait here for a reviewer.
|
||||
candidates.write_text(
|
||||
'# CANDIDATES awaiting human review (reviewer = "pending"). Not loaded by load_pair_catalog.\n'
|
||||
"# Set `reviewer` in recipes.toml after confirming vulnerability, mechanism, and license; then regenerate.\n\n"
|
||||
+ catalog_text(slice_name, pending),
|
||||
encoding="utf-8",
|
||||
)
|
||||
print(f"{len(pending)} rows await review -> {candidates}")
|
||||
elif candidates.exists():
|
||||
candidates.unlink()
|
||||
training = slice_root / "training" / "manifest.jsonl"
|
||||
training.parent.mkdir(parents=True, exist_ok=True)
|
||||
with training.open("w", encoding="utf-8") as handle:
|
||||
for row in training_rows(recipes):
|
||||
handle.write(json.dumps(row, sort_keys=True) + "\n")
|
||||
return catalog, len(recipes)
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--slice", default="vfc")
|
||||
parser.add_argument(
|
||||
"--prune", action="store_true", help="delete excerpt directories that ended up neither in the catalog nor among the candidates"
|
||||
)
|
||||
args = parser.parse_args(argv)
|
||||
catalog, count = generate(args.slice, prune=args.prune)
|
||||
print(f"wrote {catalog} ({count} pairs)")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,604 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Shared pair harvest library: extract one function from public parent/fix blobs.
|
||||
|
||||
Maintainer-only; CI stays offline (tests run the extractors on local strings).
|
||||
Does not clone mozilla-central, gecko-dev, or Chromium. Advisory-only recipes
|
||||
(no parent and commit) are rejected. FixFox is not a fetch source.
|
||||
|
||||
Modes (recipe ``mode``, default ``name``):
|
||||
- ``name`` anchor on the declarator of ``function`` (comment and string mentions ignored)
|
||||
- ``line_range`` copy ``line_start..line_end`` from the parent blob and
|
||||
``fix_line_start..fix_line_end`` (default: same) from the fix blob
|
||||
- ``hunk`` take the function enclosing the first changed hunk on each side
|
||||
- ``enclosing`` take the function enclosing ``line`` on the parent blob and
|
||||
``fix_line`` (default: same) on the fix blob; ``fix_path`` may differ
|
||||
|
||||
Languages: c, cpp, java, javascript, typescript (brace-balanced), python (indentation), perl (``sub``).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import difflib
|
||||
import re
|
||||
import sys
|
||||
import tomllib
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from urllib.parse import urlparse
|
||||
from urllib.request import Request, urlopen
|
||||
|
||||
ROOT = Path(__file__).resolve().parent
|
||||
ALLOWED_HOSTS = {
|
||||
"raw.githubusercontent.com",
|
||||
"github.com",
|
||||
"phabricator.services.mozilla.com",
|
||||
"hg.mozilla.org",
|
||||
"hg-edge.mozilla.org",
|
||||
}
|
||||
BLOCKED_TOKENS = ("fixfox", "zenodo.org/record")
|
||||
MODES = ("name", "line_range", "hunk", "enclosing")
|
||||
_SUFFIX_LANGUAGE = {
|
||||
".c": "c",
|
||||
".h": "c",
|
||||
".cc": "cpp",
|
||||
".cpp": "cpp",
|
||||
".hpp": "cpp",
|
||||
".java": "java",
|
||||
".js": "javascript",
|
||||
".mjs": "javascript",
|
||||
".cjs": "javascript",
|
||||
".ts": "typescript",
|
||||
".tsx": "typescript",
|
||||
".py": "python",
|
||||
".pl": "perl",
|
||||
".pm": "perl",
|
||||
}
|
||||
_HASH_COMMENT = {"python", "perl"}
|
||||
_DEF_RE = re.compile(r"^(\s*)(?:async\s+)?def\s+([A-Za-z_]\w*)\s*\(")
|
||||
|
||||
|
||||
class RecipeError(ValueError):
|
||||
"""Invalid harvest recipe or extraction failure."""
|
||||
|
||||
|
||||
# --- masking -----------------------------------------------------------------
|
||||
|
||||
|
||||
def mask_comments_and_strings(source: str, language: str = "c") -> str:
|
||||
"""Return ``source`` with comment and string-literal bodies replaced by spaces.
|
||||
|
||||
Same length, newlines preserved, so indexes and line numbers line up with the
|
||||
original. Braces and identifiers inside comments or strings therefore never
|
||||
anchor an extraction.
|
||||
"""
|
||||
out = list(source)
|
||||
n = len(source)
|
||||
i = 0
|
||||
hash_comments = language in _HASH_COMMENT
|
||||
slash_comments = not hash_comments
|
||||
while i < n:
|
||||
ch = source[i]
|
||||
nxt = source[i + 1] if i + 1 < n else ""
|
||||
if hash_comments and ch == "#":
|
||||
while i < n and source[i] != "\n":
|
||||
out[i] = " "
|
||||
i += 1
|
||||
continue
|
||||
if slash_comments and ch == "/" and nxt == "/":
|
||||
while i < n and source[i] != "\n":
|
||||
out[i] = " "
|
||||
i += 1
|
||||
continue
|
||||
if slash_comments and ch == "/" and nxt == "*":
|
||||
out[i] = " "
|
||||
out[i + 1] = " "
|
||||
i += 2
|
||||
while i < n and not (source[i] == "*" and i + 1 < n and source[i + 1] == "/"):
|
||||
if source[i] != "\n":
|
||||
out[i] = " "
|
||||
i += 1
|
||||
if i < n:
|
||||
out[i] = " "
|
||||
if i + 1 < n:
|
||||
out[i + 1] = " "
|
||||
i += 2
|
||||
continue
|
||||
if language == "python" and ch in "\"'" and source.startswith(ch * 3, i):
|
||||
quote = ch * 3
|
||||
for k in range(i, min(i + 3, n)):
|
||||
out[k] = " " # blank the delimiters too: a closing triple quote at column 0 is not a dedent
|
||||
i += 3
|
||||
while i < n and not source.startswith(quote, i):
|
||||
if source[i] != "\n":
|
||||
out[i] = " "
|
||||
i += 1
|
||||
for k in range(i, min(i + 3, n)):
|
||||
out[k] = " "
|
||||
i += 3
|
||||
continue
|
||||
if ch in "\"'`":
|
||||
quote = ch
|
||||
i += 1
|
||||
while i < n and source[i] != quote:
|
||||
if source[i] == "\\" and i + 1 < n:
|
||||
out[i] = " "
|
||||
if source[i + 1] != "\n":
|
||||
out[i + 1] = " "
|
||||
i += 2
|
||||
continue
|
||||
if source[i] == "\n" and quote != "`":
|
||||
break # unterminated single-line literal; stop masking at EOL
|
||||
if source[i] != "\n":
|
||||
out[i] = " "
|
||||
i += 1
|
||||
i += 1
|
||||
continue
|
||||
i += 1
|
||||
return "".join(out)
|
||||
|
||||
|
||||
# --- extraction --------------------------------------------------------------
|
||||
|
||||
|
||||
def language_for(recipe: dict[str, Any]) -> str:
|
||||
language = str(recipe.get("language") or "").lower()
|
||||
if language:
|
||||
return {"c++": "cpp", "js": "javascript", "ts": "typescript"}.get(language, language)
|
||||
return _SUFFIX_LANGUAGE.get(Path(str(recipe.get("path", ""))).suffix.lower(), "c")
|
||||
|
||||
|
||||
def extract_function(source: str, function_name: str, language: str = "c") -> str:
|
||||
"""Copy the named function from its declarator through its balanced body."""
|
||||
if language == "python":
|
||||
return extract_python_block(source, function_name)
|
||||
masked = mask_comments_and_strings(source, language)
|
||||
needle = function_name
|
||||
idx = 0
|
||||
while True:
|
||||
found = masked.find(needle, idx)
|
||||
if found < 0:
|
||||
raise RecipeError(f"function not found: {function_name}")
|
||||
idx = found + len(needle)
|
||||
before = masked[found - 1] if found else " "
|
||||
if before.isalnum() or before == "_" or before == ".":
|
||||
continue
|
||||
rest = masked[idx:].lstrip()
|
||||
if not (rest.startswith("(") or rest.startswith("<") or rest.startswith("{")):
|
||||
continue
|
||||
brace = masked.find("{", found)
|
||||
semi = masked.find(";", found)
|
||||
if brace < 0 or (0 <= semi < brace):
|
||||
continue # prototype / declaration, keep searching
|
||||
break
|
||||
start = _declarator_start(source, masked, found)
|
||||
end = _balanced_end(masked, brace)
|
||||
return source[start : end + 1].strip() + "\n"
|
||||
|
||||
|
||||
def extract_python_block(source: str, function_name: str) -> str:
|
||||
lines = source.splitlines(keepends=True)
|
||||
masked = mask_comments_and_strings(source, "python").splitlines(keepends=True)
|
||||
for index, line in enumerate(lines):
|
||||
match = _DEF_RE.match(line)
|
||||
if match is None or match.group(2) != function_name:
|
||||
continue
|
||||
indent = len(match.group(1))
|
||||
start = index
|
||||
while (
|
||||
start > 0 and lines[start - 1].lstrip().startswith("@") and (len(lines[start - 1]) - len(lines[start - 1].lstrip())) == indent
|
||||
):
|
||||
start -= 1 # decorators belong to the function
|
||||
end = _python_block_end(masked, index, indent)
|
||||
return "".join(lines[start:end]).rstrip("\n") + "\n"
|
||||
raise RecipeError(f"function not found: {function_name}")
|
||||
|
||||
|
||||
def extract_line_range(source: str, start: int, end: int) -> str:
|
||||
"""Inclusive 1-based line range."""
|
||||
lines = source.splitlines(keepends=True)
|
||||
if start < 1 or end < start or end > len(lines):
|
||||
raise RecipeError(f"line range {start}..{end} outside 1..{len(lines)}")
|
||||
return "".join(lines[start - 1 : end]).rstrip("\n") + "\n"
|
||||
|
||||
|
||||
def function_bounds_at(source: str, line: int, language: str = "c") -> tuple[int, int] | None:
|
||||
"""1-based inclusive (start, end) of the function enclosing ``line``; None when not inside one."""
|
||||
lines = source.splitlines(keepends=True)
|
||||
if line < 1 or line > len(lines):
|
||||
return None
|
||||
if language == "python":
|
||||
return _python_bounds_at(lines, line)
|
||||
masked = mask_comments_and_strings(source, language)
|
||||
offsets = _line_offsets(source)
|
||||
target = offsets[line - 1]
|
||||
depth = 0
|
||||
open_at: int | None = None
|
||||
for pos, ch in enumerate(masked):
|
||||
if ch == "{":
|
||||
if depth == 0:
|
||||
open_at = pos
|
||||
depth += 1
|
||||
elif ch == "}":
|
||||
depth -= 1
|
||||
if depth == 0 and open_at is not None:
|
||||
if open_at <= target <= pos or _line_of(offsets, open_at) <= line <= _line_of(offsets, pos):
|
||||
start = _declarator_start(source, masked, open_at)
|
||||
return _line_of(offsets, start), _line_of(offsets, pos)
|
||||
open_at = None
|
||||
return None
|
||||
|
||||
|
||||
WHOLE_FILE_MAX_LINES = 150
|
||||
|
||||
|
||||
def extract_hunk(parent_source: str, fix_source: str, language: str = "c") -> tuple[str, str]:
|
||||
"""Functions enclosing the first changed hunk that sits inside a function on both sides.
|
||||
|
||||
Fallback for small modules (npm one-file packages, route files) whose change is at
|
||||
module level: when both blobs are at most ``WHOLE_FILE_MAX_LINES`` lines, the whole
|
||||
file is the excerpt. Larger module-level changes are a RecipeError.
|
||||
"""
|
||||
parent_lines = parent_source.splitlines()
|
||||
fix_lines = fix_source.splitlines()
|
||||
matcher = difflib.SequenceMatcher(a=parent_lines, b=fix_lines, autojunk=False)
|
||||
changed = [(tag, i1, i2, j1, j2) for tag, i1, i2, j1, j2 in matcher.get_opcodes() if tag != "equal"]
|
||||
if not changed:
|
||||
raise RecipeError("parent and fix blobs are identical")
|
||||
for _tag, i1, i2, j1, j2 in changed:
|
||||
parent_line = i1 + 1 if i2 > i1 else max(i1, 1)
|
||||
fix_line = j1 + 1 if j2 > j1 else max(j1, 1)
|
||||
parent_bounds = function_bounds_at(parent_source, parent_line, language)
|
||||
fix_bounds = function_bounds_at(fix_source, fix_line, language)
|
||||
if parent_bounds is not None and fix_bounds is not None:
|
||||
return (
|
||||
extract_line_range(parent_source, *parent_bounds),
|
||||
extract_line_range(fix_source, *fix_bounds),
|
||||
)
|
||||
if len(parent_lines) <= WHOLE_FILE_MAX_LINES and len(fix_lines) <= WHOLE_FILE_MAX_LINES:
|
||||
return parent_source.rstrip("\n") + "\n", fix_source.rstrip("\n") + "\n"
|
||||
raise RecipeError("no changed hunk is inside a function on both sides and the file is too large to vendor whole")
|
||||
|
||||
|
||||
def _declarator_start(source: str, masked: str, anchor: int) -> int:
|
||||
"""Start offset of the declarator statement that owns ``anchor`` (walk back to the previous ';', '}' or preprocessor line)."""
|
||||
pos = anchor - 1
|
||||
while pos >= 0:
|
||||
ch = masked[pos]
|
||||
if ch in ";}":
|
||||
break
|
||||
if ch == "\n":
|
||||
line_start = masked.rfind("\n", 0, pos) + 1
|
||||
if masked[line_start:pos].lstrip().startswith("#"):
|
||||
break
|
||||
pos -= 1
|
||||
start = pos + 1
|
||||
# Skip blank and comment-only lines between the boundary and the declarator.
|
||||
while True:
|
||||
line_end = source.find("\n", start)
|
||||
if line_end < 0:
|
||||
break
|
||||
if masked[start:line_end].strip():
|
||||
break
|
||||
start = line_end + 1
|
||||
return start
|
||||
|
||||
|
||||
def _balanced_end(masked: str, brace: int) -> int:
|
||||
depth = 0
|
||||
for pos in range(brace, len(masked)):
|
||||
ch = masked[pos]
|
||||
if ch == "{":
|
||||
depth += 1
|
||||
elif ch == "}":
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
return pos
|
||||
raise RecipeError("unbalanced braces")
|
||||
|
||||
|
||||
def _line_offsets(source: str) -> list[int]:
|
||||
offsets = [0]
|
||||
for index, ch in enumerate(source):
|
||||
if ch == "\n":
|
||||
offsets.append(index + 1)
|
||||
return offsets
|
||||
|
||||
|
||||
def _line_of(offsets: list[int], pos: int) -> int:
|
||||
low, high = 0, len(offsets) - 1
|
||||
while low < high:
|
||||
mid = (low + high + 1) // 2
|
||||
if offsets[mid] <= pos:
|
||||
low = mid
|
||||
else:
|
||||
high = mid - 1
|
||||
return low + 1
|
||||
|
||||
|
||||
def _python_block_end(lines: list[str], def_index: int, indent: int) -> int:
|
||||
"""End index of the block starting at ``def_index``. Pass masked lines so the bodies of
|
||||
triple-quoted strings (blank after masking) never terminate the block."""
|
||||
end = def_index + 1
|
||||
last_code = def_index + 1
|
||||
while end < len(lines):
|
||||
line = lines[end]
|
||||
stripped = line.strip()
|
||||
if stripped and not stripped.startswith("#"):
|
||||
if len(line) - len(line.lstrip()) <= indent:
|
||||
break
|
||||
last_code = end + 1
|
||||
end += 1
|
||||
return last_code
|
||||
|
||||
|
||||
def _python_bounds_at(lines: list[str], line: int) -> tuple[int, int] | None:
|
||||
index = line - 1
|
||||
masked = mask_comments_and_strings("".join(lines), "python").splitlines(keepends=True)
|
||||
for def_index in range(index, -1, -1):
|
||||
match = _DEF_RE.match(lines[def_index])
|
||||
if match is None:
|
||||
continue
|
||||
indent = len(match.group(1))
|
||||
end = _python_block_end(masked, def_index, indent)
|
||||
if def_index <= index < end:
|
||||
return def_index + 1, end
|
||||
return None
|
||||
|
||||
|
||||
# --- recipes and IO ----------------------------------------------------------
|
||||
|
||||
|
||||
def provenance_header(recipe: dict[str, Any], *, side: str, sha: str, upstream_start: int | None = None) -> str:
|
||||
lines = [
|
||||
f"Provenance: {recipe.get('repo', '')} {recipe.get('function', '')} ({side}).",
|
||||
f"repo: {recipe.get('repo', '')}",
|
||||
f"commit: {sha}",
|
||||
f"parent: {recipe.get('parent', '')}",
|
||||
f"commit_url: {recipe.get('commit_url', '')}",
|
||||
f"cve: {recipe.get('cve', '')}",
|
||||
f"license: {recipe.get('license', '')}",
|
||||
f"function: {recipe.get('function', '')}",
|
||||
f"relpath: {recipe.get('relpath', recipe.get('path', ''))}",
|
||||
f"provenance: {recipe.get('provenance', 'human')}",
|
||||
f"mechanism: {recipe.get('mechanism', 'other')}",
|
||||
]
|
||||
if upstream_start is not None:
|
||||
lines.append(f"upstream_start: {upstream_start}") # 1-based line of the excerpt body in the upstream file
|
||||
language = language_for(recipe)
|
||||
if language in _HASH_COMMENT:
|
||||
return "\n".join(f"# {line}" for line in lines) + "\n"
|
||||
if language in {"cpp", "java", "javascript", "typescript"}:
|
||||
return "\n".join(f"// {line}" for line in lines) + "\n"
|
||||
return "/* " + "\n * ".join(lines) + "\n */\n"
|
||||
|
||||
|
||||
def validate_recipe(recipe: dict[str, Any]) -> None:
|
||||
name = str(recipe.get("name", ""))
|
||||
if any(token in str(recipe).lower() for token in BLOCKED_TOKENS):
|
||||
raise RecipeError(f"{name}: embargoed or blocked dataset")
|
||||
if not recipe.get("parent") or not recipe.get("commit"):
|
||||
raise RecipeError(f"{name}: advisory-only recipe rejected; parent and commit required")
|
||||
mode = str(recipe.get("mode", "name"))
|
||||
if mode not in MODES:
|
||||
raise RecipeError(f"{name}: unknown mode {mode!r}; expected one of {MODES}")
|
||||
if not recipe.get("path"):
|
||||
raise RecipeError(f"{name}: path required")
|
||||
if mode == "name" and not recipe.get("function"):
|
||||
raise RecipeError(f"{name}: path and function required")
|
||||
if mode == "line_range" and not (isinstance(recipe.get("line_start"), int) and isinstance(recipe.get("line_end"), int)):
|
||||
raise RecipeError(f"{name}: line_range mode needs integer line_start and line_end")
|
||||
if mode == "enclosing" and not isinstance(recipe.get("line"), int):
|
||||
raise RecipeError(f"{name}: enclosing mode needs an integer line")
|
||||
if not recipe.get("license"):
|
||||
raise RecipeError(f"{name}: license required")
|
||||
|
||||
|
||||
def github_raw_url(repo: str, sha: str, path: str) -> str:
|
||||
return f"https://raw.githubusercontent.com/{repo}/{sha}/{path}"
|
||||
|
||||
|
||||
def excerpt_paths(recipe: dict[str, Any], root: Path) -> tuple[Path, Path]:
|
||||
name = str(recipe["name"])
|
||||
ext = Path(str(recipe.get("path", "a.c"))).suffix or ".c"
|
||||
folder = root / name
|
||||
return folder / f"vuln{ext}", folder / f"fixed{ext}"
|
||||
|
||||
|
||||
_TOKEN_PATTERNS: tuple[tuple[re.Pattern[str], str], ...] = (
|
||||
(re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----.*?-----END [A-Z ]*PRIVATE KEY-----", re.DOTALL), "***REDACTED***"),
|
||||
(re.compile(r"(Bearer\s+)[A-Za-z0-9._\-]{12,}"), r"\1***REDACTED***"),
|
||||
(re.compile(r"\bsk-[A-Za-z0-9._\-]{16,}"), "***REDACTED***"),
|
||||
(re.compile(r"\bAKIA[0-9A-Z]{16}\b"), "***REDACTED***"),
|
||||
(re.compile(r"\bgh[pousr]_[A-Za-z0-9]{20,}"), "***REDACTED***"),
|
||||
(re.compile(r"\bxox[baprs]-[A-Za-z0-9-]{10,}"), "***REDACTED***"),
|
||||
(re.compile(r"\bAIza[A-Za-z0-9_\-]{20,}"), "***REDACTED***"),
|
||||
(re.compile(r"([a-zA-Z][a-zA-Z0-9+.\-]*://[^\s:/@]+:)[^\s:/@]{3,}(@)"), r"\1***REDACTED***\2"),
|
||||
# Generic sensitive assignment: ONLY quoted literal values, so redaction never breaks code
|
||||
# (`password = request.form[...]` and `token = getToken()` are code, not secrets).
|
||||
(
|
||||
re.compile(r"(?i)\b(api[_-]?key|secret|token|password|passwd|pwd|access[_-]?key)\b(\s*[=:]\s*)(['\"])[A-Za-z0-9/+_\-]{8,}\3"),
|
||||
r"\1\2\3***REDACTED***\3",
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def redact(body: str) -> str:
|
||||
"""Mask credential-shaped literals in an excerpt without changing code shape (mirrors openultrasast.redaction, quoted-only generic rule)."""
|
||||
for pattern, replacement in _TOKEN_PATTERNS:
|
||||
body = pattern.sub(replacement, body)
|
||||
return body
|
||||
|
||||
|
||||
def write_excerpt(path: Path, recipe: dict[str, Any], *, side: str, sha: str, body: str, upstream_start: int | None = None) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
header = provenance_header(recipe, side=side, sha=sha, upstream_start=upstream_start)
|
||||
path.write_text(header + "\n" + redact(body), encoding="utf-8")
|
||||
|
||||
|
||||
def extract_pair_with_lines(recipe: dict[str, Any], parent_src: str, fix_src: str) -> tuple[tuple[str, int], tuple[str, int]]:
|
||||
"""Like ``extract_pair`` but also returns the 1-based upstream start line of each excerpt body."""
|
||||
language = language_for(recipe)
|
||||
mode = str(recipe.get("mode", "name"))
|
||||
if mode == "line_range":
|
||||
start, end = int(recipe["line_start"]), int(recipe["line_end"])
|
||||
fix_start = int(recipe.get("fix_line_start", start))
|
||||
fix_end = int(recipe.get("fix_line_end", end))
|
||||
return (extract_line_range(parent_src, start, end), start), (extract_line_range(fix_src, fix_start, fix_end), fix_start)
|
||||
if mode == "hunk":
|
||||
return _hunk_with_lines(parent_src, fix_src, language)
|
||||
if mode == "enclosing":
|
||||
line = int(recipe["line"])
|
||||
fix_line = int(recipe.get("fix_line", line))
|
||||
parent_bounds = function_bounds_at(parent_src, line, language)
|
||||
fix_bounds = function_bounds_at(fix_src, fix_line, language)
|
||||
if parent_bounds is None or fix_bounds is None:
|
||||
raise RecipeError(f"{recipe.get('name')}: line {line}/{fix_line} is not inside a function")
|
||||
return (extract_line_range(parent_src, *parent_bounds), parent_bounds[0]), (extract_line_range(fix_src, *fix_bounds), fix_bounds[0])
|
||||
function = str(recipe["function"])
|
||||
parent_text = extract_function(parent_src, function, language)
|
||||
fix_text = extract_function(fix_src, function, language)
|
||||
return (parent_text, _start_line_of(parent_src, parent_text)), (fix_text, _start_line_of(fix_src, fix_text))
|
||||
|
||||
|
||||
def _start_line_of(source: str, excerpt: str) -> int:
|
||||
first = excerpt.splitlines()[0] if excerpt.strip() else ""
|
||||
for index, line in enumerate(source.splitlines(), start=1):
|
||||
if line.rstrip() == first.rstrip():
|
||||
return index
|
||||
return 1
|
||||
|
||||
|
||||
def _hunk_with_lines(parent_source: str, fix_source: str, language: str) -> tuple[tuple[str, int], tuple[str, int]]:
|
||||
parent_lines = parent_source.splitlines()
|
||||
fix_lines = fix_source.splitlines()
|
||||
matcher = difflib.SequenceMatcher(a=parent_lines, b=fix_lines, autojunk=False)
|
||||
changed = [(tag, i1, i2, j1, j2) for tag, i1, i2, j1, j2 in matcher.get_opcodes() if tag != "equal"]
|
||||
if not changed:
|
||||
raise RecipeError("parent and fix blobs are identical")
|
||||
for _tag, i1, i2, j1, j2 in changed:
|
||||
parent_line = i1 + 1 if i2 > i1 else max(i1, 1)
|
||||
fix_line = j1 + 1 if j2 > j1 else max(j1, 1)
|
||||
parent_bounds = function_bounds_at(parent_source, parent_line, language)
|
||||
fix_bounds = function_bounds_at(fix_source, fix_line, language)
|
||||
if parent_bounds is not None and fix_bounds is not None:
|
||||
return (extract_line_range(parent_source, *parent_bounds), parent_bounds[0]), (
|
||||
extract_line_range(fix_source, *fix_bounds),
|
||||
fix_bounds[0],
|
||||
)
|
||||
if len(parent_lines) <= WHOLE_FILE_MAX_LINES and len(fix_lines) <= WHOLE_FILE_MAX_LINES:
|
||||
return (parent_source.rstrip("\n") + "\n", 1), (fix_source.rstrip("\n") + "\n", 1)
|
||||
raise RecipeError("no changed hunk is inside a function on both sides and the file is too large to vendor whole")
|
||||
|
||||
|
||||
def extract_pair(recipe: dict[str, Any], parent_src: str, fix_src: str) -> tuple[str, str]:
|
||||
"""Apply the recipe's mode to two blobs. Pure; unit-tested offline."""
|
||||
language = language_for(recipe)
|
||||
mode = str(recipe.get("mode", "name"))
|
||||
if mode == "line_range":
|
||||
start, end = int(recipe["line_start"]), int(recipe["line_end"])
|
||||
fix_start = int(recipe.get("fix_line_start", start))
|
||||
fix_end = int(recipe.get("fix_line_end", end))
|
||||
return extract_line_range(parent_src, start, end), extract_line_range(fix_src, fix_start, fix_end)
|
||||
if mode == "hunk":
|
||||
return extract_hunk(parent_src, fix_src, language)
|
||||
if mode == "enclosing":
|
||||
line = int(recipe["line"])
|
||||
fix_line = int(recipe.get("fix_line", line))
|
||||
parent_bounds = function_bounds_at(parent_src, line, language)
|
||||
fix_bounds = function_bounds_at(fix_src, fix_line, language)
|
||||
if parent_bounds is None or fix_bounds is None:
|
||||
raise RecipeError(f"{recipe.get('name')}: line {line}/{fix_line} is not inside a function")
|
||||
return extract_line_range(parent_src, *parent_bounds), extract_line_range(fix_src, *fix_bounds)
|
||||
function = str(recipe["function"])
|
||||
return extract_function(parent_src, function, language), extract_function(fix_src, function, language)
|
||||
|
||||
|
||||
def harvest_recipe(recipe: dict[str, Any], root: Path) -> tuple[Path, Path]:
|
||||
validate_recipe(recipe)
|
||||
if str(recipe.get("host", "github")) != "github":
|
||||
raise RecipeError(f"{recipe.get('name')}: live fetch currently implemented for host=github only")
|
||||
parent_src = fetch_url(github_raw_url(str(recipe["repo"]), str(recipe["parent"]), str(recipe["path"])))
|
||||
# fix_path lets a pair take its fixed side from another file of the same snapshot
|
||||
# (Real-Vuln-Benchmark false-positive traps are the fixed twins of vulnerable findings).
|
||||
fix_src = fetch_url(
|
||||
github_raw_url(str(recipe.get("fix_repo", recipe["repo"])), str(recipe["commit"]), str(recipe.get("fix_path", recipe["path"])))
|
||||
)
|
||||
(parent_fn, parent_start), (fix_fn, fix_start) = extract_pair_with_lines(recipe, parent_src, fix_src)
|
||||
vuln_path, fixed_path = excerpt_paths(recipe, root)
|
||||
write_excerpt(vuln_path, recipe, side="vuln", sha=str(recipe["parent"]), body=parent_fn, upstream_start=parent_start)
|
||||
write_excerpt(fixed_path, recipe, side="fixed", sha=str(recipe["commit"]), body=fix_fn, upstream_start=fix_start)
|
||||
return vuln_path, fixed_path
|
||||
|
||||
|
||||
def fetch_url(url: str, timeout: int = 60) -> str:
|
||||
host = urlparse(url).hostname or ""
|
||||
if host not in ALLOWED_HOSTS:
|
||||
raise RecipeError(f"host not allowed: {host}")
|
||||
request = Request(url, headers={"User-Agent": "OpenUltraSAST-pair-harvest"})
|
||||
with urlopen(request, timeout=timeout) as response: # noqa: S310 — host allowlisted
|
||||
return response.read().decode("utf-8", "replace")
|
||||
|
||||
|
||||
def load_recipes(path: Path) -> tuple[dict[str, Any], ...]:
|
||||
payload = tomllib.loads(path.read_text())
|
||||
return tuple(dict(item) for item in payload.get("recipe", []))
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--slice", default="vfc", help="slice directory under benchmarks/pairs (default: vfc)")
|
||||
parser.add_argument("--recipes", type=Path, default=None, help="recipes file (default: <slice>/recipes.toml)")
|
||||
parser.add_argument("--name", help="recipe name to harvest (omit with --all)")
|
||||
parser.add_argument("--all", action="store_true", help="process every recipe")
|
||||
parser.add_argument("--prefix", help="process every recipe whose name starts with this prefix")
|
||||
parser.add_argument("--fetch", action="store_true", help="HTTP-fetch and write excerpts; default is dry-run validate")
|
||||
parser.add_argument("--force", action="store_true", help="overwrite existing excerpts")
|
||||
parser.add_argument("--out", type=Path, default=None, help="excerpt root (default: <slice>/)")
|
||||
args = parser.parse_args(argv)
|
||||
slice_root = ROOT / args.slice
|
||||
recipes_path = args.recipes or slice_root / "recipes.toml"
|
||||
out = args.out or slice_root
|
||||
loaded = load_recipes(recipes_path)
|
||||
recipes = {str(item["name"]): item for item in loaded}
|
||||
selected: list[dict[str, Any]]
|
||||
if args.all:
|
||||
selected = list(loaded)
|
||||
elif args.prefix:
|
||||
selected = [item for item in loaded if str(item["name"]).startswith(args.prefix)]
|
||||
elif args.name:
|
||||
if args.name not in recipes:
|
||||
print(f"unknown recipe: {args.name}", file=sys.stderr)
|
||||
return 2
|
||||
selected = [recipes[args.name]]
|
||||
else:
|
||||
print("pass --name RECIPE, --prefix PREFIX, or --all", file=sys.stderr)
|
||||
return 2
|
||||
failures = 0
|
||||
for recipe in selected:
|
||||
name = str(recipe["name"])
|
||||
try:
|
||||
validate_recipe(recipe)
|
||||
except RecipeError as exc:
|
||||
print(str(exc), file=sys.stderr)
|
||||
failures += 1
|
||||
continue
|
||||
if not args.fetch:
|
||||
print(f"ok {name} parent={recipe['parent']} commit={recipe['commit']} mode={recipe.get('mode', 'name')}")
|
||||
continue
|
||||
vuln_path, fixed_path = excerpt_paths(recipe, out)
|
||||
if vuln_path.is_file() and fixed_path.is_file() and not args.force:
|
||||
print(f"skip {name} (exists)")
|
||||
continue
|
||||
try:
|
||||
harvest_recipe(recipe, out)
|
||||
except (RecipeError, OSError) as exc:
|
||||
print(f"{name}: {exc}", file=sys.stderr)
|
||||
failures += 1
|
||||
continue
|
||||
print(f"wrote {name} -> {vuln_path.parent}")
|
||||
return 1 if failures else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -1,104 +1,16 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Rebuild catalog.toml and training/manifest.jsonl from recipes.toml + excerpts."""
|
||||
"""Thin shim: ``python benchmarks/pairs/catalog_gen.py --slice vfc``."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import tomllib
|
||||
import runpy
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parent
|
||||
RECIPES = ROOT / "recipes.toml"
|
||||
CATALOG = ROOT / "catalog.toml"
|
||||
TRAINING = ROOT / "training" / "manifest.jsonl"
|
||||
|
||||
LANG_TO_PAIR = {"c": "c_cpp", "cpp": "c_cpp", "perl": "unknown", "javascript": "javascript", "python": "python", "java": "java"}
|
||||
|
||||
|
||||
def excerpt_rel(recipe: dict[str, object]) -> tuple[str, str]:
|
||||
name = str(recipe["name"])
|
||||
ext = Path(str(recipe.get("path", "a.c"))).suffix or ".c"
|
||||
return f"{name}/vuln{ext}", f"{name}/fixed{ext}"
|
||||
|
||||
|
||||
def catalog_text(recipes: list[dict[str, object]]) -> str:
|
||||
lines = [
|
||||
"# Real-world VFC slice. Reviewed isolated functions.",
|
||||
"# Honesty dashboard — not the local 100% CI gate.",
|
||||
"# Do not add these names to LANGUAGE_MANIFESTS.",
|
||||
"# Regenerated by generate_catalog.py from recipes.toml.",
|
||||
"",
|
||||
]
|
||||
for recipe in recipes:
|
||||
vuln, fixed = excerpt_rel(recipe)
|
||||
lang = LANG_TO_PAIR.get(str(recipe.get("language", "c")), "c_cpp")
|
||||
lines += [
|
||||
"[[pair]]",
|
||||
f'name = "{recipe["name"]}"',
|
||||
'slice = "vfc"',
|
||||
f'language = "{lang}"',
|
||||
f'origin = "{recipe.get("origin", "")}"',
|
||||
f'license = "{recipe.get("license", "")}"',
|
||||
f'repo = "{recipe.get("repo", "")}"',
|
||||
f'cve = "{recipe.get("cve", "")}"',
|
||||
f'cwe = "{recipe.get("cwe", "")}"',
|
||||
f'commit = "{recipe.get("commit", "")}"',
|
||||
f'parent = "{recipe.get("parent", "")}"',
|
||||
f'commit_url = "{recipe.get("commit_url", "")}"',
|
||||
f'vuln = "{vuln}"',
|
||||
f'fixed = "{fixed}"',
|
||||
f'relpath = "{recipe.get("relpath", recipe.get("path", ""))}"',
|
||||
"min_recall = 1.0",
|
||||
'fix_policy = "silent"',
|
||||
"",
|
||||
"[[pair.expected]]",
|
||||
f'cwe = "{recipe.get("cwe", "")}"',
|
||||
f'class = "{recipe.get("class", "")}"',
|
||||
f'path = "{recipe.get("relpath", recipe.get("path", ""))}"',
|
||||
]
|
||||
if recipe.get("rule_id"):
|
||||
lines.append(f'rule_id = "{recipe["rule_id"]}"')
|
||||
if recipe.get("sink"):
|
||||
lines.append(f'sink = "{recipe["sink"]}"')
|
||||
if recipe.get("evidence"):
|
||||
evidence = str(recipe["evidence"]).replace('"', '\\"')
|
||||
lines.append(f'evidence = "{evidence}"')
|
||||
lines.append("")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def training_rows(recipes: list[dict[str, object]]) -> list[dict[str, object]]:
|
||||
rows: list[dict[str, object]] = []
|
||||
for recipe in recipes:
|
||||
vuln, fixed = excerpt_rel(recipe)
|
||||
base = {
|
||||
"pair": recipe["name"],
|
||||
"cve": recipe.get("cve", ""),
|
||||
"year": int(recipe.get("year", 0) or 0),
|
||||
"cwe": recipe.get("cwe", ""),
|
||||
"class": recipe.get("class", ""),
|
||||
"project": recipe.get("project", recipe.get("origin", "")),
|
||||
"repo": recipe.get("repo", ""),
|
||||
"license": recipe.get("license", ""),
|
||||
"function": recipe.get("function", ""),
|
||||
"relpath": recipe.get("relpath", recipe.get("path", "")),
|
||||
"commit_url": recipe.get("commit_url", ""),
|
||||
}
|
||||
rows.append({**base, "id": f"{recipe['name']}:vuln", "label": "vuln", "file": vuln, "commit": recipe.get("parent", "")})
|
||||
rows.append({**base, "id": f"{recipe['name']}:fixed", "label": "fixed", "file": fixed, "commit": recipe.get("commit", "")})
|
||||
return rows
|
||||
_LIB = Path(__file__).resolve().parents[1] / "catalog_gen.py"
|
||||
|
||||
|
||||
def main() -> int:
|
||||
recipes = list(tomllib.loads(RECIPES.read_text()).get("recipe", []))
|
||||
CATALOG.write_text(catalog_text(recipes), encoding="utf-8")
|
||||
TRAINING.parent.mkdir(parents=True, exist_ok=True)
|
||||
with TRAINING.open("w", encoding="utf-8") as handle:
|
||||
for row in training_rows(recipes):
|
||||
handle.write(json.dumps(row, sort_keys=True) + "\n")
|
||||
print(f"wrote {CATALOG} ({len(recipes)} pairs)")
|
||||
print(f"wrote {TRAINING} ({len(recipes) * 2} examples)")
|
||||
return 0
|
||||
return int(runpy.run_path(str(_LIB), run_name="catalog_gen_lib")["main"](["--slice", "vfc"]))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
+11
-188
@@ -1,203 +1,26 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Extract one function from public parent/fix blobs. Maintainer-only; CI stays offline.
|
||||
"""Thin shim: the shared harvest library lives in ``benchmarks/pairs/harvest.py``.
|
||||
|
||||
Does not clone mozilla-central, gecko-dev, or Chromium. Advisory-only recipes
|
||||
(no parent and commit) are rejected. FixFox is not a fetch source.
|
||||
Kept so existing maintainer commands and tests keep working:
|
||||
``python benchmarks/pairs/vfc/harvest.py --name openssl-cve-2014-0160``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import runpy
|
||||
import sys
|
||||
import tomllib
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from urllib.parse import urlparse
|
||||
from urllib.request import Request, urlopen
|
||||
|
||||
ALLOWED_HOSTS = {
|
||||
"raw.githubusercontent.com",
|
||||
"github.com",
|
||||
"phabricator.services.mozilla.com",
|
||||
"hg.mozilla.org",
|
||||
"hg-edge.mozilla.org",
|
||||
}
|
||||
BLOCKED_TOKENS = ("fixfox", "zenodo.org/record")
|
||||
|
||||
|
||||
class RecipeError(ValueError):
|
||||
"""Invalid harvest recipe."""
|
||||
|
||||
|
||||
def extract_function(source: str, function_name: str) -> str:
|
||||
"""Copy the named function through balanced braces."""
|
||||
needle = function_name
|
||||
idx = 0
|
||||
while True:
|
||||
found = source.find(needle, idx)
|
||||
if found < 0:
|
||||
raise RecipeError(f"function not found: {function_name}")
|
||||
before = source[found - 1] if found else " "
|
||||
after_name = source[found + len(needle) : found + len(needle) + 1]
|
||||
if before.isalnum() or before == "_":
|
||||
idx = found + len(needle)
|
||||
continue
|
||||
rest = source[found + len(needle) :].lstrip()
|
||||
if rest.startswith("(") or rest.startswith("<") or rest.startswith("{") or after_name in "(<{":
|
||||
brace = source.find("{", found)
|
||||
semi = source.find(";", found)
|
||||
if brace >= 0 and (semi < 0 or brace < semi):
|
||||
break
|
||||
idx = found + len(needle)
|
||||
line_start = source.rfind("\n", 0, found) + 1
|
||||
prev_nl = source.rfind("\n", 0, line_start - 1)
|
||||
prev_line = source[prev_nl + 1 : line_start]
|
||||
start = line_start
|
||||
stripped_prev = prev_line.strip()
|
||||
if (
|
||||
stripped_prev
|
||||
and not stripped_prev.startswith(("/", "*", "#", "//"))
|
||||
and "{" not in prev_line
|
||||
and ";" not in prev_line
|
||||
and not stripped_prev.startswith(("sub ", "def ", "class "))
|
||||
):
|
||||
start = prev_nl + 1
|
||||
brace = source.find("{", found)
|
||||
if brace < 0:
|
||||
raise RecipeError(f"no body for function: {function_name}")
|
||||
depth = 0
|
||||
for pos, char in enumerate(source[brace:], brace):
|
||||
if char == "{":
|
||||
depth += 1
|
||||
elif char == "}":
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
return source[start : pos + 1].strip() + "\n"
|
||||
raise RecipeError(f"unbalanced braces for function: {function_name}")
|
||||
|
||||
|
||||
def provenance_header(recipe: dict[str, Any], *, side: str, sha: str) -> str:
|
||||
lines = [
|
||||
f"Provenance: {recipe.get('repo', '')} {recipe.get('function', '')} ({side}).",
|
||||
f"repo: {recipe.get('repo', '')}",
|
||||
f"commit: {sha}",
|
||||
f"parent: {recipe.get('parent', '')}",
|
||||
f"commit_url: {recipe.get('commit_url', '')}",
|
||||
f"cve: {recipe.get('cve', '')}",
|
||||
f"license: {recipe.get('license', '')}",
|
||||
f"function: {recipe.get('function', '')}",
|
||||
f"relpath: {recipe.get('relpath', recipe.get('path', ''))}",
|
||||
]
|
||||
prefix = "//" if str(recipe.get("language", "c")).startswith("c") and str(recipe.get("path", "")).endswith((".cc", ".cpp", ".java", ".js")) else "/*"
|
||||
if prefix == "//":
|
||||
return "\n".join(f"// {line}" for line in lines) + "\n"
|
||||
return "/* " + "\n * ".join(lines) + "\n */\n"
|
||||
|
||||
|
||||
def validate_recipe(recipe: dict[str, Any]) -> None:
|
||||
name = str(recipe.get("name", ""))
|
||||
if any(token in str(recipe).lower() for token in BLOCKED_TOKENS):
|
||||
raise RecipeError(f"{name}: embargoed or blocked dataset")
|
||||
if not recipe.get("parent") or not recipe.get("commit"):
|
||||
raise RecipeError(f"{name}: advisory-only recipe rejected; parent and commit required")
|
||||
if not recipe.get("path") or not recipe.get("function"):
|
||||
raise RecipeError(f"{name}: path and function required")
|
||||
if not recipe.get("license"):
|
||||
raise RecipeError(f"{name}: license required")
|
||||
|
||||
|
||||
def github_raw_url(repo: str, sha: str, path: str) -> str:
|
||||
return f"https://raw.githubusercontent.com/{repo}/{sha}/{path}"
|
||||
|
||||
|
||||
def excerpt_paths(recipe: dict[str, Any], root: Path) -> tuple[Path, Path]:
|
||||
name = str(recipe["name"])
|
||||
ext = Path(str(recipe.get("path", "a.c"))).suffix or ".c"
|
||||
folder = root / name
|
||||
return folder / f"vuln{ext}", folder / f"fixed{ext}"
|
||||
|
||||
|
||||
def write_excerpt(path: Path, recipe: dict[str, Any], *, side: str, sha: str, body: str) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
header = provenance_header(recipe, side=side, sha=sha)
|
||||
path.write_text(header + "\n" + body, encoding="utf-8")
|
||||
|
||||
|
||||
def harvest_recipe(recipe: dict[str, Any], root: Path) -> tuple[Path, Path]:
|
||||
validate_recipe(recipe)
|
||||
if str(recipe.get("host", "github")) != "github":
|
||||
raise RecipeError(f"{recipe.get('name')}: live fetch currently implemented for host=github only")
|
||||
parent_src = fetch_url(github_raw_url(str(recipe["repo"]), str(recipe["parent"]), str(recipe["path"])))
|
||||
fix_src = fetch_url(github_raw_url(str(recipe["repo"]), str(recipe["commit"]), str(recipe["path"])))
|
||||
function = str(recipe["function"])
|
||||
parent_fn = extract_function(parent_src, function)
|
||||
fix_fn = extract_function(fix_src, function)
|
||||
vuln_path, fixed_path = excerpt_paths(recipe, root)
|
||||
write_excerpt(vuln_path, recipe, side="vuln", sha=str(recipe["parent"]), body=parent_fn)
|
||||
write_excerpt(fixed_path, recipe, side="fixed", sha=str(recipe["commit"]), body=fix_fn)
|
||||
return vuln_path, fixed_path
|
||||
|
||||
|
||||
def fetch_url(url: str, timeout: int = 60) -> str:
|
||||
host = urlparse(url).hostname or ""
|
||||
if host not in ALLOWED_HOSTS:
|
||||
raise RecipeError(f"host not allowed: {host}")
|
||||
request = Request(url, headers={"User-Agent": "OpenUltraSAST-vfc-harvest"})
|
||||
with urlopen(request, timeout=timeout) as response: # noqa: S310 — host allowlisted
|
||||
return response.read().decode("utf-8", "replace")
|
||||
|
||||
|
||||
def load_recipes(path: Path) -> tuple[dict[str, Any], ...]:
|
||||
payload = tomllib.loads(path.read_text())
|
||||
return tuple(dict(item) for item in payload.get("recipe", []))
|
||||
_LIB = Path(__file__).resolve().parents[1] / "harvest.py"
|
||||
_NS = runpy.run_path(str(_LIB), run_name="pair_harvest_lib")
|
||||
globals().update({key: value for key, value in _NS.items() if not key.startswith("__") and key != "main"})
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--recipes", type=Path, default=Path(__file__).with_name("recipes.toml"))
|
||||
parser.add_argument("--name", help="recipe name to harvest (omit with --all)")
|
||||
parser.add_argument("--all", action="store_true", help="process every recipe")
|
||||
parser.add_argument("--fetch", action="store_true", help="HTTP-fetch and write excerpts; default is dry-run validate")
|
||||
parser.add_argument("--force", action="store_true", help="overwrite existing excerpts")
|
||||
parser.add_argument("--out", type=Path, default=Path(__file__).resolve().parent, help="excerpt root")
|
||||
args = parser.parse_args(argv)
|
||||
loaded = load_recipes(args.recipes)
|
||||
recipes = {str(item["name"]): item for item in loaded}
|
||||
selected: list[dict[str, Any]]
|
||||
if args.all:
|
||||
selected = list(loaded)
|
||||
elif args.name:
|
||||
if args.name not in recipes:
|
||||
print(f"unknown recipe: {args.name}", file=sys.stderr)
|
||||
return 2
|
||||
selected = [recipes[args.name]]
|
||||
else:
|
||||
print("pass --name RECIPE or --all", file=sys.stderr)
|
||||
return 2
|
||||
failures = 0
|
||||
for recipe in selected:
|
||||
name = str(recipe["name"])
|
||||
try:
|
||||
validate_recipe(recipe)
|
||||
except RecipeError as exc:
|
||||
print(str(exc), file=sys.stderr)
|
||||
failures += 1
|
||||
continue
|
||||
if not args.fetch:
|
||||
print(f"ok {name} parent={recipe['parent']} commit={recipe['commit']}")
|
||||
continue
|
||||
vuln_path, fixed_path = excerpt_paths(recipe, args.out)
|
||||
if vuln_path.is_file() and fixed_path.is_file() and not args.force:
|
||||
print(f"skip {name} (exists)")
|
||||
continue
|
||||
try:
|
||||
harvest_recipe(recipe, args.out)
|
||||
except RecipeError as exc:
|
||||
print(f"{name}: {exc}", file=sys.stderr)
|
||||
failures += 1
|
||||
continue
|
||||
print(f"wrote {name} -> {vuln_path.parent}")
|
||||
return 1 if failures else 0
|
||||
args = list(sys.argv[1:] if argv is None else argv)
|
||||
if "--slice" not in args:
|
||||
args = ["--slice", "vfc", *args]
|
||||
return int(_NS["main"](args))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user