"""Anmeldung über server-seitige Sitzung im Cookie.""" from fastapi import APIRouter, Depends, HTTPException, Request, Response, status from fastapi.security import OAuth2PasswordRequestForm from sqlalchemy.orm import Session from ..config import get_settings from ..database import get_db from ..deps import get_current_user from ..models import User from ..schemas import PasswordChange, UserOut from ..security import hash_password, verify_password from ..services import sessions router = APIRouter(prefix="/auth", tags=["auth"]) def _set_session_cookie(response: Response, raw_token: str) -> None: settings = get_settings() response.set_cookie( key=settings.cookie_name, value=raw_token, max_age=settings.session_ttl_days * 86400, httponly=True, secure=settings.cookie_secure, samesite=settings.cookie_samesite, path="/", ) @router.post("/login", response_model=UserOut) def login( response: Response, request: Request, form: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db), ) -> User: user = db.query(User).filter(User.username == form.username).first() if user is None or not verify_password(form.password, user.password_hash): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Benutzername oder Passwort falsch", ) if not user.is_active: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Dieses Konto ist deaktiviert", ) raw = sessions.create_session(db, user, request.headers.get("user-agent")) _set_session_cookie(response, raw) return user @router.post("/logout", status_code=status.HTTP_204_NO_CONTENT) def logout(request: Request, response: Response, db: Session = Depends(get_db)) -> None: token = request.cookies.get(get_settings().cookie_name, "") if token: sessions.revoke(db, token) response.delete_cookie(get_settings().cookie_name, path="/") @router.get("/me", response_model=UserOut) def me(current_user: User = Depends(get_current_user)) -> User: return current_user @router.post("/change-password", response_model=UserOut) def change_password( payload: PasswordChange, current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ) -> User: if not verify_password(payload.current_password, current_user.password_hash): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Aktuelles Passwort ist falsch", ) current_user.password_hash = hash_password(payload.new_password) current_user.must_change_password = False db.commit() db.refresh(current_user) return current_user