diff --git a/backend/app/routes/auth.py b/backend/app/routes/auth.py index aae3e87..dcd64d0 100644 --- a/backend/app/routes/auth.py +++ b/backend/app/routes/auth.py @@ -1,18 +1,120 @@ from datetime import timedelta + from fastapi import APIRouter, Depends, HTTPException, status from fastapi.security import OAuth2PasswordRequestForm -from app.auth import ACCESS_TOKEN_EXPIRE_MINUTES, Token, authenticate_user, create_access_token, get_current_active_user +from pydantic import BaseModel +from sqlalchemy.orm import Session +from typing import Optional + +from app.auth import ( + ACCESS_TOKEN_EXPIRE_MINUTES, + Token, + UserOut, + authenticate_user, + create_access_token, + get_current_active_user, + get_user_by_username, + hash_password, + require_role, +) +from app.database import get_db +from app.models import UserModel router = APIRouter(prefix="/auth", tags=["auth"]) + +class UserCreate(BaseModel): + username: str + password: str + email: Optional[str] = None + role: str = "viewer" + + +class PasswordChange(BaseModel): + current_password: str + new_password: str + + +# ---------- Endpoints ---------- + @router.post("/token", response_model=Token) -async def login(form_data: OAuth2PasswordRequestForm = Depends()): - user = authenticate_user(form_data.username, form_data.password) +def login( + form_data: OAuth2PasswordRequestForm = Depends(), + db: Session = Depends(get_db), +): + user = authenticate_user(db, form_data.username, form_data.password) if not user: - raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect username or password") - access_token = create_access_token({"sub": user.username, "role": user.role}, timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)) + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Incorrect username or password", + headers={"WWW-Authenticate": "Bearer"}, + ) + access_token = create_access_token( + {"sub": user.username, "role": user.role}, + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES), + ) return Token(access_token=access_token) -@router.get("/me") -async def me(current_user=Depends(get_current_active_user)): - return {"username": current_user.username, "role": current_user.role, "disabled": current_user.disabled} + +@router.get("/me", response_model=UserOut) +def me(current_user: UserModel = Depends(get_current_active_user)): + return current_user + + +@router.patch("/me/password") +def change_my_password( + body: PasswordChange, + current_user: UserModel = Depends(get_current_active_user), + db: Session = Depends(get_db), +): + from app.auth import verify_password + if not verify_password(body.current_password, current_user.hashed_password): + raise HTTPException(status_code=400, detail="Current password is incorrect") + current_user.hashed_password = hash_password(body.new_password) + current_user.force_password_change = False + db.commit() + return {"detail": "Password updated"} + + +@router.get("/users", response_model=list[UserOut]) +def list_users( + _: UserModel = Depends(require_role("admin")), + db: Session = Depends(get_db), +): + return db.query(UserModel).order_by(UserModel.id).all() + + +@router.post("/users", response_model=UserOut, status_code=201) +def create_user( + body: UserCreate, + _: UserModel = Depends(require_role("admin")), + db: Session = Depends(get_db), +): + if get_user_by_username(db, body.username): + raise HTTPException(status_code=409, detail="Username already exists") + user = UserModel( + username=body.username, + hashed_password=hash_password(body.password), + email=body.email, + role=body.role, + disabled=False, + force_password_change=False, + ) + db.add(user) + db.commit() + db.refresh(user) + return user + + +@router.patch("/users/{username}/disable") +def disable_user( + username: str, + _: UserModel = Depends(require_role("admin")), + db: Session = Depends(get_db), +): + user = get_user_by_username(db, username) + if not user: + raise HTTPException(status_code=404, detail="User not found") + user.disabled = True + db.commit() + return {"detail": f"User {username!r} disabled"}