Spaces:
Sleeping
Sleeping
| from fastapi import HTTPException, Depends | |
| from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials | |
| from config.settings import settings | |
| import jwt | |
| import time | |
| import re | |
| import requests | |
| security = HTTPBearer() | |
| def is_jwt_token(token: str) -> bool: | |
| """Check if the token is a JWT by looking for the typical JWT pattern.""" | |
| # JWT pattern: three base64url-encoded segments separated by dots | |
| jwt_pattern = r"^[A-Za-z0-9-_=]+\.[A-Za-z0-9-_=]+\.[A-Za-z0-9-_.+/=]*$" | |
| return bool(re.match(jwt_pattern, token)) | |
| async def validate_huggingface_token(token: str) -> bool: | |
| """Validate a HuggingFace token by making a request to their API.""" | |
| try: | |
| headers = {"Authorization": f"Bearer {token}"} | |
| response = requests.get("https://huggingface.co/api/whoami-v2", headers=headers) | |
| return response.status_code == 200 | |
| except Exception: | |
| return False | |
| async def verify_auth(credentials: HTTPAuthorizationCredentials = Depends(security)): | |
| """Verify the authentication token, supporting both JWT and HuggingFace tokens.""" | |
| if not settings.auth.AUTH_ENABLED: | |
| return True | |
| try: | |
| # Get the token | |
| token = credentials.credentials | |
| # Check if it's a JWT token | |
| if is_jwt_token(token): | |
| # Decode the token without verification first to check claims | |
| unverified_payload = jwt.decode(token, options={"verify_signature": False}) | |
| # Check if token is older than 1 hour | |
| current_time = int(time.time()) | |
| if ( | |
| current_time - unverified_payload.get("iat", 0) > 3600 | |
| ): # 3600 seconds = 1 hour | |
| raise HTTPException( | |
| status_code=401, detail="Token is too old (max 1 hour)" | |
| ) | |
| # Check if subject is TEST_SERVICE1 (TODO: Change to Guild JWT Validation) | |
| if unverified_payload.get("sub") not in ["TEST_SERVICE1", "TEST_SERVICE_2"]: | |
| raise HTTPException(status_code=401, detail="Invalid subject") | |
| else: | |
| # For non-JWT tokens, try to validate as a HuggingFace token | |
| if not await validate_huggingface_token(token): | |
| raise HTTPException(status_code=401, detail="Invalid HuggingFace token") | |
| return True | |
| except jwt.InvalidTokenError as e: | |
| raise HTTPException(status_code=401, detail=f"Invalid token: {str(e)}") | |
| except Exception as e: | |
| raise HTTPException(status_code=401, detail=f"Authentication error: {str(e)}") | |