feat(access-control-prover): 11.1 entry-recognition strategy registry

This commit is contained in:
Marius Ciepluch
2026-10-29 20:19:31 +01:00
parent 82e4a1ad97
commit 86709daf2e
5 changed files with 248 additions and 6 deletions
+1 -1
View File
@@ -211,7 +211,7 @@ its final resolution, dominance, fact-acquisition and measurement semantics with
- _Depends: 7.4, 9.4, 10.5_
- [ ] 11. Target knowledge registry and acquisition
- [ ] 11.1 (P) Add the entry-recognition strategy registry
- [x] 11.1 (P) Add the entry-recognition strategy registry
- Route mapping dispatch through one registry without embedding target names in engine modules.
- Done when a fixture strategy registers, runs and records its strategy identifier and evidence.
- _Requirements: 1.1, 1.5, 1.7_
+5
View File
@@ -0,0 +1,5 @@
"""Entry-point mapping package: legacy recognisers plus the strategy registry."""
from .legacy import * # noqa: F401,F403
from .legacy import _php_methods as _php_methods # re-exported internal helpers (enforcement adapter)
from .legacy import _php_routes as _php_routes
@@ -8,7 +8,8 @@ from dataclasses import asdict, dataclass, replace
from pathlib import Path
from typing import Any, Literal, cast
from .preprocess import FileTarget
from ..preprocess import FileTarget
from .registry import registry_entries
MappingKind = Literal["semgrep", "codeql", "differential", "sharp_edge"]
AccessLevel = Literal[
@@ -107,6 +108,7 @@ def analyze_entry_points(
for target in targets:
text = source_texts[target.path]
records.extend(_language_entry_points(target, text))
records.extend(registry_entries(target, text))
if target.language == "php":
framework_entries = _php_framework_entries(target, text, source_texts)
names = {e.function_name for e in framework_entries}
@@ -138,7 +140,7 @@ def php_hook_callbacks(root: Path, targets: list[FileTarget], register_calls: Se
plugin system's event bus -- and a second framework is a row in a TOML, not a change to this function.
"""
if register_calls is None:
from .semantic.facts import load_facts
from ..semantic.facts import load_facts
register_calls = tuple(sorted({call for fact in load_facts().for_language("php").dispatches for call in fact.register}))
if not register_calls:
@@ -222,7 +224,7 @@ class _PhpMethod:
def _php_methods(text: str) -> list[_PhpMethod]:
"""CST class and method boundaries, including the method's own attributes and docblock."""
from .semantic.extra import grammar_for
from ..semantic.extra import grammar_for
grammar = grammar_for("php")
if grammar is None:
@@ -417,7 +419,7 @@ def _tornado_entries(target: FileTarget, text: str) -> list[EntryPointRecord]:
def _remix_exports(target: FileTarget, text: str) -> list[EntryPointRecord]:
if "app/routes/" not in "/" + target.path:
return []
from .semantic.extra import grammar_for
from ..semantic.extra import grammar_for
grammar = grammar_for("tsx" if Path(target.path).suffix == ".tsx" else target.language)
if grammar is None:
@@ -490,7 +492,7 @@ def entry_framework(entry: object) -> str | None:
def _java_methods(target: FileTarget, text: str) -> list[tuple[object, str, str]]:
"""Use the CST to keep class declarations, parameters and method bodies separate."""
from .semantic.extra import grammar_for
from ..semantic.extra import grammar_for
grammar = grammar_for("java")
if grammar is None:
+52
View File
@@ -0,0 +1,52 @@
"""Entry-recognition strategy registry (task 11.1, Reqs 1.1, 1.5, 1.7).
Target knowledge enters entry recognition through registered strategies; the
registry stamps every returned record with the strategy identifier unless the
strategy recorded its own provenance. The registry itself names no target.
"""
from dataclasses import replace
from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable
if TYPE_CHECKING:
from .legacy import EntryPointRecord
__all__ = ["EntryRecogniser", "register", "registry_entries", "registered_strategies"]
@runtime_checkable
class EntryRecogniser(Protocol):
"""A target-specific entry-recognition strategy registered from data or a strategy module."""
identifier: str
languages: tuple[str, ...]
def recognize(self, target: Any, text: str) -> list["EntryPointRecord"]:
"""Return the entry points this strategy recognises in one file."""
...
_STRATEGIES: dict[str, EntryRecogniser] = {}
def register(recogniser: EntryRecogniser) -> None:
"""Register a strategy; re-registering an identifier replaces the earlier strategy."""
_STRATEGIES[recogniser.identifier] = recogniser
def registered_strategies(language: str) -> tuple[EntryRecogniser, ...]:
"""Return the strategies that recognise entries for one language, ordered by identifier."""
return tuple(
strategy for strategy in sorted(_STRATEGIES.values(), key=lambda strategy: strategy.identifier) if language in strategy.languages
)
def registry_entries(target: Any, text: str) -> list["EntryPointRecord"]:
"""Run every matching strategy and stamp its identifier onto each returned record."""
records: list[EntryPointRecord] = []
for strategy in registered_strategies(getattr(target, "language", "")):
for record in strategy.recognize(target, text):
if record.provenance == "":
record = replace(record, provenance=f"strategy:{strategy.identifier}")
records.append(record)
return records
+183
View File
@@ -0,0 +1,183 @@
"""Entry-recognition strategy registry (task 11.1, Reqs 1.1, 1.5, 1.7)."""
from __future__ import annotations
import inspect
from pathlib import Path
import pytest
from openultrasast.mapping import EntryPointRecord
from openultrasast.mapping.registry import (
EntryRecogniser,
register,
registered_strategies,
registry_entries,
)
from openultrasast.preprocess import FileTarget, detect_language
@pytest.fixture(autouse=True)
def _clean_registry() -> None:
"""Each test registers its own strategies into the shared module registry."""
from openultrasast.mapping import registry
registry._STRATEGIES.clear()
class FixtureRouteStrategy:
"""A recogniser defined by the test, named nowhere in engine code."""
identifier = "test-fixture"
languages: tuple[str, ...] = ("java",)
def recognize(self, target: FileTarget, text: str) -> list[EntryPointRecord]:
marker = "fixture marker line 3"
if marker not in text:
return []
return [
EntryPointRecord(
path=target.path,
line=3,
end_line=3,
function_name="fixtureHandler",
name="fixtureHandler",
kind="route",
access_level="unknown",
trust_boundary="external",
access_evidence=[marker],
conditions=[],
provenance="",
rationale="fixture strategy",
)
]
class SelfProvenanceStrategy:
identifier = "test-self-provenance"
languages: tuple[str, ...] = ("java",)
def recognize(self, target: FileTarget, text: str) -> list[EntryPointRecord]:
return [
EntryPointRecord(
path=target.path,
line=1,
end_line=1,
function_name="selfProvenanceHandler",
name="selfProvenanceHandler",
kind="route",
access_level="unknown",
trust_boundary="external",
access_evidence=["self provenance"],
conditions=[],
provenance="strategy-own:declaration",
rationale="fixture",
)
]
def _target(path: str, language: str) -> FileTarget:
return FileTarget(
path=path,
absolute_path=f"/repo/{path}",
language=language,
loc=8,
tags=[],
has_fuzz_entry_point=False,
)
JAVA_TEXT = "package p;\nclass H {\n // fixture marker line 3\n void fixtureHandler() {}\n}\n"
def test_registered_strategy_is_listed_for_its_language() -> None:
register(FixtureRouteStrategy())
strategies = registered_strategies("java")
assert [strategy.identifier for strategy in strategies] == ["test-fixture"]
assert registered_strategies("php") == ()
def test_registry_entries_stamp_strategy_identifier_and_keep_evidence() -> None:
register(FixtureRouteStrategy())
entries = registry_entries(_target("src/H.java", "java"), JAVA_TEXT)
assert len(entries) == 1
assert entries[0].provenance == "strategy:test-fixture"
assert entries[0].access_evidence == ["fixture marker line 3"]
assert entries[0].function_name == "fixtureHandler"
def test_strategy_set_provenance_is_preserved() -> None:
register(SelfProvenanceStrategy())
entries = registry_entries(_target("src/H.java", "java"), "class A {}")
assert entries[0].provenance == "strategy-own:declaration"
def test_re_registration_replaces_the_strategy() -> None:
register(FixtureRouteStrategy())
register(FixtureRouteStrategy())
assert [s.identifier for s in registered_strategies("java")] == ["test-fixture"]
def test_registry_entries_are_deterministic() -> None:
register(FixtureRouteStrategy())
first = registry_entries(_target("src/H.java", "java"), JAVA_TEXT)
second = registry_entries(_target("src/H.java", "java"), JAVA_TEXT)
assert first == second
def test_registry_module_names_no_targets() -> None:
from openultrasast.mapping import registry
source = inspect.getsource(registry)
for target_name in (
"symfony",
"laravel",
"tornado",
"wordpress",
"stapler",
"jetty",
"django",
"flask",
"rails",
"spring",
"express",
):
assert target_name not in source.lower()
def test_dispatch_runs_registered_strategies_alongside_legacy_blocks(tmp_path: Path) -> None:
from openultrasast.mapping import analyze_entry_points
(tmp_path / "H.java").write_text(
"package p;\n"
"import hudson.model.Descriptor;\n"
"class H extends Descriptor<Thing> {\n"
" // fixture marker line 3\n"
" public void doTest(org.kohsuke.stapler.StaplerRequest req) {}\n"
"}\n",
encoding="utf-8",
)
targets = [
FileTarget(
path="H.java",
absolute_path=str(tmp_path / "H.java"),
language=detect_language(tmp_path / "H.java"),
loc=6,
tags=[],
has_fuzz_entry_point=False,
)
]
register(FixtureRouteStrategy())
entries = analyze_entry_points(tmp_path, targets, include_manifests=False)
names = {entry.function_name for entry in entries}
assert "doTest" in names # legacy Stapler block still runs
assert "fixtureHandler" in names # registry strategy contributed
fixture = next(entry for entry in entries if entry.function_name == "fixtureHandler")
assert fixture.provenance == "strategy:test-fixture"
stapler = next(entry for entry in entries if entry.function_name == "doTest")
assert stapler.provenance != "strategy:test-fixture"
def test_entry_recogniser_protocol_is_runtime_implemented() -> None:
strategy = FixtureRouteStrategy()
assert isinstance(strategy, EntryRecogniser)