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

feat(pair-corpus-honesty): shared function naming, harvest library, catalog generator

Browse Source 
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
2026-09-05 10:45:42 +02:00
co-authored by Claude Fable 5.1
parent 0116f9bea6
commit 071dea4712
7 changed files with 1530 additions and 375 deletions
Download Patch File Download Diff File Expand all files Collapse all files
benchmarks/pairs/catalog_gen.py
+395
View File
@@ -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())
benchmarks/pairs/harvest.py
+604
View File
@@ -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())
benchmarks/pairs/vfc/generate_catalog.py
+4 -92
View File
@@ -1,104 +1,16 @@
#!/usr/bin/env python3 #!/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 from __future__ import annotations
import json import runpy
import tomllib
from pathlib import Path from pathlib import Path
ROOT = Path(__file__).resolve().parent _LIB = Path(__file__).resolve().parents[1] / "catalog_gen.py"
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
def main() -> int: def main() -> int:
recipes = list(tomllib.loads(RECIPES.read_text()).get("recipe", [])) return int(runpy.run_path(str(_LIB), run_name="catalog_gen_lib")["main"](["--slice", "vfc"]))
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
if __name__ == "__main__": if __name__ == "__main__":
benchmarks/pairs/vfc/harvest.py
+11 -188
View File
@@ -1,203 +1,26 @@
#!/usr/bin/env python3 #!/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 Kept so existing maintainer commands and tests keep working:
(no parent and commit) are rejected. FixFox is not a fetch source. ``python benchmarks/pairs/vfc/harvest.py --name openssl-cve-2014-0160``.
""" """
from __future__ import annotations from __future__ import annotations
import argparse import runpy
import sys import sys
import tomllib
from pathlib import Path from pathlib import Path
from typing import Any
from urllib.parse import urlparse
from urllib.request import Request, urlopen
ALLOWED_HOSTS = { _LIB = Path(__file__).resolve().parents[1] / "harvest.py"
"raw.githubusercontent.com", _NS = runpy.run_path(str(_LIB), run_name="pair_harvest_lib")
"github.com", globals().update({key: value for key, value in _NS.items() if not key.startswith("__") and key != "main"})
"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", []))
def main(argv: list[str] | None = None) -> int: def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=__doc__) args = list(sys.argv[1:] if argv is None else argv)
parser.add_argument("--recipes", type=Path, default=Path(__file__).with_name("recipes.toml")) if "--slice" not in args:
parser.add_argument("--name", help="recipe name to harvest (omit with --all)") args = ["--slice", "vfc", *args]
parser.add_argument("--all", action="store_true", help="process every recipe") return int(_NS["main"](args))
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
if __name__ == "__main__": if __name__ == "__main__":
src/openultrasast/semantic/functions.py
+151
View File
@@ -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)
tests/test_pair_harvest.py
+365
View File
@@ -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")
tests/test_vfc_harvest.py
-95
View File
@@ -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