115 lines
3.8 KiB
Python
115 lines
3.8 KiB
Python
"""Admin-portal user management. Every endpoint requires an admin — regular
|
|
users cannot create, delete, or reset the password of any account (they can
|
|
only change their own password via ``/auth/change-password``).
|
|
"""
|
|
|
|
from fastapi import APIRouter, Depends, Request
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session as DbSession
|
|
|
|
from .. import auth
|
|
from ..auth import require_admin
|
|
from ..database import get_db
|
|
from ..errors import APIError
|
|
from ..models import Admin, Session
|
|
from ..schemas import AdminCreate, AdminOut, ResetPasswordRequest
|
|
|
|
router = APIRouter(prefix="/users", tags=["users"])
|
|
|
|
|
|
@router.get("", response_model=list[AdminOut])
|
|
def list_users(_: Admin = Depends(require_admin), db: DbSession = Depends(get_db)):
|
|
return db.execute(select(Admin).order_by(Admin.id)).scalars().all()
|
|
|
|
|
|
@router.post("", response_model=AdminOut, status_code=201)
|
|
def create_user(
|
|
payload: AdminCreate,
|
|
request: Request,
|
|
admin: Admin = Depends(require_admin),
|
|
db: DbSession = Depends(get_db),
|
|
):
|
|
exists = db.execute(
|
|
select(Admin.id).where(Admin.username == payload.username)
|
|
).scalar_one_or_none()
|
|
if exists is not None:
|
|
raise APIError(status_code=409, detail=f"User '{payload.username}' already exists")
|
|
|
|
user = Admin(
|
|
username=payload.username,
|
|
password_hash=auth.hash_password(payload.password),
|
|
is_admin=payload.is_admin,
|
|
must_change_password=payload.force_password_change,
|
|
)
|
|
db.add(user)
|
|
db.commit()
|
|
db.refresh(user)
|
|
auth.record_log(
|
|
db,
|
|
username=admin.username,
|
|
action="create_user",
|
|
detail=f"Created user '{user.username}'" + (" (admin)" if user.is_admin else ""),
|
|
request=request,
|
|
)
|
|
return user
|
|
|
|
|
|
@router.delete("/{user_id}", status_code=204)
|
|
def delete_user(
|
|
user_id: int,
|
|
request: Request,
|
|
admin: Admin = Depends(require_admin),
|
|
db: DbSession = Depends(get_db),
|
|
):
|
|
if user_id == admin.id:
|
|
raise APIError(status_code=400, detail="You cannot delete your own account")
|
|
user = db.get(Admin, user_id)
|
|
if user is None:
|
|
raise APIError(status_code=404, detail=f"User '{user_id}' not found")
|
|
if user.is_admin:
|
|
admin_count = db.execute(
|
|
select(func.count()).select_from(Admin).where(Admin.is_admin.is_(True))
|
|
).scalar_one()
|
|
if admin_count <= 1:
|
|
raise APIError(status_code=400, detail="Cannot delete the last administrator")
|
|
|
|
# Drop the user's sessions too so any live token stops working immediately.
|
|
for s in db.execute(select(Session).where(Session.admin_id == user.id)).scalars().all():
|
|
db.delete(s)
|
|
db.delete(user)
|
|
db.commit()
|
|
auth.record_log(
|
|
db,
|
|
username=admin.username,
|
|
action="delete_user",
|
|
detail=f"Deleted user '{user.username}'",
|
|
request=request,
|
|
)
|
|
|
|
|
|
@router.post("/{user_id}/reset-password", status_code=204)
|
|
def reset_password(
|
|
user_id: int,
|
|
payload: ResetPasswordRequest,
|
|
request: Request,
|
|
admin: Admin = Depends(require_admin),
|
|
db: DbSession = Depends(get_db),
|
|
):
|
|
"""Admin resets another user's password; they must change it on next login."""
|
|
user = db.get(Admin, user_id)
|
|
if user is None:
|
|
raise APIError(status_code=404, detail=f"User '{user_id}' not found")
|
|
user.password_hash = auth.hash_password(payload.new_password)
|
|
user.must_change_password = payload.force_password_change
|
|
# Invalidate existing sessions so the old password/session can't be reused.
|
|
for s in db.execute(select(Session).where(Session.admin_id == user.id)).scalars().all():
|
|
db.delete(s)
|
|
db.commit()
|
|
auth.record_log(
|
|
db,
|
|
username=admin.username,
|
|
action="reset_password",
|
|
detail=f"Reset password for '{user.username}'",
|
|
request=request,
|
|
)
|