"""后台登录 Cookie 与滑动续期中间件。 续期只发生在「已经通过 ``get_current_user`` 的请求」上: - ``get_current_user`` 把当前用户和「本次认证用的 Cookie Token」写进 ``request.state``; - 响应完成后,中间件读取该 Token,把 ``exp`` 推到 ``now + 24h``(不超过 7 天); - 登录流程、登出、未认证请求、401 响应一律不续期。 """ from __future__ import annotations import os from datetime import UTC, datetime from fastapi import Request, Response from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint from .security import AccessToken, renew_access_token SESSION_COOKIE_NAME = "access_token" # 生产环境 HTTPS 部署时设置 COOKIE_SECURE=true _COOKIE_SECURE = os.getenv("COOKIE_SECURE", "false").lower() == "true" # 登出会删除 Cookie,不能再被中间件补回来。 _NO_RENEW_PATHS = frozenset({"/api/auth/logout"}) def set_session_cookie( response: Response, token: str, expires_at: datetime, *, now: datetime | None = None, ) -> None: """写入会话 Cookie;有效期跟随 JWT 的绝对过期时刻。""" current = now or datetime.now(UTC) max_age = int((expires_at - current).total_seconds()) if max_age <= 0: return response.set_cookie( key=SESSION_COOKIE_NAME, value=token, httponly=True, secure=_COOKIE_SECURE, samesite="lax", max_age=max_age, expires=expires_at, path="/", ) def clear_session_cookie(response: Response) -> None: response.delete_cookie(key=SESSION_COOKIE_NAME, path="/") def _should_renew(request: Request, status_code: int) -> bool: """只对「已认证的 API 请求 + 非 401 + 非登出」续期。""" path = request.url.path.rstrip("/") or "/" if status_code == 401: return False if not path.startswith("/api/"): return False if path in _NO_RENEW_PATHS: return False return getattr(request.state, "current_user", None) is not None class SessionRenewalMiddleware(BaseHTTPMiddleware): """滑动续期:活跃请求自动延长 24h,但永不突破 7 天绝对上限。""" async def dispatch( self, request: Request, call_next: RequestResponseEndpoint ) -> Response: response = await call_next(request) if not _should_renew(request, response.status_code): return response token = getattr(request.state, "auth_cookie_token", None) if not token: return response renewed: AccessToken | None = renew_access_token(token) if renewed is None: return response set_session_cookie(response, renewed.token, renewed.expires_at) return response