| """Post-cutoff OOD CVE set mining (spec §3.1.4): ≥200 C/C++ single-function |
| vuln/patch pairs from CVEs disclosed 2025-01-01 onward (Qwen3-4B's training |
| cutoff treated as ~mid-2024, per spec). |
| |
| Pipeline (implemented as pure, independently-testable functions; the network |
| calls themselves live in scripts/build_ood_set.py, which composes these): |
| 1. Query NVD's CVE API for CVEs published >= 2025-01-01. |
| 2. Extract GitHub commit references from each CVE's reference URLs. |
| 3. Fetch that commit's changed files via the GitHub API; keep only C/C++ files. |
| 4. For a kept file, fetch its content at the commit's parent (vulnerable) and |
| at the commit itself (patched), then isolate the single enclosing function |
| that the diff touches — reject multi-function changes (spec: "single- |
| function-change pairs"). |
| 5. Log every rejection with a reason to collection_log.jsonl (spec: "keep a |
| collection_log.jsonl with rejection reasons"). |
| """ |
| from __future__ import annotations |
|
|
| import re |
| from dataclasses import dataclass |
|
|
| from src.cpg.statement_mapper import parse_function |
| from src.data.normalize_text import strip_comments_and_whitespace |
|
|
| _C_CPP_EXTENSIONS = {".c", ".h", ".cc", ".cpp", ".cxx", ".hpp", ".hxx", ".h++", ".c++"} |
| _GITHUB_COMMIT_RE = re.compile( |
| r"github\.com/(?P<owner>[\w.-]+)/(?P<repo>[\w.-]+)/commit/(?P<sha>[0-9a-fA-F]{7,40})" |
| ) |
|
|
|
|
| @dataclass |
| class CommitRef: |
| owner: str |
| repo: str |
| sha: str |
| url: str |
|
|
|
|
| def extract_github_commit_refs(reference_urls: list[str]) -> list[CommitRef]: |
| """spec step: "query GitHub API for advisories with fix commits" — here, |
| extracts github.com/.../commit/<sha> references from an NVD CVE's |
| reference URL list (NVD doesn't filter by language; this is the first |
| filter step before fetching the actual commit).""" |
| seen: set[tuple[str, str, str]] = set() |
| out: list[CommitRef] = [] |
| for url in reference_urls: |
| m = _GITHUB_COMMIT_RE.search(url) |
| if not m: |
| continue |
| owner, repo, sha = m.group("owner"), m.group("repo"), m.group("sha") |
| key = (owner.lower(), repo.lower(), sha.lower()) |
| if key in seen: |
| continue |
| seen.add(key) |
| out.append(CommitRef(owner=owner, repo=repo, sha=sha, url=url)) |
| return out |
|
|
|
|
| def is_c_cpp_file(filename: str) -> bool: |
| for ext in _C_CPP_EXTENSIONS: |
| if filename.lower().endswith(ext): |
| return True |
| return False |
|
|
|
|
| @dataclass |
| class FunctionChangeResult: |
| accepted: bool |
| reason: str = "" |
| func_before: str = "" |
| func_after: str = "" |
| func_name: str = "" |
|
|
|
|
| def _enclosing_function_spans(code: str) -> list[tuple[int, int, str, str]]: |
| """Returns (start_line, end_line, name, source_text) for every top-level |
| function_definition in `code` (1-based, inclusive lines).""" |
| root = parse_function(code) |
| lines = code.splitlines(keepends=True) |
| spans = [] |
| for node in root.children: |
| if node.type != "function_definition": |
| continue |
| name = "" |
| for child in node.children: |
| if child.type == "function_declarator": |
| for gc in child.children: |
| if gc.type == "identifier": |
| name = gc.text.decode("utf-8", errors="replace") |
| break |
| start_line = node.start_point[0] + 1 |
| end_line = node.end_point[0] + 1 |
| text = "".join(lines[start_line - 1 : end_line]) |
| spans.append((start_line, end_line, name, text)) |
| return spans |
|
|
|
|
| def _changed_line_numbers(before: str, after: str) -> set[int]: |
| """1-based line numbers in `before` touched by a replace/delete opcode |
| (reuses the same approach as src/data/load_primevul.py's diffing, minus |
| the insertion-anchoring nuance — here we just need "which lines moved" |
| to find the enclosing function, not a reward-grade patch_lines_vuln).""" |
| import difflib |
|
|
| before_lines = before.splitlines(keepends=True) |
| after_lines = after.splitlines(keepends=True) |
| changed: set[int] = set() |
| sm = difflib.SequenceMatcher(a=before_lines, b=after_lines, autojunk=False) |
| for tag, i1, i2, _j1, _j2 in sm.get_opcodes(): |
| if tag in ("replace", "delete"): |
| changed.update(range(i1 + 1, i2 + 1)) |
| elif tag == "insert": |
| changed.add(max(1, i1)) |
| return changed |
|
|
|
|
| def extract_single_function_change(file_before: str, file_after: str) -> FunctionChangeResult: |
| """Isolates the single function a diff touches. Rejects (with a reason) |
| if the change spans zero or multiple functions, or if the function body |
| is unchanged after dedup-normalization (a no-op/formatting-only diff).""" |
| changed_lines = _changed_line_numbers(file_before, file_after) |
| if not changed_lines: |
| return FunctionChangeResult(accepted=False, reason="no_line_level_changes_detected") |
|
|
| before_spans = _enclosing_function_spans(file_before) |
| touched = [ |
| (start, end, name, text) |
| for (start, end, name, text) in before_spans |
| if any(start <= line <= end for line in changed_lines) |
| ] |
| if len(touched) == 0: |
| return FunctionChangeResult(accepted=False, reason="changed_lines_outside_any_function") |
| if len(touched) > 1: |
| return FunctionChangeResult( |
| accepted=False, |
| reason=f"multi_function_change ({len(touched)} functions touched)", |
| ) |
|
|
| _, _, name, func_before_text = touched[0] |
| if not name: |
| return FunctionChangeResult(accepted=False, reason="could_not_resolve_function_name") |
|
|
| after_spans = _enclosing_function_spans(file_after) |
| after_match = next((t for t in after_spans if t[2] == name), None) |
| if after_match is None: |
| return FunctionChangeResult( |
| accepted=False, reason=f"function {name!r} not found (by name) in patched file" |
| ) |
| func_after_text = after_match[3] |
|
|
| if strip_comments_and_whitespace(func_before_text) == strip_comments_and_whitespace(func_after_text): |
| return FunctionChangeResult(accepted=False, reason="no_semantic_change_after_normalization") |
|
|
| return FunctionChangeResult( |
| accepted=True, func_before=func_before_text, func_after=func_after_text, func_name=name |
| ) |
|
|
