diff --git a/test/deployment/test_formal_review_tools.py b/test/deployment/test_formal_review_tools.py index 1199c76..92f9d3f 100644 --- a/test/deployment/test_formal_review_tools.py +++ b/test/deployment/test_formal_review_tools.py @@ -14,6 +14,9 @@ import deploy_review import internal_smtp_check import restore_review import verify_review +import runtime_review +from database_inventory import sqlite_inventory +from unittest.mock import patch class FormalReviewDatabaseTest(unittest.TestCase): @@ -63,5 +66,86 @@ class FormalReviewDatabaseTest(unittest.TestCase): self.assertEqual(list((root / 'data').iterdir()), []) + def test_extra_sqlite_files_cannot_hide_by_name_or_location(self): + for name in ('extra.db', 'STATE.DB', '.hidden.sqlite3', 'nested/extra.sqlite', + '../config/other.db', '../renamed.bin', 'old.db-wal', 'old.sqlite-shm'): + with self.subTest(name=name), tempfile.TemporaryDirectory() as directory: + root = Path(directory).resolve() + self.make_database(root) + path = root / 'data' / name + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(b'SQLite format 3\x00') + for module in self.modules: + with self.subTest(module=module.__name__), self.assertRaises(AssertionError): + module.database_path(root) + with self.assertRaises(AssertionError): + runtime_review.database_integrity(root) + + def test_expected_sqlite_sidecars_are_not_second_databases(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory).resolve() + self.make_database(root) + for suffix in ('-wal', '-shm', '-journal'): + (root / 'data' / ('smail.db' + suffix)).touch() + for module in self.modules: + self.assertEqual(module.database_path(root), root / 'data/smail.db') + + def test_sql_mode_rejects_renamed_database_and_orphan_sidecars(self): + for name in ('renamed.bin', 'data/.hidden.sqlite', 'data/old.db-wal'): + with self.subTest(name=name), tempfile.TemporaryDirectory() as directory: + root = Path(directory).resolve() + (root / 'data').mkdir() + self.assertEqual(sqlite_inventory(root, []), []) + (root / name).write_bytes(b'SQLite format 3\x00') + with self.assertRaises(AssertionError): + sqlite_inventory(root, []) + + def test_directory_links_are_rejected_before_reading_their_contents(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory).resolve() + self.make_database(root) + linked = root / 'data' / 'linked' + linked.mkdir() + original = Path.is_symlink + with patch.object(Path, 'is_symlink', lambda path: path == linked or original(path)): + with self.assertRaises(AssertionError): + sqlite_inventory(root, [root / 'data/smail.db']) + + def test_snapshot_and_restore_counts_close_their_connections(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory).resolve() + self.make_database(root) + source = root / 'data/smail.db' + target = root / 'snapshot.db' + deploy_review.snapshot(source, target) + moved = root / 'snapshot-moved.db' + target.rename(moved) + with closing(sqlite3.connect(moved)) as database: + self.assertEqual(database.execute('PRAGMA integrity_check').fetchone(), ('ok',)) + isolated = root / 'isolated' + isolated.mkdir() + self.make_database(isolated) + self.assertEqual(restore_review.database_counts(isolated), {'smail.db': {'marker': 0}}) + (isolated / 'data/smail.db').rename(isolated / 'data/moved.db') + + + def test_deployed_schema_requires_unified_configuration_history(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory).resolve() + self.make_database(root) + with closing(sqlite3.connect(root / 'data/smail.db')) as database: + database.executescript('CREATE TABLE schema_migration (version INTEGER); ' + 'INSERT INTO schema_migration VALUES (23); ' + 'CREATE TABLE outbound_delivery (state TEXT); ' + "INSERT INTO outbound_delivery VALUES ('pending');") + with self.assertRaises(AssertionError): + deploy_review.database_state(root) + database.executescript('INSERT INTO schema_migration VALUES (24);') + with self.assertRaises(sqlite3.OperationalError): + deploy_review.database_state(root) + database.executescript('CREATE TABLE config_records (id TEXT);') + self.assertEqual(deploy_review.database_state(root), (24, {'pending': 1})) + + if __name__ == '__main__': unittest.main() \ No newline at end of file diff --git a/test/tools/database_inventory.py b/test/tools/database_inventory.py new file mode 100644 index 0000000..a8b2bb9 --- /dev/null +++ b/test/tools/database_inventory.py @@ -0,0 +1,31 @@ +from pathlib import Path +import stat + + +def sqlite_inventory(root, expected): + root = Path(root) + assert root.is_absolute() and root.resolve(strict=True) == root + assert (root / 'data').is_dir() + expected = set(expected) + assert all(path.is_relative_to(root) for path in expected) + seen = set() + pending = [root] + while pending: + path = pending.pop() + assert not path.is_symlink() and not getattr(path, 'is_junction', lambda: False)() + mode = path.stat().st_mode + if stat.S_ISDIR(mode): + pending.extend(path.iterdir()) + continue + assert stat.S_ISREG(mode) + name = path.name.lower() + if name.endswith(('-wal', '-shm', '-journal')): + assert path.with_name(path.name.rsplit('-', 1)[0]) in expected + continue + with path.open('rb') as stream: + header = stream.read(16) + if header == b'SQLite format 3\x00' or name.endswith('.db') or '.sqlite' in name: + assert path in expected + seen.add(path) + assert seen == expected + return sorted(seen) diff --git a/test/tools/deploy_review.py b/test/tools/deploy_review.py index 66a4141..b087fb9 100644 --- a/test/tools/deploy_review.py +++ b/test/tools/deploy_review.py @@ -7,9 +7,12 @@ import os from pathlib import Path import shutil import sqlite3 +from contextlib import closing import subprocess import time +from database_inventory import sqlite_inventory + ROOT = Path('/opt/smail') STAGING = Path('/var/tmp/smail-review-20260917') @@ -27,8 +30,7 @@ def digest(path): def database_path(root): data = root / 'data' - files = sorted(path for path in data.rglob('*.db') - if path.is_file() or path.is_symlink()) + files = sqlite_inventory(root, [data / 'smail.db']) expected = data / DATABASE_NAME assert files == [expected], '正式验收要求运行目录仅保留 data/smail.db' assert expected.is_file() and not expected.is_symlink(), '活动数据库不能是符号链接' @@ -36,19 +38,31 @@ def database_path(root): def snapshot(source, target): - with sqlite3.connect(f'file:{source}?mode=ro', uri=True) as original: - with sqlite3.connect(target) as copied: + with closing(sqlite3.connect(source.as_uri() + '?mode=ro', uri=True)) as original: + with closing(sqlite3.connect(target)) as copied: original.backup(copied) assert copied.execute('PRAGMA integrity_check').fetchone()[0] == 'ok' target.chmod(0o600) + +def database_state(root): + database = database_path(root) + with closing(sqlite3.connect(database.as_uri() + '?mode=ro', uri=True)) as connection: + assert connection.execute('PRAGMA integrity_check').fetchall() == [('ok',)] + version = connection.execute('SELECT MAX(version) FROM schema_migration').fetchone()[0] + assert version == 24 + connection.execute('SELECT COUNT(*) FROM config_records').fetchone() + states = dict(connection.execute('SELECT state,COUNT(*) FROM outbound_delivery GROUP BY state')) + return version, states + + def wait_ready(): for _ in range(60): result = subprocess.run([ 'curl', '--silent', '--fail', '--max-time', '2', '--noproxy', '*', '--cacert', str(ROOT / 'ssl/custom.crt'), '--resolve', - 'mail.solit.cn:443:127.0.0.1', 'https://mail.solit.cn/api/ping', + 'web.solit.cn:443:127.0.0.1', 'https://web.solit.cn/api/ping', ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) if result.returncode == 0 and run('systemctl', 'is-active', 'smail.service') == 'active': return @@ -98,12 +112,7 @@ def main(): assert digest(ROOT / name) == checksum subprocess.run(['systemctl', 'start', 'smail.service'], check=True) wait_ready() - 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(connection.execute('SELECT state,COUNT(*) FROM outbound_delivery GROUP BY state')) + version, states = database_state(ROOT) result = {'backup': str(backup), 'archive_sha256': digest(archive), 'artifacts': expected, 'migration': version, 'queue_states': states, 'service': run('systemctl', 'is-active', 'smail.service'), diff --git a/test/tools/internal_smtp_check.py b/test/tools/internal_smtp_check.py index 373a427..9bf9923 100644 --- a/test/tools/internal_smtp_check.py +++ b/test/tools/internal_smtp_check.py @@ -5,6 +5,7 @@ import hashlib import json import smtplib import sqlite3 +from contextlib import closing import ssl import subprocess import time @@ -13,6 +14,8 @@ from email.policy import SMTP from email.utils import formatdate from pathlib import Path +from database_inventory import sqlite_inventory + ROOT = Path('/opt/smail') DATABASE_NAME = 'smail.db' @@ -24,8 +27,7 @@ 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()) + files = sqlite_inventory(root, [data / 'smail.db']) expected = data / DATABASE_NAME assert files == [expected], '正式验收要求运行目录仅保留 data/smail.db' assert expected.is_file() and not expected.is_symlink(), '活动数据库不能是符号链接' @@ -79,12 +81,12 @@ def main(): backup = Path('/var/backups') / f'smail-internal-smtp-{RUN}' backup.mkdir(mode=0o700) 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: + with closing(sqlite3.connect(database_path_value.as_uri() + '?mode=ro', uri=True)) as source: + with closing(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: + with closing(sqlite3.connect(database_path_value.as_uri() + '?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' diff --git a/test/tools/restore_review.py b/test/tools/restore_review.py index 9c79ba5..7e80794 100644 --- a/test/tools/restore_review.py +++ b/test/tools/restore_review.py @@ -7,10 +7,13 @@ import os from pathlib import Path import shutil import sqlite3 +from contextlib import closing import subprocess import tarfile import time +from database_inventory import sqlite_inventory + DATABASE_NAME = 'smail.db' @@ -21,8 +24,7 @@ def run(*args): def database_path(root): data = root / 'data' - files = sorted(path for path in data.rglob('*.db') - if path.is_file() or path.is_symlink()) + files = sqlite_inventory(root, [data / 'smail.db']) expected = data / DATABASE_NAME assert files == [expected], '正式验收要求运行目录仅保留 data/smail.db' assert expected.is_file() and not expected.is_symlink(), '活动数据库不能是符号链接' @@ -32,7 +34,7 @@ def database_path(root): def database_counts(root): counts = {} database_path_value = database_path(root) - with sqlite3.connect(f'file:{database_path_value}?mode=ro', uri=True) as database: + with closing(sqlite3.connect(database_path_value.as_uri() + '?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] = { @@ -95,7 +97,7 @@ def main(): probe = subprocess.run([ 'nsenter', '--target', pid, '--net', 'curl', '--silent', '--fail', '--max-time', '2', '--noproxy', '*', '--cacert', str(restored / 'ssl/custom.crt'), - '--resolve', 'mail.solit.cn:443:127.0.0.1', 'https://mail.solit.cn/api/ping', + '--resolve', 'web.solit.cn:443:127.0.0.1', 'https://web.solit.cn/api/ping', ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) if probe.returncode == 0: ready = True @@ -124,9 +126,9 @@ with smtplib.SMTP('127.0.0.1',25,timeout=20) as client: ''' subprocess.run(['nsenter', '--target', pid, '--net', 'python3', '-c', probe, str(restored / 'ssl/custom.crt'), stamp], check=True) - with sqlite3.connect(f'file:{restored / "data/smail.db"}?mode=ro', uri=True) as database: + with closing(sqlite3.connect(f'file:{restored / "data/smail.db"}?mode=ro', uri=True)) as database: assert database.execute('SELECT COUNT(*) FROM email WHERE msg_id=?', ('restore-' + stamp + '@review.invalid',)).fetchone()[0] == 1 - with sqlite3.connect('file:/opt/smail/data/smail.db?mode=ro', uri=True) as database: + with closing(sqlite3.connect('file:/opt/smail/data/smail.db?mode=ro', uri=True)) as database: assert database.execute('SELECT COUNT(*) FROM email WHERE msg_id=?', ('restore-' + stamp + '@review.invalid',)).fetchone()[0] == 0 result.update(ready=True, isolated=True, restored_smtp=True, live_pid_unchanged=live_pid == run( 'systemctl', 'show', 'smail.service', '--property=MainPID', '--value')) diff --git a/test/tools/runtime_review.py b/test/tools/runtime_review.py index a5775c9..8577611 100644 --- a/test/tools/runtime_review.py +++ b/test/tools/runtime_review.py @@ -13,6 +13,8 @@ import sqlite3 import subprocess import time +from database_inventory import sqlite_inventory + def run(*args): return subprocess.check_output(args, text=True).strip() @@ -28,8 +30,7 @@ def database_integrity(root): # 验收器不能创建数据库，也不能允许第二个状态库。 data = root / 'data' assert not (data / 'smail-state.db').exists(), '单数据库实例生成了旧状态库' - files = sorted(path for path in data.rglob('*.db') - if path.is_file() or path.is_symlink()) + files = sqlite_inventory(root, [data / 'smail.db']) assert files == [data / 'smail.db'] assert (data / 'smail.db').is_file() and not (data / 'smail.db').is_symlink() with closing(sqlite3.connect((data / 'smail.db').as_uri() + '?mode=ro', uri=True)) as database: @@ -173,6 +174,9 @@ def main(): shutil.chown(directory, user='smail', group='smail') shutil.copy2(binary, root / 'smail') (root / 'smail').chmod(0o755) + shutil.copy2(Path(__file__).with_name('database_inventory.py'), target / 'database_inventory.py') + (target / 'database_inventory.py').chmod(0o640) + shutil.chown(target / 'database_inventory.py', user='root', group='smail') 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') @@ -351,7 +355,7 @@ def main(): result['probe'] = json.loads((target / 'probe-result.json').read_text()) assert probe.returncode == 0, '隔离验收失败，请检查保留的 probe.log' assert result['probe']['passed'] - with sqlite3.connect(f'file:{live / "data/smail.db"}?mode=ro', uri=True) as database: + with closing(sqlite3.connect(f'file:{live / "data/smail.db"}?mode=ro', uri=True)) as database: 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 diff --git a/test/tools/runtime_sql.py b/test/tools/runtime_sql.py index b55cc97..fbb14d8 100644 --- a/test/tools/runtime_sql.py +++ b/test/tools/runtime_sql.py @@ -7,6 +7,8 @@ import re import secrets import subprocess +from database_inventory import sqlite_inventory + def command(args, sql, env=None): result = subprocess.run(args, input=sql, text=True, capture_output=True, @@ -75,8 +77,7 @@ class SQLDatabase: def integrity(self, root): # 配置历史和业务表必须位于所选 SQL 库，运行目录不能悄悄创建 SQLite。 - assert not list((root / 'data').rglob('*.db')) - assert not list((root / 'data').rglob('*.sqlite*')) + sqlite_inventory(root, []) count = int(self.query('SELECT COUNT(*) FROM config_records')) assert count > 0 return {'type': self.kind, 'database': self.settings['name'], diff --git a/test/tools/verify_review.py b/test/tools/verify_review.py index 544b021..11771f9 100644 --- a/test/tools/verify_review.py +++ b/test/tools/verify_review.py @@ -6,9 +6,12 @@ import json import os from pathlib import Path import sqlite3 +from contextlib import closing import stat import subprocess +from database_inventory import sqlite_inventory + DATABASE_NAME = 'smail.db' @@ -19,8 +22,7 @@ def run(*args): def database_path(root): data = root / 'data' - files = sorted(path for path in data.rglob('*.db') - if path.is_file() or path.is_symlink()) + files = sqlite_inventory(root, [data / 'smail.db']) expected = data / DATABASE_NAME assert files == [expected], '正式验收要求运行目录仅保留 data/smail.db' assert expected.is_file() and not expected.is_symlink(), '活动数据库不能是符号链接' @@ -72,7 +74,7 @@ def main(): summary['admin_version'] = run(str(root / 'smail-admin'), 'version') summary['database_integrity'] = {} database = database_path(root) - with sqlite3.connect(f'file:{database}?mode=ro', uri=True) as connection: + with closing(sqlite3.connect(database.as_uri() + '?mode=ro', uri=True)) as connection: integrity = connection.execute('PRAGMA integrity_check').fetchall() assert integrity == [('ok',)] summary['database_integrity'][DATABASE_NAME] = 'ok' @@ -83,11 +85,11 @@ def main(): 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'] - run(*curl, 'https://mail.solit.cn/api/ping') + '--cacert', str(root / 'ssl/custom.crt'), '--resolve', 'web.solit.cn:443:127.0.0.1'] + run(*curl, 'https://web.solit.cn/api/ping') summary['unauthenticated'] = {} for endpoint in ('/api/v1/user/info', '/api/v1/email/list'): - response = json.loads(run(*curl, '-X', 'POST', 'https://mail.solit.cn' + endpoint)) + response = json.loads(run(*curl, '-X', 'POST', 'https://web.solit.cn' + endpoint)) assert response['errorNo'] == 403 and not response.get('data') summary['unauthenticated'][endpoint] = response['errorNo'] entries = run('journalctl', '-u', 'smail.service', '--since', args.since, '--output=json', '--no-pager')