From 03b114cbeb6e8d7509deb329908e480c8a51b044 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Wed, 14 Oct 2026 19:01:07 +0200 Subject: [PATCH] feat(prover): delivery D: control flow decides dominance; the trace carries each guard's relation, effect and the operation's control context Opus agent delivery under spec access-control-prover, requirement 10. `semantic/cfg.py` builds a control-flow graph per function from the syntax tree (if/else, switch arms, loops, try/catch/finally, return, throw, break, continue, exit, assert) with dominators and a per-line control context, and leaves a function without a graph rather than a wrong one. `obligations/cfg_dominance.py::CfgDominance` discharges a statement guard only when it lies on every path to the operation and the operation is unreachable from its failing arm; a predicate whose result nothing consumes is `value_only` and never discharges; router, decorator and earlier-hop witnesses behave as before; without a graph the order rule applies and `cfg_unavailable` is recorded. The trace's guards carry `relation` and `effect`, the trace carries `operation_context`; Java parameter lists carry names only; a handler-named operation's arguments are the handler's parameters. The checker's default stays order-based. Smoke on the 23 fix changes: 15 caught, 55 of 93 locations, precision 15 of 15, no failure, no change over 120 s (unchanged from delivery B). Gate: 3575 tests pass; the reserved-repository guard passes on the export. Co-Authored-By: Claude Fable 5.1 --- .kiro/specs/access-control-prover/design.md | 6 +- .kiro/specs/access-control-prover/tasks.md | 4 +- src/openultrasast/hunt/prover.py | 68 +- src/openultrasast/semantic/cfg.py | 762 ++++++++++++++++++ src/openultrasast/semantic/cst.py | 37 + .../semantic/obligations/cfg_dominance.py | 157 ++++ .../semantic/obligations/trace.py | 28 +- tests/test_hunt_prover.py | 69 ++ tests/test_mapping_frameworks.py | 15 + tests/test_obligation_cfg_dominance.py | 113 +++ tests/test_obligation_trace.py | 60 ++ tests/test_semantic_cfg.py | 245 ++++++ 12 files changed, 1546 insertions(+), 18 deletions(-) create mode 100644 src/openultrasast/semantic/cfg.py create mode 100644 src/openultrasast/semantic/obligations/cfg_dominance.py create mode 100644 tests/test_obligation_cfg_dominance.py create mode 100644 tests/test_semantic_cfg.py diff --git a/.kiro/specs/access-control-prover/design.md b/.kiro/specs/access-control-prover/design.md index 2fee8d5..ab16b52 100644 --- a/.kiro/specs/access-control-prover/design.md +++ b/.kiro/specs/access-control-prover/design.md @@ -233,7 +233,11 @@ exit consumes before the operation is `value_only` and never discharges; router dominate as before; along a path record, a witness on an earlier hop dominates as before. Without a graph for a function, `CfgDominance` falls back to `OrderDominance` and records `cfg_unavailable`. The trace's `Guard` gains `relation` (`all_paths`, `some_paths`, `none`, `after`) and `effect` -(`throws`, `returns`, `exits`, `value_only`); the trace gains `operation_context: ControlContext`. +(`throws`, `returns`, `exits`, `branches`, `value_only`; `constrains` for an identity constraint or a +non-permissive value on the operation's own line, which has no failing arm; router and decorator witnesses carry +`exits`); the trace gains `operation_context: ControlContext`. Coarse spots accepted for this increment: a +ternary or switch expression is one node, several statements on one line share it, a guard's result held in a +variable is followed by name without reassignment tracking. ### obligations.enforcement ```python diff --git a/.kiro/specs/access-control-prover/tasks.md b/.kiro/specs/access-control-prover/tasks.md index 5693152..0113ae0 100644 --- a/.kiro/specs/access-control-prover/tasks.md +++ b/.kiro/specs/access-control-prover/tasks.md @@ -80,7 +80,7 @@ - _Requirements: 9.1, 9.2, 9.5, 9.6_ - _Depends: 4.2_ -- [ ] 4.4 `semantic/cfg.py`: structural control-flow graph per function from the syntax tree, dominators, control context (P) +- [x] 4.4 `semantic/cfg.py`: structural control-flow graph per function from the syntax tree, dominators, control context (P) - Java, PHP, Python, JavaScript and TypeScript constructs of requirement 10.1; `cfg_unavailable` otherwise - Done when `tests/test_semantic_cfg.py` shows, per language, that a guard before an unconditional return dominates, a guard inside an if-branch does not dominate a statement after the branch, a guard inside a @@ -88,7 +88,7 @@ conditions and loops - _Requirements: 10.1, 10.2, 10.4_ - _Boundary: semantic.cfg_ -- [ ] 4.5 `obligations/cfg_dominance.py::CfgDominance`, failure effects, and the trace's control components; the prover selects it; Java parameter names and handler-operation arguments cleaned +- [x] 4.5 `obligations/cfg_dominance.py::CfgDominance`, failure effects, and the trace's control components; the prover selects it; Java parameter names and handler-operation arguments cleaned - Done when `tests/test_obligation_cfg_dominance.py` shows a value-only predicate does not discharge while the same predicate guarding a throw does; a branch-only guard yields `some_paths` and no discharge; the prover smoke on the 23 fixes keeps or raises its recall; `tests/test_mapping_frameworks.py` shows Java parameter diff --git a/src/openultrasast/hunt/prover.py b/src/openultrasast/hunt/prover.py index 4033e7b..69cb782 100644 --- a/src/openultrasast/hunt/prover.py +++ b/src/openultrasast/hunt/prover.py @@ -4,17 +4,19 @@ from __future__ import annotations import re from collections import Counter -from collections.abc import Mapping +from collections.abc import Iterator, Mapping from dataclasses import asdict, dataclass, replace from pathlib import Path from typing import Any from .. import mapping, source from ..preprocess import FileTarget, count_loc, detect_language, detect_tags +from ..semantic.cfg import ControlContext, FunctionCfg +from ..semantic.cst import parse_function_nodes from ..semantic.facts import load_facts from ..semantic.ir import FileIR, FunctionIR, trailing_name +from ..semantic.obligations.cfg_dominance import CfgDominance from ..semantic.obligations.check import _on_caller_hop, _relevant, _route_path, check_obligations -from ..semantic.obligations.dominance import OrderDominance from ..semantic.obligations.enforcement import witnesses from ..semantic.obligations.facts import load_obligation_facts from ..semantic.obligations.operations import Discharge, Operation, find_discharges, find_operations, valid_witness @@ -215,7 +217,8 @@ def prove_change(root: Path, change: Mapping[str, Any], ctx: Context | None = No for op in ops: if op.fact_id in handler_ids: handler_ends[op] = returns[-1] if returns else fn.end_line + 1 - dominance = _HandlerDominance(ctx.texts, handler_ends) + graphs = CfgDominance(ctx.texts, _Graphs(irs)) + dominance = _HandlerDominance(graphs, handler_ends) extra_witnesses: list[Discharge] = [] for language in ("php", "python"): try: @@ -325,6 +328,7 @@ def prove_change(root: Path, change: Mapping[str, Any], ctx: Context | None = No trace=pair_traces[op], ), ) + degradations.extend(graphs.degradations) coverage = Coverage( dict(languages), dict(frameworks), @@ -339,16 +343,60 @@ def prove_change(root: Path, change: Mapping[str, Any], ctx: Context | None = No class _HandlerDominance: - """Judge a handler's internal guard over its body while retaining its declaration as the finding location. + """Judge a handler's internal guard at the handler's end while retaining its declaration as the finding location. - Call operations retain ordinary order dominance. The checker default and OrderDominance are unchanged. + Call operations are judged as the inner dominance judges them. The checker default and OrderDominance are unchanged. """ - def __init__(self, texts: Mapping[str, str], handler_ends: Mapping[Operation, int]) -> None: - self.order = OrderDominance(texts) + def __init__(self, inner: CfgDominance, handler_ends: Mapping[Operation, int]) -> None: + self.inner = inner self.handler_ends = handler_ends - def dominates(self, witness: Discharge, operation: Operation, path: object | None) -> bool: + def _measured(self, operation: Operation) -> Operation: end = self.handler_ends.get(operation) - measured = replace(operation, line=end) if end is not None else operation - return self.order.dominates(witness, measured, path) + return replace(operation, line=end) if end is not None else operation + + def dominates(self, witness: Discharge, operation: Operation, path: object | None) -> bool: + return self.inner.dominates(witness, self._measured(operation), path) + + def relation(self, witness: Discharge, operation: Operation) -> str: + return self.inner.relation(witness, self._measured(operation)) + + def effect(self, witness: Discharge, operation: Operation) -> str: + return self.inner.effect(witness, self._measured(operation)) + + def context(self, operation: Operation) -> ControlContext: + return self.inner.context(self._measured(operation)) + + +class _Graphs(Mapping[tuple[str, str], tuple[FunctionCfg, ...]]): + """Control-flow graphs of the parsed files' functions, built on first use and keyed by (path, function).""" + + def __init__(self, irs: Mapping[str, tuple[FileIR, str]]) -> None: + self._irs = irs + self._built: dict[str, dict[str, tuple[FunctionCfg, ...]]] = {} + + def _file(self, path: str) -> dict[str, tuple[FunctionCfg, ...]]: + if path not in self._built: + graphs: dict[str, list[FunctionCfg]] = {} + if path in self._irs: + ir, text = self._irs[path] + nodes = parse_function_nodes(path, text, ir.language) + by_line: dict[int, list[Any]] = {} + for node in nodes: + by_line.setdefault(int(node.start_point[0]) + 1, []).append(node) + for fn in ir.functions: + found = [node for node in by_line.get(fn.start_line, ()) if int(node.end_point[0]) + 1 == fn.end_line] + if len(found) == 1: + graphs.setdefault(fn.name, []).append(FunctionCfg.from_cst(found[0], ir.language)) + self._built[path] = {name: tuple(items) for name, items in graphs.items()} + return self._built[path] + + def __getitem__(self, key: tuple[str, str]) -> tuple[FunctionCfg, ...]: + return self._file(key[0])[key[1]] + + def __iter__(self) -> Iterator[tuple[str, str]]: + return iter([(path, name) for path in self._irs for name in self._file(path)]) + + def __len__(self) -> int: + return sum(len(self._file(path)) for path in self._irs) diff --git a/src/openultrasast/semantic/cfg.py b/src/openultrasast/semantic/cfg.py new file mode 100644 index 0000000..300a888 --- /dev/null +++ b/src/openultrasast/semantic/cfg.py @@ -0,0 +1,762 @@ +"""A structural control-flow graph of one function, built from its syntax tree (access-control-prover, Req 10). + +`FunctionCfg.from_cst(node, language)` turns the statements of one function into nodes and edges: statement +sequences, if and else branches, switch arms, loops with their back edges, try, catch and finally, return, throw, +break, continue and calls that end the process. Every statement a construct contains is a node; an expression is +never split. The graph answers three questions by line: does one line dominate another (every path from the entry +to the second passes the first), how does a guard on one line relate to an operation on another, and under which +branch conditions, loops and try blocks a line sits. + +A construct the builder does not model (a labelled jump, a jump hidden inside an expression, a statement nested in +an expression) leaves the function without a graph: `available` is false and `reason` says why. The caller then +judges by order and records the degradation. A partial graph would be a wrong graph. + +Exceptions are modelled conservatively: any statement inside a try body, and the try's start, may transfer to each +enclosing catch (or finally). That adds paths and never removes one, so it can only weaken dominance. +""" + +from __future__ import annotations + +import re +from collections.abc import Iterator, Sequence +from contextlib import contextmanager +from dataclasses import dataclass, field +from typing import Any + +LANGUAGES = frozenset({"java", "php", "python", "javascript", "typescript", "tsx"}) +RELATIONS = ("all_paths", "some_paths", "none", "after") +EFFECTS = ("throws", "returns", "exits", "branches", "value_only") + +_COMMENTS = frozenset({"comment", "line_comment", "block_comment", "text_interpolation", "php_tag", "empty_statement"}) +_BLOCKS = frozenset({"block", "compound_statement", "statement_block", "colon_block", "constructor_body"}) +_IF = frozenset({"if_statement"}) +_WHILE = frozenset({"while_statement"}) +_FOR = frozenset({"for_statement", "enhanced_for_statement", "for_in_statement", "foreach_statement"}) +_DO = frozenset({"do_statement"}) +_SWITCH = frozenset({"switch_statement", "switch_expression", "match_statement"}) +_TRY = frozenset({"try_statement", "try_with_resources_statement"}) +_HEADED = frozenset({"with_statement", "synchronized_statement"}) +_LABELLED = frozenset({"labeled_statement"}) +_RETURN = frozenset({"return_statement"}) +_THROW = frozenset({"throw_statement", "raise_statement"}) +_BREAK = frozenset({"break_statement"}) +_CONTINUE = frozenset({"continue_statement"}) +_EXIT_STATEMENT = frozenset({"exit_statement"}) +_COMPOUND = _IF | _WHILE | _FOR | _DO | _TRY | _HEADED | _LABELLED | frozenset({"switch_statement", "match_statement"}) +# Jumps that would leave an atomic statement without an edge for them. +_HIDDEN_JUMP = _RETURN | _THROW | _BREAK | _CONTINUE | _EXIT_STATEMENT | frozenset({"throw_expression", "yield_statement"}) +# Nested functions and classes run elsewhere; their statements are not this function's control flow. +_NESTED = frozenset( + { + "function_definition", + "function_declaration", + "function_expression", + "generator_function_declaration", + "generator_function", + "method_declaration", + "method_definition", + "constructor_declaration", + "arrow_function", + "anonymous_function", + "lambda_expression", + "lambda", + "class_definition", + "class_declaration", + "class_body", + "class", + "interface_declaration", + "enum_declaration", + "declaration_list", + "anonymous_class", + } +) +# Calls that end the process or abort the request; a call is matched by its last name segment. +_EXIT_CALLS = frozenset({"exit", "die", "_exit", "abort"}) +_JUMPS = frozenset({"return", "throw", "break", "continue", "exit"}) + + +class _Unmodelled(Exception): + """A construct the builder does not model; the function gets no graph.""" + + +@dataclass(frozen=True) +class ControlContext: + """Where a line sits: the branch conditions above it in order, loops, try blocks and exception paths.""" + + conditions: tuple[str, ...] = () + in_loop: bool = False + in_try: bool = False + catches: tuple[str, ...] = () + on_exception_path: bool = False + + +@dataclass +class _Node: + kind: str # entry, end, statement, branch, case, try, catch, finally, or a jump: return, throw, break, continue, exit + start_line: int + end_line: int + start_byte: int + context: ControlContext + condition: str = "" + located: bool = True # False for synthetic nodes that no source line names + + +@dataclass +class FunctionCfg: + """Nodes, normal and exceptional edges, and dominators of one function.""" + + start_line: int + end_line: int + available: bool + reason: str | None = None + nodes: list[_Node] = field(default_factory=list) + normal: list[set[int]] = field(default_factory=list) + exceptional: list[set[int]] = field(default_factory=list) + _dominators: dict[int, int] = field(default_factory=dict, repr=False) + + ENTRY = 0 + EXIT = 1 + + @classmethod + def from_cst(cls, node: Any, language: str) -> FunctionCfg: + start, end = _line(node), _end_line(node) + if language not in LANGUAGES: + return cls(start, end, False, "no_control_flow_builder") + builder = _Builder(start, end) + try: + builder.build(node) + except _Unmodelled as error: + return cls(start, end, False, str(error) or "unmodelled_construct") + except RecursionError: + return cls(start, end, False, "nesting_too_deep") + cfg = cls(start, end, True, None, builder.nodes, builder.normal, builder.exceptional) + cfg._dominators = _dominators(cfg) + return cfg + + # Lines to nodes. + + def node_at(self, line: int) -> int | None: + """The node a line names: the narrowest statement or condition covering it, the earliest on a tie. + + A line after the function's last line names the exit, so a handler measured at its end is judged at the exit. + """ + if not self.available: + return None + if line > self.end_line: + return self.EXIT + best: tuple[int, int, int] | None = None + for index, node in enumerate(self.nodes): + if not node.located or not node.start_line <= line <= node.end_line: + continue + key = (node.end_line - node.start_line, node.start_byte, index) + if best is None or key < best: + best = key + return best[2] if best is not None else None + + # Questions by line. + + def dominates(self, a_line: int, b_line: int) -> bool: + a, b = self.node_at(a_line), self.node_at(b_line) + return a is not None and b is not None and self._dominates(a, b) + + def relation(self, guard_line: int, operation_line: int) -> str | None: + """`all_paths`, `some_paths`, `none` or `after`; None when a line names no node.""" + guard, operation = self.node_at(guard_line), self.node_at(operation_line) + if guard is None or operation is None: + return None + if guard_line > operation_line: + return "after" + if self._dominates(guard, operation): + return "all_paths" + if guard in self._dominators and self._reaches({guard}, operation, avoid=frozenset()): + return "some_paths" + return "none" + + def effect(self, guard_line: int, operation_line: int, *, throwing: bool) -> str | None: + """What happens to the operation when the guard fails: `throws`, `returns`, `exits`, `branches` or `value_only`. + + A guard in a branch condition counts when the operation cannot be reached from one arm of that branch; the + arm's own exit names the effect, and an arm that simply does not hold the operation is `branches`. A guard + whose value is assigned counts through a later branch on that variable which lies between it and the + operation. Otherwise a throwing guard (`throwing`) is `throws` and anything else is `value_only`. + """ + guard, operation = self.node_at(guard_line), self.node_at(operation_line) + if guard is None or operation is None: + return None + found = self._branch_effect(guard, operation) + if found is None: + variable = _assigned_name(self.nodes[guard].condition) + if variable: + pattern = re.compile(r"(? ControlContext: + node = self.node_at(line) + return self.nodes[node].context if node is not None else ControlContext() + + # Graph internals. + + def _dominates(self, a: int, b: int) -> bool: + mask = self._dominators.get(b) + return mask is not None and bool(mask >> a & 1) + + def _successors(self, node: int) -> set[int]: + return self.normal[node] | self.exceptional[node] + + def _reaches(self, starts: set[int], target: int, *, avoid: frozenset[int]) -> bool: + seen: set[int] = set() + stack = [node for node in starts if node not in avoid] + while stack: + node = stack.pop() + if node == target: + return True + if node in seen: + continue + seen.add(node) + stack.extend(next_node for next_node in self._successors(node) if next_node not in avoid and next_node not in seen) + return False + + def _branch_effect(self, branch: int, operation: int) -> str | None: + arms = sorted(self.normal[branch]) + if self.nodes[branch].kind != "branch" or len(arms) < 2 or branch == operation: + return None + avoid = frozenset({branch}) + failing = [arm for arm in arms if not self._reaches({arm}, operation, avoid=avoid)] + passing = [arm for arm in arms if arm not in failing] + if not failing or not passing: + return None + jumps, falls = self._arm_ends(failing[0], avoid) + if falls or not jumps: + # The failing arm carries on without the operation: the operation lies only inside the passing branch. + return "branches" + if jumps == {"throw"}: + return "throws" + if jumps & {"exit", "break", "continue"}: + return "exits" + return "returns" + + def _arm_ends(self, start: int, avoid: frozenset[int]) -> tuple[set[str], bool]: + """The jumps the arm ends in along normal edges, and whether it reaches the function's end without one.""" + jumps: set[str] = set() + falls = False + seen: set[int] = set() + stack = [start] + while stack: + node = stack.pop() + if node in seen or node in avoid: + continue + seen.add(node) + kind = self.nodes[node].kind + if kind in _JUMPS: + jumps.add(kind) + continue + if node == self.EXIT: + falls = True + continue + stack.extend(self.normal[node]) + return jumps, falls + + +def _dominators(cfg: FunctionCfg) -> dict[int, int]: + """Dominator sets as bit masks over the nodes reachable from the entry (iterative data flow).""" + order: list[int] = [] + seen: set[int] = set() + stack: list[tuple[int, bool]] = [(FunctionCfg.ENTRY, False)] + while stack: + node, done = stack.pop() + if done: + order.append(node) + continue + if node in seen: + continue + seen.add(node) + stack.append((node, True)) + for successor in sorted(cfg._successors(node), reverse=True): + if successor not in seen: + stack.append((successor, False)) + order.reverse() + predecessors: dict[int, list[int]] = {node: [] for node in order} + for node in order: + for successor in cfg._successors(node): + if successor in predecessors: + predecessors[successor].append(node) + everything = 0 + for node in order: + everything |= 1 << node + dom = {node: everything for node in order} + dom[FunctionCfg.ENTRY] = 1 << FunctionCfg.ENTRY + changed = True + while changed: + changed = False + for node in order[1:]: + mask = everything + for predecessor in predecessors[node]: + mask &= dom[predecessor] + mask |= 1 << node + if mask != dom[node]: + dom[node] = mask + changed = True + return dom + + +@dataclass +class _Frame: + kind: str # loop or switch + breaks: list[int] = field(default_factory=list) + continues: list[int] = field(default_factory=list) + + +class _Builder: + def __init__(self, start: int, end: int) -> None: + self.nodes: list[_Node] = [] + self.normal: list[set[int]] = [] + self.exceptional: list[set[int]] = [] + self.frames: list[_Frame] = [] + self.handlers: list[list[int]] = [] + self.conditions: list[str] = [] + self.loops = 0 + self.catches: list[str] = [] + self.tries = 0 + self.exception_paths = 0 + self._new("entry", start, start, -1, located=False) + self._new("end", end, end, -1, located=False) + + # Node creation and edges. + + def _new(self, kind: str, start: int, end: int, start_byte: int, *, condition: str = "", located: bool = True) -> int: + index = len(self.nodes) + self.nodes.append( + _Node( + kind, + start, + end, + start_byte, + ControlContext(tuple(self.conditions), self.loops > 0, self.tries > 0, tuple(self.catches), self.exception_paths > 0), + condition, + located, + ) + ) + self.normal.append(set()) + self.exceptional.append(set()) + if index > FunctionCfg.EXIT: + for targets in self.handlers: + self.exceptional[index].update(targets) + return index + + def _node(self, kind: str, syntax: Any, preds: Sequence[int], *, condition: str = "") -> int: + index = self._new(kind, _line(syntax), _end_line(syntax), int(syntax.start_byte), condition=condition) + self._link(preds, index) + return index + + def _link(self, preds: Sequence[int], target: int) -> None: + for pred in preds: + self.normal[pred].add(target) + + @contextmanager + def _condition(self, text: str) -> Iterator[None]: + self.conditions.append(text) + try: + yield + finally: + self.conditions.pop() + + # Construction. + + def build(self, function: Any) -> None: + body = function.child_by_field_name("body") + if body is None: + self._link([FunctionCfg.ENTRY], FunctionCfg.EXIT) + return + out = self._statement(body, [FunctionCfg.ENTRY]) if body.type in _BLOCKS else self._simple(body, [FunctionCfg.ENTRY]) + self._link(out, FunctionCfg.EXIT) + + def _sequence(self, statements: Sequence[Any], preds: list[int]) -> list[int]: + for statement in statements: + preds = self._statement(statement, preds) + return preds + + def _statement(self, node: Any, preds: list[int]) -> list[int]: + kind = node.type + if kind in _COMMENTS: + return preds + if kind in _BLOCKS: + return self._sequence(_statements(node), preds) + if kind in _IF: + return self._if(node, preds) + if kind in _WHILE or kind in _FOR: + return self._loop(node, preds) + if kind in _DO: + return self._do(node, preds) + if kind in _SWITCH: + return self._switch(node, preds) + if kind in _TRY: + return self._try(node, preds) + if kind in _HEADED: + body = node.child_by_field_name("body") + if body is None: + raise _Unmodelled("unmodelled_construct") + header_end = max(_line(node), _line(body) - 1) + header = self._new("statement", _line(node), header_end, int(node.start_byte), condition=_header_text(node, body)) + self._link(preds, header) + return self._statement(body, [header]) + if kind in _LABELLED: + inner = [child for child in node.named_children if child.type not in {"identifier", "statement_identifier"}] + return self._sequence(inner, preds) + return self._simple(node, preds) + + def _simple(self, node: Any, preds: list[int]) -> list[int]: + kind = node.type + thrown = kind == "expression_statement" and _only_child(node) == "throw_expression" + inner = _descendants(node.named_children[0]) if thrown else _descendants(node) + if any(child.type in _HIDDEN_JUMP or child.type in _COMPOUND for child in inner): + raise _Unmodelled("jump_inside_expression") + if kind in _RETURN: + self._link([self._node("return", node, preds)], FunctionCfg.EXIT) + return [] + if kind in _THROW or thrown: + index = self._node("throw", node, preds) + if not self.handlers: + self._link([index], FunctionCfg.EXIT) + return [] + if kind in _BREAK or kind in _CONTINUE: + if any(child.type not in _COMMENTS for child in node.named_children) and not _single_level(node): + raise _Unmodelled("labelled_jump") + frame = next((f for f in reversed(self.frames) if kind in _BREAK or f.kind == "loop"), None) + if frame is None: + raise _Unmodelled("jump_without_target") + index = self._node("break" if kind in _BREAK else "continue", node, preds) + (frame.breaks if kind in _BREAK else frame.continues).append(index) + return [] + if kind in _EXIT_STATEMENT or _is_exit_call(node): + self._link([self._node("exit", node, preds)], FunctionCfg.EXIT) + return [] + if kind == "assert_statement": + index = self._node("branch", node, preds, condition=_condition_text(node)) + failed = self._new("throw", _line(node), _line(node), int(node.start_byte), located=False) + self.normal[index].add(failed) + if not self.handlers: + self._link([failed], FunctionCfg.EXIT) + return [index] + return [self._node("statement", node, preds, condition=_text(node)[:240])] + + def _branch(self, condition: Any, fallback: Any, preds: list[int]) -> tuple[int, str]: + syntax = condition if condition is not None else fallback + text = _condition_text(condition) if condition is not None else "" + if condition is not None and any(child.type in _HIDDEN_JUMP or child.type in _COMPOUND for child in _descendants(condition)): + raise _Unmodelled("jump_inside_expression") + return self._node("branch", syntax, preds, condition=text), text + + def _if(self, node: Any, preds: list[int]) -> list[int]: + condition = node.child_by_field_name("condition") + then = node.child_by_field_name("consequence") or node.child_by_field_name("body") + branch, text = self._branch(condition, node, preds) + outs: list[int] = [] + with self._condition(text): + outs.extend(self._statement(then, [branch]) if then is not None else [branch]) + rest = [branch] + negated = [f"not ({text})"] + for alternative in node.children_by_field_name("alternative"): + pushed = list(negated) + self.conditions.extend(pushed) + try: + if alternative.type in _IF: + outs.extend(self._if(alternative, rest)) + rest = [] + elif alternative.type in {"elif_clause", "else_if_clause"}: + inner = alternative.child_by_field_name("condition") + body = alternative.child_by_field_name("consequence") or alternative.child_by_field_name("body") + inner_branch, inner_text = self._branch(inner, alternative, rest) + with self._condition(inner_text): + outs.extend(self._statement(body, [inner_branch]) if body is not None else [inner_branch]) + rest = [inner_branch] + negated.append(f"not ({inner_text})") + elif alternative.type == "else_clause": + body = alternative.child_by_field_name("body") + if body is None: + body = next((child for child in alternative.named_children if child.type not in _COMMENTS), None) + outs.extend(self._statement(body, rest) if body is not None else rest) + rest = [] + else: + # A grammar that names the else statement itself as the alternative. + outs.extend(self._statement(alternative, rest)) + rest = [] + finally: + del self.conditions[len(self.conditions) - len(pushed) :] + return outs + rest + + def _loop(self, node: Any, preds: list[int]) -> list[int]: + body = node.child_by_field_name("body") + condition = node.child_by_field_name("condition") + if body is None: + raise _Unmodelled("unmodelled_construct") + header_end = max(_line(node), _line(body) - 1) + text = _condition_text(condition) if condition is not None and condition.type != "empty_statement" else _header_text(node, body) + if condition is not None and any(child.type in _HIDDEN_JUMP or child.type in _COMPOUND for child in _descendants(condition)): + raise _Unmodelled("jump_inside_expression") + header = self._new("branch", _line(node), header_end, int(node.start_byte), condition=text) + self._link(preds, header) + frame = _Frame("loop") + self.frames.append(frame) + self.loops += 1 + try: + with self._condition(text): + out = self._statement(body, [header]) + finally: + self.loops -= 1 + self.frames.pop() + self._link(out + frame.continues, header) + outs = [header] + alternative = node.child_by_field_name("alternative") + if alternative is not None: + else_body = alternative.child_by_field_name("body") + outs = self._statement(else_body, outs) if else_body is not None else outs + return outs + frame.breaks + + def _do(self, node: Any, preds: list[int]) -> list[int]: + body = node.child_by_field_name("body") + condition = node.child_by_field_name("condition") + if body is None or condition is None: + raise _Unmodelled("unmodelled_construct") + frame = _Frame("loop") + self.frames.append(frame) + self.loops += 1 + first = len(self.nodes) + try: + out = self._statement(body, preds) + finally: + self.loops -= 1 + self.frames.pop() + branch, _ = self._branch(condition, node, out + frame.continues) + self.normal[branch].add(first if first < branch else branch) + return [branch, *frame.breaks] + + def _switch(self, node: Any, preds: list[int]) -> list[int]: + subject = node.child_by_field_name("condition") or node.child_by_field_name("value") or node.child_by_field_name("subject") + body = node.child_by_field_name("body") + if body is None: + raise _Unmodelled("unmodelled_construct") + branch, subject_text = self._branch(subject, node, preds) + frame = _Frame("switch") + self.frames.append(frame) + outs: list[int] = [] + falling: list[int] = [] + has_default = False + try: + for arm in body.named_children: + if arm.type in _COMMENTS: + continue + label, statements, default, falls_through = _arm(arm) + has_default = has_default or default + case = self._new("case", _line(arm), _line(arm), int(arm.start_byte), condition=label) + self._link([branch], case) + text = "default" if default else f"case {label} of {subject_text}" + with self._condition(text): + if falls_through: + out = self._sequence(statements, [case, *falling]) + falling = out + else: + outs.extend(self._sequence(statements, [case])) + finally: + self.frames.pop() + return outs + falling + frame.breaks + ([] if has_default else [branch]) + + def _try(self, node: Any, preds: list[int]) -> list[int]: + body = node.child_by_field_name("body") + if body is None: + raise _Unmodelled("unmodelled_construct") + clauses = [child for child in node.named_children if child.type in {"catch_clause", "except_clause", "except_group_clause"}] + handler = node.child_by_field_name("handler") + if handler is not None and handler not in clauses: + clauses.append(handler) + final = next((child for child in node.named_children if child.type == "finally_clause"), None) + orelse = next((child for child in node.named_children if child.type == "else_clause"), None) + start = self._new("try", _line(node), _line(node), int(node.start_byte)) + self._link(preds, start) + entries = [self._new("catch", _line(clause), _line(clause), int(clause.start_byte)) for clause in clauses] + final_entry = self._new("finally", _line(final), _line(final), int(final.start_byte)) if final is not None else None + targets = entries or ([final_entry] if final_entry is not None else []) + self.exceptional[start].update(targets) + types = [_catch_type(clause) for clause in clauses] + self.handlers.append(targets) + self.tries += 1 + self.catches.extend(types) + try: + resources = node.child_by_field_name("resources") + out = [start] + if resources is not None: + out = [self._node("statement", resources, out, condition=_text(resources))] + out = self._statement(body, out) + finally: + del self.catches[len(self.catches) - len(types) :] + self.tries -= 1 + self.handlers.pop() + if orelse is not None: + else_body = orelse.child_by_field_name("body") + out = self._statement(else_body, out) if else_body is not None else out + normal = list(out) + if final_entry is not None: + self.handlers.append([final_entry]) + self.exception_paths += 1 + try: + for clause, entry in zip(clauses, entries, strict=True): + clause_body = clause.child_by_field_name("body") or next( + (child for child in clause.named_children if child.type in _BLOCKS), None + ) + normal.extend(self._statement(clause_body, [entry]) if clause_body is not None else [entry]) + finally: + self.exception_paths -= 1 + if final_entry is not None: + self.handlers.pop() + if final_entry is None or final is None: + return normal + self._link(normal, final_entry) + final_body = final.child_by_field_name("body") or next((child for child in final.named_children if child.type in _BLOCKS), None) + final_out = self._statement(final_body, [final_entry]) if final_body is not None else [final_entry] + # An exception that entered the finally block leaves it again towards the enclosing handler or the exit. + for index in final_out: + if self.handlers: + for targets in self.handlers: + self.exceptional[index].update(targets) + else: + self.exceptional[index].add(FunctionCfg.EXIT) + return final_out + + +def _arm(arm: Any) -> tuple[str, list[Any], bool, bool]: + """Label text, statements, whether it is the default arm, and whether control falls into the next arm.""" + kind = arm.type + named = [child for child in arm.named_children if child.type not in _COMMENTS] + if kind == "switch_block_statement_group": + labels = [child for child in named if child.type == "switch_label"] + statements = [child for child in named if child.type != "switch_label"] + return _label(labels), statements, any(_text(label).strip().startswith("default") for label in labels), True + if kind == "switch_rule": + labels = [child for child in named if child.type == "switch_label"] + statements = [child for child in named if child.type != "switch_label"] + return _label(labels), statements, any(_text(label).strip().startswith("default") for label in labels), False + if kind in {"case_statement", "switch_case"}: + value = arm.child_by_field_name("value") or (named[0] if named else None) + statements = arm.children_by_field_name("body") if kind == "switch_case" else named[1:] + return _text(value) if value is not None else "", list(statements), False, True + if kind in {"default_statement", "switch_default"}: + statements = arm.children_by_field_name("body") if kind == "switch_default" else named + return "default", list(statements), True, True + if kind == "case_clause": + consequence = arm.child_by_field_name("consequence") + patterns = [child for child in named if child != consequence] + label = ", ".join(_flat(_text(child)) for child in patterns) + default = len(patterns) == 1 and _text(patterns[0]).strip() == "_" + return label, [consequence] if consequence is not None else [], default, False + raise _Unmodelled("unmodelled_construct") + + +def _label(labels: Sequence[Any]) -> str: + return ", ".join(_flat(re.sub(r"^\s*case\s+", "", _text(label)).rstrip(":").strip()) for label in labels) + + +def _catch_type(clause: Any) -> str: + typed = clause.child_by_field_name("type") + if typed is not None: + return _flat(_text(typed)) + for child in clause.named_children: + if child.type == "catch_formal_parameter": + caught = next((item for item in child.named_children if item.type == "catch_type"), None) + if caught is not None: + return _flat(_text(caught)) + if clause.type in {"except_clause", "except_group_clause"}: + value = clause.child_by_field_name("value") + if value is None: + value = next((child for child in clause.named_children if child.type not in _BLOCKS and child.type not in _COMMENTS), None) + if value is not None: + caught = value.named_children[0] if value.type == "as_pattern" and value.named_children else value + return _flat(_text(caught)) + return "any" + + +def _statements(node: Any) -> list[Any]: + return [child for child in node.named_children if child.type not in _COMMENTS] + + +def _descendants(node: Any) -> Iterator[Any]: + for child in node.named_children: + if child.type in _NESTED: + continue + yield child + yield from _descendants(child) + + +def _only_child(node: Any) -> str: + named = [child for child in node.named_children if child.type not in _COMMENTS] + return named[0].type if len(named) == 1 else "" + + +def _single_level(node: Any) -> bool: + """`break 1;` names the innermost target, as a bare `break` does.""" + named = [child for child in node.named_children if child.type not in _COMMENTS] + return len(named) == 1 and named[0].type == "integer" and _text(named[0]).strip() == "1" + + +def _is_exit_call(node: Any) -> bool: + if node.type != "expression_statement": + return False + named = [child for child in node.named_children if child.type not in _COMMENTS] + if len(named) != 1 or named[0].type not in {"call", "call_expression", "method_invocation", "function_call_expression"}: + return False + callee = _text(named[0]).split("(", 1)[0] + return re.split(r"[.\\:>-]+", callee.strip())[-1] in _EXIT_CALLS + + +def _condition_text(node: Any) -> str: + while node.type == "parenthesized_expression" and len(node.named_children) == 1: + node = node.named_children[0] + if node.type == "assert_statement": + named = [child for child in node.named_children if child.type not in _COMMENTS] + return _flat(_text(named[0])) if named else "" + return _flat(_text(node)) + + +def _header_text(node: Any, body: Any) -> str: + header = _text(node)[: max(0, int(body.start_byte) - int(node.start_byte))] + return _flat(header.rstrip(" :{\n")) + + +def _assigned_name(text: str) -> str: + """The variable a statement assigns, when it assigns one plain name.""" + match = re.match(r"^\s*(?:(?:final|const|let|var|auto)\s+)?(?:[\w<>\[\],.?]+\s+)?(\$?[A-Za-z_]\w*)\s*(?::[^=]+)?=(?!=)", text) + return match.group(1) if match else "" + + +def _flat(text: str) -> str: + flat = " ".join(text.split()) + return flat if len(flat) <= 160 else flat[:157] + "..." + + +def _text(node: Any) -> str: + raw = getattr(node, "text", None) + if raw is None: + return "" + return raw.decode(errors="ignore") if isinstance(raw, bytes) else str(raw) + + +def _line(node: Any) -> int: + return int(node.start_point[0]) + 1 + + +def _end_line(node: Any) -> int: + return int(node.end_point[0]) + 1 + + +__all__ = ["EFFECTS", "LANGUAGES", "RELATIONS", "ControlContext", "FunctionCfg"] diff --git a/src/openultrasast/semantic/cst.py b/src/openultrasast/semantic/cst.py index 3c683f7..e8e92e5 100644 --- a/src/openultrasast/semantic/cst.py +++ b/src/openultrasast/semantic/cst.py @@ -88,6 +88,29 @@ def parse_with_cst(path: str, text: str, language: str) -> FileIR | None: return file_ir_from_tree(path=path, text=text, language=language, tree=tree) +def parse_function_nodes(path: str, text: str, language: str) -> list[Any]: + """The syntax nodes of every function in the file, in the order `file_ir_from_tree` lists them. + + The control-flow builder needs the statements of one function, which the FileIR does not keep. An unparsable + file or a language without a grammar yields no nodes. + """ + from .extra import grammar_for + + kinds = _KIND.get(language) + grammar = grammar_for(_grammar_language(language, path)) + if grammar is None or kinds is None: + return [] + try: + from tree_sitter import Parser + + tree = Parser(grammar).parse(text.encode()) + except Exception: + return [] + if bool(getattr(tree.root_node, "has_error", False)): + return [] + return [node for node in _walk(tree.root_node) if node.type in kinds["function"]] + + def _grammar_language(language: str, path: str) -> str: """TypeScript ships two grammars; a `.tsx` file needs the JSX-aware one. The FileIR keeps the language it was given.""" return "tsx" if language == "typescript" and path.endswith(".tsx") else language @@ -238,6 +261,9 @@ def _params(node: Any, kinds: dict[str, frozenset[str]]) -> tuple[str, ...]: return _identifiers(parameter) for child in _walk(node, skip=frozenset()): if child.type in kinds["params"]: + if any(param.type in {"formal_parameter", "spread_parameter", "receiver_parameter"} for param in child.named_children): + # Annotations such as `@QueryParameter` are identifiers too; keep the declared names only. + return tuple(name for param in child.named_children if (name := _java_parameter_name(param))) if any( param.type in {"simple_parameter", "variadic_parameter", "property_promotion_parameter"} for param in child.named_children ): @@ -250,6 +276,17 @@ def _params(node: Any, kinds: dict[str, frozenset[str]]) -> tuple[str, ...]: return () +def _java_parameter_name(param: Any) -> str: + if param.type == "formal_parameter": + name = param.child_by_field_name("name") + return _text(name) if name is not None else "" + if param.type == "spread_parameter": + declarator = next((child for child in param.named_children if child.type == "variable_declarator"), None) + name = declarator.child_by_field_name("name") if declarator is not None else None + return _text(name) if name is not None else "" + return "" + + def _left_name(node: Any) -> str: if node.type in {"identifier", "property_identifier", "variable_name"}: return _text(node) diff --git a/src/openultrasast/semantic/obligations/cfg_dominance.py b/src/openultrasast/semantic/obligations/cfg_dominance.py new file mode 100644 index 0000000..55df9b4 --- /dev/null +++ b/src/openultrasast/semantic/obligations/cfg_dominance.py @@ -0,0 +1,157 @@ +"""Control flow decides whether a discharge witness dominates an operation (access-control-prover, Req 10). + +`CfgDominance` implements the `Dominance` protocol over each function's structural control-flow graph. A statement +witness dominates when its node dominates the operation's node and the operation cannot be reached once the guard +fails: the failing arm throws, returns or exits, or the operation lies only inside the passing branch. A permission +predicate whose value nothing consumes before the operation (`value_only`) never dominates. Router and decorator +witnesses dominate the handler as a whole, and along a path record a witness on an earlier hop dominates, exactly +as `OrderDominance` judges them. A function without a graph is judged by `OrderDominance`, and the fallback is +recorded as a `cfg_unavailable` degradation. + +`relation`, `effect` and `context` describe a witness and an operation for the trace; `describe` gives the same +description for any `Dominance`, deriving it from order when the object has no graph. +""" + +from __future__ import annotations + +import re +from collections.abc import Mapping, Sequence +from typing import Any + +from ..cfg import ControlContext, FunctionCfg +from .dominance import Dominance, OrderDominance +from .operations import Discharge, Operation + +# Discharger kinds whose calls fail by throwing: `checkPermission`, `denyAccessUnlessGranted` and the like. +_THROWING = frozenset({"path_guard"}) +_SCOPED = frozenset({"decorator", "router"}) +_IN_OPERATION = frozenset({"identity_constraint", "non_permissive_value"}) +_EXIT_LINE = re.compile(r"^\s*(?Preturn\b|raise\b|throw\b|abort\(|sys\.exit\(|exit\(|res\.status\(|response\.status\()") + +CfgEntry = FunctionCfg | Sequence[FunctionCfg] | None + + +class CfgDominance: + """Dominance over control-flow graphs keyed by (path, function), with order as the recorded fallback.""" + + def __init__(self, texts: Mapping[str, str], cfgs: Mapping[tuple[str, str], CfgEntry]) -> None: + self.order = OrderDominance(texts) + self._texts = texts + self._cfgs = cfgs + self._unavailable: dict[tuple[str, str], str] = {} + + @property + def degradations(self) -> tuple[dict[str, object], ...]: + """One record per function judged by order instead of control flow.""" + return tuple( + {"stage": "dominance", "reason": "cfg_unavailable", "path": path, "function": function, "detail": detail} + for (path, function), detail in sorted(self._unavailable.items()) + ) + + def graph(self, path: str, function: str, line: int) -> FunctionCfg | None: + """The function's graph covering `line`, or None (recorded) when it has none.""" + entry = self._cfgs.get((path, function)) + candidates: Sequence[FunctionCfg] = (entry,) if isinstance(entry, FunctionCfg) else tuple(entry or ()) + matching = [cfg for cfg in candidates if cfg.start_line <= line <= cfg.end_line + 1] + cfg = min(matching, key=lambda item: item.end_line - item.start_line) if matching else None + if cfg is None or not cfg.available: + detail = "no_graph" if cfg is None else cfg.reason or "unmodelled_construct" + self._unavailable.setdefault((path, function), detail) + return None + return cfg + + def dominates(self, witness: Discharge, operation: Operation, path: object | None) -> bool: + if witness.scope == "hop" or (path is not None and (witness.path, witness.function) != (operation.path, operation.function)): + return self.order.dominates(witness, operation, path) + if (witness.path, witness.function) != (operation.path, operation.function): + return False + if witness.scope in _SCOPED: + return True + if witness.line is None: + return False + if witness.kind in _IN_OPERATION: + return witness.line == operation.line + if witness.line >= operation.line: + return False + cfg = self.graph(operation.path, operation.function, operation.line) + relation = cfg.relation(witness.line, operation.line) if cfg is not None else None + if cfg is None or relation is None: + if cfg is not None: + self._unavailable.setdefault((operation.path, operation.function), "line_without_node") + return self.order.dominates(witness, operation, path) + if relation != "all_paths": + return False + return cfg.effect(witness.line, operation.line, throwing=witness.kind in _THROWING) != "value_only" + + def relation(self, witness: Discharge, operation: Operation) -> str: + cfg = self._local(witness, operation) + found = cfg.relation(witness.line, operation.line) if cfg is not None and witness.line is not None else None + return found if found is not None else _order_relation(self.order, witness, operation) + + def effect(self, witness: Discharge, operation: Operation) -> str: + cfg = self._local(witness, operation) + found = ( + cfg.effect(witness.line, operation.line, throwing=witness.kind in _THROWING) + if cfg is not None and witness.line is not None + else None + ) + return found if found is not None else _order_effect(self._texts, witness, operation) + + def context(self, operation: Operation) -> ControlContext: + cfg = self.graph(operation.path, operation.function, operation.line) + return cfg.context(operation.line) if cfg is not None else ControlContext() + + def _local(self, witness: Discharge, operation: Operation) -> FunctionCfg | None: + if ( + witness.scope in _SCOPED + or witness.line is None + or witness.kind in _IN_OPERATION + or (witness.path, witness.function) != (operation.path, operation.function) + ): + return None + return self.graph(operation.path, operation.function, operation.line) + + +def describe(dominance: Dominance, witness: Discharge, operation: Operation, texts: Mapping[str, str]) -> tuple[str, str]: + """A witness's relation to an operation and its failure effect, from the graph when the dominance has one.""" + relation = getattr(dominance, "relation", None) + effect = getattr(dominance, "effect", None) + if callable(relation) and callable(effect): + return str(relation(witness, operation)), str(effect(witness, operation)) + return _order_relation(dominance, witness, operation), _order_effect(texts, witness, operation) + + +def operation_context(dominance: Dominance, operation: Operation) -> ControlContext: + context = getattr(dominance, "context", None) + found: Any = context(operation) if callable(context) else None + return found if isinstance(found, ControlContext) else ControlContext() + + +def _order_relation(dominance: Dominance, witness: Discharge, operation: Operation) -> str: + if dominance.dominates(witness, operation, None): + return "all_paths" + if witness.line is not None and witness.line > operation.line: + return "after" + return "none" + + +def _order_effect(texts: Mapping[str, str], witness: Discharge, operation: Operation) -> str: + """Without a graph: scoped witnesses end the request, throwing guards throw, others need an exit line between.""" + if witness.scope in _SCOPED: + return "exits" + if witness.kind in _IN_OPERATION: + return "constrains" + if witness.kind in _THROWING: + return "throws" + if witness.line is None or witness.line >= operation.line: + return "value_only" + lines = texts.get(witness.path, "").splitlines()[witness.line : operation.line - 1] + for line in lines: + match = _EXIT_LINE.match(line) + if match is not None: + word = match.group("word") + return "returns" if word.startswith("return") else "throws" if word.startswith(("raise", "throw")) else "exits" + return "value_only" + + +__all__ = ["CfgDominance", "describe", "operation_context"] diff --git a/src/openultrasast/semantic/obligations/trace.py b/src/openultrasast/semantic/obligations/trace.py index f41e1f9..768bf32 100644 --- a/src/openultrasast/semantic/obligations/trace.py +++ b/src/openultrasast/semantic/obligations/trace.py @@ -9,7 +9,9 @@ from collections.abc import Mapping, Sequence from dataclasses import asdict, dataclass, replace from typing import Any +from ..cfg import ControlContext from ..ir import FileIR, FunctionIR, trailing_name +from .cfg_dominance import describe, operation_context from .dominance import Dominance from .facts import ObligationFacts from .operations import Discharge, Operation, valid_witness @@ -24,6 +26,8 @@ class Guard: line: int | None provenance: str fact_id: str + relation: str # all_paths, some_paths, none or after, relative to the hop's call or the operation + effect: str # throws, returns, exits, branches or value_only: what failing the guard does to the operation @dataclass(frozen=True) @@ -49,6 +53,7 @@ class Trace: requires: tuple[str, ...] dominating: tuple[str, ...] missing: str | None + operation_context: ControlContext = ControlContext() def record(self) -> dict[str, Any]: """Return JSON values with keys sorted recursively, retaining sequence order.""" @@ -60,7 +65,8 @@ class Trace: record = self.record() components = [("entry", {key: record[key] for key in ("entry_kind", "boundary", "access", "parameters")})] components.extend(("hop", hop) for hop in record["hops"]) - components.append(("operation", {key: record[key] for key in ("operation_kind", "fact_id", "operation_line", "arguments")})) + operation_keys = ("operation_kind", "fact_id", "operation_line", "arguments", "operation_context") + components.append(("operation", {key: record[key] for key in operation_keys})) components.extend(("guard_after", guard) for guard in record["guards_after"]) components.append(("verdict", {key: record[key] for key in ("requires", "dominating", "missing")})) return "\n".join(f"{label} {json.dumps(value, sort_keys=True, separators=(',', ':'))}" for label, value in components) @@ -164,6 +170,7 @@ def build_trace( access: str | None = None, ) -> Trace: """Build one trace without changing the checker's verdict or coverage.""" + texts = {file: text for file, (_, text) in irs.items()} hops: list[Hop] = [] after_by_hop: list[list[Guard]] = [] incoming: tuple[str, ...] = () @@ -191,18 +198,28 @@ def build_trace( key=lambda w: (w.line or 0, w.kind, w.scope, w.provenance, w.fact_id), ) for witness in local: - guard = Guard(witness.kind, witness.scope, witness.line, witness.provenance, witness.fact_id) - if valid_witness(witness) and dominance.dominates(witness, measured, None): + relation, effect = describe(dominance, witness, measured, texts) + guard = Guard(witness.kind, witness.scope, witness.line, witness.provenance, witness.fact_id, relation, effect) + dominated = valid_witness(witness) and dominance.dominates(witness, measured, None) + if dominated or (valid_witness(witness) and witness.line is not None and line is not None and witness.line < line): + # A guard above the call that does not dominate it stays visible, with the relation that says why. before.append(guard) elif witness.line is not None and line is not None and witness.line > line: after.append(guard) + handler = False if final: - names = next(fact.calls for fact in scoped.operations if fact.id == operation.fact_id) + fact = next(fact for fact in scoped.operations if fact.id == operation.fact_id) + names = fact.calls + # A handler-named operation is the handler itself: its arguments are the handler's own parameters. + handler = bool(fact.handlers) and not fact.calls else: names = (path.hops[index + 1][0].rpartition("::")[2],) values = _arguments(function, line, names, text) if line is not None and names else () incoming = tuple(_origin(value, function, line or function.end_line + 1, origins, scoped) for value in values) - if final: + if handler: + own = tuple(param for param in function.params if param not in {"self", "cls", "this"}) + arguments = tuple((param, origins.get(param, "unknown")) for param in own) + elif final: arguments = tuple(zip(values, incoming, strict=True)) hops.append( Hop( @@ -226,4 +243,5 @@ def build_trace( operation.requires, tuple(sorted(dominating)), missing, + operation_context(dominance, operation), ) diff --git a/tests/test_hunt_prover.py b/tests/test_hunt_prover.py index b505274..77554a3 100644 --- a/tests/test_hunt_prover.py +++ b/tests/test_hunt_prover.py @@ -295,3 +295,72 @@ def test_remix_missing_kinds_and_public_policy(tmp_path, public, expected_missin change["files"] = [{"path": "app/routes/items.ts", "head_lines": [[1, 1]]}] result = prove_change(root, change) assert [finding["missing"] for finding in result.findings] == expected_missing + + +@pytest.mark.parametrize( + "guard,findings,relation,effect", + [ + ( + [" if (shared) {", " Jenkins.get().checkPermission(Jenkins.ADMINISTER);", " }"], + 1, + "some_paths", + "throws", + ), + ( + [" if (!Jenkins.get().hasPermission(Jenkins.ADMINISTER)) {", ' throw new AccessDeniedException("denied");', " }"], + 0, + "all_paths", + "throws", + ), + ([" boolean allowed = Jenkins.get().hasPermission(Jenkins.ADMINISTER);"], 1, "all_paths", "value_only"), + ], +) +def test_control_flow_decides_the_verdict_and_the_trace(tmp_path, guard, findings, relation, effect): + body = "\n ".join(guard) + root, change = repository( + tmp_path, + f"""class X extends Descriptor {{ + public Object doTestConnection(@QueryParameter boolean shared) {{ + {body} + return CredentialsProvider.lookupCredentials(Thing.class); + }} +}}""", + ) + result = prove_change(root, change) + assert len(result.findings) == findings + assert result.coverage.operations_discharged == 1 - findings + trace = result.traces[0] + assert trace["parameters"] == ["shared"] + assert [(g["relation"], g["effect"]) for g in trace["hops"][0]["guards_before_call"]] == [(relation, effect)] + assert not [item for item in result.degradations if item.get("reason") == "cfg_unavailable"] + if relation == "some_paths": + assert trace["missing"] == "path_guard" and trace["operation_context"]["conditions"] == [] + + +def test_handler_operation_trace_carries_handler_parameters(tmp_path): + root, change = repository( + tmp_path, + """class X extends Descriptor { + public Object doCheckValue(@QueryParameter String value, @AncestorInPath Item item) { + return 1; + } +}""", + ) + result = prove_change(root, change) + assert result.traces[0]["arguments"] == [["value", "request_input"], ["item", "request_input"]] + + +def test_unmodelled_function_is_judged_by_order_and_recorded(tmp_path): + root, change = repository( + tmp_path, + """class X extends Descriptor { + public Object doTestConnection() { + Jenkins.get().checkPermission(Jenkins.ADMINISTER); + outer: for (;;) { break outer; } + return CredentialsProvider.lookupCredentials(Thing.class); + } +}""", + ) + result = prove_change(root, change) + assert not result.findings and result.coverage.operations_discharged == 1 + assert any(item.get("reason") == "cfg_unavailable" and item.get("function") == "doTestConnection" for item in result.degradations) diff --git a/tests/test_mapping_frameworks.py b/tests/test_mapping_frameworks.py index eae1a99..7daaef5 100644 --- a/tests/test_mapping_frameworks.py +++ b/tests/test_mapping_frameworks.py @@ -239,3 +239,18 @@ export default function Items() { return
Items
; } }, ) assert [row.function_name for row in rows] == ["action"] + + +def test_java_parameters_are_names_without_annotations(): + from openultrasast.semantic.ir import parse_file + + text = """class Exposed extends Descriptor { + public Object doCheckUrl(@QueryParameter("url") final String value, @AncestorInPath Item item, StaplerRequest2 req, String... rest) { + return FormValidation.ok(); + } +} +""" + ir = parse_file("Fixture.java", text, "java") + if not ir.parse_ok: + pytest.skip("no Java grammar installed") + assert [fn.params for fn in ir.functions if fn.name == "doCheckUrl"] == [("value", "item", "req", "rest")] diff --git a/tests/test_obligation_cfg_dominance.py b/tests/test_obligation_cfg_dominance.py new file mode 100644 index 0000000..da41923 --- /dev/null +++ b/tests/test_obligation_cfg_dominance.py @@ -0,0 +1,113 @@ +"""CfgDominance: a guard counts only when control flow keeps the operation from running after it fails.""" + +import pytest + +from openultrasast.semantic.cfg import FunctionCfg +from openultrasast.semantic.cst import parse_function_nodes +from openultrasast.semantic.extra import grammar_for +from openultrasast.semantic.obligations.cfg_dominance import CfgDominance, describe, operation_context +from openultrasast.semantic.obligations.dominance import Dominance, OrderDominance +from openultrasast.semantic.obligations.operations import Discharge, Operation +from openultrasast.semantic.obligations.paths import PathRecord + +PATH = "T.java" + + +def setup(body, kind="ownership_check", marker="can("): + if grammar_for("java") is None: + pytest.skip("no Java grammar installed") + text = "class T {\n Object handle(Item item) {\n" + "".join(f" {line}\n" for line in body) + " }\n}\n" + assert len(text.encode()) > 0 + nodes = parse_function_nodes(PATH, text, "java") + assert nodes, "the parser read no function" + cfg = FunctionCfg.from_cst(nodes[0], "java") + lines = text.splitlines() + guard_line = next(number for number, line in enumerate(lines, 1) if marker in line) + op_line = next(number for number, line in enumerate(lines, 1) if "read(" in line) + witness = Discharge(PATH, guard_line, "handle", kind, "guard", "authenticated_context", "statement") + operation = Operation(PATH, op_line, "handle", "protected_read", "read", None, ("path_guard", "ownership_check")) + return CfgDominance({PATH: text}, {(PATH, "handle"): cfg}), witness, operation, text + + +def test_protocol(): + dominance, *_ = setup(["read(item);"], marker="read(") + assert isinstance(dominance, Dominance) + + +def test_value_only_predicate_does_not_discharge(): + dominance, witness, operation, _ = setup(["boolean ok = can(item);", "return read(item);"]) + assert dominance.relation(witness, operation) == "all_paths" + assert dominance.effect(witness, operation) == "value_only" + assert not dominance.dominates(witness, operation, None) + + +def test_same_predicate_guarding_a_throw_discharges(): + dominance, witness, operation, _ = setup(["if (!can(item)) {", " throw new AccessDeniedException();", "}", "return read(item);"]) + assert (dominance.relation(witness, operation), dominance.effect(witness, operation)) == ("all_paths", "throws") + assert dominance.dominates(witness, operation, None) + + +def test_operation_only_inside_passing_branch_discharges(): + dominance, witness, operation, _ = setup(["if (can(item)) {", " return read(item);", "}", "return null;"]) + assert dominance.effect(witness, operation) == "returns" + assert dominance.dominates(witness, operation, None) + dominance, witness, operation, _ = setup(["if (can(item)) {", " read(item);", "}"]) + assert dominance.effect(witness, operation) == "branches" + assert dominance.dominates(witness, operation, None) + assert operation_context(dominance, operation).conditions == ("can(item)",) + + +def test_branch_only_guard_is_some_paths_and_no_discharge(): + dominance, witness, operation, text = setup( + ["if (item.isShared()) {", " checkPermission(ADMINISTER);", "}", "return read(item);"], kind="path_guard", marker="checkPermission" + ) + assert dominance.relation(witness, operation) == "some_paths" + assert dominance.effect(witness, operation) == "throws" + assert not dominance.dominates(witness, operation, None) + # The order-based default accepts the same guard: it reads only that the guard comes first. + assert OrderDominance({PATH: text}).dominates(witness, operation, None) + + +def test_throwing_guard_in_swallowing_try_does_not_dominate(): + dominance, witness, operation, _ = setup( + ["try {", " checkPermission(ADMINISTER);", "} catch (AccessDeniedException e) {", " log(e);", "}", "return read(item);"], + kind="path_guard", + marker="checkPermission", + ) + assert not dominance.dominates(witness, operation, None) + + +def test_scopes_and_path_records_keep_their_semantics(): + dominance, witness, operation, _ = setup(["boolean ok = can(item);", "return read(item);"]) + for scope in ("decorator", "router"): + scoped = Discharge(PATH, None, "handle", "path_guard", "guard", "authenticated_context", scope) + assert dominance.dominates(scoped, operation, None) + assert describe(dominance, scoped, operation, {}) == ("all_paths", "exits") + caller = Discharge("Caller.java", 3, "entry", "path_guard", "guard", "authenticated_context", "statement") + record = PathRecord((("Caller.java::entry", 4), (f"{PATH}::handle", None)), "Caller.java::entry", True) + assert dominance.dominates(caller, operation, record) + assert not dominance.dominates(caller, operation, None) + later = PathRecord(((f"{PATH}::handle", 3), ("Caller.java::entry", None)), f"{PATH}::handle", True) + assert not dominance.dominates(caller, operation, later) + + +def test_missing_graph_falls_back_to_order_and_is_recorded(): + text = "class T {\n Object handle(Item item) {\n checkPermission(ADMINISTER);\n return read(item);\n }\n}\n" + witness = Discharge(PATH, 3, "handle", "path_guard", "guard", "authenticated_context", "statement") + operation = Operation(PATH, 4, "handle", "protected_read", "read", None, ("path_guard",)) + dominance = CfgDominance({PATH: text}, {}) + assert dominance.dominates(witness, operation, None) == OrderDominance({PATH: text}).dominates(witness, operation, None) + assert [(item["reason"], item["path"], item["function"]) for item in dominance.degradations] == [("cfg_unavailable", PATH, "handle")] + unavailable = FunctionCfg(1, 6, False, "labelled_jump") + dominance = CfgDominance({PATH: text}, {(PATH, "handle"): unavailable}) + assert dominance.dominates(witness, operation, None) + assert dominance.degradations[0]["detail"] == "labelled_jump" + assert not {"language", "framework", "repository"} & set(dominance.degradations[0]) + + +def test_order_description_without_graph(): + text = "def f():\n ok = can()\n if not ok:\n return\n read()\n" + witness = Discharge("a.py", 2, "f", "ownership_check", "guard", "authenticated_context", "statement") + operation = Operation("a.py", 5, "f", "protected_read", "read", None, ("ownership_check",)) + assert describe(OrderDominance({"a.py": text}), witness, operation, {"a.py": text}) == ("all_paths", "returns") + assert operation_context(OrderDominance({"a.py": text}), operation) == operation_context(OrderDominance({}), operation) diff --git a/tests/test_obligation_trace.py b/tests/test_obligation_trace.py index 9fefa0e..3b55ab2 100644 --- a/tests/test_obligation_trace.py +++ b/tests/test_obligation_trace.py @@ -196,3 +196,63 @@ def test_guards_after_follow_return_order(): missing="path_guard", ) assert [guard.fact_id for guard in result.guards_after] == ["final-after", "middle-after", "entry-after"] + + +def test_control_components_include_branch_guard_without_discharge(): + from openultrasast.semantic.cfg import FunctionCfg + from openultrasast.semantic.cst import parse_function_nodes + from openultrasast.semantic.obligations.cfg_dominance import CfgDominance + + text = "def entry(item):\n if enabled:\n guard()\n read(item)\n" + ir = parse_python_ast("app.py", text) + facts = ObligationFacts("1", (OperationFact("read", "protected_read", "python", ("read",), None, ("path_guard",), "high"),), ()) + nodes = parse_function_nodes("app.py", text, "python") + if not nodes: + import pytest + + pytest.skip("no Python grammar installed") + dominance = CfgDominance({"app.py": text}, {("app.py", "entry"): FunctionCfg.from_cst(nodes[0], "python")}) + result = build_trace( + path=PathRecord((("app.py::entry", None),), "app.py::entry", True), + entry=SimpleNamespace(kind="route", trust_boundary="http_request", access_level="unknown"), + operation=Operation("app.py", 4, "entry", "protected_read", "read", None, ("path_guard",)), + irs={"app.py": (ir, text)}, + facts=facts, + witnesses=(Discharge("app.py", 3, "entry", "path_guard", "guard", "authenticated_context", "statement"),), + dominance=dominance, + dominating=(), + missing="path_guard", + ) + guard = result.hops[0].guards_before_call[0] + assert (guard.relation, guard.effect) == ("some_paths", "throws") + assert result.operation_context.conditions == () and result.record()["operation_context"]["in_loop"] is False + assert '"relation":"some_paths"' in result.text() + assert '"operation_context":' in result.text().splitlines()[-2] + + +def test_order_dominance_trace_names_relation_and_effect(): + record = trace().record() + assert record["hops"][0]["guards_before_call"][0]["relation"] == "all_paths" + assert record["hops"][0]["guards_before_call"][0]["effect"] == "throws" + assert record["guards_after"][0]["relation"] == "after" + assert record["operation_context"] == {"catches": [], "conditions": [], "in_loop": False, "in_try": False, "on_exception_path": False} + + +def test_handler_operation_arguments_are_parameter_origins(): + text = "def entry(self, item):\n return item\n" + ir = parse_python_ast("app.py", text) + facts = ObligationFacts( + "1", (OperationFact("handler", "protected_read", "python", (), None, ("path_guard",), "high", handlers=("entry",)),), () + ) + result = build_trace( + path=PathRecord((("app.py::entry", None),), "app.py::entry", True), + entry=SimpleNamespace(kind="route", trust_boundary="http_request", access_level="unknown"), + operation=Operation("app.py", 1, "entry", "protected_read", "handler", None, ("path_guard",)), + irs={"app.py": (ir, text)}, + facts=facts, + witnesses=(), + dominance=OrderDominance({"app.py": text}), + dominating=(), + missing="path_guard", + ) + assert result.arguments == (("item", "request_input"),) diff --git a/tests/test_semantic_cfg.py b/tests/test_semantic_cfg.py new file mode 100644 index 0000000..998dedf --- /dev/null +++ b/tests/test_semantic_cfg.py @@ -0,0 +1,245 @@ +"""The structural control-flow graph decides dominance per language and reports a line's control context.""" + +import pytest + +from openultrasast.semantic.cfg import FunctionCfg +from openultrasast.semantic.cst import parse_function_nodes +from openultrasast.semantic.extra import grammar_for + +SOURCES = { + "java": ( + "T.java", + """class T { + Object f(int x) { + guardA(); + if (x > 0) { + guardB(); + } + for (int i = 0; i < x; i++) { + guardC(); + } + try { + guardD(); + } catch (IOException e) { + recoverE(); + } + return opF(); + } +} +""", + "x > 0", + "IOException", + ), + "php": ( + "t.php", + """ 0) { + guardB(); + } + for ($i = 0; $i < $x; $i++) { + guardC(); + } + try { + guardD(); + } catch (IOException $e) { + recoverE(); + } + return opF(); +} +""", + "$x > 0", + "IOException", + ), + "python": ( + "t.py", + """def f(x): + guardA() + if x > 0: + guardB() + for i in range(x): + guardC() + try: + guardD() + except IOError: + recoverE() + return opF() +""", + "x > 0", + "IOError", + ), + "javascript": ( + "t.js", + """function f(x) { + guardA(); + if (x > 0) { + guardB(); + } + for (let i = 0; i < x; i++) { + guardC(); + } + try { + guardD(); + } catch (e) { + recoverE(); + } + return opF(); +} +""", + "x > 0", + "any", + ), + "typescript": ( + "t.ts", + """function f(x: number): unknown { + guardA(); + if (x > 0) { + guardB(); + } + for (let i = 0; i < x; i++) { + guardC(); + } + try { + guardD(); + } catch (e) { + recoverE(); + } + return opF(); +} +""", + "x > 0", + "any", + ), +} + + +def _require(language): + if grammar_for(language) is None: + pytest.skip(f"no grammar installed for {language}") + + +def graph(language, path, text): + _require(language) + assert len(text.encode()) > 0 + nodes = parse_function_nodes(path, text, language) + assert nodes, "the parser read no function" + return FunctionCfg.from_cst(nodes[0], language) + + +def line_of(text, marker): + return next(number for number, line in enumerate(text.splitlines(), 1) if marker in line) + + +@pytest.mark.parametrize("language", sorted(SOURCES)) +def test_dominance_and_context_per_language(language): + path, text, condition, caught = SOURCES[language] + cfg = graph(language, path, text) + assert cfg.available, cfg.reason + at = {marker: line_of(text, marker) for marker in ("guardA", "guardB", "guardC", "guardD", "recoverE", "opF")} + # A guard before the unconditional return dominates it. + assert cfg.dominates(at["guardA"], at["opF"]) + assert cfg.relation(at["guardA"], at["opF"]) == "all_paths" + # A guard inside an if-branch does not dominate a statement after the branch. + assert not cfg.dominates(at["guardB"], at["opF"]) + assert cfg.relation(at["guardB"], at["opF"]) == "some_paths" + # A guard inside a loop body does not dominate a statement after the loop. + assert not cfg.dominates(at["guardC"], at["opF"]) + # A guard in a try whose catch falls through does not dominate a statement after the try. + assert not cfg.dominates(at["guardD"], at["opF"]) + assert cfg.relation(at["opF"], at["guardA"]) == "after" + assert cfg.context(at["guardB"]).conditions == (condition,) + assert cfg.context(at["guardC"]).in_loop and not cfg.context(at["opF"]).in_loop + try_context = cfg.context(at["guardD"]) + assert try_context.in_try and try_context.catches == (caught,) and not try_context.on_exception_path + assert cfg.context(at["recoverE"]).on_exception_path + assert any(node.kind == "catch" for node in cfg.nodes) + assert cfg.context(at["opF"]) == type(try_context)() + + +def python_graph(body): + text = "def f(x):\n" + "".join(f" {line}\n" for line in body) + return graph("python", "t.py", text), text + + +@pytest.mark.parametrize( + "body,effect", + [ + (["if not can(x):", " raise Denied()", "op(x)"], "throws"), + (["if not can(x):", " return None", "op(x)"], "returns"), + (["if can(x):", " op(x)"], "branches"), + (["if can(x):", " op(x)", "else:", " return None"], "returns"), + (["for item in x:", " if not can(item):", " continue", " op(item)"], "exits"), + (["if not can(x):", " sys.exit(1)", "op(x)"], "exits"), + (["allowed = can(x)", "if not allowed:", " return None", "op(x)"], "returns"), + (["allowed = can(x)", "op(x)"], "value_only"), + (["can(x)", "op(x)"], "value_only"), + (["assert can(x)", "op(x)"], "throws"), + ], +) +def test_failure_effect(body, effect): + cfg, text = python_graph(body) + assert cfg.available, cfg.reason + assert cfg.effect(line_of(text, "can("), line_of(text, "op("), throwing=False) == effect + + +def test_throwing_call_and_both_arms_reaching(): + cfg, text = python_graph(["check_permission(x)", "op(x)"]) + assert cfg.effect(line_of(text, "check_permission"), line_of(text, "op("), throwing=True) == "throws" + cfg, text = python_graph(["if can(x):", " log()", "op(x)"]) + assert cfg.effect(line_of(text, "can("), line_of(text, "op("), throwing=False) == "value_only" + + +def test_switch_arms_and_back_edges(): + text = """class T { + void f(int k) { + switch (k) { + case 1: + guardA(); + break; + default: + other(); + } + do { + guardB(); + } while (k > 0); + opC(); + } +} +""" + cfg = graph("java", "T.java", text) + assert cfg.available + assert not cfg.dominates(line_of(text, "guardA"), line_of(text, "opC")) + assert cfg.context(line_of(text, "guardA")).conditions == ("case 1 of k",) + # A do body runs at least once, so its first statement dominates what follows the loop. + assert cfg.dominates(line_of(text, "guardB"), line_of(text, "opC")) + assert cfg.context(line_of(text, "guardB")).in_loop + + +@pytest.mark.parametrize( + "language,path,text", + [ + ("java", "T.java", "class T { void f() { outer: for (;;) { for (;;) { break outer; } } } }"), + ("javascript", "t.js", "function f() { x = cond ? 1 : 2; outer: for (;;) { continue outer; } }"), + ("php", "t.php", "