提交工作区中已有的辅助测试文件删除，不包含未跟踪构建产物或安全扫描结果。
This commit is contained in:
parent
9fa25d5ce5
commit
342d0ab0a5
@ -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')
|
||||
@ -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()
|
||||
@ -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()
|
||||
@ -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()
|
||||
@ -1,2 +0,0 @@
|
||||
import sys, binascii
|
||||
open(sys.argv[1], 'wb').write(binascii.unhexlify(sys.argv[2]))
|
||||
Loading…
Reference in New Issue
Block a user