Coverage for src/lexigram/admin/services/impersonation.py: 0%
102 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 23:18 +0800
1"""User impersonation service for admin panels.
3Allows super-admin users to temporarily assume the identity of another user
4for debugging and support purposes. Every impersonation session is audit-
5logged. The original admin identity is preserved so it can be restored.
7Usage::
9 service = ImpersonationService(
10 audit_logger=audit_logger,
11 policy=my_policy,
12 )
14 result = await service.start(
15 actor=admin_user,
16 target_user_id="user-123",
17 request=request,
18 reason="Support ticket #456",
19 )
21 if result.is_ok():
22 impersonation = result.unwrap()
23 # Swap session identity to impersonation.target_user_id ...
25 # Later, to restore the original admin identity:
26 await service.stop(admin_user, request)
28.. note::
29 Wired via ``ImpersonateAction`` (``lexigram.admin.actions.standard``)
30 and ``ImpersonationController`` (``lexigram.admin.controllers.
31 impersonation``), which register ``POST /admin/impersonate/{user_id}``
32 and ``POST /admin/impersonate/stop``. See
33 ``docs/superpowers/specs/2026-08-16-security-impersonation-design.md``
34 for the security gaps that were closed before wiring (nested
35 sessions, target-role restriction, multi-worker session visibility).
36"""
38from __future__ import annotations
40from dataclasses import dataclass, field
41from datetime import UTC, datetime
42from typing import Any
43import uuid
45from lexigram.admin.config import AdminRbacConfig
46from lexigram.admin.exceptions import NotFoundError, PermissionDeniedError
47from lexigram.admin.rbac.super_admin import is_super_admin
48from lexigram.contracts.audit import AuditEntry, AuditEventSeverity, AuditLoggerProtocol
49from lexigram.di.decorators import inject
50from lexigram.logging import get_logger
51from lexigram.result import Err, Ok, Result
53logger = get_logger(__name__)
55# Session-state key used to store impersonation context
56_SESSION_KEY = "_admin_impersonation"
57# Attribute on request.state that holds original admin identity
58_ORIGINAL_USER_KEY = "_impersonation_original_user_id"
61@dataclass(frozen=True)
62class ImpersonationSession:
63 """Represents an active impersonation session.
65 Attributes:
66 id: Unique impersonation session identifier.
67 actor_id: Admin user who initiated the impersonation.
68 target_user_id: User being impersonated.
69 started_at: UTC timestamp when impersonation began.
70 reason: Optional free-text reason for audit trail.
71 """
73 id: str = field(default_factory=lambda: str(uuid.uuid4()))
74 actor_id: str = ""
75 target_user_id: str = ""
76 started_at: datetime = field(default_factory=lambda: datetime.now(UTC))
77 reason: str = ""
80class ImpersonationPolicy:
81 """Policy: only holders of the configured super-admin role may impersonate.
83 Override ``can_impersonate`` to customise authorisation logic.
84 """
86 def __init__(self, super_admin_role: str = "superadmin") -> None:
87 """Store the role name that grants impersonation rights.
89 Args:
90 super_admin_role: Role name (default ``"superadmin"``) that
91 grants super-admin rights.
92 """
93 self._super_admin_role = super_admin_role
95 def can_impersonate(self, actor: Any) -> bool:
96 """Return True if *actor* is allowed to impersonate another user.
98 Args:
99 actor: The admin user attempting to impersonate.
101 Returns:
102 True if the actor holds the configured super-admin role.
103 """
104 return is_super_admin(actor, self._super_admin_role)
106 def can_impersonate_target(self, target_roles: list[str]) -> bool:
107 """Return True unless *target_roles* includes the super-admin role.
109 Args:
110 target_roles: Role names held by the user being impersonated.
112 Returns:
113 False if the target holds the configured super-admin role
114 (impersonating another super-admin is never allowed); True
115 otherwise, including when *target_roles* is empty.
116 """
117 return self._super_admin_role not in target_roles
120@inject
121class ImpersonationService:
122 """Service that manages user impersonation sessions.
124 Args:
125 audit_logger: Optional audit logger (AuditLoggerProtocol).
126 If ``None``, impersonation events are only logged via the standard
127 logger; no structured audit record is persisted.
128 policy: Policy object that determines who may impersonate.
129 Defaults to ``ImpersonationPolicy`` (superadmin-only).
130 active_sessions: Optional pre-populated in-memory store used in tests.
131 """
133 def __init__(
134 self,
135 audit_logger: AuditLoggerProtocol | None = None,
136 policy: ImpersonationPolicy | None = None,
137 active_sessions: dict[str, ImpersonationSession] | None = None,
138 rbac_config: AdminRbacConfig | None = None,
139 ) -> None:
140 self._audit = audit_logger
141 self._policy = policy or ImpersonationPolicy(
142 super_admin_role=(rbac_config or AdminRbacConfig()).super_admin_role
143 )
144 # actor_id → active ImpersonationSession
145 self._sessions: dict[str, ImpersonationSession] = (
146 active_sessions if active_sessions is not None else {}
147 )
149 # ------------------------------------------------------------------
150 # Public API
151 # ------------------------------------------------------------------
153 async def start(
154 self,
155 actor: Any,
156 target_user_id: str,
157 reason: str = "",
158 request: Any | None = None,
159 target_roles: list[str] | None = None,
160 ) -> Result[ImpersonationSession, PermissionDeniedError]:
161 """Begin impersonating *target_user_id* on behalf of *actor*.
163 Args:
164 actor: The authenticated admin user requesting impersonation.
165 target_user_id: ID of the user to impersonate.
166 reason: Free-text reason recorded in the audit trail.
167 request: Optional Starlette request — if provided, the
168 impersonation token is stored in ``request.session``.
169 target_roles: Optional list of role names held by the target
170 user. When provided and it includes the configured
171 super-admin role, the attempt is denied.
173 Returns:
174 ``Ok(ImpersonationSession)`` on success, or
175 ``Err(PermissionDeniedError)`` if the actor is not authorised
176 or already has an active impersonation session.
177 """
178 actor_id: str = getattr(actor, "id", str(actor))
180 if not self._policy.can_impersonate(actor):
181 logger.warning(
182 "impersonation.denied",
183 actor_id=actor_id,
184 target_user_id=target_user_id,
185 )
186 return Err(
187 PermissionDeniedError(
188 f"User {actor_id!r} is not authorised to impersonate other users"
189 )
190 )
192 # D1 — nested-session guard: an actor with a live impersonation
193 # session must stop it before starting another. Silently replacing
194 # the session would orphan the original target identity and break
195 # audit attribution.
196 existing = self._sessions.get(actor_id)
197 if existing is not None:
198 logger.warning(
199 "impersonation.nested_denied",
200 actor_id=actor_id,
201 active_target_user_id=existing.target_user_id,
202 requested_target_user_id=target_user_id,
203 )
204 return Err(
205 PermissionDeniedError(
206 f"User {actor_id!r} is already impersonating "
207 f"{existing.target_user_id!r}; stop the active "
208 "impersonation session first"
209 )
210 )
212 # D2 — target-role restriction: a super-admin must never be
213 # impersonated, even by another super-admin. The caller supplies
214 # the target's roles as resolved from the user store.
215 if target_roles and not self._policy.can_impersonate_target(target_roles):
216 logger.warning(
217 "impersonation.target_denied",
218 actor_id=actor_id,
219 target_user_id=target_user_id,
220 )
221 return Err(
222 PermissionDeniedError(
223 f"Target user {target_user_id!r} holds the super-admin role "
224 "and cannot be impersonated."
225 )
226 )
228 session = ImpersonationSession(
229 actor_id=actor_id,
230 target_user_id=target_user_id,
231 reason=reason,
232 )
233 self._sessions[actor_id] = session
235 if request is not None:
236 req_session = getattr(request, "session", None)
237 if req_session is not None:
238 req_session[_SESSION_KEY] = {
239 "id": session.id,
240 "actor_id": actor_id,
241 "target_user_id": target_user_id,
242 "started_at": session.started_at.isoformat(),
243 "reason": reason,
244 }
245 req_session[_ORIGINAL_USER_KEY] = actor_id
247 await self._emit_audit(
248 action="impersonation.start",
249 actor_id=actor_id,
250 target_user_id=target_user_id,
251 session_id=session.id,
252 reason=reason,
253 request=request,
254 )
255 logger.info(
256 "impersonation.started",
257 actor_id=actor_id,
258 target_user_id=target_user_id,
259 session_id=session.id,
260 )
261 return Ok(session)
263 async def stop(
264 self,
265 actor: Any,
266 request: Any | None = None,
267 ) -> Result[str, NotFoundError]:
268 """End the active impersonation session for *actor*.
270 Args:
271 actor: The authenticated admin user (original identity).
272 request: Optional Starlette request — session cookie is cleared
273 when provided.
275 Returns:
276 ``Ok(original_user_id)`` or ``Err(NotFoundError)`` when no
277 active session exists.
278 """
279 actor_id: str = getattr(actor, "id", str(actor))
281 session = self._sessions.pop(actor_id, None)
282 if session is None and request is not None:
283 req_session = getattr(request, "session", None)
284 if req_session:
285 raw = req_session.get(_SESSION_KEY)
286 if isinstance(raw, dict):
287 actor_id = raw.get("actor_id", actor_id)
288 session = ImpersonationSession(
289 id=raw.get("id", ""),
290 actor_id=actor_id,
291 target_user_id=raw.get("target_user_id", ""),
292 reason=raw.get("reason", ""),
293 )
295 if session is None:
296 return Err(NotFoundError("No active impersonation session"))
298 if request is not None:
299 req_session = getattr(request, "session", None)
300 if req_session is not None:
301 req_session.pop(_SESSION_KEY, None)
302 req_session.pop(_ORIGINAL_USER_KEY, None)
304 await self._emit_audit(
305 action="impersonation.stop",
306 actor_id=session.actor_id,
307 target_user_id=session.target_user_id,
308 session_id=session.id,
309 reason="",
310 request=request,
311 )
312 logger.info(
313 "impersonation.stopped",
314 actor_id=session.actor_id,
315 target_user_id=session.target_user_id,
316 session_id=session.id,
317 )
318 return Ok(session.actor_id)
320 def get_active_session(
321 self, actor_id: str, request: Any | None = None
322 ) -> ImpersonationSession | None:
323 """Return the active ``ImpersonationSession`` for *actor_id*, or ``None``.
325 Args:
326 actor_id: Admin user ID to look up.
327 request: Optional Starlette request — when the in-process
328 session store has no entry (e.g. a different worker
329 handled the original ``start()`` call), the session is
330 reconstructed from ``request.session`` if present.
332 Returns:
333 Active session, or ``None`` if the user is not impersonating.
334 """
335 session = self._sessions.get(actor_id)
336 if session is not None:
337 return session
338 if request is not None:
339 req_session = getattr(request, "session", None)
340 raw = req_session.get(_SESSION_KEY) if req_session else None
341 if isinstance(raw, dict) and raw.get("actor_id") == actor_id:
342 return ImpersonationSession(
343 id=raw.get("id", ""),
344 actor_id=actor_id,
345 target_user_id=raw.get("target_user_id", ""),
346 reason=raw.get("reason", ""),
347 )
348 return None
350 def is_impersonating(self, actor_id: str, request: Any | None = None) -> bool:
351 """Return ``True`` if *actor_id* currently has an active impersonation.
353 Args:
354 actor_id: Admin user ID to check.
355 request: Optional Starlette request, passed through to
356 ``get_active_session`` for cross-worker fallback.
358 Returns:
359 True if an active session exists for this actor.
360 """
361 return self.get_active_session(actor_id, request) is not None
363 def list_active(self) -> list[ImpersonationSession]:
364 """Return all currently active impersonation sessions.
366 Returns:
367 List of active ``ImpersonationSession`` objects.
368 """
369 return list(self._sessions.values())
371 # ------------------------------------------------------------------
372 # Internal helpers
373 # ------------------------------------------------------------------
375 async def _emit_audit(
376 self,
377 action: str,
378 actor_id: str,
379 target_user_id: str,
380 session_id: str,
381 reason: str,
382 request: Any | None,
383 ) -> None:
384 """Emit an audit log entry if an audit logger is configured."""
385 if self._audit is None:
386 return
388 ip = None
389 user_agent = None
390 if request is not None:
391 client = getattr(request, "client", None)
392 ip = getattr(client, "host", None) if client else None
393 headers = getattr(request, "headers", {})
394 user_agent = headers.get("user-agent")
396 await self._audit.log(
397 AuditEntry(
398 action=action,
399 actor_id=actor_id,
400 resource_type="admin_user",
401 resource_id=target_user_id,
402 outcome="success",
403 severity=AuditEventSeverity.CRITICAL,
404 source="admin",
405 metadata={
406 "impersonation_session_id": str(session_id),
407 "reason": str(reason) if reason else "",
408 "ip_address": str(ip) if ip else "",
409 "user_agent": str(user_agent) if user_agent else "",
410 },
411 )
412 )