xuans/Smail
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

收紧正式验收为单数据库
Some checks are pending
Security Checks / security (push) Waiting to run
Details

Browse Source
This commit is contained in:
smail 2026-09-22 08:04:27 +08:00
parent 49e1f1ad98
commit 9261c8743d
4 changed files with 84 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
test/tools/deploy_review.py
View File

@ -1,4 +1,4 @@
"""仅在 47 服务器备份并升级 Smail，保留双库和失败证据。"""
"""仅在 47 服务器备份并升级 Smail，保留单库和失败证据。"""
import argparse
import hashlib
@ -13,7 +13,7 @@ import time
ROOT = Path('/opt/smail')
STAGING = Path('/var/tmp/smail-review-20260917')
DATABASES = ('smail.db', 'smail-state.db')
DATABASE_NAME = 'smail.db'
def run(*args):
@ -25,6 +25,16 @@ def digest(path):
return hashlib.file_digest(stream, 'sha256').hexdigest()
def database_path(root):
data = root / 'data'
files = sorted(path for path in data.rglob('*.db')
if path.is_file() or path.is_symlink())
expected = data / DATABASE_NAME
assert files == [expected], '正式验收要求运行目录仅保留 data/smail.db'
assert expected.is_file() and not expected.is_symlink(), '活动数据库不能是符号链接'
return expected
def snapshot(source, target):
with sqlite3.connect(f'file:{source}?mode=ro', uri=True) as original:
with sqlite3.connect(target) as copied:
@ -65,6 +75,7 @@ def main():
assert header[:5] == b'\x7fELF\x02' and header[18:20] == b'\x3e\x00'
assert digest(path) == checksum
assert not (ROOT / name).is_symlink()
database = database_path(ROOT)
backup = Path('/var/backups') / time.strftime('smail-before-outbox-%Y%m%d-%H%M%S')
backup.mkdir(mode=0o700)
@ -76,10 +87,8 @@ def main():
subprocess.run(['tar', '-czf', str(archive), '-C', '/opt', 'smail'], check=True)
subprocess.run(['tar', '-dzf', str(archive), '-C', '/opt'], check=True)
archive.chmod(0o600)
for name in DATABASES:
source = ROOT / 'data' / name
assert source.resolve() == source and not source.is_symlink()
snapshot(source, backup / name)
assert database.resolve() == database and not database.is_symlink()
snapshot(database, backup / DATABASE_NAME)
for name in expected:
shutil.copy2(ROOT / name, backup / name)
installed = True
@ -89,13 +98,12 @@ def main():
assert digest(ROOT / name) == checksum
subprocess.run(['systemctl', 'start', 'smail.service'], check=True)
wait_ready()
for name in DATABASES:
with sqlite3.connect(f'file:{ROOT / "data" / name}?mode=ro', uri=True) as database:
assert database.execute('PRAGMA integrity_check').fetchone()[0] == 'ok'
with sqlite3.connect(f'file:{ROOT / "data/smail.db"}?mode=ro', uri=True) as database:
version = database.execute('SELECT MAX(version) FROM schema_migration').fetchone()[0]
database_path(ROOT)
with sqlite3.connect(f'file:{database}?mode=ro', uri=True) as connection:
assert connection.execute('PRAGMA integrity_check').fetchone()[0] == 'ok'
version = connection.execute('SELECT MAX(version) FROM schema_migration').fetchone()[0]
assert version == 23
states = dict(database.execute('SELECT state,COUNT(*) FROM outbound_delivery GROUP BY state'))
states = dict(connection.execute('SELECT state,COUNT(*) FROM outbound_delivery GROUP BY state'))
result = {'backup': str(backup), 'archive_sha256': digest(archive),
'artifacts': expected, 'migration': version, 'queue_states': states,
'service': run('systemctl', 'is-active', 'smail.service'),
@ -106,10 +114,10 @@ def main():
except BaseException:
if installed:
subprocess.run(['systemctl', 'stop', 'smail.service'], check=True)
for name in DATABASES:
snapshot(ROOT / 'data' / name, backup / ('failed-' + name))
snapshot(backup / name, ROOT / 'data' / name)
shutil.chown(ROOT / 'data' / name, user='smail', group='smail')
database = database_path(ROOT)
snapshot(database, backup / ('failed-' + DATABASE_NAME))
snapshot(backup / DATABASE_NAME, database)
shutil.chown(database, user='smail', group='smail')
for name in expected:
shutil.copy2(backup / name, ROOT / name)
subprocess.run(['systemctl', 'start', 'smail.service'], check=True)

27
test/tools/internal_smtp_check.py
View File

@ -15,12 +15,23 @@ from pathlib import Path
ROOT = Path('/opt/smail')
DATABASE_NAME = 'smail.db'
RUN = time.strftime('%Y%m%d-%H%M%S')
DOMAIN = 'solit.cn'
BODY = '内部25端口中文正文验收：邮件必须真实入库，并且只属于指定收件人。'
ATTACHMENT = '中文附件内容，内部收信验收。'.encode('utf-8')
def database_path(root):
data = root / 'data'
files = sorted(path for path in data.rglob('*.db')
if path.is_file() or path.is_symlink())
expected = data / DATABASE_NAME
assert files == [expected], '正式验收要求运行目录仅保留 data/smail.db'
assert expected.is_file() and not expected.is_symlink(), '活动数据库不能是符号链接'
return expected
def connect():
client = smtplib.SMTP('127.0.0.1', 25, timeout=30)
client.ehlo('internal-check.invalid')
@ -64,16 +75,16 @@ def deliver(label, recipients, multipart=False):
def main():
assert ROOT.is_dir() and Path('/etc/systemd/system/smail.service').is_file()
assert subprocess.check_output(['systemctl', 'is-active', 'smail.service'], text=True).strip() == 'active'
# 只读连接真实业务库，备份两份数据库，不清除前一轮故障证据。
# 只读连接真实业务库，备份唯一数据库，不清除前一轮故障证据。
backup = Path('/var/backups') / f'smail-internal-smtp-{RUN}'
backup.mkdir(mode=0o700)
for name in ('smail.db', 'smail-state.db'):
with sqlite3.connect(f'file:{ROOT / "data" / name}?mode=ro', uri=True) as source:
with sqlite3.connect(backup / name) as target:
source.backup(target)
assert target.execute('PRAGMA integrity_check').fetchone()[0] == 'ok'
(backup / name).chmod(0o600)
with sqlite3.connect(f'file:{ROOT / "data/smail.db"}?mode=ro', uri=True) as database:
database_path_value = database_path(ROOT)
with sqlite3.connect(f'file:{database_path_value}?mode=ro', uri=True) as source:
with sqlite3.connect(backup / DATABASE_NAME) as target:
source.backup(target)
assert target.execute('PRAGMA integrity_check').fetchone()[0] == 'ok'
(backup / DATABASE_NAME).chmod(0o600)
with sqlite3.connect(f'file:{database_path_value}?mode=ro', uri=True) as database:
database.row_factory = sqlite3.Row
policy = json.loads((ROOT / 'config/email.json').read_text(encoding='utf-8')).get('policy', {})
assert policy.get('spamFilterLevel', 0) == 0, 'This fixture requires the current unfiltered policy'

29
test/tools/restore_review.py
View File

@ -12,20 +12,33 @@ import tarfile
import time
DATABASE_NAME = 'smail.db'
def run(*args):
return subprocess.check_output(args, text=True).strip()
def database_path(root):
data = root / 'data'
files = sorted(path for path in data.rglob('*.db')
if path.is_file() or path.is_symlink())
expected = data / DATABASE_NAME
assert files == [expected], '正式验收要求运行目录仅保留 data/smail.db'
assert expected.is_file() and not expected.is_symlink(), '活动数据库不能是符号链接'
return expected
def database_counts(root):
counts = {}
for name in ('smail.db', 'smail-state.db'):
with sqlite3.connect(f'file:{root / "data" / name}?mode=ro', uri=True) as database:
assert database.execute('PRAGMA integrity_check').fetchone()[0] == 'ok'
tables = database.execute("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'").fetchall()
counts[name] = {
table: database.execute('SELECT COUNT(*) FROM "' + table.replace('"', '""') + '"').fetchone()[0]
for (table,) in tables
}
database_path_value = database_path(root)
with sqlite3.connect(f'file:{database_path_value}?mode=ro', uri=True) as database:
assert database.execute('PRAGMA integrity_check').fetchone()[0] == 'ok'
tables = database.execute("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'").fetchall()
counts[DATABASE_NAME] = {
table: database.execute('SELECT COUNT(*) FROM "' + table.replace('"', '""') + '"').fetchone()[0]
for (table,) in tables
}
return counts

28
test/tools/verify_review.py
View File

@ -10,10 +10,23 @@ import stat
import subprocess
DATABASE_NAME = 'smail.db'
def run(*args):
return subprocess.check_output(args, text=True).strip()
def database_path(root):
data = root / 'data'
files = sorted(path for path in data.rglob('*.db')
if path.is_file() or path.is_symlink())
expected = data / DATABASE_NAME
assert files == [expected], '正式验收要求运行目录仅保留 data/smail.db'
assert expected.is_file() and not expected.is_symlink(), '活动数据库不能是符号链接'
return expected
def main():
parser = argparse.ArgumentParser()
parser.add_argument('--results', nargs='+', default=[])
@ -58,17 +71,16 @@ def main():
summary[name + '_sha256'] = actual
summary['admin_version'] = run(str(root / 'smail-admin'), 'version')
summary['database_integrity'] = {}
for name in ('smail.db', 'smail-state.db'):
with sqlite3.connect(f'file:{root / "data" / name}?mode=ro', uri=True) as database:
integrity = database.execute('PRAGMA integrity_check').fetchall()
assert integrity == [('ok',)]
summary['database_integrity'][name] = 'ok'
with sqlite3.connect(f'file:{root / "data/smail.db"}?mode=ro', uri=True) as database:
database = database_path(root)
with sqlite3.connect(f'file:{database}?mode=ro', uri=True) as connection:
integrity = connection.execute('PRAGMA integrity_check').fetchall()
assert integrity == [('ok',)]
summary['database_integrity'][DATABASE_NAME] = 'ok'
summary['migration'] = database.execute('SELECT MAX(version) FROM schema_migration').fetchone()[0]
summary['queue_states'] = dict(database.execute('SELECT state,COUNT(*) FROM outbound_delivery GROUP BY state'))
summary['queue_states'] = dict(connection.execute('SELECT state,COUNT(*) FROM outbound_delivery GROUP BY state'))
control = Path('/run/smail/admin-control.sock').lstat()
assert stat.S_ISSOCK(control.st_mode) and stat.S_IMODE(control.st_mode) == 0o600
assert control.st_uid == (root / 'data/smail.db').stat().st_uid
assert control.st_uid == database.stat().st_uid
summary['control_socket_mode'] = oct(stat.S_IMODE(control.st_mode))
curl = ['curl', '--silent', '--show-error', '--fail', '--max-time', '10', '--noproxy', '*',
'--cacert', str(root / 'ssl/custom.crt'), '--resolve', 'mail.solit.cn:443:127.0.0.1']