"""在私有挂载命名空间中验证固定路径安装器，不操作正式服务。""" import argparse import hashlib import json import os from pathlib import Path import subprocess import sys import tempfile def run(*args, **kwargs): return subprocess.run(args, check=True, text=True, **kwargs) def digest(path): return hashlib.sha256(path.read_bytes()).hexdigest() def snapshot(root): result = {} for path in sorted(root.rglob("*")): info = path.lstat() relative = str(path.relative_to(root)) if path.is_symlink(): result[relative] = ["link", os.readlink(path), info.st_uid, info.st_gid] elif path.is_file(): result[relative] = [digest(path), info.st_mode & 0o777, info.st_uid, info.st_gid] else: result[relative] = ["directory", info.st_mode & 0o777, info.st_uid, info.st_gid] return result def child(args): base = Path(args.child).resolve(strict=True) marker = base / "namespace-marker" if marker.read_text() != "scert-installer-test\n": raise RuntimeError("隔离目录标识不正确") source = base / "source" run("mount", "--bind", str(Path(args.source).resolve(strict=True)), str(source)) run("mount", "-o", "remount,bind,ro", str(source)) for name, target in [("install", "/scert"), ("units", "/etc/systemd/system"), ("certificates", args.certificate_root)]: run("mount", "--bind", str(base / name), target) if os.stat(base / name).st_ino != os.stat(target).st_ino: raise RuntimeError("隔离挂载未生效") environment = dict(os.environ, PATH=str(base / "commands") + ":/usr/sbin:/usr/bin:/sbin:/bin") command = ["bash", str(source / "install-linux.sh"), "--source-dir", str(source)] if not (base / "install/conf/installation.json").exists(): command.extend(["--certificate-dir", args.certificate_root]) result = subprocess.run( command, env=environment, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, ) (base / "output.txt").write_text(result.stdout) (base / "result.json").write_text(json.dumps({"exit_code": result.returncode})) def prepare(base, scenario, certificate_root): for name in ["install/conf", "install/data", "install/log", "install/backups", "install/docs", "install/web/assets", "units/Scert.service.d", "certificates", "commands", "source"]: (base / name).mkdir(parents=True, exist_ok=True) (base / "namespace-marker").write_text("scert-installer-test\n") for name in ["scert", "scert-admin", "docs/Scert-service-management.md", "docs/Scert-admin-service-management.md", "web/index.html", "web/assets/old.js"]: (base / "install" / name).write_text("old:" + name) (base / "install/conf/installation.json").write_text(json.dumps({"certificate_root": certificate_root}) + "\n") for name in ["conf/scert.json", "conf/master.key", "conf/secrets.json", "data/scert.db", "backups/keep"]: (base / "install" / name).write_text("preserve:" + name) (base / "certificates/keep.pem").write_text("preserve:certificate") for name in ["Scert.service", "Scert-admin.service", "Scert.service.d/10-certificate-path.conf"]: (base / "units" / name).write_text("old:" + name) command_log = str(base / "systemctl.log") counter = str(base / "reload-count") systemctl = base / "commands/systemctl" systemctl.write_text( '#!/usr/bin/env bash\nset -eu\n' f'printf "%s\\n" "$*" >> "{command_log}"\n' 'if [[ "$1" == "is-active" ]]; then exit 3; fi\n' '[[ "$*" == "daemon-reload" ]] || exit 97\n' + (f'if [[ ! -e "{counter}" ]]; then touch "{counter}"; exit 42; fi\n' if scenario == "reload-failure" else "") + ('exit 42\n' if scenario == "rollback-reload-failure" else "") ) systemctl.chmod(0o755) if scenario == "web-link": (base / "install/web/assets/link").symlink_to(base / "certificates/keep.pem") if scenario == "binary-directory": (base / "install/scert").rename(base / "old-scert") (base / "install/scert").mkdir() if scenario in ["unit-failure", "fresh-failure"]: command = base / "commands/install" command.write_text( '#!/usr/bin/env bash\nset -eu\n' 'if [[ "${@: -1}" == "/etc/systemd/system/Scert-admin.service" ]]; then exit 43; fi\n' 'exec /usr/bin/install "$@"\n' ) command.chmod(0o755) if scenario in ["fresh-failure", "fresh-success"]: (base / "install").rename(base / "unused-install-fixture") (base / "install").mkdir() (base / "units").rename(base / "unused-units-fixture") (base / "units").mkdir() def parent(args): if sys.platform != "linux" or os.geteuid() != 0: raise RuntimeError("测试需要 Linux root 和私有挂载命名空间") for target in ["/scert", args.certificate_root, "/etc/systemd/system"]: if not Path(target).is_dir() or Path(target).is_symlink(): raise RuntimeError("测试挂载目标必须是既有普通目录") run("getent", "passwd", "Scert", stdout=subprocess.DEVNULL) run("unshare", "--mount", "--propagation", "private", "true") output = Path(args.output).resolve(strict=True) for target in [Path("/scert"), Path(args.certificate_root), Path("/etc/systemd/system")]: if output == target or target in output.parents: raise RuntimeError("测试产物不能位于被隔离的挂载点内") source = Path(args.source).resolve(strict=True) failures = [] for scenario in args.scenarios.split(","): if scenario not in ["web-link", "binary-directory", "unit-failure", "reload-failure", "rollback-reload-failure", "success", "fresh-failure", "fresh-success"]: raise RuntimeError("未知测试场景") base = Path(tempfile.mkdtemp(prefix=scenario + "-", dir=output)) prepare(base, scenario, args.certificate_root) protected = ["conf/scert.json", "conf/master.key", "conf/secrets.json", "data/scert.db", "backups/keep"] if scenario.startswith("fresh-"): protected = [] before_protected = {name: digest(base / "install" / name) for name in protected} before_install = snapshot(base / "install") before_units = snapshot(base / "units") before_certificates = snapshot(base / "certificates") run("unshare", "--mount", "--propagation", "private", sys.executable, str(Path(__file__).resolve()), "--source", str(source), "--child", str(base), "--certificate-root", args.certificate_root) code = json.loads((base / "result.json").read_text())["exit_code"] errors = [] if (code == 0) != (scenario in ["success", "fresh-success"]): errors.append("退出码不符合预期") if {name: digest(base / "install" / name) for name in protected} != before_protected: errors.append("业务文件被修改") if snapshot(base / "certificates") != before_certificates: errors.append("证书文件被修改") if scenario in ["success", "fresh-success"]: for name in ["scert", "scert-admin", "web/index.html"]: if digest(base / "install" / name) != digest(source / name): errors.append("安装文件内容不正确：" + name) if (base / "install/web/assets/old.js").exists(): errors.append("旧前端资源仍在生效目录") else: after = snapshot(base / "install") for name, expected in before_install.items(): # 安装器会统一业务目录权限，仅比较发布文件与安装记录。 if name in ["scert", "scert-admin", "conf/installation.json"] or name.startswith(("web/", "docs/")): if after.get(name) != expected: errors.append("失败后发布文件未保留：" + name) if scenario == "fresh-failure": for name in ["scert", "scert-admin", "web", "conf/installation.json"]: if (base / "install" / name).exists(): errors.append("首次安装失败仍残留生效文件：" + name) if any(path.is_file() for path in (base / "units").rglob("*")): errors.append("首次安装失败仍残留服务文件") elif snapshot(base / "units") != before_units: errors.append("失败后服务文件未恢复") for path in (base / "install").glob(".install-*"): if path.stat().st_mode & 0o777 != 0o700 or path.stat().st_uid != 0: errors.append("升级副本目录权限不正确") text = (base / "output.txt").read_text() if scenario == "rollback-reload-failure" and "安装回退不完整，禁止启动服务" not in text: errors.append("回退失败缺少明确提示") if scenario in ["unit-failure", "reload-failure", "fresh-failure"] and "已恢复升级前发布文件" not in text: errors.append("缺少回退完成记录") command_log = base / "systemctl.log" calls = command_log.read_text().splitlines() if command_log.exists() else [] allowed = {"is-active --quiet Scert.service", "is-active --quiet Scert-admin.service", "daemon-reload"} if any(call not in allowed for call in calls): errors.append("出现不允许的服务操作") report = {"scenario": scenario, "exit_code": code, "errors": errors, "artifacts": str(base)} (base / "report.json").write_text(json.dumps(report, ensure_ascii=False, indent=2)) print(json.dumps(report, ensure_ascii=False), flush=True) failures.extend(errors) if failures: raise SystemExit(1) parser = argparse.ArgumentParser() parser.add_argument("--source", required=True) parser.add_argument("--output") parser.add_argument("--child") parser.add_argument("--certificate-root", required=True) parser.add_argument("--scenarios", default="web-link,binary-directory,unit-failure,reload-failure,rollback-reload-failure,success,fresh-failure,fresh-success") arguments = parser.parse_args() if arguments.child: child(arguments) else: parent(arguments)