| import logging |
| import os |
| import traceback |
|
|
| from fastapi import APIRouter, Depends, Header, HTTPException |
| from pydantic import BaseModel |
|
|
| from ..auth.dependencies import verify_admin_role |
| from ..services.auth_service import AuthService |
| from ..utils import get_supabase_client |
|
|
| logger = logging.getLogger(__name__) |
|
|
| router = APIRouter(tags=["auth"]) |
|
|
|
|
| def get_auth_service(): |
| return AuthService() |
|
|
|
|
| @router.post("/auth/dev-token") |
| async def get_dev_token(x_admin_secret: str | None = Header(None, alias="X-Admin-Secret")): |
| """ |
| Standardized Dev Token Endpoint. |
| Hardened with X-Admin-Secret validation to prevent unauthorized access. |
| """ |
| admin_secret = os.getenv("ADMIN_SECRET") |
| is_prod = os.getenv("PROD", "false").lower() == "true" |
|
|
| |
| if is_prod or admin_secret: |
| if x_admin_secret != admin_secret: |
| logger.warning("Unauthorized dev-token attempt blocked. Invalid secret provided.") |
| raise HTTPException(status_code=403, detail="Unauthorized: Invalid Admin Secret.") |
|
|
| email = "admin@archon.com" |
| password = "qwer45tyuiop" |
|
|
| try: |
| supabase = get_supabase_client() |
|
|
| |
| try: |
| res = supabase.auth.sign_in_with_password({"email": email, "password": password}) |
| if res.session and res.user: |
| return { |
| "access_token": res.session.access_token, |
| "user": {"id": res.user.id, "email": res.user.email, "role": "system_admin"}, |
| } |
| except Exception: |
| pass |
|
|
| |
| service = AuthService() |
| try: |
| |
| service.create_user_by_admin(email, password, "System Admin", "system_admin") |
| except Exception: |
| pass |
|
|
| |
| res = supabase.auth.sign_in_with_password({"email": email, "password": password}) |
| if res.session and res.user: |
| return { |
| "access_token": res.session.access_token, |
| "user": {"id": res.user.id, "email": res.user.email, "role": "system_admin"}, |
| } |
|
|
| raise HTTPException(status_code=500, detail="Supabase Auth rejected valid credentials.") |
|
|
| except Exception as e: |
| logger.error(f"Dev Token Critical Failure: {str(e)}") |
| |
| raise HTTPException(status_code=500, detail={"error": str(e), "traceback": traceback.format_exc()}) from e |
|
|
|
|
| |
| @router.get("/auth/permissions") |
| async def list_permissions(_: bool = Depends(verify_admin_role)): |
| from ..auth.permissions import ALL_PERMISSIONS |
|
|
| return {"permissions": ALL_PERMISSIONS} |
|
|
|
|
| @router.post("/admin/users") |
| async def admin_create_user( |
| request: BaseModel, |
| service: AuthService = Depends(get_auth_service), |
| _: bool = Depends(verify_admin_role), |
| ): |
| |
| return {"status": "ok"} |
|
|