From c52ea24663cf1adc0cefe5899c9f0af80d3339f4 Mon Sep 17 00:00:00 2001 From: smail Date: Thu, 17 Sep 2026 12:56:27 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E8=BF=9C=E7=A8=8B=E9=9A=94?= =?UTF-8?q?=E7=A6=BB=E9=A6=96=E6=AC=A1=E5=AE=89=E8=A3=85=E4=B8=8E=E8=AE=A4?= =?UTF-8?q?=E8=AF=81=E8=BF=87=E8=BD=BD=E9=AA=8C=E6=94=B6=E5=B7=A5=E5=85=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- test/tools/runtime_probe.py | 256 +++++++++++++++++++++++++++++++++++ test/tools/runtime_review.py | 127 +++++++++++++++++ 2 files changed, 383 insertions(+) create mode 100644 test/tools/runtime_probe.py create mode 100644 test/tools/runtime_review.py diff --git a/test/tools/runtime_probe.py b/test/tools/runtime_probe.py new file mode 100644 index 0000000..e13eb6b --- /dev/null +++ b/test/tools/runtime_probe.py @@ -0,0 +1,256 @@ +"""仅在隔离网络中通过正式接口验收首次安装和认证，不持久化测试密码。""" + +import argparse +import base64 +from concurrent.futures import ThreadPoolExecutor +from email.message import EmailMessage +import http.client +from http.cookies import SimpleCookie +import imaplib +import json +from pathlib import Path +import poplib +import secrets +import smtplib +import socket +import sqlite3 +import ssl +import threading +import time + + +HOST = 'mail.runtime.invalid' +ACCOUNT = 'reviewadmin' +ADDRESS = ACCOUNT + '@runtime.invalid' + + +class LocalHTTPS(http.client.HTTPSConnection): + def connect(self): + raw = socket.create_connection(('127.0.0.1', self.port), self.timeout) + self.sock = self._context.wrap_socket(raw, server_hostname=self.host) + + +class Client: + def __init__(self, context=None): + self.context = context + self.cookies = SimpleCookie() + + def request(self, path, payload=None, method='POST'): + connection = (LocalHTTPS(HOST, 443, timeout=30, context=self.context) + if self.context else http.client.HTTPConnection('127.0.0.1', 80, timeout=30)) + origin = 'https://' + HOST if self.context else 'http://127.0.0.1' + headers = {'Origin': origin, 'Content-Type': 'application/json'} + if self.cookies: + headers['Cookie'] = '; '.join(item.key + '=' + item.coded_value for item in self.cookies.values()) + if 'smail_csrf' in self.cookies: + headers['X-CSRF-Token'] = self.cookies['smail_csrf'].value + try: + connection.request(method, path, json.dumps(payload) if payload is not None else None, headers) + response = connection.getresponse() + for name, value in response.getheaders(): + if name.lower() == 'set-cookie': + self.cookies.load(value) + content = response.read() + data = json.loads(content) if 'application/json' in response.getheader('Content-Type', '') else None + return response.status, data, response.getheader('Retry-After') + finally: + connection.close() + + def success(self, path, payload=None, method='POST'): + status, data, _ = self.request(path, payload, method) + assert status == 200 and data is not None and data.get('errorNo') == 0, ( + path, status, data.get('errorNo') if data else None, data.get('errorMsg') if data else None) + return data['data'] + + +def smtp_connect(context): + client = smtplib.SMTP('127.0.0.1', 25, timeout=30) + client.ehlo('runtime.invalid') + assert client.has_extn('starttls') and not client.has_extn('auth') + code, _ = client.docmd('AUTH', 'PLAIN') + assert code in (502, 523, 538), '明文认证未被拒绝' + client._host = HOST + client.starttls(context=context) + client.ehlo('runtime.invalid') + return client + + +class LocalIMAP(imaplib.IMAP4_SSL): + def _create_socket(self, timeout): + raw = socket.create_connection(('127.0.0.1', self.port), timeout) + return self.ssl_context.wrap_socket(raw, server_hostname=self.host) + + +class LocalPOP(poplib.POP3_SSL): + def _create_socket(self, timeout): + raw = socket.create_connection(('127.0.0.1', self.port), timeout) + return self.context.wrap_socket(raw, server_hostname=self.host) + + +def protocol_operation(protocol, context, code): + if protocol == 'smtp': + client = smtp_connect(context) + + def invoke(): + try: + status, _ = client.auth('PLAIN', lambda challenge=None: '\0' + ADDRESS + '\0' + code) + assert status == 235 + return 'success' + except smtplib.SMTPAuthenticationError as error: + assert error.smtp_code == 454, ('smtp', error.smtp_code) + assert b'Temporary authentication failure' in error.smtp_error + return 'busy' + finally: + client.close() + + elif protocol == 'imap': + client = LocalIMAP(HOST, 993, ssl_context=context, timeout=30) + + def invoke(): + try: + status, _ = client.login(ADDRESS, code) + assert status == 'OK' + return 'success' + except imaplib.IMAP4.error as error: + assert 'UNAVAILABLE' in str(error), 'IMAP 未返回临时不可用' + return 'busy' + finally: + client.shutdown() + + else: + client = LocalPOP(HOST, 995, context=context, timeout=30) + client.user(ADDRESS) + + def invoke(): + try: + assert client.pass_(code).startswith(b'+OK') + return 'success' + except poplib.error_proto as error: + assert '[SYS/TEMP]' in str(error), 'POP3 未返回临时不可用' + return 'busy' + finally: + client.close() + + return invoke + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument('root', type=Path) + parser.add_argument('--live-net', required=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-') and root.name == 'instance' + result = {'stage': 'setup', 'passed': False} + started = time.monotonic() + try: + client = Client() + for _ in range(60): + try: + assert client.request('/', method='GET')[0] == 200 + break + except (OSError, AssertionError): + time.sleep(1) + else: + raise AssertionError('初始化服务未就绪') + assert 'smail_setup_token' in client.cookies, '初始化令牌未下发' + assert client.cookies['smail_setup_token']['httponly'] + assert Client().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 + password = secrets.token_urlsafe(9) + 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': ''}) + client.success(setup, {'step': 'ssl', 'action': 'set', 'ssl_type': '1', + 'key_path': '/opt/smail/ssl/runtime.key', 'crt_path': '/opt/smail/ssl/runtime.crt'}) + 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) + for _ in range(60): + try: + assert client.request('/api/ping', method='GET')[0] == 200 + break + except (OSError, AssertionError): + time.sleep(1) + else: + raise AssertionError('正式 HTTPS 服务未就绪') + login = {'account': ACCOUNT, 'password': password} + identity = client.success('/api/v1/login', login) + assert identity.get('account') == ACCOUNT and not identity.get('pending') + assert 'smail_csrf' in client.cookies + assert client.success('/api/v1/user/info').get('account') == ACCOUNT + result['setup'] = {'empty_database': True, 'token_required': True, 'login': True, 'session': True} + result['stage'] = 'protocols' + codes = {} + for protocol in ('smtp', 'imap', 'pop3'): + base = '/api/v1/security/credentials/protocols/' + client.success(base + 'enabled', {'protocol': protocol, 'enabled': True}, 'PUT') + codes[protocol] = client.success(base + 'generate', {'protocol': protocol})['code'] + assert protocol_operation(protocol, context, codes[protocol])() == 'success' + result['protocol_login'] = list(codes) + result['stage'] = 'smtp' + message = EmailMessage() + message['From'] = 'probe@runtime.invalid' + message['To'] = ADDRESS + message['Subject'] = '隔离首次安装中文收信验收' + message_id = 'runtime-' + root.parent.name + '@runtime.invalid' + message['Message-ID'] = '<' + message_id + '>' + body = '内部25端口真实收信，检查中文正文和附件。' + attachment = '隔离验收附件内容'.encode('utf-8') + 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) + 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'] = 'overload' + rounds = [] + for protocol in codes: + operations = [protocol_operation(protocol, context, codes[protocol]) for _ in range(4)] + barrier = threading.Barrier(8, timeout=20) + + def http_operation(): + status, data, retry = Client(context).request('/api/v1/login', login) + if status == 503: + assert retry == '1' and data.get('data', {}).get('code') == 'AUTH_BUSY' + return 'busy' + assert status == 200 and data.get('errorNo') == 0 and data['data'].get('account') == ACCOUNT + return 'success' + + def execute(operation): + barrier.wait() + began = time.monotonic() + return {'outcome': operation(), 'seconds': round(time.monotonic() - began, 3)} + + with ThreadPoolExecutor(max_workers=8) as pool: + responses = list(pool.map(execute, operations + [http_operation] * 4)) + assert any(item['outcome'] == 'busy' for item in responses[:4]), protocol + ' 未观察到繁忙响应' + assert any(item['outcome'] == 'busy' for item in responses[4:]), 'HTTP 未观察到繁忙响应' + assert any(item['outcome'] == 'success' for item in responses), '所有请求均失败' + assert protocol_operation(protocol, context, codes[protocol])() == 'success' + assert client.success('/api/v1/login', login).get('account') == ACCOUNT + rounds.append({'protocol': protocol, 'responses': responses, 'recovered': True}) + result['overload'] = rounds + result['stage'] = 'complete' + result['passed'] = True + finally: + result['seconds'] = round(time.monotonic() - started, 2) + (root.parent / 'probe-result.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 new file mode 100644 index 0000000..482e705 --- /dev/null +++ b/test/tools/runtime_review.py @@ -0,0 +1,127 @@ +"""在 47 服务器创建空数据副本并进行受限验收，不停止线上服务或删除文件。""" + +import argparse +import hashlib +import json +import os +from pathlib import Path +import shutil +import sqlite3 +import subprocess +import time + + +def run(*args): + return subprocess.check_output(args, text=True).strip() + + +def service_state(): + return {name: run('systemctl', 'show', name, '--property=MainPID,NRestarts,ActiveState') + for name in ('smail.service', 'mysql.service', 'Scert.service')} + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument('--main-sha', required=True) + args = parser.parse_args() + assert os.geteuid() == 0 and run('hostname') == 'ikun2' + os.umask(0o027) + live = Path('/opt/smail') + with (live / 'smail').open('rb') as stream: + assert hashlib.file_digest(stream, 'sha256').hexdigest() == args.main_sha + memory = dict(line.split(':', 1) for line in Path('/proc/meminfo').read_text().splitlines()) + assert int(memory['MemAvailable'].split()[0]) >= 272 * 1024, '可用内存不足，暂停隔离验收' + stamp = time.strftime('%Y%m%d-%H%M%S') + target = Path('/var/tmp') / ('smail-review-runtime-' + stamp) + target.mkdir(mode=0o750) + shutil.chown(target, user='smail', group='smail') + root = target / 'instance' + control = target / 'control' + for directory in (root, control, root / 'ssl'): + directory.mkdir(mode=0o750) + shutil.chown(directory, user='smail', group='smail') + shutil.copy2(live / 'smail', root / 'smail') + (root / 'smail').chmod(0o755) + shutil.copy2(Path(__file__).with_name('runtime_probe.py'), target / 'runtime_probe.py') + subprocess.run(['openssl', 'req', '-x509', '-newkey', 'rsa:2048', '-nodes', '-days', '2', + '-keyout', str(root / 'ssl/runtime.key'), '-out', str(root / 'ssl/runtime.crt'), + '-subj', '/CN=mail.runtime.invalid', '-addext', + 'subjectAltName=DNS:runtime.invalid,DNS:*.runtime.invalid'], + check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + for name in ('runtime.key', 'runtime.crt'): + path = root / 'ssl' / name + path.chmod(0o600) + shutil.chown(path, user='smail', group='smail') + assert not (root / 'data').exists() and not (root / 'config.json').exists() + before = service_state() + unit = 'smail-runtime-review-' + stamp + probe_unit = 'smail-runtime-probe-' + stamp + result = {'target': str(target), 'unit': unit, 'main_sha256': args.main_sha, 'before': before} + started = time.strftime('%Y-%m-%dT%H:%M:%S') + try: + subprocess.run([ + 'systemd-run', '--unit=' + unit, '--property=Type=exec', + '--property=User=smail', '--property=Group=smail', + '--property=PrivateNetwork=yes', '--property=PrivateTmp=yes', + '--property=ProtectSystem=strict', '--property=ProtectHome=yes', + '--property=NoNewPrivileges=yes', '--property=CapabilityBoundingSet=CAP_NET_BIND_SERVICE', + '--property=AmbientCapabilities=CAP_NET_BIND_SERVICE', + '--property=MemoryMax=128M', '--property=MemorySwapMax=0', '--property=CPUQuota=25%', + '--property=TasksMax=128', '--property=RuntimeMaxSec=300s', '--property=OOMPolicy=stop', + '--property=KillMode=control-group', '--property=TimeoutStopSec=10s', + '--property=BindPaths=' + str(root) + ':/opt/smail ' + str(control) + ':/run/smail', + '--property=ReadWritePaths=/opt/smail /run/smail', + '--property=WorkingDirectory=/opt/smail', '--setenv=SMAIL_ROOT=/opt/smail', + '--setenv=GOMEMLIMIT=64MiB', '--setenv=GOMAXPROCS=1', '/opt/smail/smail', + ], check=True) + pid = run('systemctl', 'show', unit, '--property=MainPID', '--value') + assert pid != '0' + live_net = str(Path('/proc/self/ns/net').readlink()) + 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 + probe = subprocess.run([ + 'systemd-run', '--wait', '--pipe', '--unit=' + probe_unit, + '--property=User=smail', '--property=Group=smail', + '--property=NetworkNamespacePath=/proc/' + pid + '/ns/net', + '--property=PrivateTmp=no', '--property=ProtectSystem=strict', '--property=ProtectHome=yes', + '--property=NoNewPrivileges=yes', '--property=InaccessiblePaths=/opt/smail /run/smail', + '--property=ReadWritePaths=' + str(target), '--property=MemoryMax=48M', + '--property=MemorySwapMax=0', '--property=CPUQuota=25%', '--property=TasksMax=32', + '--property=RuntimeMaxSec=240s', '--property=OOMPolicy=stop', + '--property=KillMode=control-group', '--property=TimeoutStopSec=10s', + 'python3', str(target / 'runtime_probe.py'), str(root), '--live-net', live_net, + ], 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 + assert probe.returncode == 0, '隔离验收失败，请检查保留的 probe.log' + result['probe'] = json.loads((target / 'probe-result.json').read_text()) + assert result['probe']['passed'] + with 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 + finally: + result['resources'] = run('systemctl', 'show', unit, + '--property=MemoryPeak,MemoryCurrent,Result,ExecMainStatus,NRestarts') + subprocess.run(['systemctl', 'stop', unit], check=True) + subprocess.run(['systemctl', 'stop', probe_unit], check=False, stderr=subprocess.DEVNULL) + result['after'] = service_state() + result['live_unchanged'] = result['after'] == before + result['integrity'] = {} + for name in ('smail.db', 'smail-state.db'): + path = root / 'data' / name + if path.exists(): + with sqlite3.connect(f'file:{path}?mode=ro', uri=True) as database: + result['integrity'][name] = database.execute('PRAGMA integrity_check').fetchall() + (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()) + (target / 'result.json').write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding='utf-8') + print(json.dumps(result, ensure_ascii=False, indent=2)) + assert result['live_unchanged'] and result['oom_events'] == 0 + assert result['integrity'] == {'smail.db': [['ok']], 'smail-state.db': [['ok']]} + + +if __name__ == '__main__': + main()