Tatta/DadaoTask
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(auth): 手机号也能用密码登录，验证码之外多一条路

Browse Source 
短信通道挂了人就进不来，所以给账号加第二把自助的钥匙：设过密码的用密码进，
没设过的照旧只用验证码，不强制。

- app_user.password_hash（迁移 0008 加列）+ scrypt 哈希（标准库实现，
  n=2^14/r=8/p=1，存 scrypt$n$r$p$salt$hash），不引 passlib
- POST /auth/login/password：没设过密码与密码错都回 2013，不区分账号是否存在；
  15 分钟内连错 10 次 → 2016，期间正确密码也挡住
- POST /auth/password：设/改共用，改要先验旧密码，新旧相同算弱密码 2015
- PATCH /auth/me 改昵称；用户输出走白名单 _user_out，密码哈希一个字节都不出去
- 登录页顶部「验证码 / 密码」两段切换

tests/test_account.py 10 项；全量 pytest 113 passed，ruff 全绿。
This commit is contained in:
Steven
2026-10-31 04:04:55 +08:00
parent 29aae12224
commit a5510cdb79
9 changed files with 544 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/auth.py
+132 -31
View File
@@ -1,4 +1,8 @@
"""认证：手机号 + 验证码。M0 不做密码、不做第三方登录。"""
"""认证：手机号 + 验证码，可选登录密码。
M0 只有验证码登录；密码是**可选**的第二条路：用户自己在账号设置里设一次，
之后登录页可以选「密码」；没设过密码的人只有验证码这一条，不受影响。
"""
from __future__ import annotations
@@ -14,7 +18,13 @@ from ..deps import CurrentUser, SessionDep
from ..errors import AppError, ErrorCode
from ..models import AppUser, Member, Workspace
from ..schemas import ok
from ..security import create_token_pair, decode_token
from ..security import (
create_token_pair,
decode_token,
hash_password,
password_problem,
verify_password,
)
from ..stores import get_store, sms_code_key, sms_throttle_key
router = APIRouter(prefix="/auth", tags=["认证"])
@@ -22,11 +32,36 @@ settings = get_settings()
PHONE_PATTERN = r"^1[3-9]\d{9}$"
# 密码登录连错这么多次就先锁一会儿：暴力猜密码的成本必须比猜验证码高
PASSWORD_FAIL_LIMIT = 10
PASSWORD_FAIL_TTL = 900
def mask_phone(phone: str) -> str:
return f"{phone[:3]}****{phone[-4:]}" if len(phone) == 11 else "***"
def _user_out(user: AppUser) -> dict:
"""对外只给脱敏手机号；密码本身一个字节都不出去，只告诉客户端「设没设」。"""
return {
"id": str(user.id),
"phone": mask_phone(user.phone),
"nickname": user.nickname,
"avatar_url": user.avatar_url,
"has_password": bool(user.password_hash),
"created_at": user.created_at,
}
def _tokens_out(tokens) -> dict:
return {
"access_token": tokens.access_token,
"refresh_token": tokens.refresh_token,
"token_type": "bearer",
"expires_in": tokens.expires_in,
}
class SendCodeIn(BaseModel):
phone: str = Field(pattern=PHONE_PATTERN, examples=["13800138000"])
@@ -36,6 +71,22 @@ class LoginIn(BaseModel):
code: str = Field(min_length=4, max_length=8)
class PasswordLoginIn(BaseModel):
phone: str = Field(pattern=PHONE_PATTERN)
password: str = Field(min_length=1, max_length=64)
class ProfilePatch(BaseModel):
nickname: str | None = Field(default=None, min_length=1, max_length=32)
class PasswordIn(BaseModel):
"""old_password 只在「已经设过密码」时必填；第一次设密码不需要。"""
password: str = Field(min_length=1, max_length=64)
old_password: str | None = Field(default=None, max_length=64)
class RefreshIn(BaseModel):
refresh_token: str = Field(min_length=10)
@@ -81,40 +132,54 @@ async def login(body: LoginIn, session: SessionDep) -> dict:
await session.flush()
tokens = create_token_pair(user.id)
return ok(
{
"user": {
"id": str(user.id),
"phone": mask_phone(user.phone),
"nickname": user.nickname,
"created_at": user.created_at,
},
"tokens": {
"access_token": tokens.access_token,
"refresh_token": tokens.refresh_token,
"token_type": "bearer",
"expires_in": tokens.expires_in,
},
}
)
return ok({"user": _user_out(user), "tokens": _tokens_out(tokens)})
@router.post("/login/password", summary="密码登录")
async def login_password(body: PasswordLoginIn, session: SessionDep) -> dict:
"""手机号 + 密码登录。
两个细节：
- **不区分"没这个人"和"密码错"**，同一个错误码，别让登录接口变成手机号探测工具。
- 连错 PASSWORD_FAIL_LIMIT 次锁 PASSWORD_FAIL_TTL 秒；成功即清零，正常用户不会感知。
"""
store = get_store()
fail_key = f"pwd_fail:{body.phone}"
if int(await store.get(fail_key) or 0) >= PASSWORD_FAIL_LIMIT:
raise AppError(
ErrorCode.LOGIN_TOO_MANY,
f"密码错误次数过多，请 {PASSWORD_FAIL_TTL // 60} 分钟后再试，或用验证码登录",
)
user = (
await session.execute(select(AppUser).where(AppUser.phone == body.phone))
).scalar_one_or_none()
if user is None or not verify_password(body.password, user.password_hash):
await store.incr(fail_key, PASSWORD_FAIL_TTL)
raise AppError(ErrorCode.PASSWORD_INVALID, "手机号或密码不对")
if user.status != "active":
raise AppError(ErrorCode.ACCOUNT_DISABLED, "账号已停用")
await store.delete(fail_key)
tokens = create_token_pair(user.id)
return ok({"user": _user_out(user), "tokens": _tokens_out(tokens)})
@router.post("/refresh", summary="刷新访问令牌")
async def refresh(body: RefreshIn) -> dict:
user_id = decode_token(body.refresh_token, "refresh")
tokens = create_token_pair(user_id)
return ok(
{
"access_token": tokens.access_token,
"refresh_token": tokens.refresh_token,
"token_type": "bearer",
"expires_in": tokens.expires_in,
}
)
# 这里的形状保持扁平（不带 tokens 信封），老客户端与安卓端都按这个取
return ok(_tokens_out(tokens))
@router.get("/me", summary="当前用户与所属工作区")
async def me(user: CurrentUser, session: SessionDep) -> dict:
row = await session.get(AppUser, user.id)
if row is None:
raise AppError(ErrorCode.UNAUTHORIZED, "账号不存在")
rows = (
await session.execute(
select(Workspace, Member.role)
@@ -126,11 +191,7 @@ async def me(user: CurrentUser, session: SessionDep) -> dict:
return ok(
{
"user": {
"id": str(user.id),
"phone": mask_phone(user.phone),
"nickname": user.nickname,
},
"user": _user_out(row),
"workspaces": [
{
"id": str(ws.id),
@@ -144,3 +205,43 @@ async def me(user: CurrentUser, session: SessionDep) -> dict:
],
}
)
@router.patch("/me", summary="改自己的昵称")
async def update_me(body: ProfilePatch, user: CurrentUser, session: SessionDep) -> dict:
row = await session.get(AppUser, user.id)
if row is None:
raise AppError(ErrorCode.UNAUTHORIZED, "账号不存在")
if body.nickname is not None:
nickname = body.nickname.strip()
if not nickname:
raise AppError(ErrorCode.VALIDATION_FAILED, "昵称不能为空")
row.nickname = nickname
await session.flush()
return ok({"user": _user_out(row)})
@router.post("/password", summary="设置或修改登录密码")
async def set_password(body: PasswordIn, user: CurrentUser, session: SessionDep) -> dict:
"""已经设过密码的，必须先给旧密码——否则拿到一个没锁屏的浏览器就能把密码换掉。"""
row = await session.get(AppUser, user.id)
if row is None:
raise AppError(ErrorCode.UNAUTHORIZED, "账号不存在")
if row.password_hash:
if not body.old_password:
raise AppError(ErrorCode.PASSWORD_INVALID, "请输入当前密码")
if not verify_password(body.old_password, row.password_hash):
raise AppError(ErrorCode.PASSWORD_INVALID, "当前密码不对")
if hmac.compare_digest(body.old_password, body.password):
raise AppError(ErrorCode.PASSWORD_WEAK, "新密码不能和当前密码一样")
problem = password_problem(body.password)
if problem:
raise AppError(ErrorCode.PASSWORD_WEAK, problem)
row.password_hash = hash_password(body.password)
await session.flush()
return ok({"has_password": True})
backend/app/errors.py
+8
View File
@@ -25,6 +25,10 @@ class ErrorCode(IntEnum):
SMS_CODE_INVALID = 2010
SMS_TOO_FREQUENT = 2011
ACCOUNT_DISABLED = 2012
PASSWORD_INVALID = 2013
PASSWORD_NOT_SET = 2014
PASSWORD_WEAK = 2015
LOGIN_TOO_MANY = 2016
# 3xxx 租户 / 权限
WORKSPACE_NOT_FOUND = 3001
@@ -86,6 +90,10 @@ _HTTP_STATUS: dict[ErrorCode, int] = {
ErrorCode.SMS_CODE_INVALID: 400,
ErrorCode.SMS_TOO_FREQUENT: 429,
ErrorCode.ACCOUNT_DISABLED: 403,
ErrorCode.PASSWORD_INVALID: 400,
ErrorCode.PASSWORD_NOT_SET: 400,
ErrorCode.PASSWORD_WEAK: 422,
ErrorCode.LOGIN_TOO_MANY: 429,
ErrorCode.WORKSPACE_NOT_FOUND: 404,
ErrorCode.NOT_A_MEMBER: 403,
ErrorCode.INSUFFICIENT_ROLE: 403,
backend/app/models.py
+2
View File
@@ -45,6 +45,8 @@ class AppUser(Base, Timestamps):
nickname: Mapped[str] = mapped_column(String(64), nullable=False, default="")
avatar_url: Mapped[str | None] = mapped_column(String(512))
status: Mapped[str] = mapped_column(String(16), nullable=False, default="active")
# 登录密码（scrypt 串，见 security.hash_password）。为空表示只用验证码登录。
password_hash: Mapped[str | None] = mapped_column(String(255))
class Workspace(Base, Timestamps):
backend/app/security.py
+63
View File
@@ -8,6 +8,8 @@
from __future__ import annotations
import base64
import hashlib
import hmac
import secrets
import time
@@ -78,3 +80,64 @@ def new_invite_code() -> str:
def invite_code_matches(provided: str, expected: str) -> bool:
return hmac.compare_digest(provided.strip().upper(), expected)
# —— 登录密码 ——
# 用标准库 scrypt，不引入 passlib/bcrypt：少一个依赖，也就少一处装不上的风险。
# 存储格式 `scrypt$n$r$p$salt_b64$hash_b64`，参数写进串里，将来调强度不用改代码。
_SCRYPT_N = 2**14
_SCRYPT_R = 8
_SCRYPT_P = 1
_PASSWORD_MIN = 8
_PASSWORD_MAX = 64
def hash_password(password: str) -> str:
salt = secrets.token_bytes(16)
digest = hashlib.scrypt(
password.encode("utf-8"), salt=salt, n=_SCRYPT_N, r=_SCRYPT_R, p=_SCRYPT_P, dklen=32
)
return "$".join(
[
"scrypt",
str(_SCRYPT_N),
str(_SCRYPT_R),
str(_SCRYPT_P),
base64.b64encode(salt).decode(),
base64.b64encode(digest).decode(),
]
)
def verify_password(password: str, stored: str | None) -> bool:
"""stored 为空（没设过密码）一律返回 False，别把"没密码"当成"密码对"。"""
if not stored:
return False
try:
scheme, n, r, p, salt_b64, hash_b64 = stored.split("$")
if scheme != "scrypt":
return False
digest = hashlib.scrypt(
password.encode("utf-8"),
salt=base64.b64decode(salt_b64),
n=int(n),
r=int(r),
p=int(p),
dklen=len(base64.b64decode(hash_b64)),
)
except (ValueError, TypeError):
return False
return hmac.compare_digest(digest, base64.b64decode(hash_b64))
def password_problem(password: str) -> str | None:
"""返回不合格的原因；合格返回 None。规则写在一处，设置与修改都用它。"""
if len(password) < _PASSWORD_MIN:
return f"密码至少 {_PASSWORD_MIN} 位"
if len(password) > _PASSWORD_MAX:
return f"密码最多 {_PASSWORD_MAX} 位"
has_letter = any(c.isalpha() for c in password)
has_digit = any(c.isdigit() for c in password)
if not (has_letter and has_digit):
return "密码要同时有字母和数字"
return None