Spaces:
Running
Running
File size: 4,902 Bytes
09801ca | 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 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 | """
🔐 API Dependencies - Authentication & Authorization
Enterprise-grade user isolation using JWT tokens
This is the SECURE way to get user identity:
- Extracts user_id from verified JWT token (not from request body)
- Falls back to guest ID for anonymous users
- Prevents users from accessing other users' data
"""
from fastapi import Header, HTTPException, Depends, Request
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from typing import Optional
import logging
import os
logger = logging.getLogger(__name__)
# Import the core auth module
try:
from core.auth import (
AuthenticatedUser,
get_current_user,
get_current_user_optional,
require_authenticated_user,
get_user_id_from_token,
get_user_id_from_body_deprecated,
decode_jwt_token
)
AUTH_AVAILABLE = True
except ImportError as e:
logger.warning(f"Core auth module not available: {e}")
AUTH_AVAILABLE = False
# Legacy compatibility function
async def get_current_user_id(
request: Request,
x_user_id: Optional[str] = Header(None, alias="X-User-ID"),
x_workspace_id: Optional[str] = Header(None, alias="X-Workspace-ID"),
authorization: Optional[str] = Header(None, alias="Authorization")
) -> str:
"""
🔐 Get verified user ID from JWT token or headers.
Priority:
1. JWT token in Authorization header (most secure)
2. X-User-ID header (for compatibility)
3. Generate guest ID (for anonymous users)
If X-Workspace-ID is provided, verifies membership and returns it.
"""
resolved_user_id = None
# Try JWT authentication first
if authorization and authorization.startswith("Bearer "):
token = authorization[7:]
# Check if it's a Developer API Token
if token.startswith("dv_live_") or token.startswith("dv_test_"):
try:
from database.db import AsyncSessionLocal
from database.orm import DeveloperAPIKey
from sqlalchemy import select
async with AsyncSessionLocal() as db:
result = await db.execute(select(DeveloperAPIKey).filter(DeveloperAPIKey.api_key == token))
key = result.scalars().first()
if key:
if key.status == 'revoked':
raise HTTPException(status_code=401, detail="API Key has been revoked")
key.total_calls += 1
await db.commit()
resolved_user_id = str(key.user_id)
except Exception as e:
logger.debug(f"Developer token decode failed: {e}")
# Standard JWT decode
else:
try:
if AUTH_AVAILABLE:
payload = decode_jwt_token(token)
user_id = payload.get("sub")
if user_id:
resolved_user_id = user_id
except Exception as e:
logger.debug(f"JWT decode failed: {e}")
# Fallback to X-User-ID header
if not resolved_user_id and x_user_id and x_user_id not in ["null", "undefined", "", "default"]:
resolved_user_id = x_user_id
# Generate guest ID based on request fingerprint
if not resolved_user_id:
import hashlib
ip = request.client.host if request.client else "unknown"
ua = request.headers.get("User-Agent", "unknown")[:100]
fingerprint = hashlib.sha256(f"{ip}:{ua}".encode()).hexdigest()[:12]
resolved_user_id = f"guest_{fingerprint}"
import uuid as _uuid
try:
_uuid.UUID(resolved_user_id)
except ValueError:
resolved_user_id = str(_uuid.uuid5(_uuid.NAMESPACE_OID, resolved_user_id))
return resolved_user_id
def get_verified_user_id(
body_user_id: Optional[str],
header_user_id: str
) -> str:
"""
Get the verified user ID, preferring header over body.
For migration: accepts body user_id but logs warning.
Eventually body_user_id support should be removed.
"""
# Always prefer header (verified) over body (unverified)
if header_user_id and not header_user_id.startswith("guest_"):
return header_user_id
# For guests, use body if provided (for backward compat)
if body_user_id and body_user_id not in ["default", "guest", "", "null", "undefined"]:
logger.warning(f"⚠️ DEPRECATED: user_id from body '{body_user_id}' - migrate to JWT auth")
return body_user_id
return header_user_id
# Export all auth utilities
__all__ = [
'get_current_user_id',
'get_verified_user_id',
'AuthenticatedUser',
'get_current_user',
'get_current_user_optional',
'require_authenticated_user',
'get_user_id_from_token',
]
|