50 lines
1.7 KiB
Python
50 lines
1.7 KiB
Python
"""Admin authentication: login, me, change password."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.db import get_db
|
|
from app.models import AdminUser
|
|
from app.schemas import ChangePasswordIn, LoginIn, MeOut, TokenOut
|
|
from app.security import (
|
|
create_token,
|
|
get_current_admin,
|
|
hash_password,
|
|
verify_password,
|
|
)
|
|
|
|
router = APIRouter(prefix="/auth")
|
|
|
|
|
|
@router.post("/login", response_model=TokenOut)
|
|
def login(body: LoginIn, db: Session = Depends(get_db)) -> TokenOut:
|
|
user = db.query(AdminUser).filter(AdminUser.username == body.username).first()
|
|
if not user or not verify_password(body.password, user.password_hash):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid username or password",
|
|
)
|
|
return TokenOut(access_token=create_token(user.username), username=user.username)
|
|
|
|
|
|
@router.get("/me", response_model=MeOut)
|
|
def me(admin: AdminUser = Depends(get_current_admin)) -> MeOut:
|
|
return MeOut(username=admin.username)
|
|
|
|
|
|
@router.post("/change-password", response_model=MeOut)
|
|
def change_password(
|
|
body: ChangePasswordIn,
|
|
admin: AdminUser = Depends(get_current_admin),
|
|
db: Session = Depends(get_db),
|
|
) -> MeOut:
|
|
if not verify_password(body.current_password, admin.password_hash):
|
|
raise HTTPException(status_code=400, detail="Current password is incorrect")
|
|
if len(body.new_password) < 6:
|
|
raise HTTPException(status_code=400, detail="New password must be at least 6 characters")
|
|
admin.password_hash = hash_password(body.new_password)
|
|
db.commit()
|
|
return MeOut(username=admin.username)
|