feat(huya): 浏览器兑换三目标直达并收窄扩展注入范围

- 操作列改为「直播间 / 精英宝典 / 电竞宝典」三个直达按钮，点哪个就按哪个目标写入登录态并打开，去掉弹窗内选择 - 新增 services/huya_exchange_targets.py 统一三个目标地址，接口支持可选 body {"target": "..."}（不传=精英宝典） - 扩展内容脚本从 <all_urls> 收窄到后台站点白名单，service worker 校验 sender.tab.url，拒绝其它来源的注入 - 注入改为 Cookie 写完后回执（ok/written/failed/error），前端按回执提示成功或失败，不再静默 - 扩展图标兜底改用 world:'MAIN' 读取；一个 Cookie 都没写成功时不打开目标页 - 兑换材料签发写审计 huya:exchange:kit（含账号与目标） - 控制台脚本增加 huya.com 域校验与 Secure，与扩展路径保持一致 - 修复兑换记录轮询的跨账号串扰与卸载后 setState；兑换请求加 epoch 防止过期结果覆盖弹窗 - 新增 tests/test_huya_browser_exchange.py（13 项）并同步 docs/HUYA_BROWSER_EXCHANGE.md
2026-09-11 10:31:46 +08:00
parent 874a95dd6f
commit 4518b726ac
14 changed files with 838 additions and 102 deletions
@@ -0,0 +1,247 @@
+"""虎牙「浏览器兑换」：目标页三选一、凭证签发审计与扩展注入边界。"""
+
+import json
+import re
+from pathlib import Path
+
+import pytest
+from fastapi import HTTPException
+from pydantic import ValidationError
+from sqlalchemy import create_engine, event
+from sqlalchemy.orm import sessionmaker
+
+from web.backend.database import Base
+from web.backend.huya_defaults import HUYA_DEFAULT_ROOM_PID
+from web.backend.models import AuditLog, HuyaAccount, HuyaConfig, User
+from web.backend.routers.huya import create_browser_exchange_kit
+from web.backend.schemas import HuyaBrowserExchangeRequest
+from web.backend.services.huya_exchange_targets import (
+    EXCHANGE_TARGET_KEYS,
+    exchange_target_url,
+)
+from web.backend.services.huya_runner_bind import HUYA_BIND_ROOM_ID
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+EXTENSION_DIR = REPO_ROOT / "web" / "extension" / "huya-exchange-helper"
+
+
+class TestHuyaBrowserExchangeKit:
+    def setup_method(self):
+        self.engine = create_engine("sqlite://")
+
+        @event.listens_for(self.engine, "connect")
+        def enable_foreign_keys(connection, _):
+            connection.execute("PRAGMA foreign_keys=ON")
+
+        Base.metadata.create_all(self.engine)
+        self.session = sessionmaker(bind=self.engine)()
+        self.user = User(username="operator", password_hash="hash", role="super_admin")
+        self.account = HuyaAccount(
+            uid="100", yyuid="100", cookie="yyuid=100; udb_uid=100"
+        )
+        self.session.add_all([self.user, self.account])
+        self.session.commit()
+
+    def teardown_method(self):
+        self.session.close()
+        Base.metadata.drop_all(self.engine)
+        self.engine.dispose()
+
+    def _kit(self, **kwargs):
+        return create_browser_exchange_kit(
+            self.account.id, db=self.session, current=self.user, **kwargs
+        )
+
+    def test_default_target_is_elite_and_three_options_are_returned(self):
+        kit = self._kit()
+        assert kit["target"] == "elite"
+        assert [option["key"] for option in kit["targets"]] == [
+            "live_room",
+            "elite",
+            "esports",
+        ]
+        assert [option["label"] for option in kit["targets"]] == [
+            "直播间",
+            "精英宝典",
+            "电竞宝典",
+        ]
+        # 默认目标地址=精英宝典（沿用历史行为：绑定/充值活动的 zt 活动页）
+        assert kit["target_url"].startswith(
+            "https://zt.huya.com/b02faae1/pc/index.html?"
+        )
+        assert f"pid={HUYA_DEFAULT_ROOM_PID}" in kit["target_url"]
+        assert kit["cookie_string"] == "yyuid=100; udb_uid=100"
+        assert kit["cookie_script"].endswith("})();")
+
+    def test_each_target_resolves_to_its_own_page(self):
+        live_room = exchange_target_url("live_room", HUYA_DEFAULT_ROOM_PID)
+        esports = exchange_target_url("esports", HUYA_DEFAULT_ROOM_PID)
+
+        assert live_room == f"https://www.huya.com/{HUYA_BIND_ROOM_ID}"
+        assert esports.startswith("https://zt.huya.com/50ea007f/pc/index.html?")
+        assert f"roomid={HUYA_BIND_ROOM_ID}" in esports
+        # 三个目标必须互不相同，否则选择器没有意义
+        urls = {
+            exchange_target_url(key, HUYA_DEFAULT_ROOM_PID)
+            for key in EXCHANGE_TARGET_KEYS
+        }
+        assert len(urls) == len(EXCHANGE_TARGET_KEYS)
+
+        for key in EXCHANGE_TARGET_KEYS:
+            kit = self._kit(req=HuyaBrowserExchangeRequest(target=key))
+            assert kit["target"] == key
+            expected = exchange_target_url(key, HUYA_DEFAULT_ROOM_PID)
+            assert kit["target_url"] == expected
+            # 控制台脚本里的跳转地址必须与所选目标一致
+            assert json.dumps(expected, ensure_ascii=False) in kit["cookie_script"]
+            assert [option["url"] for option in kit["targets"]].count(expected) == 1
+
+    def test_schema_rejects_unknown_target(self):
+        with pytest.raises(ValidationError):
+            HuyaBrowserExchangeRequest(target="unknown")
+        for key in EXCHANGE_TARGET_KEYS:
+            assert HuyaBrowserExchangeRequest(target=key).target == key
+            # 三个 key 都在 schemas 的正则白名单里（两处常量不许漂移）
+        assert HuyaBrowserExchangeRequest().target is None
+
+    def test_kit_issuance_is_audited_with_target(self):
+        self._kit(req=HuyaBrowserExchangeRequest(target="esports"))
+        rows = self.session.query(AuditLog).all()
+        assert len(rows) == 1
+        entry = rows[0]
+        assert entry.action == "huya:exchange:kit"
+        assert entry.target == f"huya_account:{self.account.id}"
+        assert entry.user_id == self.user.id
+        assert json.loads(entry.detail) == {
+            "account_id": self.account.id,
+            "target": "esports",
+        }
+
+    def test_account_without_cookie_is_rejected(self):
+        self.account.cookie = ""
+        self.session.commit()
+        with pytest.raises(HTTPException) as excinfo:
+            self._kit()
+        assert excinfo.value.status_code == 400
+
+    def test_support_user_only_reaches_assigned_accounts(self):
+        support = User(username="cs", password_hash="hash", role="support")
+        foreign = HuyaAccount(
+            uid="200", yyuid="200", cookie="yyuid=200", assigned_to=self.user.id
+        )
+        assigned = HuyaAccount(
+            uid="300", yyuid="300", cookie="yyuid=300", assigned_to=None
+        )
+        self.session.add_all([support, foreign, assigned])
+        self.session.commit()
+        assigned.assigned_to = support.id
+        self.session.commit()
+
+        # 客服只能兑换分配给自己的账号，他人账号按"不存在"处理
+        with pytest.raises(HTTPException) as excinfo:
+            create_browser_exchange_kit(foreign.id, db=self.session, current=support)
+        assert excinfo.value.status_code == 404
+
+        kit = create_browser_exchange_kit(assigned.id, db=self.session, current=support)
+        assert kit["cookie_string"] == "yyuid=300"
+
+    def test_user_without_task_permission_is_rejected(self):
+        guest = User(username="guest", password_hash="hash", role="guest")
+        self.session.add(guest)
+        self.session.commit()
+        with pytest.raises(HTTPException) as excinfo:
+            create_browser_exchange_kit(self.account.id, db=self.session, current=guest)
+        assert excinfo.value.status_code == 403
+        # 被拒时既不签发凭证，也不留"已签发"审计
+        assert self.session.query(AuditLog).count() == 0
+
+    def test_missing_account_returns_404(self):
+        with pytest.raises(HTTPException) as excinfo:
+            create_browser_exchange_kit(999999, db=self.session, current=self.user)
+        assert excinfo.value.status_code == 404
+
+    def test_kit_issuance_commits_audit_row(self):
+        self._kit()
+        # 未提交的审计行会停在 session.new：这条断言钉住 record_audit 之后确实 commit
+        assert not self.session.new, "审计日志未提交（缺少 db.commit()）"
+        assert self.session.query(AuditLog).count() == 1
+
+    def test_room_pid_config_changes_zt_urls_but_not_live_room(self):
+        self.session.add(HuyaConfig(room_pid="9527"))
+        self.session.commit()
+        assert (
+            "pid=9527"
+            in self._kit(req=HuyaBrowserExchangeRequest(target="esports"))["target_url"]
+        )
+        assert (
+            self._kit(req=HuyaBrowserExchangeRequest(target="live_room"))["target_url"]
+            == f"https://www.huya.com/{HUYA_BIND_ROOM_ID}"
+        )
+
+
+def _read(path: Path) -> str:
+    return path.read_text(encoding="utf-8")
+
+
+def test_extension_content_script_whitelist_is_narrow_and_parseable():
+    """内容脚本白名单必须收窄到后台站点，且每个模式都能被 background.js 解析成来源。"""
+    manifest = json.loads(_read(EXTENSION_DIR / "manifest.json"))
+    matches = [
+        pattern for entry in manifest["content_scripts"] for pattern in entry["matches"]
+    ]
+    background = _read(EXTENSION_DIR / "background.js")
+
+    assert matches, "内容脚本白名单不能为空"
+    assert "<all_urls>" not in matches
+    for pattern in matches:
+        # background.js 用 pattern.replace(/\/\*$/, '/') + new URL(...).origin 推导允许来源，
+        # 所以每个模式都必须是「具体源 + /*」，主机里不能有通配（通配会被 new URL 抛错并静默丢弃）
+        assert pattern.endswith("/*"), pattern
+        origin = pattern[: -len("/*")]
+        assert re.fullmatch(r"https?://[a-z0-9.-]+(:\d+)?", origin), pattern
+        # 内容脚本只需在后台页面生效：注入走 chrome.cookies，虎牙侧不需要内容脚本
+        assert "huya.com" not in origin
+    assert "originAllowed" in background
+
+
+def test_inject_ack_contract_is_paired_across_three_layers():
+    """注入回执链路（页面 ↔ 内容脚本 ↔ service worker）的字段必须成对，改名即失败。"""
+    page = _read(REPO_ROOT / "web" / "frontend" / "src" / "utils" / "huyaExchange.ts")
+    relay = _read(EXTENSION_DIR / "relay.js")
+    background = _read(EXTENSION_DIR / "background.js")
+
+    # 页面 → 内容脚本：事件名与参数名
+    assert "huya-exchange-kit" in page
+    assert "huya-exchange-kit" in relay
+    assert "request_id: requestId" in page
+    assert "kit.request_id" in relay
+    # 内容脚本 → 页面：回执事件名与字段
+    assert "huya-exchange-result" in page
+    assert "huya-exchange-result" in relay
+    assert "request_id: requestId" in relay
+    assert "parsed.request_id !== requestId" in page
+    assert "Boolean(result.ok)" in relay
+    assert "parsed.ok === true" in page
+    assert "result.written" in relay
+    assert "parsed.written" in page
+    # service worker → 内容脚本：延迟回执（等真正写完）与真实统计
+    assert "return true;" in background
+    assert "written > 0 && failed === 0" in background
+    assert "failedNames.push" in background
+    # 来源校验是本次安全修复的核心，不能被删掉
+    assert "originAllowed" in background
+    assert "sender.tab.url" in background
+    # 兜底路径同样要来源校验，且失败不再静默
+    assert "world: 'MAIN'" in background
+    assert "badge(tab.id, '失败')" in background
+    assert "badge(tab.id, '非后台页')" in background
+    # 一个都没写成功时不得打开目标页
+    assert "written > 0 && kit && kit.target_url" in background
+    # 只写 huya 域 Cookie
+    assert ".huya.com" in background
+
+
+def test_console_script_guards_domain_and_expiry():
+    script = _read(REPO_ROOT / "core" / "huya" / "cookie_utils.py")
+    assert "location.hostname" in script
+    assert "ttl_days: int = 180" in script