136 lines
5.6 KiB
Python
136 lines
5.6 KiB
Python
"""Auth 路由:platform_key 登录 / 管理员发用户 / 邮箱密码登录 / 改密码。
|
|
|
|
业务逻辑(bcrypt / users upsert)在 web/auth.py,这里只是 HTTP 壳。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime as _dt
|
|
from uuid import UUID
|
|
|
|
from fastapi import Depends, HTTPException
|
|
|
|
from ..auth import (
|
|
AuthConfig,
|
|
UserCreateError,
|
|
change_password,
|
|
create_user,
|
|
ensure_user_row,
|
|
get_user_profile,
|
|
mint_token,
|
|
resolve_user_by_email,
|
|
)
|
|
from ..schemas import (
|
|
AdminCreateUserRequest,
|
|
ChangePasswordRequest,
|
|
LoginRequest,
|
|
PasswordLoginRequest,
|
|
)
|
|
|
|
|
|
def register_auth_routes(app, *, require_user, auth_cfg: AuthConfig) -> None:
|
|
@app.post("/v1/auth/login", tags=["auth"])
|
|
def login(body: LoginRequest):
|
|
"""platform_key 校验通过 → 签 JWT(user_id 作为 sub)。
|
|
|
|
platform_key 错 → 403;user_id 非 UUID → 400。
|
|
user_id 未存在则幂等创建 users 行(避免下游 FK 失败);body 带 name / user_name
|
|
时一并 upsert 落库(平台侧改名每次登录自动同步,见 ensure_user_row)。
|
|
platform 服务端用此入口注入指定 user_id;dev SPA 走 /login_password。
|
|
"""
|
|
if body.platform_key != auth_cfg.platform_key:
|
|
raise HTTPException(403, "invalid platform_key")
|
|
try:
|
|
uid = UUID(body.user_id)
|
|
except (ValueError, TypeError):
|
|
raise HTTPException(400, f"invalid user_id (must be UUID): {body.user_id!r}")
|
|
ensure_user_row(uid, name=body.name, user_name=body.user_name)
|
|
token, exp = mint_token(auth_cfg, uid)
|
|
prof = get_user_profile(uid) or {}
|
|
return {
|
|
"token": token,
|
|
"expires_at": _dt.fromtimestamp(exp).isoformat(),
|
|
"user_id": str(uid),
|
|
"name": prof.get("name"),
|
|
"user_name": prof.get("user_name"),
|
|
"role": prof.get("role", "user"),
|
|
"ttl_seconds": auth_cfg.ttl_seconds,
|
|
}
|
|
|
|
@app.post("/v1/auth/admin/create_user", tags=["auth"])
|
|
def admin_create_user(body: AdminCreateUserRequest):
|
|
"""管理员发用户(dev SPA 登录页右下角入口)。
|
|
|
|
- `ZCBOT_ADMIN_TOKEN` env 未设 → 503,功能关闭
|
|
- `admin_token` 不匹配 → 403(不细分 "未设" / "错了",防探测)
|
|
- email 不合法 / password 太短 → 400
|
|
- email 已存在 → 409
|
|
- 成功 → `{"user_id": ..., "email": ...}`,前端提示 "已创建,请登录"
|
|
|
|
不签 token、不自动登录 —— 管理员发完用户用户自己登,逻辑清晰。
|
|
"""
|
|
if auth_cfg.admin_token is None:
|
|
raise HTTPException(503, "admin create_user disabled (ZCBOT_ADMIN_TOKEN not set)")
|
|
if body.admin_token != auth_cfg.admin_token:
|
|
raise HTTPException(403, "invalid admin_token")
|
|
try:
|
|
uid, email = create_user(
|
|
email=body.email, password=body.password, role=body.role
|
|
)
|
|
except UserCreateError as ex:
|
|
if ex.code in ("invalid_email", "weak_password", "invalid_role"):
|
|
raise HTTPException(400, ex.message)
|
|
if ex.code == "email_taken":
|
|
raise HTTPException(409, "email already exists")
|
|
raise HTTPException(500, f"create_user failed: {ex.message}")
|
|
return {"user_id": str(uid), "email": email, "role": body.role}
|
|
|
|
@app.post("/v1/auth/login_password", tags=["auth"])
|
|
def login_password(body: PasswordLoginRequest):
|
|
"""邮箱密码登录(dev SPA 给同事 / 自己试用)。
|
|
|
|
- users.email 未命中 / password_hash 为空 / bcrypt 校验失败 → 一律 403
|
|
(不细分错因,防探测用户存在性)
|
|
- 命中 → 直接用 DB 里现成 user_id 签 JWT(不 ensure_user_row,行已在 `user add` 时建)
|
|
- 发用户:`.venv/Scripts/python.exe main.py user add --email X --password Y`;
|
|
撤用户:`DELETE FROM users WHERE email=...`(先 DELETE 该 user 的 tasks)
|
|
"""
|
|
hit = resolve_user_by_email(body.email, body.password)
|
|
if hit is None:
|
|
raise HTTPException(403, "账号或密码错误")
|
|
uid, email = hit
|
|
token, exp = mint_token(auth_cfg, uid)
|
|
prof = get_user_profile(uid) or {}
|
|
return {
|
|
"token": token,
|
|
"expires_at": _dt.fromtimestamp(exp).isoformat(),
|
|
"user_id": str(uid),
|
|
"email": email,
|
|
"name": prof.get("name"),
|
|
"user_name": prof.get("user_name"),
|
|
"role": prof.get("role", "user"),
|
|
"ttl_seconds": auth_cfg.ttl_seconds,
|
|
}
|
|
|
|
@app.post("/v1/auth/change_password", tags=["auth"])
|
|
def change_password_route(
|
|
body: ChangePasswordRequest, user_id: UUID = Depends(require_user)
|
|
):
|
|
"""改密码(dev SPA 顶栏入口)。user_id 取自 JWT,不信任前端传值。
|
|
|
|
- 新密码 < 6 → 400
|
|
- 旧密码错 / 该账号无密码(platform_key 建的)→ 403(不细分,防探测)
|
|
- 用户不存在(JWT 有效但行没了)→ 401
|
|
成功 → `{"ok": true}`,前端提示并清空表单。
|
|
"""
|
|
try:
|
|
change_password(user_id, body.old_password, body.new_password)
|
|
except UserCreateError as ex:
|
|
if ex.code == "weak_password":
|
|
raise HTTPException(400, ex.message)
|
|
if ex.code in ("wrong_password", "no_password"):
|
|
raise HTTPException(403, ex.message)
|
|
if ex.code == "user_not_found":
|
|
raise HTTPException(401, ex.message)
|
|
raise HTTPException(500, f"change_password failed: {ex.message}")
|
|
return {"ok": True}
|