diff --git a/test/deployment/test_runtime_migration.py b/test/deployment/test_runtime_migration.py new file mode 100644 index 0000000..74cb4cd --- /dev/null +++ b/test/deployment/test_runtime_migration.py @@ -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() diff --git a/test/tools/runtime_migration.py b/test/tools/runtime_migration.py new file mode 100644 index 0000000..e0dcdea --- /dev/null +++ b/test/tools/runtime_migration.py @@ -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() diff --git a/test/tools/runtime_review.py b/test/tools/runtime_review.py index c160229..61123e6 100644 --- a/test/tools/runtime_review.py +++ b/test/tools/runtime_review.py @@ -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: