From 06d5363ce2098c7a69968a29a8870bc0ed699192 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Thu, 29 Oct 2026 18:37:06 +0100 Subject: [PATCH] feat(access-control-prover): 9.3 summary caches and explicit exploration cutoffs --- .kiro/specs/access-control-prover/tasks.md | 2 +- src/openultrasast/hunt/producers.py | 17 +- .../semantic/obligations/summaries.py | 118 ++++++++++++-- tests/test_obligation_summaries.py | 154 +++++++++++++++++- 4 files changed, 271 insertions(+), 20 deletions(-) diff --git a/.kiro/specs/access-control-prover/tasks.md b/.kiro/specs/access-control-prover/tasks.md index 9a70f0d..f3cf647 100644 --- a/.kiro/specs/access-control-prover/tasks.md +++ b/.kiro/specs/access-control-prover/tasks.md @@ -150,7 +150,7 @@ its final resolution, dominance, fact-acquisition and measurement semantics with - _Requirements: 3.7, 11.6_ - _Boundary: Function Summaries_ - _Depends: 9.1_ -- [ ] 9.3 Add summary caches and explicit exploration cutoffs +- [x] 9.3 Add summary caches and explicit exploration cutoffs - Reuse summaries per change and carry depth, fan-out, deadline and other cutoff reasons into completeness. - Done when each cutoff fixture returns accumulated evidence plus the exact incomplete reason. - _Requirements: 3.8, 6.2, 6.3, 7.2_ diff --git a/src/openultrasast/hunt/producers.py b/src/openultrasast/hunt/producers.py index 2c721db..f1b20aa 100644 --- a/src/openultrasast/hunt/producers.py +++ b/src/openultrasast/hunt/producers.py @@ -15,8 +15,9 @@ from .. import mapping, source from ..model.taxonomy import DEFAULT_FAMILIES_PATH, load_families from ..preprocess import build_file_target, detect_language from ..semantic.facts import load_facts -from ..semantic.ir import FileIR, parse_file +from ..semantic.ir import FileIR, FunctionIR, parse_file from ..semantic.obligations import check_obligations, load_obligation_facts, resolution +from ..semantic.obligations import summaries as summaries_module from ..semantic.obligations.dominance import OrderDominance from ..semantic.obligations.operations import covers, find_operations, valid_witness from .labels import label_rows @@ -116,6 +117,20 @@ class Context: files.append((path, text, language)) return resolution.SymbolIndex.build(files) + @cached_property + def summaries(self) -> tuple[summaries_module.FunctionSummary, ...]: + """Monotone function summaries over the product tree, computed once per change (Req 6.2).""" + functions_by_module: dict[str, list[FunctionIR]] = {} + for path in self.product_files: + try: + file_ir = self.ir(path) + except _Absent: + # Same guard as entry_points and symbol_index: an unreadable product file + # contributes no functions, not a failure of the summary cache. + continue + functions_by_module.setdefault(path, []).extend(file_ir.functions) + return summaries_module.monotone_summaries(functions_by_module, index=self.symbol_index) + @cached_property def hunks(self) -> dict[str, list[list[int]]]: """Line ranges per file that the push changed or added, at head.""" diff --git a/src/openultrasast/semantic/obligations/summaries.py b/src/openultrasast/semantic/obligations/summaries.py index 64701df..de57469 100644 --- a/src/openultrasast/semantic/obligations/summaries.py +++ b/src/openultrasast/semantic/obligations/summaries.py @@ -5,18 +5,26 @@ operations, local witnesses, resolved calls and local completeness WITHOUT traversing callers. Task 9.2 (Reqs 3.7, 11.6): a monotone fixpoint over strongly connected call components unions callee evidence into callers until stable, so direct and mutual recursion terminate with the accumulated -evidence. Summaries are deterministic and carry no target names beyond fact -identifiers (Req 11.7). +evidence. Task 9.3 (Reqs 3.8, 6.2, 6.3, 7.2): an exploration budget bounds +the fixpoint; depth, fan-out and deadline cutoffs stop growth, record the +exact cutoff reason and leave the affected summaries incomplete while +preserving the evidence accumulated so far. Summaries are deterministic and +carry no target names beyond fact identifiers (Req 11.7). """ from collections.abc import Iterable, Iterator, Mapping -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace +from time import monotonic from ..ir import FunctionIR from .abstract_domain import Completeness, join_completeness from .operations import Discharge, Operation from .resolution import CallTargetSet, SymbolId, SymbolIndex, resolve_call +_DEPTH_CUTOFF = "summaries:depth_cutoff" +_FAN_OUT_CUTOFF = "summaries:fan_out_cutoff" +_DEADLINE_CUTOFF = "summaries:deadline_cutoff" + @dataclass(frozen=True) class FunctionSummary: @@ -29,6 +37,20 @@ class FunctionSummary: completeness: Completeness = field(default_factory=lambda: Completeness(is_complete=True)) +@dataclass(frozen=True) +class ExplorationBudget: + """Work bounds for the summary fixpoint; ``None`` means unbounded. + + Req 3.8: every bound that truncates exploration records its exact cutoff + reason and leaves affected summaries incomplete with the evidence + accumulated so far. + """ + + depth: int | None = None + fan_out: int | None = None + deadline: float | None = None + + def _enclosing_type(index: SymbolIndex, module: str, name: str) -> str | None: """Return the enclosing type of the function ``name`` declared in ``module``.""" for symbol in index.by_name.get(name, ()): @@ -245,22 +267,80 @@ def _propagated(summary: FunctionSummary, callees: Iterable[FunctionSummary]) -> ) +def _callees_for( + member_symbol: SymbolId, + edges: Mapping[SymbolId, tuple[SymbolId, ...]], + summaries_by_symbol: Mapping[SymbolId, FunctionSummary], + fan_out: int | None, +) -> tuple[list[FunctionSummary], bool]: + """Resolve a member's exact callees, truncated to ``fan_out`` when set.""" + callee_edges = [edge for edge in edges.get(member_symbol, ()) if edge in summaries_by_symbol] + truncated = fan_out is not None and len(callee_edges) > fan_out + if truncated: + callee_edges = callee_edges[:fan_out] + return [summaries_by_symbol[edge] for edge in callee_edges], truncated + + +def _propagate_round( + members: tuple[FunctionSummary, ...], + summaries_by_symbol: dict[SymbolId, FunctionSummary], + edges: Mapping[SymbolId, tuple[SymbolId, ...]], + budget: ExplorationBudget, +) -> bool: + """One union round over the component; returns whether anything changed.""" + changed = False + for member in members: + current = summaries_by_symbol[member.symbol] + callees, truncated = _callees_for(member.symbol, edges, summaries_by_symbol, budget.fan_out) + propagated = _propagated(current, callees) + if truncated: + propagated = replace( + propagated, + completeness=join_completeness( + propagated.completeness, + Completeness(is_complete=False, reasons=(_FAN_OUT_CUTOFF,)), + ), + ) + if propagated != current: + summaries_by_symbol[member.symbol] = propagated + changed = True + return changed + + +def _apply_cutoff_reason( + members: tuple[FunctionSummary, ...], + summaries_by_symbol: dict[SymbolId, FunctionSummary], + reason: str, +) -> None: + """Mark every member of a truncated component with the cutoff reason.""" + for member in members: + current = summaries_by_symbol[member.symbol] + summaries_by_symbol[member.symbol] = replace( + current, + completeness=join_completeness(current.completeness, Completeness(is_complete=False, reasons=(reason,))), + ) + + def _stabilise_component( members: tuple[FunctionSummary, ...], summaries_by_symbol: dict[SymbolId, FunctionSummary], edges: Mapping[SymbolId, tuple[SymbolId, ...]], + budget: ExplorationBudget, ) -> None: - """Iterate a component's members until every summary is stable.""" - changed = True - while changed: - changed = False - for member in members: - current = summaries_by_symbol[member.symbol] - callees = [summaries_by_symbol[edge] for edge in edges.get(member.symbol, ()) if edge in summaries_by_symbol] - propagated = _propagated(current, callees) - if propagated != current: - summaries_by_symbol[member.symbol] = propagated - changed = True + """Iterate a component's members until stable or a budget bound fires.""" + if not any(edges.get(member.symbol) for member in members): + return + rounds = 0 + while True: + if budget.deadline is not None and monotonic() > budget.deadline: + _apply_cutoff_reason(members, summaries_by_symbol, _DEADLINE_CUTOFF) + return + if not _propagate_round(members, summaries_by_symbol, edges, budget): + return + rounds += 1 + if budget.depth is not None and rounds >= budget.depth: + _apply_cutoff_reason(members, summaries_by_symbol, _DEPTH_CUTOFF) + return def monotone_summaries( @@ -269,14 +349,18 @@ def monotone_summaries( *, operations_by_symbol: Mapping[SymbolId, tuple[Operation, ...]] | None = None, witnesses_by_symbol: Mapping[SymbolId, tuple[Discharge, ...]] | None = None, + budget: ExplorationBudget | None = None, ) -> tuple[FunctionSummary, ...]: """Summaries after the monotone fixpoint over strongly connected components. Local summaries first; then every component (callee components first) is iterated until union-stable, so recursion terminates with the accumulated - operations, witnesses and completeness reasons of its cycle. Output order - matches :func:`local_summaries`. + operations, witnesses and completeness reasons of its cycle. ``budget`` + bounds the exploration (Req 3.8): a depth, fan-out or deadline cutoff + stops growth and leaves affected summaries incomplete with the exact + cutoff reason. Output order matches :func:`local_summaries`. """ + bounded = budget if budget is not None else ExplorationBudget() local = local_summaries( functions_by_module, index, @@ -287,5 +371,5 @@ def monotone_summaries( edges = _exact_caller_edges(summaries_by_symbol) for component in _sorted_scc_components(edges, summaries_by_symbol): members = tuple(summaries_by_symbol[member] for member in component) - _stabilise_component(members, summaries_by_symbol, edges) + _stabilise_component(members, summaries_by_symbol, edges, bounded) return tuple(summaries_by_symbol[summary.symbol] for summary in local) diff --git a/tests/test_obligation_summaries.py b/tests/test_obligation_summaries.py index 2002a0e..e668d68 100644 --- a/tests/test_obligation_summaries.py +++ b/tests/test_obligation_summaries.py @@ -3,6 +3,8 @@ from __future__ import annotations import inspect +from dataclasses import FrozenInstanceError +from time import monotonic import pytest @@ -12,7 +14,12 @@ from openultrasast.semantic.obligations import summaries from openultrasast.semantic.obligations.abstract_domain import Completeness from openultrasast.semantic.obligations.operations import Discharge, Operation from openultrasast.semantic.obligations.resolution import SymbolId, SymbolIndex -from openultrasast.semantic.obligations.summaries import FunctionSummary, local_summaries, monotone_summaries +from openultrasast.semantic.obligations.summaries import ( + ExplorationBudget, + FunctionSummary, + local_summaries, + monotone_summaries, +) PHP_CLASS = """ None: assert [summary.symbol for summary in monotone_summaries(functions, index)] == [ summary.symbol for summary in local_summaries(functions, index) ] + + +JAVA_FANOUT = """ +class Fan { + void helper1() { this.helper2(); this.helper3(); } + void helper2() {} + void helper3() {} +} +""".strip() + + +def _java_operation(name: str, line: int) -> Operation: + return Operation( + path="src/C.java", + line=line, + function=name, + kind="protected_read", + fact_id=f"java:{name}-op", + resource=name, + requires=("path_guard",), + ) + + +def _fanout_functions() -> dict[str, tuple[FunctionIR, ...]]: + return { + "src/C.java": ( + FunctionIR( + "helper1", + (), + 3, + 3, + (), + (_self_call("helper2", 3), _self_call("helper3", 3)), + ), + FunctionIR("helper2", (), 4, 4, (), ()), + FunctionIR("helper3", (), 5, 5, (), ()), + ) + } + + +def _mutual_functions() -> dict[str, tuple[FunctionIR, ...]]: + return { + "src/C.java": ( + FunctionIR("a", (), 2, 4, (), (_self_call("b", 3),)), + FunctionIR("b", (), 5, 7, (), (_self_call("a", 6),)), + ) + } + + +def _operations_injected( + functions: dict[str, tuple[FunctionIR, ...]], index: SymbolIndex, injected: dict[str, Operation] +) -> dict[SymbolId, tuple[Operation, ...]]: + operations_by_symbol: dict[SymbolId, tuple[Operation, ...]] = {} + for summary in local_summaries(functions, index): + operation = injected.get(summary.symbol.name) + if operation is not None: + operations_by_symbol[summary.symbol] = (operation,) + return operations_by_symbol + + +def test_depth_cutoff_preserves_accumulated_evidence_and_reason() -> None: + index = _index((("src/C.java", JAVA_MUTUAL, "java"),)) + functions = _mutual_functions() + operations_by_symbol = _operations_injected(functions, index, {"b": _java_operation("b", 6)}) + summaries_with_cutoff = monotone_summaries( + functions, index, operations_by_symbol=operations_by_symbol, budget=ExplorationBudget(depth=1) + ) + summaries_unbounded = monotone_summaries(functions, index, operations_by_symbol=operations_by_symbol) + by_name = {summary.symbol.name: summary for summary in summaries_with_cutoff} + unbounded_by_name = {summary.symbol.name: summary for summary in summaries_unbounded} + assert by_name["a"].operations == unbounded_by_name["a"].operations + assert by_name["a"].operations == (_java_operation("b", 6),) + assert "summaries:depth_cutoff" in by_name["a"].completeness.reasons + assert "summaries:depth_cutoff" in by_name["b"].completeness.reasons + assert not by_name["a"].completeness.is_complete + + +def test_deadline_cutoff_returns_local_evidence() -> None: + index = _index((("src/C.java", JAVA_FANOUT, "java"),)) + functions = _fanout_functions() + operations_by_symbol = _operations_injected( + functions, index, {"helper2": _java_operation("helper2", 4), "helper3": _java_operation("helper3", 5)} + ) + summaries_with_cutoff = monotone_summaries( + functions, index, operations_by_symbol=operations_by_symbol, budget=ExplorationBudget(deadline=monotonic() - 1.0) + ) + by_name = {summary.symbol.name: summary for summary in summaries_with_cutoff} + assert by_name["helper1"].operations == () + assert "summaries:deadline_cutoff" in by_name["helper1"].completeness.reasons + assert not by_name["helper1"].completeness.is_complete + assert by_name["helper2"].completeness.is_complete + assert by_name["helper2"].operations == (_java_operation("helper2", 4),) + + +def test_fan_out_cutoff_merges_sorted_prefix() -> None: + index = _index((("src/C.java", JAVA_FANOUT, "java"),)) + functions = _fanout_functions() + operations_by_symbol = _operations_injected( + functions, index, {"helper2": _java_operation("helper2", 4), "helper3": _java_operation("helper3", 5)} + ) + summaries_with_cutoff = monotone_summaries( + functions, index, operations_by_symbol=operations_by_symbol, budget=ExplorationBudget(fan_out=1) + ) + by_name = {summary.symbol.name: summary for summary in summaries_with_cutoff} + assert by_name["helper1"].operations == (_java_operation("helper2", 4),) + assert "summaries:fan_out_cutoff" in by_name["helper1"].completeness.reasons + assert not by_name["helper1"].completeness.is_complete + assert by_name["helper3"].completeness.is_complete + + +def test_all_none_budget_equals_default_behaviour() -> None: + index = _index((("src/C.java", JAVA_MUTUAL, "java"),)) + functions = _mutual_functions() + operations_by_symbol = _operations_injected(functions, index, {"b": _java_operation("b", 6)}) + with_default = monotone_summaries(functions, index, operations_by_symbol=operations_by_symbol) + with_empty_budget = monotone_summaries(functions, index, operations_by_symbol=operations_by_symbol, budget=ExplorationBudget()) + assert with_default == with_empty_budget + + +def test_exploration_budget_is_frozen() -> None: + budget = ExplorationBudget(depth=1) + with pytest.raises(FrozenInstanceError): + budget.depth = 2 # type: ignore[misc] + + +def test_cutoff_reasons_join_existing_incompleteness() -> None: + index = _index((("src/C.java", JAVA_FANOUT, "java"),)) + functions = { + "src/C.java": ( + FunctionIR( + "helper1", + (), + 3, + 3, + (), + (_self_call("helper2", 3), CallSite("missing", 3, (), (), (), False, 0)), + ), + FunctionIR("helper2", (), 4, 4, (), ()), + ) + } + summaries_with_cutoff = monotone_summaries(functions, index, budget=ExplorationBudget(deadline=monotonic() - 1.0)) + by_name = {summary.symbol.name: summary for summary in summaries_with_cutoff} + reasons = by_name["helper1"].completeness.reasons + assert "resolution:no_candidate" in reasons + assert "summaries:deadline_cutoff" in reasons