121 lines
3.3 KiB
Python
121 lines
3.3 KiB
Python
from datetime import timedelta
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from fastapi.security import OAuth2PasswordRequestForm
|
|
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)
|
|
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",
|
|
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", 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"}
|