Python 64.6%
TypeScript 33.7%
CSS 0.8%
1"""Auth: e-mail + password, activation / reset links by e-mail (Resend); JWT cookie + bearer; /me.23Flow: the professor registers the students' addresses → each receives a "choose your password"4link → the password is then used on every sign-in. "Première connexion ou mot de passe oublié"5re-sends such a link to any registered address.6"""78from __future__ import annotations910from datetime import timedelta1112from fastapi import APIRouter, Depends, HTTPException, Request, Response, status13from pydantic import BaseModel, EmailStr1415from app.core.config import Settings, get_settings16from app.core.ratelimit import limiter17from app.core.security import (18 COOKIE_NAME,19 AuthUser,20 create_token,21 decode_token,22 email_allowed,23 get_current_user,24 verify_password,25)26from app.services import invites, users2728router = APIRouter(tags=["auth"])2930MIN_PASSWORD = 8313233class VerifyReq(BaseModel):34 token: str353637class PasswordLoginReq(BaseModel):38 email: EmailStr39 password: str404142class ForgotReq(BaseModel):43 email: EmailStr444546class SetPasswordReq(BaseModel):47 token: str48 password: str495051class ChangePasswordReq(BaseModel):52 current_password: str | None = None53 new_password: str545556class PrefsReq(BaseModel):57 tutoiement: bool | None = None58 course: str | None = None59 deep: bool | None = None60 locale: str | None = None61 font_scale: float | None = None62 display_name: str | None = None636465def _set_cookie(resp: Response, token: str, settings: Settings, refresh: str | None = None) -> None:66 secure = not settings.is_dev67 resp.set_cookie(COOKIE_NAME, token, httponly=True, secure=secure, samesite="lax",68 max_age=settings.JWT_TTL_HOURS * 3600, path="/")69 if refresh:70 resp.set_cookie(COOKIE_NAME + "_refresh", refresh, httponly=True, secure=secure,71 samesite="lax", max_age=settings.REFRESH_TTL_DAYS * 86400,72 path="/api/v1/auth")737475def _issue(user_id: str, role: str, settings: Settings, resp: Response) -> dict:76 token = create_token(user_id, role, settings)77 refresh = create_token(user_id, role, settings, timedelta(days=settings.REFRESH_TTL_DAYS),78 kind="refresh")79 _set_cookie(resp, token, settings, refresh)80 return {"token": token, "role": role}818283def _check_password_strength(password: str) -> None:84 if len(password) < MIN_PASSWORD:85 raise HTTPException(status.HTTP_400_BAD_REQUEST,86 detail=f"Mot de passe trop court ({MIN_PASSWORD} caractères minimum).")87 if len(password) > 128:88 raise HTTPException(status.HTTP_400_BAD_REQUEST, detail="Mot de passe trop long (128 maximum).")899091# ------------------------------------------------------------------ sign-in92@router.post("/auth/login")93async def password_login(req: PasswordLoginReq, request: Request, response: Response,94 settings: Settings = Depends(get_settings)) -> dict:95 ip = request.client.host if request.client else "?"96 limiter.check(f"pwd:{ip}", 30, 3600, "Trop de tentatives. Réessaie plus tard.")97 limiter.check(f"pwd:{req.email.lower()}", 10, 900, "Trop de tentatives pour ce compte. Patiente 15 minutes.")98 user = await users.authenticate(req.email, req.password)99 if not user:100 existing = await users.get_user_by_email(req.email)101 if existing and not existing.password_hash:102 raise HTTPException(status.HTTP_401_UNAUTHORIZED,103 detail="Ce compte n'a pas encore de mot de passe. Clique sur "104 "« Première connexion ou mot de passe oublié » pour en choisir un.")105 raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail="Courriel ou mot de passe invalide.")106 out = _issue(user.id, user.role, settings, response)107 return {"mode": "password", **out, "user": _user_dict(user)}108109110@router.post("/auth/forgot")111async def forgot_password(req: ForgotReq, request: Request,112 settings: Settings = Depends(get_settings)) -> dict:113 """First login or forgotten password: e-mail a single-use 'choose your password' link."""114 ip = request.client.host if request.client else "?"115 limiter.check(f"forgot:{ip}", 20, 3600, "Trop de demandes. Réessaie plus tard.")116 email = req.email.lower().strip()117 limiter.check(f"forgot:{email}", 5, 3600, "Trop de demandes pour cette adresse. Réessaie dans une heure.")118 user = await users.get_user_by_email(email)119 if user is None:120 # Addresses listed in the environment (professor/admin/invited) may self-register.121 if email in settings.invited_emails | settings.professor_emails | settings.admin_emails:122 user = await users.get_or_create_user(email, settings)123 else:124 hint = ("Cette adresse n'est pas inscrite au cours. Vérifie l'orthographe ou demande au "125 "professeur de t'ajouter à la liste des étudiants.")126 if not email_allowed(email, settings):127 hint = "Adresse non admise. Utilise ton courriel @uqo.ca inscrit par le professeur."128 raise HTTPException(status.HTTP_404_NOT_FOUND, detail=hint)129 sent, link = await invites.send_password_link(user, settings)130 out: dict = {"sent": sent, "first_time": not user.password_hash,131 "ttl_minutes": settings.RESET_TTL_MINUTES}132 if not sent:133 if settings.is_dev:134 out["dev_link"] = link # local development only135 else:136 out["hint"] = "Envoi de courriel indisponible pour le moment. Préviens le professeur."137 return out138139140@router.get("/auth/password-token")141async def password_token_info(token: str) -> dict:142 """Validate an activation/reset token without consuming it (drives the /mot-de-passe page)."""143 link = await users.peek_magic_link(token)144 if not link or (link.purpose or "login") not in {"invite", "reset"}:145 raise HTTPException(status.HTTP_400_BAD_REQUEST, detail="Lien invalide ou expiré.")146 user = await users.get_user_by_email(link.email)147 return {"email": link.email, "purpose": link.purpose,148 "first_time": not (user and user.password_hash),149 "display_name": user.display_name if user else None,150 "expires_at": link.expires_at.isoformat()}151152153@router.post("/auth/set-password")154async def set_password_with_token(req: SetPasswordReq, response: Response,155 settings: Settings = Depends(get_settings)) -> dict:156 """Consume an invite/reset token, store the password and open the session."""157 _check_password_strength(req.password)158 email = await users.consume_magic_link(req.token, purposes={"invite", "reset"})159 if not email:160 raise HTTPException(status.HTTP_400_BAD_REQUEST,161 detail="Lien invalide ou expiré. Demande un nouveau lien depuis la page de connexion.")162 user = await users.get_user_by_email(email) or await users.get_or_create_user(email, settings)163 await users.set_password(user.id, req.password)164 await users.invalidate_links(email, purposes={"invite", "reset"})165 user = await users.get_user(user.id)166 assert user is not None167 out = _issue(user.id, user.role, settings, response)168 return {"mode": "password_set", **out, "user": _user_dict(user)}169170171@router.post("/auth/verify")172async def verify(req: VerifyReq, response: Response,173 settings: Settings = Depends(get_settings)) -> dict:174 """Legacy direct sign-in link (purpose 'login'), still honoured for links already sent."""175 email = await users.consume_magic_link(req.token, purposes={"login"})176 if not email:177 raise HTTPException(status.HTTP_400_BAD_REQUEST, detail="Lien invalide ou expiré.")178 user = await users.get_or_create_user(email, settings)179 out = _issue(user.id, user.role, settings, response)180 return {**out, "user": _user_dict(user)}181182183@router.post("/auth/refresh")184async def refresh(request: Request, response: Response,185 settings: Settings = Depends(get_settings)) -> dict:186 raw = request.cookies.get(COOKIE_NAME + "_refresh")187 payload = decode_token(raw, settings) if raw else None188 if not payload or payload.get("kind") != "refresh":189 raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail="Session expirée.")190 user = await users.get_user(str(payload["sub"]))191 if not user:192 raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail="Compte introuvable.")193 return _issue(user.id, user.role, settings, response)194195196@router.post("/auth/logout")197async def logout(response: Response) -> dict:198 response.delete_cookie(COOKIE_NAME, path="/")199 response.delete_cookie(COOKIE_NAME + "_refresh", path="/api/v1/auth")200 return {"ok": True}201202203@router.get("/auth/config")204async def auth_config(settings: Settings = Depends(get_settings)) -> dict:205 return {"password": True, "mail": settings.mail_enabled,206 "min_password": MIN_PASSWORD,207 "domains": sorted(settings.allowed_domains), "courses": settings.courses,208 "term": settings.TERM_LABEL}209210211def _user_dict(u) -> dict: # noqa: ANN001212 return {"id": u.id, "email": u.email, "role": u.role, "display_name": u.display_name,213 "preferences": u.preferences or {}, "has_password": bool(u.password_hash),214 "consent_at": u.consent_at.isoformat() if u.consent_at else None}215216217# ------------------------------------------------------------------ me218@router.get("/me")219async def me(auth: AuthUser = Depends(get_current_user)) -> dict:220 user = await users.get_user(auth.id)221 if not user:222 raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail="Compte introuvable.")223 await users.touch(user.id)224 return _user_dict(user)225226227@router.patch("/me/preferences")228async def patch_prefs(req: PrefsReq, auth: AuthUser = Depends(get_current_user)) -> dict:229 data = req.model_dump(exclude_none=True)230 display_name = data.pop("display_name", None)231 user = await users.update_preferences(auth.id, data, display_name)232 return _user_dict(user)233234235@router.post("/me/password")236async def change_password(req: ChangePasswordReq, auth: AuthUser = Depends(get_current_user)) -> dict:237 user = await users.get_user(auth.id)238 if not user:239 raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail="Compte introuvable.")240 _check_password_strength(req.new_password)241 if user.password_hash and not verify_password(req.current_password or "", user.password_hash):242 raise HTTPException(status.HTTP_400_BAD_REQUEST, detail="Mot de passe actuel incorrect.")243 await users.set_password(auth.id, req.new_password)244 return {"ok": True}245246247@router.post("/me/consent")248async def consent(auth: AuthUser = Depends(get_current_user)) -> dict:249 await users.set_consent(auth.id)250 return {"ok": True}251252253@router.delete("/me")254async def delete_me(response: Response, auth: AuthUser = Depends(get_current_user)) -> dict:255 await users.delete_user_data(auth.id)256 response.delete_cookie(COOKIE_NAME, path="/")257 return {"ok": True, "message_fr": "Tes données ont été supprimées."}258