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" / "server" / "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" / "server" / "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" / "server" / "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" / "server" / "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" / "server" / "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") def test_scripts_and_configs_directory_integrity(self): """[结构优化验证] 验证 scripts 分层物理结构与 configs/ 目录结构及 scripts 顶层彻底无包装器""" # 1. 验证 configs/ 目录及文件 configs_dir = PROJECT_ROOT / "configs" self.assertTrue(configs_dir.is_dir(), "configs/ 目录必须存在") self.assertTrue((configs_dir / "config.server.json").is_file()) self.assertTrue((configs_dir / "config.local.json").is_file()) self.assertTrue((configs_dir / "config_router.conf").is_file()) # 2. 验证根目录保持干净（三份配置彻底归拢至 configs/，根目录不再散落冗余配置或软链接） for conf_name in ["config.server.json", "config.local.json", "config_router.conf"]: root_conf = PROJECT_ROOT / conf_name self.assertFalse(root_conf.exists(), f"根目录 {conf_name} 应已收敛至 configs/，不应散落在根目录") # 3. 验证 scripts/server/ 和 scripts/router/ 结构 self.assertTrue((PROJECT_ROOT / "scripts" / "server" / "run_update_and_commit.sh").is_file()) self.assertTrue((PROJECT_ROOT / "scripts" / "server" / "install_debian.sh").is_file()) self.assertTrue((PROJECT_ROOT / "scripts" / "server" / "uninstall_debian.sh").is_file()) self.assertTrue((PROJECT_ROOT / "scripts" / "router" / "router_local_update.sh").is_file()) self.assertTrue((PROJECT_ROOT / "scripts" / "router" / "router_local_http.sh").is_file()) # 4. 验证 scripts/ 根目录彻底干净（直接引用新结构，不保留任何包装转发脚本） wrapper_names = [ "run_update_and_commit.sh", "install_debian.sh", "uninstall_debian.sh", "router_local_update.sh", "router_local_http.sh", ] for w_name in wrapper_names: p = PROJECT_ROOT / "scripts" / w_name self.assertFalse(p.exists(), f"scripts/ 根目录不得再保留门面包装脚本 {w_name}") if __name__ == "__main__": unittest.main()