live-hub-py/tests/test_cookie_operations.py

"""CK 操作页的权限隔离、脱敏与重登入口。

2026-10-04 CK 解耦后本页面的**记录单位从「登录记录」变成「账号」**：
CK 与健康/重登进度都挂在 `accounts` 上，所有接口的 id 都是 account_id。
"""

from types import SimpleNamespace
from typing import cast
from unittest.mock import patch

import pytest

from tests.support.db import get_test_engine, make_test_session
from web.backend.models import Account, AuditLog, LoginTask, ProxyConfig, User
from web.backend.routers.cookies import (
    check_cookie_operations,
    delete_cookie,
    get_cookie,
    list_cookie_operation_tags,
    list_cookie_operations,
    list_cookies,
    relogin_invalid_cookie_operations,
)
from web.backend.services.cookie_check_service import check_douyu_cookie
from web.backend.services.login_service import (
    LoginBatchRunner,
    cleanup_orphan_login_tasks,
    cleanup_orphan_relogin_tasks,
)


class TestCookieOperation:
    def setup_method(self):
        self.engine = get_test_engine()
        self.session = make_test_session(self.engine)

        # 客服角色默认已经没有 CK 检测/重登。这里单独授权，只为保住
        # 「有这个权限时也只能动分配给自己的号、看不到 Cookie 原文」。
        self.support = User(
            username="support",
            password_hash="hash",
            role="support",
            custom_permissions=["cookie:operate"],
        )
        self.other_support = User(
            username="other", password_hash="hash", role="support"
        )
        self.session.add_all([self.support, self.other_support])
        self.session.commit()

        # CK 挂在账号上（2026-10-04 CK 解耦），所以「CK 记录」就是账号本身。
        self.owned_account = Account(
            username="owned-account",
            password="owned-password",
            email="owned@example.com",
            email_password="mail-password",
            assigned_to=self.support.id,
            cookie="owned-secret",
        )
        self.other_account = Account(
            username="other-account",
            password="other-password",
            email="other@example.com",
            email_password="mail-password",
            assigned_to=self.other_support.id,
            cookie="other-secret",
        )
        self.session.add_all([self.owned_account, self.other_account])
        self.session.commit()

    def teardown_method(self):
        self.session.close()

    def test_support_operation_list_is_scoped_and_never_returns_credentials(self):
        result = list_cookie_operations(
            search="",
            tag="",
            page=1,
            page_size=20,
            db=self.session,
            current=self.support,
        )

        assert result["total"] == 1
        assert result["items"] == [
            {
                "id": self.owned_account.id,
                "account_username": "owned-account",
                "tag": "",
                "ck_check_status": "",
                "ck_checked_at": None,
                "created_at": None,
                "relogin_status": "",
                "relogin_message": "",
                "relogin_batch_id": "",
            }
        ]
        assert "owned-secret" not in str(result)
        assert "owned-password" not in str(result)

    def test_default_support_role_cannot_open_cookie_operations(self):
        plain = User(username="plain-kf", password_hash="hash", role="support")
        self.session.add(plain)
        self.session.commit()
        with pytest.raises(Exception, match="cookie:operate"):
            list_cookie_operations(
                search="",
                tag="",
                page=1,
                page_size=20,
                db=self.session,
                current=plain,
            )

    def test_support_cannot_use_cookie_management_endpoints(self):
        with pytest.raises(Exception, match="cookie:view"):
            list_cookies(db=self.session, current=self.support)
        with pytest.raises(Exception, match="cookie:view"):
            get_cookie(self.owned_account.id, db=self.session, current=self.support)

    def test_cookie_list_filters_by_account_tag(self):
        admin = User(username="admin", password_hash="hash", role="super_admin")
        self.session.add(admin)
        self.owned_account.tag = "A组"
        self.other_account.tag = "B组"
        self.session.commit()

        result = list_cookies(
            search="",
            tag="A组",
            account_names="",
            page=1,
            page_size=20,
            include_cookie=False,
            db=self.session,
            current=admin,
        )

        assert isinstance(result, dict)
        assert result["total"] == 1
        assert [item["id"] for item in result["items"]] == [self.owned_account.id]

    def test_support_operation_tag_filter_and_list_stay_scoped(self):
        self.owned_account.tag = "我的标签"
        self.other_account.tag = "别人的标签"
        self.session.commit()

        tags = list_cookie_operation_tags(db=self.session, current=self.support)
        result = list_cookie_operations(
            search="",
            tag="别人的标签",
            page=1,
            page_size=20,
            db=self.session,
            current=self.support,
        )

        assert tags == ["我的标签"]
        assert result["total"] == 0

    @patch("web.backend.routers.cookies._check_one_cookie")
    def test_support_can_check_only_assigned_cookie(self, check_one_cookie):
        check_one_cookie.return_value = {
            "id": self.owned_account.id,
            "valid": True,
            "message": "有效",
            "fish_ball": 9,
            "nickname": "tester",
            "level": 1,
            "checked_at": "2026-01-01T00:00:00+00:00",
        }

        result = check_cookie_operations(
            ids=f"{self.owned_account.id},{self.other_account.id}",
            db=self.session,
            current=self.support,
        )

        assert [item["id"] for item in result["results"]] == [self.owned_account.id]
        self.session.refresh(self.owned_account)
        self.session.refresh(self.other_account)
        # 检测结论写在账号上，且只写了被授权的那一个
        assert self.owned_account.ck_check_status == "valid"
        assert self.other_account.ck_check_status == ""

    def test_operation_permission_does_not_allow_unrelated_users(self):
        no_permission_user = SimpleNamespace(
            id=999,
            username="no-permission",
            role="support",
            custom_permissions=[],
        )
        with pytest.raises(Exception, match="cookie:operate"):
            list_cookie_operations(
                search="",
                tag="",
                page=1,
                page_size=20,
                db=self.session,
                current=cast(User, no_permission_user),
            )

    def test_relogin_failed_account_stays_visible_and_can_be_checked(self):
        # 重登进度就写在账号上：判失效 + 有待重试时间 = 排队中
        self.owned_account.ck_check_status = "invalid"
        self.owned_account.relogin_last_error = "重新登录失败: 密码错误"
        self.session.commit()

        result = list_cookie_operations(
            search="",
            tag="",
            page=1,
            page_size=20,
            db=self.session,
            current=self.support,
        )

        assert result["total"] == 1
        assert result["items"][0]["relogin_status"] == "relogin_failed"
        assert "重新登录失败" in result["items"][0]["relogin_message"]
        # 重登失败不影响账号持有的 CK
        assert self.owned_account.cookie == "owned-secret"

    def test_relogin_pending_account_is_reported_as_pending(self):
        self.owned_account.ck_check_status = "invalid"
        from datetime import UTC, datetime, timedelta

        self.owned_account.relogin_next_at = datetime.now(UTC) + timedelta(hours=1)
        self.session.commit()

        result = list_cookie_operations(
            search="",
            tag="",
            page=1,
            page_size=20,
            db=self.session,
            current=self.support,
        )

        assert result["items"][0]["relogin_status"] == "relogin_pending"

    def test_service_restart_cleans_orphan_relogin_tasks(self):
        attempt = LoginTask(
            batch_id="orphan-relogin",
            account_id=self.owned_account.id,
            created_by=self.support.id,
            status="success",
            attempt_status="running",
            attempt_message="正在重新登录，旧 Cookie 保留中",
        )
        self.session.add(attempt)
        self.session.commit()

        cleaned = cleanup_orphan_relogin_tasks(self.session)

        assert cleaned == 1
        self.session.refresh(attempt)
        self.session.refresh(self.owned_account)
        # 只收敛尝试状态，账号的 CK 原封不动
        assert attempt.attempt_status == "failed"
        assert "服务重启" in attempt.attempt_message
        assert self.owned_account.cookie == "owned-secret"

    def test_service_restart_cleans_orphan_login_tasks(self):
        """普通登录批次遗留的在途尝试必须收敛，且不动账号的旧 CK。"""
        attempt = LoginTask(
            batch_id="orphan-login",
            account_id=self.owned_account.id,
            created_by=self.support.id,
            status="success",
            attempt_status="pending",
        )
        self.session.add(attempt)
        self.session.commit()

        cleaned = cleanup_orphan_login_tasks(self.session)

        assert cleaned == 1
        self.session.refresh(attempt)
        self.session.refresh(self.owned_account)
        assert attempt.attempt_status == "failed"
        assert "服务重启" in attempt.attempt_message
        assert self.owned_account.cookie == "owned-secret"

    def test_service_restart_cleans_legacy_in_flight_status(self):
        """升级前直接把状态写在 status 上的在途记录也要能收敛。"""
        attempt = LoginTask(
            batch_id="legacy-inflight",
            account_id=self.owned_account.id,
            created_by=self.support.id,
            status="relogin_running",
            attempt_status="",
        )
        self.session.add(attempt)
        self.session.commit()

        cleaned = cleanup_orphan_login_tasks(self.session)

        assert cleaned == 1
        self.session.refresh(attempt)
        assert attempt.attempt_status == "failed"
        assert self.owned_account.cookie == "owned-secret"

    def test_runner_copies_proxy_config_before_background_execution(self):
        proxy = ProxyConfig(
            enabled=True,
            http="http://127.0.0.1:8080",
            https="",
            api_url="",
            whitelist_enabled=False,
            whitelist_platform=None,
            whitelist_credentials=None,
            whitelist_uid="",
            whitelist_ukey="",
        )
        runner = LoginBatchRunner(
            db=self.session,
            account_ids=[],
            created_by=self.support.id,
            creator_permissions=[],
            proxy_config=proxy,
        )
        proxy.http = "http://changed.example:8080"

        proxy_dict, _ = runner._resolve_static_proxy()

        assert proxy_dict == {
            "http": "http://127.0.0.1:8080",
            "https": "http://127.0.0.1:8080",
        }

    @patch("web.backend.services.cookie_check_service.requests.get")
    def test_cookie_check_uses_new_cookie_result(self, mock_get):
        mock_get.side_effect = [
            SimpleNamespace(json=lambda: {"error": 0, "data": {"count": 9}}),
            SimpleNamespace(
                json=lambda: {"error": 0, "data": {"nn": "new-name", "lv": 12}}
            ),
        ]

        result = check_douyu_cookie("new-cookie")

        assert result["valid"]
        assert result["fish_ball"] == 9
        assert result["nickname"] == "new-name"
        assert result["level"] == 12

    @patch("web.backend.routers.cookies._start_relogin_accounts")
    def test_relogin_invalid_only_targets_support_visible_accounts(self, start_relogin):
        self.owned_account.ck_check_status = "invalid"
        self.other_account.ck_check_status = "invalid"
        self.session.commit()
        start_relogin.return_value = {
            "batch_id": "batch",
            "count": 1,
            "skipped": 0,
            "success": True,
        }

        result = relogin_invalid_cookie_operations(
            search="",
            tag="",
            db=self.session,
            current=self.support,
        )

        assert result["success"]
        selected_accounts = start_relogin.call_args.args[0]
        assert [account.id for account in selected_accounts] == [self.owned_account.id]

    def test_clearing_cookie_writes_audit_linked_to_account(self):
        """清 CK 是**不可逆**操作，审计必须能反查到账号。

        S1 给 `audit_logs` 加了 `account_id`（此前 `target` 是自由文本，账号改名
        或重名就关联不上）。这条用例锁住「清 CK 的审计真的写了 account_id」。
        """
        admin = User(username="ck-admin", password_hash="hash", role="super_admin")
        self.session.add(admin)
        self.session.commit()
        target_id = self.owned_account.id

        delete_cookie(target_id, db=self.session, current=admin)

        entry = (
            self.session.query(AuditLog)
            .filter(AuditLog.action == "cookie:delete")
            .order_by(AuditLog.id.desc())
            .first()
        )
        assert entry is not None
        assert entry.account_id == target_id
        assert entry.target == "owned-account"
        # 只清了入参那个账号，另一个不受影响
        self.session.refresh(self.owned_account)
        self.session.refresh(self.other_account)
        assert self.owned_account.cookie == ""
        assert self.other_account.cookie == "other-secret"