mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-16 03:49:26 +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__":
|
||||
|
||||
@@ -0,0 +1,151 @@
|
||||
"""Named function ranges shared by the scorer, the overlay, and the catalog generator.
|
||||
|
||||
Tree-sitter names an arrow function after its first bare parameter and gives
|
||||
object or class methods, `exports.x = function`, and IIFEs ``<anon>``. Labels in
|
||||
the pair catalogs name the declarator (``const h = ...``, ``exports.h = ...``,
|
||||
``h(req, res) {``). Every consumer must therefore name ranges the same way, or
|
||||
function-scoped matching silently fails (pair-corpus-honesty Req 1.2, design
|
||||
§Scorer). This module is the single source of that naming. Stdlib plus ``.ir``
|
||||
only; never imports ``overlay`` or ``pairs``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from collections.abc import Sequence
|
||||
|
||||
from .ir import FileIR, parse_file
|
||||
|
||||
FunctionRange = tuple[str, int, int]
|
||||
MODULE_FUNCTION = "<module>"
|
||||
ANON_FUNCTION = "<anon>"
|
||||
UNNAMED_FUNCTIONS = frozenset({MODULE_FUNCTION, ANON_FUNCTION})
|
||||
# Keywords a parser or a regex fallback can mistake for a function name; never valid labels.
|
||||
RESERVED_FUNCTION_LABELS = frozenset(
|
||||
{
|
||||
"function",
|
||||
"def",
|
||||
"return",
|
||||
"if",
|
||||
"else",
|
||||
"for",
|
||||
"while",
|
||||
"switch",
|
||||
"catch",
|
||||
"class",
|
||||
"async",
|
||||
"await",
|
||||
"lambda",
|
||||
"const",
|
||||
"let",
|
||||
"var",
|
||||
"export",
|
||||
"import",
|
||||
"module",
|
||||
"exports",
|
||||
"new",
|
||||
"this",
|
||||
"self",
|
||||
"anonymous",
|
||||
ANON_FUNCTION,
|
||||
}
|
||||
)
|
||||
|
||||
_IDENT = r"[A-Za-z_$][\w$]*"
|
||||
_ARROW_TAIL = r"(?:async\s*)?(?:\([^)]*\)|" + _IDENT + r")\s*=>"
|
||||
_DECL = re.compile(
|
||||
# def name( / function name( / export async function name(
|
||||
r"^\s*(?:export\s+)?(?:async\s+)?(?:def|function|sub)\s+(" + _IDENT + r")\s*[\({]|"
|
||||
# const name = [a.b.c =]* [async] (args) => | ident =>
|
||||
r"^\s*(?:export\s+)?(?:const|let|var)\s+(" + _IDENT + r")\s*=\s*(?:" + _IDENT + r"(?:\." + _IDENT + r")*\s*=\s*)*" + _ARROW_TAIL + r"|"
|
||||
# exports.name = / module.exports.name = [async] function | (args) =>
|
||||
r"^\s*(?:module\.)?exports\.(" + _IDENT + r")\s*=\s*(?:async\s*)?(?:function\b|" + _ARROW_TAIL + r")|"
|
||||
# object property: name: [async] function | (args) =>
|
||||
r"^\s*(?:async\s+)?(" + _IDENT + r")\s*:\s*(?:async\s*)?(?:function\b|" + _ARROW_TAIL + r")|"
|
||||
# C / Java / JS method or shorthand: [type] name(args) { (no call: the arg list carries no callback or literal)
|
||||
r"^\s*(?:[A-Za-z_][\w:<>\*\s,]*[\s\*])?(" + r"[A-Za-z_][\w:$]*" + r")\s*\(([^;{]*)\)\s*(?:const\s*)?\{?\s*$"
|
||||
)
|
||||
_CALLBACK_ARGS = re.compile(r"\bfunction\b|=>|['\"`]|^\s*\d")
|
||||
# A statement keyword before `name(` means the line calls or controls, it does not declare.
|
||||
_STATEMENT_PREFIX = re.compile(
|
||||
r"^\s*(?:return|print|yield|throw|await|new|else|case|delete|typeof|synchronized|with|elif|except|try|if|for|while|switch|catch|do)\b"
|
||||
)
|
||||
# Only these grammars name an arrow function after its bare parameter; elsewhere the parser name is trustworthy.
|
||||
_PARAM_NAMED_ARROWS = frozenset({"javascript", "typescript"})
|
||||
|
||||
|
||||
def declarator_name(line: str) -> str | None:
|
||||
"""Function name declared on ``line``, or None when the line declares nothing (a call is not a declaration)."""
|
||||
match = _DECL.match(line)
|
||||
if not match:
|
||||
return None
|
||||
groups = match.groups()
|
||||
name = next((group for group in groups[:5] if group), None)
|
||||
if name is None:
|
||||
return None
|
||||
if groups[4] is not None:
|
||||
args = groups[5] or ""
|
||||
if _CALLBACK_ARGS.search(args):
|
||||
return None # `exec('...', function (err) {` is a call, not a declaration
|
||||
if _STATEMENT_PREFIX.match(line) or name in {"if", "for", "while", "switch", "catch", "return", "elif", "with", "synchronized"}:
|
||||
return None
|
||||
name = name.rsplit("::", 1)[-1] # `bool K::m(` declares `m`; catalog labels are unqualified
|
||||
return name if name not in RESERVED_FUNCTION_LABELS else None
|
||||
|
||||
|
||||
def named_ranges_from_ir(ir: FileIR, text: str) -> tuple[FunctionRange, ...]:
|
||||
"""Declarator-named function ranges from a parsed file; anonymous functions keep ``<anon>`` only when nothing names them."""
|
||||
lines = text.splitlines()
|
||||
ranges: list[FunctionRange] = []
|
||||
for fn in ir.functions:
|
||||
if fn.name == MODULE_FUNCTION:
|
||||
continue
|
||||
line = lines[fn.start_line - 1] if 0 < fn.start_line <= len(lines) else ""
|
||||
declared = declarator_name(line)
|
||||
name = declared or fn.name
|
||||
if name in RESERVED_FUNCTION_LABELS and name != ANON_FUNCTION:
|
||||
name = ANON_FUNCTION
|
||||
elif declared is None and name != ANON_FUNCTION and ir.language in _PARAM_NAMED_ARROWS and not _declares(line, name):
|
||||
name = ANON_FUNCTION # tree-sitter named a bare-parameter arrow (`block => {`) after its parameter
|
||||
ranges.append((name, fn.start_line, fn.end_line))
|
||||
return tuple(ranges)
|
||||
|
||||
|
||||
def _declares(line: str, name: str) -> bool:
|
||||
"""True when ``line`` visibly declares ``name`` as a function (keyword form or ``name(``), not as a parameter."""
|
||||
escaped = re.escape(name)
|
||||
return bool(re.search(r"\b(?:function|def|sub)\s+" + escaped + r"\b|(?<![\w$.])" + escaped + r"\s*\(", line))
|
||||
|
||||
|
||||
def named_function_ranges(path: str, text: str, language: str) -> tuple[FunctionRange, ...] | None:
|
||||
"""Ranges for one file. None when the parser rejected the file; regex line scan only when no grammar exists."""
|
||||
ir = parse_file(path, text, language)
|
||||
if ir.parse_ok:
|
||||
return named_ranges_from_ir(ir, text)
|
||||
if ir.reason == "parse_failed":
|
||||
return None
|
||||
lines = text.splitlines()
|
||||
fallback: list[FunctionRange] = []
|
||||
for index, line in enumerate(lines, start=1):
|
||||
name = declarator_name(line)
|
||||
if name:
|
||||
fallback.append((name, index, len(lines)))
|
||||
return tuple(fallback)
|
||||
|
||||
|
||||
def function_at(ranges: Sequence[FunctionRange], line: int | None) -> str | None:
|
||||
"""Innermost *named* function whose range contains ``line``; None when unknown or only anonymous."""
|
||||
if line is None:
|
||||
return None
|
||||
best: FunctionRange | None = None
|
||||
for item in ranges:
|
||||
name, start, end = item
|
||||
if name in UNNAMED_FUNCTIONS or not (start <= line <= end):
|
||||
continue
|
||||
if best is None or (end - start) < (best[2] - best[1]):
|
||||
best = item
|
||||
return best[0] if best is not None else None
|
||||
|
||||
|
||||
def spans_named(ranges: Sequence[FunctionRange], function: str) -> tuple[tuple[int, int], ...]:
|
||||
return tuple((start, end) for name, start, end in ranges if name == function)
|
||||
@@ -0,0 +1,365 @@
|
||||
"""Maintainer VFC harvest stays offline in CI: extract and reject, no HTTP."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import runpy
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
HARVEST = Path("benchmarks/pairs/vfc/harvest.py")
|
||||
|
||||
|
||||
def _harvest() -> dict[str, object]:
|
||||
return runpy.run_path(str(HARVEST))
|
||||
|
||||
|
||||
def test_extract_function_round_trips_perl_sub_without_parens() -> None:
|
||||
harvest = _harvest()
|
||||
source = """
|
||||
sub other { return 0; }
|
||||
sub link_hash_cert {
|
||||
my ($hash, $fprint) = `"$openssl" x509 -in "$fname"`;
|
||||
}
|
||||
sub later { return 1; }
|
||||
"""
|
||||
body = harvest["extract_function"](source, "link_hash_cert")
|
||||
assert "sub link_hash_cert {" in body
|
||||
assert '`"$openssl" x509' in body
|
||||
assert "sub other" not in body
|
||||
assert "sub later" not in body
|
||||
|
||||
|
||||
def test_extract_function_round_trips_a_local_body() -> None:
|
||||
harvest = _harvest()
|
||||
source = """
|
||||
int before(void) { return 0; }
|
||||
|
||||
int target_fn(char *name, int namelen) {
|
||||
memcpy(name, name, namelen + 1);
|
||||
return namelen;
|
||||
}
|
||||
|
||||
int after(void) { return 1; }
|
||||
"""
|
||||
body = harvest["extract_function"](source, "target_fn")
|
||||
assert "int target_fn(char *name, int namelen)" in body
|
||||
assert "memcpy(name, name, namelen + 1);" in body
|
||||
assert "int before" not in body
|
||||
assert "int after" not in body
|
||||
|
||||
|
||||
def test_extract_function_skips_prototype_before_definition() -> None:
|
||||
harvest = _harvest()
|
||||
source = """
|
||||
class Loader {
|
||||
public:
|
||||
char *ArrayBufferResult();
|
||||
};
|
||||
|
||||
char *Loader::ArrayBufferResult() {
|
||||
return raw_data_->ToArrayBuffer();
|
||||
}
|
||||
"""
|
||||
body = harvest["extract_function"](source, "ArrayBufferResult")
|
||||
assert "Loader::ArrayBufferResult" in body
|
||||
assert "return raw_data_->ToArrayBuffer();" in body
|
||||
assert "class Loader" not in body
|
||||
|
||||
|
||||
def test_validate_recipe_rejects_advisory_only_and_fixfox() -> None:
|
||||
harvest = _harvest()
|
||||
validate = harvest["validate_recipe"]
|
||||
with pytest.raises(harvest["RecipeError"], match="advisory-only"):
|
||||
validate({"name": "mfsa-only", "cve": "CVE-2019-11730"})
|
||||
with pytest.raises(harvest["RecipeError"], match="embargoed"):
|
||||
validate(
|
||||
{
|
||||
"name": "fixfox",
|
||||
"parent": "abc",
|
||||
"commit": "def",
|
||||
"path": "a.c",
|
||||
"function": "f",
|
||||
"license": "MIT",
|
||||
"url": "https://zenodo.org/record/fixfox",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def test_harvest_dry_run_accepts_seed_recipes() -> None:
|
||||
harvest = _harvest()
|
||||
recipes = {str(item["name"]): item for item in harvest["load_recipes"](Path("benchmarks/pairs/vfc/recipes.toml"))}
|
||||
assert {"openssl-cve-2014-0160", "firefox-cve-2020-15667", "chromium-cve-2019-5786"} <= set(recipes)
|
||||
for recipe in recipes.values():
|
||||
harvest["validate_recipe"](recipe)
|
||||
assert harvest["main"](["--name", "openssl-cve-2014-0160"]) == 0
|
||||
|
||||
|
||||
# --- pair-corpus-honesty: shared library, comment-blind anchor, modes ---------
|
||||
|
||||
LIB = Path("benchmarks/pairs/harvest.py")
|
||||
|
||||
|
||||
def _lib() -> dict[str, object]:
|
||||
return runpy.run_path(str(LIB), run_name="pair_harvest_lib")
|
||||
|
||||
|
||||
def test_name_mode_ignores_doc_comment_mention_and_anchors_on_declarator() -> None:
|
||||
lib = _lib()
|
||||
source = """
|
||||
/*
|
||||
* Curl_follow() handles the URL redirect magic. Pass in the 'newurl' string
|
||||
* as given by the remote server and set up the new URL to request.
|
||||
*/
|
||||
CURLcode Curl_follow(struct Curl_easy *data,
|
||||
char *newurl,
|
||||
followtype type)
|
||||
{
|
||||
const char *msg = "Curl_follow(";
|
||||
return CURLE_OK;
|
||||
}
|
||||
"""
|
||||
body = lib["extract_function"](source, "Curl_follow", "c")
|
||||
assert body.startswith("CURLcode Curl_follow(struct Curl_easy *data,")
|
||||
assert "redirect magic" not in body
|
||||
assert 'const char *msg = "Curl_follow(";' in body
|
||||
|
||||
|
||||
def test_name_mode_skips_prototype_and_string_mentions() -> None:
|
||||
lib = _lib()
|
||||
source = 'int f(void);\nconst char *s = "f(";\nint f(void)\n{\n return 0;\n}\n'
|
||||
body = lib["extract_function"](source, "f", "c")
|
||||
assert body == "int f(void)\n{\n return 0;\n}\n"
|
||||
|
||||
|
||||
def test_python_block_mode_cuts_by_indentation_and_keeps_decorators() -> None:
|
||||
lib = _lib()
|
||||
source = (
|
||||
"import os\n\n"
|
||||
"@router.post('/runs')\n"
|
||||
"async def create_run(body):\n"
|
||||
" user_id = body.user_id\n"
|
||||
" if not user_id:\n"
|
||||
" return None\n"
|
||||
" return user_id\n\n"
|
||||
"def other():\n"
|
||||
" return 1\n"
|
||||
)
|
||||
body = lib["extract_function"](source, "create_run", "python")
|
||||
assert body.startswith("@router.post('/runs')\nasync def create_run(body):")
|
||||
assert "return user_id" in body
|
||||
assert "def other" not in body
|
||||
|
||||
|
||||
def test_line_range_mode_round_trips_exact_lines() -> None:
|
||||
lib = _lib()
|
||||
source = "a\nb\nc\nd\n"
|
||||
assert lib["extract_line_range"](source, 2, 3) == "b\nc\n"
|
||||
with pytest.raises(lib["RecipeError"]):
|
||||
lib["extract_line_range"](source, 3, 9)
|
||||
|
||||
|
||||
def test_hunk_mode_returns_enclosing_functions_on_both_sides() -> None:
|
||||
lib = _lib()
|
||||
parent = (
|
||||
"int a(void) { return 0; }\n\nint target(int n)\n{\n char buf[8];\n memcpy(buf, src, n);\n"
|
||||
" return n;\n}\n\nint z(void) { return 1; }\n"
|
||||
)
|
||||
fixed = parent.replace(" memcpy(buf, src, n);", " if (n > 8)\n return -1;\n memcpy(buf, src, n);")
|
||||
vuln_fn, fixed_fn = lib["extract_hunk"](parent, fixed, "c")
|
||||
assert vuln_fn.startswith("int target(int n)")
|
||||
assert "if (n > 8)" not in vuln_fn and "if (n > 8)" in fixed_fn
|
||||
assert "int a(void)" not in fixed_fn and "int z(void)" not in fixed_fn
|
||||
|
||||
|
||||
def test_hunk_mode_python() -> None:
|
||||
lib = _lib()
|
||||
parent = "def a():\n return 0\n\ndef create_run(body):\n user_id = body.user_id\n return user_id\n\ndef z():\n return 1\n"
|
||||
fixed = parent.replace(" user_id = body.user_id", " user_id = request.state.user_id")
|
||||
vuln_fn, fixed_fn = lib["extract_hunk"](parent, fixed, "python")
|
||||
assert vuln_fn.startswith("def create_run(body):") and "body.user_id" in vuln_fn
|
||||
assert "request.state.user_id" in fixed_fn and "def z" not in fixed_fn
|
||||
|
||||
|
||||
def test_extract_pair_dispatches_modes_and_validate_requires_mode_fields() -> None:
|
||||
lib = _lib()
|
||||
base = {"name": "x", "parent": "p", "commit": "c", "path": "a.py", "license": "MIT", "language": "python"}
|
||||
with pytest.raises(lib["RecipeError"], match="line_start"):
|
||||
lib["validate_recipe"]({**base, "mode": "line_range"})
|
||||
with pytest.raises(lib["RecipeError"], match="unknown mode"):
|
||||
lib["validate_recipe"]({**base, "mode": "magic"})
|
||||
lib["validate_recipe"]({**base, "mode": "hunk"})
|
||||
src = "def f():\n return 1\n"
|
||||
vuln, fixed = lib["extract_pair"]({**base, "mode": "line_range", "line_start": 1, "line_end": 2}, src, src)
|
||||
assert vuln == src and fixed == src
|
||||
|
||||
|
||||
def test_provenance_header_uses_language_comment_style() -> None:
|
||||
lib = _lib()
|
||||
py = lib["provenance_header"]({"path": "a.py", "repo": "r", "mechanism": "identity_from_request_body"}, side="vuln", sha="abc")
|
||||
assert py.startswith("# Provenance:") and "# mechanism: identity_from_request_body" in py
|
||||
ts = lib["provenance_header"]({"path": "a.ts", "repo": "r"}, side="vuln", sha="abc")
|
||||
assert ts.startswith("// Provenance:")
|
||||
c = lib["provenance_header"]({"path": "a.c", "repo": "r"}, side="vuln", sha="abc")
|
||||
assert c.startswith("/* Provenance:")
|
||||
|
||||
|
||||
def test_hunk_mode_skips_module_level_hunk_and_uses_next_function_hunk() -> None:
|
||||
lib = _lib()
|
||||
parent = "const x = 1;\n\nfunction f(a) {\n return a;\n}\n"
|
||||
fixed = "const x = 2;\n\nfunction f(a) {\n if (!a) return null;\n return a;\n}\n"
|
||||
vuln_fn, fixed_fn = lib["extract_hunk"](parent, fixed, "javascript")
|
||||
assert vuln_fn.startswith("function f(a)") and "if (!a)" in fixed_fn
|
||||
|
||||
|
||||
def test_hunk_mode_falls_back_to_whole_small_file() -> None:
|
||||
lib = _lib()
|
||||
parent = "module.exports = { cors: '*' };\n"
|
||||
fixed = "module.exports = { cors: 'https://example.test' };\n"
|
||||
vuln_fn, fixed_fn = lib["extract_hunk"](parent, fixed, "javascript")
|
||||
assert vuln_fn == parent and fixed_fn == fixed
|
||||
big = "\n".join(f"const v{i} = {i};" for i in range(200)) + "\n"
|
||||
with pytest.raises(lib["RecipeError"], match="too large"):
|
||||
lib["extract_hunk"](big, big.replace("v0 = 0", "v0 = 1"), "javascript")
|
||||
|
||||
|
||||
def test_catalog_gen_derives_function_past_the_provenance_header_and_never_anon(tmp_path: Path) -> None:
|
||||
gen = runpy.run_path(str(Path("benchmarks/pairs/catalog_gen.py")), run_name="catalog_gen_lib")
|
||||
slice_root = tmp_path / "s"
|
||||
pair = slice_root / "p"
|
||||
pair.mkdir(parents=True)
|
||||
header_v = "// Provenance: r go (vuln).\n// repo: r\n// commit: a\n// license: MIT\n\n"
|
||||
header_f = "// Provenance: r go (fixed).\n// repo: r\n// commit: b\n// license: MIT\n\n"
|
||||
body_v = "function first(a) {\n return a;\n}\n\nfunction go(cmd) {\n return exec(cmd);\n}\n"
|
||||
body_f = "function first(a) {\n return a;\n}\n\nfunction go(cmd) {\n if (!ok(cmd)) return null;\n return exec(cmd);\n}\n"
|
||||
(pair / "vuln.js").write_text(header_v + body_v)
|
||||
(pair / "fixed.js").write_text(header_f + body_f)
|
||||
recipe = {"name": "p", "path": "a.js", "language": "javascript"}
|
||||
assert gen["derive_function"](slice_root, recipe) == "go"
|
||||
# a change inside an anonymous callback resolves to the enclosing named function, never "<anon>"
|
||||
body_v2 = "function handler(req) {\n items.forEach(function (i) {\n exec(i);\n });\n}\n"
|
||||
body_f2 = "function handler(req) {\n items.forEach(function (i) {\n if (safe(i)) exec(i);\n });\n}\n"
|
||||
(pair / "vuln.js").write_text(header_v + body_v2)
|
||||
(pair / "fixed.js").write_text(header_f + body_f2)
|
||||
assert gen["derive_function"](slice_root, recipe) == "handler"
|
||||
|
||||
|
||||
def test_write_excerpt_redacts_credential_literals_but_not_code() -> None:
|
||||
lib = _lib()
|
||||
redact = lib["redact"]
|
||||
google = "AIza" + "SyDEMOKEYDEMOKEYDEMOKEYDEMOKEY0000" # assembled at runtime, never a contiguous key literal
|
||||
assert redact(f"apiKey: '{google}',") == "apiKey: '***REDACTED***',"
|
||||
assert redact('password = "hunter2hunter2"') == 'password = "***REDACTED***"'
|
||||
assert redact("password = request.form['password']") == "password = request.form['password']"
|
||||
assert redact("const token = getToken();") == "const token = getToken();"
|
||||
|
||||
|
||||
def test_catalog_gen_prefers_declarator_names_and_rejects_reserved_words(tmp_path: Path) -> None:
|
||||
gen = runpy.run_path(str(Path("benchmarks/pairs/catalog_gen.py")), run_name="catalog_gen_lib")
|
||||
slice_root = tmp_path / "s"
|
||||
pair = slice_root / "p"
|
||||
pair.mkdir(parents=True)
|
||||
header_v = "// Provenance: r x (vuln).\n// license: MIT\n\n"
|
||||
header_f = "// Provenance: r x (fixed).\n// license: MIT\n\n"
|
||||
# An arrow function assigned to a const: the label must be the const name, never the parameter name.
|
||||
body_v = "const listProcessesOnPort = (port) => {\n return exec('lsof -i :' + port);\n};\n"
|
||||
body_f = (
|
||||
"const listProcessesOnPort = (port) => {\n if (!/^\\d+$/.test(port)) throw new Error('bad');\n"
|
||||
" return exec('lsof -i :' + port);\n};\n"
|
||||
)
|
||||
(pair / "vuln.js").write_text(header_v + body_v)
|
||||
(pair / "fixed.js").write_text(header_f + body_f)
|
||||
recipe = {"name": "p", "path": "a.js", "language": "javascript"}
|
||||
assert gen["derive_function"](slice_root, recipe) == "listProcessesOnPort"
|
||||
# A bare anonymous function expression must not yield the keyword "function".
|
||||
(pair / "vuln.js").write_text(header_v + "module.exports = function (req, res) {\n exec(req.query.c);\n};\n")
|
||||
(pair / "fixed.js").write_text(header_f + "module.exports = function (req, res) {\n res.end();\n};\n")
|
||||
assert gen["derive_function"](slice_root, recipe) is None
|
||||
|
||||
|
||||
def test_catalog_gen_labels_from_sink_location_not_from_formatting_hunk(tmp_path: Path) -> None:
|
||||
gen = runpy.run_path(str(Path("benchmarks/pairs/catalog_gen.py")), run_name="catalog_gen_lib")
|
||||
slice_root = tmp_path / "s"
|
||||
pair = slice_root / "p"
|
||||
pair.mkdir(parents=True)
|
||||
# Excerpt starts at upstream line 10; the sink sits at upstream line 17 (excerpt body line 8) inside performAction.
|
||||
header_v = "// Provenance: r x (vuln).\n// license: MIT\n// upstream_start: 10\n\n"
|
||||
header_f = "// Provenance: r x (fixed).\n// license: MIT\n// upstream_start: 10\n\n"
|
||||
body_v = (
|
||||
"function trace() { }\n\nfunction other(a) {\n return a;\n}\n\nfunction performAction(yytext) {\n"
|
||||
" return eval('[' + yytext + ']');\n}\n"
|
||||
)
|
||||
body_f = (
|
||||
"function trace () { }\n\nfunction other(a) {\n return a;\n}\n\nfunction performAction(yytext) {\n"
|
||||
" return JSON.parse('[' + yytext + ']');\n}\n"
|
||||
)
|
||||
(pair / "vuln.js").write_text(header_v + body_v)
|
||||
(pair / "fixed.js").write_text(header_f + body_f)
|
||||
recipe = {"name": "p", "path": "a.js", "relpath": "a.js", "language": "javascript", "sink_location": "a.js:17:10"}
|
||||
assert gen["derive_function"](slice_root, recipe) == "performAction"
|
||||
assert recipe.get("sink") == "eval"
|
||||
# A callback callee is never a declarator and never a label: the sink inside `exec(..., function(err){` belongs to `ps`.
|
||||
body_v2 = "function ps(pid) {\n exec('ps -p ' + pid, function (err, out) {\n return out;\n });\n}\n"
|
||||
body_f2 = (
|
||||
"function ps(pid) {\n if (!/^\\d+$/.test(pid)) return null;\n exec('ps -p ' + pid, function (err, out) {\n"
|
||||
" return out;\n });\n}\n"
|
||||
)
|
||||
(pair / "vuln.js").write_text(header_v + body_v2)
|
||||
(pair / "fixed.js").write_text(header_f + body_f2)
|
||||
recipe2 = {"name": "p", "path": "a.js", "relpath": "a.js", "language": "javascript", "sink_location": "a.js:11:3"}
|
||||
assert gen["derive_function"](slice_root, recipe2) == "ps"
|
||||
assert recipe2.get("sink") == "exec"
|
||||
|
||||
|
||||
def test_extract_pair_with_lines_and_header_record_upstream_start() -> None:
|
||||
lib = _lib()
|
||||
parent = "int a(void) { return 0; }\n\nint target(int n)\n{\n char buf[8];\n memcpy(buf, src, n);\n return n;\n}\n"
|
||||
fixed = parent.replace(" memcpy(buf, src, n);", " if (n > 8)\n return -1;\n memcpy(buf, src, n);")
|
||||
(vuln_text, vuln_start), (fix_text, fix_start) = lib["extract_pair_with_lines"]({"mode": "hunk", "language": "c"}, parent, fixed)
|
||||
assert vuln_text.startswith("int target(int n)") and vuln_start == 3 and fix_start == 3
|
||||
header = lib["provenance_header"]({"path": "a.js", "repo": "r"}, side="vuln", sha="abc", upstream_start=42)
|
||||
assert "// upstream_start: 42" in header
|
||||
|
||||
|
||||
def test_python_block_survives_dedented_triple_quoted_string() -> None:
|
||||
lib = _lib()
|
||||
source = (
|
||||
"import x\n\n"
|
||||
"async def put_user(user):\n"
|
||||
" user.password = h(user.password)\n"
|
||||
" query = f'''\n"
|
||||
"UPDATE users SET password = '{user.password}'\n"
|
||||
"WHERE name = '{user.name}'\n"
|
||||
"'''\n"
|
||||
" return await db.execute(query)\n\n"
|
||||
"def other():\n return 1\n"
|
||||
)
|
||||
body = lib["extract_function"](source, "put_user", "python")
|
||||
assert body.endswith(" return await db.execute(query)\n")
|
||||
assert "def other" not in body
|
||||
import ast
|
||||
|
||||
ast.parse(body)
|
||||
|
||||
|
||||
def test_catalog_gen_fails_loud_without_the_package_and_refuses_to_prune_an_empty_catalog(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
import sys
|
||||
|
||||
gen = runpy.run_path(str(Path("benchmarks/pairs/catalog_gen.py")), run_name="catalog_gen_lib")
|
||||
slice_root = tmp_path / "s"
|
||||
pair = slice_root / "p"
|
||||
pair.mkdir(parents=True)
|
||||
(pair / "vuln.js").write_text("// Provenance: r x (vuln).\n// license: MIT\n\nconst x = 1;\n")
|
||||
(pair / "fixed.js").write_text("// Provenance: r x (fixed).\n// license: MIT\n\nconst x = 2;\n")
|
||||
(slice_root / "recipes.toml").write_text('[[recipe]]\nname = "p"\npath = "a.js"\nlanguage = "javascript"\nmechanism = "other"\n')
|
||||
gen["generate"].__globals__["ROOT"] = tmp_path # runpy returns a copy; patch the live module globals
|
||||
# No recipe can derive a label -> the generator must refuse to prune rather than delete the corpus.
|
||||
with pytest.raises(SystemExit, match="refusing"):
|
||||
gen["generate"]("s", prune=True)
|
||||
assert pair.is_dir()
|
||||
# The package import must fail loud, never degrade into "no named function".
|
||||
monkeypatch.setitem(sys.modules, "openultrasast.semantic.functions", None)
|
||||
with pytest.raises((SystemExit, ImportError)):
|
||||
runpy.run_path(str(Path("benchmarks/pairs/catalog_gen.py")), run_name="catalog_gen_lib_blocked")
|
||||
@@ -1,95 +0,0 @@
|
||||
"""Maintainer VFC harvest stays offline in CI: extract and reject, no HTTP."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import runpy
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
HARVEST = Path("benchmarks/pairs/vfc/harvest.py")
|
||||
|
||||
|
||||
def _harvest() -> dict[str, object]:
|
||||
return runpy.run_path(str(HARVEST))
|
||||
|
||||
|
||||
def test_extract_function_round_trips_perl_sub_without_parens() -> None:
|
||||
harvest = _harvest()
|
||||
source = """
|
||||
sub other { return 0; }
|
||||
sub link_hash_cert {
|
||||
my ($hash, $fprint) = `"$openssl" x509 -in "$fname"`;
|
||||
}
|
||||
sub later { return 1; }
|
||||
"""
|
||||
body = harvest["extract_function"](source, "link_hash_cert")
|
||||
assert "sub link_hash_cert {" in body
|
||||
assert "`\"$openssl\" x509" in body
|
||||
assert "sub other" not in body
|
||||
assert "sub later" not in body
|
||||
|
||||
|
||||
def test_extract_function_round_trips_a_local_body() -> None:
|
||||
harvest = _harvest()
|
||||
source = """
|
||||
int before(void) { return 0; }
|
||||
|
||||
int target_fn(char *name, int namelen) {
|
||||
memcpy(name, name, namelen + 1);
|
||||
return namelen;
|
||||
}
|
||||
|
||||
int after(void) { return 1; }
|
||||
"""
|
||||
body = harvest["extract_function"](source, "target_fn")
|
||||
assert "int target_fn(char *name, int namelen)" in body
|
||||
assert "memcpy(name, name, namelen + 1);" in body
|
||||
assert "int before" not in body
|
||||
assert "int after" not in body
|
||||
|
||||
|
||||
def test_extract_function_skips_prototype_before_definition() -> None:
|
||||
harvest = _harvest()
|
||||
source = """
|
||||
class Loader {
|
||||
public:
|
||||
char *ArrayBufferResult();
|
||||
};
|
||||
|
||||
char *Loader::ArrayBufferResult() {
|
||||
return raw_data_->ToArrayBuffer();
|
||||
}
|
||||
"""
|
||||
body = harvest["extract_function"](source, "ArrayBufferResult")
|
||||
assert "Loader::ArrayBufferResult" in body
|
||||
assert "return raw_data_->ToArrayBuffer();" in body
|
||||
assert "class Loader" not in body
|
||||
|
||||
|
||||
def test_validate_recipe_rejects_advisory_only_and_fixfox() -> None:
|
||||
harvest = _harvest()
|
||||
validate = harvest["validate_recipe"]
|
||||
with pytest.raises(harvest["RecipeError"], match="advisory-only"):
|
||||
validate({"name": "mfsa-only", "cve": "CVE-2019-11730"})
|
||||
with pytest.raises(harvest["RecipeError"], match="embargoed"):
|
||||
validate(
|
||||
{
|
||||
"name": "fixfox",
|
||||
"parent": "abc",
|
||||
"commit": "def",
|
||||
"path": "a.c",
|
||||
"function": "f",
|
||||
"license": "MIT",
|
||||
"url": "https://zenodo.org/record/fixfox",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def test_harvest_dry_run_accepts_seed_recipes() -> None:
|
||||
harvest = _harvest()
|
||||
recipes = {str(item["name"]): item for item in harvest["load_recipes"](Path("benchmarks/pairs/vfc/recipes.toml"))}
|
||||
assert {"openssl-cve-2014-0160", "firefox-cve-2020-15667", "chromium-cve-2019-5786"} <= set(recipes)
|
||||
for recipe in recipes.values():
|
||||
harvest["validate_recipe"](recipe)
|
||||
assert harvest["main"](["--name", "openssl-cve-2014-0160"]) == 0
|
||||
Reference in New Issue
Block a user