admin: add role management endpoints + Admin UI; strengthen RBAC and add tests
This commit is contained in:
@@ -45,6 +45,37 @@ def hello():
|
||||
app.include_router(oauth_router, prefix='/api/v1')
|
||||
app.include_router(auth_router, prefix='/api/v1/auth')
|
||||
|
||||
|
||||
from .rbac import require_admin
|
||||
|
||||
|
||||
@app.get('/api/v1/admin/users')
|
||||
def admin_list_users(admin_user=Depends(require_admin)):
|
||||
# placeholder; will be replaced with require_admin dependency
|
||||
db = models.SessionLocal()
|
||||
try:
|
||||
rows = db.query(models.User).all()
|
||||
return [{'id': r.id, 'email': r.email, 'role': r.role} for r in rows]
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
@app.post('/api/v1/admin/users/{user_id}/role')
|
||||
def admin_set_role(user_id: int, payload: dict, admin_user=Depends(require_admin)):
|
||||
role = payload.get('role')
|
||||
if role not in ['user', 'moderator', 'admin']:
|
||||
raise HTTPException(status_code=400, detail='invalid role')
|
||||
db = models.SessionLocal()
|
||||
try:
|
||||
user = db.query(models.User).filter_by(id=user_id).first()
|
||||
if not user:
|
||||
raise HTTPException(status_code=404, detail='user not found')
|
||||
user.role = role
|
||||
db.commit()
|
||||
return {'id': user.id, 'role': user.role}
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
# Basic user routes (demo)
|
||||
@app.post('/api/v1/users')
|
||||
def create_user(payload: dict):
|
||||
|
||||
+21
-9
@@ -1,21 +1,33 @@
|
||||
from fastapi import HTTPException
|
||||
from fastapi import HTTPException, Depends, Request
|
||||
from .auth import get_current_user
|
||||
from . import models
|
||||
|
||||
|
||||
# Role hierarchy for comparisons
|
||||
HIERARCHY = {'user': 1, 'moderator': 2, 'admin': 3}
|
||||
|
||||
|
||||
def require_role(min_role: str):
|
||||
# Simple role hierarchy
|
||||
hierarchy = {'user': 1, 'moderator': 2, 'admin': 3}
|
||||
def _inner(request=None):
|
||||
user = get_current_user(request)
|
||||
if hierarchy.get(user.role, 0) < hierarchy.get(min_role, 0):
|
||||
"""FastAPI dependency that enforces a minimum role on the calling user."""
|
||||
def _dep(user=Depends(get_current_user)):
|
||||
if HIERARCHY.get(user.role or 'user', 0) < HIERARCHY.get(min_role, 0):
|
||||
raise HTTPException(status_code=403, detail='insufficient role')
|
||||
return user
|
||||
return _inner
|
||||
return _dep
|
||||
|
||||
|
||||
def require_admin(user=Depends(get_current_user)):
|
||||
if HIERARCHY.get(user.role or 'user', 0) < HIERARCHY.get('admin'):
|
||||
raise HTTPException(status_code=403, detail='admin required')
|
||||
return user
|
||||
|
||||
|
||||
def require_owner_or_admin(resource_user_id: int):
|
||||
def _inner(request=None):
|
||||
"""Return a callable that can be used inline to check ownership/admin status.
|
||||
|
||||
Note: FastAPI path param injection into dependency factories is complex; for
|
||||
simplicity endpoints can call this helper with the resource owner id.
|
||||
"""
|
||||
def _inner(request: Request = None):
|
||||
user = get_current_user(request)
|
||||
if user.id == resource_user_id or user.role == 'admin':
|
||||
return user
|
||||
|
||||
Reference in New Issue
Block a user