增加数据库迁移实机验收并记录重启响应中断反例
Some checks are pending
Security Checks / security (push) Waiting to run

This commit is contained in:
smail 2026-09-21 05:07:08 +08:00
parent d55f94e610
commit f589f10f05
3 changed files with 426 additions and 1 deletions

View File

@ -0,0 +1,111 @@
"""迁移验收工具的只读、范围和进程证据边界测试,不访问云服务器。"""
from contextlib import closing
import json
from pathlib import Path
import sqlite3
import sys
import tempfile
import unittest
from unittest.mock import Mock, patch
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'tools'))
import runtime_migration as migration
class RuntimeMigrationTest(unittest.TestCase):
def test_pending_restart_is_completed_before_migration(self):
client = Mock()
client.success.side_effect = [
{'activeGeneration': 5, 'pendingRestart': ['database']}, {'status': 'accepted'},
]
with patch.object(migration, 'wait_ready') as wait:
self.assertEqual(migration.clear_pending_restart(client), 5)
wait.assert_called_once_with(client, 5)
self.assertEqual(client.success.call_args_list[1].args, ('/api/v1/system/restart',))
def test_no_pending_change_does_not_restart(self):
client = Mock()
client.success.return_value = {'activeGeneration': 5, 'pendingRestart': []}
with patch.object(migration, 'wait_ready') as wait:
self.assertEqual(migration.clear_pending_restart(client), 5)
wait.assert_not_called()
self.assertEqual(client.success.call_count, 1)
def test_missing_database_is_not_created(self):
with tempfile.TemporaryDirectory() as directory:
missing = Path(directory) / 'missing.db'
with self.assertRaises(sqlite3.OperationalError):
migration.database_fingerprint(missing)
self.assertFalse(missing.exists())
def test_fingerprint_detects_changes(self):
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / 'source.db'
with closing(sqlite3.connect(path)) as database:
database.execute('CREATE TABLE item (value TEXT)')
database.commit()
first = migration.database_fingerprint(path)
self.assertEqual(first, migration.database_fingerprint(path))
with closing(sqlite3.connect(path)) as database:
database.execute("INSERT INTO item VALUES ('changed')")
database.commit()
self.assertNotEqual(first, migration.database_fingerprint(path))
def test_history_requires_payload_and_digest_equality(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
(root / 'data').mkdir()
for name in ('smail.db', 'migrated.db'):
with closing(sqlite3.connect(root / 'data' / name)) as database:
database.execute('CREATE TABLE config_records (kind TEXT, id TEXT, payload BLOB, digest TEXT)')
database.execute('INSERT INTO config_records VALUES (?, ?, ?, ?)', ('backup', 'one', b'payload', 'digest'))
database.commit()
self.assertEqual(migration.verify_history_preserved(root)['source_records'], 1)
with closing(sqlite3.connect(root / 'data/migrated.db')) as database:
database.execute("UPDATE config_records SET payload = 'changed'")
database.commit()
with self.assertRaises(AssertionError):
migration.verify_history_preserved(root)
def test_integrity_distinguishes_active_and_offline_files(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
(root / 'config').mkdir()
(root / 'data').mkdir()
(root / 'config/config.json').write_text(json.dumps({
'database': {'type': 'sqlite', 'localRef': 'migrated.db'},
}))
with closing(sqlite3.connect(root / 'data/migrated.db')) as database:
database.execute('CREATE TABLE config_records (id INTEGER)')
database.execute('INSERT INTO config_records VALUES (1)')
database.commit()
(root / 'data/smail.db').touch()
(root / 'data/occupied.db').touch()
result = migration.migration_integrity(root)
self.assertEqual(result['active_database'], 'migrated.db')
self.assertEqual(result['config_records'], 1)
(root / 'data/smail-state.db').touch()
with self.assertRaises(AssertionError):
migration.migration_integrity(root)
def test_system_service_is_rejected_before_command(self):
with patch.object(migration.subprocess, 'check_output') as command:
for name in ('smail.service', 'mysql.service', 'smail-runtime-review-anything'):
with self.assertRaises(AssertionError):
migration.service_state(name)
command.assert_not_called()
def test_formal_root_rejected_before_command(self):
with patch.object(migration.subprocess, 'check_output') as command:
with self.assertRaises(AssertionError):
migration.verify_migration(Path('/opt/smail'), 'smail.service', 'not-a-secret', {})
command.assert_not_called()
def test_process_proof_rejects_zero_pid(self):
with self.assertRaises(AssertionError):
migration.process_uses_target(Path('/unused'), {'MainPID': '0'})
if __name__ == '__main__':
unittest.main()

View File

@ -0,0 +1,298 @@
"""通过隔离实机验证 SQLite 迁移、两种重启及目标不可读时的恢复。"""
import argparse
from contextlib import closing
from email.message import EmailMessage
import hashlib
import json
import os
from pathlib import Path
import re
import sqlite3
import ssl
import stat
import subprocess
import sys
import time
from runtime_probe import ACCOUNT, ADDRESS, Client, protocol_operation, smtp_connect, verify_protocol_read
def read_database(path):
# 只读打开,缺失数据库必须失败,不能由验收器创建来掩盖问题。
return closing(sqlite3.connect(path.resolve().as_uri() + '?mode=ro', uri=True, timeout=5))
def database_fingerprint(path):
with read_database(path) as database:
assert database.execute('PRAGMA integrity_check').fetchone() == ('ok',)
rows = list(database.iterdump())
return hashlib.sha256('\n'.join(rows).encode()).hexdigest()
def verify_history_preserved(root):
query = 'SELECT kind, id, payload, digest FROM config_records'
with read_database(root / 'data/smail.db') as source:
old = set(source.execute(query))
with read_database(root / 'data/migrated.db') as target:
current = set(target.execute(query))
assert old and old.issubset(current), '源库配置历史没有完整保留在目标库'
return {'source_records': len(old), 'target_records': len(current), 'all_source_records_equal': True}
def root_database(root):
data = json.loads((root / 'config/config.json').read_text())
database = data['database']
assert database['type'] == 'sqlite'
assert database['localRef'] in ('smail.db', 'migrated.db')
return database['localRef']
def migration_integrity(root):
active = root_database(root)
assert active == 'migrated.db'
names = sorted(path.name for path in (root / 'data').glob('*.db'))
assert names == ['migrated.db', 'occupied.db', 'smail.db'], '发现意外数据库文件'
with read_database(root / 'data' / active) as database:
assert database.execute('PRAGMA integrity_check').fetchone() == ('ok',)
count = database.execute('SELECT COUNT(*) FROM config_records').fetchone()[0]
assert count > 0
return {'active_database': active, 'config_records': count,
'retained_offline_files': ['smail.db', 'occupied.db'], 'integrity': 'ok'}
def wait_ready(client, generation=None):
for _ in range(70):
try:
if generation is None:
assert client.request('/api/ping', method='GET')[0] == 200
else:
status = client.success('/api/v1/config/status', method='GET')
assert status['activeGeneration'] == generation and not status['pendingRestart']
return
except (OSError, AssertionError):
time.sleep(1)
raise AssertionError('迁移后的应用服务未就绪')
def login(root, password):
context = ssl.create_default_context(cafile=str(root / 'ssl/runtime.crt'))
assert context.check_hostname and context.verify_mode == ssl.CERT_REQUIRED
client = Client(context)
wait_ready(client)
assert client.success('/api/v1/login', {'account': ACCOUNT, 'password': password})['account'] == ACCOUNT
client.success('/api/v1/security/reauth', {'password': password})
return client, context
def clear_pending_restart(client):
status = client.success('/api/v1/config/status', method='GET')
if status['pendingRestart']:
assert client.success('/api/v1/system/restart')['status'] == 'accepted'
wait_ready(client, status['activeGeneration'])
return status['activeGeneration']
def prepare(root, password):
client, _ = login(root, password)
generation = clear_pending_restart(client)
endpoint = '/api/v1/database/migrations'
# 预览时目标不存在;执行前故意占用,只能拒绝,不能覆盖已有文件。
preview = client.success(endpoint + '/preview', {
'expectedGeneration': generation, 'target': {'type': 'sqlite', 'localRef': 'occupied.db'},
})
occupied = root / 'data/occupied.db'
with occupied.open('xb') as stream:
stream.write(b'smail-migration-occupied-target\n')
occupied.chmod(0o600)
original = occupied.read_bytes()
status, rejected, _ = client.request(endpoint, {
'id': preview['migration']['id'], 'token': preview['token'],
})
assert status == 409 and rejected is not None and rejected['errorNo'] != 0
assert rejected['data']['code'] == 'DATABASE_MIGRATION_TARGET_EXISTS'
assert occupied.read_bytes() == original and root_database(root) == 'smail.db'
failed = client.success(endpoint + '/' + preview['migration']['id'], method='GET')
assert failed['status'] == 'failed' and failed['errorCode'] == 'TARGET_EXISTS'
assert client.success('/api/v1/user/info')['account'] == ACCOUNT
# 写入实际配置,证明失败后没有卡在只读维护状态。
from runtime_probe import verify_config_history
verify_config_history(client)
generation = clear_pending_restart(client)
preview = client.success(endpoint + '/preview', {
'expectedGeneration': generation, 'target': {'type': 'sqlite', 'localRef': 'migrated.db'},
})
record = client.success(endpoint, {'id': preview['migration']['id'], 'token': preview['token']})
assert record['status'] == 'succeeded'
assert root_database(root) == 'migrated.db'
pending = client.success('/api/v1/config/status', method='GET')
assert pending['activeGeneration'] > generation and pending['pendingRestart']
# 成功切换到新配置但尚未重启时,不允许继续写业务。
denied_status, denied, _ = client.request('/api/v1/security/credentials/protocols/generate', {'protocol': 'imap'})
assert denied_status == 503 and denied is not None and denied['data'] == 'DATABASE_RESTART_REQUIRED'
assert client.success('/api/v1/system/restart')['status'] == 'accepted'
wait_ready(client, pending['activeGeneration'])
assert client.success('/api/v1/user/info')['account'] == ACCOUNT
return {'occupied_target_rejected': True, 'occupied_target_unchanged': True,
'failed_record': failed['status'], 'failure_releases_maintenance': True,
'switch_succeeded': True, 'pending_write_denied': True,
'internal_restart_completed': True, 'existing_session_preserved': True,
'generation': pending['activeGeneration'], 'rejection_http_status': status}
def verify_mail(root, password, phase):
client, context = login(root, password)
assert root_database(root) == 'migrated.db'
assert client.success('/api/v1/user/info')['account'] == ACCOUNT
status = client.success('/api/v1/config/status', method='GET')
assert not status['pendingRestart']
codes = {}
for protocol in ('smtp', 'imap', 'pop3'):
codes[protocol] = client.success('/api/v1/security/credentials/protocols/generate', {'protocol': protocol})['code']
assert protocol_operation(protocol, context, codes[protocol])() == 'success'
original = json.loads((root.parent / 'probe-result.json').read_text())['smtp']['message_id']
body = '内部25端口真实收信,检查中文正文和附件。'
attachment = '隔离验收附件内容'.encode('utf-8')
historical = verify_protocol_read(context, codes, original, body, attachment)
message_id = original + '.' + phase
message = EmailMessage()
message['From'], message['To'] = 'migration@sender.invalid', ADDRESS
message['Subject'] = '切库重启后的中文邮件验收'
message['Message-ID'] = '<' + message_id + '>'
message.set_content(body)
message.add_attachment(attachment, maintype='text', subtype='plain', filename='验收附件.txt')
with smtp_connect(context) as smtp:
assert not smtp.send_message(message)
received = verify_protocol_read(context, codes, message_id, body, attachment)
with read_database(root / 'data/migrated.db') as database:
assert database.execute('SELECT COUNT(*) FROM email WHERE msg_id=?', (message_id,)).fetchone()[0] == 1
assert database.execute('SELECT COUNT(*) FROM config_records').fetchone()[0] > 0
with read_database(root / 'data/smail.db') as database:
assert database.execute('SELECT COUNT(*) FROM email WHERE msg_id=?', (message_id,)).fetchone()[0] == 0
return {'login': True, 'session': True, 'historical_mail': historical,
'new_mail': received, 'new_mail_only_in_target': True, 'message_id': message_id}
def service_state(unit):
assert re.fullmatch(r'smail-runtime-review-\d{8}-\d{6}', unit)
output = subprocess.check_output(['systemctl', 'show', unit,
'--property=MainPID,InvocationID,ActiveState,SubState,Result,ExecMainStatus'], text=True, timeout=15)
return dict(line.split('=', 1) for line in output.splitlines())
def process_uses_target(root, state):
pid = state['MainPID']
assert pid.isdigit() and int(pid) > 0
assert Path('/proc/' + pid + '/root/opt/smail/smail').stat().st_ino == (root / 'smail').stat().st_ino
paths = []
for item in Path('/proc/' + pid + '/fd').iterdir():
try:
paths.append(str(item.readlink()))
except FileNotFoundError:
pass
assert '/opt/smail/data/migrated.db' in paths, '进程未持有迁移目标库'
assert '/opt/smail/data/smail.db' not in paths, '进程仍然打开旧业务库'
def run_probe(root, unit, password, phase):
state = service_state(unit)
pid = state['MainPID']
assert pid.isdigit() and int(pid) > 0
assert Path('/proc/' + pid + '/ns/net').readlink() != Path('/proc/self/ns/net').readlink()
probe = 'smail-migration-probe-' + unit.removeprefix('smail-runtime-review-') + '-' + phase
command = ['systemd-run', '--wait', '--pipe', '--unit=' + probe,
'--property=User=smail', '--property=Group=smail',
'--property=NetworkNamespacePath=/proc/' + pid + '/ns/net',
'--property=ProtectSystem=strict', '--property=ProtectHome=yes',
'--property=NoNewPrivileges=yes', '--property=InaccessiblePaths=/opt/smail /run/smail',
'--property=ReadWritePaths=' + str(root.parent), '--property=MemoryMax=48M',
'--property=MemorySwapMax=0', '--property=CPUQuota=25%', '--property=TasksMax=32',
'--property=RuntimeMaxSec=150s', '--property=KillMode=control-group',
'--property=TimeoutStopSec=10s', 'python3', str(root.parent / 'runtime_migration.py'),
str(root), phase]
try:
completed = subprocess.run(command, input=password + '\n', text=True,
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=175)
(root.parent / ('migration-' + phase + '.log')).write_text(completed.stdout, encoding='utf-8')
assert completed.returncode == 0, '迁移探针失败,请检查独立阶段日志'
return json.loads((root.parent / ('migration-' + phase + '.json')).read_text())
finally:
# 瞬时单元可能已经回收;只有确认无进程且不活动才接受未加载返回值。
stopped = subprocess.run(['systemctl', 'stop', probe], capture_output=True, text=True, timeout=25)
output = subprocess.check_output(['systemctl', 'show', probe,
'--property=MainPID,ActiveState,LoadState'], text=True, timeout=15)
values = dict(line.split('=', 1) for line in output.splitlines())
assert values['MainPID'] == '0' and values['ActiveState'] in ('inactive', 'failed')
assert stopped.returncode == 0 or (stopped.returncode == 5 and values['LoadState'] == 'not-found')
def verify_migration(root, unit, password, result):
assert root.parent.parent == Path('/var/tmp') and re.fullmatch(
r'smail-review-runtime-\d{8}-\d{6}', root.parent.name)
assert root.name == 'instance' and not root.is_symlink()
assert root.resolve(strict=True) == root
details = result['migration'] = {'passed': False}
before = service_state(unit)
details['prepare'] = run_probe(root, unit, password, 'prepare')
internal = service_state(unit)
assert before['MainPID'] == internal['MainPID'] and before['InvocationID'] == internal['InvocationID']
process_uses_target(root, internal)
details['internal_restart_same_process'] = True
source_digest = database_fingerprint(root / 'data/smail.db')
details['internal'] = run_probe(root, unit, password, 'internal')
# 必须实际更换进程,不能把 API 的 accepted 或管理器 reload 当作系统重启。
subprocess.run(['systemctl', 'restart', unit], check=True, timeout=55)
restarted = service_state(unit)
assert restarted['MainPID'] != internal['MainPID'] and restarted['InvocationID'] != internal['InvocationID']
details['process'] = run_probe(root, unit, password, 'process')
process_uses_target(root, restarted)
details['process_states'] = {'before': before, 'internal': internal, 'restarted': restarted}
target = root / 'data/migrated.db'
assert target.is_file() and not target.is_symlink()
mode = stat.S_IMODE(target.stat().st_mode)
assert mode == 0o600
try:
# 只收紧本轮唯一目标文件权限,故障时绝不回写或清空正式库。
target.chmod(0)
subprocess.run(['systemctl', 'restart', unit], check=True, timeout=55)
for _ in range(60):
failed = service_state(unit)
if failed['MainPID'] == '0' and failed['ActiveState'] == 'failed':
break
time.sleep(1)
else:
raise AssertionError('目标库不可读时没有明确停止,可能静默回退')
assert root_database(root) == 'migrated.db'
assert database_fingerprint(root / 'data/smail.db') == source_digest
details['unavailable_target'] = {'failed_closed': True, 'source_unchanged': True,
'configuration_preserved': True, 'state': failed}
finally:
target.chmod(mode)
subprocess.run(['systemctl', 'start', unit], check=True, timeout=55)
details['recovered'] = run_probe(root, unit, password, 'recovered')
final = service_state(unit)
process_uses_target(root, final)
assert database_fingerprint(root / 'data/smail.db') == source_digest
assert final['MainPID'] not in (before['MainPID'], restarted['MainPID'])
details['history'] = verify_history_preserved(root)
details.update(source_unchanged=True, target_permissions_restored=True, final_state=final, passed=True)
(root.parent / 'migration-result.json').write_text(json.dumps(details, ensure_ascii=False, indent=2), encoding='utf-8')
def main():
parser = argparse.ArgumentParser()
parser.add_argument('root', type=Path)
parser.add_argument('phase', choices=('prepare', 'internal', 'process', 'recovered'))
args = parser.parse_args()
assert os.geteuid() != 0, 'HTTP 与邮件探针必须使用非特权用户'
password = sys.stdin.readline().strip()
assert len(password) == 12
result = prepare(args.root, password) if args.phase == 'prepare' else verify_mail(args.root, password, args.phase)
result['passed'] = True
(args.root.parent / ('migration-' + args.phase + '.json')).write_text(
json.dumps(result, ensure_ascii=False, indent=2), encoding='utf-8')
print(json.dumps(result, ensure_ascii=False, indent=2))
if __name__ == '__main__':
main()

View File

@ -79,8 +79,11 @@ def main():
parser.add_argument('--nginx-template', type=Path)
parser.add_argument('--memory-max-mib', type=int, choices=(128, 192, 256), default=128)
parser.add_argument('--restore', action='store_true')
parser.add_argument('--migration', action='store_true')
parser.add_argument('--database-type', choices=('sqlite', 'mysql', 'postgres'), default='sqlite')
args = parser.parse_args()
if args.migration and (args.database_type != 'sqlite' or args.nginx_template or args.restore):
parser.error('迁移进程验收当前只覆盖直接 TLS 的 SQLite 独立场景')
if args.restore and args.database_type != 'sqlite':
parser.error('SQL 数据库备份恢复需要独立验收,不能套用 SQLite 冷备流程')
assert os.geteuid() == 0 and run('hostname') == 'ikun2'
@ -116,6 +119,10 @@ def main():
shutil.copy2(Path(__file__).with_name('runtime_probe.py'), target / 'runtime_probe.py')
(target / 'runtime_probe.py').chmod(0o640)
shutil.chown(target / 'runtime_probe.py', user='root', group='smail')
if args.migration:
shutil.copy2(Path(__file__).with_name('runtime_migration.py'), target / 'runtime_migration.py')
(target / 'runtime_migration.py').chmod(0o640)
shutil.chown(target / 'runtime_migration.py', user='root', group='smail')
if args.database_type != 'sqlite':
shutil.copy2(Path(__file__).with_name('runtime_sql.py'), target / 'runtime_sql.py')
(target / 'runtime_sql.py').chmod(0o640)
@ -231,6 +238,9 @@ def main():
message_id = result['probe']['smtp']['message_id']
assert database.execute('SELECT COUNT(*) FROM email WHERE msg_id=?', (message_id,)).fetchone()[0] == 0
result['live_message_absent'] = True
if args.migration:
from runtime_migration import verify_migration
verify_migration(root, unit, password, result)
finally:
try:
collect_and_stop(result, unit, probe_unit, proxy_unit if args.nginx_template else None)
@ -239,7 +249,11 @@ def main():
result['live_unchanged'] = (result['after'] == before
and hashlib.file_digest(stream, 'sha256').hexdigest() == live_sha)
try:
result['integrity'] = sql_fixture.integrity(root) if sql_fixture else database_integrity(root)
if args.migration:
from runtime_migration import migration_integrity
result['integrity'] = migration_integrity(root)
else:
result['integrity'] = sql_fixture.integrity(root) if sql_fixture else database_integrity(root)
except (AssertionError, sqlite3.Error, RuntimeError, OSError, ValueError, subprocess.SubprocessError) as error:
result['integrity'] = {'error': type(error).__name__}
finally:
@ -266,6 +280,8 @@ def main():
assert not result['resource_errors'], '资源证据不完整,需要核查保留的结果'
if sql_fixture:
assert result['integrity'].get('sqlite_files') == 0 and result['integrity']['config_records'] > 0
elif args.migration:
assert result['migration']['passed'] and result['integrity']['active_database'] == 'migrated.db'
else:
assert result['integrity'] == {'smail.db': [('ok',)]}
if args.restore: