Spaces:
Running
Running
| 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 | |