From 071dea4712b78d6112c6519c52cb317d12c1c7b8 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sat, 5 Sep 2026 09:26:51 +0200 Subject: [PATCH] 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 ; 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 Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG --- benchmarks/pairs/catalog_gen.py | 395 +++++++++++++++ benchmarks/pairs/harvest.py | 604 +++++++++++++++++++++++ benchmarks/pairs/vfc/generate_catalog.py | 96 +--- benchmarks/pairs/vfc/harvest.py | 199 +------- src/openultrasast/semantic/functions.py | 151 ++++++ tests/test_pair_harvest.py | 365 ++++++++++++++ tests/test_vfc_harvest.py | 95 ---- 7 files changed, 1530 insertions(+), 375 deletions(-) create mode 100644 benchmarks/pairs/catalog_gen.py create mode 100644 benchmarks/pairs/harvest.py create mode 100644 src/openultrasast/semantic/functions.py create mode 100644 tests/test_pair_harvest.py delete mode 100644 tests/test_vfc_harvest.py diff --git a/benchmarks/pairs/catalog_gen.py b/benchmarks/pairs/catalog_gen.py new file mode 100644 index 0000000..ececf8a --- /dev/null +++ b/benchmarks/pairs/catalog_gen.py @@ -0,0 +1,395 @@ +#!/usr/bin/env python3 +"""Rebuild /catalog.toml and /training/manifest.jsonl from /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"(? 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 ```` 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()) diff --git a/benchmarks/pairs/harvest.py b/benchmarks/pairs/harvest.py new file mode 100644 index 0000000..f875073 --- /dev/null +++ b/benchmarks/pairs/harvest.py @@ -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: /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: /)") + 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()) diff --git a/benchmarks/pairs/vfc/generate_catalog.py b/benchmarks/pairs/vfc/generate_catalog.py index 9e21445..f07dbf9 100644 --- a/benchmarks/pairs/vfc/generate_catalog.py +++ b/benchmarks/pairs/vfc/generate_catalog.py @@ -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__": diff --git a/benchmarks/pairs/vfc/harvest.py b/benchmarks/pairs/vfc/harvest.py index 571b441..0ce9c58 100644 --- a/benchmarks/pairs/vfc/harvest.py +++ b/benchmarks/pairs/vfc/harvest.py @@ -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__": diff --git a/src/openultrasast/semantic/functions.py b/src/openultrasast/semantic/functions.py new file mode 100644 index 0000000..0252d05 --- /dev/null +++ b/src/openultrasast/semantic/functions.py @@ -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 ````. 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 = "" +ANON_FUNCTION = "" +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 ```` 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|(? 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) diff --git a/tests/test_pair_harvest.py b/tests/test_pair_harvest.py new file mode 100644 index 0000000..361f67e --- /dev/null +++ b/tests/test_pair_harvest.py @@ -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 "" + 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") diff --git a/tests/test_vfc_harvest.py b/tests/test_vfc_harvest.py deleted file mode 100644 index 1807b4b..0000000 --- a/tests/test_vfc_harvest.py +++ /dev/null @@ -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