File size: 3,327 Bytes
76962bf b1198f0 76962bf b1198f0 76962bf b1198f0 76962bf b1198f0 76962bf b1198f0 76962bf b1198f0 76962bf b1198f0 76962bf b1198f0 76962bf b1198f0 76962bf b1198f0 | 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 81 82 83 84 85 | 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
|