codenexus / middleware /admin_guard.py
farhan1221's picture
Upload 26 files
91990f9 verified
Raw
History Blame Contribute Delete
1.83 kB
from fastapi import Request, HTTPException
from middleware.auth_guard import get_current_user
from config.database import get_supabase_admin
async def get_admin_user(request: Request):
"""
Verifies user is authenticated AND has admin role.
Also verifies the secret admin prefix from the URL matches the one in DB.
"""
# 1. Verify User Role
user = await get_current_user(request)
admin_client = get_supabase_admin()
try:
profile = admin_client.table("users").select("role").eq("id", str(user.id)).single().execute()
if not profile.data or profile.data.get("role") != "admin":
raise HTTPException(status_code=403, detail="Admin access required")
except Exception:
raise HTTPException(status_code=403, detail="Admin access required")
# 2. Verify Secret URL Segment (X-Admin-Secret header or similar)
# To avoid changing all routes to {secret}, we'll check it from DB.
# The Admin Panel will send the current secret in the X-Admin-Secret header.
provided_secret = request.headers.get("X-Admin-Secret")
# Fallback to hardcoded if DB call fails or is empty for bootstrapping
# But ideally it should be in DB.
settings_res = admin_client.table("site_settings").select("value").eq("key", "admin_secret_url").single().execute()
db_secret = settings_res.data.get("value") if settings_res.data else "cn-admin-nc-0947"
if provided_secret != db_secret:
# We also allow the secret to be part of the path if we use the old prefix
# for backward compatibility during transition.
if "cn-admin-nc-0947" not in request.url.path and provided_secret != db_secret:
raise HTTPException(status_code=403, detail="Invalid admin secret")
return user