| 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_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: |
| |
| 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: |
| |
| 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 |
|
|