diff --git a/.kiro/specs/access-control-prover/tasks.md b/.kiro/specs/access-control-prover/tasks.md index e54d6e6..f845406 100644 --- a/.kiro/specs/access-control-prover/tasks.md +++ b/.kiro/specs/access-control-prover/tasks.md @@ -223,7 +223,7 @@ its final resolution, dominance, fact-acquisition and measurement semantics with - _Requirements: 1.2, 1.3, 1.4, 1.5, 1.6_ - _Boundary: Registration Adapter_ - _Depends: 8.4, 11.1_ -- [ ] 11.3 (P) Extract existing target knowledge into candidate records +- [x] 11.3 (P) Extract existing target knowledge into candidate records - Read legacy recognition strategies and current facts without changing promoted outputs. - Done when extraction produces deterministic candidate records with source/version/region evidence. - _Requirements: 2.9_ diff --git a/src/openultrasast/tools/__init__.py b/src/openultrasast/tools/__init__.py new file mode 100644 index 0000000..c6059bd --- /dev/null +++ b/src/openultrasast/tools/__init__.py @@ -0,0 +1 @@ +"""Tooling package for the access-control-prover acquisition workflow.""" diff --git a/src/openultrasast/tools/fact_init.py b/src/openultrasast/tools/fact_init.py new file mode 100644 index 0000000..52b2c72 --- /dev/null +++ b/src/openultrasast/tools/fact_init.py @@ -0,0 +1,112 @@ +"""Task 11.3: extract existing target knowledge into candidate records (Req 2.9). + +Extraction mode of the initialization job: the legacy recognition strategies in +``mapping/legacy.py`` and the promoted fact rows in ``ruleset/obligations/*.toml`` +are read -- never modified -- and returned as deterministic candidate records +carrying source, version and region evidence. Verification and promotion are the +acquisition workflow's job (tasks 11.4+), not extraction's. + +This is a tools module, not an engine module: naming recognition targets here is +legitimate (the whole point is to extract their vocabulary into data). +""" + +import hashlib +import re +import tomllib +from collections.abc import Iterable, Iterator +from dataclasses import dataclass +from pathlib import Path + +_RECOGNISER_PATTERN = re.compile( + r"^(def (_[a-z][a-z0-9_]*_entries)\(|def (_python_registered_routes)\(|(_PHP_REGISTRATION) = )", + re.MULTILINE, +) + + +@dataclass(frozen=True) +class CandidateRecord: + """A unit of extracted target knowledge awaiting verification (Req 2.9).""" + + kind: str # 'operation' | 'discharger' | 'implication' | 'entry-strategy' + name: str # fact id or recogniser function/constant name + source: str # file the knowledge was read from (relative-style path tail) + version: str # sha256[:12] of the source file's bytes + region: str # '::' or 'legacy.py::-' + payload: tuple[tuple[str, str], ...] = () # sorted field pairs + + +def extract_fact_rows(ruleset_dir: Path) -> tuple[CandidateRecord, ...]: + """Extract operation, discharger and implication rows as candidates.""" + records: list[CandidateRecord] = [] + for path in sorted(ruleset_dir.glob("*.toml")): + data = tomllib.loads(path.read_text(encoding="utf-8")) + version = _digest(path) + records.extend(_row_records(path.name, version, data)) + return tuple(records) + + +def extract_legacy_strategies(mapping_path: Path) -> tuple[CandidateRecord, ...]: + """Extract the legacy recogniser inventory (symbol regions) as candidates.""" + text = mapping_path.read_text(encoding="utf-8") + version = _digest(mapping_path) + records = [ + CandidateRecord( + kind="entry-strategy", + name=symbol, + source=mapping_path.name, + version=version, + region=f"legacy.py:{symbol}:{start}-{end}", + ) + for symbol, start, end in _symbol_spans(text) + ] + return tuple(records) + + +def extract_all(repo_root: Path) -> tuple[CandidateRecord, ...]: + """Extract every knowledge source: fact rows and legacy strategies.""" + ruleset_dir = repo_root / "src" / "openultrasast" / "ruleset" / "obligations" + mapping_path = repo_root / "src" / "openultrasast" / "mapping" / "legacy.py" + records = [*extract_fact_rows(ruleset_dir), *extract_legacy_strategies(mapping_path)] + return tuple(sorted(records, key=lambda record: (record.kind, record.name, record.region))) + + +def _digest(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest()[:12] + + +def _row_records(file_name: str, version: str, data: dict) -> Iterator[CandidateRecord]: + tables = (("operation", "operation"), ("discharger", "discharger"), ("implication", "implication")) + for kind, table in tables: + for row in data.get(table, []): + row_id = str(row.get("id", row.get("kind", ""))) + yield CandidateRecord( + kind=kind, + name=row_id, + source=file_name, + version=version, + region=f"{file_name}:{table}:{row_id}", + payload=_payload(row), + ) + + +def _payload(row: dict) -> tuple[tuple[str, str], ...]: + pairs = [] + for key, value in row.items(): + if isinstance(value, (list, tuple)): + pairs.append((key, ",".join(str(item) for item in value))) + else: + pairs.append((key, str(value))) + return tuple(sorted(pairs)) + + +def _symbol_spans(text: str) -> Iterable[tuple[str, int, int]]: + matches = list(_RECOGNISER_PATTERN.finditer(text)) + lines = text.splitlines() + for index, match in enumerate(matches): + symbol = match.group(2) or match.group(3) or match.group(4) + start = text.count("\n", 0, match.start()) + 1 + end_line = text.count("\n", 0, matches[index + 1].start()) + 1 if index + 1 < len(matches) else len(lines) + yield symbol, start, end_line - 1 + + +__all__ = ["CandidateRecord", "extract_all", "extract_fact_rows", "extract_legacy_strategies"] diff --git a/tests/test_fact_init.py b/tests/test_fact_init.py new file mode 100644 index 0000000..df22d0e --- /dev/null +++ b/tests/test_fact_init.py @@ -0,0 +1,119 @@ +"""Task 11.3: extraction of existing target knowledge into candidate records (Req 2.9). + +Extraction reads the legacy recognition strategies and the promoted fact rows and +returns deterministic candidate records carrying source, version and region +evidence. It never modifies promoted outputs. +""" + +import hashlib +import re +from pathlib import Path + +import pytest + +from openultrasast.tools.fact_init import ( + CandidateRecord, + extract_all, + extract_fact_rows, + extract_legacy_strategies, +) + +REPO_ROOT = Path(__file__).parents[1] +RULESET_DIR = REPO_ROOT / "src" / "openultrasast" / "ruleset" / "obligations" +MAPPING_FILE = REPO_ROOT / "src" / "openultrasast" / "mapping" / "legacy.py" + +_VERSION_PATTERN = re.compile(r"^[0-9a-f]{12}$") + + +def _file_digest(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest()[:12] + + +def _digest_dir(directory: Path) -> dict[str, str]: + return {path.name: _file_digest(path) for path in sorted(directory.glob("*.toml"))} + + +def test_fact_rows_extraction_yields_operation_and_discharger_records() -> None: + records = extract_fact_rows(RULESET_DIR) + kinds = {record.kind for record in records} + assert "operation" in kinds + assert "discharger" in kinds + assert records + + +def test_extraction_includes_java_operation_rows() -> None: + records = extract_fact_rows(RULESET_DIR) + java_operations = [record for record in records if record.kind == "operation" and record.source.endswith("java.toml")] + assert java_operations + + +def test_legacy_strategy_extraction_names_recognisers() -> None: + records = extract_legacy_strategies(MAPPING_FILE) + names = {record.name for record in records} + assert "_stapler_entries" in names + assert "_jetty_entries" in names + assert "_python_registered_routes" in names + for record in records: + assert record.region.startswith("legacy.py:") + + +def test_every_candidate_carries_source_version_and_region() -> None: + records = extract_all(REPO_ROOT) + assert records + for record in records: + assert record.source + assert _VERSION_PATTERN.match(record.version) + assert ":" in record.region + + +def test_version_digest_is_file_derived_and_stable() -> None: + first = extract_fact_rows(RULESET_DIR) + second = extract_fact_rows(RULESET_DIR) + java_rows = [record for record in first if record.source.endswith("java.toml")] + assert java_rows + expected = _file_digest(RULESET_DIR / "java.toml") + assert all(record.version == expected for record in java_rows) + assert first == second + + +def test_extraction_is_deterministic_and_sorted() -> None: + first = extract_all(REPO_ROOT) + second = extract_all(REPO_ROOT) + assert first == second + keys = [(record.kind, record.name, record.region) for record in first] + assert keys == sorted(keys) + + +def test_extraction_changes_no_promoted_outputs() -> None: + from openultrasast.mapping import registry + + tomls_before = _digest_dir(RULESET_DIR) + mapping_before = _file_digest(MAPPING_FILE) + strategies_before = len(registry._STRATEGIES) + + extract_all(REPO_ROOT) + + assert _digest_dir(RULESET_DIR) == tomls_before + assert _file_digest(MAPPING_FILE) == mapping_before + assert len(registry._STRATEGIES) == strategies_before + + +def test_operation_payload_carries_row_fields() -> None: + records = extract_fact_rows(RULESET_DIR) + java_operations = [record for record in records if record.kind == "operation" and record.source.endswith("java.toml")] + fields = {name for name, _ in java_operations[0].payload} + assert "calls" in fields + assert "kind" in fields + + +def test_candidate_record_is_frozen() -> None: + record = CandidateRecord( + kind="operation", + name="example", + source="java.toml", + version="0" * 12, + region="java.toml:operations:example", + payload=(("language", "java"),), + ) + with pytest.raises(Exception): # noqa: B017 + record.name = "other" # type: ignore[misc]