Files
server-core/app/api/routers/auth.py
T
Pine 9d60e0afc6 feat(auth): 平台成唯一账号源——小程序微信登录进 users + OPC 报名资料字段
统一用户账号体系(阶段1,非破坏):
- UserRepository 新增 get_by_wx_mini_openid。
- /auth/wx-login:小程序登录(X-Client=miniprogram)按 wx_mini_openid 建/复用
  users 账号(role=opc_member, source=mini_program, auth_type=wx_openid),
  并带 nickName/avatarUrl;非小程序仍走 wx_openid。
- /auth/me + 登录响应补齐小程序契约字段 name/phone/phoneBound。
- users 增 opc_status/topics 列(alembic 0010);desktop/web profile 也输出
  name/phone/phoneBound/status/topics,且 update-profile 支持 name/status/topics 映射。
- scripts/db/seed.py 增 accounts→users 幂等迁移:旧小程序账号并入全局 users
  (wx_mini_openid/phone 关联 + opc_member 身份),由用户执行 migrate+seed。

Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-25 19:59:34 +08:00

550 lines
22 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.
# -*- coding: utf-8 -*-
"""认证路由:登录 / 注册 / 状态 / 校验 / 资料 / 令牌管理。
这是供 PineAgents 主后端 ``src/pineagents/app/routers/auth.py`` 转发调用的
"演示 FastAPI"DEMO_API_BASE_URL,默认 http://127.0.0.1:8090)。
路由前缀 ``/auth``,字段与主后端转发模型完全一致,保证透传不丢字段。
"""
from __future__ import annotations
import json
import re
import secrets
from fastapi import APIRouter, Depends, HTTPException, Request
from ... import config
from ..dependencies import get_current_user, get_db, extract_bearer_token
from ..schemas.auth import (
AuthStatusResponse,
IdentityInfo,
LoginRequest,
LoginResponse,
PhoneLoginRequest,
ProfileResponse,
RegisterRequest,
RevokeTokenRequest,
SelectIdentityRequest,
SendCodeRequest,
SendCodeResponse,
UpdateProfileRequest,
VerifyResponse,
WxLoginRequest,
WxPhoneRequest,
)
from ...infrastructure.repositories import Database
from ...services import sms, wechat
router = APIRouter(prefix="/auth", tags=["auth"])
def _parse_topics(raw) -> list:
"""topics(JSON 文本/list) → list;失败兜底 []。"""
if not raw:
return []
if isinstance(raw, list):
return raw
if isinstance(raw, str):
try:
v = json.loads(raw)
return v if isinstance(v, list) else []
except (TypeError, ValueError):
return []
return []
# 用户来源(X-Client 头,回退 body.source;未指认则按 auth_type 推断)
_SOURCE_BY_CLIENT = {
"miniprogram": "mini_program", "mini_program": "mini_program", "mp": "mini_program",
"web": "web", "browser": "web",
"desktop": "desktop", "app": "desktop", "tauri": "desktop",
"park": "park", "admin": "admin", "enterprise": "enterprise",
"carrier": "carrier", "government": "government", "provider": "provider", "operator": "operator",
}
def _client_source(request: Request, body_source: str = "", auth_type: str = "") -> str:
"""确定用户来源:X-Client 头 > body.source > 按 auth_type 推断。"""
client = (request.headers.get("X-Client", "") or "").strip().lower()
if client in _SOURCE_BY_CLIENT:
return _SOURCE_BY_CLIENT[client]
if body_source in _SOURCE_BY_CLIENT.values():
return body_source
if body_source:
return body_source
if auth_type == "wx_openid":
return "wx"
if auth_type == "phone":
return "phone"
return "web"
def _client_ip(request: Request) -> str:
xff = request.headers.get("X-Forwarded-For", "")
if xff:
return xff.split(",")[0].strip()
return request.client.host if (request.client and request.client.address) else ""
# 账号统一使用手机号作为登录账号(11 位,1 开头)
_PHONE_RE = re.compile(r"^1\d{10}$")
def _is_phone(account: str) -> bool:
return bool(_PHONE_RE.match(account.strip()))
# 账户管理页可编辑的资料字段(username 作为登录名走 update_credentials,单独处理)
_PROFILE_FIELDS = (
"nickname",
"account",
"room",
"avatar",
"company_avatar",
"email",
"gender",
"birthday",
)
def _identity_summaries(identities: list[dict]) -> list[IdentityInfo]:
"""把 user_identities 记录裁剪成对外身份摘要。"""
return [
IdentityInfo(
id=i["id"], port=i["port"], role=i["role"], sub_role=i.get("sub_role"),
name=i.get("name", ""), org_id=i.get("org_id"), region_id=i.get("region_id"),
org_name=i.get("org_name"), region_name=i.get("region_name"),
status=i.get("status", "active"),
)
for i in identities
]
async def _issue_token(
db: Database,
user: dict,
expires_in: int | None,
identity: dict | None = None,
) -> dict:
"""为指定用户签发 JWT(可按端口身份作用域),返回 token 记录。
``identity`` 非空时按该身份解析 role/sub_role/org/region 与权限;
否则回退到 ``users.role``(单角色/中性令牌)。
"""
role = (identity or user).get("role", "opc_member")
sub_role = (identity or user).get("sub_role")
org_id = (identity or user).get("org_id")
region_id = (identity or user).get("region_id")
identity_id = identity.get("id") if identity else None
perms = await db.roles.permissions_for(role, sub_role)
scope_ids = await db.regions.visible_region_ids(region_id)
scope_level = await db.regions.level(region_id)
token_user = dict(user)
token_user.update(role=role, sub_role=sub_role, org_id=org_id, region_id=region_id)
await db.users.mark_login(user["id"]) # 记录最近登录时间(每次签发令牌=一次活动)
return await db.tokens.create(
token_user,
permissions=perms,
scope_region_ids=scope_ids,
scope_level=scope_level,
expiry_seconds=config.resolve_token_expiry(expires_in),
identity_id=identity_id,
)
async def _profile_for(
user: dict,
identity: dict | None,
db: Database,
) -> dict:
"""按身份(或回退用户单角色)构建对外资料字典。"""
role = (identity or user).get("role", "opc_member")
sub_role = (identity or user).get("sub_role")
org_id = (identity or user).get("org_id")
region_id = (identity or user).get("region_id")
profile = {f: user.get(f, "") for f in _PROFILE_FIELDS}
profile.update(username=user.get("username", ""))
profile["name"] = user.get("nickname") or user.get("username", "")
profile["phone"] = user.get("phone", "")
profile["phoneBound"] = bool(user.get("phone", ""))
profile["status"] = user.get("opc_status", "")
profile["topics"] = _parse_topics(user.get("topics", ""))
profile["source"] = user.get("source", "")
profile["auth_type"] = user.get("auth_type", "")
profile["company"] = user.get("company", "")
profile["compute_provisioned"] = bool(user.get("compute_provisioned", False))
profile["compute_quota"] = user.get("compute_quota", 0)
profile["compute_used_quota"] = user.get("compute_used_quota", 0)
profile["certification_status"] = user.get("certification_status", "uncertified")
profile["affiliation"] = user.get("affiliation", "independent")
profile["park_id"] = user.get("park_id", "")
profile["park_name"] = user.get("park_name", "")
profile["account_type"] = user.get("account_type", "opc_default")
profile["role"] = role
profile["sub_role"] = sub_role
profile["org_id"] = org_id
profile["region_id"] = region_id
profile["permissions"] = await db.roles.permissions_for(role, sub_role)
profile["scope_region_ids"] = await db.regions.visible_region_ids(region_id)
if identity:
profile["identity_id"] = identity["id"]
profile["port"] = identity["port"]
profile["identity_name"] = identity.get("name", "")
return profile
@router.post("/login", response_model=LoginResponse, summary="登录")
async def login(req: LoginRequest, db: Database = Depends(get_db)):
"""校验用户名密码,返回账号绑定的全部端口身份 + 令牌。
- 仅一个身份:令牌直接绑定该身份(自动登录)。
- 多个身份:签发中性账号令牌,前端展示身份选择,经
``/auth/select-identity`` 切换到指定身份后进入对应端口。
"""
user = await db.users.get_by_username(req.username)
if user is None or not await db.users.verify_password(user, req.password):
raise HTTPException(status_code=401, detail="Invalid username or password")
if user.get("status") != "active":
raise HTTPException(status_code=403, detail="Account is disabled")
identities = await db.identities.list_for_user(user["id"], active_only=True)
identity = identities[0] if len(identities) == 1 else None
token_record = await _issue_token(db, user, req.expires_in, identity=identity)
profile = await _profile_for(user, identity, db)
await db.audit.add(
action="login", resource="auth", resource_id=user["id"],
detail=f"login {user['username']} ({len(identities)} identities)",
user_id=user["id"],
)
return LoginResponse(
token=token_record["token"],
identities=_identity_summaries(identities),
**profile,
)
@router.post("/select-identity", response_model=LoginResponse, summary="切换端口身份")
async def select_identity(
req: SelectIdentityRequest,
user: dict = Depends(get_current_user),
db: Database = Depends(get_db),
):
"""把当前令牌切换到指定端口身份(该身份须属于当前账号且为启用状态)。"""
ident = await db.identities.get_for_user(req.identity_id, user["id"])
if ident is None or ident.get("status") != "active":
raise HTTPException(status_code=404, detail="Identity not found or disabled")
token_record = await _issue_token(db, user, None, identity=ident)
profile = await _profile_for(user, ident, db)
identities = await db.identities.list_for_user(user["id"], active_only=True)
return LoginResponse(
token=token_record["token"],
identities=_identity_summaries(identities),
**profile,
)
@router.post("/register", response_model=LoginResponse, summary="注册")
async def register(req: RegisterRequest, request: Request, db: Database = Depends(get_db)):
"""注册唯一账户(演示端已存在 pine,故返回 403)。"""
if not _is_phone(req.username):
raise HTTPException(
status_code=400,
detail="账号需为 11 位手机号(如 138xxxx1234",
)
if not config.AUTH_ENABLED:
raise HTTPException(status_code=403, detail="Authentication is not enabled")
if await db.users.has_users():
raise HTTPException(status_code=403, detail="User already registered")
if not req.username.strip() or not req.password.strip():
raise HTTPException(
status_code=400,
detail="Username and password are required",
)
user = await db.users.create(
req.username, req.password,
source=_client_source(request, getattr(req, "source", ""), "password"),
auth_type="password", register_ip=_client_ip(request),
)
identity = await db.identities.create(
user["id"], port="opc", role="opc_member",
sub_role="independent", name="独立OPC",
)
token_record = await _issue_token(db, user, req.expires_in, identity=identity)
profile = await _profile_for(user, identity, db)
return LoginResponse(
token=token_record["token"],
identities=_identity_summaries([identity]),
**profile,
)
@router.get("/status", response_model=AuthStatusResponse, summary="认证状态")
async def auth_status(db: Database = Depends(get_db)):
"""前端登录页据此判断是否展示登录表单。本期登录模式:账号密码 + 手机号验证码;微信扫码预留(未开放)。"""
return AuthStatusResponse(
enabled=config.AUTH_ENABLED,
has_users=await db.users.has_users(),
login_modes=["password", "phone"] + (["wechat"] if config.AUTH_WECHAT_LOGIN else []),
wechat_login=config.AUTH_WECHAT_LOGIN,
)
@router.get("/verify", response_model=VerifyResponse, summary="校验令牌")
async def verify(request: Request, db: Database = Depends(get_db)):
"""校验调用方 Bearer 令牌;无效令牌由依赖层直接返回 401。"""
if not config.AUTH_ENABLED:
return VerifyResponse(valid=True, username="")
user = await get_current_user(request, db)
return VerifyResponse(valid=True, username=user["username"])
@router.get("/me", response_model=dict, summary="当前用户资料")
async def me(
user: dict = Depends(get_current_user),
db: Database = Depends(get_db),
):
"""返回当前登录用户的完整资料(含角色/组织/区域/权限/数据范围)。"""
return await db.users.to_profile(user)
@router.post("/update-profile", response_model=ProfileResponse, summary="更新资料/凭据")
async def update_profile(
req: UpdateProfileRequest,
user: dict = Depends(get_current_user),
db: Database = Depends(get_db),
):
"""更新演示资料与/或用户名、密码,返回更新后的完整资料。
- 修改用户名/密码时必须提供正确的 ``current_password``
- 修改凭据后吊销该用户其余会话并签发新令牌随响应返回;
- 仅改资料字段时不重签令牌。
"""
payload = req.model_dump(exclude_none=True)
user_id = user["id"]
new_username = payload.get("new_username")
new_password = payload.get("new_password")
if new_username is not None and not new_username.strip():
raise HTTPException(status_code=400, detail="Username cannot be empty")
if new_password is not None and not new_password.strip():
raise HTTPException(status_code=400, detail="Password cannot be empty")
profile_updates = {k: payload[k] for k in _PROFILE_FIELDS if k in payload}
# 小程序契约字段 → users 列映射(name→nicknamestatus→opc_statustopics→JSON文本)
if "name" in payload:
profile_updates["nickname"] = payload["name"]
if "status" in payload:
profile_updates["opc_status"] = payload["status"]
if "topics" in payload:
t = payload["topics"]
profile_updates["topics"] = json.dumps(t, ensure_ascii=False) if isinstance(t, list) else (t or "")
changing_credentials = new_username is not None or new_password is not None
if not profile_updates and not changing_credentials:
raise HTTPException(status_code=400, detail="Nothing to update")
if changing_credentials and not await db.users.verify_password(user, req.current_password):
raise HTTPException(status_code=401, detail="Current password is incorrect")
if profile_updates:
await db.users.update_profile(user_id, profile_updates)
issued_token = ""
if changing_credentials:
await db.users.update_credentials(user_id, new_username, new_password)
await db.tokens.revoke_all(user_id)
fresh_user = await db.users.get_by_id(user_id)
token_record = await _issue_token(db, fresh_user, req.expires_in)
issued_token = token_record["token"]
fresh_user = await db.users.get_by_id(user_id)
fresh_user["permissions"] = await db.roles.permissions_for(
fresh_user.get("role", "opc_member"), fresh_user.get("sub_role"),
)
fresh_user["scope_region_ids"] = await db.regions.visible_region_ids(fresh_user.get("region_id"))
return ProfileResponse(
token=issued_token,
**await db.users.to_profile(fresh_user),
)
@router.post("/revoke-token", summary="吊销单个令牌")
async def revoke_single_token(
req: RevokeTokenRequest,
request: Request,
user: dict = Depends(get_current_user),
db: Database = Depends(get_db),
):
"""吊销指定令牌(省略则吊销当前令牌)。"""
caller_token = extract_bearer_token(request)
token_to_revoke = req.token or caller_token
is_current = token_to_revoke == caller_token
if not await db.tokens.revoke(token_to_revoke):
raise HTTPException(status_code=500, detail="Failed to revoke token")
message = (
"Current token has been revoked. Please login again."
if is_current
else "Specified token has been revoked."
)
return {
"message": message,
"revoked": True,
"revoked_current_token": is_current,
}
@router.post("/revoke-all-tokens", summary="吊销全部令牌")
async def revoke_all_sessions(
user: dict = Depends(get_current_user),
db: Database = Depends(get_db),
):
"""吊销所有令牌,所有会话需重新登录。"""
await db.tokens.revoke_all()
return {
"message": "All tokens have been revoked. Please login again.",
"revoked": True,
}
# ===========================================================================
# 统一登录(手机验证码 / 微信)—— 六端登录归一
# ===========================================================================
async def _ensure_opc_identity(db: Database, user_id: str) -> dict:
"""确保账号拥有 opc_member 身份(登录即建;重复登录复用已有)。"""
for ident in await db.identities.list_for_user(user_id, active_only=True):
if ident["port"] == "opc" and ident["role"] == "opc_member":
return ident
return await db.identities.create(
user_id, port="opc", role="opc_member",
sub_role="independent", name="独立OPC",
)
async def _login_response(
db: Database, user: dict, identity: dict, identities: list[dict] | None = None,
) -> LoginResponse:
"""以指定身份签发令牌并装配登录响应(与密码登录同构)。"""
token_record = await _issue_token(db, user, None, identity=identity)
profile = await _profile_for(user, identity, db)
return LoginResponse(
token=token_record["token"],
identities=_identity_summaries(identities or [identity]),
**profile,
)
@router.post("/send-code", response_model=SendCodeResponse, summary="发送短信验证码")
async def send_code(req: SendCodeRequest, db: Database = Depends(get_db)):
"""为手机号发送登录验证码。stub provider 仅在服务日志打印。"""
if not _is_phone(req.phone):
raise HTTPException(status_code=400, detail="手机号需为 11 位(1 开头)")
if not config.AUTH_ENABLED:
raise HTTPException(status_code=403, detail="认证未开启")
code = sms.issue(req.phone)
return SendCodeResponse(
sent=True,
stub_code=code if config.SMS_PROVIDER == "stub" else None,
)
@router.post("/phone-login", response_model=LoginResponse, summary="手机号验证码登录")
async def phone_login(req: PhoneLoginRequest, request: Request, db: Database = Depends(get_db)):
"""手机号 + 验证码登录:无账号则注册(建 opc_member 身份)。"""
if not _is_phone(req.phone):
raise HTTPException(status_code=400, detail="手机号需为 11 位(1 开头)")
if not config.AUTH_ENABLED:
raise HTTPException(status_code=403, detail="认证未开启")
try:
sms.verify(req.phone, req.code)
except sms.SmsError as exc:
raise HTTPException(status_code=401, detail=str(exc))
user = await db.users.get_by_username(req.phone)
if user is None:
user = await db.users.create(
req.phone, password=secrets.token_hex(16),
phone=req.phone, role="opc_member",
source=_client_source(request, getattr(req, "source", ""), "phone"),
auth_type="phone", register_ip=_client_ip(request),
)
if user.get("status") != "active":
raise HTTPException(status_code=403, detail="账号已禁用")
identity = await _ensure_opc_identity(db, user["id"])
await db.audit.add(
action="login", resource="auth", resource_id=user["id"],
detail=f"phone-login {req.phone}", user_id=user["id"],
)
return await _login_response(db, user, identity)
@router.post("/wx-login", response_model=LoginResponse, summary="微信扫码登录(预留)")
async def wx_login(req: WxLoginRequest, request: Request, db: Database = Depends(get_db)):
"""微信扫码登录:本期预留未开放。开启需 PINEAGENTS_WECHAT_LOGIN=true 且配置 appid/secret。"""
if not config.AUTH_WECHAT_LOGIN:
raise HTTPException(status_code=503, detail="微信登录预留中,本期未开放")
if not config.AUTH_ENABLED:
raise HTTPException(status_code=403, detail="认证未开启")
try:
openid = await wechat.code2session(req.code)
except wechat.WechatError as exc:
raise HTTPException(status_code=401, detail=str(exc))
source = _client_source(request, getattr(req, "source", ""), "wx_openid")
is_mini = source == "mini_program"
# 小程序登录存 wx_mini_openid(与网页/公众号 openid 不同);否则存 wx_openid
if is_mini:
user = await db.users.get_by_wx_mini_openid(openid) or await db.users.get_by_wx_openid(openid)
else:
user = await db.users.get_by_wx_openid(openid)
if user is None:
user = await db.users.create(
f"wx_{openid[:24]}", password=secrets.token_hex(16),
wx_openid=openid if not is_mini else "",
wx_mini_openid=openid if is_mini else "",
nickname=str(getattr(req, "nickName", "") or ""),
avatar=str(getattr(req, "avatarUrl", "") or ""),
role="opc_member", source=source, auth_type="wx_openid",
register_ip=_client_ip(request),
)
if user.get("status") != "active":
raise HTTPException(status_code=403, detail="账号已禁用")
identity = await _ensure_opc_identity(db, user["id"])
await db.audit.add(
action="login", resource="auth", resource_id=user["id"],
detail=f"wx-login {openid[:16]}", user_id=user["id"],
)
return await _login_response(db, user, identity)
@router.post("/wx-phone", summary="微信补绑手机号")
async def wx_phone(req: WxPhoneRequest, db: Database = Depends(get_db)):
"""给已登录的微信账号补绑手机号(用于兑现政策/报名等需手机的流程)。"""
if not config.AUTH_ENABLED:
raise HTTPException(status_code=403, detail="认证未开启")
try:
sms.verify(req.phone, req.code)
except sms.SmsError as exc:
raise HTTPException(status_code=401, detail=str(exc))
user = await db.users.get_by_wx_openid(req.openid)
if user is None:
raise HTTPException(status_code=404, detail="微信账号不存在,请先 wx-login")
bound = await db.users.get_by_username(req.phone)
if bound is not None and bound["id"] != user["id"]:
raise HTTPException(status_code=409, detail="该手机号已绑定其它账号")
fresh = await db.users.set_phone(user["id"], req.phone)
await db.audit.add(
action="bind-phone", resource="auth", resource_id=user["id"],
detail=f"wx bind phone {req.phone}", user_id=user["id"],
)
return {**await db.users.to_profile(fresh), "phone": fresh.get("phone", "")}