feat(auth): superadmin user-management page + CRUD
/admin/users page and /api/admin/users/* JSON CRUD endpoints, all behind
require_role("superadmin"). Temp passwords are returned once on create/reset
and never stored in plaintext. Admins get 403; password_hash is never leaked.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,103 @@
|
||||
"""Operator account management — superadmin only. Temp passwords are returned in
|
||||
the JSON response once (shown to the superadmin to hand off); only hashes persist."""
|
||||
from fastapi import APIRouter, Request, Depends, HTTPException
|
||||
from fastapi.responses import JSONResponse
|
||||
from pydantic import BaseModel
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from backend.database import get_db
|
||||
from backend.templates_config import templates
|
||||
from backend.models import OperatorUser
|
||||
from backend.operator_auth import (
|
||||
require_role, create_operator, reset_operator_password,
|
||||
set_operator_active, set_operator_role,
|
||||
)
|
||||
from backend.utils.timezone import format_local_datetime
|
||||
|
||||
router = APIRouter(tags=["operator-users"])
|
||||
_superadmin = require_role("superadmin")
|
||||
|
||||
|
||||
class NewUser(BaseModel):
|
||||
email: str
|
||||
name: str
|
||||
role: str = "admin"
|
||||
|
||||
|
||||
class RoleChange(BaseModel):
|
||||
role: str
|
||||
|
||||
|
||||
def _serialize(u: OperatorUser) -> dict:
|
||||
from datetime import datetime
|
||||
return {
|
||||
"id": u.id, "email": u.email, "display_name": u.display_name, "role": u.role,
|
||||
"active": bool(u.active), "must_change_password": bool(u.must_change_password),
|
||||
"locked": bool(u.locked_until and u.locked_until > datetime.utcnow()),
|
||||
"last_login_at": format_local_datetime(u.last_login_at, "%Y-%m-%d %H:%M") if u.last_login_at else None,
|
||||
}
|
||||
|
||||
|
||||
@router.get("/admin/users")
|
||||
async def users_page(request: Request, _=Depends(_superadmin)):
|
||||
return templates.TemplateResponse("admin/users.html", {"request": request})
|
||||
|
||||
|
||||
@router.get("/api/admin/users")
|
||||
async def list_users(_=Depends(_superadmin), db: Session = Depends(get_db)):
|
||||
users = db.query(OperatorUser).order_by(OperatorUser.display_name).all()
|
||||
return {"users": [_serialize(u) for u in users]}
|
||||
|
||||
|
||||
@router.post("/api/admin/users")
|
||||
async def add_user(body: NewUser, _=Depends(_superadmin), db: Session = Depends(get_db)):
|
||||
if body.role not in ("admin", "superadmin"):
|
||||
return JSONResponse(status_code=400, content={"detail": "role must be admin or superadmin"})
|
||||
try:
|
||||
user, raw = create_operator(db, body.email, body.name, body.role)
|
||||
except ValueError as e:
|
||||
return JSONResponse(status_code=400, content={"detail": str(e)})
|
||||
return {"user": _serialize(user), "password": raw}
|
||||
|
||||
|
||||
@router.post("/api/admin/users/{user_id}/reset-password")
|
||||
async def reset_user_password(user_id: str, _=Depends(_superadmin), db: Session = Depends(get_db)):
|
||||
user = db.query(OperatorUser).filter_by(id=user_id).first()
|
||||
if not user:
|
||||
raise HTTPException(status_code=404, detail="User not found")
|
||||
raw = reset_operator_password(db, user)
|
||||
return {"password": raw}
|
||||
|
||||
|
||||
@router.post("/api/admin/users/{user_id}/disable")
|
||||
async def disable_user(user_id: str, acting=Depends(_superadmin), db: Session = Depends(get_db)):
|
||||
if acting and acting.id == user_id:
|
||||
return JSONResponse(status_code=400, content={"detail": "Cannot disable your own account"})
|
||||
user = db.query(OperatorUser).filter_by(id=user_id).first()
|
||||
if not user:
|
||||
raise HTTPException(status_code=404, detail="User not found")
|
||||
set_operator_active(db, user, False)
|
||||
return {"active": False}
|
||||
|
||||
|
||||
@router.post("/api/admin/users/{user_id}/enable")
|
||||
async def enable_user(user_id: str, _=Depends(_superadmin), db: Session = Depends(get_db)):
|
||||
user = db.query(OperatorUser).filter_by(id=user_id).first()
|
||||
if not user:
|
||||
raise HTTPException(status_code=404, detail="User not found")
|
||||
set_operator_active(db, user, True)
|
||||
return {"active": True}
|
||||
|
||||
|
||||
@router.post("/api/admin/users/{user_id}/role")
|
||||
async def change_user_role(user_id: str, body: RoleChange,
|
||||
acting=Depends(_superadmin), db: Session = Depends(get_db)):
|
||||
if acting and acting.id == user_id:
|
||||
return JSONResponse(status_code=400, content={"detail": "Cannot change your own role"})
|
||||
if body.role not in ("admin", "superadmin"):
|
||||
return JSONResponse(status_code=400, content={"detail": "role must be admin or superadmin"})
|
||||
user = db.query(OperatorUser).filter_by(id=user_id).first()
|
||||
if not user:
|
||||
raise HTTPException(status_code=404, detail="User not found")
|
||||
set_operator_role(db, user, body.role)
|
||||
return {"role": user.role}
|
||||
Reference in New Issue
Block a user