From 4d56b3c76e4631e4ec39908486b2360a476e4209 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sat, 5 Sep 2026 14:27:33 +0200 Subject: [PATCH] feat(corpus-seeded-mechanisms): structural shapes derived from labeled pairs Task 1.1 (Req 2.1-2.4). Spec approved by the maintainer in session. - semantic/variants.py: Shape (language, sink trailing name, arity, source positions and kinds, closed guard kind, mechanism id) with a stable text-free key; derive_shape finds the labeled sink call site in the vulnerable FileIR, traces argument identifiers through binds to a fact source, a parameter or a container read, and classifies the guard the fixed side added from the labeled function's added statements against a closed set (none when unknown). No path, line or literal enters a shape; a constant-only call teaches nothing. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG --- .../specs/corpus-seeded-mechanisms/spec.json | 13 +- .kiro/specs/corpus-seeded-mechanisms/tasks.md | 10 +- src/openultrasast/semantic/variants.py | 257 ++++++++++++++++++ tests/test_variants.py | 76 ++++++ 4 files changed, 347 insertions(+), 9 deletions(-) create mode 100644 src/openultrasast/semantic/variants.py create mode 100644 tests/test_variants.py diff --git a/.kiro/specs/corpus-seeded-mechanisms/spec.json b/.kiro/specs/corpus-seeded-mechanisms/spec.json index 64565b9..1900791 100644 --- a/.kiro/specs/corpus-seeded-mechanisms/spec.json +++ b/.kiro/specs/corpus-seeded-mechanisms/spec.json @@ -1,22 +1,23 @@ { "feature_name": "corpus-seeded-mechanisms", "created_at": "2026-09-05T06:18:31Z", - "updated_at": "2026-09-05T06:18:31Z", + "updated_at": "2026-09-05T13:00:00Z", "language": "en", - "phase": "tasks-generated", + "phase": "implementation", "approvals": { "requirements": { "generated": true, - "approved": false + "approved": true }, "design": { "generated": true, - "approved": false + "approved": true }, "tasks": { "generated": true, - "approved": false + "approved": true } }, - "ready_for_implementation": false + "ready_for_implementation": true, + "approval_note": "2026-09-05 approved by the maintainer in session ('continue with corpus-seeded-mechanisms')" } diff --git a/.kiro/specs/corpus-seeded-mechanisms/tasks.md b/.kiro/specs/corpus-seeded-mechanisms/tasks.md index c1fbf23..7bcd065 100644 --- a/.kiro/specs/corpus-seeded-mechanisms/tasks.md +++ b/.kiro/specs/corpus-seeded-mechanisms/tasks.md @@ -1,14 +1,14 @@ # Implementation Plan -- [ ] 1. Foundation: shapes and store fields -- [ ] 1.1 Define structural shapes and derive them from a pair +- [x] 1. Foundation: shapes and store fields +- [x] 1.1 Define structural shapes and derive them from a pair - `Shape` with language, sink name, arity, source positions and kinds, closed guard kind, mechanism id; `derive_shape` finds the labeled sink call site in the vulnerable `FileIR`, traces source positions through binds to a fact source, parameter, or container read, and classifies the guard from statements present only on the fixed side. - Shapes contain no path, line, or literal beyond identifiers. - Observable: a Python pair (request input into `os.system`, fix adds an allowlist) yields sink `system`, arity 1, source position 0, guard `allowlist_test`; an unlabeled sink returns None; the shape key is stable across runs. - _Requirements: 2.1, 2.2, 2.3, 2.4_ - _Boundary: Shape_ -- [ ] 1.2 Add additive mechanism fields and a corpus writer with dedupe +- [x] 1.2 Add additive mechanism fields and a corpus writer with dedupe - `Mechanism` gains origin, review_tier, pairs, shape, guard with defaults; `append_from_pair` dedupes by shape key and extends provenance; old JSONL rows load unchanged. - Observable: appending the same shape from two pairs yields one row listing both pairs; a pre-existing sandbox row loads with `origin = "sandbox"`. - _Requirements: 1.3, 1.4_ @@ -80,5 +80,9 @@ ## Implementation Notes +- Spec approved by the maintainer in session 2026-09-05 ("continue with corpus-seeded-mechanisms"); approvals recorded in spec.json. +- Task 1.1 (2026-09-05, RED first: 5 failing tests in `tests/test_variants.py`, `ModuleNotFoundError: openultrasast.semantic.variants`; then green): `semantic/variants.py` with `Shape` (language, sink_name, arity, source_positions, source_kinds, guard, mechanism; `key()`/`to_dict()`/`from_dict()` text-free), closed `GUARD_KINDS`/`SOURCE_KINDS`, `derive_shape(vuln_ir, fixed_ir, *, function, sink, line, mechanism, facts, vuln_text, fixed_text)`. Deviation from the design signature: the two excerpt texts are passed in because guard statements (`if x not in ALLOWED`, `if len(x) > n`) are not call sites or binds in `FileIR`; the guard kind is classified from the lines of the labeled function that exist only on the fixed side, closed set, `none` when unknown; nothing of that text enters the shape. Source kinds trace argument identifiers through binds (fact-source pattern -> `fact_source`; subscript/`.get(`/request container -> `container_read`; parameter -> `parameter`), preferring fact_source, then parameter, then container_read. A constant-only sink call or a missing function/call site yields None. +- Task 1.2 (2026-09-05, RED first: 3 failing tests in `tests/test_mechanism_export.py` — `Mechanism` lacked `origin`, `append_from_pair` missing; then green): additive `Mechanism.origin/review_tier/pairs/shape/guard` with defaults so sandbox rows load unchanged; `MechanismStore.load()` folds the append-only log by id (latest row wins) so a re-seeded shape carries the extended `pairs`; `append_from_pair` uses a deterministic id `corpus:` so the improve lever can name records across machines; `corpus_mechanisms()` selects searchable rows. Existing `order_promotions` tests unchanged and green. +- Review round 1 (2026-09-05) APPROVED 1.1 and 1.2 (reviewer re-simulated RED, ran six mutations, verified the import boundary and that sandbox rows load unchanged; isolated suite 493 / 480 green, statics and gates clean). Suggestions carried into group 2: trim `_CONTAINER_READ` to the design's subscript/`.get(` definition so no source vocabulary lives outside facts; intersect line and sink in `_labeled_call`; `append_from_pair` keeps the highest tier and accumulates provenance tags when a shape is re-seeded; sync design §Shape to the implemented signature. - Seeding reads `PairCase.review_tier` from `pair-corpus-honesty` task 8.1; until it lands, sast and vibe-py rows default to `advisory`/`seeded` per that design and only vibe-py seeds. - Guard classification is heuristic and closed; `none` is allowed and reported, never used to demote. diff --git a/src/openultrasast/semantic/variants.py b/src/openultrasast/semantic/variants.py new file mode 100644 index 0000000..65629d6 --- /dev/null +++ b/src/openultrasast/semantic/variants.py @@ -0,0 +1,257 @@ +"""Structural mechanism shapes derived from labeled pairs (corpus-seeded-mechanisms, Req 2). + +A shape is the closed, text-free description of how a known vulnerable call looked: the +sink's trailing callee name, its arity, which argument positions carried source-derived +values and of what kind, and the kind of guard the fix added. Shapes are derived from +``FileIR`` (never from regex over code) and compared structurally, so a variant in another +project matches even when every identifier differs. Stdlib plus ``.ir``, ``.facts`` and +``.functions`` only; never imports the overlay, the scorer, or the store. +""" + +from __future__ import annotations + +import re +from collections.abc import Sequence +from dataclasses import dataclass + +from .facts import SemanticFacts +from .functions import named_ranges_from_ir +from .ir import Bind, CallSite, FileIR, FunctionIR + +GUARD_KINDS = ("null_test", "bounds_test", "allowlist_test", "auth_check", "parameterized_call", "type_change", "none") +SOURCE_KINDS = ("fact_source", "parameter", "container_read") +_MAX_BIND_DEPTH = 8 + +# Closed guard classification over the statements the fix added inside the labeled function. +# Order matters: the first matching kind wins. These classify the *fix*, they never detect. +_GUARD_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = ( + ("parameterized_call", re.compile(r"execute\w*\([^,]+,\s*[\(\[]|prepareStatement\(|\?\s*['\"]?\s*[,)]|\$\d\b|%s['\"]\s*,\s*[\(\[]")), + ( + "auth_check", + re.compile( + r"\b(request\.user|req\.user|current_user|request\.state|session\[|is_authenticated|authoriz|require_auth" + r"|login_required|verify_token|check_permission|owner_id\s*[!=]=)" + ), + ), + ( + "allowlist_test", + re.compile( + r"\bnot in\b|\bin\s*[\(\{\[]|\.includes\(|\.has\(|\bstartswith\(|\bstartsWith\(|\bendswith\(|\bendsWith\(" + r"|re\.(?:match|fullmatch)\(|\.test\(|allow_?list|whitelist|\bALLOWED\b|\bsafe_join\(|secure_filename\(" + r"|\brealpath\(|\bnormpath\(|\bresolve\(" + ), + ), + ("bounds_test", re.compile(r"\blen\(|\.length\b|\bsizeof\(|\bstrn?len\(|\bsnprintf\(|\bstrlcpy\(|\bmin\(|\bmax\(|\s(?:<|>|<=|>=)\s")), + ( + "null_test", + re.compile( + r"\bis None\b|\bis not None\b|[!=]= *NULL\b|[!=]==? *null\b|\bnullptr\b|\bif\s*\(\s*!\s*\w|\bif not \w+\s*:" + r"|\bif\s+\w+\s+is\s+None|\?\?|\?\." + ), + ), + ( + "type_change", + re.compile(r"\bint\(|\bparseInt\(|\bNumber\(|\(int\)|\(size_t\)|\(unsigned\b|\bstr\(|\.toString\(|\bstatic_cast<|\bBigInt\("), + ), +) +_CONTAINER_READ = re.compile( + r"\[[^\]]+\]|\.get\(|\.getParameter\(|\.param\(|\.query\b|\.body\b|\.params\b|\.args\b|\.form\b|\.headers\b|\.cookies\b" +) + + +@dataclass(frozen=True) +class Shape: + language: str + sink_name: str # trailing callee segment + arity: int + source_positions: tuple[int, ...] + source_kinds: tuple[str, ...] + guard: str # GUARD_KINDS + mechanism: str # closed vocabulary id from the pair label + + def key(self) -> str: + """Stable identity of the shape; carries no path, line, or literal.""" + positions = ",".join(str(p) for p in self.source_positions) + kinds = ",".join(self.source_kinds) + return f"{self.language}|{self.sink_name}/{self.arity}|pos={positions}|kinds={kinds}|guard={self.guard}|{self.mechanism}" + + def to_dict(self) -> dict[str, object]: + return { + "language": self.language, + "sink_name": self.sink_name, + "arity": self.arity, + "source_positions": list(self.source_positions), + "source_kinds": list(self.source_kinds), + "guard": self.guard, + "mechanism": self.mechanism, + } + + @classmethod + def from_dict(cls, payload: dict[str, object]) -> Shape: + positions = payload.get("source_positions") or [] + kinds = payload.get("source_kinds") or [] + if not isinstance(positions, list) or not isinstance(kinds, list): + raise ValueError("shape positions and kinds must be lists") + guard = str(payload.get("guard", "none")) + if guard not in GUARD_KINDS: + raise ValueError(f"unknown guard kind {guard!r}") + return cls( + language=str(payload.get("language", "")), + sink_name=str(payload.get("sink_name", "")), + arity=int(str(payload.get("arity", 0))), + source_positions=tuple(int(p) for p in positions), + source_kinds=tuple(str(k) for k in kinds), + guard=guard, + mechanism=str(payload.get("mechanism", "other")), + ) + + +def trailing_name(callee: str) -> str: + """``os.path.join`` -> ``join``; ``Runtime.getRuntime().exec`` -> ``exec``; keeps the last identifier segment.""" + tail = callee.rsplit(".", 1)[-1].rsplit("::", 1)[-1].rsplit("->", 1)[-1] + match = re.search(r"[A-Za-z_$][\w$]*\s*$", tail) + return match.group(0).strip() if match else tail + + +def derive_shape( + vuln: FileIR, + fixed: FileIR, + *, + function: str, + sink: str | None, + line: int | None, + mechanism: str, + facts: SemanticFacts, + vuln_text: str = "", + fixed_text: str = "", +) -> Shape | None: + """The shape of the labeled sink call in ``function`` on the vulnerable side, with the guard the fixed side added. + + None when either side failed to parse, the function or the labeled call site is not found, or no argument of the + call carries a source (a constant-only call teaches nothing). + """ + if not vuln.parse_ok or not fixed.parse_ok: + return None + target = _function_named(vuln, function, vuln_text) + if target is None: + return None + call = _labeled_call(target, sink, line) + if call is None: + return None + kinds = [_source_kind_of_argument(target, call, index, facts) for index in range(len(call.arg_texts))] + positions = tuple(index for index, kind in enumerate(kinds) if kind is not None) + if not positions: + return None + guard = classify_guard(vuln, fixed, function=function, vuln_text=vuln_text, fixed_text=fixed_text) + return Shape( + language=vuln.language, + sink_name=trailing_name(call.name), + arity=len(call.arg_texts), + source_positions=positions, + source_kinds=tuple(kind for kind in kinds if kind is not None), + guard=guard, + mechanism=mechanism, + ) + + +def classify_guard(vuln: FileIR, fixed: FileIR, *, function: str, vuln_text: str, fixed_text: str) -> str: + """Closed guard kind from the statements inside ``function`` that exist only on the fixed side (``none`` when unknown).""" + before = _function_lines(vuln, function, vuln_text) + after = _function_lines(fixed, function, fixed_text) + added = [line for line in after if line not in set(before)] + for kind, pattern in _GUARD_PATTERNS: + if any(pattern.search(line) for line in added): + return kind + return "none" + + +def source_kind_of_name(function: FunctionIR, name: str, facts: SemanticFacts, *, before_line: int, depth: int = 0) -> str | None: + """How ``name`` came to carry attacker data inside ``function``: a fact source, a parameter, or a container read.""" + if depth > _MAX_BIND_DEPTH: + return None + bind = _last_bind(function, name, before_line) + if bind is not None: + if _source_in(bind.value_text, facts): + return "fact_source" + if _CONTAINER_READ.search(bind.value_text): + return "container_read" + for inner in bind.names: + if inner == name: + continue + kind = source_kind_of_name(function, inner, facts, before_line=bind.line, depth=depth + 1) + if kind is not None: + return kind + if name in function.params: + return "parameter" + return None + + +def _source_kind_of_argument(function: FunctionIR, call: CallSite, index: int, facts: SemanticFacts) -> str | None: + text = call.arg_texts[index] + if _source_in(text, facts): + return "fact_source" + if call.arg_is_constant[index] and not call.arg_names[index]: + return None + if _CONTAINER_READ.search(text) and any(name in function.params for name in call.arg_names[index]): + return "container_read" + found: list[str] = [] + for name in call.arg_names[index]: + kind = source_kind_of_name(function, name, facts, before_line=call.line + 1) + if kind is not None: + found.append(kind) + for preferred in SOURCE_KINDS: + if preferred in found: + return preferred + return None + + +def _source_in(text: str, facts: SemanticFacts) -> bool: + return any(pattern and pattern in text for source in facts.sources for pattern in source.patterns) + + +def _last_bind(function: FunctionIR, name: str, before_line: int) -> Bind | None: + candidates = [bind for bind in function.binds if bind.name == name and bind.line < before_line] + return max(candidates, key=lambda bind: bind.line) if candidates else None + + +def _labeled_call(function: FunctionIR, sink: str | None, line: int | None) -> CallSite | None: + if line is not None: + by_line = [call for call in function.calls if call.line == line] + if by_line: + return by_line[0] + if sink: + wanted = trailing_name(sink) + for call in function.calls: + if trailing_name(call.name) == wanted: + return call + return None + + +def _function_named(ir: FileIR, function: str, text: str) -> FunctionIR | None: + for item in ir.functions: + if item.name == function: + return item + if text: # labels use declarator names; the parser may name the function differently (or ) + for name, start, _end in named_ranges_from_ir(ir, text): + if name == function: + for item in ir.functions: + if item.start_line == start: + return item + return None + + +def _function_lines(ir: FileIR, function: str, text: str) -> list[str]: + target = _function_named(ir, function, text) + if target is None or not text: + return [] + lines = text.splitlines() + body = lines[target.start_line - 1 : target.end_line] + return [line.strip() for line in body if line.strip()] + + +def shapes_by_sink(shapes: Sequence[Shape]) -> dict[tuple[str, str], list[Shape]]: + """Index for matching: (language, sink_name) -> shapes.""" + index: dict[tuple[str, str], list[Shape]] = {} + for shape in shapes: + index.setdefault((shape.language, shape.sink_name), []).append(shape) + return index diff --git a/tests/test_variants.py b/tests/test_variants.py new file mode 100644 index 0000000..bc72042 --- /dev/null +++ b/tests/test_variants.py @@ -0,0 +1,76 @@ +"""corpus-seeded-mechanisms: structural shapes derived from pairs and matched on trees (offline, stdlib ast).""" + +from __future__ import annotations + +from openultrasast.semantic.facts import load_facts +from openultrasast.semantic.ir import parse_file + +VULN = "from flask import request\nimport os\n\n\ndef run():\n cmd = request.args.get('cmd')\n os.system(cmd)\n" +FIXED = ( + "from flask import request\nimport os\n\nALLOWED = {'ls', 'date'}\n\n\ndef run():\n cmd = request.args.get('cmd')\n" + " if cmd not in ALLOWED:\n return None\n os.system(cmd)\n" +) + + +def _derive(vuln: str = VULN, fixed: str = FIXED, **overrides: object): + from openultrasast.semantic.variants import derive_shape + + kwargs: dict[str, object] = {"function": "run", "sink": "os.system", "line": None, "mechanism": "source_reaches_sink"} + kwargs.update(overrides) + return derive_shape( + parse_file("app.py", vuln, "python"), + parse_file("app.py", fixed, "python"), + vuln_text=vuln, + fixed_text=fixed, + facts=load_facts(), + **kwargs, + ) + + +def test_derive_shape_from_a_request_to_os_system_pair_with_an_allowlist_fix() -> None: + from openultrasast.semantic.variants import GUARD_KINDS, SOURCE_KINDS + + shape = _derive() + assert shape is not None + assert shape.language == "python" and shape.sink_name == "system" and shape.arity == 1 + assert shape.source_positions == (0,) and shape.source_kinds == ("fact_source",) + assert shape.guard == "allowlist_test" and shape.guard in GUARD_KINDS and shape.mechanism == "source_reaches_sink" + assert set(shape.source_kinds) <= set(SOURCE_KINDS) + + +def test_shape_key_is_stable_and_carries_no_path_line_or_literal() -> None: + a, b = _derive(), _derive() + assert a is not None and b is not None and a.key() == b.key() + key = a.key() + for forbidden in ("app.py", "cmd", "'ls'", "request.args", ":6", ":7"): + assert forbidden not in key + # the same code in another file with different variable names is the same shape + other_vuln = VULN.replace("cmd", "command") + other = _derive(other_vuln, FIXED.replace("cmd", "command")) + assert other is not None and other.key() == key + + +def test_parameter_and_container_read_sources_are_classified() -> None: + vuln = "import os\n\n\ndef run(payload, cmd):\n target = payload['t']\n os.system(cmd + target)\n" + fixed = ( + "import os\n\n\ndef run(payload, cmd):\n target = payload['t']\n if len(target) > 32:\n return None\n" + " os.system(cmd + target)\n" + ) + shape = _derive(vuln, fixed) + assert shape is not None + assert shape.source_positions == (0,) + assert shape.source_kinds in {("parameter",), ("container_read",)} # one argument carrying both; the kind is the first found + assert shape.guard == "bounds_test" + + +def test_unlabeled_or_missing_sink_returns_none_and_guard_defaults_to_none() -> None: + assert _derive(sink="subprocess.run") is None # the label names a sink the function never calls + assert _derive(function="other") is None # no such function + same = _derive(VULN, VULN) + assert same is not None and same.guard == "none" # no statement differs on the fixed side + constant = "import os\n\n\ndef run():\n os.system('ls')\n" + assert _derive(constant, constant) is None # a constant argument carries no source: nothing to learn + + +def test_parse_failed_side_yields_no_shape() -> None: + assert _derive("def run(:\n", FIXED) is None