Coverage for src/lexigram/admin/controllers/plugins.py: 91%
122 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 14:56 +0800
1"""Plugins controller — list and toggle plugins via the plugin-state toolbox.
3The page is a thin consumer of the passive ``lexigram-plugins`` primitives:
4``discover_plugins()`` lists metadata-only descriptors, and
5``load_disabled``/``update_disabled`` read and mutate the boot-file mirror
6(``.lexigram/plugins.json`` by default) under an exclusive file lock.
7Toggling applies on next boot — the mechanism stays passive; this
8controller never instantiates plugins, only the admin-facing surface over
9the same primitives an app would call.
10"""
12from __future__ import annotations
14import secrets
15from typing import TYPE_CHECKING, Any
16from urllib.parse import quote_plus
18from starlette.requests import Request
19from starlette.responses import RedirectResponse, Response
21from lexigram.admin.config import AdminRbacConfig
22from lexigram.admin.controllers.base import AdminController
23from lexigram.admin.rbac.super_admin import is_super_admin
24from lexigram.contracts.web import get, post
25from lexigram.logging import get_logger
26from lexigram.ui import el
28if TYPE_CHECKING:
29 from lexigram.admin.auth.protocols import (
30 AdminAuditLogServiceProtocol,
31 AdminCsrfServiceProtocol,
32 )
33 from lexigram.admin.engine.renderer import AdminRenderer
34 from lexigram.contracts.plugins import PluginDescriptor
36logger = get_logger(__name__)
38__all__ = ["PluginsController"]
40_REQUIRED_PERMISSIONS = frozenset({"admin.settings.edit"})
43def _load_toolbox() -> tuple[Any, Any] | None:
44 """Import the lexigram-plugins toolbox, returning None when absent.
46 Returns:
47 A ``(discovery, state)`` module pair, or ``None`` when the
48 ``lexigram-plugins`` package is not installed.
49 """
50 try:
51 from lexigram.plugins import discovery, state
53 return (discovery, state)
54 except ImportError:
55 logger.debug("plugins.controller_toolbox_missing")
56 return None
59class PluginsController(AdminController):
60 """Admin page listing discovered plugins with enable/disable toggles.
62 Routes:
63 GET /admin/plugins - Plugin listing
64 POST /admin/plugins/toggle - Toggle one plugin (disable/enable)
65 """
67 prefix = "/plugins"
69 def __init__(
70 self,
71 renderer: AdminRenderer,
72 csrf_service: AdminCsrfServiceProtocol | None = None,
73 audit_service: AdminAuditLogServiceProtocol | None = None,
74 rbac_config: AdminRbacConfig | None = None,
75 **kwargs: Any,
76 ) -> None:
77 super().__init__(renderer=renderer, **kwargs)
78 self._csrf_service = csrf_service
79 self._audit_service = audit_service
80 self._rbac_config = rbac_config
82 # ------------------------------------------------------------------
83 # Routes
84 # ------------------------------------------------------------------
86 @get("/")
87 async def index(self, request: Request) -> Response:
88 # Sec-2026-08-16-L5: no per-page permission on the read-only listing.
89 # The global AdminAuthorizationMiddleware gates every non-public
90 # request; this page exposes only entry-point metadata + the disabled
91 # set; only POST /toggle requires superadmin / admin.settings.edit.
92 # Accepted posture — see
93 # docs/superpowers/specs/2026-08-16-security-plugins-design.md.
94 """Render the plugin listing page."""
95 toolbox = _load_toolbox()
96 if toolbox is None:
97 return await self.render_admin(
98 request,
99 self._render_empty("No plugin toolbox available."),
100 title="Plugins",
101 breadcrumbs=self.generate_breadcrumbs(
102 ("Home", "/admin/"),
103 current="Plugins",
104 ),
105 )
106 discovery, state = toolbox
108 descriptors = discovery.discover_plugins()
109 disabled = state.load_disabled()
110 if not descriptors:
111 return await self.render_admin(
112 request,
113 self._render_empty("No plugins discovered."),
114 title="Plugins",
115 breadcrumbs=self.generate_breadcrumbs(
116 ("Home", "/admin/"),
117 current="Plugins",
118 ),
119 )
121 content = self._render_list(request, descriptors, disabled)
122 return await self.render_admin(
123 request,
124 content,
125 title="Plugins",
126 breadcrumbs=self.generate_breadcrumbs(
127 ("Home", "/admin/"),
128 current="Plugins",
129 ),
130 )
132 @post("/toggle")
133 async def toggle(self, request: Request) -> Response:
134 """Toggle one plugin's enabled state (applies on next boot)."""
135 form = request.scope.get("admin_form_data") or await request.form()
136 csrf_token = str(form.get("csrf_token", ""))
137 if not self._user_can_manage(request):
138 await self._audit(request, success=False, reason="permission_denied")
139 return self._error_redirect("/admin/plugins", "Permission denied.")
140 if not self._csrf_ok(request, csrf_token):
141 await self._audit(request, success=False, reason="csrf_failed")
142 return self._error_redirect("/admin/plugins", "Invalid CSRF token.")
144 plugin_name = str(form.get("plugin", "")).strip()
145 toolbox = _load_toolbox()
146 if toolbox is None:
147 await self._audit(request, success=False, reason="toolbox_missing")
148 return self._error_redirect("/admin/plugins", "Toolbox not available.")
149 discovery, state = toolbox
151 descriptors = discovery.discover_plugins()
152 entry_points = {
153 d.provider_entry_point
154 for d in descriptors
155 if self._is_known(d, plugin_name)
156 }
157 if not entry_points:
158 await self._audit(request, success=False, reason="plugin_unknown")
159 return self._error_redirect("/admin/plugins", "Unknown plugin.")
161 disabled = state.load_disabled()
162 action = "enabled" if entry_points <= disabled else "disabled"
164 def flip(current: set[str]) -> set[str]:
165 """Toggle ``entry_points`` out of (or into) ``current``."""
166 if entry_points <= current:
167 current.difference_update(entry_points)
168 else:
169 current.update(entry_points)
170 return current
172 from lexigram.plugins.exceptions import PluginStateError
174 try:
175 state.update_disabled(flip)
176 except PluginStateError:
177 await self._audit(request, success=False, reason="state_write_failed")
178 return self._error_redirect(
179 "/admin/plugins", "Could not save plugin state."
180 )
181 await self._audit(request, success=True, plugin=plugin_name, action=action)
182 return RedirectResponse(
183 url=f"/admin/plugins?notice={quote_plus(f'Plugin {action}.')}",
184 status_code=302,
185 )
187 # ------------------------------------------------------------------
188 # Rendering
189 # ------------------------------------------------------------------
191 def _render_empty(self, message: str) -> Any:
192 return el(
193 "div",
194 el(
195 "h1",
196 "Plugins",
197 class_="text-2xl font-bold text-foreground",
198 ),
199 el(
200 "p",
201 message,
202 class_="text-muted-foreground mt-2",
203 ),
204 class_="p-6",
205 )
207 def _render_list(
208 self,
209 request: Request,
210 descriptors: list[PluginDescriptor],
211 disabled: set[str],
212 ) -> Any:
213 rows = [
214 self._render_row(request, d, d.provider_entry_point in disabled)
215 for d in descriptors
216 ]
217 return el(
218 "div",
219 el("h1", "Plugins", class_="text-2xl font-bold text-foreground"),
220 el(
221 "p",
222 "Enable or disable installed plugins. Changes apply on next boot.",
223 class_="text-muted-foreground mt-1",
224 ),
225 el(
226 "div",
227 *rows,
228 class_="mt-6 divide-y divide-border border border-border rounded-xl",
229 ),
230 class_="p-6",
231 )
233 def _render_row(
234 self,
235 request: Request,
236 descriptor: PluginDescriptor,
237 is_disabled: bool,
238 ) -> Any:
239 label = "Enable" if is_disabled else "Disable"
240 status = "Disabled" if is_disabled else "Enabled"
241 status_class = (
242 "bg-red-50 text-red-700 dark:bg-red-900/30 dark:text-red-300"
243 if is_disabled
244 else "bg-green-50 text-green-700 dark:bg-green-900/30 dark:text-green-300"
245 )
246 return el(
247 "div",
248 el(
249 "div",
250 el(
251 "h3",
252 descriptor.display_name,
253 class_="text-base font-semibold text-foreground",
254 ),
255 el(
256 "p",
257 descriptor.description,
258 class_="text-sm text-muted-foreground mt-1",
259 ),
260 el(
261 "span",
262 f"v{descriptor.version}",
263 class_="text-xs text-muted-foreground",
264 ),
265 class_="flex-1",
266 ),
267 el(
268 "span",
269 status,
270 class_=f"px-2 py-1 rounded text-xs font-medium {status_class}",
271 ),
272 el(
273 "form",
274 el("input", type_="hidden", name="plugin", value=descriptor.name),
275 el(
276 "input",
277 type_="hidden",
278 name="csrf_token",
279 value=self._get_csrf_token(request),
280 ),
281 el(
282 "button",
283 label,
284 type_="submit",
285 class_="px-3 py-1.5 rounded text-sm bg-primary-600 text-white hover:bg-primary-700",
286 ),
287 method="post",
288 action="/admin/plugins/toggle",
289 class_="ml-4",
290 ),
291 class_="flex items-center gap-4 p-4",
292 )
294 def _is_known(self, descriptor: PluginDescriptor, name: str) -> bool:
295 """Return True when *name* matches this descriptor's plugin name."""
296 return descriptor.name == name
298 # ------------------------------------------------------------------
299 # Helpers
300 # ------------------------------------------------------------------
302 @staticmethod
303 def _user_permissions(request: Request) -> frozenset[str]:
304 """Return the requesting user's permission set (empty when unknown)."""
305 user = getattr(getattr(request, "state", None), "user", None)
306 return frozenset(getattr(user, "permissions", None) or ())
308 def _user_is_superadmin(self, request: Request) -> bool:
309 """Return True when the requesting user holds the superadmin role."""
310 role = (self._rbac_config or AdminRbacConfig()).super_admin_role
311 user = getattr(getattr(request, "state", None), "user", None)
312 return is_super_admin(user, role)
314 def _user_can_manage(self, request: Request) -> bool:
315 """Return True when the user may toggle plugins."""
316 return self._user_is_superadmin(request) or self._user_permissions(
317 request
318 ).issuperset(_REQUIRED_PERMISSIONS)
320 def _get_csrf_token(self, request: Request) -> str:
321 """Return a CSRF token, creating and persisting the session id."""
322 if self._csrf_service is None:
323 return ""
324 csrf_session_id = str(request.session.get("csrf_session_id", ""))
325 if not csrf_session_id:
326 csrf_session_id = secrets.token_hex(16)
327 request.session["csrf_session_id"] = csrf_session_id
328 return self._csrf_service.generate_token(csrf_session_id)
330 def _csrf_ok(self, request: Request, csrf_token: str) -> bool:
331 """Validate the submitted CSRF token against the session id."""
332 csrf_session_id = str(request.session.get("csrf_session_id", ""))
333 return bool(
334 csrf_session_id
335 and self._csrf_service
336 and self._csrf_service.validate_token(csrf_session_id, csrf_token)
337 )
339 async def _audit(
340 self,
341 request: Request,
342 success: bool,
343 **metadata: Any,
344 ) -> None:
345 """Log a plugin toggle to the security audit log, best-effort."""
346 if not self._audit_service:
347 return
348 try:
349 from lexigram.admin.auth.types import AdminSecurityEventType
351 client = getattr(request, "client", None)
352 await self._audit_service.log_event(
353 event_type=AdminSecurityEventType.SETTINGS_UPDATED,
354 ip_address=getattr(client, "host", "unknown"),
355 user_agent=request.headers.get("user-agent", "") or "",
356 success=success,
357 metadata=metadata,
358 )
359 except Exception: # noqa: BLE001 — audit failures must not break toggles
360 logger.warning("plugins.audit_failed", **metadata)
362 @staticmethod
363 def _error_redirect(url: str, message: str) -> RedirectResponse:
364 """Return a 302 redirect carrying an error flash message."""
365 return RedirectResponse(
366 url=f"{url}?error={quote_plus(message)}",
367 status_code=302,
368 )