feat(authorization-obligations): name handlers registered by call or convention

Task 2.6. The entry-point mapper named a handler only from a decorator on the
function, so projects that register routes elsewhere had no named entry point
and every obligation on them was unreachable. It now also names Python handlers
registered through add_url_rule and add_resource, the HTTP-verb methods of a
Resource, MethodView, APIView or ViewSet subclass, the handler identifier on a
JavaScript router registration with its middleware as access evidence, and the
exported HTTP-verb functions of a Next.js route module.

Access classification, evidence, conditions and the record shape are unchanged,
and the map gate output is byte-identical.

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG
This commit is contained in:
Marius Ciepluch
2026-09-06 07:31:48 +02:00
co-authored by Claude Opus 5
parent 7d0ca74bf3
commit 18452b2902
2 changed files with 209 additions and 6 deletions
+163 -6
View File
@@ -2,6 +2,8 @@ from __future__ import annotations
import ast
import json
import re
from collections.abc import Mapping, Sequence
from dataclasses import asdict, dataclass, replace
from pathlib import Path
from typing import Any, Literal, cast
@@ -194,6 +196,7 @@ def _python_entry_points(target: FileTarget, text: str) -> list[EntryPointRecord
except SyntaxError:
tree = None
if tree is not None:
records.extend(_python_registered_routes(target, tree))
for node in ast.walk(tree):
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
decorators = [_safe_unparse(decorator) for decorator in node.decorator_list]
@@ -244,20 +247,25 @@ def _python_entry_points(target: FileTarget, text: str) -> list[EntryPointRecord
def _js_entry_points(target: FileTarget, text: str) -> list[EntryPointRecord]:
records: list[EntryPointRecord] = []
route_tokens = (".get(", ".post(", ".put(", ".delete(", ".patch(", ".use(")
for number, line in enumerate(text.splitlines(), start=1):
lines = text.splitlines()
bounds = _js_function_bounds(lines)
for number, line in enumerate(lines, start=1):
stripped = line.strip()
if any(token in stripped for token in route_tokens) and any(prefix in stripped for prefix in ("app", "router", "server")):
handler, middleware = _js_registration(stripped, bounds)
access, evidence = _js_route_access(middleware, stripped)
start, end = bounds.get(handler or "", (number, number))
records.append(
_entry(
target,
number,
number,
_js_handler_name(stripped),
start,
end,
handler or _js_handler_name(stripped),
"http_handler",
"route",
"public",
access,
"http_request",
[stripped],
evidence,
_line_conditions(stripped),
)
)
@@ -276,9 +284,93 @@ def _js_entry_points(target: FileTarget, text: str) -> list[EntryPointRecord]:
_line_conditions(stripped),
)
)
records.extend(_js_route_module(target, lines, bounds))
return records
_RESOURCE_BASES = ("resource", "methodview", "apiview", "viewset", "modelviewset", "httpendpoint")
_HTTP_VERBS = ("get", "post", "put", "patch", "delete", "head", "options")
def _python_registered_routes(target: FileTarget, tree: ast.AST) -> list[EntryPointRecord]:
"""Handlers whose route is registered by a call or by convention, not by a decorator on the function itself
(authorization-obligations 2.6): `add_url_rule(path, view_func=fn)`, `api.add_resource(Cls, path)`, and the HTTP-verb
methods of a `Resource`/`MethodView`/`APIView`/`ViewSet` subclass. Without these, projects that register their routes
elsewhere have no named entry point and every obligation on them is unreachable."""
functions: dict[str, ast.FunctionDef | ast.AsyncFunctionDef] = {}
classes: dict[str, ast.ClassDef] = {}
for node in ast.walk(tree):
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
functions.setdefault(node.name, node)
elif isinstance(node, ast.ClassDef):
classes.setdefault(node.name, node)
registered_functions: set[str] = set()
registered_classes: set[str] = set()
for node in ast.walk(tree):
if not isinstance(node, ast.Call):
continue
callee = _safe_unparse(node.func)
trailing = callee.rsplit(".", 1)[-1]
if trailing == "add_url_rule":
for keyword in node.keywords:
if keyword.arg == "view_func":
registered_functions.add(_safe_unparse(keyword.value).split(".")[0])
registered_functions.update(_safe_unparse(arg) for arg in node.args[1:2] if isinstance(arg, ast.Name))
elif trailing in {"add_resource", "register_blueprint", "include_router"} or trailing == "add_view":
registered_classes.update(_safe_unparse(arg) for arg in node.args if isinstance(arg, ast.Name))
registered_functions.update(_safe_unparse(arg) for arg in node.args if isinstance(arg, ast.Name))
records: list[EntryPointRecord] = []
for name in sorted(registered_functions):
handler = functions.get(name)
if handler is None or _has_route_decorator(handler):
continue
decorators = [_safe_unparse(decorator) for decorator in handler.decorator_list]
access, evidence = _python_route_access(decorators)
records.append(
_entry(
target,
handler.lineno,
getattr(handler, "end_lineno", handler.lineno),
handler.name,
handler.name,
"route",
access,
"http_request",
evidence or [f"registered handler {handler.name}"],
_python_conditions(handler, decorators),
)
)
for name, klass in sorted(classes.items()):
bases = [_safe_unparse(base).rsplit(".", 1)[-1].lower() for base in klass.bases]
if name not in registered_classes and not any(base in _RESOURCE_BASES for base in bases):
continue
class_decorators = [_safe_unparse(decorator) for decorator in klass.decorator_list]
for child in klass.body:
if not isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)) or child.name.lower() not in _HTTP_VERBS:
continue
decorators = class_decorators + [_safe_unparse(decorator) for decorator in child.decorator_list]
access, evidence = _python_route_access(decorators)
records.append(
_entry(
target,
child.lineno,
getattr(child, "end_lineno", child.lineno),
child.name,
f"{name}.{child.name}",
"route",
access,
"http_request",
evidence or [f"{name} handler method {child.name}"],
_python_conditions(child, decorators),
)
)
return records
def _has_route_decorator(node: ast.FunctionDef | ast.AsyncFunctionDef) -> bool:
return any(_looks_like_route_decorator(_safe_unparse(decorator)) for decorator in node.decorator_list)
def _c_entry_points(target: FileTarget, text: str) -> list[EntryPointRecord]:
records: list[EntryPointRecord] = []
for number, line in enumerate(text.splitlines(), start=1):
@@ -476,6 +568,71 @@ def _looks_conditional(text: str) -> bool:
)
_JS_FUNCTION = re.compile(r"^\s*(?:export\s+)?(?:default\s+)?(?:async\s+)?function\s+([A-Za-z_$][\w$]*)")
_JS_EXPORTED_VERB = re.compile(r"^\s*export\s+(?:async\s+)?(?:function\s+([A-Z]+)|const\s+([A-Z]+)\s*=)")
_JS_IDENTIFIER = re.compile(r"^[A-Za-z_$][\w$.]*$")
_JS_AUTH_TOKENS = ("auth", "login", "session", "jwt", "token", "protect", "guard", "require")
_JS_ROLE_TOKENS = ("role", "permission", "admin", "owner")
def _js_function_bounds(lines: Sequence[str]) -> dict[str, tuple[int, int]]:
"""Declared function names to their 1-based line span, by brace balance from the declaration."""
bounds: dict[str, tuple[int, int]] = {}
for index, line in enumerate(lines):
match = _JS_FUNCTION.match(line)
if match is None:
continue
depth = 0
end = index
for cursor in range(index, len(lines)):
depth += lines[cursor].count("{") - lines[cursor].count("}")
end = cursor
if depth <= 0 and "{" in "".join(lines[index : cursor + 1]):
break
bounds.setdefault(match.group(1), (index + 1, end + 1))
return bounds
def _js_registration(line: str, bounds: Mapping[str, tuple[int, int]]) -> tuple[str | None, list[str]]:
"""`router.get('/books', requireAuth, listBooks)` -> the handler identifier and the middleware identifiers before it."""
body = line[line.find("(") + 1 : line.rfind(")")] if "(" in line and line.rfind(")") > line.find("(") else ""
parts = [part.strip() for part in body.split(",") if part.strip()]
identifiers = [part for part in parts[1:] if _JS_IDENTIFIER.match(part)]
if not identifiers:
return None, [line]
handler = next((name for name in reversed(identifiers) if name in bounds), identifiers[-1])
middleware = [name for name in identifiers if name != handler]
return handler, middleware
def _js_route_access(middleware: Sequence[str], line: str) -> tuple[AccessLevel, list[str]]:
lowered = " ".join(middleware).lower()
evidence = [f"middleware {name}" for name in middleware] or [line]
if any(token in lowered for token in _JS_ROLE_TOKENS):
return "role-restricted", evidence
if any(token in lowered for token in _JS_AUTH_TOKENS):
return "authenticated", evidence
return "public", evidence
def _js_route_module(target: FileTarget, lines: Sequence[str], bounds: Mapping[str, tuple[int, int]]) -> list[EntryPointRecord]:
"""Next.js App Router convention: exported HTTP-verb functions in a `route.<ext>` module are the handlers."""
if Path(target.path).stem != "route":
return []
verbs = {verb.upper() for verb in _HTTP_VERBS}
records: list[EntryPointRecord] = []
for index, line in enumerate(lines, start=1):
match = _JS_EXPORTED_VERB.match(line)
name = (match.group(1) or match.group(2)) if match else None
if name is None or name not in verbs:
continue
start, end = bounds.get(name, (index, index))
records.append(
_entry(target, start, end, name, f"route.{name}", "route", "public", "http_request", [f"route module export {name}"], [])
)
return records
def _js_handler_name(line: str) -> str | None:
if "function " in line:
return line.split("function ", 1)[1].split("(", 1)[0].strip() or None
+46
View File
@@ -215,3 +215,49 @@ def _sarif_payload(analyzer: str, rule_id: str, result_path: str) -> dict[str, o
}
],
}
def test_entry_points_name_handlers_registered_by_call_or_convention(tmp_path: Path) -> None:
"""authorization-obligations 2.6: a handler registered by a call or by convention is still a named route entry."""
from openultrasast.mapping import analyze_entry_points
from openultrasast.preprocess import preprocess_repository
(tmp_path / "api_views").mkdir()
(tmp_path / "api_views" / "books.py").write_text(
"from flask_restx import Resource\n\n\n"
"def get_by_title(book_title):\n return Book.query.filter_by(book_title=book_title).first()\n\n\n"
"class UserProfile(Resource):\n"
" @login_required\n"
" def get(self):\n return db.users.find_one({'id': request.args.get('user_id')})\n\n"
" def delete(self, user_id):\n return db.users.delete_one({'id': user_id})\n\n\n"
"app.add_url_rule('/books/<book_title>', view_func=get_by_title)\n"
"api.add_resource(UserProfile, '/users/profile')\n"
)
(tmp_path / "routes.js").write_text(
"function listBooks(req, res) { return res.json(Book.find({})); }\n"
"function stats(req, res) { return res.json({}); }\n"
"router.get('/books', requireAuth, listBooks);\n"
"router.get('/stats', stats);\n"
)
(tmp_path / "app").mkdir()
(tmp_path / "app" / "api").mkdir()
(tmp_path / "app" / "api" / "route.ts").write_text(
"export async function GET(request: NextRequest) {\n return NextResponse.json(await db.from('opportunities').select());\n}\n"
"export async function PATCH(request: NextRequest) {\n return NextResponse.json(await db.from('claims').update({}));\n}\n"
"function helper() { return 1; }\n"
)
_, targets = preprocess_repository(tmp_path)
routes = {(e.path, e.function_name): e for e in analyze_entry_points(tmp_path, targets) if e.kind == "route" and e.function_name}
named = {(path.replace("\\", "/"), function) for path, function in routes}
assert ("api_views/books.py", "get_by_title") in named # registered by add_url_rule, no decorator on the function
assert ("api_views/books.py", "get") in named and ("api_views/books.py", "delete") in named # Resource verb methods
assert ("routes.js", "listBooks") in named and ("routes.js", "stats") in named
assert ("app/api/route.ts", "GET") in named and ("app/api/route.ts", "PATCH") in named
assert ("app/api/route.ts", "helper") not in named # only the HTTP-verb exports are handlers
assert routes[("routes.js", "listBooks")].access_level == "authenticated" # requireAuth middleware on the registration
assert any("requireAuth" in item for item in routes[("routes.js", "listBooks")].access_evidence)
assert routes[("routes.js", "stats")].access_level == "public"
assert routes[("api_views/books.py", "get")].access_level == "authenticated" # @login_required on the verb method
assert routes[("api_views/books.py", "delete")].access_level == "public"
body = routes[("api_views/books.py", "get_by_title")]
assert body.line == 4 and body.end_line == 5 # the record spans the handler, not the registration line