From 22d4a84db75365bfcea70226bbabf853ba13f8d8 Mon Sep 17 00:00:00 2001 From: smail Date: Mon, 21 Sep 2026 00:59:59 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A1=A5=E5=85=85=E5=8D=95=E5=BA=93=E9=A6=96?= =?UTF-8?q?=E6=AC=A1=E5=AE=89=E8=A3=85=E5=92=8C=E9=82=AE=E4=BB=B6=E5=8D=8F?= =?UTF-8?q?=E8=AE=AE=E5=AE=9E=E6=9C=BA=E9=AA=8C=E6=94=B6=E5=99=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- test/deployment/test_runtime_sql.py | 126 +++++++++++++++++++++++++++ test/tools/runtime_probe.py | 71 ++++++++++++--- test/tools/runtime_review.py | 62 +++++++++++--- test/tools/runtime_sql.py | 128 ++++++++++++++++++++++++++++ test/tools/runtime_sql_bridge.py | 64 ++++++++++++++ 5 files changed, 428 insertions(+), 23 deletions(-) create mode 100644 test/deployment/test_runtime_sql.py create mode 100644 test/tools/runtime_sql.py create mode 100644 test/tools/runtime_sql_bridge.py diff --git a/test/deployment/test_runtime_sql.py b/test/deployment/test_runtime_sql.py new file mode 100644 index 0000000..9bb81f0 --- /dev/null +++ b/test/deployment/test_runtime_sql.py @@ -0,0 +1,126 @@ +"""SQL 实机验收器的范围限制、敏感信息和命名空间恢复回归。""" + +from pathlib import Path +import subprocess +import sys +import tempfile +import unittest +from unittest.mock import MagicMock, call, patch + + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'tools')) +import runtime_sql +import runtime_sql_bridge + + +def settings(kind='mysql'): + return {'kind': kind, 'name': 'smail_runtime_test_123456abcdef', + 'user': 'smail_rt_123456abcdef', 'password': 'a' * 40} + + +class RuntimeSQLTest(unittest.TestCase): + def test_rejects_other_projects_before_any_command(self): + for key, value in [('name', 'smail'), ('name', 'production'), + ('name', "smail_runtime_test_123456abcdef'; DROP TABLE user"), + ('user', 'root'), ('kind', 'sqlite'), ('password', "a';SELECT 1")]: + with self.subTest(key=key), patch.object(runtime_sql, 'command') as command: + candidate = settings() + candidate[key] = value + with self.assertRaises(AssertionError): + runtime_sql.SQLDatabase(candidate) + command.assert_not_called() + + def test_password_is_not_in_command_arguments(self): + for kind, key in [('mysql', 'MYSQL_PWD'), ('postgres', 'PGPASSWORD')]: + with self.subTest(kind=kind), patch.object(runtime_sql, 'command', return_value='0') as command: + database = runtime_sql.SQLDatabase(settings(kind)) + database.assert_empty_users() + args, sql, env = command.call_args.args + self.assertNotIn(settings()['password'], ' '.join(args)) + self.assertEqual(env[key], settings()['password']) + self.assertIn('COUNT(*)', sql) + + def test_driver_error_details_are_not_exposed(self): + failure = subprocess.CompletedProcess([], 1, '', 'sensitive-test-value') + with patch.object(runtime_sql.subprocess, 'run', return_value=failure): + with self.assertRaises(RuntimeError) as caught: + runtime_sql.command(['mysql'], 'SELECT 1') + self.assertNotIn('sensitive-test-value', str(caught.exception)) + + def test_fixture_grants_only_exact_database_and_locks_account(self): + for kind in ('mysql', 'postgres'): + with self.subTest(kind=kind): + fixture = runtime_sql.SQLFixture(kind) + with patch.object(fixture, 'admin', return_value='0') as admin: + fixture.create() + fixture.lock() + statements = '\n'.join(item.args[0] for item in admin.call_args_list) + self.assertNotIn('DROP ', statements) + self.assertNotIn('*.*', statements) + self.assertNotIn('CREATEDB', statements) + if kind == 'mysql': + self.assertIn(fixture.settings['name'].replace('_', r'\_'), statements) + self.assertIn('ACCOUNT LOCK', statements) + else: + self.assertIn('NOLOGIN', statements) + + def test_uncreated_account_does_not_trigger_admin_command(self): + fixture = runtime_sql.SQLFixture('mysql') + with patch.object(fixture, 'admin') as admin: + fixture.lock() + admin.assert_not_called() + + def test_nonempty_database_fails_first_install_check(self): + fixture = runtime_sql.SQLFixture('postgres') + with patch.object(fixture, 'admin', return_value='3'): + with self.assertRaises(AssertionError): + fixture.create() + + def test_sql_mode_rejects_hidden_second_sqlite_database(self): + database = runtime_sql.SQLDatabase(settings()) + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / 'data/nested').mkdir(parents=True) + (root / 'data/nested/smail-state.db').touch() + with patch.object(database, 'query') as query: + with self.assertRaises(AssertionError): + database.integrity(root) + query.assert_not_called() + + def test_sql_mode_requires_configuration_history_in_same_database(self): + database = runtime_sql.SQLDatabase(settings()) + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / 'data').mkdir() + with patch.object(database, 'query', return_value='0'): + with self.assertRaises(AssertionError): + database.integrity(root) + with patch.object(database, 'query', return_value='5'): + self.assertEqual(database.integrity(root)['sqlite_files'], 0) + + def test_bridge_rejects_non_database_port_before_opening_socket(self): + with patch.object(runtime_sql_bridge.socket, 'socket') as socket: + with self.assertRaises(AssertionError): + runtime_sql_bridge.DatabaseBridge(1234, 22) + socket.assert_not_called() + + def test_bridge_restores_host_namespace_when_bind_fails(self): + handles = [] + for descriptor in (11, 12): + handle = MagicMock() + handle.__enter__.return_value = handle + handle.fileno.return_value = descriptor + handles.append(handle) + with patch.object(runtime_sql_bridge.Path, 'readlink', side_effect=[Path('net2'), Path('net1')]), \ + patch('builtins.open', side_effect=handles), \ + patch.object(runtime_sql_bridge.os, 'setns', create=True) as setns, \ + patch.object(runtime_sql_bridge.socketserver.ThreadingTCPServer, '__init__', side_effect=OSError()), \ + patch.object(runtime_sql_bridge.threading, 'Thread') as thread: + with self.assertRaises(OSError): + runtime_sql_bridge.DatabaseBridge(1234, 3306) + self.assertEqual(setns.call_args_list, [call(12, 0), call(11, 0)]) + thread.assert_not_called() + + +if __name__ == '__main__': + unittest.main() diff --git a/test/tools/runtime_probe.py b/test/tools/runtime_probe.py index 6287af3..0f083f6 100644 --- a/test/tools/runtime_probe.py +++ b/test/tools/runtime_probe.py @@ -185,6 +185,38 @@ def protocol_operation(protocol, context, code): return invoke +def verify_imap_append(context, code, message_id, body, attachment): + appended_id = 'append-' + message_id + message = EmailMessage() + message['From'] = ADDRESS + message['To'] = ADDRESS + message['Subject'] = '邮件协议追加和删除验收' + message['Message-ID'] = '<' + appended_id + '>' + message.set_content(body) + message.add_attachment(attachment, maintype='text', subtype='plain', filename='验收附件.txt') + client = LocalIMAP(HOST, 993, ssl_context=context, timeout=30) + try: + assert client.login(ADDRESS, code)[0] == 'OK' + assert client.select('INBOX')[0] == 'OK' + assert client.append('INBOX', None, None, message.as_bytes(policy=email_policy.clone(linesep='\r\n')))[0] == 'OK' + status, values = client.response('APPENDUID') + assert status == 'APPENDUID' and len(values) == 1 and values[0] + validity, uid = map(int, values[0].split()) + assert 0 < validity <= 2**32 - 1 and 0 < uid <= 2**32 - 1 + assert client.noop()[0] == 'OK' + status, contents = client.uid('FETCH', str(uid), '(BODY.PEEK[])') + assert status == 'OK' + raw = next(item[1] for item in contents if isinstance(item, tuple)) + verify_message_content(raw, appended_id, body, attachment) + assert client.uid('STORE', str(uid), '+FLAGS.SILENT', r'(\Deleted)')[0] == 'OK' + assert client.uid('EXPUNGE', str(uid))[0] == 'OK' + status, matches = client.uid('SEARCH', None, 'HEADER', 'Message-ID', '"<' + appended_id + '>"') + assert status == 'OK' and not matches[0].split() + finally: + client.logout() + return {'append_uid': True, 'body_attachment': True, 'uid_expunge': True} + + def verify_restored(root, context, password, message_id, port): client = Client(context, port=port) for _ in range(60): @@ -229,12 +261,18 @@ def main(): parser.add_argument('--proxy', action='store_true') parser.add_argument('--password-stdin', action='store_true') parser.add_argument('--restore-message-id') + parser.add_argument('--sql-stdin', action='store_true') args = parser.parse_args() assert str(Path('/proc/self/ns/net').readlink()) != args.live_net, '禁止在宿主网络执行测试' root = args.root.resolve(strict=True) assert root.parent.name.startswith(('smail-review-runtime-', 'smail-review-restore-')) and root.name == 'instance' password = sys.stdin.readline().strip() if args.password_stdin else secrets.token_urlsafe(9) assert len(password) == 12 + sql_database = None + if args.sql_stdin: + assert args.password_stdin and not args.restore_message_id + from runtime_sql import SQLDatabase + sql_database = SQLDatabase(json.loads(sys.stdin.readline())) if args.restore_message_id: assert args.password_stdin context = ssl.create_default_context(cafile=str(root / 'ssl/runtime.crt')) @@ -258,9 +296,14 @@ def main(): assert Client(port=setup_port).request('/api/v1/setup', {'step': 'database', 'action': 'get'})[0] == 403 assert not (root / 'data/smail.db').exists(), '业务库必须从空数据开始' setup = '/api/v1/setup' - client.success(setup, {'step': 'database', 'action': 'set', 'db_type': 'sqlite', 'db_dsn': ''}) - with sqlite3.connect(f'file:{root / "data/smail.db"}?mode=ro', uri=True) as database: - assert database.execute('SELECT COUNT(*) FROM user').fetchone()[0] == 0 + if sql_database: + client.success(setup, sql_database.setup_payload()) + sql_database.assert_empty_users() + assert not (root / 'data/smail.db').exists() + else: + client.success(setup, {'step': 'database', 'action': 'set', 'db_type': 'sqlite', 'db_dsn': ''}) + with sqlite3.connect(f'file:{root / "data/smail.db"}?mode=ro', uri=True) as database: + assert database.execute('SELECT COUNT(*) FROM user').fetchone()[0] == 0 client.success(setup, {'step': 'password', 'action': 'set', 'account': ACCOUNT, 'password': password}) client.success(setup, {'step': 'domain', 'action': 'set', 'smtp_domain': 'runtime.invalid', 'web_domain': HOST, 'multi_domain': ''}) @@ -320,18 +363,22 @@ def main(): message.add_attachment(attachment, maintype='text', subtype='plain', filename='验收附件.txt') with smtp_connect(context) as smtp: assert not smtp.send_message(message) - with sqlite3.connect(f'file:{root / "data/smail.db"}?mode=ro', uri=True) as database: - rows = database.execute('SELECT id,text,attachments FROM email WHERE msg_id=?', (message_id,)).fetchall() - assert len(rows) == 1 and body in rows[0][1] - attachments = json.loads(rows[0][2]) - assert len(attachments) == 1 and attachments[0]['Filename'] == '验收附件.txt' - assert base64.b64decode(attachments[0]['Content']) == attachment - owners = database.execute('SELECT u.account FROM user_email e JOIN user u ON u.id=e.user_id WHERE e.email_id=?', - (rows[0][0],)).fetchall() - assert owners == [(ACCOUNT,)] + if sql_database: + sql_database.verify_message(message_id, ACCOUNT, body, attachment) + else: + with sqlite3.connect(f'file:{root / "data/smail.db"}?mode=ro', uri=True) as database: + rows = database.execute('SELECT id,text,attachments FROM email WHERE msg_id=?', (message_id,)).fetchall() + assert len(rows) == 1 and body in rows[0][1] + attachments = json.loads(rows[0][2]) + assert len(attachments) == 1 and attachments[0]['Filename'] == '验收附件.txt' + assert base64.b64decode(attachments[0]['Content']) == attachment + owners = database.execute('SELECT u.account FROM user_email e JOIN user u ON u.id=e.user_id WHERE e.email_id=?', + (rows[0][0],)).fetchall() + assert owners == [(ACCOUNT,)] result['smtp'] = {'message_id': message_id, 'stored': True, 'owners': 1, 'attachment': True} result['stage'] = 'protocol_read' result['protocol_read'] = verify_protocol_read(context, codes, message_id, body, attachment) + result['imap_append'] = verify_imap_append(context, codes['imap'], message_id, body, attachment) result['stage'] = 'overload' rounds = [] result['overload'] = rounds diff --git a/test/tools/runtime_review.py b/test/tools/runtime_review.py index a866d21..c160229 100644 --- a/test/tools/runtime_review.py +++ b/test/tools/runtime_review.py @@ -20,7 +20,8 @@ def run(*args): def service_state(): return {name: run('systemctl', 'show', name, '--property=MainPID,NRestarts,ActiveState,InvocationID') - for name in ('smail.service', 'nginx.service', 'mysql.service', 'Scert.service')} + for name in ('smail.service', 'nginx.service', 'mysql.service', 'Scert.service', + 'fdm-studio.service', 'postgresql@18-main.service')} def database_integrity(root): @@ -78,7 +79,10 @@ 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('--database-type', choices=('sqlite', 'mysql', 'postgres'), default='sqlite') args = parser.parse_args() + if args.restore and args.database_type != 'sqlite': + parser.error('SQL 数据库备份恢复需要独立验收，不能套用 SQLite 冷备流程') assert os.geteuid() == 0 and run('hostname') == 'ikun2' os.umask(0o027) live = Path('/opt/smail') @@ -112,6 +116,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.database_type != 'sqlite': + shutil.copy2(Path(__file__).with_name('runtime_sql.py'), target / 'runtime_sql.py') + (target / 'runtime_sql.py').chmod(0o640) + shutil.chown(target / 'runtime_sql.py', user='root', group='smail') if args.nginx_template: shutil.copy2(Path(__file__).with_name('runtime_proxy.py'), target / 'runtime_proxy.py') (target / 'runtime_proxy.py').chmod(0o640) @@ -136,6 +144,13 @@ def main(): 'topology': 'nginx-8001-8002' if args.nginx_template else 'direct-80-443'} service_env = [] probe_args = [] + sql_fixture = None + bridge = None + result['database_type'] = args.database_type + if args.database_type != 'sqlite': + from runtime_sql import SQLFixture + sql_fixture = SQLFixture(args.database_type) + result['sql_database'] = sql_fixture.settings['name'] if args.nginx_template: from runtime_proxy import prepare_nginx prepare_nginx(target, root, args.nginx_template) @@ -146,6 +161,9 @@ def main(): started = time.strftime('%Y-%m-%dT%H:%M:%S') password = secrets.token_urlsafe(9) try: + if sql_fixture: + sql_fixture.create() + probe_args.append('--sql-stdin') subprocess.run([ 'systemd-run', '--unit=' + unit, '--property=Type=exec', '--property=User=smail', '--property=Group=smail', @@ -168,6 +186,9 @@ def main(): assert str(Path('/proc/' + pid + '/ns/net').readlink()) != live_net assert Path('/proc/' + pid + '/root/opt/smail/smail').stat().st_ino == (root / 'smail').stat().st_ino assert (root / 'smail').stat().st_ino != (live / 'smail').stat().st_ino + if sql_fixture: + from runtime_sql_bridge import DatabaseBridge + bridge = DatabaseBridge(int(pid), sql_fixture.port) if args.nginx_template: subprocess.run([ 'systemd-run', '--unit=' + proxy_unit, '--property=Type=exec', @@ -198,7 +219,8 @@ def main(): '--property=KillMode=control-group', '--property=TimeoutStopSec=10s', 'python3', str(target / 'runtime_probe.py'), str(root), '--live-net', live_net, '--password-stdin', *probe_args, - ], input=password + '\n', text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=270) + ], input=password + '\n' + (json.dumps(sql_fixture.settings) + '\n' if sql_fixture else ''), + text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=270) (target / 'probe.log').write_text(probe.stdout, encoding='utf-8') result['probe_code'] = probe.returncode if (target / 'probe-result.json').exists(): @@ -210,15 +232,30 @@ def main(): assert database.execute('SELECT COUNT(*) FROM email WHERE msg_id=?', (message_id,)).fetchone()[0] == 0 result['live_message_absent'] = True finally: - collect_and_stop(result, unit, probe_unit, proxy_unit if args.nginx_template else None) - result['after'] = service_state() - with (live / 'smail').open('rb') as stream: - result['live_unchanged'] = (result['after'] == before - and hashlib.file_digest(stream, 'sha256').hexdigest() == live_sha) try: - result['integrity'] = database_integrity(root) - except (AssertionError, sqlite3.Error) as error: - result['integrity'] = {'error': type(error).__name__} + collect_and_stop(result, unit, probe_unit, proxy_unit if args.nginx_template else None) + result['after'] = service_state() + with (live / 'smail').open('rb') as stream: + 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) + except (AssertionError, sqlite3.Error, RuntimeError, OSError, ValueError, subprocess.SubprocessError) as error: + result['integrity'] = {'error': type(error).__name__} + finally: + result.setdefault('cleanup_errors', []) + if bridge: + try: + bridge.close() + result['database_bridge_stopped'] = True + except Exception as error: + result['cleanup_errors'].append({'action': 'database_bridge', 'error': type(error).__name__}) + if sql_fixture: + try: + sql_fixture.lock() + result['test_database_account_locked'] = True + except Exception as error: + result['cleanup_errors'].append({'action': 'database_account', 'error': type(error).__name__}) (target / 'service.log').write_text(run('journalctl', '-u', unit, '--no-pager'), encoding='utf-8') kernel = run('journalctl', '-k', '--since', started, '--no-pager') result['oom_events'] = sum('oom-kill:constraint=' in line for line in kernel.splitlines()) @@ -227,7 +264,10 @@ def main(): assert result['live_unchanged'] and result['oom_events'] == 0 assert not result['cleanup_errors'], '测试单元未全部停止，需要核查保留的结果' assert not result['resource_errors'], '资源证据不完整，需要核查保留的结果' - assert result['integrity'] == {'smail.db': [('ok',)]} + if sql_fixture: + assert result['integrity'].get('sqlite_files') == 0 and result['integrity']['config_records'] > 0 + else: + assert result['integrity'] == {'smail.db': [('ok',)]} if args.restore: from runtime_restore import verify_backup_restore restored = verify_backup_restore(root, result, password, args.nginx_template is not None) diff --git a/test/tools/runtime_sql.py b/test/tools/runtime_sql.py new file mode 100644 index 0000000..2a87c8a --- /dev/null +++ b/test/tools/runtime_sql.py @@ -0,0 +1,128 @@ +"""隔离实机验收的 SQL 数据库适配；不删除数据库，不记录连接凭据。""" + +import base64 +import json +import os +import re +import secrets +import subprocess + + +def command(args, sql, env=None): + result = subprocess.run(args, input=sql, text=True, capture_output=True, + timeout=30, env=env, encoding='utf-8') + if result.returncode: + # 驱动错误可能包含连接参数，不能直接传播到验收日志。 + raise RuntimeError('隔离测试数据库操作失败，请在受保护环境核查') + return result.stdout.strip() + + +class SQLDatabase: + def __init__(self, settings): + assert set(settings) == {'kind', 'name', 'user', 'password'} + assert settings['kind'] in ('mysql', 'postgres') + assert re.fullmatch(r'smail_runtime_test_[0-9a-f]{12}', settings['name']) + assert re.fullmatch(r'smail_rt_[0-9a-f]{12}', settings['user']) + assert re.fullmatch(r'[A-Za-z0-9_-]{32,64}', settings['password']) + self.settings = settings.copy() + self.kind = settings['kind'] + self.port = 3306 if self.kind == 'mysql' else 5432 + + def setup_payload(self): + return {'step': 'database', 'action': 'set', 'db_type': self.kind, + 'db_mode': 'fields', 'db_host': '127.0.0.1', 'db_port': str(self.port), + 'db_name': self.settings['name'], 'db_user': self.settings['user'], + 'db_password': self.settings['password']} + + def query(self, sql): + env = os.environ.copy() + if self.kind == 'mysql': + env['MYSQL_PWD'] = self.settings['password'] + args = ['mysql', '--protocol=tcp', '--host=127.0.0.1', '--port=3306', + '--user=' + self.settings['user'], '--database=' + self.settings['name'], + '--default-character-set=utf8mb4', '--batch', '--raw', '--skip-column-names'] + else: + env['PGPASSWORD'] = self.settings['password'] + env['PGCLIENTENCODING'] = 'UTF8' + args = ['psql', '-X', '-v', 'ON_ERROR_STOP=1', '-At', '--host=127.0.0.1', + '--port=5432', '--username=' + self.settings['user'], + '--dbname=' + self.settings['name']] + return command(args, sql, env) + + def assert_empty_users(self): + table = '`user`' if self.kind == 'mysql' else '"user"' + assert self.query('SELECT COUNT(*) FROM ' + table) == '0' + + def verify_message(self, message_id, account, body, attachment): + assert re.fullmatch(r'[a-zA-Z0-9@._-]+', message_id) + constructor = 'JSON_OBJECT' if self.kind == 'mysql' else 'json_build_object' + content = self.query("SELECT " + constructor + "('id',id,'text',text,'attachments',attachments)" + " FROM email WHERE msg_id='" + message_id + "'") + rows = content.splitlines() + assert len(rows) == 1 + record = json.loads(rows[0]) + assert body in record['text'] + attachments = record['attachments'] + if isinstance(attachments, str): + attachments = json.loads(attachments) + assert len(attachments) == 1 and attachments[0]['Filename'] == '验收附件.txt' + assert base64.b64decode(attachments[0]['Content']) == attachment + table = '`user`' if self.kind == 'mysql' else '"user"' + assert isinstance(record['id'], int) and record['id'] > 0 + owners = self.query('SELECT u.account FROM user_email e JOIN ' + table + + ' u ON u.id=e.user_id WHERE e.email_id=' + str(record['id'])) + assert owners.splitlines() == [account] + + def integrity(self, root): + # 配置历史和业务表必须位于所选 SQL 库，运行目录不能悄悄创建 SQLite。 + assert not list((root / 'data').rglob('*.db')) + assert not list((root / 'data').rglob('*.sqlite*')) + count = int(self.query('SELECT COUNT(*) FROM config_records')) + assert count > 0 + return {'type': self.kind, 'database': self.settings['name'], + 'config_records': count, 'sqlite_files': 0} + + +class SQLFixture(SQLDatabase): + def __init__(self, kind): + token = secrets.token_hex(6) + super().__init__({'kind': kind, 'name': 'smail_runtime_test_' + token, + 'user': 'smail_rt_' + token, 'password': secrets.token_urlsafe(30)}) + self.account_created = False + + def admin(self, sql, database=None): + if self.kind == 'mysql': + args = ['mysql', '--protocol=socket', '--batch', '--skip-column-names'] + else: + args = ['runuser', '-u', 'postgres', '--', 'psql', '-X', '-v', 'ON_ERROR_STOP=1', '-At'] + if database: + assert database == self.settings['name'] + args += ['--dbname=' + database] + return command(args, sql) + + def create(self): + name, user, password = (self.settings[key] for key in ('name', 'user', 'password')) + if self.kind == 'mysql': + self.admin("CREATE USER '" + user + "'@'localhost' IDENTIFIED BY '" + password + + "' WITH MAX_USER_CONNECTIONS 24;") + self.account_created = True + self.admin('CREATE DATABASE `' + name + '` CHARACTER SET utf8mb4;') + # 转义授权模式中的下划线，只授权本轮唯一数据库。 + self.admin('GRANT ALL ON `' + name.replace('_', r'\_') + "`.* TO '" + user + "'@'localhost';") + count = self.admin("SELECT COUNT(*) FROM information_schema.tables WHERE table_schema='" + name + "';") + else: + self.admin('CREATE ROLE ' + user + " LOGIN CONNECTION LIMIT 24 PASSWORD '" + password + "';") + self.account_created = True + self.admin('CREATE DATABASE ' + name + ' OWNER ' + user + ';') + self.admin('GRANT ALL ON SCHEMA public TO ' + user + ';', name) + count = self.admin("SELECT COUNT(*) FROM information_schema.tables WHERE table_schema='public';", name) + assert count == '0', '首次安装必须使用空数据库' + + def lock(self): + if not self.account_created: + return + user = self.settings['user'] + if self.kind == 'mysql': + self.admin("ALTER USER '" + user + "'@'localhost' ACCOUNT LOCK;") + else: + self.admin('ALTER ROLE ' + user + ' NOLOGIN;') diff --git a/test/tools/runtime_sql_bridge.py b/test/tools/runtime_sql_bridge.py new file mode 100644 index 0000000..5a62135 --- /dev/null +++ b/test/tools/runtime_sql_bridge.py @@ -0,0 +1,64 @@ +"""只向隔离网络提供单个本机数据库端口，不向宿主网络新增监听端口。""" + +from contextlib import ExitStack +import os +from pathlib import Path +import select +import socket +import socketserver +import threading + + +class DatabaseBridge: + def __init__(self, pid, port): + assert isinstance(pid, int) and pid > 1 + assert port in (3306, 5432) + namespace = Path('/proc') / str(pid) / 'ns/net' + assert namespace.readlink() != Path('/proc/self/ns/net').readlink() + limit = threading.BoundedSemaphore(24) + + class Handler(socketserver.BaseRequestHandler): + def handle(self): + if not limit.acquire(blocking=False): + return + try: + with socket.create_connection(('127.0.0.1', port), timeout=15) as upstream: + streams = (self.request, upstream) + for stream in streams: + stream.settimeout(30) + while True: + ready, _, _ = select.select(streams, [], [], 30) + if not ready: + return + for source in ready: + data = source.recv(65536) + if not data: + return + target = upstream if source is self.request else self.request + target.sendall(data) + except (OSError, TimeoutError): + return + finally: + limit.release() + + class Server(socketserver.ThreadingTCPServer): + daemon_threads = True + allow_reuse_address = False + + with ExitStack() as stack: + host = stack.enter_context(open('/proc/self/ns/net', 'rb')) + isolated = stack.enter_context(open(namespace, 'rb')) + try: + os.setns(isolated.fileno(), 0) + self.server = Server(('127.0.0.1', port), Handler) + finally: + os.setns(host.fileno(), 0) + # 监听套接字属于隔离网络，后续工作线程在恢复后的宿主网络连接数据库。 + self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) + self.thread.start() + + def close(self): + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=5) + assert not self.thread.is_alive(), '数据库验收桥接未停止'