161 lines
5.0 KiB
Python
161 lines
5.0 KiB
Python
"""Authentication routes: register, login, logout and current-user lookup.
|
|
|
|
Identity is a self-hosted username/password pair. Passwords are hashed with
|
|
Argon2; sessions are opaque random tokens whose SHA-256 is persisted, delivered
|
|
in an httpOnly cookie.
|
|
"""
|
|
from datetime import timedelta
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, Response
|
|
from sqlalchemy import delete, select
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import Session as DbSession
|
|
|
|
from app import config, security
|
|
from app.db import get_db
|
|
from app.deps import get_current_user, require_same_origin
|
|
from app.models import User, UserSession, utcnow
|
|
from app.schemas import AuthResponse, LoginRequest, RegisterRequest, UserOut
|
|
|
|
router = APIRouter(prefix="/api/v1/auth", tags=["auth"])
|
|
|
|
throttle = security.LoginThrottle(
|
|
config.LOGIN_MAX_ATTEMPTS, config.LOGIN_WINDOW_SECONDS
|
|
)
|
|
|
|
_COOKIE_MAX_AGE = config.SESSION_TTL_DAYS * 24 * 60 * 60
|
|
|
|
|
|
def _set_session_cookie(response: Response, token: str) -> None:
|
|
response.set_cookie(
|
|
key=config.COOKIE_NAME,
|
|
value=token,
|
|
max_age=_COOKIE_MAX_AGE,
|
|
httponly=True,
|
|
samesite="lax",
|
|
secure=config.COOKIE_SECURE,
|
|
path="/",
|
|
)
|
|
|
|
|
|
def _issue_session(db: DbSession, user: User) -> str:
|
|
token = security.new_session_token()
|
|
now = utcnow()
|
|
# Drop this user's expired sessions while we are here.
|
|
db.execute(
|
|
delete(UserSession).where(
|
|
UserSession.user_id == user.id, UserSession.expires_at <= now
|
|
)
|
|
)
|
|
db.add(
|
|
UserSession(
|
|
token_hash=security.hash_token(token),
|
|
user_id=user.id,
|
|
created_at=now,
|
|
expires_at=now + timedelta(days=config.SESSION_TTL_DAYS),
|
|
)
|
|
)
|
|
db.commit()
|
|
return token
|
|
|
|
|
|
@router.post(
|
|
"/register",
|
|
status_code=201,
|
|
response_model=AuthResponse,
|
|
dependencies=[Depends(require_same_origin)],
|
|
)
|
|
def register(payload: RegisterRequest, response: Response, db: DbSession = Depends(get_db)):
|
|
username = payload.username.strip()
|
|
if not security.valid_username(username):
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail="Username must be 3-32 characters: letters, digits, '.', '_' or '-'",
|
|
)
|
|
|
|
username_key = security.normalize_username(username)
|
|
existing = db.execute(
|
|
select(User).where(User.username_key == username_key)
|
|
).scalar_one_or_none()
|
|
if existing is not None:
|
|
raise HTTPException(status_code=409, detail="Username is already taken")
|
|
|
|
user = User(
|
|
username=username,
|
|
username_key=username_key,
|
|
password_hash=security.hash_password(payload.password),
|
|
)
|
|
db.add(user)
|
|
try:
|
|
db.commit()
|
|
except IntegrityError:
|
|
db.rollback()
|
|
raise HTTPException(status_code=409, detail="Username is already taken")
|
|
|
|
db.refresh(user)
|
|
_set_session_cookie(response, _issue_session(db, user))
|
|
return {"user": UserOut.model_validate(user)}
|
|
|
|
|
|
@router.post(
|
|
"/login",
|
|
response_model=AuthResponse,
|
|
dependencies=[Depends(require_same_origin)],
|
|
)
|
|
def login(payload: LoginRequest, request: Request, response: Response, db: DbSession = Depends(get_db)):
|
|
username_key = security.normalize_username(payload.username)
|
|
ip = security.client_ip(request)
|
|
|
|
wait = throttle.retry_after(username_key, ip)
|
|
if wait:
|
|
raise HTTPException(
|
|
status_code=429,
|
|
detail="Too many failed login attempts. Try again later.",
|
|
headers={"Retry-After": str(wait)},
|
|
)
|
|
|
|
user = db.execute(
|
|
select(User).where(User.username_key == username_key)
|
|
).scalar_one_or_none()
|
|
|
|
if user is None:
|
|
security.burn_password_check(payload.password)
|
|
throttle.record_failure(username_key, ip)
|
|
raise HTTPException(status_code=401, detail="Invalid username or password")
|
|
|
|
if not security.verify_password(payload.password, user.password_hash):
|
|
throttle.record_failure(username_key, ip)
|
|
raise HTTPException(status_code=401, detail="Invalid username or password")
|
|
|
|
throttle.reset(username_key, ip)
|
|
if security.needs_rehash(user.password_hash):
|
|
user.password_hash = security.hash_password(payload.password)
|
|
db.commit()
|
|
|
|
_set_session_cookie(response, _issue_session(db, user))
|
|
return {"user": UserOut.model_validate(user)}
|
|
|
|
|
|
@router.post(
|
|
"/logout",
|
|
status_code=204,
|
|
dependencies=[Depends(require_same_origin)],
|
|
)
|
|
def logout(request: Request, db: DbSession = Depends(get_db)):
|
|
token = request.cookies.get(config.COOKIE_NAME)
|
|
if token:
|
|
db.execute(
|
|
delete(UserSession).where(
|
|
UserSession.token_hash == security.hash_token(token)
|
|
)
|
|
)
|
|
db.commit()
|
|
response = Response(status_code=204)
|
|
response.delete_cookie(config.COOKIE_NAME, path="/")
|
|
return response
|
|
|
|
|
|
@router.get("/me", response_model=AuthResponse)
|
|
def me(user: User = Depends(get_current_user)):
|
|
return {"user": UserOut.model_validate(user)}
|