103 lines
4.0 KiB
Python
103 lines
4.0 KiB
Python
import os
|
|
import strawberry
|
|
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
|
|
|
|
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, 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"
|
|
|
|
@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" |