diff --git a/test/generate_files.py b/test/generate_files.py deleted file mode 100644 index 58c7a7a..0000000 --- a/test/generate_files.py +++ /dev/null @@ -1,8 +0,0 @@ -import sys -import binascii -with open(sys.argv[1], 'wb') as fp: - for line in sys.stdin: - line = line.strip() - if line: - fp.write(binascii.unhexlify(line)) -print('done') diff --git a/test/run_deploy.py b/test/run_deploy.py deleted file mode 100644 index e07a86b..0000000 --- a/test/run_deploy.py +++ /dev/null @@ -1,19 +0,0 @@ -import paramiko -ssh = paramiko.SSHClient() -ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) -from tools.remote_ssh import connect_remote -connect_remote(ssh) -sftp = ssh.open_sftp() -print('Uploading out/smail to /opt/smail/smail.new ...') -sftp.put('out/smail', '/opt/smail/smail.new') -sftp.close() - -stdin, stdout, stderr = ssh.exec_command('chmod +x /opt/smail/smail.new && killall -9 smail; mv /opt/smail/smail.new /opt/smail/smail && cd /opt/smail && nohup ./smail -p 80 > log/runtime.log 2>&1 &') -print(stdout.read().decode()) -print(stderr.read().decode()) - -import time -time.sleep(2) -stdin, stdout, stderr = ssh.exec_command('ps aux | grep smail') -print(stdout.read().decode()) -ssh.close() diff --git a/test/runtime_data_check.py b/test/runtime_data_check.py deleted file mode 100644 index b23defb..0000000 --- a/test/runtime_data_check.py +++ /dev/null @@ -1,108 +0,0 @@ -"""只读核对运行目录迁移前后的文件及 SQLite 内容，不输出用户数据或密钥。""" - -import argparse -import hashlib -import json -import sqlite3 -from pathlib import Path - - -CONFIG_FILES = {"config.json", "smtp.json", "email.json", "ssl.json", "plugins.json"} -LEGACY_FILES = {"config.dev.json", "config_mysql.json", "config.json.legacy"} - - -def digest_file(path): - digest = hashlib.sha256() - with path.open("rb") as stream: - for chunk in iter(lambda: stream.read(1024 * 1024), b""): - digest.update(chunk) - return digest.hexdigest() - - -def database_snapshot(path): - # 必须停服调用。只读连接不会创建空库，不使用会忽略 WAL 的 immutable 模式。 - connection = sqlite3.connect(path.resolve().as_uri() + "?mode=ro", uri=True) - try: - connection.execute("PRAGMA query_only = ON") - connection.execute("BEGIN") - if connection.execute("PRAGMA integrity_check").fetchall() != [("ok",)]: - raise RuntimeError("数据库完整性检查失败") - tables = {} - for (name,) in connection.execute( - "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name" - ).fetchall(): - quoted = '"' + name.replace('"', '""') + '"' - rows = [] - for row in connection.execute(f"SELECT * FROM {quoted}"): - encoded = [ - {"blob": value.hex()} if isinstance(value, bytes) else value - for value in row - ] - payload = json.dumps(encoded, ensure_ascii=False, separators=(",", ":")) - rows.append(hashlib.sha256(payload.encode("utf-8")).hexdigest()) - tables[name] = { - "rows": len(rows), - "sha256": hashlib.sha256("".join(sorted(rows)).encode("ascii")).hexdigest(), - } - return {"integrity": "ok", "tables": tables} - finally: - connection.close() - - -def snapshot(root, layout): - config = root / "config" - files = {} - databases = {} - source = config if layout == "old" else root / "data" - for path in sorted(source.rglob("*")): - if path.is_symlink(): - raise RuntimeError("拒绝跟随运行目录符号链接") - if not path.is_file(): - continue - relative = path.relative_to(source).as_posix() - # 完整迁移前备份是新建的独立副本，已单独核对，不属于待迁移的原始集合。 - if layout == "new" and path.relative_to(source).parts[0] == "runtime-migration-backups": - continue - if layout == "old": - if relative in CONFIG_FILES: - key = "config/" + relative - elif relative in LEGACY_FILES: - key = "data/legacy-config/" + relative - else: - key = "data/" + relative - else: - key = "data/" + relative - files[key] = {"size": path.stat().st_size, "sha256": digest_file(path)} - if path.suffix in {".db", ".sqlite", ".sqlite3"}: - databases[key] = database_snapshot(path) - if layout == "new": - if {path.name for path in config.iterdir()} != CONFIG_FILES: - raise RuntimeError("config 目录必须恰好包含五个配置文件") - for name in sorted(CONFIG_FILES): - path = config / name - files["config/" + name] = {"size": path.stat().st_size, "sha256": digest_file(path)} - generation = json.loads((config / "config.json").read_text(encoding="utf-8"))["generation"] - return {"generation": generation, "files": files, "databases": databases} - - -def main(): - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--root", required=True, type=Path) - parser.add_argument("--layout", required=True, choices=("old", "new")) - parser.add_argument("--output", required=True, type=Path) - parser.add_argument("--baseline", type=Path) - args = parser.parse_args() - result = snapshot(args.root.resolve(strict=True), args.layout) - if args.baseline: - before = json.loads(args.baseline.read_text(encoding="utf-8")) - if before != result: - raise RuntimeError("迁移前后文件或数据库内容不一致，请保留现场") - # 不覆盖既有证据。 - with args.output.open("x", encoding="utf-8") as stream: - json.dump(result, stream, ensure_ascii=False, indent=2) - stream.write("\n") - print(f"核对通过：配置代次 {result['generation']}，文件 {len(result['files'])} 个，数据库 {len(result['databases'])} 个") - - -if __name__ == "__main__": - main() diff --git a/test/ssl-runtime-check.py b/test/ssl-runtime-check.py deleted file mode 100644 index d5414c6..0000000 --- a/test/ssl-runtime-check.py +++ /dev/null @@ -1,34 +0,0 @@ -"""核对本地邮件 TLS 监听实际使用的证书，仅用于开发环境验收。""" - -import argparse -import hashlib -from pathlib import Path -import socket -import ssl - - -def main(): - parser = argparse.ArgumentParser(description="核对邮件 TLS 端口的证书指纹") - parser.add_argument("certificate", type=Path) - parser.add_argument("--host", default="127.0.0.1") - parser.add_argument("--server-name", required=True) - parser.add_argument("--ports", type=int, nargs="+", default=[465, 993, 995]) - args = parser.parse_args() - pem = args.certificate.read_text(encoding="utf-8") - leaf = pem.split("-----END CERTIFICATE-----", 1)[0] + "-----END CERTIFICATE-----\n" - expected = hashlib.sha256(ssl.PEM_cert_to_DER_cert(leaf)).hexdigest() - context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) - # 本地自签名证书用精确指纹验证，不改变应用或系统的信任配置。 - context.check_hostname = False - context.verify_mode = ssl.CERT_NONE - for port in args.ports: - with socket.create_connection((args.host, port), timeout=5) as connection: - with context.wrap_socket(connection, server_hostname=args.server_name) as stream: - actual = hashlib.sha256(stream.getpeercert(binary_form=True)).hexdigest() - if actual != expected: - raise RuntimeError(f"端口 {port} 未加载预期证书") - print(f"端口 {port}：TLS 握手成功，证书指纹一致，协议 {stream.version()}") - - -if __name__ == "__main__": - main() diff --git a/test/write_helper.py b/test/write_helper.py deleted file mode 100644 index b72f1c7..0000000 --- a/test/write_helper.py +++ /dev/null @@ -1,2 +0,0 @@ -import sys, binascii -open(sys.argv[1], 'wb').write(binascii.unhexlify(sys.argv[2]))