76 lines
2.9 KiB
Python
76 lines
2.9 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 api_key:
|
|
token = api_key
|
|
elif credentials:
|
|
token = credentials.credentials
|
|
|
|
# If no token is provided
|
|
if not token:
|
|
# Check if environment allows default dev token or requires strict 401
|
|
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"}
|
|
)
|
|
|
|
# Derive patient/user identity context from token or header
|
|
patient_id = x_patient_id or ("PAT-DEFAULT" if token == API_SECRET_KEY else token.replace("test_token_", ""))
|
|
return AuthContext(user_id=patient_id, role="patient", patient_id=patient_id)
|
|
|
|
|
|
# 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
|