mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-21 05:59:29 +00:00
feat(hunt): feature table, producers, labels, metric, artifact, the DSPy hunt program, runner and CLI
hunter-learning-loop groups 1-3 (first increment). The feature table is the contract: one row per touched function with diff, source, semantic and taxonomy columns, provenance per producer, labels by the benchmark's rule. Producers read the repository at head only (a planted advisory id and commit text never reach a feature). The hunt program is a DSPy ReAct loop over a signature whose inputs are the unmasked columns in a fixed order; answers outside the closed family list or citing unseen lines score as unknown. The runner writes records the benchmark scorer reads, resumable by change, program id and mask, and refuses the held-out pilot unless the run is a report run. Research extra [hunt] = dspy, pandas; the core install is unchanged. 43 tests on a scripted model, no network. Co-Authored-By: Claude Fable 5.1 <[email protected]>
This commit is contained in:
Marius Ciepluch
co-authored by
Claude Fable 5.1
Claude Fable 5.1
parent
e55e44d981
commit
d9d6335ab3
@@ -0,0 +1,294 @@
|
||||
"""Four fixed annotations computed from head source and the tool's own facts."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
import subprocess
|
||||
from collections.abc import Callable, Mapping
|
||||
from dataclasses import dataclass, field
|
||||
from functools import cached_property
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from .. import mapping, source
|
||||
from ..model.taxonomy import DEFAULT_FAMILIES_PATH, load_families
|
||||
from ..preprocess import build_file_target, detect_language
|
||||
from ..semantic.facts import load_facts
|
||||
from ..semantic.ir import FileIR, parse_file
|
||||
from ..semantic.obligations import check_obligations, load_obligation_facts
|
||||
from ..semantic.obligations.dominance import OrderDominance
|
||||
from ..semantic.obligations.operations import covers, find_operations, valid_witness
|
||||
from .labels import label_rows
|
||||
from .table import FEATURE_NAMES
|
||||
|
||||
Change = Mapping[str, Any]
|
||||
|
||||
|
||||
class _Absent(ValueError):
|
||||
def __init__(self, reason: str) -> None:
|
||||
self.reason = reason
|
||||
super().__init__(reason)
|
||||
|
||||
|
||||
def _git(root: Path, *args: str) -> str:
|
||||
result = subprocess.run(["git", "-C", str(root), *args], capture_output=True, check=True)
|
||||
return result.stdout.decode("utf-8", errors="replace")
|
||||
|
||||
|
||||
def _safe_path(root: Path, path: str) -> Path:
|
||||
relative = Path(path)
|
||||
resolved = (root / relative).resolve()
|
||||
if relative.is_absolute() or any(part.startswith(".env") or part == ".git" for part in relative.parts):
|
||||
raise ValueError("private or absolute source path")
|
||||
if not resolved.is_relative_to(root.resolve()):
|
||||
raise ValueError("source path is outside the repository")
|
||||
return resolved
|
||||
|
||||
|
||||
@dataclass
|
||||
class Context:
|
||||
"""Cache work shared by every candidate of one checked-out change."""
|
||||
|
||||
root: Path
|
||||
change: Change
|
||||
texts: dict[str, str] = field(default_factory=dict)
|
||||
spans: dict[str, tuple[source.FunctionRange, ...]] = field(default_factory=dict)
|
||||
range_errors: dict[str, str] = field(default_factory=dict)
|
||||
irs: dict[str, FileIR] = field(default_factory=dict)
|
||||
diff_bytes: int = 0
|
||||
|
||||
@cached_property
|
||||
def head_verified(self) -> bool:
|
||||
"""True once the checkout is proven to be at the change's head; absent otherwise."""
|
||||
if _git(self.root, "rev-parse", "HEAD").strip() != _git(self.root, "rev-parse", str(self.change["head"])).strip():
|
||||
raise _Absent("head_mismatch")
|
||||
return True
|
||||
|
||||
def verify_head(self) -> None:
|
||||
if not self.head_verified:
|
||||
raise _Absent("head_mismatch")
|
||||
|
||||
def read(self, path: str) -> str:
|
||||
safe = _safe_path(self.root, path)
|
||||
if path not in self.texts:
|
||||
self.verify_head()
|
||||
data = safe.read_bytes()
|
||||
if not data:
|
||||
raise _Absent("empty_source")
|
||||
self.texts[path] = data.decode("utf-8", errors="replace")
|
||||
return self.texts[path]
|
||||
|
||||
def ranges(self, path: str) -> tuple[source.FunctionRange, ...]:
|
||||
if path not in self.spans:
|
||||
self.spans[path] = source.ranges(path, self.read(path), source.canonical_language(detect_language(self.root / path)))
|
||||
return self.spans[path]
|
||||
|
||||
@cached_property
|
||||
def product_files(self) -> list[str]:
|
||||
return source.product_files(self.root)
|
||||
|
||||
@cached_property
|
||||
def entry_points(self) -> list[mapping.EntryPointRecord]:
|
||||
for path in self.product_files:
|
||||
self.read(path)
|
||||
return mapping.analyze_entry_points(self.root, [build_file_target(self.root, self.root / path) for path in self.product_files])
|
||||
|
||||
@cached_property
|
||||
def hunks(self) -> dict[str, list[list[int]]]:
|
||||
self.verify_head()
|
||||
# Exclude private files before git opens the patch; only hunk coordinates become features.
|
||||
patch = _git(self.root, "diff", "--no-ext-diff", "--no-textconv", "-U0",
|
||||
f"{self.change['base']}..{self.change['head']}", "--", ".",
|
||||
":(glob,exclude)**/.env*", ":(exclude).env*")
|
||||
self.diff_bytes = len(patch.encode("utf-8"))
|
||||
result: dict[str, list[list[int]]] = {}
|
||||
path = None
|
||||
for text in patch.splitlines():
|
||||
if text.startswith("+++ "):
|
||||
token = text[4:]
|
||||
token = json.loads(token) if token.startswith('"') else token
|
||||
path = token[2:] if token.startswith("b/") else None
|
||||
elif path is not None and (match := re.match(r"@@ -\d+(?:,\d+)? \+(\d+)(?:,(\d+))? @@", text)):
|
||||
start, count = int(match[1]), int(match[2] or "1")
|
||||
if count:
|
||||
result.setdefault(path, []).append([start, start + count - 1])
|
||||
return result
|
||||
|
||||
def ir(self, path: str) -> FileIR:
|
||||
if path not in self.irs:
|
||||
self.irs[path] = parse_file(path, self.read(path), source.canonical_language(detect_language(self.root / path)))
|
||||
return self.irs[path]
|
||||
|
||||
|
||||
Compute = Callable[[Path, Change, str, int, Context], dict[str, object]]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Producer:
|
||||
"""A versioned producer with a closed output column list and private failure handling."""
|
||||
|
||||
name: str
|
||||
version: str
|
||||
columns: tuple[str, ...]
|
||||
compute: Compute
|
||||
|
||||
|
||||
def _producer(name: str, compute: Compute) -> Producer:
|
||||
columns = tuple(f"feature.{name}.{column}" for column in FEATURE_NAMES[name])
|
||||
provenance = tuple(f"provenance.{name}.{column}" for column in ("ok", "reason", "version", "bytes_read"))
|
||||
|
||||
def guarded(root: Path, change: Change, path: str, line: int, ctx: Context) -> dict[str, object]:
|
||||
result: dict[str, object] = dict.fromkeys(columns)
|
||||
reason = None
|
||||
before = set(ctx.texts)
|
||||
try:
|
||||
values = compute(root, change, path, line, ctx)
|
||||
if set(values) != set(FEATURE_NAMES[name]):
|
||||
raise ValueError("producer output does not match its columns")
|
||||
result.update({f"feature.{name}.{key}": value for key, value in values.items()})
|
||||
except Exception as error:
|
||||
reason = error.reason if isinstance(error, _Absent) else type(error).__name__
|
||||
bytes_read = sum(len(text.encode("utf-8")) for key, text in ctx.texts.items() if key not in before or key == path)
|
||||
if name == "diff":
|
||||
bytes_read = ctx.diff_bytes
|
||||
elif name == "taxonomy" and reason is None:
|
||||
bytes_read = len(DEFAULT_FAMILIES_PATH.read_bytes())
|
||||
result.update({f"provenance.{name}.ok": reason is None, f"provenance.{name}.reason": reason,
|
||||
f"provenance.{name}.version": "1", f"provenance.{name}.bytes_read": bytes_read})
|
||||
return result
|
||||
|
||||
return Producer(name, "1", columns + provenance, guarded)
|
||||
|
||||
|
||||
def _diff(root: Path, change: Change, path: str, line: int, ctx: Context) -> dict[str, object]:
|
||||
hunks = ctx.hunks.get(path, [])
|
||||
distance = min((max(start - line, 0, line - end) for start, end in hunks), default=None)
|
||||
return {"hunks": hunks, "on_changed_line": distance == 0, "distance": distance}
|
||||
|
||||
|
||||
def _source(root: Path, change: Change, path: str, line: int, ctx: Context) -> dict[str, object]:
|
||||
ctx.read(path)
|
||||
if path in ctx.range_errors:
|
||||
raise _Absent(ctx.range_errors[path])
|
||||
facts = source.facts_for(root, path, line, entry_points=ctx.entry_points)
|
||||
# Select the occurrence at this line when several methods share a name.
|
||||
spans = [r for r in ctx.ranges(path) if r.name == facts.function and r.start <= line <= r.end]
|
||||
span = min(spans, key=lambda r: r.end - r.start) if spans else None
|
||||
return {"function": facts.function, "span": [span.start, span.end] if span else None,
|
||||
"callers": [list(caller) for caller in facts.callers],
|
||||
"entry_points": [{"path": entry.path, "line": entry.line, "kind": entry.kind} for entry in facts.entry_points],
|
||||
"language": facts.language}
|
||||
|
||||
|
||||
def _roles(function: Any, facts: Any) -> dict[str, str]:
|
||||
names = set(function.params)
|
||||
names.update(call.name for call in function.calls)
|
||||
for bind in function.binds:
|
||||
names.update((bind.name, *bind.names))
|
||||
if bind.call_name:
|
||||
names.add(bind.call_name)
|
||||
for call in function.calls:
|
||||
for arguments in call.arg_names:
|
||||
names.update(arguments)
|
||||
roles = {}
|
||||
for role, entries, attribute in (("source", facts.sources, "patterns"), ("sink", facts.sinks, "calls"),
|
||||
("sanitizer", facts.sanitizers, "calls")):
|
||||
for fact in entries:
|
||||
for pattern in getattr(fact, attribute):
|
||||
token = pattern.rstrip("(")
|
||||
for name in sorted(names):
|
||||
if name == token or name.endswith("." + token) or token.startswith(name + ".") and role == "source":
|
||||
# A shared sink/sanitizer name remains a sink; parameterization needs more than a name.
|
||||
roles.setdefault(name, role)
|
||||
return roles
|
||||
|
||||
|
||||
def _obligations(path: str, ctx: Context) -> list[dict[str, object]]:
|
||||
ir, text = ctx.ir(path), ctx.read(path)
|
||||
facts, flow = load_obligation_facts(), load_facts()
|
||||
scoped = facts.for_language(ir.language)
|
||||
if not scoped.operations and not scoped.dischargers:
|
||||
raise _Absent("obligations_language_unsupported")
|
||||
dominance = OrderDominance({path: text})
|
||||
result = check_obligations(irs={path: (ir, text)}, entries=ctx.entry_points, facts=facts, flow_facts=flow,
|
||||
policy=None, paths=(), dominance=dominance, store_shapes=(), min_siblings=3)
|
||||
# No finding does not imply discharge: the checker also gates on entry-point reachability.
|
||||
return [{"operation": operation.kind, "line": operation.line,
|
||||
"discharged": any(valid_witness(witness) and covers(witness, operation) and
|
||||
dominance.dominates(witness, operation, None) for witness in result.discharges)}
|
||||
for operation in find_operations(ir, scoped, text=text)]
|
||||
|
||||
|
||||
def _semantic(root: Path, change: Change, path: str, line: int, ctx: Context) -> dict[str, object]:
|
||||
ir = ctx.ir(path)
|
||||
if not ir.parse_ok:
|
||||
raise _Absent("language_unsupported" if ir.reason == "language_unsupported" else "parse_failed")
|
||||
functions = [fn for fn in ir.functions if fn.start_line <= line <= fn.end_line]
|
||||
if not functions:
|
||||
raise _Absent("function_unavailable")
|
||||
function = min(functions, key=lambda fn: fn.end_line - fn.start_line)
|
||||
return {"roles": _roles(function, load_facts().for_language(ir.language)),
|
||||
"def_use": {"binds": [{"line": b.line, "name": b.name, "names": list(b.names), "call_name": b.call_name}
|
||||
for b in function.binds],
|
||||
"calls": [{"line": c.line, "name": c.name, "names": [list(names) for names in c.arg_names]}
|
||||
for c in function.calls]},
|
||||
"obligations": _obligations(path, ctx)}
|
||||
|
||||
|
||||
def _taxonomy(root: Path, change: Change, path: str, line: int, ctx: Context) -> dict[str, object]:
|
||||
return {"families": [{"id": f.id, "definition": " ".join(f.description.split())} for f in load_families().families]}
|
||||
|
||||
|
||||
PRODUCERS = (_producer("diff", _diff), _producer("source", _source),
|
||||
_producer("semantic", _semantic), _producer("taxonomy", _taxonomy))
|
||||
|
||||
|
||||
def _candidates(root: Path, change: Change, ctx: Context) -> list[dict[str, Any]]:
|
||||
selected: set[tuple[str, int]] = set()
|
||||
files = list(change.get("files", []))
|
||||
if change.get("kind") == "vulnerability" and not any(f["path"] == change["path"] for f in files):
|
||||
files.append({"path": change["path"], "head_lines": []})
|
||||
for file in files:
|
||||
path = file["path"]
|
||||
_safe_path(root, path)
|
||||
try:
|
||||
ranges = ctx.ranges(path)
|
||||
except Exception as error:
|
||||
ctx.range_errors[path] = error.reason if isinstance(error, _Absent) else type(error).__name__
|
||||
line = next((start for start, _ in file.get("head_lines", [])), 1)
|
||||
selected.add((path, max(1, line)))
|
||||
continue
|
||||
touched = [r for r in ranges if r.kind == "function" and
|
||||
any(r.start <= end and start <= r.end for start, end in file.get("head_lines", []))]
|
||||
selected.update((path, r.start) for r in touched)
|
||||
if not touched and file.get("head_lines") and ranges:
|
||||
selected.add((path, max(1, file["head_lines"][0][0])))
|
||||
if change.get("kind") == "vulnerability" and path == change["path"]:
|
||||
labelled = [r for r in ranges if r.name == change["function"] and
|
||||
r.start <= change["function_lines"][1] and change["function_lines"][0] <= r.end]
|
||||
selected.update((path, r.start) for r in labelled)
|
||||
if not labelled:
|
||||
selected.add((path, change["function_lines"][0]))
|
||||
return [{"change_id": change.get("id", change.get("case_id")), "repo": change.get("repo", change.get("repository")),
|
||||
"head": change["head"], "path": path, "line": line, "kind": change["kind"], "fold": change.get("fold")}
|
||||
for path, line in sorted(selected)]
|
||||
|
||||
|
||||
def candidate_rows(root: Path, change: Change) -> list[dict[str, Any]]:
|
||||
"""Enumerate touched head functions and the positive fix site, retaining unreadable files."""
|
||||
root = Path(root)
|
||||
return _candidates(root, change, Context(root, change))
|
||||
|
||||
|
||||
def compute_rows(root: Path, change: Change, producers: tuple[Producer, ...] = PRODUCERS) -> list[dict[str, Any]]:
|
||||
"""Fill candidate rows, then join labels after every producer has finished."""
|
||||
root = Path(root)
|
||||
# Producers receive only revision coordinates; advisory and label fields never cross this boundary.
|
||||
coordinates = {key: change[key] for key in ("base", "head")}
|
||||
ctx = Context(root, coordinates)
|
||||
rows = _candidates(root, change, ctx)
|
||||
for row in rows:
|
||||
for producer in producers:
|
||||
row.update(producer.compute(root, coordinates, row["path"], row["line"], ctx))
|
||||
return label_rows(rows, change)
|
||||
Reference in New Issue
Block a user