Spaces:
Sleeping
Sleeping
| """``/api/auth`` — proxy to the Supabase ``apex-auth`` edge function.""" | |
| import logging | |
| import httpx | |
| from fastapi import APIRouter, Depends, HTTPException, Request | |
| from slowapi import Limiter | |
| from slowapi.util import get_remote_address | |
| from app.core.auth import get_current_user | |
| from app.core.config import SUPABASE_EDGE_URL | |
| from app.types import AuthRequest | |
| log = logging.getLogger("auth-api") | |
| router = APIRouter() | |
| # Rate limiter: 5 auth attempts per minute per IP (prevents brute-force) | |
| limiter = Limiter(key_func=get_remote_address) | |
| async def handle_auth(request: Request, req: AuthRequest): | |
| async with httpx.AsyncClient(timeout=30) as client: | |
| resp = await client.post( | |
| f"{SUPABASE_EDGE_URL}/apex-auth", | |
| json={"action": req.action, "email": req.email, "password": req.password, "name": req.name}, | |
| ) | |
| data = resp.json() | |
| if resp.status_code != 200: | |
| raise HTTPException(status_code=resp.status_code, detail="Authentication failed") | |
| # Extract user_id from various possible fields in the response | |
| user_id = None | |
| # Check for common user identification fields | |
| if "user_id" in data: | |
| user_id = data["user_id"] | |
| elif "id" in data: | |
| user_id = data["id"] | |
| elif "user" in data and isinstance(data["user"], dict): | |
| user_id = data["user"].get("id") or data["user"].get("user_id") or data["user"].get("sub") | |
| # Also check for user_id in JWT token if present in response | |
| if not user_id and "access_token" in data: | |
| try: | |
| import jwt | |
| token_data = jwt.decode( | |
| data["access_token"], | |
| options={"verify_signature": False, "verify_aud": False}, | |
| algorithms=["HS256"] | |
| ) | |
| user_id = token_data.get("sub") | |
| except Exception: | |
| pass | |
| # Last resort: check if response is just a user object | |
| if not user_id and len(data) == 1 and isinstance(list(data.values())[0], dict): | |
| possible_user = list(data.values())[0] | |
| user_id = possible_user.get("id") or possible_user.get("user_id") or possible_user.get("sub") | |
| if user_id: | |
| data["user_id"] = user_id | |
| log.info("[auth] User logged in, user_id=%s", user_id) | |
| else: | |
| log.warning("[auth] No user_id in auth response - keys: %s", list(data.keys())) | |
| # For debugging: log the full response (masked) | |
| import json | |
| masked_data = json.dumps(data) | |
| if len(masked_data) > 500: | |
| masked_data = masked_data[:500] + "..." | |
| log.warning("[auth] Full response (masked): %s", masked_data) | |
| # Log the response structure (without secrets) to help debug token issues | |
| top_keys = list(data.keys()) if isinstance(data, dict) else type(data).__name__ | |
| log.info("[auth] Edge response keys: %s", top_keys) | |
| if "access_token" in data: | |
| log.info("[auth] access_token present (len=%d)", len(str(data["access_token"]))) | |
| if "session" in data and isinstance(data["session"], dict): | |
| log.info("[auth] session.access_token present: %s", "access_token" in data["session"]) | |
| if "token" in data: | |
| log.info("[auth] token present (len=%d)", len(str(data["token"]))) | |
| # Normalize response: promote access_token / session to top level | |
| # if the edge function wrapped them inside a "data" key | |
| if "access_token" not in data and "session" not in data: | |
| nested = data.get("data") | |
| if isinstance(nested, dict): | |
| if "access_token" in nested: | |
| data["access_token"] = nested["access_token"] | |
| log.info("[auth] Promoted access_token from data.%s", "access_token") | |
| if "session" in nested and isinstance(nested["session"], dict): | |
| data["session"] = nested["session"] | |
| log.info("[auth] Promoted session from data.session") | |
| if "user" in nested and isinstance(nested["user"], dict) and "user_id" not in data: | |
| uid = nested["user"].get("id") or nested["user"].get("user_id") or nested["user"].get("sub") | |
| if uid: | |
| data["user_id"] = uid | |
| log.info("[auth] Promoted user_id from data.user") | |
| return data | |
| async def get_current_auth_user(user: dict = Depends(get_current_user)): | |
| """Get current authenticated user information.""" | |
| return user | |