diff --git a/alembic/versions/phone_rebind_log.py b/alembic/versions/phone_rebind_log.py new file mode 100644 index 0000000..d6c0265 --- /dev/null +++ b/alembic/versions/phone_rebind_log.py @@ -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") diff --git a/app/core/config.py b/app/core/config.py index b8d554d..fe01e31 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -44,9 +44,11 @@ class Settings(BaseSettings): JWT_ALGORITHM: str = "HS256" JWT_ACCESS_TOKEN_EXPIRE_MINUTES: int = 120 JWT_REFRESH_TOKEN_EXPIRE_DAYS: int = 30 - # 微信登录未命中 openid 时签发的"待绑手机"令牌有效期(分钟)。 - # 见 auth.wechat_login / security.create_bind_ticket。 + # 微信登录未命中 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 无法越权访问后台。 @@ -84,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 包全流程联调用)===== @@ -109,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: diff --git a/app/models/__init__.py b/app/models/__init__.py index 694aa00..b6303b7 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -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 diff --git a/app/models/phone_rebind_log.py b/app/models/phone_rebind_log.py new file mode 100644 index 0000000..c28be70 --- /dev/null +++ b/app/models/phone_rebind_log.py @@ -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_) + 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 + ) diff --git a/tests/test_wechat_conflict.py b/tests/test_wechat_conflict.py new file mode 100644 index 0000000..093bc9c --- /dev/null +++ b/tests/test_wechat_conflict.py @@ -0,0 +1,49 @@ +"""微信登录 M2 测试:conflict_ticket 令牌、继续绑定(attach/只登入)、换绑(建号+软删+30天限)。 + +沿用 tests/test_wechat_login.py 风格:HTTP 走 client;微信 code→openid 用 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"