diff --git a/app/graphql/mutations.py b/app/graphql/mutations.py index f26f51c..a8be3ec 100644 --- a/app/graphql/mutations.py +++ b/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 + +from app.graphql.context import CustomContext +from auth.tokens import decode_token, create_jwt_token, create_refresh_token +from app.utils.format import get_image_url -@strawberry.type -class AuthResponse: - access_token: str - 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 这种涉及第三方重定向的流程, - # 依然保留现有的 RESTful (/login/google 和 /auth/callback) 会更简单。 - # 登录成功后重定向回前端,前端拿到 token,后续所有数据交互全部走 GraphQL。 \ No newline at end of file + @strawberry.mutation + async def refresh_session(self, info: Info[CustomContext, None]) -> str: + """ + 刷新 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" \ No newline at end of file