4fbcff2bb4
- 修改get_current_user函数同时支持Authorization header和cookie获取token - 在logout接口中添加JSONResponse并清除webview cookie登录态 - 新增webview模块实现URL token到cookie的转换和安全重定向功能 - 实现域名白名单校验防止开放重定向漏洞 - 在main.py中注册webview路由
126 lines
4.0 KiB
Python
126 lines
4.0 KiB
Python
import logging
|
||
import uuid
|
||
from datetime import datetime, timedelta
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException, Request
|
||
from pydantic import BaseModel
|
||
|
||
from .. import config, db, security, wechat
|
||
from ..user import user_payload
|
||
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
router = APIRouter(prefix="/api/v1/auth", tags=["auth"])
|
||
|
||
|
||
class WeChatLoginRequest(BaseModel):
|
||
code: str
|
||
|
||
|
||
class BindPhoneRequest(BaseModel):
|
||
code: str
|
||
phone_code: str
|
||
|
||
|
||
def get_current_user(request: Request):
|
||
token = ""
|
||
|
||
# 1. 优先从 Authorization header 取（小程序 API 调用方式）
|
||
auth = request.headers.get("Authorization", "")
|
||
if auth.startswith("Bearer "):
|
||
token = auth[7:]
|
||
|
||
# 2. header 没有则从 cookie 取（webview H5 页面内请求方式）
|
||
if not token:
|
||
token = request.cookies.get("auth_token", "")
|
||
|
||
if not token:
|
||
raise HTTPException(status_code=401, detail="缺少 token")
|
||
|
||
session = db.fetch_one(
|
||
"SELECT s.user_id AS id FROM auth_sessions s "
|
||
"WHERE s.token=%s AND s.expires_at>NOW()",
|
||
(token,),
|
||
)
|
||
if not session:
|
||
raise HTTPException(status_code=401, detail="token 无效或过期")
|
||
user = db.fetch_one("SELECT * FROM auth_users WHERE id=%s", (session["id"],))
|
||
if not user:
|
||
raise HTTPException(status_code=401, detail="用户不存在")
|
||
return user
|
||
|
||
|
||
def _record_session(user_id):
|
||
"""创建会话 token 并写入 auth_sessions。"""
|
||
token = uuid.uuid4().hex + uuid.uuid4().hex[:16]
|
||
expires = datetime.now() + timedelta(seconds=config.JWT_TTL_SECONDS)
|
||
db.execute(
|
||
"INSERT INTO auth_sessions (user_id, token, expires_at) VALUES (%s, %s, %s)",
|
||
(user_id, token, expires),
|
||
)
|
||
return token
|
||
|
||
|
||
@router.post("/wechat-login")
|
||
def wechat_login(req: WeChatLoginRequest):
|
||
try:
|
||
openid = wechat.code_to_openid(req.code)
|
||
except wechat.WeChatError as exc:
|
||
logger.warning("微信登录 code 换取 openid 失败: %s", exc)
|
||
raise HTTPException(status_code=400, detail=str(exc))
|
||
|
||
user = db.fetch_one("SELECT * FROM auth_users WHERE wechat_openid=%s", (openid,))
|
||
if user:
|
||
token = _record_session(user["id"])
|
||
return {"success": True, "is_new": False, "token": token, **user_payload(user)}
|
||
|
||
return {"success": False, "need_phone": True, "error": "需要绑定手机号"}
|
||
|
||
|
||
@router.post("/bind-phone")
|
||
def bind_phone(req: BindPhoneRequest):
|
||
try:
|
||
openid = wechat.code_to_openid(req.code)
|
||
phone = wechat.get_phone_number(req.phone_code)
|
||
except wechat.WeChatError as exc:
|
||
logger.warning("绑定手机号时获取微信信息失败: %s", exc)
|
||
raise HTTPException(status_code=400, detail=str(exc))
|
||
|
||
if not phone:
|
||
raise HTTPException(status_code=400, detail="获取手机号失败")
|
||
|
||
encrypted = security.encrypt_phone(phone)
|
||
user = db.fetch_one("SELECT * FROM auth_users WHERE phone_encrypted=%s", (encrypted,))
|
||
if not user:
|
||
return {"success": False, "error": "手机号未匹配，无法登录"}
|
||
|
||
db.execute("UPDATE auth_users SET wechat_openid=%s WHERE id=%s", (openid, user["id"]))
|
||
user["wechat_openid"] = openid
|
||
token = _record_session(user["id"])
|
||
return {"success": True, "is_new": True, "token": token, **user_payload(user)}
|
||
|
||
|
||
@router.get("/verify")
|
||
def verify(user=Depends(get_current_user)):
|
||
return {"success": True, **user_payload(user)}
|
||
|
||
|
||
@router.post("/logout")
|
||
def logout(request: Request, user=Depends(get_current_user)):
|
||
from fastapi.responses import JSONResponse
|
||
|
||
auth = request.headers.get("Authorization", "")
|
||
if auth.startswith("Bearer "):
|
||
db.execute("DELETE FROM auth_sessions WHERE token=%s", (auth[7:],))
|
||
# 退出登录时解除该用户手机号绑定的微信 openid，下次需重新授权手机号绑定
|
||
db.execute(
|
||
"UPDATE auth_users SET wechat_openid=NULL WHERE id=%s",
|
||
(user["id"],),
|
||
)
|
||
resp = JSONResponse(content={"success": True})
|
||
# 同时清除 webview cookie 登录态
|
||
resp.delete_cookie(key="auth_token", path="/")
|
||
return resp
|
||
|