File size: 3,121 Bytes
a5ddf46
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
"""
Authentication helper for validating Better-Auth sessions.

Provides utilities for session validation and user extraction.
"""

import httpx
import logging
from typing import Optional, Dict, Any
from fastapi import Request
from config import Config

logger = logging.getLogger(__name__)


async def validate_session(request: Request) -> Optional[str]:
    """
    Extract and validate session from request cookies.

    Calls the Better-Auth backend to validate the session and returns
    the user ID if valid.

    Args:
        request: FastAPI Request object

    Returns:
        User ID string if session is valid, None otherwise
    """
    # Extract session cookie
    session_token = request.cookies.get("better-auth.session_token")

    if not session_token:
        logger.debug("No session cookie found")
        return None

    try:
        # Call auth backend to validate session
        async with httpx.AsyncClient(timeout=5.0) as client:
            response = await client.get(
                f"{Config.AUTH_BACKEND_URL}/api/auth/get-session",
                cookies={"better-auth.session_token": session_token}
            )

            if response.status_code != 200:
                logger.debug(f"Session validation failed: {response.status_code}")
                return None

            data = response.json()
            user_id = data.get("user", {}).get("id")

            if not user_id:
                logger.debug("No user ID in session response")
                return None

            return user_id

    except httpx.TimeoutException:
        logger.warning("Session validation timed out")
        return None
    except httpx.RequestError as e:
        logger.error(f"Session validation request failed: {e}")
        return None
    except Exception as e:
        logger.error(f"Unexpected error validating session: {e}")
        return None


async def get_session_data(request: Request) -> Optional[Dict[str, Any]]:
    """
    Get full session data from the auth backend.

    Args:
        request: FastAPI Request object

    Returns:
        Session data dictionary if valid, None otherwise
    """
    session_token = request.cookies.get("better-auth.session_token")

    if not session_token:
        return None

    try:
        async with httpx.AsyncClient(timeout=5.0) as client:
            response = await client.get(
                f"{Config.AUTH_BACKEND_URL}/api/auth/get-session",
                cookies={"better-auth.session_token": session_token}
            )

            if response.status_code != 200:
                return None

            return response.json()

    except Exception as e:
        logger.error(f"Error getting session data: {e}")
        return None


def extract_user_id_from_cookies(request: Request) -> Optional[str]:
    """
    Quick check for session cookie presence without validation.

    Useful for early bailout when no cookie exists.

    Args:
        request: FastAPI Request object

    Returns:
        Session token if present, None otherwise
    """
    return request.cookies.get("better-auth.session_token")