Files
shaguabijia-app-server/app/core/pangle.py
T
OuYingJun1024 0722d7b0d5 feat: 福利钱包+签到+省钱战绩+激励广告发奖+微信提现 后端
- 福利钱包: 金币/现金余额、流水、兑换 (api/v1/wallet.py, crud/wallet.py, models/wallet.py, schemas/welfare.py)
- 每日签到: 连续天数 + 档位奖励 (api/v1/signin.py, crud/signin.py, models/signin.py)
- 任务系统: "打开消息提醒"等任务领奖 (api/v1/tasks.py, crud/task.py, models/task.py)
- 省钱战绩: 省钱汇总/战绩/店铺菜品 (api/v1/savings.py, crud/savings.py, models/savings.py)
- 激励广告发奖: 穿山甲服务端回调 + 发奖规则 + 限流 (api/v1/ad.py, core/pangle.py, core/rewards.py, core/ratelimit.py, crud/ad_reward.py, schemas/ad.py, docs/ad_reward_golive_checklist.md)
- 微信提现: 商家转账到零钱 V3 (core/wxpay.py); user 表加微信 openid/nickname/avatar
- DB 迁移: 8 个 alembic (welfare 表/cash_transaction/openid 唯一约束/savings 店铺菜品/savings_record/ad_reward/withdraw_order+openid/user 微信字段)
- 运维脚本: reconcile_withdraws(对账) + reset_signin/reset_welfare + sim_pangle_callback(模拟穿山甲回调)
- 测试: test_welfare / test_withdraw / test_ad_reward
- 配置: .env.example + config.py 新增福利/广告/微信支付项; main.py 挂 5 个新路由

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-27 17:38:08 +08:00

43 lines
1.9 KiB
Python

"""穿山甲激励视频"服务端发奖回调"的验签。
发奖闭环:激励视频播完 → 穿山甲服务器 GET 回调本服务(带 user_id / trans_id / sign 等)
→ 本服务**验签**确认回调确实来自穿山甲(而非伪造请求) → 幂等发金币。
客户端不参与发奖,被破解也刷不到钱。
⚠️ 关于验签方案:穿山甲后台的"奖励校验"实际签名算法以控制台为准(不同接入方式可能是
HMAC 共享密钥,也可能是穿山甲 RSA 公钥验签)。这里用 **HMAC-SHA256(排序参数, 密钥)** 作为
默认实现并配套自验签测试。真实接入时**只需改本文件的 `verify_callback_sign` / `build_sign`
这一处**——发奖、幂等、每日限额逻辑(crud/ad_reward.py)与验签方案完全无关,不受影响。
"""
from __future__ import annotations
import hashlib
import hmac
# 验签时排除的参数(sign 自身不参与签名;空值参数也不计入)
_SIGN_EXCLUDE = {"sign"}
def _canonical(params: dict[str, str]) -> str:
"""把参与签名的参数按 key 升序拼成 `k=v&k=v`。空值剔除,保证两端构造一致。"""
items = sorted(
(k, v) for k, v in params.items() if k not in _SIGN_EXCLUDE and v not in (None, "")
)
return "&".join(f"{k}={v}" for k, v in items)
def build_sign(params: dict[str, str], secret: str) -> str:
"""对参数计算 HMAC-SHA256 十六进制签名(测试 / 自验用)。"""
msg = _canonical(params).encode("utf-8")
return hmac.new(secret.encode("utf-8"), msg, hashlib.sha256).hexdigest()
def verify_callback_sign(params: dict[str, str], secret: str) -> bool:
"""校验回调签名。`params` 是回调全部 query 参数(含 sign)。密钥为空一律判失败。"""
if not secret:
return False
got = params.get("sign") or ""
expect = build_sign(params, secret)
# 定长比较防时序侧信道
return hmac.compare_digest(got, expect)