vmess-domain-rotator/tests/test_shell_lifecycle.py

import os
import pathlib
import shutil
import subprocess
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)


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