285 lines
10 KiB
Python
285 lines
10 KiB
Python
# Copyright 2026 Umbra Tools (umbratools.com)
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""Authentication routes: ``/api/auth/*``.
|
|
|
|
Handlers are synchronous so Argon2 verification runs in FastAPI's threadpool.
|
|
The session cookie is written only here; the raw token is never logged or
|
|
returned in a body.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Depends, Query, Request, Response
|
|
|
|
from .app import (
|
|
clear_csrf_cookie,
|
|
clear_session_cookie,
|
|
ensure_csrf_token,
|
|
get_principal_allowing_password_change,
|
|
get_service,
|
|
get_settings,
|
|
read_csrf_token,
|
|
request_id,
|
|
require_principal,
|
|
set_session_cookie,
|
|
)
|
|
from .models import Principal
|
|
from .schemas import (
|
|
ChangeFirstPasswordRequest,
|
|
ChangeOwnPasswordRequest,
|
|
LoginRequest,
|
|
LoginResponse,
|
|
LogoutResponse,
|
|
PasswordResetAvailableResponse,
|
|
ResendVerificationRequest,
|
|
ResendVerificationResponse,
|
|
ResetPasswordRequest,
|
|
SessionUserResponse,
|
|
UserResponse,
|
|
VerifyEmailRequest,
|
|
VerifyEmailResponse,
|
|
)
|
|
from .service import AccountService
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
def _profile(service: AccountService, user_id: object) -> UserResponse:
|
|
"""Return the sanitized profile for one account."""
|
|
|
|
return UserResponse.from_view(service.self_view(user_id)) # type: ignore[arg-type]
|
|
|
|
|
|
def _csrf_for(request: Request, response: Response) -> str:
|
|
"""Reuse the caller's CSRF token, or issue one if the request had none."""
|
|
|
|
token = read_csrf_token(request)
|
|
if token:
|
|
return token
|
|
return ensure_csrf_token(response, settings=get_settings(request))
|
|
|
|
|
|
@router.post("/login", response_model=LoginResponse)
|
|
def login(payload: LoginRequest, request: Request, response: Response) -> LoginResponse:
|
|
"""Authenticate a local password and open a server-side session.
|
|
|
|
Failures are reported with a single generic error, so the response cannot
|
|
be used to discover whether an account exists.
|
|
"""
|
|
|
|
settings = get_settings(request)
|
|
service: AccountService = get_service(request)
|
|
result = service.authenticate(
|
|
email=payload.email,
|
|
password=payload.password,
|
|
request_id=request_id(request),
|
|
)
|
|
set_session_cookie(response, token=result.session.token, settings=settings)
|
|
return LoginResponse(
|
|
user=_profile(service, result.user.id),
|
|
must_change_password=result.session.must_change_password,
|
|
csrf_token=_csrf_for(request, response),
|
|
idle_expires_at=result.session.idle_expires_at,
|
|
absolute_expires_at=result.session.absolute_expires_at,
|
|
)
|
|
|
|
|
|
@router.post("/change-first-password", response_model=LoginResponse)
|
|
def change_first_password(
|
|
payload: ChangeFirstPasswordRequest,
|
|
request: Request,
|
|
response: Response,
|
|
principal: Principal = Depends(get_principal_allowing_password_change),
|
|
) -> LoginResponse:
|
|
"""Complete the mandatory first-login change and rotate the session."""
|
|
|
|
settings = get_settings(request)
|
|
service: AccountService = get_service(request)
|
|
result = service.change_first_password(
|
|
principal,
|
|
current_password=payload.current_password,
|
|
new_password=payload.new_password,
|
|
request_id=request_id(request),
|
|
)
|
|
set_session_cookie(response, token=result.session.token, settings=settings)
|
|
return LoginResponse(
|
|
user=_profile(service, result.user.id),
|
|
must_change_password=result.session.must_change_password,
|
|
csrf_token=_csrf_for(request, response),
|
|
idle_expires_at=result.session.idle_expires_at,
|
|
absolute_expires_at=result.session.absolute_expires_at,
|
|
)
|
|
|
|
|
|
@router.post("/logout", response_model=LogoutResponse)
|
|
def logout(
|
|
request: Request,
|
|
response: Response,
|
|
principal: Principal = Depends(get_principal_allowing_password_change),
|
|
) -> LogoutResponse:
|
|
"""Revoke the caller's current session and clear the cookie."""
|
|
|
|
settings = get_settings(request)
|
|
service: AccountService = get_service(request)
|
|
revoked = service.logout(principal, request_id=request_id(request))
|
|
clear_session_cookie(response, settings=settings)
|
|
clear_csrf_cookie(response, settings=settings)
|
|
return LogoutResponse(revoked_sessions=revoked)
|
|
|
|
|
|
@router.post("/logout-everywhere", response_model=LogoutResponse)
|
|
def logout_everywhere(
|
|
request: Request,
|
|
response: Response,
|
|
principal: Principal = Depends(get_principal_allowing_password_change),
|
|
) -> LogoutResponse:
|
|
"""Revoke every session for the caller, including the current one."""
|
|
|
|
settings = get_settings(request)
|
|
service: AccountService = get_service(request)
|
|
revoked = service.logout_everywhere(principal, request_id=request_id(request))
|
|
clear_session_cookie(response, settings=settings)
|
|
clear_csrf_cookie(response, settings=settings)
|
|
return LogoutResponse(revoked_sessions=revoked)
|
|
|
|
|
|
@router.post("/verify-email", response_model=VerifyEmailResponse)
|
|
def verify_email(payload: VerifyEmailRequest, request: Request) -> VerifyEmailResponse:
|
|
"""Consume a verification link and activate the account.
|
|
|
|
The token is a bearer capability taken from a mailed link, so it is not
|
|
echoed back and the response reveals nothing beyond the account it belongs
|
|
to. It may be a query parameter when the link is followed in a browser.
|
|
"""
|
|
|
|
service: AccountService = get_service(request)
|
|
account = service.verify_email(payload.token, request_id=request_id(request))
|
|
return VerifyEmailResponse(
|
|
account=UserResponse.from_view(account),
|
|
message="Your email address is confirmed. You can sign in now.",
|
|
)
|
|
|
|
|
|
@router.get("/verify-email", response_model=VerifyEmailResponse)
|
|
def verify_email_from_link(
|
|
request: Request,
|
|
token: str = Query(min_length=1, max_length=512),
|
|
) -> VerifyEmailResponse:
|
|
"""The same operation for a link followed in a browser."""
|
|
|
|
service: AccountService = get_service(request)
|
|
account = service.verify_email(token, request_id=request_id(request))
|
|
return VerifyEmailResponse(
|
|
account=UserResponse.from_view(account),
|
|
message="Your email address is confirmed. You can sign in now.",
|
|
)
|
|
|
|
|
|
@router.get("/reset-password", response_model=PasswordResetAvailableResponse)
|
|
def reset_password_available(token: str, request: Request) -> PasswordResetAvailableResponse:
|
|
"""Report whether a reset link is usable, without consuming it.
|
|
|
|
The page needs this to avoid showing a password form for a link that will be
|
|
refused on submit. It is deliberately one answer for every reason - unknown,
|
|
already used, expired - so it cannot be used to test whether a link exists.
|
|
"""
|
|
|
|
service: AccountService = get_service(request)
|
|
available = service.password_reset_is_usable(token)
|
|
return PasswordResetAvailableResponse(
|
|
available=available,
|
|
message=(
|
|
"This link works once and expires in one hour. Choose a new password below."
|
|
if available
|
|
else "This link is no longer valid. It may already have been used, or it may have expired."
|
|
),
|
|
)
|
|
|
|
|
|
@router.post("/reset-password", response_model=ResendVerificationResponse)
|
|
def reset_password(payload: ResetPasswordRequest, request: Request) -> ResendVerificationResponse:
|
|
"""Consume a reset link and set a new password.
|
|
|
|
Unauthenticated by design: the token is the credential. Every session on the
|
|
account is revoked when it succeeds, so anyone holding a cookie captured
|
|
before the reset loses access.
|
|
"""
|
|
|
|
service: AccountService = get_service(request)
|
|
service.reset_password_with_token(
|
|
payload.token,
|
|
payload.new_password,
|
|
request_id=request_id(request),
|
|
)
|
|
return ResendVerificationResponse(
|
|
message="Your password has been changed. You can sign in with it now."
|
|
)
|
|
|
|
|
|
@router.post("/change-password", response_model=LoginResponse)
|
|
def change_own_password(
|
|
payload: ChangeOwnPasswordRequest,
|
|
request: Request,
|
|
principal: Principal = Depends(get_principal_allowing_password_change),
|
|
) -> LoginResponse:
|
|
"""Change the signed-in person's own password.
|
|
|
|
Requires the current password and returns a fresh session, because every
|
|
other session on the account is revoked. The response shape matches sign-in
|
|
so the caller can continue without a second round trip.
|
|
"""
|
|
|
|
service: AccountService = get_service(request)
|
|
result = service.change_own_password(
|
|
principal,
|
|
current_password=payload.current_password,
|
|
new_password=payload.new_password,
|
|
request_id=request_id(request),
|
|
)
|
|
response = Response()
|
|
set_session_cookie(response, token=result.session.token, settings=request.app.state.settings)
|
|
return LoginResponse(
|
|
user=UserResponse.from_view(result.view),
|
|
must_change_password=False,
|
|
csrf_token=result.session.csrf_token,
|
|
idle_expires_at=result.session.idle_expires_at,
|
|
absolute_expires_at=result.session.absolute_expires_at,
|
|
)
|
|
|
|
|
|
@router.post("/resend-verification", response_model=ResendVerificationResponse)
|
|
def resend_verification(
|
|
payload: ResendVerificationRequest,
|
|
request: Request,
|
|
) -> ResendVerificationResponse:
|
|
"""Re-mail a verification link.
|
|
|
|
The response is the same whether an account was found or not, so this cannot
|
|
be used to discover which addresses have accounts.
|
|
"""
|
|
|
|
service: AccountService = get_service(request)
|
|
service.resend_email_verification(payload.email, request_id=request_id(request))
|
|
return ResendVerificationResponse(
|
|
message="If that address has an account awaiting confirmation, a new link is on its way."
|
|
)
|
|
|
|
|
|
@router.get("/me", response_model=SessionUserResponse)
|
|
def me(
|
|
request: Request,
|
|
response: Response,
|
|
principal: Principal = Depends(require_principal),
|
|
) -> SessionUserResponse:
|
|
"""Return the caller's own identity and the CSRF token to echo back."""
|
|
|
|
service: AccountService = get_service(request)
|
|
return SessionUserResponse(
|
|
user=_profile(service, principal.user_id),
|
|
role=principal.role,
|
|
organization_id=principal.organization_id,
|
|
must_change_password=principal.must_change_password,
|
|
csrf_token=_csrf_for(request, response),
|
|
)
|