diff --git a/.kiro/specs/access-control-prover/tasks.md b/.kiro/specs/access-control-prover/tasks.md index 6833486..62fb4ec 100644 --- a/.kiro/specs/access-control-prover/tasks.md +++ b/.kiro/specs/access-control-prover/tasks.md @@ -18,7 +18,7 @@ - _Requirements: 4.1a_ - _Boundary: obligations.check (opt-in only)_ -- [ ] 2. Entry recognisers +- [x] 2. Entry recognisers - [x] 2.1 Stapler and Jetty recognisers in `mapping.py` (P) - Public `do[A-Z]...` methods on exposed types or with Stapler-typed or `@QueryParameter` parameters; Jetty `AbstractHandler`, `HttpServlet`, JAX-RS resources; evidence strings; `public` only for `UnprotectedRootAction` @@ -26,13 +26,13 @@ helper named `doX`, and the existing mapping tests still pass - _Requirements: 1.1, 1.6, 1.7, 1.8_ - _Boundary: mapping_ -- [ ] 2.2 Symfony, Livewire and Laravel controller, Remix export and Tornado recognisers in `mapping.py` (P) +- [x] 2.2 Symfony, Livewire and Laravel controller, Remix export and Tornado recognisers in `mapping.py` (P) - Done when fixture files of each shape yield `route` entries with evidence and the negative fixtures yield none - _Requirements: 1.2, 1.3, 1.4, 1.5, 1.7, 1.8_ - _Boundary: mapping_ - [ ] 3. Enforcement witnesses and facts -- [ ] 3.1 `semantic/obligations/enforcement.py` producing router and decorator `Discharge` witnesses (P) +- [x] 3.1 `semantic/obligations/enforcement.py` producing router and decorator `Discharge` witnesses (P) - Laravel route middleware and controller `middleware`/`authorizeResource`; Symfony `IsGranted`, `Security`, security.yaml `access_control`; Tornado `authenticated`; names listed in the facts headers - Done when `tests/test_obligation_enforcement.py` shows a guarded Laravel route, a Symfony attribute and a @@ -50,15 +50,15 @@ - _Requirements: 2.3, 2.4, 2.7_ - _Boundary: ruleset/obligations_ -- [ ] 3.3 Handler-named operations: `handlers` and `framework` fields in the facts loader, `find_operations(..., handlers_of=)`, Stapler form-validation rows in `java.toml`, Tornado handler rows in `python.toml`, Remix action rows in `typescript.toml` +- [x] 3.3 Handler-named operations: `handlers` and `framework` fields in the facts loader, `find_operations(..., handlers_of=)`, Stapler form-validation rows in `java.toml`, Tornado handler rows in `python.toml`, Remix action rows in `typescript.toml` - Done when `tests/test_obligation_facts.py` loads a handler-named row, `tests/test_obligation_operations.py` (or the existing operations test file) yields the operation at the declaration line only for listed handlers, and the prover smoke on the 23 fixes reaches and checks the Jenkins form-validation methods - _Requirements: 2.8_ - _Boundary: obligations.facts, obligations.operations_ -- [ ] 4. The prover -- [ ] 4.1 `hunt/prover.py`: `prove_change` over the hunter `Context`, `Coverage`, findings as hunt defects +- [x] 4. The prover +- [x] 4.1 `hunt/prover.py`: `prove_change` over the hunter `Context`, `Coverage`, findings as hunt defects - Touched functions from head lines, build output left out, targets include direct callees holding operations, IRs of hop files passed to the checker, extra witnesses from enforcement, coverage counts by reason, no model - Done when `tests/test_hunt_prover.py` shows: an unguarded Stapler handler calling `lookupCredentials` gives @@ -103,3 +103,40 @@ requested static gate passed (format, Ruff and mypy); the exact pytest gate passed with 112 passed, zero failed, exit 0. UV_NO_SYNC=1 was set; UV_CACHE_DIR was placed under /tmp because the default cache is outside the writable sandbox. The unchanged unjustified-path checker test remains intact. + +- Delivery B (2026-10-14) implements task 2.2's Symfony, Livewire/Laravel, Remix and Tornado + entries, task 3.1's scoped enforcement witnesses, task 3.3's handler facts and operation discovery, + and task 4.1's integration. Changes remain uncommitted as the brief requires. +- Handler facts are explicitly opt-in through file-qualified handler maps in the checker. The prover + uses handler obligations where a Stapler or Tornado entry reaches no listed call operation; existing + call-site findings and Delivery A tests remain unchanged. Remix retains its handler operation even + when listed calls exist, and independently requires a path guard and an ownership check. Public + access waives only the path guard for these handler facts. The checker default is unchanged. +- The prover supplies a local dominance adapter for handler facts: it checks guards within the handler + body while reporting the operation at its declaration. A final ordinary return does not turn a bare + permission predicate into a guard. OrderDominance and the existing producers and arms are unchanged. +- Remix arrow exports are named locally in the prover's IR copy, including TSX route modules. Symfony + recognises positional and named route paths and class prefixes. Tornado recognises local subclasses + and the pool terminal's terminado base when it explicitly invokes Tornado's WebSocketHandler method. +- PHP CST Point.row access crashed on the larger Kimai controller with the installed native parser. + New recognisers derive line numbers from byte offsets instead; a fixture past line 320 and the + 23-change source smoke complete without a crash. Existing CST code was not modified. +- Task 3.2 remains partially complete: TypeScript adds clone-verified requireUserId and + authenticateApiRequest, retaining requireUser and authorize; PHP rows are retained and their header + records enforcement sources. Python retains the existing authenticated decorator and gains the + Tornado handler operation. No get_current_user fact was added: Git searches of Mage HEAD, the pool + revision and websocket fix 66440e4d3 found neither that API nor an authenticated decorator use, and + local framework documentation was unavailable. The brief forbids network access and unsourced names. +- Review used the main-context fallback because the brief prohibits model calls. Verdict: APPROVED + for tasks 2.2, 3.1, 3.3 and 4.1; task 3.2 remains open for its source-verification gap. No new + placeholder markers, credentials, dependencies, producer edits or arm edits were introduced. +- Fresh verification: UV_NO_SYNC=1 and UV_CACHE_DIR=/tmp/ousast-uv. The exact static gate passed: + ruff format --check ., ruff check ., mypy src/openultrasast (178 source files). The exact requested + pytest gate plus test_obligation_operations.py passed with 145 passed, zero failed, exit 0. + Initial missing-feature tests produced 24 failed and 14 passed; integration tests initially produced + eight failed and 20 passed. No test_search_* test was run. +- A separate file-scoped smoke read immutable Git source objects for all 23 access-control changes, + validated their commit objects and source digests, and recorded nonzero input bytes for every change. + All 23 executions completed, zero errors; 16 changes checked operations and seven remained unchecked. + It materialised only changed source files in temporary directories, so it is not a full-repository + recall measurement. The maintainer-owned task 6.1 and release readiness remain unclaimed. diff --git a/src/openultrasast/hunt/prover.py b/src/openultrasast/hunt/prover.py index 082b418..c5c91e0 100644 --- a/src/openultrasast/hunt/prover.py +++ b/src/openultrasast/hunt/prover.py @@ -2,9 +2,10 @@ from __future__ import annotations +import re from collections import Counter from collections.abc import Mapping -from dataclasses import asdict, dataclass +from dataclasses import asdict, dataclass, replace from pathlib import Path from typing import Any @@ -14,8 +15,9 @@ from ..semantic.facts import load_facts from ..semantic.ir import FileIR, FunctionIR, trailing_name from ..semantic.obligations.check import _on_caller_hop, _relevant, check_obligations from ..semantic.obligations.dominance import OrderDominance +from ..semantic.obligations.enforcement import witnesses from ..semantic.obligations.facts import load_obligation_facts -from ..semantic.obligations.operations import Operation, find_operations, valid_witness +from ..semantic.obligations.operations import Discharge, Operation, find_discharges, find_operations, valid_witness from ..semantic.obligations.paths import compose_paths from ..semantic.obligations.policy import DEFAULT_POLICY_RELPATH, load_declared_policy from .producers import Context, _safe_path @@ -70,35 +72,30 @@ def prove_change(root: Path, change: Mapping[str, Any], ctx: Context | None = No readable.append(path) except Exception as error: failed.add(path) - reasons[str(getattr(error, "reason", None) or type(error).__name__)] += 1 - degradations.append({"path": path, "reason": str(getattr(error, "reason", None) or type(error).__name__)}) + reasons[type(error).__name__] += 1 + degradations.append({"path": path, "reason": type(error).__name__}) + for config in mapping.framework_config_paths(root): + try: + ctx.read(config) + except Exception as error: + degradations.append({"path": config, "reason": type(error).__name__}) if "entry_points" not in ctx.__dict__: - # External manifest enforcement is supplied by delivery B; this delivery uses source entries. - mapped: list[mapping.EntryPointRecord] = [] - for target in targets: - try: - mapped.extend(mapping.analyze_entry_points(root, [target], texts=ctx.texts, include_manifests=False)) - except Exception as error: - failed.add(target.path) - reasons[str(getattr(error, "reason", None) or type(error).__name__)] += 1 - degradations.append({"path": target.path, "reason": str(getattr(error, "reason", None) or type(error).__name__)}) - ctx.__dict__["entry_points"] = mapped - entries = [entry for entry in ctx.entry_points if entry.function_name and entry.defines_handler and entry.line is not None] + ctx.__dict__["entry_points"] = mapping.analyze_entry_points(root, targets, texts=ctx.texts, include_manifests=False) + entries = [ + entry + for entry in ctx.entry_points + if entry.function_name and entry.defines_handler and entry.line is not None and entry.kind == "route" + ] recognised = {entry.path for entry in entries} - for path in recognised: - evidence = " ".join(item for entry in entries if entry.path == path for item in entry.access_evidence) - framework = ( - "stapler" - if "Stapler" in evidence - else "jetty" - if "Jetty" in evidence - else "servlet" - if "Servlet" in evidence - else "jaxrs" - if "JAX-RS" in evidence - else source.canonical_language(detect_language(root / path)) - ) - frameworks[framework] += 1 + handlers_of: dict[str, dict[str, str]] = {} + framework_files: set[tuple[str, str]] = set() + for entry in entries: + framework = mapping.entry_framework(entry) + if framework is None: + framework = source.canonical_language(detect_language(root / entry.path)) + framework_files.add((entry.path, framework)) + handlers_of.setdefault(entry.path, {})[str(entry.function_name)] = framework + frameworks.update(framework for _, framework in framework_files) def read_ir(path: str) -> FileIR | None: if path in failed: @@ -107,6 +104,14 @@ def prove_change(root: Path, change: Mapping[str, Any], ctx: Context | None = No return irs[path][0] try: ir, text = ctx.ir(path), ctx.read(path) + aliases = {e.line: e.function_name for e in entries if e.path == path and mapping.entry_framework(e) == "remix"} + ir = replace( + ir, + functions=tuple( + replace(fn, name=str(aliases[fn.start_line])) if fn.name == "" and fn.start_line in aliases else fn + for fn in ir.functions + ), + ) if not ir.parse_ok: reasons["language_unsupported" if ir.reason == "language_unsupported" else "parse_failed"] += 1 failed.add(path) @@ -117,16 +122,16 @@ def prove_change(root: Path, change: Mapping[str, Any], ctx: Context | None = No return None irs[path] = ir, text ctx.spans.setdefault(path, source.ranges_from_ir(ir, text)) - found = find_operations(ir, facts.for_language(ir.language), text=text) + found = find_operations(ir, facts.for_language(ir.language), text=text, handlers_of=handlers_of.get(path)) for fn in ir.functions: if fn.name != "": functions[path, fn.name] = fn operations[path, fn.name] = tuple(op for op in found if op.function == fn.name) return ir except Exception as error: - reasons[str(getattr(error, "reason", None) or type(error).__name__)] += 1 + reasons[type(error).__name__] += 1 failed.add(path) - degradations.append({"path": path, "reason": str(getattr(error, "reason", None) or type(error).__name__)}) + degradations.append({"path": path, "reason": type(error).__name__}) return None for file in change.get("files", ()): @@ -171,12 +176,24 @@ def prove_change(root: Path, change: Mapping[str, Any], ctx: Context | None = No if any(span.name in call_names for span in ctx.ranges(path)): read_ir(path) except Exception as error: - reasons[str(getattr(error, "reason", None) or type(error).__name__)] += 1 - degradations.append({"path": path, "reason": str(getattr(error, "reason", None) or type(error).__name__)}) + reasons[type(error).__name__] += 1 + degradations.append({"path": path, "reason": type(error).__name__}) target_functions.update(key for key, ops in operations.items() if key[1] in call_names and ops) paths, unreached = compose_paths( callers=callers, entries={f"{e.path}::{e.function_name}" for e in entries}, targets=sorted(target_functions), max_depth=max_depth ) + # Handler obligations fill gaps in the call inventory. Preserve the established call-site verdict + # when that entry already reaches a listed operation, including one in a directly called helper. + handler_ids = {fact.id for fact in facts.operations if fact.handlers} + for record in paths: + target_path, _, target_name = record.hops[-1][0].rpartition("::") + if any(op.fact_id not in handler_ids for op in operations.get((target_path, target_name), ())): + entry_path, _, entry_name = record.entry.rpartition("::") + if handlers_of.get(entry_path, {}).get(entry_name) != "remix": + handlers_of[entry_path].pop(entry_name, None) + for (path, name), ops in operations.items(): + if name not in handlers_of.get(path, {}): + operations[path, name] = tuple(op for op in ops if op.fact_id not in handler_ids) path_nodes = {node for record in paths for node, _ in record.hops} for node in path_nodes: read_ir(node.rpartition("::")[0]) @@ -185,7 +202,35 @@ def prove_change(root: Path, change: Mapping[str, Any], ctx: Context | None = No reasons["unreached" if path in recognised else "framework_unrecognised"] += 1 entry_index = {f"{e.path}::{e.function_name}": e for e in entries} policy = load_declared_policy(_safe_path(root, DEFAULT_POLICY_RELPATH.as_posix())) - dominance = OrderDominance(ctx.texts) + handler_ends: dict[Operation, int] = {} + for (path, name), ops in operations.items(): + fn = functions[path, name] + returns = [ + index + for index, line in enumerate(ctx.texts[path].splitlines(), 1) + if fn.start_line <= index <= fn.end_line and re.match(r"^\s*return\b", line) + ] + for op in ops: + if op.fact_id in handler_ids: + handler_ends[op] = returns[-1] if returns else fn.end_line + 1 + dominance = _HandlerDominance(ctx.texts, handler_ends) + extra_witnesses: list[Discharge] = [] + for language in ("php", "python"): + try: + extra_witnesses.extend(witnesses(root=root, texts=ctx.texts, entries=entries, language=language)) + except Exception as error: + degradations.append({"stage": "enforcement", "reason": type(error).__name__}) + # Boolean permission predicates need an exit; they are not throwing guards. + for op in handler_ends: + ir, text = irs[op.path] + for witness in find_discharges(ir, facts.for_language(ir.language), flow.for_language(ir.language), text=text): + if ( + witness.function == op.function + and witness.kind == "ownership_check" + and dominance.dominates(witness, op, None) + and handlers_of.get(op.path, {}).get(op.function) == "stapler" + ): + extra_witnesses.append(replace(witness, kind="path_guard")) findings: dict[str, dict[str, Any]] = {} checked: set[Operation] = set() undischarged: set[Operation] = set() @@ -204,7 +249,8 @@ def prove_change(root: Path, change: Mapping[str, Any], ctx: Context | None = No dominance=dominance, store_shapes=(), min_siblings=3, - extra_witnesses=(), + extra_witnesses=extra_witnesses, + handlers_of=handlers_of, assume_entries_protected=True, ) degradations.extend(result.degradations) @@ -212,13 +258,15 @@ def prove_change(root: Path, change: Mapping[str, Any], ctx: Context | None = No for op in operations.get((target_path, target_name), ()): checked.add(op) # Silence from a public premise is not evidence that a check dominated the operation. - if not any( - w.kind in op.requires + dominated = { + w.kind + for w in result.discharges + if w.kind in op.requires and valid_witness(w) and dominance.dominates(w, op, record) and (_relevant(w, op) or _on_caller_hop(w, record, op, dominance)) - for w in result.discharges - ): + } + if (set(op.requires) - dominated) if op.fact_id in handler_ids else not dominated: undischarged.add(op) for finding in result.findings: op = finding.operation @@ -232,7 +280,7 @@ def prove_change(root: Path, change: Mapping[str, Any], ctx: Context | None = No if call.line == op.line and any(call.name == name or call.name.endswith("." + name) for name in op_fact.calls) } ) - entry = asdict(entry_index[record.entry]) + entry_payload = asdict(entry_index[record.entry]) hops = [list(hop) for hop in record.hops] findings.setdefault( identity, @@ -247,7 +295,7 @@ def prove_change(root: Path, change: Mapping[str, Any], ctx: Context | None = No f"operation {op.fact_id} ({op.kind}) at {op.path}:{op.line}; missing {finding.missing}", ], id=identity, - entry=entry, + entry=entry_payload, hops=hops, operation=dict(id=op.fact_id, kind=op.kind, calls=calls, line=op.line), missing=finding.missing, @@ -265,3 +313,19 @@ def prove_change(root: Path, change: Mapping[str, Any], ctx: Context | None = No dict(reasons), ) return ProverResult(tuple(findings.values()), coverage, tuple(degradations)) + + +class _HandlerDominance: + """Judge a handler's internal guard over its body while retaining its declaration as the finding location. + + Call operations retain ordinary order dominance. The checker default and OrderDominance are unchanged. + """ + + def __init__(self, texts: Mapping[str, str], handler_ends: Mapping[Operation, int]) -> None: + self.order = OrderDominance(texts) + self.handler_ends = handler_ends + + def dominates(self, witness: Discharge, operation: Operation, path: object | None) -> bool: + end = self.handler_ends.get(operation) + measured = replace(operation, line=end) if end is not None else operation + return self.order.dominates(witness, measured, path) diff --git a/src/openultrasast/mapping.py b/src/openultrasast/mapping.py index 7d66257..78a8360 100644 --- a/src/openultrasast/mapping.py +++ b/src/openultrasast/mapping.py @@ -100,10 +100,18 @@ def analyze_entry_points( root: Path, targets: list[FileTarget], *, texts: Mapping[str, str] | None = None, include_manifests: bool = True ) -> list[EntryPointRecord]: records: list[EntryPointRecord] = [] + source_texts = dict(texts) if texts is not None else {t.path: (root / t.path).read_text(errors="ignore") for t in targets} + if texts is None: + for config in framework_config_paths(root): + source_texts[config] = (root / config).read_text(errors="ignore") for target in targets: - path = root / target.path - text = texts[target.path] if texts is not None else path.read_text(errors="ignore") + text = source_texts[target.path] records.extend(_language_entry_points(target, text)) + if target.language == "php": + framework_entries = _php_framework_entries(target, text, source_texts) + names = {e.function_name for e in framework_entries} + records = [e for e in records if not (e.path == target.path and e.kind == "function" and e.function_name in names)] + records.extend(framework_entries) records.extend(_tag_entry_points(target)) if include_manifests: records.extend(_route_manifest_entry_points(root, targets)) @@ -178,9 +186,9 @@ def _language_entry_points(target: FileTarget, text: str) -> list[EntryPointReco if target.language == "java": return _stapler_entries(target, text) + _jetty_entries(target, text) if target.language == "python": - return _python_entry_points(target, text) + return _python_entry_points(target, text) + _tornado_entries(target, text) if target.language in {"javascript", "typescript"}: - return _js_entry_points(target, text) + return _js_entry_points(target, text) + _remix_exports(target, text) if target.language in {"c", "cpp"}: return _c_entry_points(target, text) if target.language == "solidity": @@ -190,6 +198,296 @@ def _language_entry_points(target: FileTarget, text: str) -> list[EntryPointReco return [] +def framework_config_paths(root: Path) -> list[str]: + """Small, declared configuration locations outside the product source census.""" + candidates = [ + root / "config" / name for name in ("routes.yaml", "routes.yml", "routes.xml", "packages/security.yaml", "packages/security.yml") + ] + candidates.extend((root / "config/routes").glob("*.yaml")) + candidates.extend((root / "config/routes").glob("*.yml")) + return sorted(path.relative_to(root).as_posix() for path in candidates if path.is_file() and not path.is_symlink()) + + +@dataclass(frozen=True) +class _PhpMethod: + owner: str + owner_header: str + owner_attributes: str + name: str + line: int + end_line: int + header: str + body: str + + +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 + + grammar = grammar_for("php") + if grammar is None: + return [] + from tree_sitter import Parser + + data = text.encode() + parser = Parser(grammar) + tree = parser.parse(data) + if tree.root_node.has_error: + return [] + result: list[_PhpMethod] = [] + + def value(node: Any) -> str: + return data[node.start_byte : node.end_byte].decode() + + def walk(node: Any, owner: str = "", header: str = "", attributes: str = "", prior_comment: str = "") -> None: + if node.type == "class_declaration": + name, body = node.child_by_field_name("name"), node.child_by_field_name("body") + owner = value(name) if name is not None else "" + header = data[node.start_byte : body.start_byte if body else node.end_byte].decode() + attributes = header + prior_comment + if node.type == "method_declaration" and owner: + name, body = node.child_by_field_name("name"), node.child_by_field_name("body") + if name is not None and body is not None: + own_header = data[node.start_byte : body.start_byte].decode() + declaration = re.search(r"\bfunction\b", own_header) + line = data[: node.start_byte].count(b"\n") + 1 + own_header[: declaration.start() if declaration else 0].count("\n") + own_header = prior_comment + own_header + result.append( + _PhpMethod( + owner, header, attributes, value(name), line, data[: node.end_byte].count(b"\n") + 1, own_header, value(body) + ) + ) + return + previous_comment = "" + for child in node.named_children: + walk(child, owner, header, attributes, previous_comment) + previous_comment = value(child) if child.type == "comment" else "" + + walk(tree.root_node) + return result + + +@dataclass(frozen=True) +class _PhpRoute: + source: str + owner: str + method: str + path: str + statement: str + + +def _php_routes(texts: Mapping[str, str]) -> list[_PhpRoute]: + """Resolve literal Laravel array and Controller@method registrations without name-only joins.""" + routes: list[_PhpRoute] = [] + pattern = re.compile(r"\bRoute::(?:get|post|put|patch|delete|any|match)\s*\([^;]+;", re.S) + for source, text in texts.items(): + if not source.endswith(".php"): + continue + # Comments cannot register a route. Preserve strings, which carry the route arguments. + clean = re.sub(r"/\*.*?\*/|//[^\n]*", "", text, flags=re.S) + for match in pattern.finditer(clean): + statement = match.group() + target = re.search(r"\[\s*([\w\\]+)::class\s*,\s*['\"](\w+)['\"]\s*\]", statement) + if target is None: + target = re.search(r"['\"]([\w\\]+)@(\w+)['\"]", statement) + path = re.search(r"\(\s*['\"]([^'\"]+)['\"]", statement) + if target: + routes.append(_PhpRoute(source, target[1].split("\\")[-1], target[2], path[1] if path else "", statement)) + return routes + + +def _php_framework_entries(target: FileTarget, text: str, texts: Mapping[str, str]) -> list[EntryPointRecord]: + records: list[EntryPointRecord] = [] + routes = _php_routes(texts) + for method in _php_methods(text): + if re.search(r"\b(private|protected)\b", method.header) or method.name.startswith("_"): + continue + route_attribute = re.search(r"(?:#\[|@)(?:[\w\\]+\\)?Route\s*\(\s*(?:path\s*[:=]\s*)?['\"]([^'\"]+)['\"]", method.header) + configured: list[tuple[str, str]] = [] + if re.search(r"extends\s+(?:[\w\\]+\\)?AbstractController\b", method.owner_header): + for source, config in texts.items(): + if "route" not in source or not source.endswith((".yaml", ".yml", ".xml", ".php")): + continue + if re.search(r"\b" + re.escape(method.owner) + r"::" + re.escape(method.name) + r"\b", config): + configured.append((source, method.name)) + if route_attribute or configured: + route_name = route_attribute[1] if route_attribute else configured[0][1] + class_route = re.search(r"(?:#\[|@)(?:[\w\\]+\\)?Route\s*\(\s*(?:path\s*[:=]\s*)?['\"]([^'\"]+)['\"]", method.owner_attributes) + if class_route and route_attribute: + route_name = class_route[1].rstrip("/") + "/" + route_name.lstrip("/") + evidence = [f"Symfony Route {route_name}"] + [f"Symfony controller registration in {source}" for source, _ in configured] + records.append( + _entry(target, method.line, method.end_line, method.name, route_name, "route", "unknown", "http_request", evidence, []) + ) + continue + component = re.search(r"extends\s+(?:Livewire\\)?Component\b", method.owner_header) + if component and ("Livewire\\" in text or "Livewire\\" in method.owner_header): + bindings = [ + source + for source, body in texts.items() + if re.search(r"wire:[\w.:-]+\s*=\s*['\"]" + re.escape(method.name) + r"(?:\(|['\"])", body) + ] + records.append( + _entry( + target, + method.line, + method.end_line, + method.name, + method.name, + "route", + "unknown", + "http_request", + [f"Livewire Component {method.owner}"] + [f"Livewire wire binding in {source}" for source in bindings], + [], + ) + ) + continue + for route in routes: + if (route.owner, route.method) != (method.owner, method.name): + continue + access: AccessLevel = "unknown" if "middleware" in route.statement else "public" + records.append( + _entry( + target, + method.line, + method.end_line, + method.name, + route.path or method.name, + "route", + access, + "http_request", + [f"Laravel Route registration in {route.source}: {route.path}"], + [], + access_declared=access == "public", + ) + ) + return records + + +def _tornado_entries(target: FileTarget, text: str) -> list[EntryPointRecord]: + try: + tree = ast.parse(text) + except SyntaxError: + return [] + records: list[EntryPointRecord] = [] + classes = [node for node in ast.walk(tree) if isinstance(node, ast.ClassDef)] + bases = {"RequestHandler", "WebSocketHandler"} + for _ in classes: + for node in classes: + if any(_safe_unparse(base).split(".")[-1] in bases for base in node.bases): + bases.add(node.name) + for node in classes: + direct = any(_safe_unparse(base).split(".")[-1] in bases for base in node.bases) + # The pool's terminal socket inherits through terminado and explicitly invokes Tornado's base method. + explicit_super = any(_safe_unparse(base) == "terminado.TermSocket" for base in node.bases) and any( + isinstance(call, ast.Call) and _safe_unparse(call.func).startswith("tornado.websocket.WebSocketHandler.") + for call in ast.walk(node) + ) + if not direct and not explicit_super: + continue + for method in node.body: + if not isinstance(method, (ast.FunctionDef, ast.AsyncFunctionDef)) or method.name not in { + "get", + "post", + "put", + "delete", + "patch", + "head", + "options", + "open", + }: + continue + records.append( + _entry( + target, + method.lineno, + method.end_lineno, + method.name, + method.name, + "route", + "unknown", + "http_request", + [f"Tornado {node.name}.{method.name}", *[_safe_unparse(d) for d in method.decorator_list]], + [], + ) + ) + return records + + +def _remix_exports(target: FileTarget, text: str) -> list[EntryPointRecord]: + if "app/routes/" not in "/" + target.path: + return [] + from .semantic.extra import grammar_for + + grammar = grammar_for("tsx" if Path(target.path).suffix == ".tsx" else target.language) + if grammar is None: + return [] + from tree_sitter import Parser + + data = text.encode() + parser = Parser(grammar) + tree = parser.parse(data) + if tree.root_node.has_error: + return [] + records: list[EntryPointRecord] = [] + + def walk(node: Any) -> None: + if node.type == "export_statement": + for declaration in node.named_children: + candidates = ( + declaration.named_children if declaration.type in {"lexical_declaration", "variable_declaration"} else [declaration] + ) + for candidate in candidates: + name_node = candidate.child_by_field_name("name") + if name_node is None: + continue + name = data[name_node.start_byte : name_node.end_byte].decode() + if name not in {"action", "loader"}: + continue + function = candidate.child_by_field_name("value") if candidate.type == "variable_declarator" else candidate + if function is None or function.type not in {"function_declaration", "arrow_function", "function_expression"}: + continue + records.append( + _entry( + target, + data[: function.start_byte].count(b"\n") + 1, + data[: function.end_byte].count(b"\n") + 1, + name, + name, + "route", + "unknown", + "http_request", + [f"Remix route export {name}"], + [], + ) + ) + return + for child in node.named_children: + walk(child) + + walk(tree.root_node) + return records + + +def entry_framework(entry: object) -> str | None: + """The recogniser id recorded in a handler entry's evidence.""" + evidence = " ".join(getattr(entry, "access_evidence", ())) + for name, identifier in ( + ("Stapler", "stapler"), + ("Jetty", "jetty"), + ("Servlet", "servlet"), + ("JAX-RS", "jaxrs"), + ("Symfony", "symfony"), + ("Livewire", "livewire"), + ("Laravel", "laravel"), + ("Remix", "remix"), + ("Tornado", "tornado"), + ): + if name in evidence: + return identifier + return 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 diff --git a/src/openultrasast/ruleset/obligations/java.toml b/src/openultrasast/ruleset/obligations/java.toml index 4e40ad3..17f8ff7 100644 --- a/src/openultrasast/ruleset/obligations/java.toml +++ b/src/openultrasast/ruleset/obligations/java.toml @@ -1,3 +1,5 @@ +# Handler obligations: Jenkins form validation and list filling require permission checks. +# https://www.jenkins.io/doc/developer/security/form-validation/ (reference in the approved brief). # Delivery A additionally checked actual calls in Git HEAD objects (2026-10-14): # p4-plugin: CredentialsHelper.lookupCredentialsInItem, P4PasswordImpl and P4TicketImpl # call ConnectionFactory.testConnection, ConnectionFactory calls connect, ReviewNotifier @@ -79,3 +81,11 @@ id = "identity-from-context" kind = "identity_constraint" identity_sources = ["SecurityContextHolder.getContext().getAuthentication()", "Jenkins.getAuthentication()", "Jenkins.getAuthentication2()", "getUserPrincipal()"] request_sources = ["getParameter(", "getHeader(", "getCookies(", "getQueryString("] + +[[operation]] +id = "stapler-handler" +kind = "privileged_action" +handlers = ["^do(?:Check.*|Fill.*Items|Test.*|Validate.*|Verify.*)$"] +framework = "stapler" +requires = ["path_guard"] +sensitivity = "high" diff --git a/src/openultrasast/ruleset/obligations/php.toml b/src/openultrasast/ruleset/obligations/php.toml index 8e12368..1b5e2e2 100644 --- a/src/openultrasast/ruleset/obligations/php.toml +++ b/src/openultrasast/ruleset/obligations/php.toml @@ -1,3 +1,9 @@ +# Delivery B enforcement sources: shopperlabs__shopper Git HEAD packages/admin/routes/web.php +# uses chained middleware and groups. kimai__kimai at the pool revision uses @Route(path=...) +# and @Security in src/Controller/InvoiceController.php (class and action docblocks). +# Recognised contract names from the approved spec and the references below: Laravel auth, can:, +# role and permission middleware, constructor middleware/authorizeResource; Symfony IsGranted, +# Security and security.yaml access_control. No PHP operation or discharger row was changed. # Closed authorization facts for Laravel, WordPress and Symfony. # Checked shopperlabs__shopper Git HEAD source objects in /home/mc/ousast-results/valid/train/hist/: # CollectionProducts.authorize, SetupGuide.can, AssetController.abort_unless, diff --git a/src/openultrasast/ruleset/obligations/python.toml b/src/openultrasast/ruleset/obligations/python.toml index 9d2a866..96a89e4 100644 --- a/src/openultrasast/ruleset/obligations/python.toml +++ b/src/openultrasast/ruleset/obligations/python.toml @@ -1,3 +1,6 @@ +# Checked mage-ai__mage-ai Git HEAD and websocket fix 66440e4d3: no authenticated +# decorator or get_current_user definition found. authenticated is already an existing fact. +# Handler names follow the Tornado RequestHandler/WebSocketHandler contract in the approved spec. # Obligated operations and dischargers for Python web code (authorization-obligations, Req 1). Closed data. version = "1" language = "python" @@ -84,3 +87,11 @@ permissive_values = ["False"] id = "validated-input" kind = "validated_input" calls = ["validate", "schema.load", "parse_obj", "model_validate", "clean", "is_valid", "secure_filename", "safe_join"] + +[[operation]] +id = "tornado-handler" +kind = "privileged_action" +handlers = ["^(?:get|post|put|delete|patch|head|options|open)$"] +framework = "tornado" +requires = ["path_guard"] +sensitivity = "high" diff --git a/src/openultrasast/ruleset/obligations/typescript.toml b/src/openultrasast/ruleset/obligations/typescript.toml index d7e0757..1f94179 100644 --- a/src/openultrasast/ruleset/obligations/typescript.toml +++ b/src/openultrasast/ruleset/obligations/typescript.toml @@ -1,3 +1,6 @@ +# Checked triggerdotdev__trigger.dev Git HEAD: apps/webapp/app/services/session.server.ts +# defines requireUser and requireUserId; apiAuth.server.ts defines authenticateApiRequest; +# routes/resources.dashboard-agent.alerts.$channelId.unsubscribe.tsx defines and calls authorize. # Obligated operations and dischargers for TypeScript web code (same surface as JavaScript) (authorization-obligations, Req 1). Closed data. version = "1" language = "typescript" @@ -72,7 +75,7 @@ sensitivity = "medium" [[discharger]] id = "route-guard" kind = "path_guard" -calls = ["authenticate", "requireAuth", "ensureAuthenticated", "isAuthenticated", "verifyToken", "checkAuth", "protect", "authorize", "requireUser", "getServerSession", "auth", "withAuth", "jwt.verify", "passport.authenticate"] +calls = ["authenticate", "requireAuth", "ensureAuthenticated", "isAuthenticated", "verifyToken", "checkAuth", "protect", "authorize", "requireUser", "requireUserId", "authenticateApiRequest", "getServerSession", "auth", "withAuth", "jwt.verify", "passport.authenticate"] decorators = ["UseGuards", "Authorized", "Roles"] # `req.session` is the authenticated context itself, and it is listed because a destructuring spells the @@ -112,3 +115,11 @@ permissive_values = ["false"] id = "validated-input" kind = "validated_input" calls = ["parse", "safeParse", "validate", "validateSync", "sanitize", "escape", "validationResult", "z.object", "assert"] + +[[operation]] +id = "remix-handler" +kind = "privileged_action" +handlers = ["^action$"] +framework = "remix" +requires = ["path_guard", "ownership_check"] +sensitivity = "high" diff --git a/src/openultrasast/semantic/obligations/check.py b/src/openultrasast/semantic/obligations/check.py index 756c2f5..a541b8e 100644 --- a/src/openultrasast/semantic/obligations/check.py +++ b/src/openultrasast/semantic/obligations/check.py @@ -66,6 +66,7 @@ def check_obligations( min_siblings: int, extra_witnesses: Sequence[Discharge] = (), assume_entries_protected: bool = False, + handlers_of: Mapping[str, Mapping[str, str]] | None = None, ) -> ObligationResult: operations: list[Operation] = [] witnesses: list[Discharge] = list(extra_witnesses) @@ -81,7 +82,7 @@ def check_obligations( continue scoped = facts.for_language(ir.language) flow = flow_facts.for_language(ir.language) - operations.extend(find_operations(ir, scoped, text=text)) + operations.extend(find_operations(ir, scoped, text=text, handlers_of=(handlers_of or {}).get(path))) witnesses.extend(find_discharges(ir, scoped, flow, text=text, entries=entries)) sets = sibling_sets(entries, operations, witnesses) anomalies, under = consistency_anomalies(sets, min_siblings=min_siblings) @@ -122,11 +123,15 @@ def check_obligations( # (a) the operation's own obligation: discharged when any accepted discharger dominates. A declared public route # waives the path guard here too (Req 4.4), so the operation falls through to the next kind it accepts. required = [kind for kind in operation.requires if not (kind == "path_guard" and declared_access == "public")] - if required and not (set(required) & dominated_kinds): - missing = next( - (kind for kind in required if (handler, kind) in anomaly_index), - required[0], - ) + handler_fact = next((f for f in facts.operations if f.id == operation.fact_id and f.handlers), None) + missing_kinds = ( + [kind for kind in required if kind not in dominated_kinds] + if handler_fact is not None and handlers_of is not None + else [next((kind for kind in required if (handler, kind) in anomaly_index), required[0])] + if required and not (set(required) & dominated_kinds) + else [] + ) + for missing in missing_kinds: finding = _finding( operation, missing, @@ -139,6 +144,7 @@ def check_obligations( paths, facts, entry if assume_entries_protected else None, + handler_obligation=handler_fact is not None and handlers_of is not None, ) if finding is not None: findings.append(finding) @@ -225,6 +231,8 @@ def _finding( paths: Sequence[object], facts: ObligationFacts, assumed_entry: object | None = None, + *, + handler_obligation: bool = False, ) -> ObligationFinding | None: wrong = next((w for w in same if w.kind == missing and not valid_witness(w)), None) declared = policy.resources.get(operation.resource or "") if policy is not None else None @@ -247,14 +255,25 @@ def _finding( label = "consistency_violation" evidence = anomaly.siblings + anomaly.evidence[len(anomaly.siblings) :] elif ( - paths and assumed_entry is not None and getattr(assumed_entry, "access_level", "public") != "public" and declared_access != "public" + paths + and assumed_entry is not None + and ( + (getattr(assumed_entry, "access_level", "public") != "public" and declared_access != "public") + or (handler_obligation and missing != "path_guard") + ) ): label = "unguarded_path" evidence = ( f"entry {getattr(assumed_entry, 'kind', '')}: {getattr(assumed_entry, 'path', '')}::" f"{getattr(assumed_entry, 'function_name', '')}", *(str(item) for item in getattr(assumed_entry, "access_evidence", ())), - "no declared policy clause and no sibling norm; the prover assumes a recognised entry is protected", + ( + "no declared policy clause and no sibling norm; the handler fact requires " + + missing + + "; public access waives only path_guard" + if handler_obligation and (declared_access == "public" or getattr(assumed_entry, "access_level", "public") == "public") + else "no declared policy clause and no sibling norm; the prover assumes a recognised entry is protected" + ), ) elif paths: return None # path-aware mode: without a clause or a sibling norm, nothing justifies the obligation (gating 3) diff --git a/src/openultrasast/semantic/obligations/enforcement.py b/src/openultrasast/semantic/obligations/enforcement.py new file mode 100644 index 0000000..32b4832 --- /dev/null +++ b/src/openultrasast/semantic/obligations/enforcement.py @@ -0,0 +1,104 @@ +"""Framework enforcement outside a handler, resolved from cached source and route declarations.""" + +from __future__ import annotations + +import ast +import re +from collections.abc import Mapping, Sequence +from pathlib import Path + +import yaml + +from ...mapping import EntryPointRecord, _php_methods, _php_routes, entry_framework +from .operations import Discharge + +_MIDDLEWARE = re.compile(r"['\"](auth(?::[^'\"]+)?|can:[^'\"]+|role(?::[^'\"]+)?|permission(?::[^'\"]+)?)['\"]") + + +def witnesses(*, root: Path, texts: Mapping[str, str], entries: Sequence[EntryPointRecord], language: str) -> list[Discharge]: + """One scoped path guard per enforcement source and covered handler; unrelated declarations discharge nothing. + + Texts are supplied by the caller's cache. Root names the checkout; this function never rereads source files. + """ + del root + found: list[Discharge] = [] + routes = _php_routes(texts) if language == "php" else [] + security = _security_rules(texts) if language == "php" else [] + methods_by_path = {path: _php_methods(text) for path, text in texts.items() if language == "php" and path.endswith(".php")} + for entry in entries: + if not entry.function_name or entry.line is None or not entry.defines_handler: + continue + framework = entry_framework(entry) + sources: list[tuple[str, str]] = [] + if language == "python" and framework == "tornado": + tree = ast.parse(texts.get(entry.path, "")) + for py_method in ast.walk(tree): + if isinstance(py_method, (ast.FunctionDef, ast.AsyncFunctionDef)) and py_method.lineno == entry.line: + for decorator in py_method.decorator_list: + name = ast.unparse(decorator) + if name in {"authenticated", "tornado.web.authenticated"}: + sources.append(("decorator", f"Tornado {name} in {entry.path}")) + if language == "php" and framework in {"symfony", "laravel", "livewire"}: + methods = methods_by_path.get(entry.path, []) + method = next((m for m in methods if m.name == entry.function_name and m.line == entry.line), None) + if method is None: + continue + if framework == "symfony": + for header in (method.header, method.owner_attributes): + if re.search(r"(?:#\[|@)(?:[\w\\]+\\)?(?:IsGranted|Security)\s*\(", header): + sources.append(("decorator", f"Symfony IsGranted/Security in {entry.path}")) + # Symfony evaluates access_control in order: a public match cannot be skipped for a later role. + for source, path, roles in security: + if re.search(path, entry.name): + if roles and not any(role in {"PUBLIC_ACCESS", "IS_AUTHENTICATED_ANONYMOUSLY"} for role in roles): + sources.append(("router", f"Symfony access_control in {source}")) + break + if framework in {"laravel", "livewire"}: + for route in routes: + if (route.owner, route.method) == (method.owner, method.name): + middleware = re.search(r"->middleware\s*\((.*?)\)", route.statement, re.S) + if middleware and _MIDDLEWARE.search(middleware[1]): + sources.append(("router", f"Laravel route middleware in {route.source}")) + for constructor in methods: + if constructor.owner != method.owner or constructor.name != "__construct": + continue + for match in re.finditer(r"\$this->(?:middleware|authorizeResource)\s*\([^;]+;", constructor.body, re.S): + statement = match.group() + only = re.search(r"->only\s*\((.*?)\)", statement, re.S) + exclude = re.search(r"->except\s*\((.*?)\)", statement, re.S) + if only and not re.search(r"['\"]" + re.escape(method.name) + r"['\"]", only[1]): + continue + if exclude and re.search(r"['\"]" + re.escape(method.name) + r"['\"]", exclude[1]): + continue + if "authorizeResource" in statement or _MIDDLEWARE.search(statement): + sources.append(("router", f"Laravel constructor enforcement in {entry.path}")) + for scope, detail in dict.fromkeys(sources): + found.append( + Discharge( + entry.path, None, entry.function_name, "path_guard", f"{framework}-enforcement", "authenticated_context", scope, detail + ) + ) + return list(dict.fromkeys(found)) + + +def _security_rules(texts: Mapping[str, str]) -> list[tuple[str, str, tuple[str, ...]]]: + result: list[tuple[str, str, tuple[str, ...]]] = [] + for source, text in texts.items(): + if not source.endswith(("config/packages/security.yaml", "config/packages/security.yml")): + continue + payload = yaml.safe_load(text) + if not isinstance(payload, dict) or not isinstance(payload.get("security"), dict): + continue + rules = payload["security"].get("access_control", []) + if not isinstance(rules, list): + continue + for rule in rules: + if not isinstance(rule, dict) or not isinstance(rule.get("path"), str): + continue + roles = rule.get("roles", ()) + if isinstance(roles, str): + roles = (roles,) + if not isinstance(roles, (list, tuple)): + continue + result.append((source, rule["path"], tuple(str(role) for role in roles))) + return result diff --git a/src/openultrasast/semantic/obligations/facts.py b/src/openultrasast/semantic/obligations/facts.py index 38c016a..4a37965 100644 --- a/src/openultrasast/semantic/obligations/facts.py +++ b/src/openultrasast/semantic/obligations/facts.py @@ -7,6 +7,7 @@ data bug and fails loud. Nothing detects an obligation that is not named here. from __future__ import annotations +import re import tomllib from dataclasses import dataclass from pathlib import Path @@ -20,7 +21,9 @@ PROVENANCE_KINDS = ("authenticated_context", "request_input", "constant", "unkno SENSITIVITIES = ("low", "medium", "high") DEFAULT_OBLIGATION_FACTS_DIR = Path(__file__).resolve().parents[2] / "ruleset" / "obligations" -_OPERATION_FIELDS = frozenset({"id", "kind", "calls", "resource_arg", "requires", "sensitivity", "query_document_arg"}) +_OPERATION_FIELDS = frozenset( + {"id", "kind", "calls", "resource_arg", "requires", "sensitivity", "query_document_arg", "handlers", "framework"} +) _DISCHARGER_FIELDS = frozenset( {"id", "kind", "calls", "decorators", "identity_sources", "constraint_params", "request_sources", "permissive_values"} ) @@ -45,6 +48,8 @@ class OperationFact: # calls were a selector or a predicate. The position is 1-based over the real arguments; the arbiter that # can see argument shapes decides it, and a language whose calls are unambiguous never sets this. query_document_arg: int | None = None + handlers: tuple[str, ...] = () + framework: str | None = None @dataclass(frozen=True) @@ -109,6 +114,17 @@ def _load_obligation_facts_uncached(directory: Path | None = None) -> Obligation document_arg = row.get("query_document_arg") if document_arg is not None and (not isinstance(document_arg, int) or document_arg < 1): raise ObligationFactsError(f"{path}: operation {row.get('id')!r} query_document_arg must be a positive integer") + handlers = _strings(row.get("handlers"), "handlers", path) + framework = row.get("framework") + if framework is not None and (not isinstance(framework, str) or not framework): + raise ObligationFactsError("invalid operation framework") + if handlers and framework is None: + raise ObligationFactsError("handler operation needs a framework") + for pattern in handlers: + try: + re.compile(pattern) + except re.error as exc: + raise ObligationFactsError(type(exc).__name__) from exc operations.append( OperationFact( id=str(row["id"]), @@ -119,6 +135,8 @@ def _load_obligation_facts_uncached(directory: Path | None = None) -> Obligation requires=requires, sensitivity=sensitivity, query_document_arg=document_arg, + handlers=handlers, + framework=framework, ) ) for row in _rows(payload.get("discharger"), path): diff --git a/src/openultrasast/semantic/obligations/operations.py b/src/openultrasast/semantic/obligations/operations.py index c5f14a7..df2c993 100644 --- a/src/openultrasast/semantic/obligations/operations.py +++ b/src/openultrasast/semantic/obligations/operations.py @@ -10,7 +10,7 @@ the authenticated context (a constraint bound from the request or a constant dis from __future__ import annotations import re -from collections.abc import Sequence +from collections.abc import Mapping, Sequence from dataclasses import dataclass from ..facts import SemanticFacts @@ -44,7 +44,9 @@ class Discharge: detail: str = "" # identifier only: the decorator or call name that witnessed the discharge -def find_operations(ir: FileIR, facts: ObligationFacts, *, text: str) -> tuple[Operation, ...]: +def find_operations( + ir: FileIR, facts: ObligationFacts, *, text: str, handlers_of: Mapping[str, str] | None = None +) -> tuple[Operation, ...]: """Every obligated operation in the file's named functions, in source order.""" if not ir.parse_ok: return () @@ -54,6 +56,14 @@ def find_operations(ir: FileIR, facts: ObligationFacts, *, text: str) -> tuple[O for function in ir.functions: if function.name == "": continue + for fact in facts.operations: + if ( + fact.handlers + and not fact.calls + and (handlers_of or {}).get(function.name) == fact.framework + and any(re.search(pattern, function.name) for pattern in fact.handlers) + ): + found.append(Operation(ir.path, function.start_line, function.name, fact.kind, fact.id, None, fact.requires)) for site in _operations(function, facts, lines, kind=None): key = (function.name, site.call.line, site.fact.id) if key in seen: diff --git a/src/openultrasast/semantic/obligations/paths.py b/src/openultrasast/semantic/obligations/paths.py index c4f5850..a89b68a 100644 --- a/src/openultrasast/semantic/obligations/paths.py +++ b/src/openultrasast/semantic/obligations/paths.py @@ -30,6 +30,19 @@ def compose_paths( raise ValueError("invalid path bounds") found: dict[tuple[tuple[str, int | None], ...], PathRecord] = {} unreached: list[str] = [] + if not entries: + # Nothing can complete a chain; one ordinary change of a 3000-file tree spent eleven minutes walking + # callers towards entries that did not exist. + return [], [f"{path}::{function}" for path, function in dict.fromkeys(targets)] + # Each node's callers are looked up once per change: the walk revisits nodes across chains and targets. + known: dict[str, list[tuple[str, int]]] = {} + + def callers_of_node(node: str) -> list[tuple[str, int]]: + if node not in known: + node_path, _, node_function = node.rpartition("::") + known[node] = list(dict.fromkeys(callers(node_path, node_function)))[:fan_out] + return known[node] + for path, function in dict.fromkeys(targets): target = f"{path}::{function}" reached = False @@ -43,9 +56,8 @@ def compose_paths( continue if len(hops) - 1 >= max_depth: continue - node_path, _, node_function = node.rpartition("::") seen = {hop[0] for hop in hops} - for caller, line in list(dict.fromkeys(callers(node_path, node_function)))[:fan_out]: + for caller, line in callers_of_node(node): if caller not in seen: pending.append(((caller, line), *hops)) if not reached: diff --git a/tests/test_hunt_prover.py b/tests/test_hunt_prover.py index daefde3..271a025 100644 --- a/tests/test_hunt_prover.py +++ b/tests/test_hunt_prover.py @@ -181,3 +181,112 @@ def test_nonpublic_unreached_function_in_partly_covered_repo(tmp_path): assert result.coverage.frameworks_recognised == {"stapler": 1} assert result.coverage.unchecked_by_reason["framework_unrecognised"] == 1 assert not result.findings + + +@pytest.mark.parametrize("guard,expected", [("", 1), ("checkPermission(ADMINISTER);", 0)]) +def test_handler_without_listed_call(tmp_path, guard, expected): + root, change = repository( + tmp_path, + f"""class X extends Descriptor {{ + public Object doCheckValue() {{ + {guard} + return 1; + }} +}}""", + ) + result = prove_change(root, change) + assert len(result.findings) == expected and result.coverage.operations_checked == 1 + if expected: + assert result.findings[0]["line"] == 2 + assert result.findings[0]["operation"]["id"] == "stapler-handler" + + +@pytest.mark.parametrize("middleware,expected", [("->middleware(['auth'])", 0), ("->middleware(['throttle:10'])", 1)]) +def test_laravel_enforcement_in_prover(tmp_path, middleware, expected): + root, change = repository( + tmp_path, + "class Plain {}", + { + "Editor.php": "" if arrow else "export async function action()" + text = f"""{declaration} {{ + const user = await requireUser(request); + return deactivateRecord(requestId); +}}""" + root, change = repository(tmp_path, "class Plain {}", {"app/routes/items.ts": text}) + change["files"] = [{"path": "app/routes/items.ts", "head_lines": [[1, 4]]}] + result = prove_change(root, change) + assert result.coverage.operations_checked == 1 and result.coverage.operations_discharged == 0 + assert len(result.findings) == 1 and result.findings[0]["missing"] == "ownership_check" + + +@pytest.mark.parametrize("decorator,expected", [("", 1), ("@tornado.web.authenticated", 0)]) +def test_tornado_enforcement_in_prover(tmp_path, decorator, expected): + text = f"""import tornado.web +class Handler(tornado.web.RequestHandler): + {decorator} + def get(self): + return 1 +""" + root, change = repository(tmp_path, "class Plain {}", {"app.py": text}) + change["files"] = [{"path": "app.py", "head_lines": [[1, 5]]}] + result = prove_change(root, change) + assert len(result.findings) == expected and result.coverage.operations_checked == 1 + + +def test_symfony_yaml_is_read_by_prover(tmp_path): + root, change = repository( + tmp_path, + "class Plain {}", + { + "Controller.php": ( + " 0 + + +@pytest.mark.parametrize("exits,expected", [(True, 0), (False, 1)]) +def test_handler_permission_predicate_requires_exit(tmp_path, exits, expected): + root, change = repository( + tmp_path, + f"""class X extends Descriptor {{ + public Object doCheckValue() {{ + boolean allowed = hasPermission(ADMINISTER); + {"if (!allowed) {" if exits else ""} + {"return null;" if exits else ""} + {"}" if exits else ""} + return 1; + }} +}}""", + ) + result = prove_change(root, change) + assert len(result.findings) == expected and result.coverage.operations_discharged == 1 - expected + + +@pytest.mark.parametrize("public,expected_missing", [(False, ["path_guard", "ownership_check"]), (True, ["ownership_check"])]) +def test_remix_missing_kinds_and_public_policy(tmp_path, public, expected_missing): + extra = {"app/routes/items.ts": "export function action() { return deactivateRecord(requestId); }"} + if public: + extra[".openultrasast/obligations.toml"] = 'version = 1\n[[route]]\npath = "action"\naccess = "public"\n' + root, change = repository(tmp_path, "class Plain {}", extra) + change["files"] = [{"path": "app/routes/items.ts", "head_lines": [[1, 1]]}] + result = prove_change(root, change) + assert [finding["missing"] for finding in result.findings] == expected_missing diff --git a/tests/test_mapping_frameworks.py b/tests/test_mapping_frameworks.py index 7c5e6e2..eae1a99 100644 --- a/tests/test_mapping_frameworks.py +++ b/tests/test_mapping_frameworks.py @@ -67,3 +67,175 @@ def test_jetty_servlet_and_jaxrs(tmp_path, text, name): def test_unrecognised_and_comments(tmp_path): assert entries(tmp_path, "/* class X extends Descriptor { public void doFake() {} } */\nclass Plain { public void doCheck() {} }") == [] + + +def framework_entries(tmp_path, files): + targets = [] + for name, text in files.items(): + path = tmp_path / name + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text) + targets.append(build_file_target(tmp_path, path)) + return [ + e for e in analyze_entry_points(tmp_path, targets, texts=files, include_manifests=False) if e.kind == "route" and e.function_name + ] + + +@pytest.mark.parametrize("annotation", ['#[Route("/items")]', '/** @Route("/items") */']) +def test_symfony_actions(tmp_path, annotation): + rows = framework_entries( + tmp_path, + { + "Controller.php": f"""Save', + }, + ) + assert [(e.path, e.function_name) for e in rows] == [("Component.php", "store")] + assert "Livewire" in " ".join(rows[0].access_evidence) + + +@pytest.mark.parametrize("middleware,access", [("", "public"), ("->middleware(['auth', 'can:edit'])", "unknown")]) +def test_laravel_cross_file_registration(tmp_path, middleware, access): + rows = framework_entries( + tmp_path, + { + "app/Http/Controllers/Editor.php": """ { return 1; }", + "export const action: ActionFunction = async () => { return 1; }", + ], +) +def test_remix_exports(tmp_path, declaration): + rows = framework_entries(tmp_path, {"apps/web/app/routes/items.ts": declaration + "\nexport function loader() { return 1; }"}) + assert [e.function_name for e in rows] == ["action", "loader"] + assert all("Remix" in " ".join(e.access_evidence) and e.access_level == "unknown" for e in rows) + assert framework_entries(tmp_path, {"app/helpers/items.ts": declaration}) == [] + + +@pytest.mark.parametrize("base", ["tornado.web.RequestHandler", "tornado.websocket.WebSocketHandler", "RequestHandler"]) +def test_tornado_methods(tmp_path, base): + rows = framework_entries( + tmp_path, + { + "app.py": f"""from tornado.web import RequestHandler +class Handler({base}): + def get(self): + return 1 + def open(self): + pass + def helper(self): + pass +class Plain: + def get(self): + pass +""" + }, + ) + assert [e.function_name for e in rows] == ["get", "open"] + assert all("Tornado" in " ".join(e.access_evidence) for e in rows) + + +@pytest.mark.parametrize("prefix", ["", "\n" * 320]) +def test_symfony_named_paths_and_class_prefix(tmp_path, prefix): + rows = framework_entries( + tmp_path, + { + "Controller.php": prefix + + """ { return 1; }; +export default function Items() { return
Items
; } +""" + }, + ) + assert [row.function_name for row in rows] == ["action"] diff --git a/tests/test_obligation_enforcement.py b/tests/test_obligation_enforcement.py new file mode 100644 index 0000000..71d95d7 --- /dev/null +++ b/tests/test_obligation_enforcement.py @@ -0,0 +1,157 @@ +"""Framework enforcement is scoped to the registered handler, never a nearby method.""" + +import pytest + +from openultrasast.mapping import analyze_entry_points +from openultrasast.preprocess import build_file_target +from openultrasast.semantic.obligations.dominance import OrderDominance +from openultrasast.semantic.obligations.operations import Operation + + +def mapped(tmp_path, files): + targets = [] + for name, text in files.items(): + path = tmp_path / name + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text) + if path.suffix in {".php", ".py"}: + targets.append(build_file_target(tmp_path, path)) + return [ + e for e in analyze_entry_points(tmp_path, targets, texts=files, include_manifests=False) if e.kind == "route" and e.function_name + ] + + +@pytest.mark.parametrize("guard", ["auth", "can:edit", "role:admin", "permission:edit"]) +def test_laravel_route_middleware(tmp_path, guard): + from openultrasast.semantic.obligations.enforcement import witnesses + + files = { + "Editor.php": "middleware(['{guard}']);", + } + rows = witnesses(root=tmp_path, texts=files, entries=mapped(tmp_path, files), language="php") + (witness,) = rows + assert (witness.path, witness.function, witness.line, witness.scope) == ("Editor.php", "store", None, "router") + assert "routes/web.php" in witness.detail + assert OrderDominance(files).dominates( + witness, Operation("Editor.php", 1, "store", "protected_write", "x", None, ("path_guard",)), None + ) + + +@pytest.mark.parametrize("guard", ["$this->middleware('auth');", "$this->authorizeResource(Item::class, 'item');"]) +def test_laravel_constructor(tmp_path, guard): + from openultrasast.semantic.obligations.enforcement import witnesses + + files = { + "Editor.php": f"""only(['store'])", "->except(['other'])"]) +def test_constructor_restrictions_do_not_guard_other_handlers(tmp_path, restriction): + from openultrasast.semantic.obligations.enforcement import witnesses + + files = { + "Editor.php": f"""middleware('auth'){restriction}; }} + public function store() {{}} + public function other() {{}} +}} +Route::post('/items', [Editor::class, 'store']); +Route::get('/other', [Editor::class, 'other']); +""" + } + rows = witnesses(root=tmp_path, texts=files, entries=mapped(tmp_path, files), language="php") + assert [row.function for row in rows] == ["store"] + + +def test_symfony_first_public_rule_does_not_become_a_guard(tmp_path): + from openultrasast.semantic.obligations.enforcement import witnesses + + files = { + "Editor.php": "middleware('auth'); +Route::post('/items', [Editor::class, 'store'])->middleware('throttle:10'); +""", + "app.py": "class Handler(RequestHandler):\n @not_authenticated\n def get(self):\n pass\n", + } + entries = mapped(tmp_path, files) + assert witnesses(root=tmp_path, texts=files, entries=entries, language="php") == [] + assert witnesses(root=tmp_path, texts=files, entries=entries, language="python") == [] diff --git a/tests/test_obligation_facts.py b/tests/test_obligation_facts.py index ff294bb..de64cf5 100644 --- a/tests/test_obligation_facts.py +++ b/tests/test_obligation_facts.py @@ -115,3 +115,30 @@ def test_delivery_a_java_operations_and_dischargers_resolve(): assert "path_guard" in operations[call].requires assert dischargers["checkPermission"].kind == dischargers["checkAnyPermission"].kind == "path_guard" assert dischargers["hasPermission"].kind == "ownership_check" + + +def test_handler_rows_and_validation(tmp_path): + from openultrasast.semantic.obligations.facts import ObligationFactsError, load_obligation_facts + + text = """version = "1" +language = "python" +[[operation]] +id = "handler" +kind = "privileged_action" +handlers = ["^open$"] +framework = "tornado" +requires = ["path_guard"] +sensitivity = "high" +""" + path = tmp_path / "python.toml" + path.write_text(text) + (row,) = load_obligation_facts(tmp_path).operations + assert row.handlers == ("^open$",) and row.framework == "tornado" and row.calls == () + for invalid in [ + text.replace('framework = "tornado"', ""), + text.replace('"^open$"', '"["'), + text.replace('framework = "tornado"', "framework = 12"), + ]: + path.write_text(invalid) + with pytest.raises(ObligationFactsError): + load_obligation_facts(tmp_path) diff --git a/tests/test_obligation_operations.py b/tests/test_obligation_operations.py index cc73404..da2eb57 100644 --- a/tests/test_obligation_operations.py +++ b/tests/test_obligation_operations.py @@ -111,3 +111,33 @@ def test_discharge_witness_covers_operation_only_when_kind_and_provenance_fit() assert covering["constrained"] == ["identity_constraint"] and covering["constrained_too"] == ["identity_constraint"] assert covering["leaky"] == [] # request-bound identity does not cover assert covering["guarded"] == [] # a path guard does not cover a protected read (only privileged actions) + + +def test_handler_named_operations_are_framework_scoped(): + from openultrasast.semantic.obligations import load_obligation_facts + from openultrasast.semantic.obligations.operations import find_operations + + text = "class Handler:\n def open(self):\n pass\n def helper(self):\n pass\n" + ir = parse_file("app.py", text, "python") + facts = load_obligation_facts().for_language("python") + assert find_operations(ir, facts, text=text) == () + assert find_operations(ir, facts, text=text, handlers_of={"open": "other"}) == () + (op,) = find_operations(ir, facts, text=text, handlers_of={"open": "tornado", "helper": "tornado"}) + assert (op.function, op.line, op.requires) == ("open", 2, ("path_guard",)) + + +def test_handler_rows_in_java_and_typescript(): + from openultrasast.semantic.obligations import load_obligation_facts + from openultrasast.semantic.obligations.operations import find_operations + + facts = load_obligation_facts() + cases = [ + ("java", "stapler", "class Editor { public void doFillNamesItems() {} }", "doFillNamesItems", ("path_guard",)), + ("typescript", "remix", "export function action() { return 1; }", "action", ("path_guard", "ownership_check")), + ] + for language, framework, text, name, requires in cases: + ir = parse_file(f"app.{language}", text, language) + assert ir.parse_ok + assert find_operations(ir, facts.for_language(language), text=text) == () + (operation,) = find_operations(ir, facts.for_language(language), text=text, handlers_of={name: framework}) + assert operation.line == 1 and operation.requires == requires and operation.function == name diff --git a/tests/test_obligation_paths.py b/tests/test_obligation_paths.py index 5807a1b..f15260f 100644 --- a/tests/test_obligation_paths.py +++ b/tests/test_obligation_paths.py @@ -63,3 +63,21 @@ def test_records_are_consumed_by_the_checker(): ) (finding,) = result.findings assert finding.operation.function == "entry" and finding.label == "unguarded_path" + + +def test_callers_are_looked_up_once_per_node_and_never_without_entries(): + from openultrasast.semantic.obligations.paths import compose_paths + + calls: list[str] = [] + graph = {"a.py::leaf": [("a.py::mid", 3)], "a.py::mid": [("web.py::view", 9), ("a.py::other", 4)], "a.py::other": [("a.py::mid", 5)]} + + def callers(path, function): + calls.append(f"{path}::{function}") + return graph.get(f"{path}::{function}", []) + + paths, unreached = compose_paths(callers=callers, entries={"web.py::view"}, targets=[("a.py", "leaf"), ("a.py", "mid")], max_depth=3) + assert [p.hops[0][0] for p in paths] == ["web.py::view"] * len(paths) and len(paths) >= 2 and unreached == [] + assert len(calls) == len(set(calls)) # once per node across both targets + calls.clear() + paths, unreached = compose_paths(callers=callers, entries=set(), targets=[("a.py", "leaf")]) + assert paths == [] and unreached == ["a.py::leaf"] and calls == []