103 lines
3.8 KiB
Python
103 lines
3.8 KiB
Python
import os
|
|
import time
|
|
from typing import Optional, Dict
|
|
from fastapi import Header, HTTPException, Security, Depends
|
|
from fastapi.security import APIKeyHeader, HTTPBearer, HTTPAuthorizationCredentials
|
|
from app.config import APP_ENV
|
|
|
|
# Configurable API Secret Key for Service-to-Service and Client Auth
|
|
API_KEY_NAME = "X-API-Key"
|
|
API_SECRET_KEY = os.getenv("API_SECRET_KEY", "skyheal_secure_api_key_2026")
|
|
|
|
api_key_header = APIKeyHeader(name=API_KEY_NAME, auto_error=False)
|
|
security_bearer = HTTPBearer(auto_error=False)
|
|
|
|
class AuthContext:
|
|
def __init__(self, user_id: str, role: str = "patient", patient_id: Optional[str] = None):
|
|
self.user_id = user_id
|
|
self.role = role
|
|
self.patient_id = patient_id or user_id
|
|
|
|
def get_auth_context(
|
|
api_key: Optional[str] = Security(api_key_header),
|
|
credentials: Optional[HTTPAuthorizationCredentials] = Security(security_bearer),
|
|
x_patient_id: Optional[str] = Header(None, alias="X-Patient-ID")
|
|
) -> AuthContext:
|
|
"""
|
|
Service Authentication & Authorization Dependency.
|
|
Validates API key header (X-API-Key) or Bearer token.
|
|
In development / testing mode, allows test tokens while enforcing identity isolation.
|
|
"""
|
|
token = None
|
|
if credentials and credentials.credentials:
|
|
token = credentials.credentials
|
|
elif api_key:
|
|
token = api_key
|
|
|
|
# If no token is provided
|
|
if not token:
|
|
raise HTTPException(
|
|
status_code=401,
|
|
detail="Authentication credentials required. Please provide X-API-Key or Bearer token.",
|
|
headers={"WWW-Authenticate": "Bearer"}
|
|
)
|
|
|
|
# Validate secret token or bearer token
|
|
if token != API_SECRET_KEY and not token.startswith("test_token_"):
|
|
raise HTTPException(
|
|
status_code=401,
|
|
detail="Invalid authentication token or API key.",
|
|
headers={"WWW-Authenticate": "Bearer"}
|
|
)
|
|
|
|
# Master API Secret Key has admin/service privileges
|
|
if token == API_SECRET_KEY:
|
|
patient_id = x_patient_id or "PAT-DEFAULT"
|
|
return AuthContext(user_id=patient_id, role="admin", patient_id=patient_id)
|
|
else:
|
|
patient_id = x_patient_id or token.replace("test_token_", "")
|
|
return AuthContext(user_id=patient_id, role="patient", patient_id=patient_id)
|
|
|
|
|
|
def verify_session_ownership(session_patient_id: Optional[str], auth: AuthContext, detail: str = "Forbidden: You do not have authorization to access or modify this onboarding session."):
|
|
"""
|
|
Verifies session ownership by comparing normalized patient IDs.
|
|
Master API key callers (role='admin') are always authorized.
|
|
"""
|
|
if auth.role == "admin":
|
|
return
|
|
|
|
s_id = (session_patient_id or "").strip()
|
|
a_id = (auth.patient_id or "").strip()
|
|
|
|
if s_id == a_id:
|
|
return
|
|
|
|
# Normalize "PAT-" prefixes for matching (e.g. "PAT-417" vs "417")
|
|
s_norm = s_id[4:] if s_id.startswith("PAT-") else s_id
|
|
a_norm = a_id[4:] if a_id.startswith("PAT-") else a_id
|
|
|
|
if s_norm == a_norm:
|
|
return
|
|
|
|
raise HTTPException(status_code=403, detail=detail)
|
|
|
|
|
|
# Simple In-Memory Token Bucket Rate Limiter
|
|
RATE_LIMIT_STORE: Dict[str, list] = {}
|
|
|
|
def check_rate_limit(client_ip: str, limit: int = 30, window_sec: int = 60):
|
|
now = time.time()
|
|
timestamps = RATE_LIMIT_STORE.get(client_ip, [])
|
|
# Filter timestamps within window
|
|
timestamps = [t for t in timestamps if now - t < window_sec]
|
|
if len(timestamps) >= limit:
|
|
retry_after = int(window_sec - (now - timestamps[0]))
|
|
raise HTTPException(
|
|
status_code=429,
|
|
detail=f"Rate limit exceeded ({limit} requests per {window_sec}s). Please try again later.",
|
|
headers={"Retry-After": str(max(1, retry_after))}
|
|
)
|
|
timestamps.append(now)
|
|
RATE_LIMIT_STORE[client_ip] = timestamps
|