Coverage for src / lexigram / admin / services / impersonation.py: 0%
84 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-11 02:25 +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)
27"""
29from __future__ import annotations
31from dataclasses import dataclass, field
32from datetime import UTC, datetime
33from typing import Any
34import uuid
36from lexigram.admin.exceptions import PermissionDeniedError
37from lexigram.contracts.audit import AuditEntry, AuditEventSeverity, AuditLoggerProtocol
38from lexigram.di.decorators import inject
39from lexigram.logging import get_logger
40from lexigram.result import Err, Ok, Result
42logger = get_logger(__name__)
44# Session-state key used to store impersonation context
45_SESSION_KEY = "_admin_impersonation"
46# Attribute on request.state that holds original admin identity
47_ORIGINAL_USER_KEY = "_impersonation_original_user_id"
50@dataclass(frozen=True)
51class ImpersonationSession:
52 """Represents an active impersonation session.
54 Attributes:
55 id: Unique impersonation session identifier.
56 actor_id: Admin user who initiated the impersonation.
57 target_user_id: User being impersonated.
58 started_at: UTC timestamp when impersonation began.
59 reason: Optional free-text reason for audit trail.
60 """
62 id: str = field(default_factory=lambda: str(uuid.uuid4()))
63 actor_id: str = ""
64 target_user_id: str = ""
65 started_at: datetime = field(default_factory=lambda: datetime.now(UTC))
66 reason: str = ""
69class ImpersonationPolicy:
70 """Default policy: only users with the ``superadmin`` role may impersonate.
72 Override ``can_impersonate`` to customise authorisation logic.
73 """
75 def can_impersonate(self, actor: Any) -> bool:
76 """Return True if *actor* is allowed to impersonate another user.
78 Args:
79 actor: The admin user attempting to impersonate.
81 Returns:
82 True if the actor has the ``superadmin`` role.
83 """
84 roles = getattr(actor, "roles", [])
85 if isinstance(roles, list):
86 return "superadmin" in roles
87 return False
90@inject
91class ImpersonationService:
92 """Service that manages user impersonation sessions.
94 Args:
95 audit_logger: Optional audit logger (AuditLoggerProtocol).
96 If ``None``, impersonation events are only logged via the standard
97 logger; no structured audit record is persisted.
98 policy: Policy object that determines who may impersonate.
99 Defaults to ``ImpersonationPolicy`` (superadmin-only).
100 active_sessions: Optional pre-populated in-memory store used in tests.
101 """
103 def __init__(
104 self,
105 audit_logger: AuditLoggerProtocol | None = None,
106 policy: ImpersonationPolicy | None = None,
107 active_sessions: dict[str, ImpersonationSession] | None = None,
108 ) -> None:
109 self._audit = audit_logger
110 self._policy = policy or ImpersonationPolicy()
111 # actor_id → active ImpersonationSession
112 self._sessions: dict[str, ImpersonationSession] = (
113 active_sessions if active_sessions is not None else {}
114 )
116 # ------------------------------------------------------------------
117 # Public API
118 # ------------------------------------------------------------------
120 async def start(
121 self,
122 actor: Any,
123 target_user_id: str,
124 reason: str = "",
125 request: Any | None = None,
126 ) -> Result[ImpersonationSession, PermissionDeniedError]:
127 """Begin impersonating *target_user_id* on behalf of *actor*.
129 Args:
130 actor: The authenticated admin user requesting impersonation.
131 target_user_id: ID of the user to impersonate.
132 reason: Free-text reason recorded in the audit trail.
133 request: Optional Starlette request — if provided, the
134 impersonation token is stored in ``request.session``.
136 Returns:
137 ``Ok(ImpersonationSession)`` on success, or
138 ``Err(PermissionDeniedError)`` if the actor is not authorised.
139 """
140 actor_id: str = getattr(actor, "id", str(actor))
142 if not self._policy.can_impersonate(actor):
143 logger.warning(
144 "impersonation.denied",
145 actor_id=actor_id,
146 target_user_id=target_user_id,
147 )
148 return Err(
149 PermissionDeniedError(
150 f"User {actor_id!r} is not authorised to impersonate other users"
151 )
152 )
154 session = ImpersonationSession(
155 actor_id=actor_id,
156 target_user_id=target_user_id,
157 reason=reason,
158 )
159 self._sessions[actor_id] = session
161 if request is not None:
162 req_session = getattr(request, "session", None)
163 if req_session is not None:
164 req_session[_SESSION_KEY] = {
165 "id": session.id,
166 "actor_id": actor_id,
167 "target_user_id": target_user_id,
168 "started_at": session.started_at.isoformat(),
169 "reason": reason,
170 }
171 req_session[_ORIGINAL_USER_KEY] = actor_id
173 await self._emit_audit(
174 action="impersonation.start",
175 actor_id=actor_id,
176 target_user_id=target_user_id,
177 session_id=session.id,
178 reason=reason,
179 request=request,
180 )
181 logger.info(
182 "impersonation.started",
183 actor_id=actor_id,
184 target_user_id=target_user_id,
185 session_id=session.id,
186 )
187 return Ok(session)
189 async def stop(
190 self,
191 actor: Any,
192 request: Any | None = None,
193 ) -> Result[str, str]:
194 """End the active impersonation session for *actor*.
196 Args:
197 actor: The authenticated admin user (original identity).
198 request: Optional Starlette request — session cookie is cleared
199 when provided.
201 Returns:
202 ``Ok(original_user_id)`` or ``Err("no_active_session")``.
203 """
204 actor_id: str = getattr(actor, "id", str(actor))
206 session = self._sessions.pop(actor_id, None)
207 if session is None and request is not None:
208 req_session = getattr(request, "session", None)
209 if req_session:
210 raw = req_session.get(_SESSION_KEY)
211 if isinstance(raw, dict):
212 actor_id = raw.get("actor_id", actor_id)
213 session = ImpersonationSession(
214 id=raw.get("id", ""),
215 actor_id=actor_id,
216 target_user_id=raw.get("target_user_id", ""),
217 reason=raw.get("reason", ""),
218 )
220 if session is None:
221 return Err("no_active_session")
223 if request is not None:
224 req_session = getattr(request, "session", None)
225 if req_session is not None:
226 req_session.pop(_SESSION_KEY, None)
227 req_session.pop(_ORIGINAL_USER_KEY, None)
229 await self._emit_audit(
230 action="impersonation.stop",
231 actor_id=session.actor_id,
232 target_user_id=session.target_user_id,
233 session_id=session.id,
234 reason="",
235 request=request,
236 )
237 logger.info(
238 "impersonation.stopped",
239 actor_id=session.actor_id,
240 target_user_id=session.target_user_id,
241 session_id=session.id,
242 )
243 return Ok(session.actor_id)
245 def get_active_session(self, actor_id: str) -> ImpersonationSession | None:
246 """Return the active ``ImpersonationSession`` for *actor_id*, or ``None``.
248 Args:
249 actor_id: Admin user ID to look up.
251 Returns:
252 Active session, or ``None`` if the user is not impersonating.
253 """
254 return self._sessions.get(actor_id)
256 def is_impersonating(self, actor_id: str) -> bool:
257 """Return ``True`` if *actor_id* currently has an active impersonation.
259 Args:
260 actor_id: Admin user ID to check.
262 Returns:
263 True if an active session exists for this actor.
264 """
265 return actor_id in self._sessions
267 def list_active(self) -> list[ImpersonationSession]:
268 """Return all currently active impersonation sessions.
270 Returns:
271 List of active ``ImpersonationSession`` objects.
272 """
273 return list(self._sessions.values())
275 # ------------------------------------------------------------------
276 # Internal helpers
277 # ------------------------------------------------------------------
279 async def _emit_audit(
280 self,
281 action: str,
282 actor_id: str,
283 target_user_id: str,
284 session_id: str,
285 reason: str,
286 request: Any | None,
287 ) -> None:
288 """Emit an audit log entry if an audit logger is configured."""
289 if self._audit is None:
290 return
292 ip = None
293 user_agent = None
294 if request is not None:
295 client = getattr(request, "client", None)
296 ip = getattr(client, "host", None) if client else None
297 headers = getattr(request, "headers", {})
298 user_agent = headers.get("user-agent")
300 await self._audit.log(
301 AuditEntry(
302 action=action,
303 actor_id=actor_id,
304 resource_type="admin_user",
305 resource_id=target_user_id,
306 outcome="success",
307 severity=AuditEventSeverity.CRITICAL,
308 source="admin",
309 metadata={
310 "impersonation_session_id": str(session_id),
311 "reason": str(reason) if reason else "",
312 "ip_address": str(ip) if ip else "",
313 "user_agent": str(user_agent) if user_agent else "",
314 },
315 )
316 )