vmess-domain-rotator/tests/test_shell_lifecycle.py

import json
import os
import pathlib
import shutil
import subprocess
import sys
import tempfile
import unittest

PROJECT_ROOT = pathlib.Path(__file__).resolve().parent.parent


class TestShellLifecycleP0P1Bugs(unittest.TestCase):
    """测试并锁定 Shell 部署与发布脚本中的高危安全及逻辑漏洞"""

    def setUp(self):
        self.tmp_dir = tempfile.TemporaryDirectory()
        self.work_dir = pathlib.Path(self.tmp_dir.name)

    def tearDown(self):
        self.tmp_dir.cleanup()

    def test_p0_uninstall_debian_empty_service_name_deletes_var_lib(self):
        """[P0/P2 修复验证] 验证 uninstall_debian.sh 传空服务名立即报错退出，绝不执行 rm -rf /var/lib/"""
        mock_bin = self.work_dir / "bin"
        mock_bin.mkdir()
        # Mock id 绕过 root 检测
        (mock_bin / "id").write_text("#!/bin/sh\necho 0\n")
        # Mock systemctl
        (mock_bin / "systemctl").write_text("#!/bin/sh\nexit 0\n")
        # Mock rm 记录所有删除操作
        rm_log = self.work_dir / "rm_calls.log"
        (mock_bin / "rm").write_text(f"#!/bin/sh\necho \"rm $*\" >> {rm_log}\n")
        for p in mock_bin.iterdir():
            p.chmod(0o755)

        script_path = PROJECT_ROOT / "scripts" / "uninstall_debian.sh"
        env = os.environ.copy()
        env["PATH"] = f"{mock_bin}:{env.get('PATH', '')}"

        # 传入非法空参数 --service-name ""
        proc = subprocess.run(
            ["bash", str(script_path), "--service-name", ""],
            env=env,
            capture_output=True,
            text=True,
            cwd=str(self.work_dir),
        )

        self.assertNotEqual(proc.returncode, 0, "空服务名必须报错退出")
        rm_content = rm_log.read_text() if rm_log.exists() else ""
        self.assertNotIn("rm -rf /var/lib/", rm_content, "修复验证确认：绝不执行 rm -rf /var/lib/")

    def test_p0_build_router_package_trap_prematurely_deletes_tmp_dir(self):
        """[P0/P2 修复验证] 验证 build_router_package.sh 延迟注册 trap，遇到前置报错绝不误删已有目录"""
        victim_dir = self.work_dir / "victim_data_dir"
        victim_dir.mkdir()
        canary_file = victim_dir / "data.txt"
        canary_file.write_text("critical data")

        script_path = PROJECT_ROOT / "build_router_package.sh"
        env = os.environ.copy()
        env["TMP_DIR"] = str(victim_dir)
        env["ROUTER_CONF"] = "/nonexistent/path/config_router.conf"

        # 执行打包脚本，因缺少配置文件而必然报错退出
        proc = subprocess.run(
            ["sh", str(script_path)],
            env=env,
            capture_output=True,
            text=True,
            cwd=str(self.work_dir),
        )
        self.assertNotEqual(proc.returncode, 0)
        # 验证修复成功：victim_dir 完好无损，没有被提前的 trap cleanup 误删
        self.assertTrue(victim_dir.exists(), "修复验证确认：build_router_package.sh 未误删 victim_dir")

    def test_p1_run_update_and_commit_passes_token_in_env_not_argv(self):
        """[P1 修复验证] 验证 run_update_and_commit.sh 采用 GIT_CONFIG_* 环境变量传递 Token，argv 零泄露"""
        script_path = PROJECT_ROOT / "scripts" / "run_update_and_commit.sh"
        script_content = script_path.read_text()
        # 确认彻底移除了命令行 -c "http.extraHeader=..."
        self.assertNotIn('git_auth_opts=(-c "http.extraHeader', script_content, "Token 严禁出现在命令行 argv 参数中")
        self.assertIn("GIT_CONFIG_COUNT=1", script_content, "使用 Git 原生配置环境变量安全传递")
        self.assertIn("GIT_CONFIG_KEY_0=http.extraHeader", script_content)
        self.assertIn("GIT_CONFIG_VALUE_0=Authorization: Basic", script_content)

    def test_p1_run_update_and_commit_ls_remote_distinguishes_error_code(self):
        """[P1 修复验证] 验证 run_update_and_commit.sh 明确区分 ls-remote 退出码 2 与 128"""
        script_path = PROJECT_ROOT / "scripts" / "run_update_and_commit.sh"
        script_content = script_path.read_text()
        self.assertIn("ls_remote_rc", script_content, "已捕获并校验 ls-remote 退出码")
        self.assertIn("$ls_remote_rc -eq 2", script_content, "显式匹配 2 作为分支不存在")
        self.assertIn("fatal: git ls-remote failed", script_content, "非 0/2 致命错误直接报错退出，决不误建孤儿分支")

    def test_p1_run_update_and_commit_fallback_silence(self):
        """[P1 修复验证] 验证 run_update_and_commit.sh 在回退状态且值未变时执行回退静默"""
        script_path = PROJECT_ROOT / "scripts" / "run_update_and_commit.sh"
        script_content = script_path.read_text()
        self.assertIn('is_fallback="0"', script_content)
        self.assertIn('"error_use_last_good"', script_content)
        self.assertIn('is_fallback="1"', script_content)
        self.assertIn('"$is_fallback" == "1"', script_content)

    def test_mode_a_git_history_rolling_prune(self):
        """[方案B 模式A验证] 验证 run_update_and_commit.sh 在达到阈值时自动修剪孤儿分支历史至 max_state_commits"""
        repo_dir = self.work_dir / "state_repo"
        repo_dir.mkdir()
        # 初始化 git repo
        subprocess.run(["git", "init", "-b", "main"], cwd=str(repo_dir), check=True, capture_output=True)
        subprocess.run(["git", "config", "user.name", "Test Runner"], cwd=str(repo_dir), check=True)
        subprocess.run(["git", "config", "user.email", "test@example.com"], cwd=str(repo_dir), check=True)

        # 模拟生成 5 个初始历史 commit
        target_file = repo_dir / "current.txt"
        for i in range(1, 6):
            target_file.write_text(f"domain{i}.example.com\n")
            subprocess.run(["git", "add", "."], cwd=str(repo_dir), check=True)
            subprocess.run(["git", "commit", "-m", f"update {i}"], cwd=str(repo_dir), check=True, capture_output=True)

        initial_count = int(subprocess.check_output(["git", "rev-list", "--count", "HEAD"], cwd=str(repo_dir)).decode().strip())
        self.assertEqual(initial_count, 5)

        # 准备真实 updater 所需的测试配置与数据源
        csv_file = self.work_dir / "result.csv"
        csv_file.write_text("IP 地址,已用时间,下载速度 (MB/s)\nnew-best.example.com,20 ms,12.50\n")

        test_config = {
            "source": {"type": "cfst_local"},
            "cfst_local": {
                "work_dir": str(self.work_dir),
                "binary": "echo",
                "result_file": str(csv_file),
                "skip_run": True,
                "run_args": [],
            },
            "output": {
                "runtime_dir": str(repo_dir),
                "selected_value_file": "current.txt",
                "selected_value_json": "current.json",
                "selected_value_json_key": "domain",
                "state_file": "state.json",
                "state_last_good_key": "last_good",
                "export_vars_file": "vars.json",
                "substore_value_key": "DOMAIN",
            },
        }
        config_path = self.work_dir / "test_config.json"
        config_path.write_text(json.dumps(test_config))

        script_path = PROJECT_ROOT / "scripts" / "run_update_and_commit.sh"
        env = os.environ.copy()
        env["APP_DIR"] = str(repo_dir)
        env["GIT_RUNTIME_BRANCH"] = "main"
        env["GIT_PUSH_ENABLED"] = "0"
        env["GIT_MAX_STATE_COMMITS"] = "3"  # 设置阈值为 3
        env["PYTHON_BIN"] = sys.executable

        # 必须带上 scripts 目录（含 domain_updater.py 与 rotator 子包）在 APP_DIR
        app_scripts_dir = repo_dir / "scripts"
        shutil.copytree(PROJECT_ROOT / "scripts", app_scripts_dir)

        # 运行 run_update_and_commit.sh
        proc = subprocess.run(
            [
                "bash",
                str(script_path),
                "--force-commit",
                str(config_path),
            ],
            env=env,
            capture_output=True,
            text=True,
            cwd=str(self.work_dir),
        )

        combined_output = f"{proc.stdout}\n{proc.stderr}"
        self.assertIn("trimming older history (retaining last 3)", combined_output)
        self.assertIn("history successfully pruned to 3 commits", combined_output)

        # 检查 commit 总数是否被精确修剪至 3
        pruned_count = int(subprocess.check_output(["git", "rev-list", "--count", "HEAD"], cwd=str(repo_dir)).decode().strip())
        self.assertEqual(pruned_count, 3, "历史 Commit 数量必须严格限制在 max_state_commits = 3")

        # 检查最新内容是否完整保存
        self.assertEqual(target_file.read_text().strip(), "new-best.example.com")


if __name__ == "__main__":
    unittest.main()