fix(auth):harden session cookie transport

This commit is contained in:
Anish Sarkar 2026-06-24 03:55:39 +05:30
parent 9b127a8533
commit fbecbb98b5
3 changed files with 53 additions and 13 deletions

View file

@ -3,13 +3,20 @@
from __future__ import annotations
from datetime import UTC, datetime, timedelta
from enum import Enum
from typing import Any
import jwt
from fastapi import Request, Response
from app.config import config
class TransportMode(Enum):
COOKIE = "cookie"
HEADER = "header"
def _cookie_secure(request: Request | None = None) -> bool:
policy = config.SESSION_COOKIE_SECURE_POLICY
if policy == "always":
@ -49,7 +56,7 @@ def _set_persistent_cookie(
def write_session(
response: Response,
access: str,
refresh: str,
refresh: str | None = None,
request: Request | None = None,
) -> None:
_set_persistent_cookie(
@ -59,13 +66,14 @@ def write_session(
max_age=config.ACCESS_TOKEN_LIFETIME_SECONDS,
request=request,
)
_set_persistent_cookie(
response,
key=config.REFRESH_COOKIE_NAME,
value=refresh,
max_age=config.REFRESH_TOKEN_LIFETIME_SECONDS,
request=request,
)
if refresh is not None:
_set_persistent_cookie(
response,
key=config.REFRESH_COOKIE_NAME,
value=refresh,
max_age=config.REFRESH_TOKEN_LIFETIME_SECONDS,
request=request,
)
def clear_session(response: Response, request: Request | None = None) -> None:
@ -80,10 +88,41 @@ def clear_session(response: Response, request: Request | None = None) -> None:
)
def read_refresh(request: Request, body: Any | None = None) -> str | None:
def read_refresh(request: Request, body: Any | None = None) -> tuple[str | None, TransportMode]:
cookie = request.cookies.get(config.REFRESH_COOKIE_NAME)
if cookie:
return cookie
return cookie, TransportMode.COOKIE
if body is None:
return None
return getattr(body, "refresh_token", None)
return None, TransportMode.HEADER
return getattr(body, "refresh_token", None), TransportMode.HEADER
def access_expires_at(access_token: str) -> int:
payload = jwt.decode(
access_token,
config.SECRET_KEY,
algorithms=["HS256"],
options={"verify_aud": False},
)
return int(payload["exp"])
def issue(
response: Response,
mode: TransportMode,
*,
access: str,
refresh: str | None,
access_expires_at: int,
request: Request | None = None,
) -> dict:
if mode is TransportMode.COOKIE:
write_session(response, access, refresh, request)
return {"authenticated": True, "access_expires_at": access_expires_at}
return {
"access_token": access,
"refresh_token": refresh,
"token_type": "bearer",
"access_expires_at": access_expires_at,
}