Coverage for src / lexigram / admin / controllers / settings.py: 31%
123 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-13 22:14 +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.controllers.base import AdminController
16from lexigram.admin.settings.panel import BooleanNode
17from lexigram.admin.settings.panel.layout import ConfigLayout
18from lexigram.admin.settings.panel.registry import ConfigRegistry
19from lexigram.admin.settings.panel.types import ConfigCategory, get_default_categories
20from lexigram.admin.settings.panel.ui import ConfigDashboardUI
21from lexigram.contracts.web import get, post
22from lexigram.logging import get_logger
24if TYPE_CHECKING:
25 from lexigram.admin.auth.protocols import AdminCsrfServiceProtocol
26 from lexigram.admin.engine.renderer import AdminRenderer
27 from lexigram.admin.services.settings_service import AdminSettingsService
29logger = get_logger(__name__)
31__all__ = ["SettingsController"]
33_SYSTEM_CATEGORY = "system"
36class SettingsController(AdminController):
37 """Spec-driven settings controller.
39 Routes:
40 GET /admin/settings - Redirect to first editable spec
41 GET /admin/settings/{namespace} - Spec edit form
42 POST /admin/settings/{namespace} - Save spec values
43 """
45 prefix = "/settings"
47 def __init__(
48 self,
49 renderer: AdminRenderer,
50 settings_service: AdminSettingsService | None = None,
51 csrf_service: AdminCsrfServiceProtocol | None = None,
52 audit_service: Any = None,
53 registry: ConfigRegistry | None = None,
54 ) -> None:
55 super().__init__(renderer=renderer, settings_service=settings_service)
56 self._csrf_service = csrf_service
57 self._audit_service = audit_service
58 self._registry = registry or ConfigRegistry.with_defaults()
60 # -- helpers --
62 def _store_name(self) -> str:
63 """Use the DB store when registered, else the in-memory default."""
64 return "db" if self._registry.has_store("db") else "default"
66 @staticmethod
67 def _user_permissions(request: Request) -> frozenset[str]:
68 """Return the requesting user's permission set (empty when unknown)."""
69 user = getattr(getattr(request, "state", None), "user", None)
70 return frozenset(getattr(user, "permissions", None) or ())
72 @staticmethod
73 def _user_is_superadmin(request: Request) -> bool:
74 """Return True when the requesting user holds the superadmin role.
76 Superadmin bypasses per-spec permission gating so accounts created
77 with an empty permission set (e.g. via the setup wizard) can still
78 manage system configurations.
79 """
80 user = getattr(getattr(request, "state", None), "user", None)
81 roles = getattr(user, "roles", None) or ()
82 return "superadmin" in roles
84 def _build_categories(
85 self, request: Request
86 ) -> tuple[list[ConfigCategory], list[Any]]:
87 """Build categories with the visible specs for the requesting user."""
88 permissions = self._user_permissions(request)
89 is_superadmin = self._user_is_superadmin(request)
90 visible = [
91 spec
92 for spec in self._registry.get_specs(_SYSTEM_CATEGORY)
93 if not spec.required_permissions
94 or is_superadmin
95 or permissions.issuperset(spec.required_permissions)
96 ]
97 categories = get_default_categories()
98 for cat in categories:
99 if cat.name == _SYSTEM_CATEGORY:
100 cat.specs.extend(visible)
101 return categories, visible
103 def _get_csrf_token(self, request: Request) -> str | None:
104 """Resolve the CSRF token for form rendering, if available."""
105 if not self._csrf_service:
106 return None
107 try:
108 session = getattr(request, "session", {})
109 session_id: str = session.get("admin_user_id", "")
110 return self._csrf_service.generate_token(session_id)
111 except Exception: # noqa: BLE001 — non-fatal for form rendering
112 logger.warning("settings.csrf_token_unavailable")
113 return None
115 async def _audit(
116 self,
117 request: Request,
118 success: bool = True,
119 event_type: AdminSecurityEventType = AdminSecurityEventType.SETTINGS_UPDATED,
120 **metadata: Any,
121 ) -> None:
122 """Append a settings change to the security audit log, best-effort."""
123 if not self._audit_service:
124 return
125 try:
126 client = getattr(request, "client", None)
127 await self._audit_service.log_event(
128 event_type=event_type,
129 ip_address=getattr(client, "host", "unknown"),
130 user_agent=request.headers.get("user-agent", "") or "",
131 success=success,
132 metadata=metadata,
133 )
134 except Exception: # noqa: BLE001 — audit failures must not break saves
135 logger.warning("settings.audit_failed", **metadata)
137 # -- routes --
139 @get("/")
140 async def index(self, request: Request) -> Response:
141 """Redirect to the first editable spec, or render an empty state."""
142 _, visible = self._build_categories(request)
143 if visible:
144 return RedirectResponse(
145 url=f"/admin/settings/{visible[0].namespace}",
146 status_code=302,
147 )
149 categories = get_default_categories()
150 layout = ConfigLayout(
151 categories=categories,
152 active_category=None,
153 active_namespace=None,
154 content=None,
155 title="Settings",
156 )
157 return await self.render_admin(
158 request,
159 layout,
160 title="Settings",
161 breadcrumbs=self.generate_breadcrumbs(
162 ("Home", "/admin/"),
163 current="Settings",
164 ),
165 )
167 @get("/{namespace:path}")
168 async def spec_view(self, request: Request) -> Response:
169 """Spec detail/edit view."""
170 namespace = request.path_params.get("namespace", "")
171 spec = self._registry.get_spec(namespace)
172 if not spec or not spec.get_nodes():
173 self.flash(f"Configuration '{namespace}' not found.", "error")
174 return RedirectResponse(url="/admin/settings", status_code=302)
176 categories, _ = self._build_categories(request)
177 values = await self._registry.get_values(namespace, self._store_name())
179 ui = ConfigDashboardUI()
180 form_content = ui.render_config_form(
181 spec=spec.to_dict(),
182 values=values,
183 action=f"/admin/settings/{namespace}",
184 csrf_token=self._get_csrf_token(request),
185 )
187 layout = ConfigLayout(
188 categories=categories,
189 active_category=_SYSTEM_CATEGORY,
190 active_namespace=namespace,
191 content=form_content,
192 title="Settings",
193 )
195 return await self.render_admin(
196 request,
197 layout,
198 title=f"{spec.label or namespace} - Settings",
199 breadcrumbs=self.generate_breadcrumbs(
200 ("Home", "/admin/"),
201 ("Settings", "/admin/settings"),
202 current=spec.label or namespace,
203 ),
204 )
206 @post("/{namespace:path}")
207 async def save_spec(self, request: Request) -> Response:
208 """Save configuration changes for a spec."""
209 namespace = request.path_params.get("namespace", "")
210 spec = self._registry.get_spec(namespace)
211 if not spec or not spec.get_nodes():
212 self.flash(f"Configuration '{namespace}' not found.", "error")
213 return RedirectResponse(url="/admin/settings", status_code=302)
215 permissions = self._user_permissions(request)
216 if (
217 spec.required_permissions
218 and not self._user_is_superadmin(request)
219 and not permissions.issuperset(spec.required_permissions)
220 ):
221 await self._audit(
222 request,
223 success=False,
224 event_type=AdminSecurityEventType.PERMISSION_DENIED,
225 reason="permission_denied",
226 )
227 self.flash("You do not have permission to edit this setting.", "error")
228 return RedirectResponse(
229 url=f"/admin/settings/{namespace}",
230 status_code=302,
231 )
233 form = request.scope.get("admin_form_data")
234 if form is None:
235 form = await request.form()
237 nodes = spec.get_nodes()
238 updates = {
239 key: (
240 "true"
241 if isinstance(nodes[key], BooleanNode) and value == "on"
242 else value
243 )
244 for key, value in form.items()
245 if not key.startswith("_") and key in nodes
246 }
248 invalid = [
249 key
250 for key, value in updates.items()
251 if str(nodes[key].validate(value)).lower() != value.lower()
252 ]
253 await self._registry.save_values(namespace, updates, self._store_name())
255 await self._audit(
256 request,
257 namespace=namespace,
258 keys=sorted(updates),
259 invalid=invalid,
260 )
262 if request.headers.get("hx-request") == "true":
263 from lexigram.ui.core.base import render_to_string
265 self._flash_messages.clear()
266 if invalid:
267 toast_message = (
268 "Settings saved. Invalid values reset to defaults: "
269 + ", ".join(invalid)
270 )
271 toast_kind = "warning"
272 else:
273 toast_message = "Settings saved successfully."
274 toast_kind = "success"
275 toast_html = render_to_string(self._render_toast(toast_message, toast_kind))
276 flash_oob = (
277 f'<div id="flash-container" hx-swap-oob="true">{toast_html}</div>'
278 )
280 values = await self._registry.get_values(namespace, self._store_name())
281 ui = ConfigDashboardUI()
282 form_content = ui.render_config_form(
283 spec=spec.to_dict(),
284 values=values,
285 action=f"/admin/settings/{namespace}",
286 csrf_token=self._get_csrf_token(request),
287 )
288 form_html = render_to_string(form_content)
290 return HTMLResponse(flash_oob + form_html)
292 if invalid:
293 self.flash(
294 f"Saved. Invalid values reset to defaults: {', '.join(invalid)}",
295 "warning",
296 )
297 else:
298 self.flash("Settings saved successfully.", "success")
299 return RedirectResponse(
300 url=f"/admin/settings/{namespace}",
301 status_code=302,
302 )
304 def _render_toast(self, message: str, kind: str) -> str:
305 """Build a toast component for htmx responses."""
306 return f'<div class="toast toast-{kind}">{message}</div>'