Download src/utils/auth.py from DiabetesCareChatbot/dmChatbotBackend: direct link, hf CLI and curl.
- Browser
- Download file 3.33 kB
-
https://huggingface.co/spaces/DiabetesCareChatbot/dmChatbotBackend/resolve/main/src/utils/auth.py
- Command line
-
hf download hf://spaces/DiabetesCareChatbot/dmChatbotBackend/src/utils/auth.py
-
curl -L -o auth.py https://huggingface.co/spaces/DiabetesCareChatbot/dmChatbotBackend/resolve/main/src/utils/auth.py
3.33 kB
| import os | |
| import jwt | |
| import datetime | |
| import uuid | |
| from fastapi import Request, HTTPException, Depends | |
| from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials | |
| from src.utils.logger import setup_logger | |
| logger = setup_logger("Auth") | |
| security = HTTPBearer(auto_error=False) | |
| # Default patient fallback is allowed only in development when explicitly enabled | |
| DEFAULT_PATIENT_ID = os.getenv("DEFAULT_PATIENT_UUID") | |
| DEV_ALLOW_DEFAULT_PATIENT = os.getenv("DEV_ALLOW_DEFAULT_PATIENT", "false").lower() in ("1", "true", "yes") | |
| def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)): | |
| """ | |
| Verifies the Supabase JWT and returns the user ID (sub). | |
| """ | |
| if not credentials: | |
| raise HTTPException(status_code=401, detail="Missing credentials") | |
| token = credentials.credentials | |
| jwt_secret = os.getenv("SUPABASE_JWT_SECRET") | |
| if not jwt_secret: | |
| logger.error("SUPABASE_JWT_SECRET not found in environment") | |
| raise HTTPException(status_code=500, detail="JWT secret missing") | |
| try: | |
| # Supabase uses HS256 for signing JWTs with the project secret | |
| payload = jwt.decode(token, jwt_secret, algorithms=["HS256"], audience="authenticated") | |
| user_id = payload.get("sub") | |
| if not user_id: | |
| raise HTTPException(status_code=401, detail="Invalid token: missing sub") | |
| return user_id | |
| except jwt.ExpiredSignatureError: | |
| raise HTTPException(status_code=401, detail="Token has expired") | |
| except jwt.InvalidTokenError as e: | |
| logger.warning(f"Invalid token: {e}") | |
| raise HTTPException(status_code=401, detail="Invalid token") | |
| def get_active_user(credentials: HTTPAuthorizationCredentials = Depends(security)): | |
| """ | |
| Returns the user ID from the token if valid. | |
| In development mode (DEV_ALLOW_DEFAULT_PATIENT) it falls back to DEFAULT_PATIENT_ID when no token provided. | |
| In production it will raise 401 for missing/invalid tokens. | |
| """ | |
| if not credentials: | |
| if DEV_ALLOW_DEFAULT_PATIENT and DEFAULT_PATIENT_ID: | |
| logger.info("No credentials provided; falling back to DEFAULT_PATIENT_ID (dev mode)") | |
| return DEFAULT_PATIENT_ID | |
| raise HTTPException(status_code=401, detail="Missing credentials") | |
| try: | |
| return get_current_user(credentials) | |
| except HTTPException as e: | |
| # In dev mode we may still fall back | |
| if DEV_ALLOW_DEFAULT_PATIENT and DEFAULT_PATIENT_ID: | |
| logger.info("Invalid credentials; falling back to DEFAULT_PATIENT_ID (dev mode)") | |
| return DEFAULT_PATIENT_ID | |
| raise e | |
| def create_dev_token(patient_id: str, expires_minutes: int = 60): | |
| """Create a development JWT signed with SUPABASE_JWT_SECRET for local dev/testing. | |
| The token uses `sub` to store the patient/user id and `aud` of 'authenticated' to match validation. | |
| """ | |
| jwt_secret = os.getenv("SUPABASE_JWT_SECRET") | |
| if not jwt_secret: | |
| raise RuntimeError("SUPABASE_JWT_SECRET not set; cannot create token") | |
| now = datetime.datetime.utcnow() | |
| payload = { | |
| "sub": patient_id, | |
| "iat": now, | |
| "exp": now + datetime.timedelta(minutes=expires_minutes), | |
| "aud": "authenticated", | |
| } | |
| token = jwt.encode(payload, jwt_secret, algorithm="HS256") | |
| return token | |