diff --git a/pyproject.toml b/pyproject.toml index ebe7bbe..2d88260 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -33,6 +33,7 @@ semantic = [ "tree-sitter-c>=0.23", "tree-sitter-cpp>=0.23", "tree-sitter-java>=0.23", + "tree-sitter-php>=0.23", ] [project.scripts] diff --git a/src/openultrasast/learn/cli.py b/src/openultrasast/learn/cli.py index d0aff8b..e2b04a4 100644 --- a/src/openultrasast/learn/cli.py +++ b/src/openultrasast/learn/cli.py @@ -15,6 +15,14 @@ from typing import Any def add_commands(learn_sub: Any) -> None: + mechanisms = learn_sub.add_parser("mechanisms", help="deterministic candidate discharge facts (no admission)") + mechanisms_sub = mechanisms.add_subparsers(dest="mechanisms_command", required=True) + mine = mechanisms_sub.add_parser("mine", help="mine fix pairs; no model or fact-table writes") + mine.add_argument("--memory", help="the example store (default OUSAST_MEMORY)") + mine.add_argument("--out", type=Path, required=True, help="candidate JSONL under the results root, outside this repository") + mine.add_argument("--profile", choices=("static", "plane"), default="static") + mine.add_argument("--spec", type=Path, help="decision engine compile spec defining the fold seed/split") + mine.add_argument("--root", type=Path, default=Path("."), help="repository root for the counts-only measurement") memory = learn_sub.add_parser("memory", help="the decision engine's local memory of labelled examples") memory_sub = memory.add_subparsers(dest="memory_command", required=True) build = memory_sub.add_parser("build", help="label + feature record + excerpt -> example rows (no model)") @@ -97,6 +105,8 @@ def add_commands(learn_sub: Any) -> None: def run(args: argparse.Namespace) -> int: + if args.learn_command == "mechanisms": + return _mechanisms_mine(args) if args.learn_command == "audit-leaks": return _audit_leaks(args) if args.learn_command == "memory" and args.memory_command == "build": @@ -111,6 +121,25 @@ def run(args: argparse.Namespace) -> int: return 2 +def _mechanisms_mine(args: argparse.Namespace) -> int: + from ..plane.memory import MemoryStoreError, open_store + from ..plane.reconciler import results_root + from .mechanisms import mine_store, write_result + + try: + root = args.root.resolve() + out = args.out.resolve() + if out.is_relative_to(root) or not out.is_relative_to(results_root().resolve()): + raise ValueError("--out must be under the results root and outside the repository") + result = mine_store(open_store(args.memory), profile=args.profile, spec_path=args.spec) + write_result(result, out, root=root, results=results_root()) + except (ValueError, OSError, MemoryStoreError) as exc: + print(f"learn mechanisms mine: {exc}", file=sys.stderr) + return 2 + print(json.dumps(result.record, sort_keys=True)) + return 0 if result.record["pairs_mined"] else 2 + + def _experiment(args: argparse.Namespace) -> int: """register | units | run | analyse. Exit 2 on a refused manifest (unregistered, modified, uncommitted), 3 when a ceiling stops the run (``unfinished``: a rerun resumes from the outcome rows).""" diff --git a/src/openultrasast/learn/mechanisms.py b/src/openultrasast/learn/mechanisms.py new file mode 100644 index 0000000..5cfeb58 --- /dev/null +++ b/src/openultrasast/learn/mechanisms.py @@ -0,0 +1,612 @@ +"""Deterministic discharge proposals, never admitted facts. + +The shared tree-sitter IR supplies function/call identity. Its CST supplies bounded +expression containment and structured if branches; loops, exceptions, alias writes, +and unmodelled expressions fail closed. Parameters and fact-table sources are +potential taint, as in learn.slice. A return is an output boundary only for +output_encoding (a proposal, not a claim that every return renders HTML). + +Patterns use zero-based argument positions. Only >=2 independent repository +proposers survive. Each fold carries its own training evidence; global evidence +must never be used to admit a fact for a held-out fold. No fact table/store writes. +""" + +from __future__ import annotations + +import ast +import hashlib +import json +import re +from collections import Counter, defaultdict +from collections.abc import Callable, Mapping, Sequence +from dataclasses import dataclass +from pathlib import Path +from textwrap import dedent +from typing import Any + +from ..model.specs import taint_specs +from ..plane.memory import MemoryStore, repo_key +from ..semantic.cst import _KIND, _walk, file_ir_from_tree +from ..semantic.extra import grammar_for +from ..semantic.facts import load_facts +from ..semantic.ir import FileIR, trailing_name +from ..semantic.taint import _call_matches, _source_in +from .examples import Example +from .folds import Fold, compile_split, folds_digest, outer_folds + +RECORD = Path("benchmarks/measurements/2026-10-04-fix-mechanisms-mining/record.json") +SUPPORTED = frozenset({"python", "javascript", "typescript", "tsx", "php"}) +_NUMBERED = re.compile(r"^\s*(\d+)(?: (.*))?$") + + +@dataclass(frozen=True) +class Pair: + id: str + repository: str + group: str + language: str + family: str + vulnerable: str + fixed: str + source: str = "pairs" + frameworks: tuple[str, ...] = () + package: str = "" + example_ids: tuple[str, ...] = () + + +@dataclass +class MiningResult: + candidates: list[dict[str, Any]] + record: dict[str, Any] + + +class Unsupported(ValueError): + """A counted skip, never an empty successful analysis.""" + + +def _text(node: Any) -> str: + return bytes(node.text).decode("utf-8") + + +def _hash(value: str) -> str: + return hashlib.sha256(value.encode()).hexdigest() + + +def _canonical(value: Any) -> str: + return json.dumps(value, sort_keys=True, separators=(",", ":")) + + +def _parse(text: str, language: str) -> tuple[Any, FileIR]: + if language not in SUPPORTED: + raise Unsupported("unsupported_language") + grammar = grammar_for(language) + if grammar is None: + raise Unsupported("grammar_unavailable") + lines = text.split("\ndiff:\n", 1)[0].splitlines() + matches = [_NUMBERED.match(line) for line in lines] + if any(matches): + if not all(matches) or len({m[1] for m in matches if m}) != len(lines): + raise Unsupported("unparseable") + lines = [(m[2] or "") for m in matches if m] + code = dedent("\n".join(lines)) + if language == "php" and not code.lstrip().startswith(" str: + name = name.strip().replace("->", ".").replace("::", ".") + if "(" in name or not re.fullmatch(r"[\w$\\]+(?:\.[\w$]+)*", name): + raise Unsupported("unsupported_shape") + if name.split(".")[0] in variables or name.startswith("$"): + if "." not in name: + raise Unsupported("unsupported_shape") + return trailing_name(name) + return name.lstrip("\\") + + +@dataclass(frozen=True) +class Flow: + roots: frozenset[str] = frozenset() + patterns: frozenset[str] = frozenset() + + +def _union(flows: Sequence[Flow]) -> Flow: + return Flow(frozenset().union(*(f.roots for f in flows)), frozenset().union(*(f.patterns for f in flows))) + + +@dataclass +class Site: + callee: str + flow: Flow + arguments: dict[str, Any] + slot: tuple[int, ...] + guards: frozenset[str] + + +class Analysis: + """Small structured walker over the shipped IR's call sites, not a new parser.""" + + def __init__(self, language: str, family: str, function: Any, ir: Any) -> None: + self.language, self.family = language, family + self.kinds = _KIND[language] + self.calls = {call.order: call for call in ir.calls} + self.variables = set(ir.params) | {bind.name for bind in ir.binds} + self.states = {name: Flow(frozenset({name})) for name in ir.params} + self.sites: list[Site] = [] + self.facts = load_facts().for_language(language) + self.spec = taint_specs(language=language).get(family) + self.function = function + + def name(self, node: Any) -> str: + call = self.calls.get(node.end_byte) + if call is None: + raise Unsupported("unsupported_shape") + return _callee(call.name, self.variables) + + def args(self, node: Any) -> list[Any]: + args = node.child_by_field_name("arguments") + if args is None: + args = next((c for c in node.named_children if c.type in self.kinds["args"]), None) + return list(args.named_children) if args else [] + + def expr(self, node: Any, slot: tuple[int, ...], guards: frozenset[str]) -> Flow: + kind = node.type + children = list(node.named_children) + if kind in {"identifier", "variable_name"}: + return self.states.get(_text(node), Flow()) + if kind in {"string", "string_literal", "integer", "float", "number", "true", "false", "null", "none", "boolean"}: + return Flow() + if kind in self.kinds["call"]: + name = self.name(node) + args = self.args(node) + values = [self.expr(arg, slot + (i,), guards) for i, arg in enumerate(args)] + flow = _union(values) + if _source_in(self.calls[node.end_byte].name + "(", self.facts): + flow = Flow(flow.roots | {""}, flow.patterns) + arguments = _arguments(args) + self.sites.append(Site(name, flow, arguments, slot, guards)) + added = {_canonical({"callee": name, "argument": i}) for i, value in enumerate(values) if value.roots} + return Flow(flow.roots, flow.patterns | added) + if kind in {"attribute", "member_expression", "subscript", "subscript_expression", "member_access_expression"}: + flow = self.expr(children[0], slot, guards) if children else Flow() + if _source_in(_text(node), self.facts): + return Flow(flow.roots | {""}, flow.patterns) + return flow + if kind in {"keyword_argument", "argument", "parenthesized_expression", "unary_expression", "not_operator"}: + return self.expr(children[-1], slot, guards) if children else Flow() + if kind in { + "binary_expression", + "binary_operator", + "comparison_operator", + "list", + "tuple", + "array", + "array_creation_expression", + "array_element_initializer", + "pair", + "dictionary", + "object", + }: + return _union([self.expr(child, slot + (i,), guards) for i, child in enumerate(children)]) + if kind in {"property_identifier", "name", "comment", "string_content", "escape_sequence"}: + return Flow() + raise Unsupported("unsupported_shape") + + def predicate(self, node: Any, truth: bool) -> frozenset[str]: + while node.type == "parenthesized_expression": + node = node.named_children[0] + if node.type in {"not_operator", "unary_expression"}: + if not _text(node).lstrip().startswith(("not ", "!")): + raise Unsupported("unsupported_shape") + return self.predicate(node.named_children[-1], not truth) + if node.type not in self.kinds["call"]: + raise Unsupported("unsupported_shape") + name = self.name(node) + predicates = [] + for index, arg in enumerate(self.args(node)): + # Evaluating a condition is not an output path or a sanitizer. + value = self.expr(arg, (), frozenset()) + if value.roots: + predicates.append( + _canonical( + {"callee": name, "argument": index, "predicate": "truthy" if truth else "falsy", "_roots": sorted(value.roots)} + ) + ) + return frozenset(predicates) + + def block(self, node: Any, prefix: tuple[int, ...] = (), guards: frozenset[str] = frozenset()) -> bool: + children = list(node.named_children) if node.type in {"block", "statement_block", "compound_statement"} else [node] + for index, statement in enumerate(children): + slot = prefix + (index,) + kind = statement.type + if kind in self.kinds["function"] or kind in {"comment", "pass_statement"}: + continue + if kind == "if_statement": + condition = statement.child_by_field_name("condition") + consequence = statement.child_by_field_name("consequence") or statement.child_by_field_name("body") + alternative = statement.child_by_field_name("alternative") + if condition is None or consequence is None or alternative is not None: + raise Unsupported("unsupported_shape") + before = dict(self.states) + yes, no = self.predicate(condition, True), self.predicate(condition, False) + exits = self.block(consequence, slot, guards | yes) + # Branch writes need SSA/joins; no invented textual-order dataflow. + if self.states != before: + raise Unsupported("unsupported_shape") + if exits: + guards |= no + continue + if kind in {"return_statement", "echo_statement", "throw_statement", "raise_statement"}: + flow = _union([self.expr(c, slot, guards) for c in statement.named_children]) + if self.family == "output_encoding" and kind in {"return_statement", "echo_statement"}: + self.sites.append(Site("" if kind == "return_statement" else "echo", flow, {}, slot, guards)) + if kind != "echo_statement": + return True + continue + if kind in {"lexical_declaration", "variable_declaration", "expression_statement"}: + if self.block(statement.named_children[0], slot, guards): + return True + if len(statement.named_children) != 1: + raise Unsupported("unsupported_shape") + continue + if kind in self.kinds["assign"]: + left = statement.child_by_field_name("left") or statement.child_by_field_name("name") + right = statement.child_by_field_name("right") or statement.child_by_field_name("value") + if left is None or right is None or left.type not in {"identifier", "variable_name"}: + raise Unsupported("unsupported_shape") + if kind.startswith("augmented"): + raise Unsupported("unsupported_shape") + self.states[_text(left)] = self.expr(right, slot, guards) + continue + if kind in self.kinds["call"]: + self.expr(statement, slot, guards) + if self.name(statement) in {"abort", "exit", "die"}: + return True + continue + raise Unsupported("unsupported_shape") + return False + + def run(self) -> list[Site]: + body = self.function.child_by_field_name("body") + if body is None: + raise Unsupported("unsupported_shape") + self.block(body) + return self.sites + + def sink(self, site: Site) -> bool: + return (self.family == "output_encoding" and site.callee in {"", "echo"}) or bool( + self.spec and any(_call_matches(site.callee, sink) for sink in self.spec.sinks) + ) + + +def _literal(node: Any) -> tuple[bool, Any]: + text = _text(node) + if text.lower() in {"true", "false", "null", "none"}: + return True, {"true": True, "false": False, "null": None, "none": None}[text.lower()] + if node.type not in {"string", "string_literal", "integer", "number", "float"}: + return False, None + try: + value = ast.literal_eval(text) + except (ValueError, SyntaxError): + return False, None + return (True, value) if isinstance(value, str | int | float | bool) else (False, None) + + +def _arguments(args: list[Any]) -> dict[str, Any]: + result: dict[str, Any] = {} + for position, original in enumerate(args): + arg = original.named_children[-1] if original.type == "argument" else original + key = "" + if arg.type == "keyword_argument": + key = _text(arg.named_children[0]) + arg = arg.named_children[-1] + ok, value = _literal(arg) + if ok: + result[_canonical({"key": key} if key else {"argument": position, "key": key})] = value + elif arg.type in {"array_creation_expression", "dictionary", "object"}: + for entry in arg.named_children: + if entry.type not in {"array_element_initializer", "pair"} or len(entry.named_children) != 2: + continue + valid_key, key_value = _literal(entry.named_children[0]) + valid_value, value = _literal(entry.named_children[-1]) + if valid_key and isinstance(key_value, str) and valid_value: + result[_canonical({"argument": position, "key": key_value})] = value + return result + + +def _functions(tree: Any, ir: FileIR) -> dict[str, tuple[Any, Any]]: + nodes = [n for n in _walk(tree.root_node) if n.type in _KIND[ir.language]["function"]] + functions = list(ir.functions[1:]) + if not functions or len({f.name for f in functions}) != len(functions): + raise Unsupported("unsupported_shape") + return {f.name: (node, f) for node, f in zip(nodes, functions, strict=True)} + + +def _proposals(before: Analysis, after: Analysis) -> set[tuple[str, str]]: + old, new = before.run(), after.run() + result: set[tuple[str, str]] = set() + old_sinks = [s for s in old if before.sink(s) and s.flow.roots] + for sink in old_sinks: + # Ambiguous repeated operations are not aligned by textual similarity. + matches = [s for s in new if s.callee == sink.callee and s.flow.roots == sink.flow.roots] + if len([s for s in old_sinks if s.callee == sink.callee]) != 1: + raise Unsupported("ambiguous_sink") + if len(matches) == 1: + fixed = matches[0] + result.update(("sanitizer", p) for p in fixed.flow.patterns - sink.flow.patterns) + # Only predicates involving a value on this sink's path are retained below. + for encoded in fixed.guards - sink.guards: + predicate = json.loads(encoded) + roots = set(predicate.pop("_roots")) + if roots & fixed.flow.roots: + result.add(("guard", _canonical(predicate))) + for key, value in fixed.arguments.items(): + if key not in sink.arguments or _canonical(sink.arguments[key]) != _canonical(value): + result.add(("safe_argument", _canonical({"callee": fixed.callee, **json.loads(key), "value": value}))) + elif not matches: + replacements = [s for s in new if s.flow.roots == sink.flow.roots] + if len(replacements) == 1 and replacements[0].slot == sink.slot and replacements[0].callee != sink.callee: + result.add(("safe_alternative", _canonical({"callee": sink.callee, "replacement": replacements[0].callee}))) + elif replacements: + raise Unsupported("ambiguous_sink") + else: + raise Unsupported("ambiguous_sink") + return result + + +def _known(language: str, family: str, form: str, pattern: Mapping[str, Any]) -> bool: + spec = taint_specs(language=language).get(family) + if spec is None: + return False + name = str(pattern.get("replacement") if form == "safe_alternative" else pattern.get("callee")) + if form == "guard" and pattern.get("predicate") != "truthy": + return False + names = spec.guards if form == "guard" else spec.sanitizers + spec.quoted_sanitizers + # Shape sinks alone do not establish a particular keyword/literal restriction. + return form != "safe_argument" and any(_call_matches(name, known) for known in names) + + +def _project_specific(pattern: Mapping[str, Any], pair: Pair, local: set[str]) -> str | None: + names = [str(pattern.get(k, "")) for k in ("callee", "replacement")] + if any(trailing_name(n) in local for n in names if n): + return "local_definition" + identities = [pair.repository.rstrip("/").removesuffix(".git").rsplit("/", 1)[-1], pair.package] + compact = re.sub(r"[^a-z0-9]", "", _canonical(pattern).lower()) + if any((token := re.sub(r"[^a-z0-9]", "", identity.lower())) and token in compact for identity in identities): + return "project_identifier" + return None + + +def mine_pairs(pairs: Sequence[Pair], *, folds: Sequence[Fold] = ()) -> MiningResult: + skips: Counter[str] = Counter() + parsed: dict[int, tuple[dict[str, tuple[Any, Any]], dict[str, tuple[Any, Any]]]] = {} + definitions: dict[tuple[str, str], set[str]] = defaultdict(set) + for index, pair in enumerate(pairs): + try: + before = _functions(*_parse(pair.vulnerable, pair.language)) + after = _functions(*_parse(pair.fixed, pair.language)) + parsed[index] = before, after + definitions[repo_key(pair.repository), pair.language].update(before.keys() | after.keys()) + except Unsupported as exc: + skips[str(exc)] += 1 + proposals: dict[tuple[str, str, str, str], list[Pair]] = defaultdict(list) + mined = 0 + for index, (old, new) in parsed.items(): + pair = pairs[index] + try: + common = old.keys() & new.keys() + if not common: + raise Unsupported("unmatched_function") + found: set[tuple[str, str]] = set() + for name in sorted(common): + found |= _proposals(Analysis(pair.language, pair.family, *old[name]), Analysis(pair.language, pair.family, *new[name])) + mined += 1 + if not found: + skips["no_supported_change"] += 1 + for form, pattern in sorted(found): + reason = _project_specific(json.loads(pattern), pair, definitions[repo_key(pair.repository), pair.language]) + if reason: + skips[reason] += 1 + else: + proposals[pair.language, pair.family, form, pattern].append(pair) + except Unsupported as exc: + skips[str(exc)] += 1 + candidates = [] + cells: dict[str, dict[str, int]] = {} + tops: dict[str, list[dict[str, Any]]] = defaultdict(list) + for (language, family, form, encoded), evidence in sorted(proposals.items()): + repos = {_hash(repo_key(p.repository)) for p in evidence} + pattern = json.loads(encoded) + known = _known(language, family, form, pattern) + cell = cells.setdefault(f"{language}/{family}/{form}", {"total": 0, "at_least_two_repositories": 0, "known": 0}) + cell["total"] += 1 + cell["known"] += int(known) + if len(repos) < 2: + skips["fewer_than_two_repositories"] += 1 + continue + cell["at_least_two_repositories"] += 1 + candidate = { + "language": language, + "family": family, + "form": form, + "pattern": pattern, + "known": known, + "proposing_pairs": sorted({p.id for p in evidence}), + "proposing_examples": {p.id: sorted(p.example_ids) for p in sorted(evidence, key=lambda p: p.id)}, + "repository_count": len(repos), + "repository_ids": sorted(repos), + } + training = [] + for fold in folds: + held_repos = {repo_key(p.repository) for p in pairs if p.group in fold.eval_groups} + train = [ + p + for p in evidence + if p.group not in fold.eval_groups + and repo_key(p.repository) not in held_repos + and p.source != fold.held_source + and (not fold.held_framework or fold.held_framework not in p.frameworks) + ] + train_repos = sorted({_hash(repo_key(p.repository)) for p in train}) + training.append( + { + "fold": fold.name, + "repository_count": len(train_repos), + "repository_ids": train_repos, + "proposing_pairs": sorted({p.id for p in train}), + "supported": len(train_repos) >= 2, + "eligible": len(train_repos) >= 2 and not known and fold.status == "ok", + } + ) + candidate["folds"] = training + candidates.append(candidate) + tops[form].append({"language": language, "family": family, "pattern": pattern, "repository_count": len(repos)}) + return MiningResult( + candidates, + { + "pairs_read": len(pairs), + "pairs_mined": mined, + "skips": dict(sorted(skips.items())), + "candidates": cells, + "top_patterns": { + form: sorted(rows, key=lambda r: (-r["repository_count"], _canonical(r)))[:10] for form, rows in sorted(tops.items()) + }, + }, + ) + + +def pairs_from_rows( + rows: Sequence[Mapping[str, Any]], excerpt_text: Callable[[str], str | None], *, profile: str = "static" +) -> tuple[list[Pair], Counter[str], int]: + """Never pair on repository group alone. Legacy rows use an unambiguous candidate. + + Explicit pair/source_ref keys take precedence; repeated revisions/profiles and + missing sides are counted, not guessed. Delta rows duplicate pin evidence. + """ + from .labels import load_sources + + allowed = {source.id for source in load_sources().sources} + buckets: dict[tuple[str, str, str, str, str, str], list[Mapping[str, Any]]] = defaultdict(list) + skips: Counter[str] = Counter() + for row in rows: + if row.get("profile", "static") != profile or row.get("unit", "pin") != "pin": + continue + source = str(row.get("source", "")) + if source not in allowed or row.get("origin") == "user" or row.get("conditional"): + skips["excluded_source"] += 1 + continue + repo, group = str(row.get("repo") or ""), str(row.get("group") or "") + candidate = str(row.get("candidate") or "") + if not repo or not group or "/" not in repo or not candidate: + skips["missing_pair_identity"] += 1 + continue + explicit = str(row.get("pair_id") or row.get("pair") or row.get("source_ref") or "") + identity = explicit or candidate + key = (repo_key(repo), group, source, str(row.get("language", "")), str(row.get("family", "")), identity) + buckets[key].append(row) + pairs: list[Pair] = [] + for key, members in sorted(buckets.items()): + sides = {label: [r for r in members if r.get("label") == label] for label in (0, 1)} + if len(members) != 2 or any(len(side) != 1 for side in sides.values()): + skips["incomplete_pair" if any(not s for s in sides.values()) else "ambiguous_pair"] += 1 + continue + fixed, vulnerable = sides[0][0], sides[1][0] + if any(r.get("pin_role", role) != role for r, role in ((fixed, "fixed"), (vulnerable, "vulnerable"))): + skips["label_side_mismatch"] += 1 + continue + texts = [excerpt_text(str(r.get("excerpt_sha") or "")) for r in (vulnerable, fixed)] + if any(not text for text in texts): + skips["unreadable_excerpt"] += 1 + continue + repo, group, source, language, family, identity = key + if not any(vulnerable.get(k) for k in ("pair_id", "pair", "source_ref")): + identity = "legacy-" + _hash(_canonical(key)) + pairs.append( + Pair( + identity, + repo, + group, + language, + family, + texts[0] or "", + texts[1] or "", + source, + tuple(sorted(set(vulnerable.get("frameworks") or ()) | set(fixed.get("frameworks") or ()))), + str(vulnerable.get("package") or ""), + tuple(sorted(str(r["id"]) for r in members if r.get("id"))), + ) + ) + return pairs, skips, len(buckets) + + +def mine_store(store: MemoryStore, *, profile: str = "static", spec_path: Path | None = None) -> MiningResult: + from .compile import load_spec + from .labels import load_sources + + rows = [r.row for r in store.rows(kind="example")] + allowed = {source.id for source in load_sources().sources} + examples = [ + Example.from_row(row) + for row in rows + if row.get("profile") == profile and row.get("source") in allowed and row.get("origin") != "user" and not row.get("conditional") + ] + spec = load_spec(spec_path) if spec_path else load_spec() + split = compile_split(examples, fraction=float(spec["split"].get("compile_fraction", 0.25)), seed=spec.seed) + folds = outer_folds(examples, seed=spec.seed, exclude=split.groups) + read_bytes = 0 + blobs = 0 + + def excerpt_text(sha: str) -> str | None: + nonlocal read_bytes, blobs + if not re.fullmatch(r"[0-9a-f]{64}", sha): + return None + data = store.get_blob("excerpts", sha) + if data is None: + return None + read_bytes += len(data) + blobs += 1 + try: + return data.decode("utf-8") + except UnicodeDecodeError: + return None + + pairs, skips, pairs_read = pairs_from_rows(rows, excerpt_text, profile=profile) + result = mine_pairs(pairs, folds=folds) + combined = Counter(result.record["skips"]) + skips + result.record.update( + { + "rows_read": len(rows), + "pairs_read": pairs_read, + "complete_pairs": len(pairs), + "excerpt_blobs_read": blobs, + "excerpt_bytes_read": read_bytes, + "skips": dict(sorted(combined.items())), + "folds_digest": folds_digest(split, folds), + "compile_spec_sha256": spec.sha256, + "fold_seed": spec.seed, + "profile": profile, + "status": "mined" if result.record["pairs_mined"] else "no_mineable_pairs", + } + ) + return result + + +def write_result(result: MiningResult, out: Path, *, root: Path, results: Path) -> None: + """Counts-only checked-in record; candidate evidence must stay under results outside the repo.""" + out, root, results = out.resolve(), root.resolve(), results.resolve() + if out.is_relative_to(root) or not out.is_relative_to(results): + raise ValueError("--out must be under the results root and outside the repository") + record = root / RECORD + if not record.resolve().is_relative_to(root): + raise ValueError("measurement record must stay inside the repository") + out.parent.mkdir(parents=True, exist_ok=True) + out.write_text("".join(_canonical(candidate) + "\n" for candidate in result.candidates), encoding="utf-8") + record.parent.mkdir(parents=True, exist_ok=True) + record.write_text(json.dumps(result.record, indent=2, sort_keys=True) + "\n", encoding="utf-8") diff --git a/src/openultrasast/semantic/cst.py b/src/openultrasast/semantic/cst.py index b96b379..b1174c9 100644 --- a/src/openultrasast/semantic/cst.py +++ b/src/openultrasast/semantic/cst.py @@ -43,6 +43,14 @@ _KIND = { _KIND["cpp"] = _KIND["c"] _KIND["typescript"] = _KIND["javascript"] _KIND["tsx"] = _KIND["javascript"] +_KIND["php"] = { + "function": frozenset({"function_definition", "method_declaration"}), + "assign": frozenset({"assignment_expression", "augmented_assignment_expression"}), + "call": frozenset({"function_call_expression", "member_call_expression", "scoped_call_expression"}), + "args": frozenset({"arguments"}), + "string": frozenset({"string", "encapsed_string"}), + "params": frozenset({"formal_parameters"}), +} _CONST = frozenset( { @@ -183,6 +191,11 @@ def _call(node: Any, kinds: dict[str, frozenset[str]]) -> CallSite: def _callee_name(node: Any, args_node: Any) -> str: + if node.type in {"member_call_expression", "scoped_call_expression"}: + receiver = node.child_by_field_name("object") or node.child_by_field_name("scope") + name = node.child_by_field_name("name") + if receiver is not None and name is not None: + return f"{_text(receiver)}->{_text(name)}" idents = [ _text(child) for child in getattr(node, "named_children", ()) or () @@ -207,7 +220,7 @@ def _call_name(node: Any, kinds: dict[str, frozenset[str]]) -> str | None: def _function_name(node: Any) -> str: for child in getattr(node, "named_children", ()) or (): - if child.type == "identifier": + if child.type in {"identifier", "name"}: return _text(child) if child.type == "function_declarator": for grandchild in getattr(child, "named_children", ()) or (): @@ -240,7 +253,7 @@ def _left_name(node: Any) -> str: def _identifiers(node: Any) -> tuple[str, ...]: names: list[str] = [] for child in _walk(node): - if child.type == "identifier": + if child.type in {"identifier", "variable_name"}: names.append(_text(child)) return tuple(names) diff --git a/src/openultrasast/semantic/extra.py b/src/openultrasast/semantic/extra.py index 62df964..08092ca 100644 --- a/src/openultrasast/semantic/extra.py +++ b/src/openultrasast/semantic/extra.py @@ -17,6 +17,7 @@ _GRAMMAR_MODULES = { "c": ("tree_sitter_c", "language"), "cpp": ("tree_sitter_cpp", "language"), "java": ("tree_sitter_java", "language"), + "php": ("tree_sitter_php", "language_php"), } _PROBE_ENV = "OPENULTRASAST_TREE_SITTER_PROBE" _PROBE_OFF = frozenset({"0", "off", "false", "unavailable", "no"}) diff --git a/tests/test_learn_mechanisms.py b/tests/test_learn_mechanisms.py new file mode 100644 index 0000000..071e881 --- /dev/null +++ b/tests/test_learn_mechanisms.py @@ -0,0 +1,289 @@ +"""Authored fix pairs only: no store, models, containers, or population files.""" + +from __future__ import annotations + +import importlib.util +import json +from dataclasses import replace + +import pytest + +from openultrasast.learn.folds import Fold +from openultrasast.learn.mechanisms import Pair, mine_pairs + + +def pair(before: str, after: str, *, language: str = "python", family: str = "output_encoding", repo: str = "owner/alpha") -> Pair: + return Pair("pair-" + repo, repo, repo, language, family, before, after) + + +def repeated(p: Pair, **kwargs): + return mine_pairs([p, replace(p, id="pair-beta", repository="owner/beta", group="owner/beta")], **kwargs) + + +def test_learn_mechanisms_sanitizer_return_and_def_use(): + p = pair("def f(x):\n return x", "def f(x):\n y = html.escape(x)\n return y") + result = repeated(p) + assert [(c["form"], c["pattern"]) for c in result.candidates] == [("sanitizer", {"callee": "html.escape", "argument": 0})] + assert result.candidates[0]["repository_count"] == 2 + assert "owner/alpha" not in json.dumps(result.record) + assert result.record["pairs_mined"] == 2 + + +def test_learn_mechanisms_javascript(): + p = pair("function f(x) { res.send(x); }", "function f(x) { const y = escapeHtml(x); res.send(y); }", language="javascript") + assert repeated(p).candidates[0]["pattern"] == {"callee": "escapeHtml", "argument": 0} + + +def test_learn_mechanisms_guard_and_known(): + p = pair( + "def f(u):\n redirect(u)", + "def f(u):\n if not is_safe_url(u):\n abort()\n redirect(u)", + family="untrusted_destination", + ) + candidate = repeated(p).candidates[0] + assert candidate["form"] == "guard" + assert candidate["pattern"] == {"callee": "is_safe_url", "argument": 0, "predicate": "truthy"} + assert candidate["known"] is True + assert not any(f["eligible"] for f in candidate["folds"]) + + +@pytest.mark.skipif(importlib.util.find_spec("tree_sitter_php") is None, reason="PHP tree-sitter wheel unavailable") +def test_learn_mechanisms_php_safe_argument_and_echo(): + p = pair( + "function f($d) { return unserialize($d); }", + "function f($d) { return unserialize($d, ['allowed_classes' => false]); }", + language="php", + family="deserialization", + ) + assert repeated(p).candidates[0]["pattern"] == {"callee": "unserialize", "argument": 1, "key": "allowed_classes", "value": False} + assert repeated(p).candidates[0]["form"] == "safe_argument" + p = pair("function f($x) { echo $x; }", "function f($x) { echo htmlspecialchars($x); }", language="php") + assert repeated(p).candidates[0]["form"] == "sanitizer" + + +def test_learn_mechanisms_safe_alternative(): + p = pair("def f(x):\n return yaml.load(x)", "def f(x):\n return yaml.safe_load(x)", family="deserialization") + candidate = repeated(p).candidates[0] + assert candidate["form"] == "safe_alternative" + assert candidate["pattern"] == {"callee": "yaml.load", "replacement": "yaml.safe_load"} + assert candidate["known"] is True + + +def test_learn_mechanisms_keyword(): + p = pair("def f(x):\n subprocess.run(x, shell=True)", "def f(x):\n subprocess.run(x, shell=False)", family="injection") + result = repeated(p) + assert any(c["pattern"] == {"callee": "subprocess.run", "key": "shell", "value": False} for c in result.candidates) + + +def test_learn_mechanisms_one_repository_and_local_helpers_rejected(): + p = pair("def f(x):\n return x", "def f(x):\n return cleanse(x)") + assert mine_pairs([p]).candidates == [] + local = replace(p, fixed="def cleanse(x):\n return x\ndef f(x):\n return cleanse(x)") + result = repeated(local) + assert result.candidates == [] + assert result.record["skips"]["local_definition"] == 2 + assert repeated(replace(p, fixed="def f(x):\n return alpha_escape(x)")).candidates == [] + + +def test_learn_mechanisms_folds_exclude_proposers(): + p = pair("def f(x):\n return x", "def f(x):\n return cleanse(x)") + folds = [Fold("outer", "outer-0", frozenset({p.group})), Fold("outer", "outer-1", frozenset({"elsewhere"}))] + candidate = repeated(p, folds=folds).candidates[0] + assert [(f["repository_count"], f["eligible"]) for f in candidate["folds"]] == [(1, False), (2, True)] + assert candidate["folds"][0]["proposing_pairs"] == ["pair-beta"] + + +@pytest.mark.parametrize( + "fixed", + [ + "def f(x):\n cleanse(x)\n return x", + "def f(x):\n y = cleanse(x)\n y = 'constant'\n return y", + "def f(x):\n if is_safe_url(x):\n pass\n redirect(x)", + ], +) +def test_learn_mechanisms_unrelated_calls_are_not_candidates(fixed): + before = "def f(x):\n redirect(x)" if "redirect" in fixed else "def f(x):\n return x" + family = "untrusted_destination" if "redirect" in fixed else "output_encoding" + assert repeated(pair(before, fixed, family=family)).candidates == [] + + +def test_learn_mechanisms_skips_and_determinism(): + p = pair("def f(x):\n return x", "def f(x):\n return cleanse(x)") + assert repeated(p) == repeated(p) + assert repeated(replace(p, fixed="def broken(")).record["skips"]["unparseable"] == 2 + assert repeated(replace(p, language="ruby")).record["skips"]["unsupported_language"] == 2 + assert ( + repeated(replace(p, fixed="def f(x):\n for y in x:\n cleanse(y)\n return x")).record["skips"]["unsupported_shape"] == 2 + ) + + +def test_learn_mechanisms_guard_must_test_sink_value(): + p = pair( + "def f(x, u):\n redirect(u)", + "def f(x, u):\n if not check(x):\n abort()\n redirect(u)", + family="untrusted_destination", + ) + assert repeated(p).candidates == [] + + +def test_learn_mechanisms_existing_path_call_is_not_new(): + p = pair("def f(x):\n return html.escape(x)", "def f(x):\n y = html.escape(x)\n return y") + assert repeated(p).candidates == [] + + +def test_learn_mechanisms_missing_grammar_is_counted(monkeypatch): + monkeypatch.setattr("openultrasast.learn.mechanisms.grammar_for", lambda language: None) + result = repeated(pair("def f(x):\n return x", "def f(x):\n return cleanse(x)")) + assert result.candidates == [] + assert result.record["skips"] == {"grammar_unavailable": 2} + + +def test_learn_mechanisms_pair_identity_and_unread_inputs(): + from openultrasast.learn.mechanisms import pairs_from_rows + + base = { + "repo": "owner/alpha", + "group": "owner/alpha", + "source": "pairs", + "candidate": "app.py::f", + "language": "python", + "family": "output_encoding", + "profile": "static", + "unit": "pin", + } + rows = [{**base, "label": label, "excerpt_sha": str(label)} for label in (0, 1)] + pairs, skips, count = pairs_from_rows(rows, lambda sha: "def f(x):\n return x") + assert len(pairs) == count == 1 and not skips + assert pairs_from_rows(rows + [rows[0]], lambda sha: "code")[1] == {"ambiguous_pair": 1} + assert pairs_from_rows(rows[:1], lambda sha: "code")[1] == {"incomplete_pair": 1} + assert pairs_from_rows(rows, lambda sha: None)[1] == {"unreadable_excerpt": 1} + forbidden = [{**row, "source": "population-v3"} for row in rows] + assert pairs_from_rows(forbidden, lambda sha: pytest.fail("must not read excluded excerpts"))[1] == {"excluded_source": 2} + + +def memory_store(tmp_path): + from openultrasast.plane.memory import FileStore, row_id + + store = FileStore(tmp_path / "memory") + rows = [] + for repo in ("owner/alpha", "owner/beta", "owner/gamma"): + for label in (0, 1): + code = "def f(x):\n return x" if label else "def f(x):\n return cleanse(x)" + sha = store.put_blob("excerpts", code.encode()) + rows.append( + { + "id": row_id("example", "test", "test", f"{repo}:{label}"), + "kind": "example", + "repo": repo, + "pin": str(label) * 40, + "run": "test", + "task": "test", + "population": "pairs", + "split": "pairs", + "image": "host", + "candidate": "app.py::f", + "family": "output_encoding", + "language": "python", + "profile": "static", + "label": label, + "source": "pairs", + "group": repo, + "unit": "pin", + "x": {}, + "instruments": {}, + "excerpt_sha": sha, + } + ) + store.put_rows(rows) + return store + + +def test_learn_mechanisms_store_cli_and_privacy(tmp_path, monkeypatch): + from openultrasast.cli import main + from openultrasast.learn.mechanisms import RECORD, mine_store + + store = memory_store(tmp_path) + result = mine_store(store) + assert result.record["complete_pairs"] == 3 + assert result.record["excerpt_bytes_read"] > 0 + assert len(result.candidates) == 1 + assert len(result.candidates[0]["folds"]) == 5 + monkeypatch.setenv("OUSAST_RESULTS", str(tmp_path / "results")) + root = tmp_path / "repo" + out = tmp_path / "results" / "plane" / "mining" / "candidates.jsonl" + assert main(["learn", "mechanisms", "mine", "--memory", f"file://{tmp_path / 'memory'}", "--out", str(out), "--root", str(root)]) == 0 + record = (root / RECORD).read_text() + assert all(repo not in record for repo in ("alpha", "beta", "gamma")) + assert json.loads(out.read_text())["repository_count"] == 3 + assert main(["learn", "mechanisms", "mine", "--memory", f"file://{tmp_path / 'empty'}", "--out", str(out), "--root", str(root)]) == 2 + assert json.loads((root / RECORD).read_text())["status"] == "no_mineable_pairs" + + +def test_learn_mechanisms_output_location_rejected(tmp_path): + from openultrasast.learn.mechanisms import MiningResult, write_result + + root = tmp_path / "repo" + with pytest.raises(ValueError, match="outside"): + write_result(MiningResult([], {}), root / "candidate.jsonl", root=root, results=tmp_path) + assert not root.exists() + + +@pytest.mark.parametrize("signature,source", [("", "input()"), ("request", "request.get_json()")]) +def test_learn_mechanisms_source_calls(signature, source): + p = pair(f"def f({signature}):\n return {source}", f"def f({signature}):\n return html.escape({source})") + assert repeated(p).candidates[0]["pattern"] == {"callee": "html.escape", "argument": 0} + + +def test_learn_mechanisms_inserted_log_is_not_sink_replacement(): + p = pair("def f(x):\n yaml.load(x)", "def f(x):\n logger.info(x)\n yaml.safe_load(x)", family="deserialization") + result = repeated(p) + assert result.candidates == [] + assert result.record["skips"]["ambiguous_sink"] == 2 + + +def test_learn_mechanisms_keyword_reordering_is_not_change(): + p = pair( + "def f(x):\n subprocess.run(x, shell=False, timeout=10)", + "def f(x):\n subprocess.run(x, timeout=10, shell=False)", + family="injection", + ) + assert repeated(p).candidates == [] + + +def test_learn_mechanisms_legacy_pair_ids_are_repository_scoped(tmp_path): + from openultrasast.learn.mechanisms import mine_store + + result = mine_store(memory_store(tmp_path)) + candidate = result.candidates[0] + assert len(candidate["proposing_pairs"]) == 3 + assert all(len(ids) == 2 for ids in candidate["proposing_examples"].values()) + assert len({i for ids in candidate["proposing_examples"].values() for i in ids}) == 6 + + +def test_learn_mechanisms_runtime_callable_is_unsupported(): + p = pair("def f(x, clean):\n return x", "def f(x, clean):\n return clean(x)") + result = repeated(p) + assert result.candidates == [] + assert result.record["skips"]["unsupported_shape"] == 2 + + +def test_learn_mechanisms_language_support_is_not_pooled(): + p = pair("def f(x):\n return x", "def f(x):\n return cleanse(x)") + js = pair("function f(x) { return x; }", "function f(x) { return cleanse(x); }", language="javascript", repo="owner/beta") + assert mine_pairs([p, js]).candidates == [] + + +def test_learn_mechanisms_guard_inside_branch_and_numbered_excerpt(): + p = pair( + " 10 def f(x):\n 11 redirect(x)", + " 20 def f(x):\n 21 if is_safe_url(x):\n 22 redirect(x)", + family="untrusted_destination", + ) + candidate = repeated(p).candidates[0] + assert candidate["form"] == "guard" and candidate["pattern"]["predicate"] == "truthy" + + +@pytest.mark.skipif(importlib.util.find_spec("tree_sitter_php") is None, reason="PHP tree-sitter wheel unavailable") +def test_learn_mechanisms_php_receiver_normalization(): + p = pair("function f($x, $helper) { echo $x; }", "function f($x, $helper) { echo $helper->escape($x); }", language="php") + assert repeated(p).candidates[0]["pattern"] == {"callee": "escape", "argument": 0}