Coverage for src/lexigram/admin/controllers/settings.py: 25%
143 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 23:39 +0800
1"""Spec-driven settings controller for the admin interface.
3Renders editable configuration specs (branding, caching, security) through
4the config panel UI and persists values to the DB-backed store.
5"""
7from __future__ import annotations
9from typing import TYPE_CHECKING, Any
11from starlette.requests import Request
12from starlette.responses import HTMLResponse, RedirectResponse, Response
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
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
33logger = get_logger(__name__)
35__all__ = ["SettingsController"]
38class SettingsController(AdminController):
39 """Spec-driven settings controller.
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 """
47 prefix = "/settings"
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
64 # -- helpers --
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 )
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 ())
78 def _user_is_superadmin(self, request: Request) -> bool:
79 """Return True when the requesting user holds the superadmin role.
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)
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)
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 )
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
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
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)
160 # -- routes --
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 )
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 )
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)
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)
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 )
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 )
231 layout = ConfigLayout(
232 categories=categories,
233 active_category=spec.package_source,
234 active_namespace=namespace,
235 content=form_content,
236 title="Settings",
237 )
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 )
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)
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 )
277 form = request.scope.get("admin_form_data")
278 if form is None:
279 form = await request.form()
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 }
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 }
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)
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 )
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 )
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 )
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)
357 return HTMLResponse(flash_oob + form_html)
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 )
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))