marco
b50495bebe
短信(SMS_MOCK 切 mock/real): - integrations/sms.py 重写: real 模式走极光短信 REST /v1/messages 自定义验证码(本服务 secrets 生成 6 位码 + 进程内存 + 本地校验一次性/防爆破), 鉴权复用极光一键登录 JG_APP_KEY/MASTER_SECRET (同一极光应用, 上线只需 SMS_MOCK=false); mock 仍"任意6位通过"不动其余测试 - 防刷四层: 单号冷却 + 单号每日上限 + 单IP rate_limit(/sms/send 10/min、/sms/login 20/min) + 单码失败次数作废; SmsError 带 status_code 映射 429/503/400 - config 增 SMS_SEND_ENDPOINT/SIGN_ID/TEMPLATE_ID/CODE_LENGTH/DAILY_LIMIT/MAX_VERIFY_ATTEMPTS; test_auth 加 real 模式单测; sms.md/后端技术实现/待办账本同步 admin 后台(app/admin/ 独立子应用, uvicorn app.admin.main:admin_app :8771): - 复用主仓 models/repositories/integrations + 同库, 鉴权完全隔离(ADMIN_JWT_SECRET≠JWT_SECRET_KEY + payload typ=admin + bcrypt 密码 + 可选 IP 白名单); 主 app 不 import 本包, admin 崩不影响主进程 - 路由: 登录 / 账号管理(RBAC: super_admin·finance·operator) / 用户列表+360详情+封禁+手动调币 / 钱包流水 / 提现重试对账 / 反馈工单 / 数据大盘; 全写操作落 admin_audit_log(涉钱与业务写同事务) - 涉钱逻辑(调微信/退款/对账)复用 app.repositories.wallet 不重写 - 新增 models/admin.py(AdminUser/AdminAuditLog) + admin_tables 迁移 + create_admin.py + deploy/shaguabijia-admin.service; 依赖加 bcrypt Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
56 lines
1.8 KiB
Python
56 lines
1.8 KiB
Python
"""Admin JWT:与 App 用户 token 完全隔离。
|
|
|
|
用独立 secret(settings.ADMIN_JWT_SECRET ≠ JWT_SECRET_KEY)+ payload typ="admin",
|
|
双重保证 App 用户的 access_token 无法当 admin token 用(secret 不同直接验签失败)。
|
|
admin 无 refresh:过期(默认 12h)重新登录,简单。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any
|
|
|
|
import jwt
|
|
|
|
from app.core.config import settings
|
|
|
|
_ALGO = "HS256"
|
|
|
|
|
|
class AdminTokenError(Exception):
|
|
"""admin token 解析/校验失败,api 层捕获后返回 401。"""
|
|
|
|
|
|
def _now() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
def create_admin_token(*, admin_id: int, role: str) -> tuple[str, int]:
|
|
"""签发 admin access token,返回 (token, expires_in_seconds)。"""
|
|
now = _now()
|
|
expire = now + timedelta(minutes=settings.ADMIN_JWT_EXPIRE_MINUTES)
|
|
payload: dict[str, Any] = {
|
|
"sub": str(admin_id),
|
|
"role": role,
|
|
"typ": "admin",
|
|
"iat": int(now.timestamp()),
|
|
"exp": int(expire.timestamp()),
|
|
}
|
|
token = jwt.encode(payload, settings.ADMIN_JWT_SECRET, algorithm=_ALGO)
|
|
return token, int((expire - now).total_seconds())
|
|
|
|
|
|
def decode_admin_token(token: str) -> dict[str, Any]:
|
|
"""解析校验 admin token。签名错/过期/typ 非 admin → AdminTokenError。"""
|
|
try:
|
|
payload = jwt.decode(token, settings.ADMIN_JWT_SECRET, algorithms=[_ALGO])
|
|
except jwt.ExpiredSignatureError as e:
|
|
raise AdminTokenError("token expired") from e
|
|
except jwt.InvalidTokenError as e:
|
|
raise AdminTokenError(f"invalid token: {e}") from e
|
|
|
|
if payload.get("typ") != "admin":
|
|
raise AdminTokenError("not an admin token")
|
|
if "sub" not in payload:
|
|
raise AdminTokenError("token missing sub")
|
|
return payload
|