Files
DadaoTask/backend/tests/test_account.py
T
Steven 7558637567
CI / backend (push) Canceled after 0s
feat(auth): 短信登录默认关,换成手机号/邮箱 + 密码,发信邮箱进后台
用户要求:内测的固定验证码 123456 等于"知道手机号就能进任何账号",先关掉;
登录改用账号密码,并多一种"邮箱当账号名"的方式;发信邮箱要能在后台改。

短信那条路(默认关):
- `SMS_LOGIN_ENABLED=false`,`/auth/sms/send` 与 `/auth/login` 统一回 2017,不是静默失败;
- 生产校验改成"开着短信登录就必须配真通道",不开时 provider 是 dev 也不拦;
- 新增 `GET /auth/methods`,登录页据此渲染,关着的路不出现(少一次"点进去才发现不能用")。

邮箱登录:
- `app_user.email` + `uq_app_user_email`(`lower(email)` 部分索引),登录与绑定都归一成小写;
- `POST /auth/login/password` 收 `phone` 或 `email`(二选一),失败统一 2013、连错 10 次锁 15 分钟;
- 绑定在 `PATCH /auth/me`:格式错 2018、被占用 2019、留空串表示解绑。

系统设置(跨租户,只有平台管理员能改):
- 新表 `system_setting`(不挂租户、无 RLS,写权限由 `deps.platform_admin` 把关);
- `GET/PUT /settings/smtp`、`POST /settings/smtp/test`,授权码只写不读,留空表示沿用旧值;
- `services/mail.py` 只用标准库 smtplib:465 走 SSL、587 走 STARTTLS,认证/拒收/网络失败都归 2020 但消息不同;
- Web 新增「系统设置」页(状态卡 + 测试发信),侧栏只对平台管理员显示入口;账号页补「邮箱」。

验证:pytest 130 passed、ruff check/format 全绿、tsc 通过、vite build 通过;
`web/e2e/login-email-smtp.mjs` 15 项全绿(含用 126 授权码真发一封测试邮件)。

顺带记下的两个坑:`platform_admin_phones` 是 list,pydantic-settings 会按 JSON 解析,
要用 `NoDecode` 交给 before-validator 拆逗号;迁移号写成 0009 会撞既有的 0009,一并改成 0010。
2026-11-03 00:00:11 +08:00

190 lines
6.2 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""账号功能:昵称、密码设置/修改、密码登录。
这些都是"账号被占"的高风险入口,所以断言不只看 200,还要看:
- 密码一字节都不回给客户端(只回 has_password)
- 没设过密码的人不会被密码登录误放进去
- 改完密码,旧密码立刻失效
- 连错太多次要先锁一会儿,别让登录接口变成离线爆破的放大镜
"""
from __future__ import annotations
from httpx import AsyncClient
from .conftest import auth, login, unwrap
async def _me(client: AsyncClient, user: dict) -> dict:
return unwrap(await client.get("/api/v1/auth/me", headers=user["headers"]))
async def test_me_reports_profile_without_leaking_password(client: AsyncClient) -> None:
user = await login(client)
data = await _me(client, user)
assert data["user"]["has_password"] is False
assert data["user"]["phone"].endswith(user["phone"][-4:])
assert "****" in data["user"]["phone"]
# 只回「设没设」,不回密码本身:字段白名单比黑名单可靠
assert set(data["user"]) == {
"id",
"phone",
"email",
"nickname",
"avatar_url",
"has_password",
"platform_admin",
"created_at",
}
async def test_patch_nickname(client: AsyncClient) -> None:
user = await login(client)
data = unwrap(
await client.patch(
"/api/v1/auth/me", json={"nickname": " 老王 "}, headers=user["headers"]
)
)
assert data["user"]["nickname"] == "老王"
# 空昵称要被拦住,不能变成"无名氏"
resp = await client.patch("/api/v1/auth/me", json={"nickname": " "}, headers=user["headers"])
assert resp.json()["code"] == 1002
async def test_set_password_then_login_with_it(client: AsyncClient) -> None:
user = await login(client)
unwrap(
await client.post(
"/api/v1/auth/password",
json={"password": "dadao2026"},
headers=user["headers"],
)
)
assert (await _me(client, user))["user"]["has_password"] is True
# 验证码登录照旧可用,两条路并存
data = unwrap(
await client.post(
"/api/v1/auth/login/password",
json={"phone": user["phone"], "password": "dadao2026"},
)
)
assert data["tokens"]["access_token"]
assert data["user"]["id"] == user["user_id"]
async def test_password_login_rejects_wrong_password(client: AsyncClient) -> None:
user = await login(client)
unwrap(
await client.post(
"/api/v1/auth/password", json={"password": "dadao2026"}, headers=user["headers"]
)
)
resp = await client.post(
"/api/v1/auth/login/password", json={"phone": user["phone"], "password": "wrong-pass1"}
)
assert resp.status_code == 400
assert resp.json()["code"] == 2013
async def test_password_login_on_account_without_password(client: AsyncClient) -> None:
user = await login(client)
resp = await client.post(
"/api/v1/auth/login/password", json={"phone": user["phone"], "password": "dadao2026"}
)
# 没设过密码 = 密码不对,不能因为「没设置」就报 2014 把信息漏出去
assert resp.json()["code"] == 2013
async def test_unknown_phone_looks_the_same_as_wrong_password(client: AsyncClient) -> None:
user = await login(client)
unwrap(
await client.post(
"/api/v1/auth/password", json={"password": "dadao2026"}, headers=user["headers"]
)
)
resp = await client.post(
"/api/v1/auth/login/password", json={"phone": "13511112222", "password": "dadao2026"}
)
assert resp.json()["code"] == 2013
async def test_change_password_requires_old_one_and_invalidates_it(client: AsyncClient) -> None:
user = await login(client)
unwrap(
await client.post(
"/api/v1/auth/password", json={"password": "dadao2026"}, headers=user["headers"]
)
)
# 不给旧密码 → 拒绝
resp = await client.post(
"/api/v1/auth/password", json={"password": "dadao2027"}, headers=user["headers"]
)
assert resp.json()["code"] == 2013
# 旧密码给错 → 拒绝
resp = await client.post(
"/api/v1/auth/password",
json={"password": "dadao2027", "old_password": "nope-1234"},
headers=user["headers"],
)
assert resp.json()["code"] == 2013
unwrap(
await client.post(
"/api/v1/auth/password",
json={"password": "dadao2027", "old_password": "dadao2026"},
headers=user["headers"],
)
)
# 新密码能登,旧密码不能
unwrap(
await client.post(
"/api/v1/auth/login/password", json={"phone": user["phone"], "password": "dadao2027"}
)
)
resp = await client.post(
"/api/v1/auth/login/password", json={"phone": user["phone"], "password": "dadao2026"}
)
assert resp.json()["code"] == 2013
async def test_weak_password_rejected(client: AsyncClient) -> None:
user = await login(client)
for weak in ["short1", "12345678", "abcdefgh"]:
resp = await client.post(
"/api/v1/auth/password", json={"password": weak}, headers=user["headers"]
)
assert resp.json()["code"] == 2015, weak
async def test_password_endpoints_require_login(client: AsyncClient) -> None:
resp = await client.post("/api/v1/auth/password", json={"password": "dadao2026"})
assert resp.status_code == 401
resp = await client.patch("/api/v1/auth/me", json={"nickname": "x"}, headers=auth("nope"))
assert resp.status_code == 401
async def test_repeated_failures_are_throttled(client: AsyncClient) -> None:
user = await login(client)
unwrap(
await client.post(
"/api/v1/auth/password", json={"password": "dadao2026"}, headers=user["headers"]
)
)
for _ in range(10):
await client.post(
"/api/v1/auth/login/password", json={"phone": user["phone"], "password": "bad-pass1"}
)
resp = await client.post(
"/api/v1/auth/login/password", json={"phone": user["phone"], "password": "dadao2026"}
)
# 次数用光之后,连正确密码也要先等一会儿——锁的是这段时间,不是这一次请求
assert resp.json()["code"] == 2016