fix(admin): review 加固——超管防自锁/时区/并发发钱/审计

代码 review 后端运营后台后修复 7 项:
- admins: 降级/禁用 super_admin 前校验仍≥1 个 active 超管,防零超管死局;审计记 before/after
- queries: 时间筛选统一 tz-aware(列为 timestamptz),比较绝对时刻、不再依赖 DB 会话时区
- price_report: approve/reject 加行锁,防并发/连点双倍发奖
- users/wallet: get_or_create_account 增可选 lock 参(默认关,不动 C 端);调金币/现金读余额加行锁防双写错位
- ad_audit: 复算排序补 id 次级键,时间戳并列时顺序确定
- withdraw: health-check 限 finance/super,不暴露密钥路径给全员
- schemas: 调账/拒绝 reason 加 trim 校验,拒纯空白

测试: admin 套件 37 passed,全量 140 passed,无新增失败。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
OuYingJun1024
2026-06-14 12:07:50 +08:00
parent 47812f7fcc
commit 436b2a36da
9 changed files with 122 additions and 36 deletions
+3 -3
View File
@@ -50,7 +50,7 @@ def _reward_video_rows(
AdRewardRecord.reward_date == date, AdRewardRecord.reward_date == date,
AdRewardRecord.reward_scene == "reward_video", AdRewardRecord.reward_scene == "reward_video",
) )
.order_by(AdRewardRecord.user_id, AdRewardRecord.created_at) .order_by(AdRewardRecord.user_id, AdRewardRecord.created_at, AdRewardRecord.id)
) )
if user_id is not None: if user_id is not None:
stmt = stmt.where(AdRewardRecord.user_id == user_id) stmt = stmt.where(AdRewardRecord.user_id == user_id)
@@ -127,7 +127,7 @@ def _feed_rows(db: Session, *, date: str, user_id: int | None) -> list[dict]:
stmt = ( stmt = (
select(AdFeedRewardRecord) select(AdFeedRewardRecord)
.where(AdFeedRewardRecord.reward_date == date) .where(AdFeedRewardRecord.reward_date == date)
.order_by(AdFeedRewardRecord.user_id, AdFeedRewardRecord.created_at) .order_by(AdFeedRewardRecord.user_id, AdFeedRewardRecord.created_at, AdFeedRewardRecord.id)
) )
if user_id is not None: if user_id is not None:
stmt = stmt.where(AdFeedRewardRecord.user_id == user_id) stmt = stmt.where(AdFeedRewardRecord.user_id == user_id)
@@ -205,7 +205,7 @@ def ad_coin_audit(
rows.extend(_reward_video_rows(db, date=date, user_id=user_id)) rows.extend(_reward_video_rows(db, date=date, user_id=user_id))
if scene in (None, "feed"): if scene in (None, "feed"):
rows.extend(_feed_rows(db, date=date, user_id=user_id)) rows.extend(_feed_rows(db, date=date, user_id=user_id))
rows.sort(key=lambda r: r["created_at"], reverse=True) rows.sort(key=lambda r: (r["created_at"], r["record_id"]), reverse=True)
total = len(rows) total = len(rows)
mismatch_count = sum(1 for r in rows if not r["matched"]) mismatch_count = sum(1 for r in rows if not r["matched"])
+45 -14
View File
@@ -57,7 +57,7 @@ def list_users(
"""用户列表(admin 全量)。支持手机号前缀 / 渠道 / 状态 / 昵称模糊 / 注册·最近登录时间范围筛选, """用户列表(admin 全量)。支持手机号前缀 / 渠道 / 状态 / 昵称模糊 / 注册·最近登录时间范围筛选,
按 id·注册时间·最近登录排序。**offset 分页**(cursor=offset):任意列排序下游标语义统一, 按 id·注册时间·最近登录排序。**offset 分页**(cursor=offset):任意列排序下游标语义统一,
代价是翻页期间数据变动可能错位一条——admin 低频场景可接受(同 [list_all_withdraw_orders])。 代价是翻页期间数据变动可能错位一条——admin 低频场景可接受(同 [list_all_withdraw_orders])。
日期入参统一转 UTC naive 比较(User 时间均为 UTC naive,见 _as_utc_naive)。""" 日期入参统一转 tz-aware UTC 比较(列为 timestamptz,见 _as_utc)。"""
stmt = select(User) stmt = select(User)
if phone: if phone:
stmt = stmt.where(User.phone.like(f"{phone}%")) # 前缀匹配 stmt = stmt.where(User.phone.like(f"{phone}%")) # 前缀匹配
@@ -68,13 +68,13 @@ def list_users(
if nickname and nickname.strip(): if nickname and nickname.strip():
stmt = stmt.where(User.nickname.ilike(f"%{nickname.strip()}%")) stmt = stmt.where(User.nickname.ilike(f"%{nickname.strip()}%"))
if created_from is not None: if created_from is not None:
stmt = stmt.where(User.created_at >= _as_utc_naive(created_from)) stmt = stmt.where(User.created_at >= _as_utc(created_from))
if created_to is not None: if created_to is not None:
stmt = stmt.where(User.created_at <= _as_utc_naive(created_to)) stmt = stmt.where(User.created_at <= _as_utc(created_to))
if last_login_from is not None: if last_login_from is not None:
stmt = stmt.where(User.last_login_at >= _as_utc_naive(last_login_from)) stmt = stmt.where(User.last_login_at >= _as_utc(last_login_from))
if last_login_to is not None: if last_login_to is not None:
stmt = stmt.where(User.last_login_at <= _as_utc_naive(last_login_to)) stmt = stmt.where(User.last_login_at <= _as_utc(last_login_to))
sort_cols = { sort_cols = {
"id": User.id, "id": User.id,
@@ -190,16 +190,16 @@ def list_all_withdraw_orders(
date_col = WithdrawOrder.updated_at if date_field == "updated_at" else WithdrawOrder.created_at date_col = WithdrawOrder.updated_at if date_field == "updated_at" else WithdrawOrder.created_at
if date_from is not None: if date_from is not None:
stmt = stmt.where(date_col >= _as_utc_naive(date_from)) stmt = stmt.where(date_col >= _as_utc(date_from))
if date_to is not None: if date_to is not None:
stmt = stmt.where(date_col <= _as_utc_naive(date_to)) stmt = stmt.where(date_col <= _as_utc(date_to))
now = datetime.now(timezone.utc).replace(tzinfo=None) # tz-aware:列为 timestamptz,比较绝对时刻、与 DB 会话时区无关(同 _as_utc / stats.py)
now = datetime.now(timezone.utc)
today_start = ( today_start = (
datetime.now(ZoneInfo("Asia/Shanghai")) datetime.now(ZoneInfo("Asia/Shanghai"))
.replace(hour=0, minute=0, second=0, microsecond=0) .replace(hour=0, minute=0, second=0, microsecond=0)
.astimezone(timezone.utc) .astimezone(timezone.utc)
.replace(tzinfo=None)
) )
if quick_filter == "abnormal": if quick_filter == "abnormal":
stmt = stmt.where( stmt = stmt.where(
@@ -254,11 +254,16 @@ def list_all_withdraw_orders(
return items, next_cursor return items, next_cursor
def _as_utc_naive(value: datetime) -> datetime: def _as_utc(value: datetime) -> datetime:
"""前端传 ISO 时间;DB 当前按 UTC naive 比较最稳(SQLite/本地开发一致)。""" """前端传 ISO 时间 → 统一成 tz-aware UTC 再比较。
所有时间列均为 `DateTime(timezone=True)`(Postgres timestamptz);用 tz-aware 绑定参数
比较的是绝对时刻,与 DB 会话时区无关、恒正确。曾用 naive UTC,正确性依赖会话 TimeZone=UTC,
生产会话非 UTC 时筛选边界会整体偏移——故统一 tz-aware(与 stats.py / withdraw_summary 一致)。
无时区入参按 UTC 解释。"""
if value.tzinfo is None: if value.tzinfo is None:
return value return value.replace(tzinfo=timezone.utc)
return value.astimezone(timezone.utc).replace(tzinfo=None) return value.astimezone(timezone.utc)
def list_feedbacks( def list_feedbacks(
@@ -266,15 +271,41 @@ def list_feedbacks(
*, *,
status: str | None = None, status: str | None = None,
user_id: int | None = None, user_id: int | None = None,
content: str | None = None,
created_from: datetime | None = None,
created_to: datetime | None = None,
sort_by: str = "id",
sort_order: str = "desc",
limit: int = 20, limit: int = 20,
cursor: int | None = None, cursor: int | None = None,
) -> tuple[list[Feedback], int | None]: ) -> tuple[list[Feedback], int | None]:
"""反馈工单列表。支持 状态 / 用户ID / 内容模糊 / 提交时间范围 筛选,按 id·提交时间排序。
**offset 分页**(cursor=offset):任意列排序下游标语义统一(同 [list_users]),代价是翻页期间
数据变动可能错位一条——admin 低频场景可接受。created_at 为 timestamptz,日期入参统一转 tz-aware UTC 比较。"""
stmt = select(Feedback) stmt = select(Feedback)
if status: if status:
stmt = stmt.where(Feedback.status == status) stmt = stmt.where(Feedback.status == status)
if user_id is not None: if user_id is not None:
stmt = stmt.where(Feedback.user_id == user_id) stmt = stmt.where(Feedback.user_id == user_id)
return cursor_paginate(db, stmt, Feedback.id, limit=limit, cursor=cursor) if content and content.strip():
stmt = stmt.where(Feedback.content.ilike(f"%{content.strip()}%"))
if created_from is not None:
stmt = stmt.where(Feedback.created_at >= _as_utc(created_from))
if created_to is not None:
stmt = stmt.where(Feedback.created_at <= _as_utc(created_to))
sort_cols = {"id": Feedback.id, "created_at": Feedback.created_at}
sort_col = sort_cols.get(sort_by, Feedback.id)
order_fn = asc if sort_order == "asc" else desc
id_order = asc(Feedback.id) if sort_order == "asc" else desc(Feedback.id)
stmt = stmt.order_by(order_fn(sort_col), id_order)
offset = max(cursor or 0, 0)
rows = list(db.execute(stmt.offset(offset).limit(limit + 1)).scalars().all())
has_more = len(rows) > limit
items = rows[:limit]
next_cursor = offset + limit if has_more else None
return items, next_cursor
def get_withdraw_by_out_bill_no(db: Session, out_bill_no: str) -> WithdrawOrder | None: def get_withdraw_by_out_bill_no(db: Session, out_bill_no: str) -> WithdrawOrder | None:
+24 -5
View File
@@ -17,6 +17,12 @@ router = APIRouter(
) )
def _active_super_count(db: AdminDb) -> int:
return sum(
1 for a in admin_repo.list_admins(db) if a.role == "super_admin" and a.status == "active"
)
@router.get("", response_model=list[AdminOut], summary="管理员列表") @router.get("", response_model=list[AdminOut], summary="管理员列表")
def list_admins(db: AdminDb) -> list[AdminOut]: def list_admins(db: AdminDb) -> list[AdminOut]:
return [AdminOut.model_validate(a) for a in admin_repo.list_admins(db)] return [AdminOut.model_validate(a) for a in admin_repo.list_admins(db)]
@@ -48,16 +54,29 @@ def update_admin(
if admin_id == admin.id and body.status == "disabled": if admin_id == admin.id and body.status == "disabled":
raise HTTPException(status_code=400, detail="不能禁用自己") raise HTTPException(status_code=400, detail="不能禁用自己")
# 防自锁:降级 / 禁用某个 super_admin 前,确认操作后仍至少剩 1 个 active super_admin,
# 否则会进入「零可用超管」死局——本路由仅 super 可进,只能改库恢复。
demotes_super = (
target.role == "super_admin"
and target.status == "active"
and (
(body.role is not None and body.role != "super_admin")
or body.status == "disabled"
)
)
if demotes_super and _active_super_count(db) <= 1:
raise HTTPException(status_code=400, detail="不能降级/禁用最后一个超级管理员")
changes: dict = {} changes: dict = {}
if body.role is not None: if body.role is not None and body.role != target.role:
changes["role"] = {"before": target.role, "after": body.role}
target.role = body.role target.role = body.role
changes["role"] = body.role if body.status is not None and body.status != target.status:
if body.status is not None: changes["status"] = {"before": target.status, "after": body.status}
target.status = body.status target.status = body.status
changes["status"] = body.status
if body.password is not None: if body.password is not None:
target.password_hash = hash_password(body.password)
changes["password"] = "reset" changes["password"] = "reset"
target.password_hash = hash_password(body.password)
if not changes: if not changes:
raise HTTPException(status_code=400, detail="无任何变更字段") raise HTTPException(status_code=400, detail="无任何变更字段")
db.commit() db.commit()
+4 -2
View File
@@ -60,7 +60,9 @@ def approve_price_report(
admin: Annotated[AdminUser, Depends(require_role("operator"))], admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb, db: AdminDb,
) -> OkResponse: ) -> OkResponse:
rep = db.get(PriceReport, report_id) # 行锁(SELECT FOR UPDATE):并发/连点双请求会都读到 pending → 各发一次金币双倍发奖,
# 锁住该行串行化,第二个请求拿锁后看到 approved → 走 400。SQLite 下 FOR UPDATE 为 no-op。
rep = db.get(PriceReport, report_id, with_for_update=True)
if rep is None: if rep is None:
raise HTTPException(status_code=404, detail="上报记录不存在") raise HTTPException(status_code=404, detail="上报记录不存在")
if rep.status != "pending": if rep.status != "pending":
@@ -88,7 +90,7 @@ def reject_price_report(
admin: Annotated[AdminUser, Depends(require_role("operator"))], admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb, db: AdminDb,
) -> OkResponse: ) -> OkResponse:
rep = db.get(PriceReport, report_id) rep = db.get(PriceReport, report_id, with_for_update=True) # 行锁,同 approve(防并发重复审核)
if rep is None: if rep is None:
raise HTTPException(status_code=404, detail="上报记录不存在") raise HTTPException(status_code=404, detail="上报记录不存在")
if rep.status != "pending": if rep.status != "pending":
+10 -6
View File
@@ -126,7 +126,8 @@ def grant_user_coins(
if body.mode == "set": if body.mode == "set":
if body.amount < 0: if body.amount < 0:
raise HTTPException(status_code=400, detail="目标金币值不能为负") raise HTTPException(status_code=400, detail="目标金币值不能为负")
before = wallet_repo.get_or_create_account(db, user_id, commit=False).coin_balance # lock=True:锁账户行,防连点/并发各读同一 before 算同一 delta 双写,余额错位
before = wallet_repo.get_or_create_account(db, user_id, commit=False, lock=True).coin_balance
delta = body.amount - before delta = body.amount - before
if delta == 0: if delta == 0:
raise HTTPException(status_code=400, detail=f"当前金币已为 {body.amount},无需调整") raise HTTPException(status_code=400, detail=f"当前金币已为 {body.amount},无需调整")
@@ -134,9 +135,9 @@ def grant_user_coins(
if body.amount == 0: if body.amount == 0:
raise HTTPException(status_code=400, detail="amount 不能为 0") raise HTTPException(status_code=400, detail="amount 不能为 0")
delta = body.amount delta = body.amount
# 负数扣减时不允许扣成负余额(运营误操作保护) # 负数扣减时不允许扣成负余额(运营误操作保护);lock=True 防并发扣穿
if delta < 0: if delta < 0:
acc_now = wallet_repo.get_or_create_account(db, user_id, commit=False) acc_now = wallet_repo.get_or_create_account(db, user_id, commit=False, lock=True)
if acc_now.coin_balance + delta < 0: if acc_now.coin_balance + delta < 0:
raise HTTPException( raise HTTPException(
status_code=400, detail=f"扣减后金币为负(当前余额 {acc_now.coin_balance})" status_code=400, detail=f"扣减后金币为负(当前余额 {acc_now.coin_balance})"
@@ -174,7 +175,10 @@ def grant_user_cash(
if body.mode == "set": if body.mode == "set":
if body.amount_cents < 0: if body.amount_cents < 0:
raise HTTPException(status_code=400, detail="目标现金值不能为负") raise HTTPException(status_code=400, detail="目标现金值不能为负")
before = wallet_repo.get_or_create_account(db, user_id, commit=False).cash_balance_cents # lock=True:锁账户行,防连点/并发各读同一 before 算同一 delta 双写,余额错位
before = wallet_repo.get_or_create_account(
db, user_id, commit=False, lock=True
).cash_balance_cents
delta = body.amount_cents - before delta = body.amount_cents - before
if delta == 0: if delta == 0:
raise HTTPException( raise HTTPException(
@@ -184,9 +188,9 @@ def grant_user_cash(
if body.amount_cents == 0: if body.amount_cents == 0:
raise HTTPException(status_code=400, detail="amount_cents 不能为 0") raise HTTPException(status_code=400, detail="amount_cents 不能为 0")
delta = body.amount_cents delta = body.amount_cents
# 负数扣减时不允许扣成负余额(运营误操作保护) # 负数扣减时不允许扣成负余额(运营误操作保护);lock=True 防并发扣穿
if delta < 0: if delta < 0:
acc_now = wallet_repo.get_or_create_account(db, user_id, commit=False) acc_now = wallet_repo.get_or_create_account(db, user_id, commit=False, lock=True)
if acc_now.cash_balance_cents + delta < 0: if acc_now.cash_balance_cents + delta < 0:
raise HTTPException( raise HTTPException(
status_code=400, detail=f"扣减后现金为负(当前余额 {acc_now.cash_balance_cents} 分)" status_code=400, detail=f"扣减后现金为负(当前余额 {acc_now.cash_balance_cents} 分)"
+6 -1
View File
@@ -87,7 +87,12 @@ def withdraws_summary(db: AdminDb) -> WithdrawSummaryOut:
return WithdrawSummaryOut(**queries.withdraw_summary(db)) return WithdrawSummaryOut(**queries.withdraw_summary(db))
@router.get("/health-check", response_model=WxpayHealthCheckOut, summary="提现配置健康检查") @router.get(
"/health-check",
response_model=WxpayHealthCheckOut,
summary="提现配置健康检查",
dependencies=[Depends(require_role("finance"))], # 暴露密钥路径/配置,限财务+super
)
def withdraw_health_check() -> WxpayHealthCheckOut: def withdraw_health_check() -> WxpayHealthCheckOut:
private_path = wxpay._resolve_config_path(settings.WXPAY_MCH_PRIVATE_KEY_PATH) # noqa: SLF001 private_path = wxpay._resolve_config_path(settings.WXPAY_MCH_PRIVATE_KEY_PATH) # noqa: SLF001
public_path = wxpay._resolve_config_path(settings.WXPAY_PUBLIC_KEY_PATH) # noqa: SLF001 public_path = wxpay._resolve_config_path(settings.WXPAY_PUBLIC_KEY_PATH) # noqa: SLF001
+9 -1
View File
@@ -3,7 +3,7 @@ from __future__ import annotations
from datetime import datetime from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field from pydantic import BaseModel, ConfigDict, Field, field_validator
class PriceReportOut(BaseModel): class PriceReportOut(BaseModel):
@@ -36,6 +36,14 @@ class PriceReportOut(BaseModel):
class PriceReportRejectRequest(BaseModel): class PriceReportRejectRequest(BaseModel):
reason: str = Field(min_length=1, max_length=256, description="拒绝理由,用户端记录页会看到") reason: str = Field(min_length=1, max_length=256, description="拒绝理由,用户端记录页会看到")
@field_validator("reason")
@classmethod
def _reason_not_blank(cls, v: str) -> str:
# min_length=1 放过纯空白(" "),trim 后再校验非空,避免审计/用户端记录到空理由
if not v.strip():
raise ValueError("拒绝理由不能为空")
return v.strip()
class PriceReportSummary(BaseModel): class PriceReportSummary(BaseModel):
"""审核台顶部各状态计数。""" """审核台顶部各状态计数。"""
+12 -1
View File
@@ -4,7 +4,7 @@ from __future__ import annotations
from datetime import datetime from datetime import datetime
from typing import Literal from typing import Literal
from pydantic import BaseModel, ConfigDict, Field from pydantic import BaseModel, ConfigDict, Field, field_validator
class AdminUserListItem(BaseModel): class AdminUserListItem(BaseModel):
@@ -37,6 +37,13 @@ class AdminUserOverview(BaseModel):
feedback_total: int feedback_total: int
def _strip_reason(v: str) -> str:
# min_length=1 放过纯空白(" "),trim 后再校验非空,避免审计记到空原因
if not v.strip():
raise ValueError("操作原因不能为空")
return v.strip()
class GrantCoinsRequest(BaseModel): class GrantCoinsRequest(BaseModel):
mode: Literal["delta", "set"] = Field( mode: Literal["delta", "set"] = Field(
"delta", description="delta=增减(amount 为变动量) / set=设为(amount 为目标值,须≥0)" "delta", description="delta=增减(amount 为变动量) / set=设为(amount 为目标值,须≥0)"
@@ -47,6 +54,8 @@ class GrantCoinsRequest(BaseModel):
) )
reason: str = Field(..., min_length=1, max_length=128, description="操作原因(必填,入审计)") reason: str = Field(..., min_length=1, max_length=128, description="操作原因(必填,入审计)")
_v_reason = field_validator("reason")(_strip_reason)
class GrantCashRequest(BaseModel): class GrantCashRequest(BaseModel):
mode: Literal["delta", "set"] = Field( mode: Literal["delta", "set"] = Field(
@@ -58,6 +67,8 @@ class GrantCashRequest(BaseModel):
) )
reason: str = Field(..., min_length=1, max_length=128, description="操作原因(必填,入审计)") reason: str = Field(..., min_length=1, max_length=128, description="操作原因(必填,入审计)")
_v_reason = field_validator("reason")(_strip_reason)
class SetUserStatusRequest(BaseModel): class SetUserStatusRequest(BaseModel):
status: Literal["active", "disabled"] = Field( status: Literal["active", "disabled"] = Field(
+9 -3
View File
@@ -78,9 +78,15 @@ class WithdrawNotReviewable(Exception):
"""提现单当前状态不可审核(非 reviewing,可能已被处理过)。""" """提现单当前状态不可审核(非 reviewing,可能已被处理过)。"""
def get_or_create_account(db: Session, user_id: int, *, commit: bool = True) -> CoinAccount: def get_or_create_account(
"""取用户金币账户,不存在则建一个空账户。""" db: Session, user_id: int, *, commit: bool = True, lock: bool = False
acc = db.get(CoinAccount, user_id) ) -> CoinAccount:
"""取用户金币账户,不存在则建一个空账户。
lock=True 时对已存在的账户行加 SELECT FOR UPDATE(--写余额的调用方串行化,防并发
双写余额错位, admin set 模式连点);默认 False 不改 C 端发奖行为SQLite 下为 no-op
"""
acc = db.get(CoinAccount, user_id, with_for_update=True) if lock else db.get(CoinAccount, user_id)
if acc is None: if acc is None:
acc = CoinAccount( acc = CoinAccount(
user_id=user_id, user_id=user_id,