Coverage for src/lexigram/admin/relations/routes.py: 95%

1"""Route registration for relation managers.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Callable 

6from typing import TYPE_CHECKING, Any 

7 

8from starlette.responses import HTMLResponse 

9from starlette.routing import Route 

10 

11from lexigram.logging import get_logger 

12from lexigram.ui import el, render_to_string 

13 

14if TYPE_CHECKING: 

15 from starlette.requests import Request 

16 

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 

21 

22logger = get_logger(__name__) 

23 

24 

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. 

33 

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. 

42 

43 Returns: 

44 The list of Starlette routes for the relation manager. 

45 """ 

46 prefix = f"/{resource_name}" 

47 

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) 

63 

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>") 

79 

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) 

95 

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 ) 

119 

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) 

138 

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("") 

156 

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 ] 

189 

190 

191def _create_manager( 

192 manager_class: type[RelationManager], parent_id: Any 

193) -> RelationManager: 

194 return manager_class(parent_id=parent_id) 

195 

196 

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 

203 

204 

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. 

210 

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. 

215 

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 

229 

230 

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. 

238 

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, ...). 

245 

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 

259 

260 

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. 

266 

267 The authorization middleware normally redirects unauthenticated 

268 requests before routing; this gate keeps handlers fail-closed when 

269 invoked directly. 

270 

271 Args: 

272 request: The request whose ``state`` carries the resolved user. 

273 audit_service: Optional audit service for best-effort denial logging. 

274 

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 

282 

283 

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) 

293 

294 

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 

306 

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) 

318 

319 

320__all__ = [ 

321 "register_relation_routes", 

322]