From 8ef2629e0b33ab380a3e0a24918c2c550c8cf0f9 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Thu, 29 Oct 2026 16:04:51 +0100 Subject: [PATCH] feat(access-control-prover): 8.3 exact, finite-candidate and unresolved call targets --- .kiro/specs/access-control-prover/tasks.md | 2 +- src/openultrasast/hunt/prover.py | 88 +++++-- .../semantic/obligations/resolution.py | 150 +++++++++++ tests/test_call_resolution.py | 247 ++++++++++++++++++ 4 files changed, 459 insertions(+), 28 deletions(-) create mode 100644 tests/test_call_resolution.py diff --git a/.kiro/specs/access-control-prover/tasks.md b/.kiro/specs/access-control-prover/tasks.md index 62d0a9a..57ba5f1 100644 --- a/.kiro/specs/access-control-prover/tasks.md +++ b/.kiro/specs/access-control-prover/tasks.md @@ -122,7 +122,7 @@ its final resolution, dominance, fact-acquisition and measurement semantics with - _Requirements: 3.2, 3.5, 6.2_ - _Boundary: Symbol Index_ - _Depends: 8.1_ -- [ ] 8.3 Resolve exact, finite-candidate and unresolved call targets +- [x] 8.3 Resolve exact, finite-candidate and unresolved call targets - Return complete exact targets, incomplete finite candidate sets, or unresolved results with reasons. - Remove global bare-name fallback from prover-mode resolution. - Done when ambiguity never merges methods and unresolved calls cannot produce complete paths. diff --git a/src/openultrasast/hunt/prover.py b/src/openultrasast/hunt/prover.py index cd2bccd..44095ec 100644 --- a/src/openultrasast/hunt/prover.py +++ b/src/openultrasast/hunt/prover.py @@ -4,8 +4,8 @@ from __future__ import annotations import re from collections import Counter -from collections.abc import Iterator, Mapping -from dataclasses import asdict, dataclass, replace +from collections.abc import Callable, Iterator, Mapping +from dataclasses import asdict, dataclass, field, replace from pathlib import Path from typing import Any @@ -23,6 +23,7 @@ from ..semantic.obligations.facts import load_obligation_facts from ..semantic.obligations.operations import Discharge, Operation, find_discharges, find_operations, valid_witness from ..semantic.obligations.paths import compose_paths from ..semantic.obligations.policy import DEFAULT_POLICY_RELPATH, load_declared_policy +from ..semantic.obligations.resolution import CallBindings, bind_calls from ..semantic.obligations.trace import build_trace from .producers import Context, _safe_path @@ -41,6 +42,9 @@ class Coverage: # until the prover opts into abstract judgment (task 10.6). Indeterminate pairs are counted # neither clean nor discharged. pair_judgments: PairJudgmentCounts = PairJudgmentCounts() + # Requirement 3.3/3.4 (access-control-prover): call-resolution outcome totals. Prover-mode + # resolution has no bare-name fallback; every call lands in exactly one bucket. + resolution_counts: dict[str, int] = field(default_factory=dict) @dataclass(frozen=True) @@ -51,6 +55,52 @@ class ProverResult: traces: tuple[dict[str, Any], ...] = () +def _resolution_bindings( + readable: list[str], + read_ir: Callable[[str], FileIR | None], + functions: Mapping[tuple[str, str], FunctionIR], + ctx: Context, + reasons: Counter[str], + degradations: list[dict[str, str]], +) -> CallBindings: + """Parse every readable file and resolve all calls (Reqs 3.3/3.4).""" + for path in readable: + try: + read_ir(path) + except Exception as error: + reasons[type(error).__name__] += 1 + degradations.append({"path": path, "reason": type(error).__name__}) + functions_by_module: dict[str, list[FunctionIR]] = {} + for (module, _name), function_ir in functions.items(): + functions_by_module.setdefault(module, []).append(function_ir) + return bind_calls(functions_by_module=functions_by_module, index=ctx.symbol_index) + + +def _exact_callers(bindings: CallBindings, ctx: Context, path: str, function: str) -> list[tuple[str, int]]: + """Caller sites from exact bindings only; no bare-name fallback (Reqs 3.3/3.4).""" + out = [] + for symbol in ctx.symbol_index.by_name.get(function, ()): + if symbol.module != path: + continue + for binding in bindings.exact_callers(symbol): + out.append((f"{binding.module}::{binding.function}", binding.line)) + return sorted(set(out)) + + +def _targets_from_bindings( + touched: set[tuple[str, str]], + functions: Mapping[tuple[str, str], FunctionIR], + bindings: CallBindings, +) -> set[tuple[str, str]]: + """Targets join by resolved symbol identity, never by trailing name (Req 3.3).""" + target_functions = set(touched) + for key in touched: + for call in functions[key].calls: + for target_set in bindings.resolutions.get((key[0], key[1], call.line), ()): + target_functions.update((symbol.module, symbol.name) for symbol in target_set.targets) + return target_functions + + def prove_change(root: Path, change: Mapping[str, Any], ctx: Context | None = None, *, max_depth: int = 3) -> ProverResult: """Check touched head functions and their direct operation-bearing callees, with no model.""" if max_depth < 0: @@ -162,32 +212,15 @@ def prove_change(root: Path, change: Mapping[str, Any], ctx: Context | None = No if file.get("head_lines") and not selected: reasons["function_unavailable"] += 1 - # The existing name-based index supplies call sites; retain file-qualified enclosing functions. - def callers(path: str, function: str) -> list[tuple[str, int]]: - del path - out = [] - for caller_path, line in source.callers_of(root, function, files=readable, texts=ctx.texts, index=ctx.call_index): - ir = read_ir(caller_path) - if ir is None: - continue - spans = [fn for fn in ir.functions if fn.name != "" and fn.start_line <= line <= fn.end_line] - if spans: - fn = min(spans, key=lambda item: item.end_line - item.start_line) - out.append((f"{caller_path}::{fn.name}", line)) - return out + # Full IR coverage first: conservative resolution (Req 3.3/3.4) needs every readable file parsed. + bindings = _resolution_bindings(readable, read_ir, functions, ctx, reasons, degradations) - target_functions = set(touched) - call_names = {trailing_name(call.name) for key in touched for call in functions[key].calls} - if call_names: - for path in readable: - # Ranges are cached and avoid parsing unrelated files just to look for direct callees. - try: - if any(span.name in call_names for span in ctx.ranges(path)): - read_ir(path) - except Exception as error: - reasons[type(error).__name__] += 1 - degradations.append({"path": path, "reason": type(error).__name__}) - target_functions.update(key for key, ops in operations.items() if key[1] in call_names and ops) + # Conservative caller resolution: only exact (complete, single-target) bindings supply callers; + # the global bare-name index fallback is removed from the prover path (Req 3.3/3.4). + def callers(path: str, function: str) -> list[tuple[str, int]]: + return _exact_callers(bindings, ctx, path, function) + + target_functions = _targets_from_bindings(touched, functions, bindings) paths, unreached = compose_paths( callers=callers, entries={f"{e.path}::{e.function_name}" for e in entries}, targets=sorted(target_functions), max_depth=max_depth ) @@ -343,6 +376,7 @@ def prove_change(root: Path, change: Mapping[str, Any], ctx: Context | None = No len(checked), len(checked - undischarged), dict(reasons), + resolution_counts=dict(bindings.counts), ) return ProverResult(tuple(findings.values()), coverage, tuple(degradations), tuple(traces)) diff --git a/src/openultrasast/semantic/obligations/resolution.py b/src/openultrasast/semantic/obligations/resolution.py index 9eee343..1a01247 100644 --- a/src/openultrasast/semantic/obligations/resolution.py +++ b/src/openultrasast/semantic/obligations/resolution.py @@ -12,6 +12,8 @@ from collections.abc import Iterable, Mapping from dataclasses import dataclass, field from ..cst import DeclarationScope, declaration_scopes +from ..ir import CallSite +from .abstract_domain import Completeness @dataclass(frozen=True) @@ -74,3 +76,151 @@ def _inheritance_edges(module: str, scopes: Iterable[DeclarationScope]) -> dict[ if scope.enclosing_type is not None: edges[(module, scope.enclosing_type)] = scope.enclosing_base_types return edges + + +@dataclass(frozen=True) +class CallTargetSet: + """The outcome of resolving one call: complete exact targets, incomplete candidates, or unresolved.""" + + targets: tuple[SymbolId, ...] + completeness: Completeness + + +_SELF_RECEIVERS = frozenset({"self", "$this", "this", "cls", "static"}) + + +@dataclass(frozen=True) +class CallerBinding: + """One call site bound to a target declaration, with whether the binding is exact.""" + + module: str + function: str + line: int + exact: bool + + +@dataclass(frozen=True) +class CallBindings: + """Forward and reverse views of every resolved call in one change.""" + + call_targets: tuple[CallTargetSet, ...] + exact_caller_index: Mapping[SymbolId, tuple[CallerBinding, ...]] + counts: Mapping[str, int] + resolutions: Mapping[tuple[str, str, int], tuple[CallTargetSet, ...]] = field(default_factory=dict) + + def exact_callers(self, target: SymbolId) -> tuple[CallerBinding, ...]: + """Call sites whose resolution is complete and lands on this target.""" + return self.exact_caller_index.get(target, ()) + + +def resolve_call(call: CallSite, *, module: str, index: SymbolIndex) -> CallTargetSet: + """Resolve one call site against the repository symbol index. + + Exact receiver and full signature produce one complete target; partial information + produces a finite candidate set with an incompleteness reason; nothing falls back to + a global bare-name key (Reqs 3.3, 3.4, 11.1, 11.6). + """ + same_name = index.by_name.get(call.name, ()) + receiver = (call.receiver or "").strip() + if not receiver: + return _module_scope_targets(same_name, module, call.arity) + if receiver in _SELF_RECEIVERS: + return _method_targets(same_name, module, call.arity, "receiver_self_shared_across_classes") + if _is_declared_type(same_name, receiver): + return _type_receiver_targets(same_name, receiver, call.arity) + return _variable_receiver_targets(same_name, call.arity) + + +def bind_calls(*, functions_by_module: Mapping[str, Iterable], index: SymbolIndex) -> CallBindings: + """Resolve every call in the given function IRs and index the exact caller edges.""" + call_targets: list[CallTargetSet] = [] + caller_index: dict[SymbolId, list[CallerBinding]] = {} + resolutions: dict[tuple[str, str, int], list[CallTargetSet]] = {} + counts = {"complete_calls": 0, "incomplete_calls": 0, "unresolved_calls": 0} + for module, functions in functions_by_module.items(): + for function_ir in functions: + for call in function_ir.calls: + target_set = resolve_call(call, module=module, index=index) + call_targets.append(target_set) + resolutions.setdefault((module, function_ir.name, call.line), []).append(target_set) + _count_target_set(target_set, counts) + if not target_set.completeness.reasons and len(target_set.targets) == 1: + binding = CallerBinding(module=module, function=function_ir.name, line=call.line, exact=True) + bindings = caller_index.setdefault(target_set.targets[0], []) + if binding not in bindings: + bindings.append(binding) + return CallBindings( + call_targets=tuple(call_targets), + exact_caller_index={target: tuple(bindings) for target, bindings in sorted(caller_index.items())}, + counts=dict(sorted(counts.items())), + resolutions={key: tuple(sets) for key, sets in sorted(resolutions.items())}, + ) + + +def _count_target_set(target_set: CallTargetSet, counts: dict[str, int]) -> None: + if not target_set.targets: + counts["unresolved_calls"] += 1 + elif target_set.completeness.reasons: + counts["incomplete_calls"] += 1 + else: + counts["complete_calls"] += 1 + + +def _module_scope_targets(same_name: tuple[SymbolId, ...], module: str, arity: int | None) -> CallTargetSet: + # A receiver-less call (implicit this in class languages, plain call in + # module languages) resolves against every same-name declaration in the + # calling module: module-level functions and class methods alike. Complete + # only when exactly one candidate exists; a shared name stays a finite + # candidate set with an incompleteness reason, never a bare-name merge. + module_scope = tuple(symbol for symbol in same_name if symbol.module == module) + narrowed = _arity_narrow(module_scope, arity) + if module_scope and not narrowed: + return CallTargetSet(targets=(), completeness=Completeness(is_complete=False, reasons=("resolution:arity_mismatch",))) + return _finish(narrowed, shared_reason="name_shared_across_module_scopes") + + +def _method_targets(same_name: tuple[SymbolId, ...], module: str, arity: int | None, shared_reason: str) -> CallTargetSet: + methods = tuple(symbol for symbol in same_name if symbol.enclosing_type is not None and symbol.module == module) + return _narrowed_or_unresolved(methods, arity, shared_reason) + + +def _type_receiver_targets(same_name: tuple[SymbolId, ...], type_name: str, arity: int | None) -> CallTargetSet: + methods = tuple(symbol for symbol in same_name if symbol.enclosing_type == type_name) + return _narrowed_or_unresolved(methods, arity, "class_shared_across_modules" if len({s.module for s in methods}) > 1 else "") + + +def _variable_receiver_targets(same_name: tuple[SymbolId, ...], arity: int | None) -> CallTargetSet: + methods = tuple(symbol for symbol in same_name if symbol.enclosing_type is not None) + narrowed = _arity_narrow(methods, arity) + if not narrowed: + return CallTargetSet(targets=(), completeness=Completeness(is_complete=False, reasons=("resolution:no_candidate",))) + return CallTargetSet( + targets=narrowed, + completeness=Completeness(is_complete=False, reasons=("resolution:receiver_type_unknown",)), + ) + + +def _narrowed_or_unresolved(methods: tuple[SymbolId, ...], arity: int | None, shared_reason: str) -> CallTargetSet: + narrowed = _arity_narrow(methods, arity) + if methods and not narrowed: + return CallTargetSet(targets=(), completeness=Completeness(is_complete=False, reasons=("resolution:arity_mismatch",))) + return _finish(narrowed, shared_reason=shared_reason) + + +def _is_declared_type(same_name: tuple[SymbolId, ...], receiver: str) -> bool: + return any(symbol.enclosing_type == receiver for symbol in same_name) + + +def _arity_narrow(symbols: tuple[SymbolId, ...], arity: int | None) -> tuple[SymbolId, ...]: + if arity is None: + return symbols + narrowed = tuple(symbol for symbol in symbols if len(symbol.signature or ()) == arity) + return narrowed or symbols + + +def _finish(targets: tuple[SymbolId, ...], shared_reason: str) -> CallTargetSet: + if not targets: + return CallTargetSet(targets=(), completeness=Completeness(is_complete=False, reasons=("resolution:no_candidate",))) + if len(targets) == 1: + return CallTargetSet(targets=targets, completeness=Completeness(is_complete=True)) + return CallTargetSet(targets=targets, completeness=Completeness(is_complete=False, reasons=(f"resolution:{shared_reason}",))) diff --git a/tests/test_call_resolution.py b/tests/test_call_resolution.py new file mode 100644 index 0000000..580e54d --- /dev/null +++ b/tests/test_call_resolution.py @@ -0,0 +1,247 @@ +"""Call resolution tests: exact, finite-candidate and unresolved targets (task 8.3). + +Req 3.3: multiple feasible targets stay a finite candidate set marked incomplete, +never merged under a bare name. +Req 3.4: no bounded candidate set -> unresolved, unusable for complete paths. +Req 11.1/11.6: grades distinguish definite/possible/unknown; precision loss never +flips a complete result to a different complete result. +""" + +from __future__ import annotations + +import pytest + +from openultrasast.semantic.ir import CallSite, FunctionIR +from openultrasast.semantic.obligations.abstract_domain import Completeness +from openultrasast.semantic.obligations.resolution import ( + CallBindings, + CallerBinding, + CallTargetSet, + SymbolId, + SymbolIndex, + bind_calls, + resolve_call, +) + + +def _complete() -> Completeness: + return Completeness(is_complete=True) + + +def _call(name: str, line: int, receiver: str | None = None, arity: int | None = None) -> CallSite: + return CallSite( + name=name, + line=line, + arg_texts=(), + arg_is_constant=(), + arg_names=(), + extra_arg_is_sequence=False, + receiver=receiver, + arity=arity, + ) + + +JAVA_ONE_CLASS = ( + "src/a/Handler.java", + """package pkg; +public class Handler { + public void run(String command) { } + public void handle() { run("x"); this.run("y"); } +} +""", + "java", +) + +JAVA_TWO_CLASSES_SAME_METHOD = ( + "src/a/Two.java", + """package pkg; +public class Auditor { public void check(String item) { } } +public class Gate { public void check(String item) { } } +""", + "java", +) + +PYTHON_MODULE_SCOPE = ( + "src/svc.py", + """def audit(request): + pass + +def other(request): + audit(request) +""", + "python", +) + + +def index_of(*files: tuple[str, str, str]) -> SymbolIndex: + return SymbolIndex.build(files) + + +class TestResolveCall: + def test_exact_self_receiver_in_one_class_is_complete(self) -> None: + index = index_of(JAVA_ONE_CLASS) + call = _call(name="run", line=5, receiver="this", arity=1) + target_set = resolve_call(call, module="src/a/Handler.java", index=index) + assert target_set.targets == (SymbolId("src/a/Handler.java", "Handler", "run", ("command",)),) + assert target_set.completeness.reasons == () + + def test_two_classes_same_method_stay_candidates_never_bare(self) -> None: + index = index_of(JAVA_TWO_CLASSES_SAME_METHOD) + call = _call(name="check", line=2, receiver="auditor", arity=1) + target_set = resolve_call(call, module="src/a/Two.java", index=index) + assert set(target_set.targets) == { + SymbolId("src/a/Two.java", "Auditor", "check", ("item",)), + SymbolId("src/a/Two.java", "Gate", "check", ("item",)), + } + assert target_set.completeness.reasons # incomplete, not merged + + def test_overload_arity_narrows_but_variable_receiver_stays_incomplete(self) -> None: + index = index_of( + ( + "src/a/Ops.java", + """public class Ops { + public void run() { } + public void run(String command) { } +} +""", + "java", + ) + ) + narrowed = resolve_call( + _call(name="run", line=1, receiver="ops", arity=1), + module="src/a/Ops.java", + index=index, + ) + assert narrowed.targets == (SymbolId("src/a/Ops.java", "Ops", "run", ("command",)),) + assert narrowed.completeness.reasons # partial type information stays incomplete + + def test_unknown_arity_keeps_overloads_as_candidates(self) -> None: + index = index_of( + ( + "src/a/Ops.java", + """public class Ops { + public void run() { } + public void run(String command) { } +} +""", + "java", + ) + ) + candidates = resolve_call( + _call(name="run", line=1, receiver="ops", arity=None), + module="src/a/Ops.java", + index=index, + ) + assert len(candidates.targets) == 2 + assert candidates.completeness.reasons + + def test_no_candidate_is_unresolved_with_reason(self) -> None: + index = index_of(JAVA_ONE_CLASS) + call = _call(name="missing", line=3, receiver="helper", arity=1) + target_set = resolve_call(call, module="src/a/Handler.java", index=index) + assert target_set.targets == () + assert target_set.completeness.reasons + + def test_module_scope_unique_bare_call_is_complete(self) -> None: + index = index_of(PYTHON_MODULE_SCOPE) + call = _call(name="audit", line=5, receiver=None, arity=1) + target_set = resolve_call(call, module="src/svc.py", index=index) + assert target_set.targets == (SymbolId("src/svc.py", None, "audit", ("request",)),) + assert target_set.completeness.reasons == () + + def test_module_scope_shared_name_is_incomplete(self) -> None: + index = index_of( + ( + "src/twin.py", + """def audit(request): + pass + +def audit(request, extra): + pass +""", + "python", + ) + ) + call = _call(name="audit", line=4, receiver=None, arity=None) + target_set = resolve_call(call, module="src/twin.py", index=index) + assert len(target_set.targets) == 2 + assert target_set.completeness.reasons + + def test_precision_loss_never_flips_complete_to_complete(self) -> None: + single = index_of( + ( + "src/a/One.java", + """public class Gate { + public void check(String item) { } + public void audit() { check("x"); } +} +""", + "java", + ) + ) + doubled = index_of( + ( + "src/a/One.java", + """public class Gate { + public void check(String item) { } + public void audit() { check("x"); } +} +public class Other { + public void check(String item) { } +} +""", + "java", + ) + ) + call = _call(name="check", line=3, receiver="this", arity=1) + before = resolve_call(call, module="src/a/One.java", index=single) + after = resolve_call(call, module="src/a/One.java", index=doubled) + assert before.completeness.reasons == () + assert after.completeness.reasons # grew ambiguous, never silently complete + + +class TestBindCalls: + def test_exact_binding_feeds_reverse_caller_index_and_counts(self) -> None: + index = index_of(JAVA_ONE_CLASS) + bindings = bind_calls( + functions_by_module={ + "src/a/Handler.java": ( + _ir_stub( + calls=( + _call(name="run", line=5, receiver="this", arity=1), + _call(name="run", line=5, receiver="this", arity=1), + ), + name="handle", + ), + ) + }, + index=index, + ) + assert isinstance(bindings, CallBindings) + target = SymbolId("src/a/Handler.java", "Handler", "run", ("command",)) + assert bindings.exact_callers(target) == (CallerBinding(module="src/a/Handler.java", function="handle", line=5, exact=True),) + assert bindings.counts["complete_calls"] == 2 + assert bindings.counts["unresolved_calls"] == 0 + + def test_candidate_only_binding_is_not_an_exact_caller(self) -> None: + index = index_of(JAVA_TWO_CLASSES_SAME_METHOD) + bindings = bind_calls( + functions_by_module={"src/a/Two.java": (_ir_stub(calls=(_call(name="check", line=3, receiver="gate", arity=1),)),)}, + index=index, + ) + target = SymbolId("src/a/Two.java", "Gate", "check", ("item",)) + assert bindings.exact_callers(target) == () + assert bindings.counts["incomplete_calls"] == 1 + + +def _ir_stub(calls: tuple[CallSite, ...], name: str = "caller") -> FunctionIR: + return FunctionIR(name=name, params=(), start_line=1, end_line=9, binds=(), calls=calls) + + +@pytest.mark.parametrize("fixture", [JAVA_ONE_CLASS, PYTHON_MODULE_SCOPE]) +def test_call_target_set_is_canonical_and_frozen(fixture: tuple[str, str, str]) -> None: + index = index_of(fixture) + assert index.declarations == tuple(sorted(index.declarations)) + target_set = CallTargetSet(targets=index.declarations, completeness=_complete()) + with pytest.raises(Exception): # noqa: B017 - frozen contract + target_set.targets = () # type: ignore[misc]