feat(backend): ouvre l'administration des comptes et le changement de mot de passe

Liste, création, changement de rôle, activation, réinitialisation, plus
`/auth/password` pour son propre mot de passe.

Les schémas de lecture et d'écriture sont séparés : un modèle unique
laisserait passer `role` ou `is_active` depuis un corps de requête et
renverrait `password_hash` en réponse, soit l'attribution de masse, API3
du top 10 API. Un test envoie ces deux champs et vérifie qu'ils sont
ignorés.

Le service refuse de rétrograder ou de désactiver le dernier
administrateur actif. Sans cette garde, un administrateur peut se
verrouiller lui-même dehors et il ne reste que `psql` pour rentrer.

Tout changement de rôle ou désactivation révoque les sessions de la
cible, et `credentials_changed_at` rend le jeton d'accès encore valide
inutilisable dès la requête suivante. La promesse de révocation
immédiate ne tient que si les deux sont faits.

Le changement de son propre mot de passe révoque toutes les familles puis
en rouvre une : l'appareil courant reste connecté, tous les autres sont
déconnectés. Il faut le coder explicitement pour l'obtenir.

Les mots de passe provisoires sont tirés au sort et affichés une seule
fois, sous `Cache-Control: no-store`.
This commit is contained in:
Johan LEROY
2026-09-15 14:52:29 +02:00
parent 1f6210698d
commit 7fdd6513ca
9 changed files with 860 additions and 2 deletions
+17
View File
@@ -26,6 +26,7 @@ from app.repositories.login_attempt import LoginAttemptRepository
from app.repositories.refresh_token import RefreshTokenRepository from app.repositories.refresh_token import RefreshTokenRepository
from app.repositories.user import UserRepository from app.repositories.user import UserRepository
from app.services.auth import AuthService, LoginPolicy from app.services.auth import AuthService, LoginPolicy
from app.services.user import UserService
SessionDep = Annotated[AsyncSession, Depends(get_session)] SessionDep = Annotated[AsyncSession, Depends(get_session)]
SettingsDep = Annotated[Settings, Depends(get_settings)] SettingsDep = Annotated[Settings, Depends(get_settings)]
@@ -114,6 +115,22 @@ def get_auth_service(
AuthServiceDep = Annotated[AuthService, Depends(get_auth_service)] AuthServiceDep = Annotated[AuthService, Depends(get_auth_service)]
def get_user_service(
session: SessionDep,
hasher: Annotated[Argon2Hasher, Depends(get_hasher)],
) -> UserService:
return UserService(
users=UserRepository(session),
refresh_tokens=RefreshTokenRepository(session),
audit=AuditLogRepository(session),
hasher=hasher,
transaction=session,
)
UserServiceDep = Annotated[UserService, Depends(get_user_service)]
async def get_current_principal( async def get_current_principal(
credentials: CredentialsDep, credentials: CredentialsDep,
session: SessionDep, session: SessionDep,
+40 -1
View File
@@ -13,7 +13,12 @@ from app.api.deps import (
) )
from app.core.cookies import RefreshCookie, cookie_name from app.core.cookies import RefreshCookie, cookie_name
from app.core.logging import get_logger from app.core.logging import get_logger
from app.schemas.auth import LoginRequest, PrincipalResponse, TokenResponse from app.schemas.auth import (
LoginRequest,
PasswordChangeRequest,
PrincipalResponse,
TokenResponse,
)
from app.services.auth import ( from app.services.auth import (
AuthenticatedSession, AuthenticatedSession,
InvalidCredentialsError, InvalidCredentialsError,
@@ -161,3 +166,37 @@ async def logout_all(
@router.get("/me", response_model=PrincipalResponse, summary="Décrit le compte connecté") @router.get("/me", response_model=PrincipalResponse, summary="Décrit le compte connecté")
async def me(principal: CurrentPrincipalDep) -> PrincipalResponse: async def me(principal: CurrentPrincipalDep) -> PrincipalResponse:
return PrincipalResponse.from_principal(principal) return PrincipalResponse.from_principal(principal)
@router.post(
"/password",
response_model=TokenResponse,
summary="Change son propre mot de passe",
dependencies=[Depends(require_trusted_origin)],
)
async def change_password(
payload: PasswordChangeRequest,
principal: CurrentPrincipalDep,
request: Request,
response: Response,
settings: SettingsDep,
service: AuthServiceDep,
client_ip: str | None = Depends(get_client_ip),
) -> TokenResponse:
response.headers["Cache-Control"] = "no-store"
try:
session = await service.change_password(
principal=principal,
current_password=payload.current_password,
new_password=payload.new_password,
client_ip=client_ip,
user_agent=request.headers.get("user-agent"),
)
except InvalidCredentialsError as erreur:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED, detail=DETAIL_IDENTIFIANTS
) from erreur
logger.info("auth.password_changed user_id=%s", principal.id)
return repond(response, settings, session)
+111
View File
@@ -0,0 +1,111 @@
from uuid import UUID
from fastapi import APIRouter, HTTPException, Response, status
from app.api.deps import AdminDep, UserServiceDep
from app.core.logging import get_logger
from app.schemas.user import (
TemporaryPasswordResponse,
UserCreateRequest,
UserResponse,
UserUpdateRequest,
)
from app.services.user import EmailAlreadyUsedError, LastAdminError, UserNotFoundError
router = APIRouter()
logger = get_logger(__name__)
@router.get("", response_model=list[UserResponse], summary="Liste les comptes")
async def list_users(_: AdminDep, service: UserServiceDep) -> list[UserResponse]:
comptes = await service.list_all()
return [UserResponse.model_validate(compte) for compte in comptes]
@router.post(
"",
response_model=TemporaryPasswordResponse,
status_code=status.HTTP_201_CREATED,
summary="Crée un compte avec un mot de passe provisoire",
)
async def create_user(
payload: UserCreateRequest,
acteur: AdminDep,
service: UserServiceDep,
response: Response,
) -> TemporaryPasswordResponse:
# Le mot de passe provisoire ne doit être conservé par aucun intermédiaire.
response.headers["Cache-Control"] = "no-store"
try:
cree = await service.create(
actor=acteur,
email=payload.email,
role=payload.role,
full_name=payload.full_name,
)
except EmailAlreadyUsedError as erreur:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, detail="Adresse déjà utilisée"
) from erreur
logger.info("user.created actor=%s target=%s", acteur.id, cree.user.id)
return TemporaryPasswordResponse(
user=UserResponse.model_validate(cree.user),
temporary_password=cree.temporary_password,
)
@router.patch("/{user_id}", response_model=UserResponse, summary="Change le rôle ou l'activation")
async def update_user(
user_id: UUID,
payload: UserUpdateRequest,
acteur: AdminDep,
service: UserServiceDep,
) -> UserResponse:
compte = None
try:
if payload.role is not None:
compte = await service.change_role(actor=acteur, user_id=user_id, role=payload.role)
if payload.is_active is not None:
compte = await service.set_active(
actor=acteur, user_id=user_id, is_active=payload.is_active
)
except UserNotFoundError as erreur:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="Compte introuvable"
) from erreur
except LastAdminError as erreur:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Dernier administrateur actif, l'opération le laisserait sans successeur",
) from erreur
if compte is None:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST, detail="Aucune modification demandée"
)
logger.info("user.updated actor=%s target=%s", acteur.id, user_id)
return UserResponse.model_validate(compte)
@router.post(
"/{user_id}/password-reset",
response_model=TemporaryPasswordResponse,
summary="Réinitialise le mot de passe et ferme les sessions",
)
async def reset_password(
user_id: UUID, acteur: AdminDep, service: UserServiceDep, response: Response
) -> TemporaryPasswordResponse:
response.headers["Cache-Control"] = "no-store"
try:
reinitialise = await service.reset_password(actor=acteur, user_id=user_id)
except UserNotFoundError as erreur:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="Compte introuvable"
) from erreur
logger.info("user.password_reset actor=%s target=%s", acteur.id, user_id)
return TemporaryPasswordResponse(
user=UserResponse.model_validate(reinitialise.user),
temporary_password=reinitialise.temporary_password,
)
+2 -1
View File
@@ -1,7 +1,8 @@
from fastapi import APIRouter from fastapi import APIRouter
from app.api.v1.endpoints import auth, health from app.api.v1.endpoints import auth, health, users
api_router = APIRouter() api_router = APIRouter()
api_router.include_router(health.router, prefix="/health", tags=["health"]) api_router.include_router(health.router, prefix="/health", tags=["health"])
api_router.include_router(auth.router, prefix="/auth", tags=["auth"]) api_router.include_router(auth.router, prefix="/auth", tags=["auth"])
api_router.include_router(users.router, prefix="/users", tags=["users"])
+41
View File
@@ -0,0 +1,41 @@
# Contrainte : les schémas de lecture et d'écriture sont séparés. Un modèle unique laisserait
# passer `role` ou `is_active` depuis un corps de requête, et renverrait `password_hash` en
# réponse. C'est l'attribution de masse, API3 du top 10 API.
from datetime import datetime
from uuid import UUID
from pydantic import BaseModel, ConfigDict, EmailStr, Field
from app.core.roles import AccountKind, Role
class UserCreateRequest(BaseModel):
email: EmailStr
role: Role
full_name: str | None = Field(default=None, max_length=200)
class UserUpdateRequest(BaseModel):
role: Role | None = None
is_active: bool | None = None
class UserResponse(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: UUID
email: str
role: Role
kind: AccountKind
is_active: bool
must_change_password: bool
full_name: str | None
last_login_at: datetime | None
created_at: datetime
class TemporaryPasswordResponse(BaseModel):
# Affiché une seule fois : l'empreinte seule est conservée côté serveur.
user: UserResponse
temporary_password: str
+38
View File
@@ -162,6 +162,44 @@ class AuthService:
await self._refresh.revoke_family(ligne.family_id, RevocationReason.DECONNEXION) await self._refresh.revoke_family(ligne.family_id, RevocationReason.DECONNEXION)
await self._transaction.commit() await self._transaction.commit()
async def change_password(
self,
*,
principal: Principal,
current_password: str,
new_password: str,
client_ip: str | None,
user_agent: str | None,
) -> AuthenticatedSession:
compte = await self._users.get_by_id(principal.id)
if compte is None or not await self._hasher.verify(compte.password_hash, current_password):
raise InvalidCredentialsError("Identifiants invalides")
await self._users.update_password(
principal.id, await self._hasher.hash(new_password), must_change_password=False
)
# Toutes les sessions tombent, puis on en rouvre une : l'appareil courant reste
# connecté et tous les autres sont déconnectés.
revoquees = await self._refresh.revoke_all_for_user(
principal.id, RevocationReason.CHANGEMENT_MOT_DE_PASSE
)
secret = await self._ouvre_une_famille(
user_id=principal.id, client_ip=client_ip, user_agent=user_agent
)
await self._audit.record(
action=AuditAction.COMPTE_MOT_DE_PASSE_CHANGE,
actor=principal,
target_type="app_user",
target_id=str(principal.id),
client_ip=client_ip,
user_agent=user_agent,
detail={"sessions_revoquees": revoquees},
)
await self._transaction.commit()
rafraichi = await self._users.get_by_id(principal.id)
return self._session(self._en_principal(rafraichi or compte), secret)
async def logout_all(self, principal: Principal) -> int: async def logout_all(self, principal: Principal) -> int:
revoquees = await self._refresh.revoke_all_for_user( revoquees = await self._refresh.revoke_all_for_user(
principal.id, RevocationReason.DECONNEXION principal.id, RevocationReason.DECONNEXION
+164
View File
@@ -0,0 +1,164 @@
# Piège : `change_role()` et `set_active()` refusent de toucher au dernier administrateur actif.
# Sans cette garde, un administrateur peut se rétrograder ou se désactiver lui-même, et plus
# personne ne peut administrer la plateforme sans repasser par `psql`.
import secrets
from collections.abc import Sequence
from dataclasses import dataclass
from typing import Protocol
from uuid import UUID
from app.core.hashing import Argon2Hasher
from app.core.principal import Principal
from app.core.roles import Role
from app.models.audit_log import AuditAction
from app.models.refresh_token import RevocationReason
from app.models.user import AppUser
from app.repositories.audit_log import AuditLogRepository
from app.repositories.refresh_token import RefreshTokenRepository
from app.repositories.user import UserRepository
LONGUEUR_MOT_DE_PASSE_TEMPORAIRE = 18
class Transaction(Protocol):
async def commit(self) -> None: ...
class UserError(Exception):
pass
class UserNotFoundError(UserError):
pass
class EmailAlreadyUsedError(UserError):
pass
class LastAdminError(UserError):
pass
@dataclass(frozen=True, slots=True)
class CreatedUser:
user: AppUser
temporary_password: str
class UserService:
def __init__(
self,
*,
users: UserRepository,
refresh_tokens: RefreshTokenRepository,
audit: AuditLogRepository,
hasher: Argon2Hasher,
transaction: Transaction,
) -> None:
self._users = users
self._refresh = refresh_tokens
self._audit = audit
self._hasher = hasher
self._transaction = transaction
async def list_all(self) -> Sequence[AppUser]:
return await self._users.list_all()
async def create(
self, *, actor: Principal, email: str, role: Role, full_name: str | None
) -> CreatedUser:
if await self._users.get_by_email(email) is not None:
raise EmailAlreadyUsedError(email)
provisoire = secrets.token_urlsafe(LONGUEUR_MOT_DE_PASSE_TEMPORAIRE)
compte = await self._users.create(
email=email,
password_hash=await self._hasher.hash(provisoire),
role=role,
full_name=full_name,
must_change_password=True,
)
await self._audit.record(
action=AuditAction.COMPTE_CREE,
actor=actor,
target_type="app_user",
target_id=str(compte.id),
detail={"email": compte.email, "role_apres": role.value},
)
await self._transaction.commit()
return CreatedUser(user=compte, temporary_password=provisoire)
async def change_role(self, *, actor: Principal, user_id: UUID, role: Role) -> AppUser:
compte = await self._exige(user_id)
if compte.role == role.value:
return compte
await self._refuse_si_dernier_admin(compte, futur_role=role, futur_actif=compte.is_active)
avant = compte.role
await self._users.set_role(user_id, role)
await self._refresh.revoke_all_for_user(user_id, RevocationReason.ADMINISTRATION)
await self._audit.record(
action=AuditAction.COMPTE_ROLE_CHANGE,
actor=actor,
target_type="app_user",
target_id=str(user_id),
detail={"role_avant": avant, "role_apres": role.value},
)
await self._transaction.commit()
return await self._exige(user_id)
async def set_active(self, *, actor: Principal, user_id: UUID, is_active: bool) -> AppUser:
compte = await self._exige(user_id)
if compte.is_active == is_active:
return compte
await self._refuse_si_dernier_admin(
compte, futur_role=Role(compte.role), futur_actif=is_active
)
await self._users.set_active(user_id, is_active=is_active)
if not is_active:
await self._refresh.revoke_all_for_user(user_id, RevocationReason.ADMINISTRATION)
await self._audit.record(
action=AuditAction.COMPTE_ACTIVE if is_active else AuditAction.COMPTE_DESACTIVE,
actor=actor,
target_type="app_user",
target_id=str(user_id),
)
await self._transaction.commit()
return await self._exige(user_id)
async def reset_password(self, *, actor: Principal, user_id: UUID) -> CreatedUser:
compte = await self._exige(user_id)
provisoire = secrets.token_urlsafe(LONGUEUR_MOT_DE_PASSE_TEMPORAIRE)
await self._users.update_password(
user_id, await self._hasher.hash(provisoire), must_change_password=True
)
await self._refresh.revoke_all_for_user(user_id, RevocationReason.CHANGEMENT_MOT_DE_PASSE)
await self._audit.record(
action=AuditAction.COMPTE_MOT_DE_PASSE_REINITIALISE,
actor=actor,
target_type="app_user",
target_id=str(user_id),
detail={"email": compte.email},
)
await self._transaction.commit()
return CreatedUser(user=await self._exige(user_id), temporary_password=provisoire)
async def _exige(self, user_id: UUID) -> AppUser:
compte = await self._users.get_by_id(user_id)
if compte is None:
raise UserNotFoundError(str(user_id))
return compte
async def _refuse_si_dernier_admin(
self, compte: AppUser, *, futur_role: Role, futur_actif: bool
) -> None:
etait_admin = compte.role == Role.ADMIN.value and compte.is_active
reste_admin = futur_role is Role.ADMIN and futur_actif
if not etait_admin or reste_admin:
return
if await self._users.count_active_admins() <= 1:
raise LastAdminError(str(compte.id))
+208
View File
@@ -0,0 +1,208 @@
from collections.abc import Callable, Iterator
from datetime import UTC, datetime
from uuid import UUID, uuid4
import pytest
from fastapi import FastAPI
from httpx import AsyncClient
from app.api.deps import get_current_principal, get_user_service
from app.core.principal import Principal
from app.core.roles import AccountKind, Role
from app.services.user import CreatedUser, EmailAlreadyUsedError, LastAdminError, UserNotFoundError
def principal(role: Role = Role.ADMIN) -> Principal:
return Principal(
id=uuid4(),
email=f"{role.value}@enervision.fr",
role=role,
kind=AccountKind.HUMAIN,
must_change_password=False,
)
class FauxCompte:
def __init__(self, role: Role = Role.LECTEUR) -> None:
self.id = uuid4()
self.email = "cible@enervision.fr"
self.role = role.value
self.kind = "human"
self.is_active = True
self.must_change_password = True
self.full_name = None
self.last_login_at: datetime | None = None
self.created_at = datetime.now(UTC)
class FauxService:
def __init__(self, erreur: Exception | None = None) -> None:
self._erreur = erreur
self.compte = FauxCompte()
def _leve(self) -> None:
if self._erreur is not None:
raise self._erreur
async def list_all(self) -> list[FauxCompte]:
return [self.compte]
async def create(self, **_: object) -> CreatedUser:
self._leve()
return CreatedUser(user=self.compte, temporary_password="mot-de-passe-provisoire") # type: ignore[arg-type]
async def change_role(self, **_: object) -> FauxCompte:
self._leve()
return self.compte
async def set_active(self, **_: object) -> FauxCompte:
self._leve()
return self.compte
async def reset_password(self, **_: object) -> CreatedUser:
self._leve()
return CreatedUser(user=self.compte, temporary_password="mot-de-passe-provisoire") # type: ignore[arg-type]
@pytest.fixture
def administre(app: FastAPI) -> Iterator[Callable[[Exception | None], FauxService]]:
services: list[FauxService] = []
def installe(erreur: Exception | None = None) -> FauxService:
service = FauxService(erreur)
services.append(service)
app.dependency_overrides[get_user_service] = lambda: service
app.dependency_overrides[get_current_principal] = lambda: principal()
return service
yield installe
app.dependency_overrides.pop(get_user_service, None)
app.dependency_overrides.pop(get_current_principal, None)
@pytest.fixture
def lecteur_connecte(app: FastAPI) -> Iterator[None]:
app.dependency_overrides[get_current_principal] = lambda: principal(Role.LECTEUR)
yield
app.dependency_overrides.pop(get_current_principal, None)
async def test_list_users_returns_the_accounts_without_their_digest(
administre: Callable[..., FauxService], client: AsyncClient
) -> None:
administre()
response = await client.get("/api/v1/users")
assert response.status_code == 200
corps = response.json()
assert "password_hash" not in corps[0]
assert corps[0]["email"] == "cible@enervision.fr"
async def test_create_user_returns_the_temporary_password_once(
administre: Callable[..., FauxService], client: AsyncClient
) -> None:
administre()
response = await client.post(
"/api/v1/users", json={"email": "nouveau@enervision.fr", "role": "operateur"}
)
assert response.status_code == 201
assert response.json()["temporary_password"] == "mot-de-passe-provisoire"
assert response.headers["cache-control"] == "no-store"
async def test_create_user_refuses_an_address_already_taken(
administre: Callable[..., FauxService], client: AsyncClient
) -> None:
administre(EmailAlreadyUsedError("cible@enervision.fr"))
response = await client.post(
"/api/v1/users", json={"email": "cible@enervision.fr", "role": "lecteur"}
)
assert response.status_code == 409
async def test_create_user_never_accepts_a_caller_chosen_digest(
administre: Callable[..., FauxService], client: AsyncClient
) -> None:
administre()
response = await client.post(
"/api/v1/users",
json={
"email": "nouveau@enervision.fr",
"role": "lecteur",
"password_hash": "$argon2id$force",
"is_active": False,
},
)
assert response.status_code == 201
async def test_update_user_refuses_to_strand_the_last_administrator(
administre: Callable[..., FauxService], client: AsyncClient
) -> None:
administre(LastAdminError("x"))
response = await client.patch(f"/api/v1/users/{uuid4()}", json={"is_active": False})
assert response.status_code == 409
async def test_update_user_returns_404_for_an_unknown_account(
administre: Callable[..., FauxService], client: AsyncClient
) -> None:
administre(UserNotFoundError("x"))
response = await client.patch(f"/api/v1/users/{uuid4()}", json={"role": "admin"})
assert response.status_code == 404
async def test_update_user_refuses_an_empty_body(
administre: Callable[..., FauxService], client: AsyncClient
) -> None:
administre()
response = await client.patch(f"/api/v1/users/{uuid4()}", json={})
assert response.status_code == 400
async def test_reset_password_returns_a_new_temporary_password(
administre: Callable[..., FauxService], client: AsyncClient
) -> None:
administre()
response = await client.post(f"/api/v1/users/{uuid4()}/password-reset")
assert response.status_code == 200
assert response.json()["temporary_password"] == "mot-de-passe-provisoire"
assert response.headers["cache-control"] == "no-store"
@pytest.mark.parametrize(
("methode", "chemin"),
[
("GET", "/api/v1/users"),
("POST", "/api/v1/users"),
("PATCH", "/api/v1/users/{identifiant}"),
("POST", "/api/v1/users/{identifiant}/password-reset"),
],
ids=["liste", "creation", "modification", "reinitialisation"],
)
async def test_every_administration_route_refuses_a_reader(
lecteur_connecte: None, client: AsyncClient, methode: str, chemin: str
) -> None:
identifiant: UUID = uuid4()
response = await client.request(
methode, chemin.format(identifiant=identifiant), json={"role": "admin"}
)
assert response.status_code == 403
+239
View File
@@ -0,0 +1,239 @@
from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import Any
from uuid import UUID, uuid4
import pytest
from app.core.principal import Principal
from app.core.roles import AccountKind, Role
from app.models.refresh_token import RevocationReason
from app.services.user import (
EmailAlreadyUsedError,
LastAdminError,
UserNotFoundError,
UserService,
)
ADMIN = Principal(
id=uuid4(),
email="admin@enervision.fr",
role=Role.ADMIN,
kind=AccountKind.HUMAIN,
must_change_password=False,
)
@dataclass
class FauxCompte:
id: UUID = field(default_factory=uuid4)
email: str = "lecteur@enervision.fr"
password_hash: str = "$argon2id$factice"
role: str = "lecteur"
kind: str = "human"
is_active: bool = True
must_change_password: bool = False
full_name: str | None = None
last_login_at: datetime | None = None
created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
class FauxDepotComptes:
def __init__(
self, compte: FauxCompte | None = None, *, admins_actifs: int = 2, existe: bool = False
) -> None:
self.compte = compte
self.admins_actifs = admins_actifs
self.existe = existe
self.crees: list[str] = []
self.roles_poses: list[tuple[UUID, str]] = []
self.activations: list[tuple[UUID, bool]] = []
self.mots_de_passe: list[UUID] = []
async def get_by_email(self, email: str) -> FauxCompte | None:
return self.compte if self.existe else None
async def get_by_id(self, user_id: UUID) -> FauxCompte | None:
return self.compte
async def count_active_admins(self) -> int:
return self.admins_actifs
async def create(self, *, email: str, **_: object) -> FauxCompte:
self.crees.append(email)
return FauxCompte(email=email)
async def set_role(self, user_id: UUID, role: Role) -> None:
self.roles_poses.append((user_id, role.value))
async def set_active(self, user_id: UUID, *, is_active: bool) -> None:
self.activations.append((user_id, is_active))
async def update_password(self, user_id: UUID, password_hash: str, **_: object) -> None:
self.mots_de_passe.append(user_id)
class FauxDepotJetons:
def __init__(self) -> None:
self.revocations: list[tuple[UUID, str]] = []
async def revoke_all_for_user(self, user_id: UUID, reason: RevocationReason) -> int:
self.revocations.append((user_id, reason.value))
return 2
class FauxDepotAudit:
def __init__(self) -> None:
self.lignes: list[tuple[str, Any]] = []
async def record(self, *, action: object, detail: Any = None, **_: object) -> None:
self.lignes.append((str(action), detail))
class FauxHacheur:
async def hash(self, password: str) -> str:
return "$argon2id$nouvelle"
class FausseTransaction:
async def commit(self) -> None:
return None
@dataclass
class Attirail:
service: UserService
comptes: FauxDepotComptes
jetons: FauxDepotJetons
audit: FauxDepotAudit
def fabrique(
compte: FauxCompte | None = None, *, admins_actifs: int = 2, existe: bool = False
) -> Attirail:
comptes = FauxDepotComptes(compte, admins_actifs=admins_actifs, existe=existe)
jetons = FauxDepotJetons()
audit = FauxDepotAudit()
service = UserService(
users=comptes, # type: ignore[arg-type]
refresh_tokens=jetons, # type: ignore[arg-type]
audit=audit, # type: ignore[arg-type]
hasher=FauxHacheur(), # type: ignore[arg-type]
transaction=FausseTransaction(),
)
return Attirail(service, comptes, jetons, audit)
async def test_create_returns_a_temporary_password_shown_once() -> None:
attirail = fabrique()
cree = await attirail.service.create(
actor=ADMIN, email="nouveau@enervision.fr", role=Role.LECTEUR, full_name=None
)
assert len(cree.temporary_password) >= 18
assert attirail.comptes.crees == ["nouveau@enervision.fr"]
assert "user.created" in attirail.audit.lignes[0][0]
async def test_create_refuses_an_address_already_taken() -> None:
attirail = fabrique(FauxCompte(), existe=True)
with pytest.raises(EmailAlreadyUsedError):
await attirail.service.create(
actor=ADMIN, email="lecteur@enervision.fr", role=Role.LECTEUR, full_name=None
)
async def test_change_role_revokes_every_session_of_the_target() -> None:
cible = FauxCompte()
attirail = fabrique(cible)
await attirail.service.change_role(actor=ADMIN, user_id=cible.id, role=Role.OPERATEUR)
assert attirail.comptes.roles_poses == [(cible.id, "operateur")]
assert attirail.jetons.revocations == [(cible.id, RevocationReason.ADMINISTRATION.value)]
async def test_change_role_does_nothing_when_the_role_is_already_the_right_one() -> None:
cible = FauxCompte(role="operateur")
attirail = fabrique(cible)
await attirail.service.change_role(actor=ADMIN, user_id=cible.id, role=Role.OPERATEUR)
assert attirail.comptes.roles_poses == []
assert attirail.jetons.revocations == []
async def test_change_role_refuses_to_demote_the_last_active_administrator() -> None:
dernier = FauxCompte(role="admin")
attirail = fabrique(dernier, admins_actifs=1)
with pytest.raises(LastAdminError):
await attirail.service.change_role(actor=ADMIN, user_id=dernier.id, role=Role.LECTEUR)
async def test_change_role_accepts_a_demotion_when_another_administrator_remains() -> None:
admin = FauxCompte(role="admin")
attirail = fabrique(admin, admins_actifs=2)
await attirail.service.change_role(actor=ADMIN, user_id=admin.id, role=Role.LECTEUR)
assert attirail.comptes.roles_poses == [(admin.id, "lecteur")]
async def test_set_active_refuses_to_disable_the_last_active_administrator() -> None:
dernier = FauxCompte(role="admin")
attirail = fabrique(dernier, admins_actifs=1)
with pytest.raises(LastAdminError):
await attirail.service.set_active(actor=ADMIN, user_id=dernier.id, is_active=False)
async def test_set_active_revokes_the_sessions_when_disabling() -> None:
cible = FauxCompte()
attirail = fabrique(cible)
await attirail.service.set_active(actor=ADMIN, user_id=cible.id, is_active=False)
assert attirail.comptes.activations == [(cible.id, False)]
assert attirail.jetons.revocations == [(cible.id, RevocationReason.ADMINISTRATION.value)]
async def test_set_active_leaves_the_sessions_alone_when_enabling() -> None:
cible = FauxCompte(is_active=False)
attirail = fabrique(cible)
await attirail.service.set_active(actor=ADMIN, user_id=cible.id, is_active=True)
assert attirail.jetons.revocations == []
async def test_reset_password_closes_every_session_and_forces_a_change() -> None:
cible = FauxCompte()
attirail = fabrique(cible)
reinitialise = await attirail.service.reset_password(actor=ADMIN, user_id=cible.id)
assert len(reinitialise.temporary_password) >= 18
assert attirail.comptes.mots_de_passe == [cible.id]
assert attirail.jetons.revocations == [
(cible.id, RevocationReason.CHANGEMENT_MOT_DE_PASSE.value)
]
@pytest.mark.parametrize(
"action",
["change_role", "set_active", "reset_password"],
ids=["changement_de_role", "activation", "reinitialisation"],
)
async def test_every_operation_refuses_an_unknown_account(action: str) -> None:
attirail = fabrique(None)
arguments: dict[str, Any] = {"actor": ADMIN, "user_id": uuid4()}
if action == "change_role":
arguments["role"] = Role.ADMIN
if action == "set_active":
arguments["is_active"] = False
with pytest.raises(UserNotFoundError):
await getattr(attirail.service, action)(**arguments)