安全加固:feed 广告金币钳制 + prod 弱密钥 fail-fast

#1 feed-reward 防刷(eCPM/时长由客户端上报,原本金额无上界):
- rewards.calculate_ad_reward_coin 在唯一计算口把 eCPM 钳到 AD_ECPM_MAX_FEN
  (¥500 CPM),feed 与 reward_video 回退客户端 eCPM 的路径一并护住;阈值高于
  所有真实值,不影响正规发奖。
- ad_feed_reward 单事件时长上限由 24h(8640 份)收紧为 FEED_MAX_DURATION_SECONDS
  =120s(12 份),挡伪造超长时长刷份数。叠加每日 500 条上限锁住日产出。

#2 prod 启动期强校验密钥(Settings._enforce_prod_secrets):
- APP_ENV=prod 时 JWT_SECRET_KEY / ADMIN_JWT_SECRET 为默认值/空/长度<16 即 raise,
  挡住 token 被伪造导致账号与后台失陷。INTERNAL_API_SECRET 默认空=端点关闭的安全
  默认态,不强制。dev 不触发。

测试:tests/ 140 passed(3 个失败为既有的 compare/coupon 透传层问题,与本改动无关)。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

app/core/config.py

@@ -9,11 +9,16 @@ from functools import lru_cache
 from pathlib import Path
 from typing import Literal
 
-from pydantic import Field
+from pydantic import Field, model_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 _PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
 
+# 生产环境 JWT secret 的最小可接受长度(字节)。HS256 推荐高熵随机串;<16 视为弱密钥。
+_MIN_PROD_SECRET_LEN = 16
+# 已知的占位默认值(代码里写死的 default),prod 下绝不能沿用。
+_INSECURE_SECRET_DEFAULTS = frozenset({"change-me", "change-me-admin", ""})
+
 
 class Settings(BaseSettings):
     model_config = SettingsConfigDict(
@@ -198,6 +203,29 @@ class Settings(BaseSettings):
     def is_prod(self) -> bool:
         return self.APP_ENV == "prod"
 
+    @model_validator(mode="after")
+    def _enforce_prod_secrets(self) -> "Settings":
+        """prod 下强校验 JWT secret,弱/默认/空即启动报错(fail-fast,挡住 token 被伪造)。
+
+        只校验两个签发凭证:App 用户的 JWT_SECRET_KEY、后台的 ADMIN_JWT_SECRET——它们沿用默认值
+        时任何人都能伪造 access/admin token → 账号与后台失陷。INTERNAL_API_SECRET 默认空 = 内部端点
+        关闭(返 503),是安全的默认态,故不在此强制。dev 不触发,便于本地直接起。
+        """
+        if not self.is_prod:
+            return self
+        weak: list[str] = []
+        for name in ("JWT_SECRET_KEY", "ADMIN_JWT_SECRET"):
+            value = getattr(self, name)
+            if value in _INSECURE_SECRET_DEFAULTS or len(value) < _MIN_PROD_SECRET_LEN:
+                weak.append(name)
+        if weak:
+            raise ValueError(
+                f"APP_ENV=prod 但检测到弱/默认密钥: {', '.join(weak)} —— 必须改成 "
+                f"≥{_MIN_PROD_SECRET_LEN} 位高熵随机串(否则 JWT 可被伪造 → 用户/后台账号失陷)。"
+                f"生成示例: python -c \"import secrets; print(secrets.token_urlsafe(48))\""
+            )
+        return self
+
 
 @lru_cache(maxsize=1)
 def get_settings() -> Settings:

app/core/rewards.py

@@ -143,6 +143,13 @@ AD_LT_FACTOR_TABLE: tuple[tuple[float, int, int | None], ...] = (
     (1.0, 11, None),
 )
 
+# 客户端可影响的 eCPM 可信上限(分/千次展示):信息流广告一期由客户端上报 eCPM,伪造天价 eCPM
+# 可铸出天量金币(见 calculate_ad_reward_coin)。真实 eCPM 一般 <¥100 CPM(=10000 分),档位表顶档
+# 为 >¥400(=40000 分);取 ¥500 CPM=50000 分,留足真实头部余量又封死伪造值。钳在唯一计算口
+# calculate_ad_reward_coin,故 feed 与 reward_video(回退客户端上报 eCPM 时)一并护住;阈值设在所有
+# 真实值之上,不会少发正规奖励。
+AD_ECPM_MAX_FEN: int = 50_000
+
 
 def parse_ecpm_fen(ecpm: str | int | float | None) -> float:
     """解析 eCPM 原始值(穿山甲 getEcpm 原值,单位=分/千次展示)。非法/缺失→0。"""
@@ -187,8 +194,12 @@ def calculate_ad_reward_coin(ecpm: str | int | float | None, count_after_this: i
     eCPM 是穿山甲 getEcpm 原值,单位【分/千次展示】;先 ÷100 转成元(因子判档 + 收益换算都用元)。
     单次收益(元)= eCPM元 ÷ 1000(每千次→单次) × 因子1(eCPM 元档) × 因子2(LT);
     再按 1 元=10000 金币取整。count_after_this 为账号累计第 N 次看视频(LT 因子用,不按天重置)。
+
+    eCPM 在此先钳到 AD_ECPM_MAX_FEN(¥500 CPM):信息流广告一期 eCPM 由客户端上报,伪造天价值
+    会铸天量金币;钳在这唯一入口,feed 与 reward_video 回退客户端 eCPM 的路径都护住,且阈值高于
+    所有真实值,不影响正规发奖。
     """
-    ecpm_yuan = parse_ecpm_yuan(ecpm)
+    ecpm_yuan = min(parse_ecpm_yuan(ecpm), AD_ECPM_MAX_FEN / 100.0)
     yuan = (ecpm_yuan / 1000.0) * ad_ecpm_factor(ecpm_yuan) * ad_lt_factor(count_after_this)
     return max(0, round(yuan * COIN_PER_YUAN))
 

app/repositories/ad_feed_reward.py

@@ -16,6 +16,10 @@ from app.repositories import wallet as crud_wallet
 
 
 FEED_REWARD_UNIT_SECONDS = 10
+# 单个 feed 事件的时长上限(秒):一期 duration_seconds 由客户端上报,伪造超长时长会刷份数
+# (每 10 秒 1 份)。真实单条信息流视频远小于此;取 120s=12 份封顶,挡刷量、不影响正规单。
+# 与 rewards.AD_ECPM_MAX_FEN(eCPM 钳顶)合起来,把单事件可铸金币锁进有限区间。
+FEED_MAX_DURATION_SECONDS = 120
 
 
 def _find_by_event(db: Session, client_event_id: str) -> AdFeedRewardRecord | None:
@@ -66,13 +70,19 @@ def grant_feed_reward(
     adn: str | None = None,
     slot_id: str | None = None,
 ) -> AdFeedRewardRecord:
-    """完成一条信息流广告后结算奖励。client_event_id 幂等,同号重试不重复发。"""
+    """完成一条信息流广告后结算奖励。client_event_id 幂等,同号重试不重复发。
+
+    一期 eCPM/时长均由客户端上报,故服务端两道硬闸防刷:时长钳到 FEED_MAX_DURATION_SECONDS
+    限单事件份数,eCPM 在 rewards.calculate_ad_reward_coin 内钳到 AD_ECPM_MAX_FEN 限单份金额;
+    叠加每日 get_ad_daily_limit 条数上限,把单用户日产出锁进有限区间。
+    """
     existing = _find_by_event(db, client_event_id)
     if existing is not None:
         return existing
 
     today = cn_today().isoformat()
-    safe_duration = max(0, min(duration_seconds, 24 * 60 * 60))
+    # 客户端上报时长先钳到 FEED_MAX_DURATION_SECONDS,防伪造超长时长刷份数(见常量注释)。
+    safe_duration = max(0, min(duration_seconds, FEED_MAX_DURATION_SECONDS))
     unit_count = safe_duration // FEED_REWARD_UNIT_SECONDS
 
     if _granted_today(db, user_id, today) >= rewards.get_ad_daily_limit(db):