feat(graphql): implement secure session refresh mutation with token rotation

- Rename get_current_user to get_current_user_id in auth/deps.py to clarify intent
- Remove legacy /me, /logout routes, and placeholder function from auth/__init__.py
- Remove dependency_overrides from app/main.py
- Update CustomContext to resolve and cache full User objects via db query

.cliprepo.yaml

@@ -6,21 +6,27 @@ prompts:
 
 preset:
   auth:
-    - "app/auth.py"
+    - "auth/*"
-    - "app/auth/*"
 
   s3:
     - "app/models/base.py"
     - "app/models/user.py"
     - "app/utils/s3.py"
-    - "app/auth.py"
+    - "auth.py"
 
   db_model:
     - "app/models/*"
     - "app/db.py"
-    # - "app/auth.py"
     - "app/main.py"
 
   graphql:
     - "app/graphql/*"
     - "app/main.py"
+
+  get_current_user:
+    - "app/main.py"
+    - "auth/__init__.py"
+    - "auth/deps.py"
+    - "auth/tokens.py"
+    - "auth/oauth/google.py"
+    - "app/graphql/*"

app/graphql/context.py

@@ -1,9 +1,9 @@
 from app.models.user import User
-from auth.deps import get_current_user
+from auth.deps import get_current_user_id
 from fastapi import Request
 from strawberry.fastapi import BaseContext
 from strawberry.dataloader import DataLoader
-from typing import List
+from typing import List, Optional
 from collections import defaultdict
 from app.db import SessionLocal
 from app.models import WorkspaceMember
@@ -31,13 +31,22 @@ class CustomContext(BaseContext):
         super().__init__(request)
         self.db = SessionLocal()
         self.request = request
+        self._current_user: Optional[User] = None
         
         self.members_loader = DataLoader(load_fn=load_members_by_workspace_ids)
 
     async def get_current_user(self) -> User | None:
-        # 复用现有的 auth 逻辑
+        """在 Context 级别负责把 user_id 变成真正的数据库 User 对象"""
+        if self._current_user is not None:
+            return self._current_user
+            
         try:
-            return await get_current_user(self.request)
+            # 借用 auth 包的纯工具解出 ID
+            user_id = get_current_user_id(self.request)
+            # 在 Context 的数据库连接中查询完整用户
+            user = self.db.query(User).filter(User.id == user_id).first()
+            self._current_user = user
+            return user
         except Exception:
             return None
 

app/graphql/mutations.py

@@ -1,18 +1,103 @@
+import os
 import strawberry
-from auth.tokens import create_jwt_token, create_refresh_token
+from strawberry.types import Info
+from dotenv import load_dotenv
+from urllib.parse import quote
+import json
         
-@strawberry.type
+from app.graphql.context import CustomContext
-class AuthResponse:
+from auth.tokens import decode_token, create_jwt_token, create_refresh_token
-    access_token: str
+from app.utils.format import get_image_url
-    refresh_token: str
+
-    token_type: str
+load_dotenv()
+
+ENV_MODE = os.getenv("ENV_MODE")
+IS_PROD = ENV_MODE == "prod"
+ACCESS_TOKEN_EXPIRE_MINUTES = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", "60"))
+ACCESS_TOKEN_MAX_AGE = ACCESS_TOKEN_EXPIRE_MINUTES * 60
 
 @strawberry.type
 class Mutation:
     @strawberry.mutation
-    def logout(self) -> str:
+    def logout(self, info: Info[CustomContext, None]) -> str:
+        """退出登录：清除浏览器的 HttpOnly Cookies"""
+        response = info.context.response
+        
+        # 让浏览器立刻过期这些 cookie
+        response.delete_cookie("access_token")
+        response.delete_cookie("refresh_token")
+        response.delete_cookie("user_info")
+        
         return "ok"
     
-    # 提示：像 Google OAuth 这种涉及第三方重定向的流程，
+    @strawberry.mutation
-    # 依然保留现有的 RESTful (/login/google 和 /auth/callback) 会更简单。
+    async def refresh_session(self, info: Info[CustomContext, None]) -> str:
-    # 登录成功后重定向回前端，前端拿到 token，后续所有数据交互全部走 GraphQL。
+        """
+        刷新 Session：如果 access_token 过期了，前端可以调用这个接口用 refresh_token 换新的 access_token。
+        同时如果 refresh_token 已经过半寿命了，这个接口也会顺便轮转一个新的 refresh_token，确保用户体验的连续性。
+        """
+        request = info.context.request
+        response = info.context.response
+        
+        # 从 Cookie 获取并校验 refresh_token
+        refresh_token = request.cookies.get("refresh_token")
+        if not refresh_token:
+            raise Exception("No refresh token found in cookies")
+            
+        try:
+            payload = decode_token(refresh_token)
+        except Exception:
+            raise Exception("Invalid or expired refresh token")
+            
+        if payload.get("type") != "refresh":
+            raise Exception("Token type mismatch")
+
+        user = await info.context.get_current_user()
+        
+        # 如果 access_token 已经彻底过期导致 Context 没捞到用户，
+        # 我们用 refresh_token 里的 sub 去 Context 的 db 里再查一遍用户
+        from app.models.user import User
+        if not user:
+            user_id = payload.get("sub")
+            user = info.context.db.query(User).filter(User.id == user_id).first()
+        if not user:
+            raise Exception("User no longer exists")
+            
+        # 始终生成并写入新的短效 access_token
+        token_data = {"sub": str(user.id), "email": user.email}
+        new_access = create_jwt_token(token_data)
+        
+        ACCESS_TOKEN_MAX_AGE = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", "60")) * 60
+        response.set_cookie(
+            key="access_token", value=new_access, httponly=True,
+            max_age=ACCESS_TOKEN_MAX_AGE, samesite="lax", secure=IS_PROD,
+        )
+        
+        # 检查 refresh_token 是否已经消耗了超过一半的生命周期
+        import time
+        exp = payload.get("exp")  # 过期绝对秒数
+        iat = payload.get("iat")  # 签发绝对秒数
+        now = int(time.time())
+        
+        # 如果当前时间已经过了总寿命的一半，就触发轮转，下发新的 refresh_token
+        if (now - iat) > (exp - iat) / 2:
+            new_refresh = create_refresh_token(token_data)
+            REFRESH_TOKEN_MAX_AGE = int(os.getenv("REFRESH_TOKEN_EXPIRE_DAYS", "7")) * 24 * 60 * 60
+            response.set_cookie(
+                key="refresh_token", value=new_refresh, httponly=True,
+                max_age=REFRESH_TOKEN_MAX_AGE, samesite="lax", secure=IS_PROD,
+            )
+
+        # 同步写入并续期前端所需的 user_info        
+        response.set_cookie(
+            key="user_info",
+            value=quote(json.dumps({
+                "email": user.email,
+                "name": user.name,
+                "avatar": get_image_url(user.avatar) if user.avatar else None
+            })),
+            max_age=ACCESS_TOKEN_MAX_AGE,
+            httponly=False
+        )
+        
+        return "ok"

app/main.py

@@ -5,7 +5,7 @@ from strawberry import Schema
 from strawberry.fastapi import GraphQLRouter
 from dotenv import load_dotenv
 
-from auth import router as auth_router, get_user_by_id, get_current_user
+from auth import router as auth_router
 
 from app.db import engine, SessionLocal
 from app.models import Base, User
@@ -46,23 +46,7 @@ graphql_router = GraphQLRouter(schema, context_getter=get_context)
 app.include_router(auth_router)                         # 保留现有的 auth 路由供 Google OAuth 回调使用
 app.include_router(graphql_router, prefix="/graphql")   # 挂载 GraphQL 终点，前端以后只需要请求 /graphql
 
-def override_get_user_by_id(user_id: str = Depends(get_current_user)):
+# 打印路由测试
-    db = SessionLocal()
-    try:
-        user = db.query(User).filter(User.id == user_id).first()
-        if not user:
-            raise HTTPException(status_code=404, detail="User not found")
-        return {
-            "email": user.email,
-            "name": user.name,
-            "avatar": get_image_url(user.avatar),
-        }
-    finally:
-        db.close()
-        
-app.dependency_overrides[get_user_by_id] = override_get_user_by_id
-
-# test code
 for route in app.routes:
     methods = getattr(route, "methods", "N/A")
     print(f"Path: {route.path} | Name: {route.name} | Methods: {methods}")

auth/__init__.py

@@ -1,53 +1,18 @@
 import os
-from fastapi import APIRouter, Depends, HTTPException, Request
+from fastapi import APIRouter
 
-from .deps import get_current_user
+from .deps import get_current_user_id
-from .tokens import create_jwt_token, create_refresh_token, verify_access_token, decode_token
+from .tokens import create_jwt_token, create_refresh_token, verify_access_token
 from .oauth import oauth_router
 
-# 创建全局 auth 路由器
+# 创建全局 auth 路由器并挂载第三方 OAuth 的路由器
 router = APIRouter(tags=["Authentication"])
-
-# 挂载第三方 OAuth 的路由器
 router.include_router(oauth_router)
 
-# 定义一个动态获取当前完整用户对象的依赖项
-# 这样 auth 包不需要知道 User 模型长什么样
-# 也不需要知道 SessionLocal 是什么
-# 全部由外部动态提供
-def get_user_by_id(user_id: str = Depends(get_current_user)):
-    """
-    这是一个占位/契约函数。
-    我们在 main.py 中可以通过 app.dependency_overrides 或者直接在使用时，
-    把具体的数据库查询逻辑绑定到这里。
-    """
-    raise NotImplementedError("Please override or inject the implementation of this function in the external `main.py` file.")
-
-@router.get("/me")
-def me(current_user = Depends(get_user_by_id)):
-    """
-    current_user 是由外部解析并传进来的完整用户对象（或者是字典）
-    """
-    return current_user
-    
-@router.post("/refresh")
-def refresh_token(refresh_token: str):
-    payload = decode_token(refresh_token)
-    if payload.get("type") != "refresh":
-        raise HTTPException(status_code=401)
-    user_id = payload.get("sub")
-    new_access = create_jwt_token({"sub": user_id})
-    return {"access_token": new_access}
-
-@router.post("/logout")
-def logout():
-    return {"message": "ok"}
-
 # 显式声明暴露的接口
 __all__ = [
     "router",
-    "get_current_user",
+    "get_current_user_id",
-    "get_user_by_id",
     "create_jwt_token",
     "create_refresh_token",
     "verify_access_token",

auth/deps.py

@@ -9,7 +9,8 @@ load_dotenv()
 JWT_SECRET = os.getenv("JWT_SECRET")
 JWT_ALGORITHM = os.getenv("JWT_ALGORITHM", "HS256")
 
-def get_current_user(request: Request):
+def get_current_user_id(request: Request):
+    """仅从 Cookie 中提取并解密出用户的 user_id (sub)"""
     token = request.cookies.get("access_token")
 
     if not token:

auth/oauth/google.py

@@ -9,7 +9,6 @@ import json
 from urllib.parse import quote
 from dotenv import load_dotenv
 
-from auth.deps import get_current_user
 from app.db import SessionLocal
 from app.models import User
 from auth.tokens import create_jwt_token, create_refresh_token, decode_token
@@ -214,28 +213,3 @@ async def auth_callback(request: Request, code: str, state: str, background_task
     response.delete_cookie("oauth_state")
     
     return response
-
-# @router.get("/me")
-# def me(user = Depends(get_current_user)):
-#     return {
-#         "email": user.email,
-#         "name": user.name,
-#         "avatar": get_image_url(user.avatar),
-#     }
-    
-# @router.post("/refresh")
-# def refresh_token(refresh_token: str):
-#     payload = decode_token(refresh_token)
-
-#     if payload.get("type") != "refresh":
-#         raise HTTPException(status_code=401)
-
-#     user_id = payload.get("sub")
-
-#     new_access = create_jwt_token({"sub": user_id})
-
-#     return {"access_token": new_access}
-
-# @router.post("/logout")
-# def logout():
-#     return {"message": "ok"}