yml2213
4518b726ac
- 操作列改为「直播间 / 精英宝典 / 电竞宝典」三个直达按钮，点哪个就按哪个目标写入登录态并打开，去掉弹窗内选择
- 新增 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
248 lines
10 KiB
Python
248 lines
10 KiB
Python
"""虎牙「浏览器兑换」：目标页三选一、凭证签发审计与扩展注入边界。"""
|
||
|
||
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
|