Coverage for src/lexigram/admin/controllers/settings.py: 25%

1"""Spec-driven settings controller for the admin interface. 

2 

3Renders editable configuration specs (branding, caching, security) through 

4the config panel UI and persists values to the DB-backed store. 

5""" 

6 

7from __future__ import annotations 

8 

9from typing import TYPE_CHECKING, Any 

10 

11from starlette.requests import Request 

12from starlette.responses import HTMLResponse, RedirectResponse, Response 

13 

14from lexigram.admin.auth.types import AdminSecurityEventType 

15from lexigram.admin.config import AdminRbacConfig 

16from lexigram.admin.controllers.base import AdminController 

17from lexigram.admin.multitenancy.adapter import resolve_tenant_id 

18from lexigram.admin.rbac.super_admin import is_super_admin 

19from lexigram.admin.settings.panel import BooleanNode, SecretNode 

20from lexigram.admin.settings.panel.layout import ConfigLayout 

21from lexigram.admin.settings.panel.registry import ConfigRegistry 

22from lexigram.admin.settings.panel.types import ConfigCategory 

23from lexigram.admin.settings.panel.ui import ConfigDashboardUI 

24from lexigram.contracts.web import get, post 

25from lexigram.logging import get_logger 

26from lexigram.ui import el, render_to_string 

27 

28if TYPE_CHECKING: 

29 from lexigram.admin.auth.protocols import AdminCsrfServiceProtocol 

30 from lexigram.admin.engine.renderer import AdminRenderer 

31 from lexigram.admin.services.settings_service import AdminSettingsService 

32 

33logger = get_logger(__name__) 

34 

35__all__ = ["SettingsController"] 

36 

37 

38class SettingsController(AdminController): 

39 """Spec-driven settings controller. 

40 

41 Routes: 

42 GET /admin/settings - Redirect to first editable spec 

43 GET /admin/settings/{namespace} - Spec edit form 

44 POST /admin/settings/{namespace} - Save spec values 

45 """ 

46 

47 prefix = "/settings" 

48 

49 def __init__( 

50 self, 

51 renderer: AdminRenderer, 

52 settings_service: AdminSettingsService | None = None, 

53 csrf_service: AdminCsrfServiceProtocol | None = None, 

54 audit_service: Any = None, 

55 registry: ConfigRegistry | None = None, 

56 rbac_config: AdminRbacConfig | None = None, 

57 ) -> None: 

58 super().__init__(renderer=renderer, settings_service=settings_service) 

59 self._csrf_service = csrf_service 

60 self._audit_service = audit_service 

61 self._registry = registry or ConfigRegistry.with_defaults() 

62 self._rbac_config = rbac_config 

63 

64 # -- helpers -- 

65 

66 def _store_name(self, spec: type[Any]) -> str: 

67 """Use the spec's configured store when registered, else the in-memory default.""" 

68 return ( 

69 spec.store_name if self._registry.has_store(spec.store_name) else "default" 

70 ) 

71 

72 @staticmethod 

73 def _user_permissions(request: Request) -> frozenset[str]: 

74 """Return the requesting user's permission set (empty when unknown).""" 

75 user = getattr(getattr(request, "state", None), "user", None) 

76 return frozenset(getattr(user, "permissions", None) or ()) 

77 

78 def _user_is_superadmin(self, request: Request) -> bool: 

79 """Return True when the requesting user holds the superadmin role. 

80 

81 Superadmin bypasses per-spec permission gating so accounts created 

82 with an empty permission set (e.g. via the setup wizard) can still 

83 manage system configurations. 

84 """ 

85 role = (self._rbac_config or AdminRbacConfig()).super_admin_role 

86 user = getattr(getattr(request, "state", None), "user", None) 

87 return is_super_admin(user, role) 

88 

89 def _build_categories( 

90 self, request: Request 

91 ) -> tuple[list[ConfigCategory], list[Any]]: 

92 """Build one category per package source, with visible specs for the user.""" 

93 permissions = self._user_permissions(request) 

94 is_superadmin = self._user_is_superadmin(request) 

95 

96 def _is_visible(spec: Any) -> bool: 

97 return ( 

98 not spec.required_permissions 

99 or is_superadmin 

100 or permissions.issuperset(spec.required_permissions) 

101 ) 

102 

103 categories: list[ConfigCategory] = [] 

104 visible: list[Any] = [] 

105 for order, package_source in enumerate(self._registry.get_package_sources()): 

106 specs = [ 

107 spec 

108 for spec in self._registry.get_specs_by_package(package_source) 

109 if _is_visible(spec) 

110 ] 

111 visible.extend(specs) 

112 categories.append( 

113 ConfigCategory( 

114 name=package_source, 

115 label=package_source.replace("-", " ").replace("_", " ").title(), 

116 order=order * 10, 

117 specs=specs, 

118 ) 

119 ) 

120 return categories, visible 

121 

122 def _get_csrf_token(self, request: Request) -> str | None: 

123 """Resolve the CSRF token for form rendering, if available.""" 

124 if not self._csrf_service: 

125 return None 

126 try: 

127 session = getattr(request, "session", {}) 

128 # Canonical session-id resolution — keep in sync with 

129 # middleware/csrf.py _validate_csrf and resources/handler.py. 

130 session_id: str = session.get("csrf_session_id") or session.get( 

131 "admin_user_id", "anonymous" 

132 ) 

133 return self._csrf_service.generate_token(session_id) 

134 except Exception: # noqa: BLE001 — non-fatal for form rendering 

135 logger.warning("settings.csrf_token_unavailable") 

136 return None 

137 

138 async def _audit( 

139 self, 

140 request: Request, 

141 success: bool = True, 

142 event_type: AdminSecurityEventType = AdminSecurityEventType.SETTINGS_UPDATED, 

143 **metadata: Any, 

144 ) -> None: 

145 """Append a settings change to the security audit log, best-effort.""" 

146 if not self._audit_service: 

147 return 

148 try: 

149 client = getattr(request, "client", None) 

150 await self._audit_service.log_event( 

151 event_type=event_type, 

152 ip_address=getattr(client, "host", "unknown"), 

153 user_agent=request.headers.get("user-agent", "") or "", 

154 success=success, 

155 metadata=metadata, 

156 ) 

157 except Exception: # noqa: BLE001 — audit failures must not break saves 

158 logger.warning("settings.audit_failed", **metadata) 

159 

160 # -- routes -- 

161 

162 @get("/") 

163 async def index(self, request: Request) -> Response: 

164 """Redirect to the first editable spec, or render an empty state.""" 

165 categories, visible = self._build_categories(request) 

166 if visible: 

167 return RedirectResponse( 

168 url=f"/admin/settings/{visible[0].namespace}", 

169 status_code=302, 

170 ) 

171 

172 layout = ConfigLayout( 

173 categories=categories, 

174 active_category=None, 

175 active_namespace=None, 

176 content=None, 

177 title="Settings", 

178 ) 

179 return await self.render_admin( 

180 request, 

181 layout, 

182 title="Settings", 

183 breadcrumbs=self.generate_breadcrumbs( 

184 ("Home", "/admin/"), 

185 current="Settings", 

186 ), 

187 ) 

188 

189 @get("/{namespace:path}") 

190 async def spec_view(self, request: Request) -> Response: 

191 """Spec detail/edit view.""" 

192 namespace = request.path_params.get("namespace", "") 

193 spec = self._registry.get_spec(namespace) 

194 if not spec or not spec.get_nodes(): 

195 self.flash(f"Configuration '{namespace}' not found.", "error") 

196 return RedirectResponse(url="/admin/settings", status_code=302) 

197 

198 permissions = self._user_permissions(request) 

199 if ( 

200 spec.required_permissions 

201 and not self._user_is_superadmin(request) 

202 and not permissions.issuperset(spec.required_permissions) 

203 ): 

204 await self._audit( 

205 request, 

206 success=False, 

207 event_type=AdminSecurityEventType.PERMISSION_DENIED, 

208 reason="permission_denied", 

209 ) 

210 self.flash("You do not have permission to view this setting.", "error") 

211 return RedirectResponse(url="/admin/settings", status_code=302) 

212 

213 categories, _ = self._build_categories(request) 

214 tenant_id = ( 

215 await resolve_tenant_id(request, default="default") 

216 if spec.scope == "tenant" 

217 else None 

218 ) 

219 values = await self._registry.get_values( 

220 namespace, self._store_name(spec), tenant_id=tenant_id 

221 ) 

222 

223 ui = ConfigDashboardUI() 

224 form_content = ui.render_config_form( 

225 spec=spec.to_dict(), 

226 values=values, 

227 action=f"/admin/settings/{namespace}", 

228 csrf_token=self._get_csrf_token(request), 

229 ) 

230 

231 layout = ConfigLayout( 

232 categories=categories, 

233 active_category=spec.package_source, 

234 active_namespace=namespace, 

235 content=form_content, 

236 title="Settings", 

237 ) 

238 

239 return await self.render_admin( 

240 request, 

241 layout, 

242 title=f"{spec.label or namespace} - Settings", 

243 breadcrumbs=self.generate_breadcrumbs( 

244 ("Home", "/admin/"), 

245 ("Settings", "/admin/settings"), 

246 current=spec.label or namespace, 

247 ), 

248 ) 

249 

250 @post("/{namespace:path}") 

251 async def save_spec(self, request: Request) -> Response: 

252 """Save configuration changes for a spec.""" 

253 namespace = request.path_params.get("namespace", "") 

254 spec = self._registry.get_spec(namespace) 

255 if not spec or not spec.get_nodes(): 

256 self.flash(f"Configuration '{namespace}' not found.", "error") 

257 return RedirectResponse(url="/admin/settings", status_code=302) 

258 

259 permissions = self._user_permissions(request) 

260 if ( 

261 spec.required_permissions 

262 and not self._user_is_superadmin(request) 

263 and not permissions.issuperset(spec.required_permissions) 

264 ): 

265 await self._audit( 

266 request, 

267 success=False, 

268 event_type=AdminSecurityEventType.PERMISSION_DENIED, 

269 reason="permission_denied", 

270 ) 

271 self.flash("You do not have permission to edit this setting.", "error") 

272 return RedirectResponse( 

273 url=f"/admin/settings/{namespace}", 

274 status_code=302, 

275 ) 

276 

277 form = request.scope.get("admin_form_data") 

278 if form is None: 

279 form = await request.form() 

280 

281 nodes = spec.get_nodes() 

282 multi = getattr(form, "multi_items", None) 

283 raw_items = list(multi()) if multi else list(form.items()) 

284 updates = { 

285 key: ( 

286 "true" 

287 if isinstance(nodes[key], BooleanNode) 

288 and any(_value == "on" for _key, _value in raw_items if _key == key) 

289 else value 

290 ) 

291 for key, value in raw_items 

292 if not key.startswith("_") and key in nodes 

293 } 

294 

295 ignored_readonly = sorted(key for key in updates if nodes[key].readonly) 

296 editable_updates = { 

297 key: value for key, value in updates.items() if not nodes[key].readonly 

298 } 

299 

300 cleared_secrets: list[str] = [] 

301 for key in list(editable_updates): 

302 if isinstance(nodes[key], SecretNode) and not editable_updates[key]: 

303 editable_updates.pop(key) 

304 cleared_secrets.append(key) 

305 

306 invalid = [ 

307 key 

308 for key, value in editable_updates.items() 

309 if str(nodes[key].validate(value)).lower() != value.lower() 

310 ] 

311 tenant_id = ( 

312 await resolve_tenant_id(request, default="default") 

313 if spec.scope == "tenant" 

314 else None 

315 ) 

316 await self._registry.save_values( 

317 namespace, editable_updates, self._store_name(spec), tenant_id=tenant_id 

318 ) 

319 

320 await self._audit( 

321 request, 

322 namespace=namespace, 

323 keys=sorted(editable_updates), 

324 invalid=invalid, 

325 ignored_readonly=ignored_readonly, 

326 cleared_secrets=sorted(cleared_secrets), 

327 ) 

328 

329 if request.headers.get("hx-request") == "true": 

330 self._flash_messages.clear() 

331 if invalid: 

332 toast_message = ( 

333 "Settings saved. Invalid values reset to defaults: " 

334 + ", ".join(invalid) 

335 ) 

336 toast_kind = "warning" 

337 else: 

338 toast_message = "Settings saved successfully." 

339 toast_kind = "success" 

340 toast_html = self._render_toast(toast_message, toast_kind) 

341 flash_oob = ( 

342 f'<div id="flash-container" hx-swap-oob="true">{toast_html}</div>' 

343 ) 

344 

345 values = await self._registry.get_values( 

346 namespace, self._store_name(spec), tenant_id=tenant_id 

347 ) 

348 ui = ConfigDashboardUI() 

349 form_content = ui.render_config_form( 

350 spec=spec.to_dict(), 

351 values=values, 

352 action=f"/admin/settings/{namespace}", 

353 csrf_token=self._get_csrf_token(request), 

354 ) 

355 form_html = render_to_string(form_content) 

356 

357 return HTMLResponse(flash_oob + form_html) 

358 

359 if invalid: 

360 self.flash( 

361 f"Saved. Invalid values reset to defaults: {', '.join(invalid)}", 

362 "warning", 

363 ) 

364 else: 

365 self.flash("Settings saved successfully.", "success") 

366 return RedirectResponse( 

367 url=f"/admin/settings/{namespace}", 

368 status_code=302, 

369 ) 

370 

371 def _render_toast(self, message: str, kind: str) -> str: 

372 """Build a toast component for htmx responses.""" 

373 return render_to_string(el("div", {"class": f"toast toast-{kind}"}, message))