Coverage for src/lexigram/admin/relations/routes.py: 0%
156 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 23:18 +0800
1"""Route registration for relation managers."""
3from __future__ import annotations
5from collections.abc import Callable
6from typing import TYPE_CHECKING, Any
8from starlette.responses import HTMLResponse
9from starlette.routing import Route
11from lexigram.logging import get_logger
12from lexigram.ui import el, render_to_string
14if TYPE_CHECKING:
15 from starlette.requests import Request
17 from lexigram.admin.auth.protocols import AdminAuditLogServiceProtocol
18 from lexigram.admin.exceptions import PermissionDeniedError
19 from lexigram.admin.relations.manager_ext import RelationManager
20 from lexigram.result import Result
22logger = get_logger(__name__)
25def register_relation_routes(
26 resource_name: str,
27 manager_class: type[RelationManager],
28 *,
29 parent_data_source: Any = None,
30 audit_service: AdminAuditLogServiceProtocol | None = None,
31) -> list[Route]:
32 """Create Starlette Route objects for a relation manager.
34 Args:
35 resource_name: Registered resource name, used as the route prefix.
36 manager_class: Relation manager class to mount.
37 parent_data_source: Optional data source used to resolve the parent
38 record before rendering (parent-IDOR gate). When ``None`` no
39 parent gate is applied.
40 audit_service: Optional audit service for best-effort
41 permission-denied logging.
43 Returns:
44 The list of Starlette routes for the relation manager.
45 """
46 prefix = f"/{resource_name}"
48 async def _handle_list(request: Request) -> HTMLResponse:
49 parent_id = request.path_params.get("parent_id", "")
50 denied = await _require_user(request, audit_service)
51 if denied:
52 return denied
53 mgr = _create_manager(manager_class, parent_id)
54 parent, denied = await _require_parent(mgr, parent_data_source)
55 if denied:
56 return denied
57 if parent is not None:
58 check = await _check(mgr.can_view_parent, request, audit_service, parent)
59 if check:
60 return check
61 html = await mgr.render(request, resource_name)
62 return HTMLResponse(html)
64 async def _handle_create_form(request: Request) -> HTMLResponse:
65 parent_id = request.path_params.get("parent_id", "")
66 denied = await _require_user(request, audit_service)
67 if denied:
68 return denied
69 mgr = _create_manager(manager_class, parent_id)
70 parent, denied = await _require_parent(mgr, parent_data_source)
71 if denied:
72 return denied
73 if parent is not None:
74 check = await _check(mgr.can_view_parent, request, audit_service, parent)
75 if check:
76 return check
77 form = mgr.create_form()
78 return HTMLResponse(form or "<div>No create form available</div>")
80 async def _handle_create(request: Request) -> HTMLResponse:
81 parent_id = request.path_params.get("parent_id", "")
82 denied = await _require_user(request, audit_service)
83 if denied:
84 return denied
85 mgr = _create_manager(manager_class, parent_id)
86 _, denied = await _require_parent(mgr, parent_data_source)
87 if denied:
88 return denied
89 check = await _check(mgr.can_create, request, audit_service)
90 if check:
91 return check
92 await mgr.get_query()
93 html = await mgr.render(request, resource_name)
94 return HTMLResponse(html)
96 async def _handle_edit_form(request: Request) -> HTMLResponse:
97 parent_id = request.path_params.get("parent_id", "")
98 record_id = request.path_params.get("record_id", "")
99 denied = await _require_user(request, audit_service)
100 if denied:
101 return denied
102 mgr = _create_manager(manager_class, parent_id)
103 parent, denied = await _require_parent(mgr, parent_data_source)
104 if denied:
105 return denied
106 if parent is not None:
107 check = await _check(mgr.can_view_parent, request, audit_service, parent)
108 if check:
109 return check
110 record = await _get_record(mgr, record_id)
111 if record is not None:
112 check = await _check(mgr.can_edit, request, audit_service, record)
113 if check:
114 return check
115 form = mgr.edit_form(record) if record else None
116 return HTMLResponse(
117 form or render_to_string(el("div", "Edit form for ", record_id))
118 )
120 async def _handle_update(request: Request) -> HTMLResponse:
121 parent_id = request.path_params.get("parent_id", "")
122 record_id = request.path_params.get("record_id", "")
123 denied = await _require_user(request, audit_service)
124 if denied:
125 return denied
126 mgr = _create_manager(manager_class, parent_id)
127 _, denied = await _require_parent(mgr, parent_data_source)
128 if denied:
129 return denied
130 record = await _get_record(mgr, record_id)
131 if record is None:
132 return HTMLResponse("Not found", status_code=404)
133 check = await _check(mgr.can_edit, request, audit_service, record)
134 if check:
135 return check
136 html = await mgr.render(request, resource_name)
137 return HTMLResponse(html)
139 async def _handle_delete(request: Request) -> HTMLResponse:
140 parent_id = request.path_params.get("parent_id", "")
141 record_id = request.path_params.get("record_id", "")
142 denied = await _require_user(request, audit_service)
143 if denied:
144 return denied
145 mgr = _create_manager(manager_class, parent_id)
146 _, denied = await _require_parent(mgr, parent_data_source)
147 if denied:
148 return denied
149 record = await _get_record(mgr, record_id)
150 if record is None:
151 return HTMLResponse("Not found", status_code=404)
152 check = await _check(mgr.can_delete, request, audit_service, record)
153 if check:
154 return check
155 return HTMLResponse("")
157 return [
158 Route(
159 path=f"{prefix}/{{parent_id}}/relations/{{rel_name}}",
160 endpoint=_handle_list,
161 methods=["GET"],
162 ),
163 Route(
164 path=f"{prefix}/{{parent_id}}/relations/{{rel_name}}/new",
165 endpoint=_handle_create_form,
166 methods=["GET"],
167 ),
168 Route(
169 path=f"{prefix}/{{parent_id}}/relations/{{rel_name}}",
170 endpoint=_handle_create,
171 methods=["POST"],
172 ),
173 Route(
174 path=f"{prefix}/{{parent_id}}/relations/{{rel_name}}/{{record_id}}/edit",
175 endpoint=_handle_edit_form,
176 methods=["GET"],
177 ),
178 Route(
179 path=f"{prefix}/{{parent_id}}/relations/{{rel_name}}/{{record_id}}",
180 endpoint=_handle_update,
181 methods=["PUT"],
182 ),
183 Route(
184 path=f"{prefix}/{{parent_id}}/relations/{{rel_name}}/{{record_id}}",
185 endpoint=_handle_delete,
186 methods=["DELETE"],
187 ),
188 ]
191def _create_manager(
192 manager_class: type[RelationManager], parent_id: Any
193) -> RelationManager:
194 return manager_class(parent_id=parent_id)
197async def _get_record(mgr: RelationManager, record_id: str) -> Any:
198 items = await mgr.get_query()
199 for item in items:
200 if str(getattr(item, "id", "")) == record_id:
201 return item
202 return None
205async def _require_parent(
206 mgr: RelationManager,
207 parent_data_source: Any,
208) -> tuple[Any | None, HTMLResponse | None]:
209 """Resolve the parent record through the data source, or a 404.
211 Args:
212 mgr: The relation manager whose ``parent_id`` is being resolved.
213 parent_data_source: The resource data source, or ``None`` to skip
214 the parent gate entirely.
216 Returns:
217 ``(parent, None)`` on success with the resolved parent also
218 attached to ``mgr.parent``; ``(None, None)`` when no data source
219 is mounted; ``(None, 404 response)`` when the parent record does
220 not exist.
221 """
222 if parent_data_source is None:
223 return None, None
224 parent = await parent_data_source.find_one(mgr.parent_id)
225 if parent is None:
226 return None, HTMLResponse("Parent not found", status_code=404)
227 mgr.parent = parent
228 return parent, None
231async def _check(
232 predicate: Callable[..., Result[None, PermissionDeniedError]],
233 request: Request,
234 audit_service: AdminAuditLogServiceProtocol | None,
235 *args: Any,
236) -> HTMLResponse | None:
237 """Run a permission predicate, denying with a 403 response on failure.
239 Args:
240 predicate: Manager predicate returning ``Result[None,
241 PermissionDeniedError]``.
242 request: The request whose ``state`` carries the resolved user.
243 audit_service: Optional audit service for best-effort denial logging.
244 *args: Extra predicate arguments (parent entity, record, ...).
246 Returns:
247 A 403 HTMLResponse when the predicate denies or no user is
248 present (fail-closed), ``None`` to let the handler proceed.
249 """
250 action = f"relation.{predicate.__name__}"
251 parent_id = request.path_params.get("parent_id", "")
252 user = getattr(getattr(request, "state", None), "user", None)
253 if user is None:
254 return await _deny(request, audit_service, action, parent_id=parent_id)
255 result = predicate(*args, user)
256 if result.is_err():
257 return await _deny(request, audit_service, action, parent_id=parent_id)
258 return None
261async def _require_user(
262 request: Request,
263 audit_service: AdminAuditLogServiceProtocol | None,
264) -> HTMLResponse | None:
265 """Deny unauthenticated route traffic with a 403, fail-closed.
267 The authorization middleware normally redirects unauthenticated
268 requests before routing; this gate keeps handlers fail-closed when
269 invoked directly.
271 Args:
272 request: The request whose ``state`` carries the resolved user.
273 audit_service: Optional audit service for best-effort denial logging.
275 Returns:
276 A 403 HTMLResponse when no user is present, ``None`` otherwise.
277 """
278 user = getattr(getattr(request, "state", None), "user", None)
279 if user is None:
280 return await _deny(request, audit_service, "relation.access")
281 return None
284async def _deny(
285 request: Request,
286 audit_service: AdminAuditLogServiceProtocol | None,
287 action: str,
288 parent_id: Any = "",
289) -> HTMLResponse:
290 """Return a 403 denial response, recording the event best-effort."""
291 await _audit_denial(request, audit_service, action=action, parent_id=parent_id)
292 return HTMLResponse("Permission denied", status_code=403)
295async def _audit_denial(
296 request: Request,
297 audit_service: AdminAuditLogServiceProtocol | None,
298 *,
299 action: str,
300 parent_id: Any = "",
301) -> None:
302 """Append a permission denial to the security audit log, best-effort."""
303 if not audit_service:
304 return
305 from lexigram.admin.auth.types import AdminSecurityEventType
307 try:
308 client = getattr(request, "client", None)
309 await audit_service.log_event(
310 event_type=AdminSecurityEventType.PERMISSION_DENIED,
311 ip_address=getattr(client, "host", "unknown"),
312 user_agent=request.headers.get("user-agent", "") or "",
313 success=False,
314 metadata={"action": action, "parent_id": str(parent_id)},
315 )
316 except Exception: # noqa: BLE001 — audit failures must not break denials
317 logger.warning("relations.audit_failed", action=action)
320__all__ = [
321 "register_relation_routes",
322]