feat(access-control-prover): 11.2 generic call-site registration strategy

This commit is contained in:
Marius Ciepluch
2026-10-29 20:23:50 +01:00
parent 86709daf2e
commit c279513def
3 changed files with 250 additions and 1 deletions
+1 -1
View File
@@ -217,7 +217,7 @@ its final resolution, dominance, fact-acquisition and measurement semantics with
- _Requirements: 1.1, 1.5, 1.7_
- _Boundary: Mapping Registry_
- _Depends: 7.1_
- [ ] 11.2 Add the generic call-site registration strategy
- [x] 11.2 Add the generic call-site registration strategy
- Resolve repository-defined registered handlers through stable symbols and return unresolved registration coverage.
- Done when exact and unresolved registration fixtures produce entries or coverage respectively.
- _Requirements: 1.2, 1.3, 1.4, 1.5, 1.6_
@@ -0,0 +1,121 @@
"""Generic call-site registration adapter (task 11.2).
Resolves handlers registered at call sites through the same-file stable
symbols: a registration call whose argument names a class declared in the
file yields one entry per declared method of that class, with the
registration site and the resolved handler identity as evidence. A
registration whose argument names no declared class yields no entry and is
reported as unresolved coverage. The registration vocabulary is data
supplied to the strategy; this module names no target.
"""
from __future__ import annotations
import re
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any
from ..semantic.cst import declaration_scopes
if TYPE_CHECKING:
from .legacy import EntryPointRecord
@dataclass
class RegistrationSiteStrategy:
"""Recognise handlers registered at call sites named by vocabulary."""
identifier: str
languages: tuple[str, ...]
vocabulary: tuple[str, ...] = ()
_matchers: tuple[tuple[str, re.Pattern[str]], ...] = field(default=(), init=False, repr=False, compare=False)
def __post_init__(self) -> None:
self._matchers = tuple((name, re.compile(rf"(?<!\w){re.escape(name)}\s*\(")) for name in self.vocabulary)
def recognize(self, target: Any, text: str) -> list[EntryPointRecord]:
records: list[EntryPointRecord] = []
declared_classes = _declared_classes(target.path, text, target.language)
for name, matcher in self._matchers:
for line_number, line_text in enumerate(text.splitlines(), start=1):
if not matcher.search(line_text):
continue
handler = _resolve_handler(line_text, declared_classes)
if handler is None:
continue
records.extend(
_handler_entries(
path=target.path,
handler=handler,
call_name=name,
call_line=line_number,
text=text,
language=target.language,
)
)
return records
def registration_coverage(self, target: Any, text: str) -> dict[str, int]:
declared_classes = _declared_classes(target.path, text, target.language)
unresolved = 0
for _, matcher in self._matchers:
for line_text in text.splitlines():
if not matcher.search(line_text):
continue
if _resolve_handler(line_text, declared_classes) is None:
unresolved += 1
return {"registration_unresolved": unresolved} if unresolved else {}
def _declared_classes(path: str, text: str, language: str) -> frozenset[str]:
classes = {scope.enclosing_type for scope in declaration_scopes(path, text, language) if scope.enclosing_type is not None}
return frozenset(classes)
def _resolve_handler(line_text: str, declared_classes: frozenset[str]) -> str | None:
for class_name in sorted(declared_classes):
if re.search(rf"\b{re.escape(class_name)}\b", line_text):
return class_name
return None
def _handler_entries(
*,
path: str,
handler: str,
call_name: str,
call_line: int,
text: str,
language: str,
) -> list[EntryPointRecord]:
from .legacy import EntryPointRecord
entries = []
for scope in declaration_scopes(path, text, language):
if scope.enclosing_type != handler:
continue
entries.append(
EntryPointRecord(
path=path,
line=scope.start_line,
end_line=scope.start_line,
function_name=scope.function_name,
name=scope.function_name,
kind="route",
access_level="unknown",
trust_boundary="external",
access_evidence=[
f"registration site {path}:{call_line}",
f"registered handler {handler}",
],
conditions=[],
provenance="",
rationale="registered handler request method",
defines_handler=True,
registered_by=call_name,
)
)
return entries
__all__ = ["RegistrationSiteStrategy"]
+128
View File
@@ -0,0 +1,128 @@
"""Registration-site strategy tests (task 11.2).
The generic call-site registration adapter resolves handlers registered at
call sites through the same-file stable symbols and reports unresolved
registrations as coverage. Vocabulary is data supplied to the strategy; the
module itself names no target.
"""
from __future__ import annotations
import inspect
from openultrasast.mapping.registration_sites import RegistrationSiteStrategy
from openultrasast.mapping.registry import register, registry_entries
from openultrasast.preprocess import FileTarget
PATH = "src/Main.java"
JAVA_EXACT = """class Handler {
void handle() {
}
void respond() {
}
}
class Registry {
void install() {
addHandler(new Handler());
}
}
"""
JAVA_UNRESOLVED = """class Registry {
void install() {
addHandler(new Elsewhere());
}
}
"""
def _target(path: str = PATH, text: str = JAVA_EXACT) -> FileTarget:
return FileTarget(
path=path,
absolute_path=f"/repo/{path}",
language="java",
loc=max(len(text.splitlines()), 1),
tags=[],
has_fuzz_entry_point=False,
)
def _strategy(vocabulary: tuple[str, ...] = ("addHandler",)) -> RegistrationSiteStrategy:
return RegistrationSiteStrategy(
identifier="test-registration",
languages=("java",),
vocabulary=vocabulary,
)
def test_registered_handler_methods_become_entries() -> None:
entries = _strategy().recognize(_target(), JAVA_EXACT)
by_function = {entry.function_name: entry for entry in entries}
assert set(by_function) == {"handle", "respond"}
entry = by_function["handle"]
assert entry.path == PATH
assert entry.kind == "route"
assert entry.function_name == "handle"
assert entry.access_level == "unknown"
assert entry.access_evidence == [
f"registration site {PATH}:11",
"registered handler Handler",
]
assert entry.registered_by == "addHandler"
def test_unresolved_registration_produces_no_entry_but_coverage() -> None:
strategy = _strategy()
assert strategy.recognize(_target(text=JAVA_UNRESOLVED), JAVA_UNRESOLVED) == []
assert strategy.registration_coverage(_target(text=JAVA_UNRESOLVED), JAVA_UNRESOLVED) == {"registration_unresolved": 1}
def test_exact_registration_reports_zero_unresolved() -> None:
assert _strategy().registration_coverage(_target(), JAVA_EXACT) == {}
def test_empty_vocabulary_matches_nothing() -> None:
strategy = _strategy(vocabulary=())
assert strategy.recognize(_target(), JAVA_EXACT) == []
assert strategy.registration_coverage(_target(), JAVA_EXACT) == {}
def test_multiple_entries_are_deterministic() -> None:
strategy = _strategy()
first = strategy.recognize(_target(), JAVA_EXACT)
second = strategy.recognize(_target(), JAVA_EXACT)
assert first == second
assert [entry.function_name for entry in first] == sorted(
entry.function_name or "" for entry in first
)
def test_registry_dispatch_stamps_strategy_provenance() -> None:
strategy = _strategy()
register(strategy)
stamped = registry_entries(_target(), JAVA_EXACT)
fixture_entries = [entry for entry in stamped if entry.provenance == "strategy:test-registration"]
assert {entry.function_name for entry in fixture_entries} == {"handle", "respond"}
def test_module_names_no_targets() -> None:
from openultrasast.mapping import registration_sites
source = inspect.getsource(registration_sites).lower()
for name in (
"symfony",
"laravel",
"tornado",
"wordpress",
"stapler",
"jetty",
"django",
"flask",
"rails",
"spring",
"express",
):
assert name not in source