This commit is contained in:
@@ -0,0 +1,114 @@
|
||||
"""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,
|
||||
)
|
||||
Reference in New Issue
Block a user