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

fix(learn): fixed-side negatives only where the fixed side declares the function

Browse Source 
A pair now gives a fixed-side negative only when the fixed side's source declares the
labelled function (the excerpt builder's matcher; a name on a code line where the matcher
resolves neither side), or when a deterministic rename at a distinct fix commit replaces it
(one new function, same parameter list, changed body: provenance fixed_side_moved). Anything
else is counted under pair_fixed_side.absent_fixed_side, not labelled.

108 pair labels lose their negative: 80 seeded pointer pairs whose fixed side is a guard
helper of the same commit, and 28 Real-Vuln-Benchmark traps whose fixed side is a sibling
handler of the same snapshot (4 of those handlers are labelled positives themselves). The
memory build no longer substitutes the catalog's fix_function on the fixed side. The pair
cache is an allowed label-source tree so pointer sides are read through the guard.

Snapshot: benchmarks/measurements/2026-10-01-decision-engine-labels (counts only).

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo
This commit is contained in:
Marius Ciepluch
2026-10-01 11:03:54 +02:00
co-authored by Claude Opus 5.5
parent 7f6bbacb6e
commit 658f2d5e39
5 changed files with 711 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/openultrasast/learn/cli.py
-1
View File
@@ -211,7 +211,6 @@ def _memory_build(args: argparse.Namespace) -> int:
try:
build = build_examples(
labels, records_by_key(rows), reader, sources=sources, store=store, profile=args.profile, license_of=reader.license,
function_of=reader.function,
) # fmt: skip
except ExampleBuildError as exc:
print(f"learn memory build: {exc}", file=sys.stderr)
src/openultrasast/learn/examples.py
+5 -9
View File
@@ -15,8 +15,11 @@ Memory examples carry **no rationale** (code, signals, roles, family and label o
- conditional rows (``benign_control``, ``assumed_benign``) are skipped until their condition is evaluated;
- a candidate whose excerpt is empty or whose function is not found is never stored; when more than 10% of a
source's recorded candidates have no excerpt the build fails that source (exit 2 in the CLI): an unread clone is
not a small memory. A fixed side whose file was read but no longer declares the labelled function (a pointer
pair whose fix lives in another function) is a label defect, counted apart as ``absent_fixed_side``.
not a small memory. A fixed side whose file was read but no longer declares the labelled function is a label
defect, counted apart as ``absent_fixed_side`` (the label builder emits a fixed-side negative only where the fixed
side declares its candidate function, so this count should stay 0). A pair label names the function its side
declares -- the labelled one, or the renamed one of a ``fixed_side_moved`` negative -- never the catalog's
``fix_function`` (a different handler of the same snapshot).
Pair labels have no commit pin (the corpus holds excerpts); their envelope pin is :func:`label_pin`, a stable
40-hex digest of the pair and its side, so the vulnerable and the fixed side stay distinct rows.
@@ -255,13 +258,6 @@ class SideReader:
self._pairs = {case.name: case for case in load_pair_catalog(self.root / source.files[0])}
return self._pairs
def function(self, label: Mapping[str, Any], pin_role: str) -> str | None:
"""The function a pair's fixed side holds when it is not the labelled one (``fix_function``)."""
if label.get("source") != "pairs" or pin_role != "fixed":
return None
case = self._pair_cases().get(str(label.get("source_ref")))
return str(getattr(case, "fix_function", "") or "") or None
def license(self, label: Mapping[str, Any]) -> str:
if label.get("source") != "pairs":
return ""
src/openultrasast/learn/labels.py
+128 -4
View File
@@ -13,7 +13,11 @@ Rules per source (design section 3):
protocol assumption, **conditional** on a candidate there having a signal at the tip and none at the base
(``benign_control``). A vulnerable-pin candidate matching no site is unlabelled, never negative.
- *Pairs*: the labelled function on the vulnerable side (positive) and on the fixed side (negative); unscorable
pairs and the ``title`` tier are excluded (the tier is a sensitivity arm, ``--include-title``).
pairs and the ``title`` tier are excluded (the tier is a sensitivity arm, ``--include-title``). The negative is
emitted only where the fixed side's source declares the function (the excerpt builder's matcher) or a
deterministic rename at a distinct fix commit replaces it (``provenance = fixed_side_moved``); a fixed side that
holds another function of the same snapshot (a guard helper, a sibling handler) gives no negative and is counted
as ``absent_fixed_side``.
- *Recipes* (dev-php): the reviewed ``[[known]]`` function at the recipe commit (positive).
- *Adjudications*: recorded ``True``/``TP`` verdicts positive (``privileged`` kept), ``False``/``FP`` negative, with
the recorded reason as evidence.
@@ -41,7 +45,7 @@ from pathlib import Path
from typing import Any
from ..model.taxonomy import load_families
from ..plane.tasks.repo_facts import _EXCLUDED_DIRS, _NOT_PRODUCT, DECLARATION, GLOBAL, enclosing
from ..plane.tasks.repo_facts import _EXCLUDED_DIRS, _NOT_PRODUCT, DECLARATION, GLOBAL, _declared_name, declared_functions, enclosing
from ..preprocess import detect_language
from ..ruleset.frameworks import load_frameworks
@@ -134,6 +138,10 @@ class SourceGuard:
self.files.add(resolved)
if source.kind == "pairs":
self.trees.add(resolved.parent)
if any(source.kind == "pairs" for source in sources.sources):
from ..pairs import pair_cache_dir
self.trees.add(pair_cache_dir().resolve()) # pointer pairs: the catalog's sides, materialised outside the tree
self.opened: list[Path] = []
def check(self, path: Path) -> Path:
@@ -317,6 +325,7 @@ class Label:
conditional: str | None = None
design_informed: bool = False
privileged: bool = False
provenance: str = "" # how a negative was derived when it is not the labelled function itself (``fixed_side_moved``)
def _family_of(cwe: str | None, family: str | None) -> str:
@@ -432,6 +441,50 @@ def changed_functions(clone: Clone, pin: str, ranges: Mapping[str, Sequence[tupl
return out
# --- the fixed side of a pair ----------------------------------------------------------------------------------------
FIXED_DECLARED = "declared"
FIXED_MOVED = "fixed_side_moved"
ABSENT_UNREAD = "fixed side unread"
ABSENT_UNMATCHED = "fixed side does not declare the function"
ABSENT_SAME_SNAPSHOT = "fixed side is another function of the same snapshot"
ABSENT_NO_EQUIVALENT = "fixed side does not declare the function, no renamed equivalent"
_HEADER = re.compile(r"^\s*(?:#|//|/\*|\*)") # comment lines: the excerpt header names the labelled function
_PIN_LINE = re.compile(r"^\W*(commit|parent):\s*([0-9a-f]{7,40})\b", re.IGNORECASE)
@dataclass(frozen=True)
class FixedSide:
verdict: str
function: str = ""
def _fix_commits(case: Any, fixed_lines: Sequence[str]) -> tuple[str, str] | None:
"""``(vulnerable, fix)`` commits of a pair: the recipe's ``parent``/``commit`` (pointer pairs) or the excerpt
header's; ``None`` when either is unrecorded (a fix that cannot be told apart from its snapshot)."""
recorded = dict(getattr(case, "recipe", ()) or ())
header = {m.group(1).lower(): m.group(2) for line in fixed_lines[:20] if (m := _PIN_LINE.match(line))}
parent = str(recorded.get("parent") or header.get("parent") or "")
commit = str(getattr(case, "commit", "") or recorded.get("commit") or header.get("commit") or "")
return (parent.lower(), commit.lower()) if parent and commit else None
def _signature(lines: Sequence[str], start: int, function: str) -> str | None:
"""The parameter list after ``function`` on its declaration (up to 5 lines), whitespace-normalised."""
text = "\n".join(lines[start : start + 5])
at = re.search(rf"(?<![\w$]){re.escape(function)}\s*\(", text)
if at is None:
return None
depth, out = 0, []
for char in text[at.end() - 1 :]:
depth += char == "("
depth -= char == ")"
out.append(char)
if depth == 0:
return re.sub(r"\s+", " ", "".join(out))
return None
@dataclass
class Build:
"""The rows of one build and what it could not label (reported, never silent)."""
@@ -439,6 +492,7 @@ class Build:
labels: list[Label] = field(default_factory=list)
skipped: dict[str, int] = field(default_factory=dict)
history: dict[str, Any] = field(default_factory=dict)
absent_fixed_side: dict[str, int] = field(default_factory=dict) # family -> pair labels with no fixed-side negative
def skip(self, reason: str, n: int = 1) -> None:
self.skipped[reason] = self.skipped.get(reason, 0) + n
@@ -584,8 +638,71 @@ class LabelBuilder:
} # fmt: skip
self._row(label=1, unit="pin", **common)
self._row(label=1, unit="delta", direction="introduce", **common)
self._row(label=0, unit="pin", pin_role="fixed", **common)
self._row(label=0, unit="delta", direction="repair", pin_role="fixed", **common)
fixed = self._fixed_side(case, expected.function)
if fixed.verdict == FIXED_DECLARED:
self._row(label=0, unit="pin", pin_role="fixed", **common)
self._row(label=0, unit="delta", direction="repair", pin_role="fixed", **common)
elif fixed.verdict == FIXED_MOVED:
rename = f"{FIXED_MOVED}: {expected.function} -> {fixed.function} (same file, same signature, body changed)"
moved = {
**common, "candidate": f"{path}::{fixed.function}", "provenance": FIXED_MOVED,
"evidence": f"{common['evidence']}; {rename}",
} # fmt: skip
self._row(label=0, unit="pin", pin_role="fixed", **moved)
self._row(label=0, unit="delta", direction="repair", pin_role="fixed", **moved)
else:
family = str(common["family"])
self.build.absent_fixed_side[family] = self.build.absent_fixed_side.get(family, 0) + 1
self.build.skip(f"pairs: no fixed-side negative ({fixed.verdict})")
def _pair_lines(self, file: Path) -> list[str] | None:
"""A pair side's lines through the guard (catalog excerpts, or pointer files in the pair cache); ``None`` unread."""
try:
return self.guard.text(file).splitlines() if Path(file).is_file() else None
except (OSError, UnicodeDecodeError):
return None
def _fixed_side(self, case: Any, function: str) -> FixedSide:
"""Whether the fixed side gives a defensible negative for ``function`` (Req 2.1: the same function at the fix).
Verified from the fixed side's source with the excerpt builder's own matcher: the fixed side must declare the
labelled function. Where it does not, the only other negative accepted is a deterministic rename at a real
fix (a fix commit distinct from the vulnerable one): exactly one function the fixed side declares that the
vulnerable side does not, with the labelled function's parameter list and a changed body (``fixed_side_moved``).
A different function of the same snapshot (a guard helper, a sibling handler) is never a negative."""
from .excerpt import LANGUAGE_ALIASES, function_bounds
fixed_lines, vuln_lines = self._pair_lines(case.fixed_file), self._pair_lines(case.vuln_file)
if not fixed_lines:
return FixedSide(ABSENT_UNREAD)
language = LANGUAGE_ALIASES.get(case.language, case.language)
if function_bounds(fixed_lines, language, function):
return FixedSide(FIXED_DECLARED)
vuln_span = function_bounds(vuln_lines or (), language, function)
if vuln_span is None:
# the matcher resolves the function on neither side: its limit, not evidence of absence. The fixed side
# still declares it when its name is on a code line (the excerpt header names it in a comment).
named = re.compile(rf"(?<![\w$]){re.escape(function)}(?![\w$])")
code = [line for line in fixed_lines if not _HEADER.match(line)]
return FixedSide(FIXED_DECLARED if any(named.search(line) for line in code) else ABSENT_UNMATCHED)
commits = _fix_commits(case, fixed_lines)
if commits is not None and commits[0] == commits[1]:
return FixedSide(ABSENT_SAME_SNAPSHOT)
pattern = DECLARATION.get(language)
if pattern is None or vuln_lines is None or commits is None:
return FixedSide(ABSENT_NO_EQUIVALENT)
vulnerable_names = set(declared_functions(vuln_lines, language))
fixed_names = [name for name in (_declared_name(pattern, line) for line in fixed_lines) if name]
signature = _signature(vuln_lines, vuln_span[0], function)
body = "\n".join(vuln_lines[slice(*vuln_span)][1:]).strip()
moved = []
for name in dict.fromkeys(fixed_names):
span = function_bounds(fixed_lines, language, name)
if name in vulnerable_names or fixed_names.count(name) != 1 or span is None or signature is None:
continue
if _signature(fixed_lines, span[0], name) == signature and "\n".join(fixed_lines[slice(*span)][1:]).strip() != body:
moved.append(name)
return FixedSide(FIXED_MOVED, moved[0]) if len(moved) == 1 else FixedSide(ABSENT_NO_EQUIVALENT)
# -- development recipes ------------------------------------------------------------------------------------------
@@ -795,6 +912,13 @@ def snapshot(build: Build, sources: Sources) -> dict[str, Any]:
for source in sorted({r.source for r in conditional})
},
"assumed_benign_history": build.history,
"pair_fixed_side": {
"absent_fixed_side": {
"labels": sum(build.absent_fixed_side.values()),
"by_family": dict(sorted(build.absent_fixed_side.items())),
},
"fixed_side_moved": sum(1 for r in pins if r.provenance == FIXED_MOVED),
},
"skipped": dict(sorted(build.skipped.items())),
"floors": {
"per_family": PER_FAMILY_FLOOR,