Compare commits

..

17 Commits

Author SHA1 Message Date
guke 433555da3c chore(alembic): 合并 pages_override 与领券成功率两个迁移 head,修复 upgrade head(单数)多 head 报错
pull main 带入 #126 的 admin_user_pages_override,与本分支领券成功率的
coupon_session_platform_success → coupon_claim_app_env 都从 admin_user_plain_password
分叉,形成两个 alembic head。多 head 会让部署/run.sh 用的 `alembic upgrade head`(单数)
报 "Multiple head revisions are present" 而中断。

本迁移为纯 merge:upgrade/downgrade 均空,不含任何表结构或数据改动,仅把两个 head
收敛为单 head(merge_pages_override_coupon_slot),使 `alembic upgrade head` 恢复正常。
2026-07-16 17:36:06 +08:00
guke 1b729f7098 Merge branch 'main' of https://gitea.shaguabijia.com/WonderableAI/shaguabijia-app-server into feat/wechat-login
# Conflicts:
#	app/api/v1/auth.py
2026-07-16 14:53:10 +08:00
zhuzihao e9fd51d119 fix(auth): 发码防刷改为按成功计数 + 新增每设备每日发码上限 (#136)
- 修复:发码限流原为原子「判+记」,被单号 60s 冷却挡下的重发也占设备额度
  → 正常用户连点重发可能被误锁 1 小时。改为「先判后记、只对成功发码计数」:
  check 判在真发之前(超限直接 429、不真发),record 只在 send_code 成功后调;
  被单号冷却 / 供应商失败抛 429 时直接返回、不计数。
- 新增:同一设备(device_id)+ IP 每天最多 20 次发码上限,与原每小时 5 次两道闸并存,
  均按成功计数,叠一层日封顶挡低频长时间轰炸。
- 基建:ratelimit.py 新增 RateLimitRule + check_rate_limits / record_rate_limits
  (peek/commit 拆分);原子的 enforce_rate_limit 仍保留给登录爆破(失败也计)不变。
- 测试:补 2 个用例(冷却挡下不占额度 / 每日上限)。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

---------

Co-authored-by: zzhyyyyy <2685922758@qq.com>
Reviewed-on: #136
Co-authored-by: zhuzihao <zhuzihao@wonderable.ai>
Co-committed-by: zhuzihao <zhuzihao@wonderable.ai>
2026-07-16 09:40:56 +08:00
guke 7a8bd59517 feat(auth): M3 §10 绑微信按默认替换展示昵称头像(共享 helper 接两路径)
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-15 09:35:07 +08:00
guke 1567a9ac74 feat(auth): M2 POST /wechat/conflict/rebind(换绑=注销X重建Y,单事务)
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-14 17:46:58 +08:00
guke 80652ca78d feat(auth): M2 POST /wechat/conflict/continue(继续绑定=登录老账号)
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-14 17:40:22 +08:00
guke f411e81b07 feat(auth): M2 占用响应加 conflict_ticket/has_wechat/rebind_available
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-14 17:34:11 +08:00
guke f8b5d31d2e feat(auth): M2 conflict_ticket 令牌(编码已验证 openid+phone)
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-14 17:21:52 +08:00
guke af615a9853 feat(auth): M2 phone_rebind_log 换绑台账模型 + 配置 + 迁移
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-14 17:10:28 +08:00
guke 0177d467bf Merge branch 'main' of https://gitea.shaguabijia.com/WonderableAI/shaguabijia-app-server into feat/wechat-login 2026-07-14 10:52:09 +08:00
guke 3146224944 test(auth): 补 jverify 绑号 401/502 错误分支测试
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-13 15:31:46 +08:00
guke 3621d606ad feat(auth): POST /wechat/bind-phone/jverify(极光本机号绑号)
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-13 15:27:00 +08:00
guke c1b00eca0c test(auth): 补 phone_occupied 的 avatar_url 断言(锁定 M2 冲突页契约)
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-13 15:23:22 +08:00
guke a5fdcb53d8 feat(auth): POST /wechat/bind-phone/sms(短信绑号:建号/占用)
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-13 15:14:42 +08:00
guke fe9b749154 style(auth): 微信登录 import 去前向未用 + isort 排序
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-13 15:10:22 +08:00
guke 88313cefdf feat(auth): POST /wechat-login(openid 命中即登入,否则发绑号令牌)
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-13 14:57:36 +08:00
guke 2a82b20365 feat(auth): 微信登录 bind_ticket 令牌 + 配置
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-13 14:40:47 +08:00
25 changed files with 1425 additions and 591 deletions
+32
View File
@@ -0,0 +1,32 @@
"""phone_rebind_log 表(M2 换绑 30 天限制台账)
Revision ID: phone_rebind_log
Revises: comparison_llm_cost
"""
from alembic import op
import sqlalchemy as sa
revision = "phone_rebind_log"
down_revision = "comparison_llm_cost"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.create_table(
"phone_rebind_log",
sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
sa.Column("phone", sa.String(length=20), nullable=False),
sa.Column("old_user_id", sa.Integer(), nullable=True),
sa.Column("new_user_id", sa.Integer(), nullable=False),
sa.Column("source", sa.String(length=32), nullable=False, server_default="wechat_conflict"),
sa.Column("rebound_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
)
op.create_index("ix_phone_rebind_log_phone", "phone_rebind_log", ["phone"])
op.create_index("ix_phone_rebind_log_rebound_at", "phone_rebind_log", ["rebound_at"])
def downgrade() -> None:
op.drop_index("ix_phone_rebind_log_rebound_at", table_name="phone_rebind_log")
op.drop_index("ix_phone_rebind_log_phone", table_name="phone_rebind_log")
op.drop_table("phone_rebind_log")
+299 -16
View File
@@ -12,19 +12,36 @@ from __future__ import annotations
import logging
from fastapi import APIRouter, HTTPException, Request, status
from fastapi import APIRouter, HTTPException, Request
from sqlalchemy.exc import IntegrityError
from app.api.deps import CurrentUser, DbSession
from app.core import test_account
from app.core.ratelimit import enforce_rate_limit
from app.core.security import TokenError, decode_token, issue_token_pair
from app.core.ratelimit import (
RateLimitRule,
check_rate_limits,
enforce_rate_limit,
record_rate_limits,
)
from app.core.security import (
TokenError,
create_bind_ticket,
create_conflict_ticket,
decode_bind_ticket,
decode_conflict_ticket,
decode_token,
issue_token_pair,
)
from app.integrations import wxpay
from app.integrations.jiguang import JiguangError, mask_phone, verify_and_get_phone
from app.integrations.sms import SmsError, send_code, verify_code
from app.repositories import onboarding as onboarding_repo
from app.repositories import phone_rebind as rebind_repo
from app.repositories import user as user_repo
from app.schemas.auth import (
JverifyLoginRequest,
LogoutResponse,
OccupiedAccountInfo,
RefreshRequest,
SmsLoginRequest,
SmsSendRequest,
@@ -32,6 +49,13 @@ from app.schemas.auth import (
TokenPair,
TokenWithUser,
UserOut,
WechatBindPhoneJverifyRequest,
WechatBindPhoneSmsRequest,
WechatBindResultResponse,
WechatConflictContinueRequest,
WechatConflictRebindRequest,
WechatLoginRequest,
WechatLoginResponse,
)
logger = logging.getLogger("shagua.auth")
@@ -40,9 +64,10 @@ router = APIRouter(prefix="/api/v1/auth", tags=["auth"])
# 手机号登录防刷:同一设备(device_id) + 同一 IP 每小时最多的登录尝试次数(成功/失败都计)。
SMS_LOGIN_MAX_PER_HOUR = 5
# 发码防刷:同一设备(device_id) + 同一 IP 每小时最多的发码次数。
# 发码防刷(同一设备 device_id + 同一 IP,**只按成功发码计数**;被单号 60s 冷却挡下的重发不占额度):
# 堵「换手机号绕开单号 60s 冷却」的洞 —— 冷却是单号维度,一机换号能绕开。
SMS_SEND_MAX_PER_HOUR_PER_DEVICE = 5
SMS_SEND_MAX_PER_HOUR_PER_DEVICE = 5 # 每小时上限
SMS_SEND_MAX_PER_DAY_PER_DEVICE = 20 # 每天上限(再叠一层日封顶,挡低频长时间轰炸)
def _login_response(
@@ -99,23 +124,26 @@ def sms_send(req: SmsSendRequest, request: Request) -> SmsSendResponse:
logger.info("test_account sms_send short-circuit (不真发)")
return SmsSendResponse(sent=True, mock=True, cooldown_sec=0)
# 防刷:同一设备(device_id) + 同一 IP 每小时最多 SMS_SEND_MAX_PER_HOUR_PER_DEVICE 次发码
# 补「换手机号绕开单号 60s 冷却」的洞(冷却是单号维度,一机换号能绕);设备维度按机器封顶,
# 挡短信轰炸/烧钱。放在真发(send_code)之前 → 超限直接拦下、不真发短信。
enforce_rate_limit(
request,
scope="sms-send-device",
subject=req.device_id,
limit=SMS_SEND_MAX_PER_HOUR_PER_DEVICE,
window_sec=3600,
detail="操作过于频繁,请稍后再试",
)
# 发码防刷:同一设备(device_id) + 同一 IP,每小时 / 每天两道闸,**均只按成功发码计数**
# 补「换手机号绕开单号 60s 冷却」的洞(冷却是单号维度,一机换号能绕);设备维度按机器封顶,挡短信轰炸/烧钱。
# 关键:被单号 60s 冷却挡下的重发是「没真发、没烧钱」→ 不该占额度。故 check(先判)放在真发之前
# (超限直接 429、不真发),record(计数)只在 send_code 成功后调 —— 冷却/供应商失败抛 429 时直接返回、不计数。
send_rules = [
RateLimitRule("sms-send-device", SMS_SEND_MAX_PER_HOUR_PER_DEVICE, 3600,
"操作过于频繁,请稍后再试"),
RateLimitRule("sms-send-device-daily", SMS_SEND_MAX_PER_DAY_PER_DEVICE, 86400,
"今日验证码发送次数过多,请明天再试"),
]
check_rate_limits(request, subject=req.device_id, rules=send_rules)
try:
cooldown = send_code(req.phone)
except SmsError as e:
raise HTTPException(status_code=e.status_code, detail=str(e)) from e
# 发码成功 → 两道闸各 +1(被单号冷却挡下的重发走不到这里,故不占额度)
record_rate_limits(request, subject=req.device_id, rules=send_rules)
from app.core.config import settings # 局部 import 避免循环
return SmsSendResponse(sent=True, mock=settings.SMS_MOCK, cooldown_sec=cooldown)
@@ -166,6 +194,261 @@ def sms_login(req: SmsLoginRequest, request: Request, db: DbSession) -> TokenWit
return _login_response(user, onboarding_completed=completed)
# ===================== 微信登录 =====================
@router.post(
"/wechat-login",
response_model=WechatLoginResponse,
summary="微信登录(openid 命中即登入,否则发绑号令牌)",
)
def wechat_login(req: WechatLoginRequest, db: DbSession) -> WechatLoginResponse:
from app.core.config import settings # 局部 import,避免循环
# 微信登录只需 code→openid(sns/oauth2),不需要商户转账证书;故只校验 APP_ID/SECRET。
if not (settings.WECHAT_APP_ID and settings.WECHAT_APP_SECRET):
raise HTTPException(status_code=503, detail="wechat login not configured")
try:
info = wxpay.code_to_userinfo(req.code) # {openid, nickname, avatar_url, raw};失败抛 ValueError
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
openid = info["openid"]
user = user_repo.get_user_by_wechat_openid(db, openid)
if user is not None:
# openid 命中 → 直接登入(绝不套用提现 bind-wechat 的"撞号即 409"逻辑)
if user.status != "active":
raise HTTPException(status_code=403, detail="account disabled")
user_repo.touch_last_login(db, user)
completed = onboarding_repo.is_completed(db, user_id=user.id, device_id=req.device_id)
logger.info("wechat_login hit user_id=%d openid=%s*** onboarded=%s", user.id, openid[:6], completed)
return WechatLoginResponse(
status="logged_in",
token=_login_response(user, onboarding_completed=completed),
)
# 未命中 → 签发短时 bind_ticket,进手机号绑定流程(账号此刻还不建)
ticket = create_bind_ticket(
openid=openid,
wechat_nickname=info["nickname"],
wechat_avatar_url=info["avatar_url"],
)
logger.info("wechat_login new openid=%s*** issue bind_ticket", openid[:6])
return WechatLoginResponse(
status="need_bind_phone",
bind_ticket=ticket,
wechat_nickname=info["nickname"],
wechat_avatar_url=info["avatar_url"],
)
def _finish_wechat_bind(
db,
*,
openid: str,
wechat_nickname: str | None,
wechat_avatar_url: str | None,
phone: str,
device_id: str,
) -> WechatBindResultResponse:
"""绑手机建号的公共尾段:手机号被占用 → 返回 phone_occupied(M2 处理 3 选 1);
未占用 → 新建微信账号(channel=wechat,昵称头像取微信)→ 签 token 登入。"""
existing = user_repo.get_user_by_phone(db, phone)
if existing is not None:
from app.core.config import settings # 局部 import,避免循环
ticket = create_conflict_ticket(
openid=openid,
wechat_nickname=wechat_nickname,
wechat_avatar_url=wechat_avatar_url,
phone=phone,
)
blocked = rebind_repo.rebound_within_days(db, phone, settings.PHONE_REBIND_LIMIT_DAYS)
logger.info(
"wechat bind phone occupied phone=%s by user_id=%d has_wechat=%s",
mask_phone(phone), existing.id, bool(existing.wechat_openid),
)
return WechatBindResultResponse(
status="phone_occupied",
occupied_account=OccupiedAccountInfo(
nickname=existing.nickname,
avatar_url=existing.avatar_url,
created_at=existing.created_at,
has_wechat=bool(existing.wechat_openid),
),
conflict_ticket=ticket,
rebind_available=not blocked,
rebind_blocked_days=(
rebind_repo.remaining_block_days(db, phone, settings.PHONE_REBIND_LIMIT_DAYS)
if blocked else 0
),
)
user = user_repo.create_wechat_user(
db,
phone=phone,
openid=openid,
wechat_nickname=wechat_nickname,
wechat_avatar_url=wechat_avatar_url,
)
completed = onboarding_repo.is_completed(db, user_id=user.id, device_id=device_id)
logger.info("wechat bind ok user_id=%d phone=%s openid=%s*** onboarded=%s",
user.id, mask_phone(phone), openid[:6], completed)
return WechatBindResultResponse(
status="logged_in",
token=_login_response(user, onboarding_completed=completed),
)
@router.post(
"/wechat/bind-phone/sms",
response_model=WechatBindResultResponse,
summary="微信登录·其他手机号(短信)绑定",
)
def wechat_bind_phone_sms(
req: WechatBindPhoneSmsRequest, request: Request, db: DbSession
) -> WechatBindResultResponse:
try:
claims = decode_bind_ticket(req.bind_ticket)
except TokenError as e:
raise HTTPException(status_code=401, detail="授权已过期,请重新用微信登录") from e
# 防刷:同 sms/login,按 设备+IP 每小时限流(放在验证码校验之前,失败也计数)
enforce_rate_limit(
request,
scope="wechat-bind-sms-device",
subject=req.device_id,
limit=SMS_LOGIN_MAX_PER_HOUR,
window_sec=3600,
detail="登录尝试过于频繁,请稍后再试",
)
if not verify_code(req.phone, req.code):
raise HTTPException(status_code=400, detail="invalid sms code")
return _finish_wechat_bind(
db,
openid=claims["openid"],
wechat_nickname=claims["wnk"],
wechat_avatar_url=claims["wav"],
phone=req.phone,
device_id=req.device_id,
)
@router.post(
"/wechat/bind-phone/jverify",
response_model=WechatBindResultResponse,
summary="微信登录·本机号(极光)绑定",
)
def wechat_bind_phone_jverify(
req: WechatBindPhoneJverifyRequest, db: DbSession
) -> WechatBindResultResponse:
try:
claims = decode_bind_ticket(req.bind_ticket)
except TokenError as e:
raise HTTPException(status_code=401, detail="授权已过期,请重新用微信登录") from e
try:
phone = verify_and_get_phone(req.login_token)
except JiguangError as e:
logger.error("[JG] verify+decrypt failed: %s", e, exc_info=True)
raise HTTPException(status_code=502, detail=f"jiguang verify failed: {e}") from e
return _finish_wechat_bind(
db,
openid=claims["openid"],
wechat_nickname=claims["wnk"],
wechat_avatar_url=claims["wav"],
phone=phone,
device_id=req.device_id,
)
# ===================== 微信占用冲突(M2) =====================
@router.post(
"/wechat/conflict/continue",
response_model=WechatBindResultResponse,
summary="微信占用冲突·继续绑定(登录老账号,能绑就绑)",
)
def wechat_conflict_continue(
req: WechatConflictContinueRequest, request: Request, db: DbSession
) -> WechatBindResultResponse:
try:
claims = decode_conflict_ticket(req.conflict_ticket)
except TokenError as e:
raise HTTPException(status_code=401, detail="操作超时,请重新用微信登录") from e
enforce_rate_limit(
request, scope="wechat-conflict-device", subject=req.device_id,
limit=SMS_LOGIN_MAX_PER_HOUR, window_sec=3600, detail="操作过于频繁,请稍后再试",
)
user = user_repo.get_user_by_phone(db, claims["phone"])
if user is None:
# P 期间被腾空(老账号改号/注销)→ 前提已变,让前端重走
raise HTTPException(status_code=409, detail="账号状态已变化,请重新登录")
if user.status != "active":
raise HTTPException(status_code=403, detail="account disabled")
if user.wechat_openid is None:
try:
user_repo.attach_wechat_to_user(
db, user, openid=claims["openid"],
wechat_nickname=claims["wnk"], wechat_avatar_url=claims["wav"],
)
except IntegrityError:
db.rollback() # openid 被别处绑走 → 只登入不绑
user_repo.touch_last_login(db, user)
else:
user_repo.touch_last_login(db, user) # X 已绑别的微信 → 只登入,丢弃本次 openid
completed = onboarding_repo.is_completed(db, user_id=user.id, device_id=req.device_id)
logger.info("wechat conflict continue user_id=%d openid=%s***", user.id, claims["openid"][:6])
return WechatBindResultResponse(
status="logged_in",
token=_login_response(user, onboarding_completed=completed),
)
@router.post(
"/wechat/conflict/rebind",
response_model=WechatBindResultResponse,
summary="微信占用冲突·换绑(注销老账号+用该号重建全新账号)",
)
def wechat_conflict_rebind(
req: WechatConflictRebindRequest, request: Request, db: DbSession
) -> WechatBindResultResponse:
from app.core.config import settings # 局部 import,避免循环
try:
claims = decode_conflict_ticket(req.conflict_ticket)
except TokenError as e:
raise HTTPException(status_code=401, detail="操作超时,请重新用微信登录") from e
enforce_rate_limit(
request, scope="wechat-conflict-device", subject=req.device_id,
limit=SMS_LOGIN_MAX_PER_HOUR, window_sec=3600, detail="操作过于频繁,请稍后再试",
)
phone = claims["phone"]
if rebind_repo.rebound_within_days(db, phone, settings.PHONE_REBIND_LIMIT_DAYS):
days = rebind_repo.remaining_block_days(db, phone, settings.PHONE_REBIND_LIMIT_DAYS)
raise HTTPException(status_code=409, detail=f"该手机号 {days} 天内已换绑过,暂不能再次换绑")
user = user_repo.rebind_account(
db, phone=phone, openid=claims["openid"],
wechat_nickname=claims["wnk"], wechat_avatar_url=claims["wav"],
)
completed = onboarding_repo.is_completed(db, user_id=user.id, device_id=req.device_id)
logger.info("wechat conflict rebind new_user_id=%d phone=%s openid=%s***",
user.id, mask_phone(phone), claims["openid"][:6])
return WechatBindResultResponse(
status="logged_in",
token=_login_response(user, onboarding_completed=completed),
)
# ===================== Refresh =====================
@router.post("/refresh", response_model=TokenPair, summary="用 refresh_token 换新 token 对")
+3 -18
View File
@@ -14,24 +14,18 @@ import logging
from fastapi import APIRouter
from app.api.deps import CurrentUser, DbSession
from app.core import wx_poc_signal
from app.repositories import device as device_repo
from app.schemas.device import (
DeviceOut,
DeviceRegisterRequest,
HeartbeatRequest,
HeartbeatResponse,
LivenessAckRequest,
LivenessOut,
OkResponse,
PendingCompare,
)
logger = logging.getLogger("shagua.device")
# PoC:源平台代号 → Android 包名(前端 launch 用)。当前只用美团。
_SOURCE_PACKAGES = {"meituan": "com.sankuai.meituan"}
router = APIRouter(prefix="/api/v1/device", tags=["device"])
@@ -58,12 +52,12 @@ def register_device(
return DeviceOut.model_validate(device)
@router.post("/heartbeat", response_model=HeartbeatResponse, summary="上报心跳")
@router.post("/heartbeat", response_model=OkResponse, summary="上报心跳")
def report_heartbeat(
req: HeartbeatRequest,
user: CurrentUser,
db: DbSession,
) -> HeartbeatResponse:
) -> OkResponse:
device_repo.touch_heartbeat(
db,
user_id=user.id,
@@ -71,16 +65,7 @@ def report_heartbeat(
accessibility_enabled=req.accessibility_enabled,
registration_id=req.registration_id,
)
# PoC:该设备有"待比价"信号则带回(只对写死的测试设备生效, 取走即清、只弹一次)
source = wx_poc_signal.pop_pending(req.device_id)
pending = None
if source:
pending = PendingCompare(
source_platform=source,
source_package=_SOURCE_PACKAGES.get(source, ""),
)
logger.info("heartbeat 下发比价信号 device=%s source=%s", req.device_id, source)
return HeartbeatResponse(pending_compare=pending)
return OkResponse()
@router.get("/liveness", response_model=LivenessOut, summary="查询本机掉线告警(后置检测)")
-159
View File
@@ -1,159 +0,0 @@
"""微信客服(企业微信)消息接收回调 /wx/kf/callback(裸路径, 非 /api/v1)。
对应企业微信「微信客服」→ 客服账号 → 接收消息回调里填的 URL。是服务号版(wx_mp.py)的
平行实现:下游(set_pending → 心跳下发 → 前端弹窗)完全一致, 只有上游"如何拿到用户消息"不同 ——
服务号是微信把消息直接 POST 给你;微信客服是 POST 一个事件通知, 你再用 <Token> 主动 sync_msg 拉。
GET : URL 接入验证 → msg_signature 验签(含密文 echostr) → 解密 echostr → 返回明文。
⚠️ 与公众号明文模式不同:企业微信 echostr 是密文, 必须解密后返回。
POST : 收事件(密文) → msg_signature 验签 → AES 解密 → 解析 XML。
Event=kf_msg_or_event → 取 <Token>/<OpenKfId> → sync_msg 增量拉消息 →
客户(origin=3)发来的图片(image) → set_pending(测试设备, "meituan") + 下载落盘。
其余消息/事件记日志忽略。一律返回 "success"(异常也吞掉, 不让企业微信重试轰炸)。
回调加密与公众号「安全模式」同源, 直接复用 integrations/wx_mp_crypto(验签 + AES-256-CBC 解密),
差别只在 receiveid 用 corpid(公众号是 appid)。MVP 只做"收图打信号 + 落盘证明链路通"
"""
from __future__ import annotations
import logging
import xml.etree.ElementTree as ET
from pathlib import Path
from typing import Any
from fastapi import APIRouter, Request
from fastapi.responses import PlainTextResponse
from app.core import wx_poc_signal
from app.core.config import settings
from app.integrations import wx_kf_client, wx_mp_crypto
logger = logging.getLogger("shagua.wx_kf")
router = APIRouter(prefix="/wx/kf", tags=["wx-kf"])
# 收到的截图暂存目录(MVP 验证用;接上识别后可改为不落盘或定期清理)
_INBOX = Path(settings.MEDIA_ROOT) / "wx_kf_inbox"
@router.get("/callback", include_in_schema=False)
async def verify(
msg_signature: str = "", timestamp: str = "", nonce: str = "", echostr: str = ""
) -> PlainTextResponse:
"""企业微信 URL 接入验证:验签(密文 echostr 参与)→ 解密 echostr → 返回明文。"""
if not settings.wx_kf_callback_configured:
logger.warning("wx_kf verify: 回调凭证未配齐(CorpId/Token/AESKey/Secret)")
return PlainTextResponse("", status_code=503)
if not wx_mp_crypto.verify_msg_signature(
settings.WX_KF_TOKEN, timestamp, nonce, echostr, msg_signature
):
logger.warning("wx_kf verify: signature 校验失败 ts=%s nonce=%s", timestamp, nonce)
return PlainTextResponse("invalid signature", status_code=403)
try:
plain = wx_mp_crypto.decrypt_message(
settings.WX_KF_AES_KEY, settings.WX_KF_CORP_ID, echostr
)
except Exception:
logger.exception("wx_kf verify: echostr 解密失败")
return PlainTextResponse("decrypt failed", status_code=403)
return PlainTextResponse(plain)
@router.post("/callback", include_in_schema=False)
async def receive(request: Request) -> PlainTextResponse:
"""收事件通知(安全模式)。任何异常都吞掉返回 success, 不让企业微信重试轰炸, 靠日志排查。"""
if not settings.wx_kf_callback_configured:
logger.warning("wx_kf receive: 回调凭证未配齐")
return PlainTextResponse("success")
try:
body = (await request.body()).decode("utf-8")
qp = request.query_params
# 安全模式外层 XML 只有 ToUserName + Encrypt。来源=企业微信服务器(HTTPS)+ 下面验签,
# 且 stdlib ET 不扩展外部实体, XXE 不适用。
encrypt = ET.fromstring(body).findtext("Encrypt") or ""
if not wx_mp_crypto.verify_msg_signature(
settings.WX_KF_TOKEN,
qp.get("timestamp", ""),
qp.get("nonce", ""),
encrypt,
qp.get("msg_signature", ""),
):
logger.warning("wx_kf receive: msg_signature 校验失败")
return PlainTextResponse("success")
xml = wx_mp_crypto.decrypt_message(
settings.WX_KF_AES_KEY, settings.WX_KF_CORP_ID, encrypt
)
await _handle_event(ET.fromstring(xml))
except Exception:
logger.exception("wx_kf receive: 处理异常")
return PlainTextResponse("success")
async def _handle_event(root: ET.Element) -> None:
"""客服事件:MsgType=event 且 Event=kf_msg_or_event → 用 Token 增量拉消息。"""
event = root.findtext("Event") or ""
if event != "kf_msg_or_event":
logger.info("wx_kf 收到事件(暂忽略): event=%s", event)
return
token = root.findtext("Token") or ""
open_kfid = root.findtext("OpenKfId") or ""
msgs = await wx_kf_client.sync_messages(token, open_kfid)
logger.info("wx_kf sync_msg 拉到 %d 条 open_kfid=%s", len(msgs), open_kfid)
for msg in msgs:
await _handle_message(msg)
async def _handle_message(msg: dict[str, Any]) -> None:
"""处理单条客服消息。只关心【客户发来的】(origin=3)图片 / 小程序卡片:打比价信号(+ 落盘)。
origin:3=客户发的 / 4=系统推送 / 5=接待人员发的。接待人员或系统消息一律不触发比价。
两种触发物对应场景:image=用户发结算页截图;miniprogram=用户分享美团小程序卡片(见需求截图)。
"""
if msg.get("origin") != 3:
return
msgtype = msg.get("msgtype") or ""
external_userid = msg.get("external_userid") or ""
if msgtype == "image":
media_id = (msg.get("image") or {}).get("media_id", "")
logger.info(
"wx_kf 收到图片: external_userid=%s media_id=%s", external_userid, media_id
)
_fire_pending("image")
await _save_media(media_id, external_userid)
elif msgtype == "miniprogram":
mp = msg.get("miniprogram") or {}
logger.info(
"wx_kf 收到小程序卡片: external_userid=%s title=%s appid=%s",
external_userid,
mp.get("title"),
mp.get("appid"),
)
# 卡片无可下载媒体, 只打信号;源平台/订单识别(可据 appid 判平台)留后续, 现写死美团。
_fire_pending("miniprogram")
else:
logger.info(
"wx_kf 收到消息(暂忽略): external_userid=%s type=%s", external_userid, msgtype
)
def _fire_pending(kind: str) -> None:
"""PoC:给写死的测试设备打"从美团比价"信号(下次心跳带回前端弹窗, 与服务号版完全一致)。"""
poc_dev = settings.WX_POC_TEST_DEVICE_ID
if not poc_dev:
return
wx_poc_signal.set_pending(poc_dev, "meituan")
logger.info("wx_kf PoC: 已给测试设备 %s 打比价信号 source=meituan (触发=%s)", poc_dev, kind)
async def _save_media(media_id: str, external_userid: str) -> None:
"""图片消息只给 media_id, 走临时素材下载接口拉回落盘(PoC 验证用)。"""
content = await wx_kf_client.download_media(media_id)
if not content:
return
try:
_INBOX.mkdir(parents=True, exist_ok=True)
dest = _INBOX / f"{external_userid[:16]}_{media_id[:16]}.jpg"
dest.write_bytes(content)
logger.info("wx_kf 图片已落盘: %s (%d bytes)", dest, len(content))
except Exception:
logger.exception("wx_kf 图片落盘失败 media_id=%s", media_id)
-112
View File
@@ -1,112 +0,0 @@
"""微信服务号消息接收回调 /wx/mp/callback(裸路径, 非 /api/v1)。
对应公众平台"设置与开发 → 基本配置 → 服务器配置"里填的 URL。
GET : URL 接入验证 → 校验 signature → 原样返回 echostr(明文)
POST : 收用户消息(安全模式密文) → msg_signature 验签 → AES 解密 → 解析 XML
图片消息(MsgType=image): 打日志(openid/PicUrl/MediaId) + 用 PicUrl 下载落盘
其余消息/事件: 记一条日志忽略。一律返回 "success"(微信据此不再重试)。
MVP 只做"收到图片并落盘"证明链路通;后续接:识别平台+订单 → 打 pending_compare 标记
→ 心跳带回前端弹窗,以及 openid↔device 绑定。任何异常都返回 "success"、不让微信重试轰炸。
"""
from __future__ import annotations
import logging
import xml.etree.ElementTree as ET
from pathlib import Path
import httpx
from fastapi import APIRouter, Request
from fastapi.responses import PlainTextResponse
from app.core import wx_poc_signal
from app.core.config import settings
from app.integrations import wx_mp_crypto
logger = logging.getLogger("shagua.wx_mp")
router = APIRouter(prefix="/wx/mp", tags=["wx-mp"])
# 收到的截图暂存目录(MVP 验证用;接上识别后可改为不落盘或定期清理)
_INBOX = Path(settings.MEDIA_ROOT) / "wx_inbox"
@router.get("/callback", include_in_schema=False)
async def verify(
signature: str = "", timestamp: str = "", nonce: str = "", echostr: str = ""
) -> PlainTextResponse:
"""微信 URL 接入验证:校验 signature 通过则原样返回 echostr。"""
if not settings.WX_MP_TOKEN:
logger.warning("wx_mp verify: WX_MP_TOKEN 未配置")
return PlainTextResponse("", status_code=503)
if wx_mp_crypto.verify_url_signature(
settings.WX_MP_TOKEN, timestamp, nonce, signature
):
return PlainTextResponse(echostr)
logger.warning("wx_mp verify: signature 校验失败 ts=%s nonce=%s", timestamp, nonce)
return PlainTextResponse("invalid signature", status_code=403)
@router.post("/callback", include_in_schema=False)
async def receive(request: Request) -> PlainTextResponse:
"""收用户消息(安全模式)。任何异常都吞掉返回 success,不让微信重试轰炸,靠日志排查。"""
if not settings.wx_mp_callback_configured:
logger.warning("wx_mp receive: 回调凭证未配齐(Token/AESKey/AppID)")
return PlainTextResponse("success")
try:
body = (await request.body()).decode("utf-8")
qp = request.query_params
# 安全模式外层 XML 只有 ToUserName + Encrypt。来源=微信服务器(HTTPS)+ 下面验签,
# 且 stdlib ET 不扩展外部实体, XXE 不适用。
encrypt = ET.fromstring(body).findtext("Encrypt") or ""
if not wx_mp_crypto.verify_msg_signature(
settings.WX_MP_TOKEN,
qp.get("timestamp", ""),
qp.get("nonce", ""),
encrypt,
qp.get("msg_signature", ""),
):
logger.warning("wx_mp receive: msg_signature 校验失败")
return PlainTextResponse("success")
xml = wx_mp_crypto.decrypt_message(
settings.WX_MP_AES_KEY, settings.WX_MP_APPID, encrypt
)
await _handle_message(ET.fromstring(xml))
except Exception:
logger.exception("wx_mp receive: 处理异常")
return PlainTextResponse("success")
async def _handle_message(root: ET.Element) -> None:
openid = root.findtext("FromUserName") or ""
msg_type = root.findtext("MsgType") or ""
if msg_type == "image":
pic_url = root.findtext("PicUrl") or ""
media_id = root.findtext("MediaId") or ""
logger.info(
"wx_mp 收到图片: openid=%s media_id=%s pic_url=%s", openid, media_id, pic_url
)
# PoC: 写死"该测试设备要从美团比价", 下次心跳带回前端弹选平台窗
poc_dev = settings.WX_POC_TEST_DEVICE_ID
if poc_dev:
wx_poc_signal.set_pending(poc_dev, "meituan")
logger.info("wx_mp PoC: 已给测试设备 %s 打比价信号 source=meituan", poc_dev)
await _download(pic_url, openid, media_id)
else:
logger.info("wx_mp 收到消息(暂忽略): openid=%s type=%s", openid, msg_type)
async def _download(pic_url: str, openid: str, media_id: str) -> None:
"""用 PicUrl 直接下载图片落盘。PicUrl 是临时公网链接,无需 access_token / IP 白名单。"""
if not pic_url:
return
try:
_INBOX.mkdir(parents=True, exist_ok=True)
async with httpx.AsyncClient(timeout=15) as client:
resp = await client.get(pic_url)
resp.raise_for_status()
dest = _INBOX / f"{openid[:12]}_{media_id[:16]}.jpg"
dest.write_bytes(resp.content)
logger.info("wx_mp 图片已落盘: %s (%d bytes)", dest, len(resp.content))
except Exception:
logger.exception("wx_mp 图片下载失败 pic_url=%s", pic_url)
+9 -35
View File
@@ -44,6 +44,11 @@ class Settings(BaseSettings):
JWT_ALGORITHM: str = "HS256"
JWT_ACCESS_TOKEN_EXPIRE_MINUTES: int = 120
JWT_REFRESH_TOKEN_EXPIRE_DAYS: int = 30
# 微信登录未命中 openid 时签发的"待绑手机"令牌有效期(JWT_SECRET_KEY 签名,typ=wechat_bind;
# 见 security.create_bind_ticket)。需覆盖"授权→输手机号→收短信→输验证码"整个绑定流程。
WECHAT_BIND_TICKET_EXPIRE_MINUTES: int = 10
# 一个手机号 30 天内最多换绑一次(微信占用冲突页的"换绑"动作)。见 phone_rebind_log。
PHONE_REBIND_LIMIT_DAYS: int = 30
# ===== Admin 后台 =====
# admin 用独立 JWT secret(≠ JWT_SECRET_KEY),App 用户 token 无法越权访问后台。
@@ -81,6 +86,7 @@ class Settings(BaseSettings):
SMS_SIGN_ID: int = 31729 # 极光短信签名 ID(非机密,可被 .env 覆盖)
SMS_TEMPLATE_ID: int = 1 # 极光短信模板 ID(变量名 code,有效期 5 分钟)
SMS_CODE_LENGTH: int = 6 # 验证码位数(本服务生成;前端 code 字段 4-8 位兼容)
SMS_DAILY_LIMIT_PER_PHONE: int = 10 # 单手机号每日发送上限(防刷 + 控费)
SMS_MAX_VERIFY_ATTEMPTS: int = 5 # 单个验证码最多校验失败次数,超过即作废(防爆破)
# ===== 测试账号(release 包全流程联调用)=====
@@ -106,6 +112,9 @@ class Settings(BaseSettings):
# 美团调用走的代理。本机开发直连美团会 SSL EOF,需填 http://127.0.0.1:7897;
# 线上国内服务器留空(=直连)。见 .env.example 与 integrations/meituan.py。
MT_CPS_PROXY: str = ""
# 本地开发:开启后 /feed 接口直接返回 mock 数据,不调美团 API、不查离线库,
# 方便前端联调 feed 卡片样式、分页、距离排序等 UI。生产必须 false。
MT_CPS_MOCK_FEED: bool = True
@property
def mt_cps_configured(self) -> bool:
@@ -136,26 +145,6 @@ class Settings(BaseSettings):
# (不跳授权、不拿 openid),整套微信代码保留。审核通过后 .env 置 true + 重启即启用,无需改代码。
WX_MP_OAUTH_ENABLED: bool = False
# ===== 微信服务号(消息接收回调) =====
# 用户发消息给服务号 → 微信 POST 到 /wx/mp/callback(安全模式:Token 验签 + AESKey 解密)。
# 区别于上面的网页授权(那是落地页拿 openid);这里是被动收用户主动发来的消息(截图比价入口)。
WX_MP_TOKEN: str = "" # 公众平台"服务器配置"的 Token(URL 验签 + 消息验签)
WX_MP_AES_KEY: str = "" # EncodingAESKey(43 位, 消息 AES-256-CBC 加解密)
# ===== 微信截图比价 PoC(临时验证链路, 验证后删) =====
# 只对这台写死的测试设备下发"从美团比价"信号;空=不触发任何设备(部署上线零风险)。
# 收图 → 给此 device_id 打 pending → 该设备下次心跳带回 → 前端弹选平台窗。
WX_POC_TEST_DEVICE_ID: str = ""
# ===== 微信客服(企业微信;截图比价上游入口 —— 服务号版的平行实现) =====
# 用户在「微信客服」会话发消息 → 企业微信 POST 事件到 /wx/kf/callback → 后端 sync_msg 拉消息。
# 回调加密与公众号「安全模式」同源(复用 wx_mp_crypto), 但 receiveid 用 corpid 且 echostr 需解密。
# 拉消息 / 下载图片走 qyapi, 需 access_token(corpid + secret 换取)。全空=回调端点拒收, 零风险。
WX_KF_CORP_ID: str = "" # 企业微信 corpid(回调解密 receiveid + 换 access_token)
WX_KF_SECRET: str = "" # 「微信客服」Secret(换 access_token)
WX_KF_TOKEN: str = "" # 客服「接收消息」回调的 Token(URL 验签 + 消息验签)
WX_KF_AES_KEY: str = "" # 客服「接收消息」回调的 EncodingAESKey(AES-256-CBC 解密)
@property
def wx_mp_configured(self) -> bool:
"""服务号网页授权凭证齐全(缺则落地页不发起授权,降级为无 openid)。"""
@@ -166,21 +155,6 @@ class Settings(BaseSettings):
"""落地页是否真正发起微信授权 = 凭证齐全 且 总开关开。"""
return self.wx_mp_configured and self.WX_MP_OAUTH_ENABLED
@property
def wx_mp_callback_configured(self) -> bool:
"""消息接收回调凭证齐全(Token + AESKey + AppID)。缺则回调端点拒绝处理消息。"""
return bool(self.WX_MP_TOKEN and self.WX_MP_AES_KEY and self.WX_MP_APPID)
@property
def wx_kf_callback_configured(self) -> bool:
"""微信客服回调凭证齐全(CorpId + Token + AESKey + Secret)。缺则回调端点拒绝处理。"""
return bool(
self.WX_KF_CORP_ID
and self.WX_KF_TOKEN
and self.WX_KF_AES_KEY
and self.WX_KF_SECRET
)
# ===== 微信支付(商家转账到零钱 / 提现)=====
# 真实凭证放 .env(已 gitignore),证书 .pem 放 secrets/。WECHAT_APP_ID 同时用于
# 微信登录(code 换 openid)与转账,必须与 App 端开放平台 appid 一致。
+93 -8
View File
@@ -9,29 +9,41 @@ from __future__ import annotations
import threading
import time
from typing import NamedTuple
from fastapi import HTTPException, Request, status
from app.core.config import settings
# key -> (window_start_ts, count)
_buckets: dict[str, tuple[float, int]] = {}
# key -> (window_start_ts, count, window_sec)
# 存每个 key 自己的 window_sec:_buckets 混着不同窗口(60s 广告 / 3600s 登录 / 86400s 日闸)的 key,
# GC 必须按各 key 自己的窗口判过期(见 [_purge_expired]),否则短窗口调用触发的 GC 会误删长窗口 key。
_buckets: dict[str, tuple[float, int, float]] = {}
_lock = threading.Lock()
_GC_THRESHOLD = 10000 # _buckets 超此阈值才顺手清过期 key(仿 sms.py;测试可 monkeypatch 调小强制每次扫)
def _purge_expired(now: float) -> None:
"""清过期 key(**仅在持有 _lock 时调用**)。按每个 key 自己存的 window_sec 判过期,而非调用方的窗口
—— _buckets 是全局共享、混着 60s(广告)/3600s(登录)/86400s(日闸)不同窗口的 key;若用调用方窗口,
高频的 60s 广告端点触发 GC 时会把本该活 3600s/86400s 的登录/日闸计数一并删掉,使其在规模上(超阈值才
触发本清理)被反复清零而失效。仅在超阈值时扫,低频、开销可忽略。"""
if len(_buckets) <= _GC_THRESHOLD:
return
for k in [k for k, (s, _, w) in _buckets.items() if now - s >= w]:
_buckets.pop(k, None)
def _hit(key: str, limit: int, window_sec: float) -> bool:
"""记一次访问。返回 True=放行,False=超限。"""
now = time.monotonic()
with _lock:
start, count = _buckets.get(key, (now, 0))
start, count, _ = _buckets.get(key, (now, 0, window_sec))
if now - start >= window_sec: # 窗口过期,重置
start, count = now, 0
count += 1
_buckets[key] = (start, count)
# 顺手清过期 key,防内存无限涨(低频访问足够)
if len(_buckets) > 10000:
for k in [k for k, (s, _) in _buckets.items() if now - s >= window_sec]:
_buckets.pop(k, None)
_buckets[key] = (start, count, window_sec)
_purge_expired(now) # 顺手清过期 key(按各自窗口),防内存无限涨
return count <= limit
@@ -83,3 +95,76 @@ def enforce_rate_limit(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail=detail,
)
# ===================== 先判 / 后记(只按「成功」计数)=====================
# _hit 是原子「判+记」:一调用就 +1,适合登录爆破(失败尝试也要计)。但对「短信发码」这类
# **只想给成功动作计数**的场景不合适 —— 被单号冷却挡下的重发没真发、没烧钱,不该占额度。
# 故拆成 _peek(只判不记)+ _commit(只记):check_rate_limits 先判 → 动作 → 成功后 record。
class RateLimitRule(NamedTuple):
"""一条限流规则。scope 区分不同闸(不同 key 前缀);同一 (subject, IP) 在 window_sec
内最多 limit 次,超限抛 429 用 detail 文案。
(scope, window_sec) 成对绑在一条规则里 —— check(先判)与 record(计数)复用同一条,
避免两处把窗口/scope 写歪导致 key 对不上。
"""
scope: str
limit: int
window_sec: float
detail: str = "操作过于频繁,请稍后再试"
def _peek(key: str, limit: int, window_sec: float) -> bool:
"""只读:当前窗口内是否还没到上限(count < limit)。**不改计数**。
与 [_commit] 配对实现「先判后记」——只在动作成功后才 _commit。"""
now = time.monotonic()
with _lock:
start, count, _ = _buckets.get(key, (now, 0, window_sec))
if now - start >= window_sec: # 窗口已过期 → 视作已重置(count 归零)
count = 0
return count < limit
def _commit(key: str, window_sec: float) -> None:
"""记一次访问(+1)。窗口过期则以本次为起点重置。仅在动作成功后调用。"""
now = time.monotonic()
with _lock:
start, count, _ = _buckets.get(key, (now, 0, window_sec))
if now - start >= window_sec: # 窗口过期,重置
start, count = now, 0
_buckets[key] = (start, count + 1, window_sec)
_purge_expired(now) # 顺手清过期 key(按各自窗口,同 [_hit])
def check_rate_limits(request: Request, subject: str, rules: list[RateLimitRule]) -> None:
"""【先判】一组限流:任一规则已达上限即抛 429,且**不改计数**。
配合 [record_rate_limits] 实现「只按成功计数」:先 check 所有闸(全未超才继续)→ 执行动作
→ 动作**成功后**再 record。动作被下游挡下(如短信单号冷却)、没真正发生时不 record → 不占额度。
key = `scope:subject:client_ip`(与 [enforce_rate_limit] 同款)。
"""
if not settings.RATE_LIMIT_ENABLED:
return
ip = _client_ip(request)
for rule in rules:
if not _peek(f"{rule.scope}:{subject}:{ip}", rule.limit, rule.window_sec):
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail=rule.detail,
)
def record_rate_limits(request: Request, subject: str, rules: list[RateLimitRule]) -> None:
"""【记一次】一组限流(每条规则 +1)。仅在动作成功后调用,与 [check_rate_limits] 配对。
⚠️ check→动作→record 非原子:并发突发下计数可能略超 limit(每个在途请求各 +1)。对
「防脚本/防轰炸」的安全网定位可接受;要精确配额需迁 Redis(见模块 docstring)。
"""
if not settings.RATE_LIMIT_ENABLED:
return
ip = _client_ip(request)
for rule in rules:
_commit(f"{rule.scope}:{subject}:{ip}", rule.window_sec)
+85
View File
@@ -87,6 +87,91 @@ def issue_token_pair(user_id: int) -> dict[str, Any]:
}
def create_bind_ticket(
*, openid: str, wechat_nickname: str | None, wechat_avatar_url: str | None
) -> str:
"""微信登录未命中 openid 时,签发短时"待绑手机"令牌,承载 openid + 微信昵称头像。
typ='wechat_bind'、sub=openid;有效期 settings.WECHAT_BIND_TICKET_EXPIRE_MINUTES 分钟。
与 access/refresh 用同一 JWT_SECRET_KEY 签名,靠 typ 区分,decode_bind_ticket 校验 typ。
"""
now = _now()
expire = now + timedelta(minutes=settings.WECHAT_BIND_TICKET_EXPIRE_MINUTES)
payload: dict[str, Any] = {
"sub": openid,
"typ": "wechat_bind",
"wnk": wechat_nickname,
"wav": wechat_avatar_url,
"iat": int(now.timestamp()),
"exp": int(expire.timestamp()),
}
return jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM)
def decode_bind_ticket(token: str) -> dict[str, Any]:
"""解析"待绑手机"令牌,校验签名/过期/类型。失败抛 TokenError。
返回 {'openid': str, 'wnk': str|None, 'wav': str|None}。
"""
try:
payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=[settings.JWT_ALGORITHM])
except jwt.ExpiredSignatureError as e:
raise TokenError("bind ticket expired") from e
except jwt.InvalidTokenError as e:
raise TokenError(f"invalid bind ticket: {e}") from e
if payload.get("typ") != "wechat_bind":
raise TokenError(f"wrong token type: want=wechat_bind got={payload.get('typ')}")
if "sub" not in payload:
raise TokenError("bind ticket missing sub")
return {"openid": payload["sub"], "wnk": payload.get("wnk"), "wav": payload.get("wav")}
def create_conflict_ticket(
*, openid: str, wechat_nickname: str | None, wechat_avatar_url: str | None, phone: str
) -> str:
"""手机号占用时签发的短时"冲突处理"令牌。
比 bind_ticket 多编码 **已验证的手机号 phone** —— 换绑/继续绑定只认它,证明"这对
openid/手机号刚在绑号时验证通过",免用户重输验证码,又堵住"拿自己 openid + 任意手机号
去夺号"的接管漏洞。typ='wechat_conflict';有效期复用 WECHAT_BIND_TICKET_EXPIRE_MINUTES。
"""
now = _now()
expire = now + timedelta(minutes=settings.WECHAT_BIND_TICKET_EXPIRE_MINUTES)
payload: dict[str, Any] = {
"sub": openid,
"typ": "wechat_conflict",
"wnk": wechat_nickname,
"wav": wechat_avatar_url,
"phn": phone,
"iat": int(now.timestamp()),
"exp": int(expire.timestamp()),
}
return jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM)
def decode_conflict_ticket(token: str) -> dict[str, Any]:
"""解析"冲突处理"令牌,校验签名/过期/类型。失败抛 TokenError。
返回 {'openid': str, 'wnk': str|None, 'wav': str|None, 'phone': str}。
"""
try:
payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=[settings.JWT_ALGORITHM])
except jwt.ExpiredSignatureError as e:
raise TokenError("conflict ticket expired") from e
except jwt.InvalidTokenError as e:
raise TokenError(f"invalid conflict ticket: {e}") from e
if payload.get("typ") != "wechat_conflict":
raise TokenError(f"wrong token type: want=wechat_conflict got={payload.get('typ')}")
if "sub" not in payload or "phn" not in payload:
raise TokenError("conflict ticket missing sub/phn")
return {
"openid": payload["sub"],
"wnk": payload.get("wnk"),
"wav": payload.get("wav"),
"phone": payload["phn"],
}
# ===================== 密码 hash(admin 后台账号用)=====================
# 用户侧是手机号+验证码登录,不存密码;仅 admin 账号用 username+password 登录。
-27
View File
@@ -1,27 +0,0 @@
"""微信截图比价 PoC:进程内存的"待比价"信号(单 worker 够用, 重启即失效, PoC 可接受)。
收图端点 set_pending(device_id, source) → 该设备下次心跳 pop_pending 取走并清除 →
心跳响应带回 → 前端弹选平台窗。只对 settings.WX_POC_TEST_DEVICE_ID 写入, 不碰其他设备。
后续接真识别 / openid↔device 绑定时整体替换本模块。
"""
from __future__ import annotations
import threading
_lock = threading.Lock()
_pending: dict[str, str] = {} # device_id -> source_platform(如 "meituan")
def set_pending(device_id: str, source_platform: str) -> None:
if not device_id:
return
with _lock:
_pending[device_id] = source_platform
def pop_pending(device_id: str) -> str | None:
"""取出并清除该设备的待比价信号;无则 None。心跳每帧调, 取到即消费(只弹一次)。"""
if not device_id:
return None
with _lock:
return _pending.pop(device_id, None)
+2 -1
View File
@@ -13,7 +13,8 @@ worker / 多机时内存不共享 → 冷却、校验都会失效,届时迁移
防刷两层(短信花钱 + `/sms/send` 在登录前无法 JWT 鉴权):
1. 单号 `SMS_SEND_INTERVAL_SEC` 冷却(本文件)
2. 单设备(device_id)每小时频控(api 层 auth.sms_send 内 enforce_rate_limit)+ 极光控制台 IP 白名单/防轰炸(运维侧)。
2. 单设备(device_id)+ IP 每小时 / 每天频控(api 层 auth.sms_send 的 check/record_rate_limits,
**只按成功发码计数** —— 被本文件单号冷却挡下的重发不占额度)+ 极光控制台 IP 白名单/防轰炸(运维侧)。
⚠️ 原「单 IP 频控(rate_limit 依赖)」2026-06-26 按产品要求删除、改设备维度;但 device_id 客户端可伪造/轮换,
脚本轮换 id 能绕过本层 → 挡脚本狂发主要靠极光控制台侧(+ 可选 nginx 限流)。
⚠️ 原「单号每日上限」2026-07-03 按精简要求删除(mentor 定:登录风控只留单号冷却 + 单设备频控);
-126
View File
@@ -1,126 +0,0 @@
"""微信客服(企业微信)API 客户端:access_token 缓存 + sync_msg 拉消息 + 临时素材下载。
对应 api/v1/wx_kf.py:收到 `kf_msg_or_event` 回调后, 用这里的函数增量拉取用户消息。
- access_token(7200s)进程内缓存, 提前 300s 刷新(单 worker 够用, 重启即失效)。
- sync_msg 游标 next_cursor 按 open_kfid 进程内存续(PoC 重启后从最近 3 天重拉, 可接受)。
凭证来自 settings(WX_KF_CORP_ID / WX_KF_SECRET);纯外部 HTTP, 不含业务逻辑。
"""
from __future__ import annotations
import asyncio
import logging
import time
from typing import Any
import httpx
from app.core.config import settings
logger = logging.getLogger("shagua.wx_kf")
_QYAPI = "https://qyapi.weixin.qq.com/cgi-bin"
# access_token 进程内缓存(单 worker)。token 空 / 未到刷新点直接用。
_token_lock = asyncio.Lock()
_token_cache: dict[str, Any] = {"token": "", "expire_at": 0.0}
# sync_msg 游标:open_kfid -> next_cursor(PoC 进程内存;重启从最近 3 天重拉)
_cursors: dict[str, str] = {}
async def _get_access_token() -> str:
"""取企业微信 access_token, 进程内缓存, 提前 300s 过期刷新。失败返回空串(调用方降级)。"""
now = time.time()
if _token_cache["token"] and _token_cache["expire_at"] - 300 > now:
return _token_cache["token"]
async with _token_lock:
now = time.time() # 拿锁后复检, 避免并发重复刷新
if _token_cache["token"] and _token_cache["expire_at"] - 300 > now:
return _token_cache["token"]
try:
async with httpx.AsyncClient(timeout=10) as client:
resp = await client.get(
f"{_QYAPI}/gettoken",
params={
"corpid": settings.WX_KF_CORP_ID,
"corpsecret": settings.WX_KF_SECRET,
},
)
data = resp.json()
except Exception:
logger.exception("wx_kf gettoken 请求失败")
return ""
if data.get("errcode"):
logger.warning(
"wx_kf gettoken errcode=%s errmsg=%s", data.get("errcode"), data.get("errmsg")
)
return ""
_token_cache["token"] = data.get("access_token", "")
_token_cache["expire_at"] = now + int(data.get("expires_in", 7200))
return _token_cache["token"]
async def sync_messages(token: str, open_kfid: str) -> list[dict[str, Any]]:
"""收到 kf_msg_or_event 回调后调用:用回调带的一次性 token 增量拉消息, has_more 循环拉完。
token: 回调事件里的 <Token>(消息拉取凭证, 非配置 Token;不传会有严格频控)。
open_kfid:客服账号 id(回调里的 <OpenKfId>)。
返回本次新拉到的 msg_list(已合并多页);游标按 open_kfid 进程内存续。
"""
access_token = await _get_access_token()
if not access_token:
return []
out: list[dict[str, Any]] = []
cursor = _cursors.get(open_kfid, "")
try:
async with httpx.AsyncClient(timeout=15) as client:
for _ in range(20): # 最多 20 页护栏, 防异常时死循环
body: dict[str, Any] = {"cursor": cursor, "token": token, "limit": 1000}
if open_kfid:
body["open_kfid"] = open_kfid
resp = await client.post(
f"{_QYAPI}/kf/sync_msg",
params={"access_token": access_token},
json=body,
)
data = resp.json()
if data.get("errcode"):
logger.warning(
"wx_kf sync_msg errcode=%s errmsg=%s",
data.get("errcode"),
data.get("errmsg"),
)
break
out.extend(data.get("msg_list", []))
cursor = data.get("next_cursor", cursor)
_cursors[open_kfid] = cursor
if not data.get("has_more"):
break
except Exception:
logger.exception("wx_kf sync_msg 请求失败 open_kfid=%s", open_kfid)
return out
async def download_media(media_id: str) -> bytes | None:
"""临时素材下载(图片消息只给 media_id)。成功返回二进制, 失败返回 None。"""
if not media_id:
return None
access_token = await _get_access_token()
if not access_token:
return None
try:
async with httpx.AsyncClient(timeout=20) as client:
resp = await client.get(
f"{_QYAPI}/media/get",
params={"access_token": access_token, "media_id": media_id},
)
resp.raise_for_status()
# 成功=二进制文件流;失败=JSON(errcode)。据 Content-Type 区分。
ctype = resp.headers.get("Content-Type", "")
if "application/json" in ctype or "text/plain" in ctype:
logger.warning("wx_kf media/get 非文件响应: %s", resp.text[:200])
return None
return resp.content
except Exception:
logger.exception("wx_kf media/get 下载失败 media_id=%s", media_id)
return None
-69
View File
@@ -1,69 +0,0 @@
"""微信服务号消息接收回调的验签与解密(安全模式)。
服务号"服务器配置"选安全模式后:
- URL 接入验证(GET): sha1(sort(token, timestamp, nonce)) == signature 原样返回 echostr
- 消息(POST): body <Encrypt> 是密文;
msg_signature = sha1(sort(token, timestamp, nonce, encrypt))
密文 AES-256-CBC 解出: random(16B) + msg_len(4B big-endian) + msg + appid, PKCS7(=32) padding
只做验签 + 解密(收消息)被动回复(加密)MVP 不需要 收到后走客服消息异步回执
密钥/口令来自 settings(WX_MP_TOKEN / WX_MP_AES_KEY / WX_MP_APPID),本模块只做纯算法不读配置
"""
from __future__ import annotations
import base64
import hashlib
import hmac
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
def verify_url_signature(token: str, timestamp: str, nonce: str, signature: str) -> bool:
"""GET 接入验证:token/timestamp/nonce 三者字典序排序拼接后 sha1。"""
return _consteq(_sha1(token, timestamp, nonce), signature)
def verify_msg_signature(
token: str, timestamp: str, nonce: str, encrypt: str, msg_signature: str
) -> bool:
"""POST 消息验签:四者(含密文 encrypt)字典序排序拼接后 sha1。"""
return _consteq(_sha1(token, timestamp, nonce, encrypt), msg_signature)
def decrypt_message(aes_key_b64: str, expected_appid: str, encrypt_b64: str) -> str:
"""解密 <Encrypt> 密文, 返回明文消息 XML。appid 不符抛 ValueError。
aes_key_b64: EncodingAESKey(43 , 不含结尾 '='), '=' base64 解出 32 字节 AES-256 key
"""
aes_key = base64.b64decode(aes_key_b64 + "=") # 43 → 32 bytes
iv = aes_key[:16]
decryptor = Cipher(algorithms.AES(aes_key), modes.CBC(iv)).decryptor()
plain = decryptor.update(base64.b64decode(encrypt_b64)) + decryptor.finalize()
plain = _pkcs7_unpad(plain)
# random(16) + msg_len(4, big-endian) + msg(msg_len) + from_appid
content = plain[16:]
msg_len = int.from_bytes(content[:4], "big")
msg = content[4 : 4 + msg_len]
from_appid = content[4 + msg_len :].decode("utf-8")
if expected_appid and from_appid != expected_appid:
raise ValueError(f"appid mismatch: {from_appid!r} != {expected_appid!r}")
return msg.decode("utf-8")
def _sha1(*parts: str) -> str:
return hashlib.sha1("".join(sorted(parts)).encode("utf-8")).hexdigest()
def _consteq(a: str, b: str) -> bool:
return hmac.compare_digest(a, b)
def _pkcs7_unpad(data: bytes) -> bytes:
"""微信用块大小 32 的 PKCS7,末字节即 padding 长度(1..32)。越界则原样返回(容错)。"""
if not data:
return data
pad = data[-1]
if pad < 1 or pad > 32:
return data
return data[:-pad]
-6
View File
@@ -39,8 +39,6 @@ from app.api.v1.signin import router as signin_router
from app.api.v1.tasks import router as tasks_router
from app.api.v1.user import router as user_router
from app.api.v1.wallet import router as wallet_router
from app.api.v1.wx_kf import router as wx_kf_router
from app.api.v1.wx_mp import router as wx_mp_router
from app.api.v1.wxpay import router as wxpay_router
from app.core.config import settings
from app.core.daily_exchange_worker import (
@@ -142,10 +140,6 @@ app.include_router(internal_launch_confirm_router)
app.include_router(platform_router)
# CPS 群发短链跳转 /c/{code}(公网无鉴权:记点击 → 302 跳美团)
app.include_router(cps_redirect_router)
# 微信服务号消息接收回调 /wx/mp/callback(公网无鉴权:微信服务器验签, 截图比价入口)
app.include_router(wx_mp_router)
# 微信客服(企业微信)消息接收回调 /wx/kf/callback(服务号版的平行实现, 收事件→sync_msg 拉图)
app.include_router(wx_kf_router)
# 用户上传文件(头像)静态服务。生产可改由 nginx 直接 serve MEDIA_ROOT。
_media_root = Path(settings.MEDIA_ROOT)
+1
View File
@@ -32,6 +32,7 @@ from app.models.invite_fingerprint import InviteFingerprint # noqa: F401
from app.models.launch_confirm_sample import LaunchConfirmSample # noqa: F401
from app.models.meituan_coupon import MeituanCoupon # noqa: F401
from app.models.onboarding import OnboardingCompletion # noqa: F401
from app.models.phone_rebind_log import PhoneRebindLog # noqa: F401
from app.models.ops_marquee_seed import OpsMarqueeSeed # noqa: F401
from app.models.ops_stat_config import OpsStatConfig # noqa: F401
from app.models.price_observation import PriceObservation # noqa: F401
+31
View File
@@ -0,0 +1,31 @@
"""手机号换绑台账。
记录"手机号从老账号被夺走、重建为新账号(X 注销 → Y)"这一破坏性事件,支撑"一个手机号
30 天内最多换绑一次"的限制。手机号级、渠道无关(source 标来源);普通微信绑定不写此表。
M2 spec §4.1
"""
from __future__ import annotations
from datetime import datetime
from sqlalchemy import DateTime, Integer, String, func
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
class PhoneRebindLog(Base):
__tablename__ = "phone_rebind_log"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
# 被换绑的真实手机号(注意:存真实号,不是老账号被腾号后的 deleted_<id>)
phone: Mapped[str] = mapped_column(String(20), index=True, nullable=False)
# 被注销的老账号 X;P 换绑时已被腾空(极边界)则为空
old_user_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
# 换绑后新建的账号 Y
new_user_id: Mapped[int] = mapped_column(Integer, nullable=False)
# 换绑来源。手机号级配额、渠道无关,留字段给未来其他换绑路径共用同一份 30 天限制。
source: Mapped[str] = mapped_column(String(32), nullable=False, default="wechat_conflict")
rebound_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
)
+39
View File
@@ -0,0 +1,39 @@
"""手机号换绑台账(phone_rebind_log)的查询与写入。见 M2 spec §4.1。"""
from __future__ import annotations
import math
from datetime import datetime, timedelta, timezone
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from app.models.phone_rebind_log import PhoneRebindLog
def rebound_within_days(db: Session, phone: str, days: int) -> bool:
"""该手机号在最近 days 天内是否换绑过(命中 → 禁止再次换绑)。"""
since = datetime.now(timezone.utc) - timedelta(days=days)
stmt = (
select(PhoneRebindLog.id)
.where(PhoneRebindLog.phone == phone, PhoneRebindLog.rebound_at >= since)
.limit(1)
)
return db.execute(stmt).first() is not None
def remaining_block_days(db: Session, phone: str, days: int) -> int:
"""距离该手机号可再次换绑还剩几天(向上取整;无记录返回 0)。"""
last = db.execute(
select(func.max(PhoneRebindLog.rebound_at)).where(PhoneRebindLog.phone == phone)
).scalar_one_or_none()
if last is None:
return 0
if last.tzinfo is None: # SQLite 取回 naive datetime,按 UTC 归一
last = last.replace(tzinfo=timezone.utc)
remaining = (last + timedelta(days=days) - datetime.now(timezone.utc)).total_seconds()
return max(0, math.ceil(remaining / 86400))
def add_rebind_log(db: Session, *, phone: str, old_user_id: int | None, new_user_id: int, source: str) -> None:
"""写一条换绑台账(**不 commit**,交给调用方 rebind_account 的单事务)。"""
db.add(PhoneRebindLog(phone=phone, old_user_id=old_user_id, new_user_id=new_user_id, source=source))
+132
View File
@@ -12,6 +12,7 @@ from sqlalchemy import select
from sqlalchemy.orm import Session
from app.models.user import User
from app.repositories import phone_rebind
# ===== 创建时分配的标识:用户名(对外展示账号 ID)+ 默认昵称 =====
@@ -58,6 +59,17 @@ def is_default_nickname(nickname: str | None) -> bool:
)
def apply_wechat_display_identity(
user: User, *, wechat_nickname: str | None, wechat_avatar_url: str | None
) -> None:
"""§10:用已有账号绑微信时,仅当展示字段仍为默认才用微信昵称/头像替换(两规则独立);
自定义(改过昵称/传过头像)则保留只改内存对象,由调用方 commit"""
if is_default_nickname(user.nickname) and wechat_nickname:
user.nickname = wechat_nickname
if user.avatar_url is None and wechat_avatar_url:
user.avatar_url = wechat_avatar_url
def get_user_by_username(db: Session, username: str) -> User | None:
return db.execute(
select(User).where(User.username == username)
@@ -86,6 +98,85 @@ def get_user_by_phone(db: Session, phone: str) -> User | None:
return db.execute(stmt).scalar_one_or_none()
def get_user_by_wechat_openid(db: Session, openid: str) -> User | None:
stmt = select(User).where(User.wechat_openid == openid)
return db.execute(stmt).scalar_one_or_none()
def touch_last_login(db: Session, user: User) -> User:
"""openid 命中登录时更新 last_login_at(手机号登录在 upsert_user_for_login 里已更新)。"""
user.last_login_at = datetime.now(timezone.utc)
db.commit()
db.refresh(user)
return user
def attach_wechat_to_user(
db: Session, user: User, *, openid: str, wechat_nickname: str | None, wechat_avatar_url: str | None
) -> User:
"""继续绑定:把微信 openid + 微信源字段并入已存在账号(调用方保证 user.wechat_openid 为空)。
wechat_openid / wechat_nickname / wechat_avatar_url,并按 §10 规则回填展示字段:
仅当昵称仍为默认值(is_default_nickname)时用微信昵称替换,仅当头像为 null 时用微信头像替换;
用户已自定义的展示昵称/头像始终保留,两规则相互独立
openid 唯一约束(O 期间被别处绑走,极罕见)时由调用方捕获 IntegrityError 兜底降级为"只登入不绑"
"""
user.wechat_openid = openid
user.wechat_nickname = wechat_nickname
user.wechat_avatar_url = wechat_avatar_url
user.last_login_at = datetime.now(timezone.utc)
apply_wechat_display_identity(user, wechat_nickname=wechat_nickname, wechat_avatar_url=wechat_avatar_url)
db.commit()
db.refresh(user)
return user
def _build_wechat_user(
db: Session,
*,
phone: str,
openid: str,
wechat_nickname: str | None,
wechat_avatar_url: str | None,
) -> User:
"""构造并 db.add 一个微信账号行(register_channel='wechat',展示昵称头像取微信,缺则默认),
** commit**create_wechat_user rebind_account 共用,保证建号逻辑单一来源"""
user = User(
phone=phone,
username=_gen_unique_username(db),
nickname=wechat_nickname or _gen_nickname(),
avatar_url=wechat_avatar_url,
register_channel="wechat",
wechat_openid=openid,
wechat_nickname=wechat_nickname,
wechat_avatar_url=wechat_avatar_url,
last_login_at=datetime.now(timezone.utc),
)
db.add(user)
return user
def create_wechat_user(
db: Session,
*,
phone: str,
openid: str,
wechat_nickname: str | None,
wechat_avatar_url: str | None,
) -> User:
"""微信登录新建账号(未占用分支)。见 _build_wechat_user。
openid 唯一约束是并发/重复绑定的最终防线(极罕见,openid wechat-login 刚查过为空)
"""
user = _build_wechat_user(
db, phone=phone, openid=openid,
wechat_nickname=wechat_nickname, wechat_avatar_url=wechat_avatar_url,
)
db.commit()
db.refresh(user)
return user
def upsert_user_for_login(
db: Session,
*,
@@ -154,3 +245,44 @@ def soft_delete_account(db: Session, user: User) -> None:
# 释放邀请码唯一槽
user.invite_code = None
db.commit()
def rebind_account(
db: Session,
*,
phone: str,
openid: str,
wechat_nickname: str | None,
wechat_avatar_url: str | None,
source: str = "wechat_conflict",
) -> User:
"""换绑:**单事务内**注销老账号 X(腾出手机号)+ 用该号建全新微信账号 Y + 写换绑台账。
- 老账号可能已不存在(P 被腾空) old_user_id=None,直接建 Y(幂等更稳)
- 手机号唯一约束靠时序:先把 X.phone 改名并 flush 腾号,再插 Y
- 全程不中途 commit,任一步失败整体回滚,绝不出现"X 删了 Y 没建"
X 的字段变更等价 soft_delete_account(软删 + 匿名化 + 释放 openid/邀请码唯一槽),但不在此 commit
"""
old = get_user_by_phone(db, phone)
old_id = old.id if old is not None else None
if old is not None:
old.status = "deleted"
old.phone = f"deleted_{old.id}"
old.nickname = None
old.avatar_url = None
old.wechat_openid = None
old.wechat_nickname = None
old.wechat_avatar_url = None
old.invite_code = None
db.flush() # 先落 phone 改名,腾出手机号唯一约束,才能给 Y 用
new_user = _build_wechat_user(
db, phone=phone, openid=openid,
wechat_nickname=wechat_nickname, wechat_avatar_url=wechat_avatar_url,
)
db.flush() # 拿 new_user.id
phone_rebind.add_rebind_log(
db, phone=phone, old_user_id=old_id, new_user_id=new_user.id, source=source
)
db.commit()
db.refresh(new_user)
return new_user
+2
View File
@@ -20,6 +20,7 @@ from app.core.config import settings
from app.core.rewards import COIN_PER_CENT, coins_to_cents
from app.integrations import wxpay
from app.models.user import User
from app.repositories.user import apply_wechat_display_identity
from app.models.wallet import (
CashTransaction,
CoinAccount,
@@ -374,6 +375,7 @@ def bind_wechat_openid(db: Session, user_id: int, code: str) -> dict:
user.wechat_openid = info["openid"]
user.wechat_nickname = info["nickname"]
user.wechat_avatar_url = info["avatar_url"]
apply_wechat_display_identity(user, wechat_nickname=info["nickname"], wechat_avatar_url=info["avatar_url"])
db.commit()
return info
+61
View File
@@ -102,3 +102,64 @@ class RefreshRequest(BaseModel):
class LogoutResponse(BaseModel):
ok: bool = True
# ===== 微信登录 =====
class WechatLoginRequest(BaseModel):
code: str = Field(..., min_length=1, description="微信 App 授权拿到的 code(单次有效)")
device_id: str = Field(
"", max_length=64,
description="硬件级设备标识(Android ANDROID_ID),用于新手引导按 设备+账号 去重;空=按未完成处理",
)
class WechatLoginResponse(BaseModel):
# status="logged_in" → openid 命中,token 有值;"need_bind_phone" → 未命中,bind_ticket 有值
status: str
token: TokenWithUser | None = None
bind_ticket: str | None = None
wechat_nickname: str | None = None
wechat_avatar_url: str | None = None
class OccupiedAccountInfo(BaseModel):
"""手机号被占用时返回的原账号脱敏展示信息(供冲突页)。"""
nickname: str | None = None
avatar_url: str | None = None
created_at: datetime
has_wechat: bool = False
class WechatBindResultResponse(BaseModel):
# status="logged_in" → 未占用,已建号登入,token 有值;
# "phone_occupied" → 手机号被占用,occupied_account + conflict_ticket 有值,token 为 None
status: str
token: TokenWithUser | None = None
occupied_account: OccupiedAccountInfo | None = None
conflict_ticket: str | None = None # 占用时签发,换绑/继续绑定只认它
rebind_available: bool | None = None # 该手机号 30 天内是否还能换绑(给换绑按钮预置禁用态)
rebind_blocked_days: int | None = None # 被限时剩余天数(rebind_available=False 时>0)
class WechatBindPhoneSmsRequest(BaseModel):
bind_ticket: str = Field(..., min_length=1)
phone: str = Field(..., min_length=11, max_length=11, pattern=r"^1\d{10}$")
code: str = Field(..., min_length=4, max_length=8)
device_id: str = Field("", max_length=64)
class WechatBindPhoneJverifyRequest(BaseModel):
bind_ticket: str = Field(..., min_length=1)
login_token: str = Field(..., min_length=1, description="客户端 loginAuth 拿到的 loginToken")
device_id: str = Field("", max_length=64)
class WechatConflictContinueRequest(BaseModel):
conflict_ticket: str = Field(..., min_length=1)
device_id: str = Field("", max_length=64)
class WechatConflictRebindRequest(BaseModel):
conflict_ticket: str = Field(..., min_length=1)
device_id: str = Field("", max_length=64)
-14
View File
@@ -36,20 +36,6 @@ class OkResponse(BaseModel):
ok: bool = True
class PendingCompare(BaseModel):
"""PoC:后端通过心跳下发的"用户要从某源平台比价"信号。前端据此弹选平台窗 + launch 源平台。"""
source_platform: str # pricebot 源平台代号(PoC 写死 "meituan")
source_package: str # 源平台 Android 包名(前端 launch 用)
class HeartbeatResponse(BaseModel):
"""心跳响应。常规只回 ok;PoC 期带回 pending_compare 触发比价(exclude_none 省略 null)。"""
model_config = ConfigDict()
ok: bool = True
pending_compare: PendingCompare | None = None
class LivenessOut(BaseModel):
"""本机掉线告警状态(后置检测 pull 版)。客户端只需这一个布尔判断要不要弹「开启自启动」引导,
故只返回 kill_alert_pending(不暴露设备详情 / 内部 liveness_state )从未注册过 默认 False(无告警)"""
+61
View File
@@ -107,6 +107,67 @@ def test_sms_send_device_ip_rate_limit(client, monkeypatch) -> None:
assert r.status_code == 200, r.text
def test_sms_send_cooldown_reject_not_counted(client, monkeypatch) -> None:
"""发码额度只算「成功发码」:被单号 60s 冷却挡下的重发(429)不占设备额度。
做法:同号狂发只成功 1 其余被冷却挡下;把小时额度设 2,证明换号后仍能再成功发 1
若冷却重发也计数,额度早被那几次耗尽"""
from app.api.v1 import auth
from app.core import ratelimit
monkeypatch.setattr(ratelimit.settings, "RATE_LIMIT_ENABLED", True)
monkeypatch.setattr(auth, "SMS_SEND_MAX_PER_HOUR_PER_DEVICE", 2)
ratelimit._buckets.clear()
device = "dev-cooldown"
phone_a = "13710137000"
# 首发成功(小时闸计 1/2)
assert client.post(
"/api/v1/auth/sms/send", json={"phone": phone_a, "device_id": device}
).status_code == 200
# 同号连发 3 次:都被单号 60s 冷却挡下 → 429,且**不占**设备额度
for _ in range(3):
r = client.post(
"/api/v1/auth/sms/send", json={"phone": phone_a, "device_id": device}
)
assert r.status_code == 429, r.text
# 换号再发:设备额度只用了 1/2(冷却那几次没算)→ 仍放行(计到 2/2)
assert client.post(
"/api/v1/auth/sms/send", json={"phone": "13710137001", "device_id": device}
).status_code == 200
# 又换号:此时小时闸已 2/2 → 429(反证成功发码确实各计了 1)
r = client.post(
"/api/v1/auth/sms/send", json={"phone": "13710137002", "device_id": device}
)
assert r.status_code == 429, r.text
def test_sms_send_daily_cap(client, monkeypatch) -> None:
"""每天发码上限(设备 + IP):成功发码累计到日上限即 429(用不同手机号绕开单号冷却)。
抬高小时闸单独测日闸;超限文案含今日以便前端提示明天再来"""
from app.api.v1 import auth
from app.core import ratelimit
monkeypatch.setattr(ratelimit.settings, "RATE_LIMIT_ENABLED", True)
monkeypatch.setattr(auth, "SMS_SEND_MAX_PER_HOUR_PER_DEVICE", 100) # 抬高小时闸,不干扰
monkeypatch.setattr(auth, "SMS_SEND_MAX_PER_DAY_PER_DEVICE", 3)
ratelimit._buckets.clear()
device = "dev-daily"
for i in range(3):
r = client.post(
"/api/v1/auth/sms/send",
json={"phone": f"13720137{i:03d}", "device_id": device},
)
assert r.status_code == 200, f"{i + 1} 次应放行: {r.text}"
# 第 4 次:同设备同 IP 当日超限 → 429
r = client.post(
"/api/v1/auth/sms/send",
json={"phone": "13720137999", "device_id": device},
)
assert r.status_code == 429, r.text
assert "今日" in r.json()["detail"]
def test_sms_login_device_ip_rate_limit(client, monkeypatch) -> None:
"""防刷:同一设备(device_id) + 同一 IP 每小时最多 SMS_LOGIN_MAX_PER_HOUR 次登录尝试,超出 429。
conftest 默认 RATE_LIMIT_ENABLED=false(内存计数跨用例累加),本用例临时打开并清空计数隔离"""
+57
View File
@@ -0,0 +1,57 @@
"""ratelimit 内存桶过期清理(GC)测试。
回归重点:_buckets **全局共享**混着不同窗口(60s 广告 / 3600s 登录 / 86400s 日闸) key
GC 必须按每个 key 自己存的 window_sec判过期,而不是当前调用方的窗口 否则高频的 60s 端点
触发 GC 时会把本该存活更久的 3600s/86400s 计数(如短信日闸)一并删掉,使其被反复清零限流失效
monkeypatch _GC_THRESHOLD 0 强制每次都扫,免造上万条(仿 test_auth 里对 sms._GC_THRESHOLD 的做法)
"""
from __future__ import annotations
from app.core import ratelimit
def test_purge_expired_respects_each_key_own_window(monkeypatch) -> None:
"""短窗口(60s)触发的 GC 只删真正过期的 key,不得删掉仍在自身窗口内的长窗口 key。"""
monkeypatch.setattr(ratelimit, "_GC_THRESHOLD", 0) # 强制每次都扫
ratelimit._buckets.clear()
now = 1_000_000.0
# 日闸:100s 前开窗、window=86400 → 远未过期,必须保留
ratelimit._buckets["sms-send-device-daily:D:IP"] = (now - 100, 7, 86400.0)
# 登录:1800s、window=3600 → 未过期,保留
ratelimit._buckets["sms-login-device:D:IP"] = (now - 1800, 2, 3600.0)
# 广告:120s、window=60 → 已过期,应删
ratelimit._buckets["ad-watch-report:IP2"] = (now - 120, 3, 60.0)
ratelimit._purge_expired(now)
assert "sms-send-device-daily:D:IP" in ratelimit._buckets
assert "sms-login-device:D:IP" in ratelimit._buckets
assert "ad-watch-report:IP2" not in ratelimit._buckets
def test_purge_expired_keeps_long_window_key_older_than_short_window(monkeypatch) -> None:
"""反证旧 bug:日闸 key 已老于 3600s,旧代码在 60s/3600s 端点触发 GC 时会误删它;
现在按自身 86400s 窗口判 未过期 必须保留"""
monkeypatch.setattr(ratelimit, "_GC_THRESHOLD", 0)
ratelimit._buckets.clear()
now = 2_000_000.0
# 3700s 前开窗(> 1 小时),但 window=86400 → 未过期
ratelimit._buckets["sms-send-device-daily:D:IP"] = (now - 3700, 20, 86400.0)
ratelimit._purge_expired(now)
assert "sms-send-device-daily:D:IP" in ratelimit._buckets
def test_purge_expired_noop_below_threshold(monkeypatch) -> None:
"""未超阈值时不扫(即便有过期 key 也不动),避免每次请求都 O(n) 扫全表。"""
monkeypatch.setattr(ratelimit, "_GC_THRESHOLD", 10)
ratelimit._buckets.clear()
now = 3_000_000.0
ratelimit._buckets["stale:IP"] = (now - 999, 1, 60.0) # 早过期,但没超阈值
ratelimit._purge_expired(now)
assert "stale:IP" in ratelimit._buckets # 桶数没超阈值 → 不清理
+232
View File
@@ -0,0 +1,232 @@
"""微信登录 M2 测试:conflict_ticket 令牌、继续绑定(attach/只登入)、换绑(建号+软删+30天限)。
沿用 tests/test_wechat_login.py 风格:HTTP client;微信 codeopenid monkeypatch;
短信走 SMS_MOCK(任意 6 位过)数据变更用"再走一遍 wechat-login 看 openid 落在哪个账号"做行为断言
"""
from __future__ import annotations
import pytest
from app.api.v1 import auth # noqa: F401 (后续测试打桩 verify_and_get_phone 用)
from app.core import security
from app.integrations import wxpay
from app.models.phone_rebind_log import PhoneRebindLog
def _fake_userinfo(openid: str, nickname: str | None = "微信昵称", avatar: str | None = "http://x/a.png"):
def _f(code: str) -> dict:
return {"openid": openid, "nickname": nickname, "avatar_url": avatar, "raw": {}}
return _f
def _sms_occupy(client, phone: str) -> int:
"""用普通短信登录占用一个手机号(register_channel=sms),返回该账号 id。"""
assert client.post("/api/v1/auth/sms/send", json={"phone": phone}).status_code == 200
r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
assert r.status_code == 200, r.text
return r.json()["user"]["id"]
def _occupy_via_conflict(client, monkeypatch, openid: str, phone: str, device_id: str) -> dict:
"""微信登录(新 openid)→ 绑同一手机号 → 返回 phone_occupied 的响应体(含 conflict_ticket)。"""
monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo(openid))
ticket = client.post(
"/api/v1/auth/wechat-login", json={"code": "c", "device_id": device_id}
).json()["bind_ticket"]
r = client.post(
"/api/v1/auth/wechat/bind-phone/sms",
json={"bind_ticket": ticket, "phone": phone, "code": "123456", "device_id": device_id},
)
assert r.status_code == 200, r.text
body = r.json()
assert body["status"] == "phone_occupied"
return body
# ===== Task 1: 模型可导入(建表由 conftest 的 create_all 完成) =====
def test_phone_rebind_log_model_importable() -> None:
assert PhoneRebindLog.__tablename__ == "phone_rebind_log"
# ===== Task 2: conflict_ticket 令牌 =====
def test_conflict_ticket_roundtrip() -> None:
token = security.create_conflict_ticket(
openid="oid1", wechat_nickname="", wechat_avatar_url="http://a", phone="13900139000"
)
claims = security.decode_conflict_ticket(token)
assert claims["openid"] == "oid1"
assert claims["wnk"] == ""
assert claims["wav"] == "http://a"
assert claims["phone"] == "13900139000"
def test_conflict_ticket_wrong_type_rejected() -> None:
# bind_ticket 冒充 conflict_ticket → TokenError(typ 不匹配)
bind = security.create_bind_ticket(openid="oid", wechat_nickname=None, wechat_avatar_url=None)
with pytest.raises(security.TokenError):
security.decode_conflict_ticket(bind)
def test_conflict_ticket_expired_rejected(monkeypatch) -> None:
monkeypatch.setattr(security.settings, "WECHAT_BIND_TICKET_EXPIRE_MINUTES", -1)
token = security.create_conflict_ticket(
openid="oid", wechat_nickname=None, wechat_avatar_url=None, phone="13900139000"
)
with pytest.raises(security.TokenError):
security.decode_conflict_ticket(token)
# ===== Task 3: 占用响应扩展 =====
def test_phone_occupied_returns_conflict_ticket_and_flags(client, monkeypatch) -> None:
phone = "13900139101"
_sms_occupy(client, phone) # 老账号 X(sms,无微信)
body = _occupy_via_conflict(client, monkeypatch, "openid_occ_101", phone, "devO1")
assert body["conflict_ticket"]
assert body["rebind_available"] is True # 首次,未换绑过
assert body["rebind_blocked_days"] == 0
assert body["occupied_account"]["has_wechat"] is False # X 是 sms 账号
# ===== Task 4: 继续绑定 =====
def test_continue_attaches_wechat_and_logs_into_existing(client, monkeypatch) -> None:
"""X 无微信 → 继续绑定并入 openid + 登入 X;之后同 openid 登录直接命中 X。"""
phone = "13900139201"
x_id = _sms_occupy(client, phone) # X:sms 账号,无微信
body = _occupy_via_conflict(client, monkeypatch, "openid_cont_201", phone, "devC1")
r = client.post(
"/api/v1/auth/wechat/conflict/continue",
json={"conflict_ticket": body["conflict_ticket"], "device_id": "devC1"},
)
assert r.status_code == 200, r.text
assert r.json()["status"] == "logged_in"
assert r.json()["token"]["user"]["id"] == x_id # 登入的是老账号 X
# openid 现已并入 X:再走 wechat-login 直接命中 X
r = client.post("/api/v1/auth/wechat-login", json={"code": "c", "device_id": "devC1"})
assert r.json()["status"] == "logged_in"
assert r.json()["token"]["user"]["id"] == x_id
def test_continue_when_existing_has_wechat_logs_in_and_discards_openid(client, monkeypatch) -> None:
"""X 已绑别的微信 → 继续绑定只登入 X、丢弃本次 openid(不覆盖)。"""
phone = "13900139202"
# 先建一个已绑微信 O1 的账号 X(微信登录 O1 + 短信绑号)
monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_o1_202"))
t = client.post("/api/v1/auth/wechat-login", json={"code": "c", "device_id": "devC2"}).json()["bind_ticket"]
x = client.post(
"/api/v1/auth/wechat/bind-phone/sms",
json={"bind_ticket": t, "phone": phone, "code": "123456", "device_id": "devC2"},
).json()
x_id = x["token"]["user"]["id"]
# 新 openid O2 撞同号 → 占用(has_wechat=True)→ 继续绑定
body = _occupy_via_conflict(client, monkeypatch, "openid_o2_202", phone, "devC2b")
assert body["occupied_account"]["has_wechat"] is True
r = client.post(
"/api/v1/auth/wechat/conflict/continue",
json={"conflict_ticket": body["conflict_ticket"], "device_id": "devC2b"},
)
assert r.status_code == 200, r.text
assert r.json()["token"]["user"]["id"] == x_id # 登入 X
# O2 被丢弃:再走 wechat-login(O2)→ 仍未命中(need_bind_phone)
monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_o2_202"))
assert client.post(
"/api/v1/auth/wechat-login", json={"code": "c", "device_id": "devC2b"}
).json()["status"] == "need_bind_phone"
def test_continue_expired_ticket_returns_401(client, monkeypatch) -> None:
monkeypatch.setattr(security.settings, "WECHAT_BIND_TICKET_EXPIRE_MINUTES", -1)
expired = security.create_conflict_ticket(
openid="oid", wechat_nickname=None, wechat_avatar_url=None, phone="13900139209"
)
r = client.post(
"/api/v1/auth/wechat/conflict/continue",
json={"conflict_ticket": expired, "device_id": "devC3"},
)
assert r.status_code == 401, r.text
# ===== Task 5: 换绑 =====
def test_rebind_creates_new_account_and_binds_openid(client, monkeypatch) -> None:
"""换绑 → 建全新微信账号 Y(≠X)+ openid 落到 Y;老账号 X 被注销(手机号归 Y)。"""
phone = "13900139301"
x_id = _sms_occupy(client, phone)
body = _occupy_via_conflict(client, monkeypatch, "openid_rb_301", phone, "devR1")
r = client.post(
"/api/v1/auth/wechat/conflict/rebind",
json={"conflict_ticket": body["conflict_ticket"], "device_id": "devR1"},
)
assert r.status_code == 200, r.text
y = r.json()["token"]["user"]
assert r.json()["status"] == "logged_in"
assert y["phone"] == phone
assert y["register_channel"] == "wechat"
assert y["id"] != x_id # 是全新账号,不是老账号
# openid 落到 Y:再走 wechat-login 命中 Y
r = client.post("/api/v1/auth/wechat-login", json={"code": "c", "device_id": "devR1"})
assert r.json()["status"] == "logged_in"
assert r.json()["token"]["user"]["id"] == y["id"]
# ===== §10: continue 路径也应用展示身份回填规则 =====
def test_continue_applies_section10(client, monkeypatch) -> None:
"""§10 via M2 attach 路径: X 是默认昵称+null头像的 sms 账号;
continue 绑定微信后,展示昵称/头像应被微信值替换,并体现在响应的 token.user """
phone = "13900139211"
_sms_occupy(client, phone) # 建 X:默认昵称, null avatar
body = _occupy_via_conflict(
client, monkeypatch, "openid_s10", phone, "devS10"
) # fake_userinfo 默认 nickname="微信昵称", avatar="http://x/a.png"
r = client.post(
"/api/v1/auth/wechat/conflict/continue",
json={"conflict_ticket": body["conflict_ticket"], "device_id": "devS10"},
)
assert r.status_code == 200, r.text
user_out = r.json()["token"]["user"]
assert user_out["nickname"] == "微信昵称"
assert user_out["avatar_url"] == "http://x/a.png"
def test_rebind_blocked_within_30_days(client, monkeypatch) -> None:
"""同一手机号 30 天内二次换绑 → 409;占用响应 rebind_available=False。"""
phone = "13900139302"
_sms_occupy(client, phone)
body = _occupy_via_conflict(client, monkeypatch, "openid_rb_302a", phone, "devR2")
assert client.post(
"/api/v1/auth/wechat/conflict/rebind",
json={"conflict_ticket": body["conflict_ticket"], "device_id": "devR2"},
).status_code == 200
# 第二次:新 openid 撞同号 → 占用响应此时 rebind_available=False
body2 = _occupy_via_conflict(client, monkeypatch, "openid_rb_302b", phone, "devR2b")
assert body2["rebind_available"] is False
assert body2["rebind_blocked_days"] >= 1
r = client.post(
"/api/v1/auth/wechat/conflict/rebind",
json={"conflict_ticket": body2["conflict_ticket"], "device_id": "devR2b"},
)
assert r.status_code == 409, r.text
def test_rebind_expired_ticket_returns_401(client, monkeypatch) -> None:
monkeypatch.setattr(security.settings, "WECHAT_BIND_TICKET_EXPIRE_MINUTES", -1)
expired = security.create_conflict_ticket(
openid="oid", wechat_nickname=None, wechat_avatar_url=None, phone="13900139309"
)
r = client.post(
"/api/v1/auth/wechat/conflict/rebind",
json={"conflict_ticket": expired, "device_id": "devR3"},
)
assert r.status_code == 401, r.text
+198
View File
@@ -0,0 +1,198 @@
"""微信登录 M1 测试:bind_ticket 令牌、wechat-login(openid 命中/未命中)、
bind-phone(建号/占用/令牌过期)
沿用 tests/test_auth.py 风格:HTTP client fixture;微信 codeopenid monkeypatch
拦掉(conftest WECHAT_APP_ID/SECRET dummy,不真连微信);短信走 SMS_MOCK(任意 6 位通过)
"""
from __future__ import annotations
import pytest
from app.api.v1 import auth
from app.core import security
from app.integrations import wxpay
def _fake_userinfo(openid: str, nickname: str | None = "微信昵称", avatar: str | None = "http://x/a.png"):
"""返回一个可传给 monkeypatch 的假 code_to_userinfo(忽略 code,固定返回给定 openid)。"""
def _f(code: str) -> dict:
return {"openid": openid, "nickname": nickname, "avatar_url": avatar, "raw": {}}
return _f
# ===== Task 1: bind_ticket 令牌 =====
def test_bind_ticket_roundtrip() -> None:
token = security.create_bind_ticket(openid="oid1", wechat_nickname="", wechat_avatar_url="http://a")
claims = security.decode_bind_ticket(token)
assert claims["openid"] == "oid1"
assert claims["wnk"] == ""
assert claims["wav"] == "http://a"
def test_bind_ticket_wrong_type_rejected() -> None:
# 用 access token 冒充 bind_ticket → TokenError(typ 不匹配)
access, _ = security.create_token(user_id=1, token_type="access")
with pytest.raises(security.TokenError):
security.decode_bind_ticket(access)
def test_bind_ticket_expired_rejected(monkeypatch) -> None:
monkeypatch.setattr(security.settings, "WECHAT_BIND_TICKET_EXPIRE_MINUTES", -1)
token = security.create_bind_ticket(openid="oid", wechat_nickname=None, wechat_avatar_url=None)
with pytest.raises(security.TokenError):
security.decode_bind_ticket(token)
# ===== Task 2: wechat-login =====
def test_wechat_login_new_openid_returns_bind_ticket(client, monkeypatch) -> None:
monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_new_1", "小明", "http://x/m.png"))
r = client.post("/api/v1/auth/wechat-login", json={"code": "wxcode1", "device_id": "devA"})
assert r.status_code == 200, r.text
body = r.json()
assert body["status"] == "need_bind_phone"
assert body["bind_ticket"]
assert body["wechat_nickname"] == "小明"
assert body["wechat_avatar_url"] == "http://x/m.png"
assert body["token"] is None
def test_wechat_login_invalid_code_returns_400(client, monkeypatch) -> None:
def _raise(code: str) -> dict:
raise ValueError("微信授权失败: invalid code")
monkeypatch.setattr(wxpay, "code_to_userinfo", _raise)
r = client.post("/api/v1/auth/wechat-login", json={"code": "bad", "device_id": "devA"})
assert r.status_code == 400, r.text
# ===== Task 3: bind-phone/sms =====
def test_wechat_bind_sms_creates_account_then_openid_logs_in(client, monkeypatch) -> None:
"""未占用 → 建微信账号(channel=wechat,昵称头像取微信);再次同 openid 登录 → 直接登入同一账号。"""
monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_flow_2", "阿花", "http://x/h.png"))
phone = "13900139002"
# 1) 微信登录 → 未命中 → 拿 ticket
r = client.post("/api/v1/auth/wechat-login", json={"code": "c1", "device_id": "devB"})
ticket = r.json()["bind_ticket"]
assert ticket
# 2) 短信绑号(SMS_MOCK:任意 6 位通过)→ 建号 + 登入
r = client.post(
"/api/v1/auth/wechat/bind-phone/sms",
json={"bind_ticket": ticket, "phone": phone, "code": "123456", "device_id": "devB"},
)
assert r.status_code == 200, r.text
body = r.json()
assert body["status"] == "logged_in"
user = body["token"]["user"]
assert user["phone"] == phone
assert user["register_channel"] == "wechat"
assert user["nickname"] == "阿花"
assert user["avatar_url"] == "http://x/h.png"
uid = user["id"]
# 3) 再次微信登录(同 openid)→ 命中 → 直接登入同一账号
r = client.post("/api/v1/auth/wechat-login", json={"code": "c2", "device_id": "devB"})
assert r.status_code == 200, r.text
body = r.json()
assert body["status"] == "logged_in"
assert body["token"]["user"]["id"] == uid
def test_wechat_bind_sms_phone_occupied(client, monkeypatch) -> None:
"""手机号已被其他账号占用 → 返回 phone_occupied + 原账号信息(不建号)。"""
phone = "13900139003"
# 先用普通短信登录占用该手机号(register_channel=sms)
assert client.post("/api/v1/auth/sms/send", json={"phone": phone}).status_code == 200
r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
assert r.status_code == 200, r.text
occupied_nickname = r.json()["user"]["nickname"]
# 微信登录(新 openid)→ 未命中 → ticket
monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_occ_3"))
ticket = client.post(
"/api/v1/auth/wechat-login", json={"code": "c", "device_id": "devC"}
).json()["bind_ticket"]
# 绑同一手机号 → 占用
r = client.post(
"/api/v1/auth/wechat/bind-phone/sms",
json={"bind_ticket": ticket, "phone": phone, "code": "123456", "device_id": "devC"},
)
assert r.status_code == 200, r.text
body = r.json()
assert body["status"] == "phone_occupied"
assert body["token"] is None
assert body["occupied_account"]["nickname"] == occupied_nickname
assert body["occupied_account"]["avatar_url"] is None # 短信注册账号无头像 → 序列化为 null
assert body["occupied_account"]["created_at"]
def test_wechat_bind_sms_expired_ticket_returns_401(client, monkeypatch) -> None:
"""过期 bind_ticket → 401。"""
monkeypatch.setattr(security.settings, "WECHAT_BIND_TICKET_EXPIRE_MINUTES", -1)
expired = security.create_bind_ticket(openid="openid_exp", wechat_nickname="x", wechat_avatar_url=None)
r = client.post(
"/api/v1/auth/wechat/bind-phone/sms",
json={"bind_ticket": expired, "phone": "13900139009", "code": "123456", "device_id": "devD"},
)
assert r.status_code == 401, r.text
# ===== Task 4: bind-phone/jverify =====
def test_wechat_bind_jverify_creates_account(client, monkeypatch) -> None:
"""本机号(极光)绑定路径:verify_and_get_phone 拦掉,未占用 → 建号登入。"""
monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_jv_5", "极光用户", None))
phone = "13900139005"
# 极光 loginToken→手机号 在 auth 模块命名空间打桩(auth.py 顶部 from ...jiguang import verify_and_get_phone)
monkeypatch.setattr(auth, "verify_and_get_phone", lambda token: phone)
ticket = client.post(
"/api/v1/auth/wechat-login", json={"code": "c", "device_id": "devE"}
).json()["bind_ticket"]
r = client.post(
"/api/v1/auth/wechat/bind-phone/jverify",
json={"bind_ticket": ticket, "login_token": "jgtoken", "device_id": "devE"},
)
assert r.status_code == 200, r.text
body = r.json()
assert body["status"] == "logged_in"
user = body["token"]["user"]
assert user["phone"] == phone
assert user["register_channel"] == "wechat"
assert user["nickname"] == "极光用户"
# 微信 userinfo 隐私脱敏 avatar=None → 头像为空(客户端兜底默认头像)
assert user["avatar_url"] is None
def test_wechat_bind_jverify_expired_ticket_returns_401(client, monkeypatch) -> None:
"""过期 bind_ticket → 401(极光绑号路径,decode 先于极光核验)。"""
monkeypatch.setattr(security.settings, "WECHAT_BIND_TICKET_EXPIRE_MINUTES", -1)
expired = security.create_bind_ticket(openid="openid_jv_exp", wechat_nickname="x", wechat_avatar_url=None)
r = client.post(
"/api/v1/auth/wechat/bind-phone/jverify",
json={"bind_ticket": expired, "login_token": "jgtoken", "device_id": "devE"},
)
assert r.status_code == 401, r.text
def test_wechat_bind_jverify_jiguang_error_returns_502(client, monkeypatch) -> None:
"""极光核验失败(JiguangError)→ 502。"""
monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_jv_err"))
def _raise(token: str) -> str:
raise auth.JiguangError("mock jg failure")
monkeypatch.setattr(auth, "verify_and_get_phone", _raise)
ticket = client.post(
"/api/v1/auth/wechat-login", json={"code": "c", "device_id": "devE"}
).json()["bind_ticket"]
r = client.post(
"/api/v1/auth/wechat/bind-phone/jverify",
json={"bind_ticket": ticket, "login_token": "badtoken", "device_id": "devE"},
)
assert r.status_code == 502, r.text
+88
View File
@@ -312,6 +312,94 @@ def test_bind_rejects_openid_already_bound(client, monkeypatch) -> None:
assert r.status_code == 409, r.text
# ===== §10 绑微信时展示身份回填规则 =====
def test_bind_replaces_default_nickname_and_null_avatar(client, monkeypatch) -> None:
"""§10-A: 全默认(昵称=系统默认,头像=null) → 绑微信后两个展示字段都替换为微信值。"""
monkeypatch.setattr(
"app.integrations.wxpay.code_to_userinfo",
lambda code: {"openid": "openid_s10_a", "nickname": "微信昵称A", "avatar_url": "https://x/a.png", "raw": {}},
)
token = _login(client, "13800003001")
r = client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
assert r.status_code == 200, r.text
r = client.get("/api/v1/auth/me", headers=_auth(token))
assert r.status_code == 200, r.text
body = r.json()
assert body["nickname"] == "微信昵称A"
assert body["avatar_url"] == "https://x/a.png"
def test_bind_keeps_customized_nickname(client, monkeypatch) -> None:
"""§10-B: 用户已改过昵称 → 绑微信后昵称保留,但 null 头像仍替换为微信头像。"""
monkeypatch.setattr(
"app.integrations.wxpay.code_to_userinfo",
lambda code: {"openid": "openid_s10_b", "nickname": "微信昵称B", "avatar_url": "https://x/b.png", "raw": {}},
)
token = _login(client, "13800003002")
# 先把昵称改成非默认值
r = client.patch(
"/api/v1/user/profile", json={"nickname": "我的名字"}, headers=_auth(token)
)
assert r.status_code == 200, r.text
r = client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
assert r.status_code == 200, r.text
r = client.get("/api/v1/auth/me", headers=_auth(token))
body = r.json()
assert body["nickname"] == "我的名字" # 保留自定义昵称
assert body["avatar_url"] == "https://x/b.png" # null 头像被微信头像替换
def test_bind_keeps_customized_avatar(client, monkeypatch) -> None:
"""§10-C: 已有自定义头像 → 绑微信后头像保留,但默认昵称替换为微信昵称。"""
monkeypatch.setattr(
"app.integrations.wxpay.code_to_userinfo",
lambda code: {"openid": "openid_s10_c", "nickname": "微信昵称C", "avatar_url": "https://x/c.png", "raw": {}},
)
token = _login(client, "13800003003")
phone = "13800003003"
# 直接用 DB 给用户写入自定义头像(保持默认昵称)
db = SessionLocal()
try:
user = db.execute(select(User).where(User.phone == phone)).scalar_one()
user.avatar_url = "https://custom/av.png"
db.commit()
finally:
db.close()
r = client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
assert r.status_code == 200, r.text
r = client.get("/api/v1/auth/me", headers=_auth(token))
body = r.json()
assert body["nickname"] == "微信昵称C" # 默认昵称被替换
assert body["avatar_url"] == "https://custom/av.png" # 自定义头像保留
def test_bind_wechat_empty_keeps_default(client, monkeypatch) -> None:
"""§10-D: 微信侧 nickname/avatar 均为 None → 不覆盖,展示字段维持原默认值。"""
monkeypatch.setattr(
"app.integrations.wxpay.code_to_userinfo",
lambda code: {"openid": "openid_s10_d", "nickname": None, "avatar_url": None, "raw": {}},
)
token = _login(client, "13800003004")
r = client.get("/api/v1/auth/me", headers=_auth(token))
original_nickname = r.json()["nickname"] # 系统分配的默认昵称
assert original_nickname.startswith("用户")
r = client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
assert r.status_code == 200, r.text
r = client.get("/api/v1/auth/me", headers=_auth(token))
body = r.json()
assert body["nickname"] == original_nickname # 默认昵称不变
assert body["avatar_url"] is None # 头像仍为 null
def test_withdraw_reject_refunds(client, monkeypatch) -> None:
"""管理员审核拒绝 → 退回现金 + 单 rejected + 理由写入 fail_reason(用户可见)。"""
monkeypatch.setattr("app.integrations.wxpay.code_to_userinfo", lambda code: {"openid": "openid_reject", "nickname": None, "avatar_url": None, "raw": {}})