Senko-san
14c1bc16e0
Add POST /auth/register: creates a non-superuser then auto-logs in, returning the same TokenResponse as login. Gated by the new allow_registration setting (env ALLOW_REGISTRATION, default true); when disabled it raises PermissionDeniedError (403). Accounts remain admin-only for superusers. Tests cover create+login, duplicate (409), short password (422), and the disabled (403) path. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
63 lines
2.2 KiB
Python
63 lines
2.2 KiB
Python
"""Auth endpoints: login, refresh (rotation), logout, and current-user."""
|
|
|
|
from fastapi import APIRouter, status
|
|
|
|
from app.api.deps import AuthServiceDep, CurrentUser, UserServiceDep
|
|
from app.api.schemas.auth import (
|
|
LoginRequest,
|
|
RefreshRequest,
|
|
RegisterRequest,
|
|
TokenResponse,
|
|
)
|
|
from app.api.schemas.user import UserResponse
|
|
from app.core.config import get_settings
|
|
from app.domain.errors import PermissionDeniedError
|
|
from app.domain.tokens import TokenPair
|
|
|
|
router = APIRouter(prefix="/auth", tags=["auth"])
|
|
|
|
|
|
def _to_token_response(pair: TokenPair) -> TokenResponse:
|
|
return TokenResponse(
|
|
access_token=pair.access.encoded,
|
|
refresh_token=pair.refresh.encoded,
|
|
)
|
|
|
|
|
|
@router.post("/login", response_model=TokenResponse)
|
|
async def login(body: LoginRequest, auth: AuthServiceDep) -> TokenResponse:
|
|
pair = await auth.login(body.username, body.password)
|
|
return _to_token_response(pair)
|
|
|
|
|
|
@router.post("/register", response_model=TokenResponse, status_code=status.HTTP_201_CREATED)
|
|
async def register(
|
|
body: RegisterRequest, users: UserServiceDep, auth: AuthServiceDep
|
|
) -> TokenResponse:
|
|
"""Public self-service sign-up (gated by ``ALLOW_REGISTRATION``).
|
|
|
|
Registered accounts are always regular users — superusers are created
|
|
admin-only. On success the new account is logged straight in.
|
|
"""
|
|
if not get_settings().allow_registration:
|
|
raise PermissionDeniedError("Registration is disabled on this instance.")
|
|
await users.create_user(username=body.username, password=body.password, is_superuser=False)
|
|
pair = await auth.login(body.username, body.password)
|
|
return _to_token_response(pair)
|
|
|
|
|
|
@router.post("/refresh", response_model=TokenResponse)
|
|
async def refresh(body: RefreshRequest, auth: AuthServiceDep) -> TokenResponse:
|
|
pair = await auth.refresh(body.refresh_token)
|
|
return _to_token_response(pair)
|
|
|
|
|
|
@router.post("/logout", status_code=status.HTTP_204_NO_CONTENT)
|
|
async def logout(body: RefreshRequest, auth: AuthServiceDep) -> None:
|
|
await auth.logout(body.refresh_token)
|
|
|
|
|
|
@router.get("/me", response_model=UserResponse)
|
|
async def me(user: CurrentUser) -> UserResponse:
|
|
return UserResponse.from_entity(user)
|