Spaces:
Sleeping
Sleeping
File size: 2,043 Bytes
1397957 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 | from typing import Optional
from fastapi import HTTPException, Depends, Request
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
from jose import jwt, JWTError
from .config import settings
from .supabase import get_client, is_enabled as supabase_enabled
security = HTTPBearer(auto_error=False)
class AuthUser(BaseModel):
id: str
email: Optional[str] = None
role: Optional[str] = None
def decode_supabase_jwt(token: str) -> Optional[dict]:
if not settings.supabase_jwt_secret:
return None
try:
payload = jwt.decode(
token,
settings.supabase_jwt_secret,
algorithms=["HS256"],
audience="authenticated"
)
return payload
except JWTError:
return None
async def get_current_user(
credentials: Optional[HTTPAuthorizationCredentials] = Depends(security)
) -> Optional[AuthUser]:
if not credentials:
return None
token = credentials.credentials
# Check for HF TOKEN secret
if settings.token and token == settings.token:
return AuthUser(id="hf_user", role="admin")
if not supabase_enabled():
return None
payload = decode_supabase_jwt(token)
if not payload:
return None
return AuthUser(
id=payload.get("sub"),
email=payload.get("email"),
role=payload.get("role")
)
async def require_auth(
user: Optional[AuthUser] = Depends(get_current_user)
) -> AuthUser:
if not user:
if settings.token:
raise HTTPException(status_code=401, detail="Invalid or missing TOKEN")
if not supabase_enabled():
raise HTTPException(status_code=503, detail="Authentication not configured")
raise HTTPException(status_code=401, detail="Invalid or missing authentication token")
return user
async def optional_auth(
user: Optional[AuthUser] = Depends(get_current_user)
) -> Optional[AuthUser]:
return user
|