Coverage for src/lexigram/auth/authn/user_service.py: 56%
135 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-26 00:58 +0800
1"""User management services for CRUD operations."""
3from __future__ import annotations
5import asyncio
6from typing import TYPE_CHECKING, Any, cast
8from lexigram.auth.authn.security import PasswordHasher, PasswordPolicy
9from lexigram.auth.events import PasswordChanged
10from lexigram.auth.exceptions import (
11 AuthorizationError,
12 EmailExistsError,
13 InvalidCredentialsError,
14 PasswordPolicyError,
15 UserNotFoundError,
16)
17from lexigram.auth.models.user import User, UserCredentials
18from lexigram.contracts.exceptions import ValidationError
19from lexigram.di.decorators import inject
20from lexigram.logging import get_logger
21from lexigram.result import Err, Ok, Result
23if TYPE_CHECKING:
24 from lexigram.contracts.events.protocols import EventBusProtocol
26logger = get_logger(__name__)
29@inject
30class UserService:
31 """Service for user management operations.
33 Handles user CRUD, password management, and administrative operations.
34 """
36 def __init__(
37 self,
38 password_policy: PasswordPolicy,
39 user_store: Any,
40 event_bus: EventBusProtocol | None = None,
41 ) -> None:
42 self.password_policy = password_policy
43 self.user_store = user_store
44 self._event_bus = event_bus
45 # Background tasks kept alive to prevent GC before completion
46 self._background_tasks: set[asyncio.Task[object]] = set()
48 def _emit(self, event: object) -> None:
49 """Fire-and-forget event publication.
51 Schedules a background task to publish *event* via the event bus (if
52 one is configured). The task reference is stored in
53 ``_background_tasks`` to prevent premature garbage collection; it
54 removes itself from the set upon completion.
55 """
56 if self._event_bus is None:
57 return
58 task: asyncio.Task[object] = asyncio.create_task(self._event_bus.publish(event))
59 self._background_tasks.add(task)
60 task.add_done_callback(self._background_tasks.discard)
62 async def create_user(
63 self,
64 name: str,
65 email: str,
66 password: str,
67 roles: list[str] | None = None,
68 ) -> Result[User, EmailExistsError | PasswordPolicyError]:
69 """Create a new user.
71 Returns:
72 ``Ok(User)`` on success.
73 ``Err(PasswordPolicyError)`` if the password does not satisfy policy.
74 ``Err(EmailExistsError)`` if the email address is already registered.
76 Infrastructure exceptions (e.g. DB connectivity) propagate unchanged.
77 """
78 try:
79 self.password_policy.validate(password)
80 except ValueError as e:
81 return Err(PasswordPolicyError(str(e)))
83 hashed_password = await PasswordHasher().hash(password)
84 try:
85 user = await self.user_store.create_user(
86 name,
87 email,
88 hashed_password,
89 roles,
90 )
91 except EmailExistsError as e:
92 return Err(e)
93 except ValueError as e:
94 # In-memory store raises ValueError for duplicate email.
95 return Err(EmailExistsError(str(e)))
96 return Ok(user)
98 async def get_user(self, user_id: str) -> User | None:
99 """Get user by ID."""
100 return cast("User | None", await self.user_store.get_user_by_id(user_id))
102 async def update_user(
103 self, user: User
104 ) -> Result[User, UserNotFoundError | ValidationError]:
105 """Update user information.
107 Returns:
108 ``Ok(User)`` on success.
109 ``Err(UserNotFoundError)`` if the user does not exist.
110 ``Err(ValidationError)`` if the supplied user data is invalid
111 (e.g. empty email).
113 Infrastructure exceptions (e.g. DB connectivity) propagate unchanged.
114 """
115 if not user.email:
116 return Err(ValidationError("User email must not be empty"))
117 existing = await self.user_store.get_user_by_id(user.user_id)
118 if not existing:
119 return Err(UserNotFoundError(user.user_id))
120 await self.user_store.update_user(user)
121 return Ok(user)
123 async def delete_user(self, user_id: str) -> Result[None, UserNotFoundError]:
124 """Delete a user.
126 Returns:
127 ``Ok(None)`` on success.
128 ``Err(UserNotFoundError)`` if the user does not exist.
130 Raises:
131 AuthorizationError: If attempting to delete the protected ``admin``
132 account (security boundary — not a recoverable Result path).
134 Infrastructure exceptions (e.g. DB connectivity) propagate unchanged.
135 """
136 user = await self.get_user(user_id)
137 if not user:
138 return Err(UserNotFoundError(user_id))
139 if user.name == "admin":
140 raise AuthorizationError(
141 "The system administrator account ('admin') cannot be deleted.",
142 )
143 await self.user_store.delete_user(user_id)
144 return Ok(None)
146 async def lock_user(self, user_id: str) -> Result[None, UserNotFoundError]:
147 """Deactivate (lock) a user account.
149 Sets ``is_active = False`` so the user cannot log in until unlocked.
151 Returns:
152 ``Ok(None)`` on success.
153 ``Err(UserNotFoundError)`` if the user does not exist.
155 Infrastructure exceptions propagate unchanged.
156 """
157 import dataclasses
159 user = await self.get_user(user_id)
160 if not user:
161 return Err(UserNotFoundError(user_id))
162 locked = dataclasses.replace(user, is_active=False)
163 await self.user_store.update_user(locked)
164 return Ok(None)
166 async def unlock_user(self, user_id: str) -> Result[None, UserNotFoundError]:
167 """Reactivate (unlock) a previously locked user account.
169 Sets ``is_active = True`` so the user may log in again.
171 Returns:
172 ``Ok(None)`` on success.
173 ``Err(UserNotFoundError)`` if the user does not exist.
175 Infrastructure exceptions propagate unchanged.
176 """
177 import dataclasses
179 user = await self.get_user(user_id)
180 if not user:
181 return Err(UserNotFoundError(user_id))
182 unlocked = dataclasses.replace(user, is_active=True)
183 await self.user_store.update_user(unlocked)
184 return Ok(None)
186 async def change_user_password(
187 self,
188 user_id: str,
189 current_password: str,
190 new_password: str,
191 ) -> Result[None, InvalidCredentialsError | PasswordPolicyError]:
192 """Change a user's password (requires current password).
194 Returns:
195 ``Ok(None)`` on success.
196 ``Err(InvalidCredentialsError)`` if the current password is wrong or
197 the user does not exist.
198 ``Err(PasswordPolicyError)`` if the new password violates policy or
199 has been used recently.
201 Infrastructure exceptions (e.g. DB connectivity) propagate unchanged.
202 """
203 user = await self.get_user(user_id)
204 if not user:
205 # Return the same error as wrong password to avoid leaking user existence.
206 return Err(InvalidCredentialsError("Current password is incorrect"))
208 creds = await self.user_store.get_credentials(user_id)
209 if (
210 not creds
211 or not creds.hashed_password
212 or not await PasswordHasher().verify(
213 current_password,
214 creds.hashed_password,
215 )
216 ):
217 return Err(InvalidCredentialsError("Current password is incorrect"))
219 try:
220 self.password_policy.validate(new_password)
221 except ValueError as e:
222 return Err(PasswordPolicyError(str(e)))
224 if getattr(self.password_policy, "prevent_reuse", False):
225 history_size = getattr(self.password_policy, "history_size", 5)
226 checks = [h for h in [creds.hashed_password, *creds.previous_hashes] if h]
227 for old_hash in checks[: history_size + 1]:
228 if await PasswordHasher().verify(new_password, old_hash):
229 return Err(
230 PasswordPolicyError(
231 "New password must not match recent passwords",
232 )
233 )
235 new_hash = await PasswordHasher().hash(new_password)
236 new_prev = (
237 [creds.hashed_password, *creds.previous_hashes]
238 if creds.hashed_password
239 else list(creds.previous_hashes)
240 )
241 history_size = getattr(self.password_policy, "history_size", 5)
242 new_prev = list(filter(lambda h: h, new_prev))[:history_size]
244 updated_creds = UserCredentials(
245 user_id=user_id,
246 hashed_password=new_hash,
247 previous_hashes=new_prev,
248 )
249 await self.user_store.update_credentials(updated_creds)
250 self._emit(PasswordChanged(user_id=user_id))
251 return Ok(None)
253 async def set_user_password(
254 self,
255 user_id: str,
256 new_password: str,
257 force: bool = False,
258 ) -> None:
259 """Set a user's password (admin operation)."""
260 user = await self.get_user(user_id)
261 if not user:
262 raise ValueError("User not found")
264 try:
265 self.password_policy.validate(new_password)
266 except ValueError as e:
267 raise ValueError(f"Password policy violation: {e}") from e
269 creds = await self.user_store.get_credentials(user_id)
271 if not force and getattr(self.password_policy, "prevent_reuse", False):
272 if creds:
273 history_size = getattr(self.password_policy, "history_size", 5)
274 checks = [
275 h for h in [creds.hashed_password, *creds.previous_hashes] if h
276 ]
277 for old_hash in checks[: history_size + 1]:
278 if await PasswordHasher().verify(new_password, old_hash):
279 raise PasswordPolicyError(
280 "New password must not match recent passwords",
281 )
283 new_hash = await PasswordHasher().hash(new_password)
284 if creds:
285 new_prev = (
286 [creds.hashed_password, *creds.previous_hashes]
287 if creds.hashed_password
288 else list(creds.previous_hashes)
289 )
290 history_size = getattr(self.password_policy, "history_size", 5)
291 new_prev = list(filter(lambda h: h, new_prev))[:history_size]
292 else:
293 new_prev = []
295 updated_creds = UserCredentials(
296 user_id=user_id,
297 hashed_password=new_hash,
298 previous_hashes=new_prev,
299 )
300 await self.user_store.update_credentials(updated_creds)
302 async def list_users(self, skip: int = 0, limit: int = 100) -> list[User]:
303 """List users with pagination."""
304 return cast("list[User]", await self.user_store.list_users(skip, limit))
306 async def count_users(self) -> int:
307 """Count total users."""
308 return cast("int", await self.user_store.count_users())
310 def __repr__(self) -> str:
311 """Return a string representation of this service."""
312 return f"UserService(user_store={type(self.user_store).__name__})"
314 async def shutdown(self) -> None:
315 """Cancel and await all pending background event tasks."""
316 tasks = list(self._background_tasks)
317 for task in tasks:
318 task.cancel()
319 if tasks:
320 await asyncio.gather(*tasks, return_exceptions=True)
321 self._background_tasks.clear()
324__all__ = ["UserService"]