From 86709daf2e5c4996d32b9d57123e90b736e10dce Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Thu, 29 Oct 2026 20:19:31 +0100 Subject: [PATCH] feat(access-control-prover): 11.1 entry-recognition strategy registry --- .kiro/specs/access-control-prover/tasks.md | 2 +- src/openultrasast/mapping/__init__.py | 5 + .../{mapping.py => mapping/legacy.py} | 12 +- src/openultrasast/mapping/registry.py | 52 +++++ tests/test_mapping_registry.py | 183 ++++++++++++++++++ 5 files changed, 248 insertions(+), 6 deletions(-) create mode 100644 src/openultrasast/mapping/__init__.py rename src/openultrasast/{mapping.py => mapping/legacy.py} (99%) create mode 100644 src/openultrasast/mapping/registry.py create mode 100644 tests/test_mapping_registry.py diff --git a/.kiro/specs/access-control-prover/tasks.md b/.kiro/specs/access-control-prover/tasks.md index 09dc9ca..aaa2ac8 100644 --- a/.kiro/specs/access-control-prover/tasks.md +++ b/.kiro/specs/access-control-prover/tasks.md @@ -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_ diff --git a/src/openultrasast/mapping/__init__.py b/src/openultrasast/mapping/__init__.py new file mode 100644 index 0000000..1da771d --- /dev/null +++ b/src/openultrasast/mapping/__init__.py @@ -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 diff --git a/src/openultrasast/mapping.py b/src/openultrasast/mapping/legacy.py similarity index 99% rename from src/openultrasast/mapping.py rename to src/openultrasast/mapping/legacy.py index 78a8360..66ea8d0 100644 --- a/src/openultrasast/mapping.py +++ b/src/openultrasast/mapping/legacy.py @@ -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: diff --git a/src/openultrasast/mapping/registry.py b/src/openultrasast/mapping/registry.py new file mode 100644 index 0000000..af634cb --- /dev/null +++ b/src/openultrasast/mapping/registry.py @@ -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 diff --git a/tests/test_mapping_registry.py b/tests/test_mapping_registry.py new file mode 100644 index 0000000..615d65f --- /dev/null +++ b/tests/test_mapping_registry.py @@ -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 {\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)