95 lines
3.1 KiB
Python
95 lines
3.1 KiB
Python
"""Login, logout, current-user and self-service password change.
|
|
|
|
``POST /auth/login`` is the only unauthenticated endpoint here; the rest resolve
|
|
the caller from their Bearer session token.
|
|
"""
|
|
|
|
from fastapi import APIRouter, Depends, Request, Security
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session as DbSession
|
|
|
|
from .. import auth
|
|
from ..auth import bearer_scheme, get_current_admin
|
|
from ..database import get_db
|
|
from ..errors import APIError
|
|
from ..models import Admin, Session
|
|
from ..schemas import (
|
|
ChangePasswordRequest,
|
|
CurrentUser,
|
|
LoginRequest,
|
|
LoginResponse,
|
|
)
|
|
|
|
router = APIRouter(prefix="/auth", tags=["auth"])
|
|
|
|
|
|
@router.post("/login", response_model=LoginResponse)
|
|
def login(payload: LoginRequest, request: Request, db: DbSession = Depends(get_db)):
|
|
admin = db.execute(
|
|
select(Admin).where(Admin.username == payload.username)
|
|
).scalar_one_or_none()
|
|
if admin is None or not auth.verify_password(payload.password, admin.password_hash):
|
|
# Generic message so we don't leak which usernames exist.
|
|
raise APIError(status_code=401, detail="Invalid username or password")
|
|
|
|
token = auth.create_session(db, admin)
|
|
auth.record_log(
|
|
db, username=admin.username, action="login", detail="Signed in", request=request
|
|
)
|
|
return LoginResponse(
|
|
token=token,
|
|
username=admin.username,
|
|
is_admin=admin.is_admin,
|
|
must_change_password=admin.must_change_password,
|
|
)
|
|
|
|
|
|
@router.post("/logout", status_code=204)
|
|
def logout(
|
|
request: Request,
|
|
admin: Admin = Depends(get_current_admin),
|
|
creds=Security(bearer_scheme),
|
|
db: DbSession = Depends(get_db),
|
|
):
|
|
if creds and creds.credentials:
|
|
session = db.get(Session, creds.credentials)
|
|
if session is not None:
|
|
db.delete(session)
|
|
db.commit()
|
|
auth.record_log(db, username=admin.username, action="logout", detail="Signed out", request=request)
|
|
|
|
|
|
@router.get("/me", response_model=CurrentUser)
|
|
def me(admin: Admin = Depends(get_current_admin)):
|
|
return admin
|
|
|
|
|
|
@router.post("/change-password", status_code=204)
|
|
def change_password(
|
|
payload: ChangePasswordRequest,
|
|
request: Request,
|
|
admin: Admin = Depends(get_current_admin),
|
|
db: DbSession = Depends(get_db),
|
|
):
|
|
"""Change your OWN password. Any authenticated user may do this.
|
|
|
|
When the account is flagged ``must_change_password`` (a forced first-login or
|
|
post-reset change) the current password is not required — the user just
|
|
authenticated with it. A normal self-service change still verifies it.
|
|
"""
|
|
if not admin.must_change_password:
|
|
if not payload.current_password or not auth.verify_password(
|
|
payload.current_password, admin.password_hash
|
|
):
|
|
raise APIError(status_code=400, detail="Current password is incorrect")
|
|
admin.password_hash = auth.hash_password(payload.new_password)
|
|
admin.must_change_password = False
|
|
db.commit()
|
|
auth.record_log(
|
|
db,
|
|
username=admin.username,
|
|
action="change_password",
|
|
detail="Changed own password",
|
|
request=request,
|
|
)
|