File size: 6,224 Bytes
5bfc214 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 | """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
)
|