fix(security): resolve 403 Forbidden on voice turn - normalize patient ID matching with PAT-prefix support - Bearer token now prioritized over X-API-Key for identity resolution - Master API key gets admin role for service-to-service bypass - Centralized verify_session_ownership helper across all IDOR checks - Files: security.py, voice_onboarding.py, onboarding.py, documents.py

ai-patient-onboarding/app/api/documents.py

@@ -8,7 +8,7 @@ from app.schemas.documents import DocumentExtractionResponse
 from app.services.onboarding_service import OnboardingService
 
 from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile
-from app.core.security import get_auth_context, AuthContext, check_rate_limit
+from app.core.security import get_auth_context, AuthContext, check_rate_limit, verify_session_ownership
 
 router = APIRouter(
     prefix="/v1/documents",
@@ -38,11 +38,7 @@ async def upload_medical_document(
         raise HTTPException(status_code=404, detail="Onboarding session not found")
 
     # 1b. Verify Session Ownership (IDOR Prevention)
-    if session.patient_id != auth.patient_id and auth.role != "admin":
-        raise HTTPException(
-            status_code=403,
-            detail="Forbidden: You do not have authorization to upload documents to this session.",
-        )
+    verify_session_ownership(session.patient_id, auth, detail="Forbidden: You do not have authorization to upload documents to this session.")
 
     # Enforce MIME type and File Size Limits (10MB MAX)
     allowed_mimes = ["image/jpeg", "image/png", "image/webp", "application/pdf"]

ai-patient-onboarding/app/api/onboarding.py

@@ -39,7 +39,7 @@ from app.services.voice_service import VoiceService
 
 from fastapi import APIRouter, Depends, HTTPException, Request
 from sqlalchemy.orm import Session
-from app.core.security import get_auth_context, AuthContext, check_rate_limit
+from app.core.security import get_auth_context, AuthContext, check_rate_limit, verify_session_ownership
 
 router = APIRouter(
     prefix="/v1/onboarding",
@@ -147,11 +147,7 @@ async def send_message(
         )
 
     # 1b. Verify Session Ownership (IDOR Prevention)
-    if session.patient_id != auth.patient_id and auth.role != "admin":
-        raise HTTPException(
-            status_code=403,
-            detail="Forbidden: You do not have authorization to access or modify this patient onboarding session.",
-        )
+    verify_session_ownership(session.patient_id, auth, detail="Forbidden: You do not have authorization to access or modify this patient onboarding session.")
 
     # 2. Check session
     if session.status != "in_progress":
@@ -298,11 +294,7 @@ async def complete_onboarding(
         )
 
     # 1b. Verify Session Ownership (IDOR Prevention)
-    if session.patient_id != auth.patient_id and auth.role != "admin":
-        raise HTTPException(
-            status_code=403,
-            detail="Forbidden: You do not have authorization to complete this onboarding session.",
-        )
+    verify_session_ownership(session.patient_id, auth, detail="Forbidden: You do not have authorization to complete this onboarding session.")
 
     # 2. Load state
     state = onboarding_service.get_state(

ai-patient-onboarding/app/api/voice_onboarding.py

@@ -32,7 +32,7 @@ from app.services.patient_submission_service import PatientSubmissionService
 from app.services.result_extraction_service import ResultExtractionService
 from app.services.voice_service import VoiceService
 
-from app.core.security import get_auth_context, AuthContext, check_rate_limit
+from app.core.security import get_auth_context, AuthContext, check_rate_limit, verify_session_ownership
 
 router = APIRouter(
     prefix="/v1/voice-onboarding",
@@ -159,11 +159,7 @@ async def voice_turn(
         )
 
     # 1b. IDOR Session Ownership Check
-    if session.patient_id != auth.patient_id and auth.role != "admin":
-        raise HTTPException(
-            status_code=403,
-            detail="Forbidden: You do not have authorization to access or modify this voice session.",
-        )
+    verify_session_ownership(session.patient_id, auth, detail="Forbidden: You do not have authorization to access or modify this voice session.")
 
     if session.status == "completed":
         session.status = "in_progress"
@@ -393,11 +389,7 @@ async def complete_voice_onboarding(
         )
 
     # 1b. Verify Session Ownership (IDOR Prevention)
-    if session.patient_id != auth.patient_id and auth.role != "admin":
-        raise HTTPException(
-            status_code=403,
-            detail="Forbidden: You do not have authorization to complete this voice session.",
-        )
+    verify_session_ownership(session.patient_id, auth, detail="Forbidden: You do not have authorization to complete this voice session.")
 
     state = onboarding_service.get_state(db, session_id)
 

ai-patient-onboarding/app/core/security.py

@@ -29,14 +29,13 @@ def get_auth_context(
     In development / testing mode, allows test tokens while enforcing identity isolation.
     """
     token = None
-    if api_key:
-        token = api_key
-    elif credentials:
+    if credentials and credentials.credentials:
         token = credentials.credentials
+    elif api_key:
+        token = api_key
 
     # 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.",
@@ -51,9 +50,37 @@ def get_auth_context(
             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)
+    # 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