hz4th_coder
6c1225f588
- 登录: QQ/163/126/189/搜狐/Gmail/Outlook/Yahoo/iCloud/企业邮/自定义, 自动识别+测试连接 - 收信: IMAP 多文件夹同步/未读/星标/搜索(本地+服务器) - 读信: HTML渲染/CID内嵌图/原文查看/附件下载 - 写信: 收件人/抄送/密送/附件/富文本/草稿/回复转发 - 存储: SQLite + 系统安全存储(DPAPI/Keychain/Keystore) - 测试: 单元测试 + Mock IMAP/SMTP 服务器集成测试(23项全过) - 可迁移: Windows/Android/iOS/macOS/Linux 一套代码
200 lines
8.9 KiB
Python
200 lines
8.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Mock IMAP 服务器 —— 用于协议栈集成测试
|
|
支持: LOGIN/AUTHENTICATE PLAIN, CAPABILITY, LIST, SELECT, UID SEARCH,
|
|
UID FETCH(含多段字面量), UID STORE, LOGOUT
|
|
用法: python3 mock_imap_server.py <port> [--auth-user USER --auth-pass PASS]
|
|
"""
|
|
import socket
|
|
import sys
|
|
import threading
|
|
import argparse
|
|
import base64
|
|
|
|
MAILBOXES = [
|
|
('INBOX', []),
|
|
('Sent', []),
|
|
('[Gmail]', [r'\HasChildren', r'\Noselect']),
|
|
]
|
|
|
|
HEADER_1 = ('Date: Mon, 17 Aug 2026 08:23:45 +0800\r\n'
|
|
'From: "Zhang San" <zhangsan@example.com>\r\n'
|
|
'To: lisi@example.com\r\n'
|
|
'Subject: =?UTF-8?B?5L2g5aW955qE5L2T5YaF?=\r\n'
|
|
'Message-ID: <msg-1@example.com>\r\n'
|
|
'MIME-Version: 1.0\r\n'
|
|
'Content-Type: text/plain; charset="utf-8"\r\n').encode('utf-8')
|
|
|
|
BODY_1 = ('Content-Type: text/plain; charset="utf-8"\r\n\r\n'
|
|
'你好，这是一封测试邮件！\r\n').encode('utf-8')
|
|
|
|
HEADER_2 = ('Date: Tue, 18 Aug 2026 09:00:00 +0800\r\n'
|
|
'From: "Wang Wu" <wangwu@example.com>\r\n'
|
|
'To: zhangsan@example.com\r\n'
|
|
'Subject: Re: 测试邮件\r\n'
|
|
'Message-ID: <msg-2@example.com>\r\n'
|
|
'MIME-Version: 1.0\r\n').encode('utf-8')
|
|
|
|
BODY_2 = ('Content-Type: text/plain; charset="utf-8"\r\n\r\n'
|
|
'收到，谢谢！\r\n').encode('utf-8')
|
|
|
|
MESSAGES = {
|
|
1: (HEADER_1, BODY_1, [r'\Seen'], 1234, '17-Aug-2026 08:23:45 +0800'),
|
|
2: (HEADER_2, BODY_2, [], 567, '18-Aug-2026 09:00:00 +0800'),
|
|
3: (HEADER_1, BODY_1, [r'\Flagged'], 1234, '18-Aug-2026 10:00:00 +0800'),
|
|
}
|
|
|
|
|
|
def parse_uid_set(mset):
|
|
uids = []
|
|
for part in mset.split(','):
|
|
if ':' in part:
|
|
a, b = part.split(':')
|
|
uids.extend(range(int(a), int(b) + 1))
|
|
else:
|
|
uids.append(int(part))
|
|
return uids
|
|
|
|
|
|
def handle(conn, auth_user, auth_pass):
|
|
conn.sendall(b'* OK [CAPABILITY IMAP4rev1 AUTH=PLAIN] Mock IMAP ready\r\n')
|
|
buf = b''
|
|
try:
|
|
while True:
|
|
data = conn.recv(65536)
|
|
if not data:
|
|
break
|
|
buf += data
|
|
while b'\r\n' in buf:
|
|
line, buf = buf.split(b'\r\n', 1)
|
|
text = line.decode('utf-8', 'replace').strip()
|
|
if not text:
|
|
continue
|
|
parts = text.split(' ')
|
|
tag = parts[0]
|
|
# UID 前缀命令
|
|
if len(parts) > 1 and parts[1].upper() == 'UID':
|
|
sub = parts[2].upper() if len(parts) > 2 else ''
|
|
cmd = 'UID_' + sub
|
|
rest = ' '.join(parts[3:]) if len(parts) > 3 else ''
|
|
else:
|
|
cmd = parts[1].upper() if len(parts) > 1 else ''
|
|
rest = ' '.join(parts[2:]) if len(parts) > 2 else ''
|
|
|
|
if cmd == 'CAPABILITY':
|
|
conn.sendall(b'* CAPABILITY IMAP4rev1 AUTH=PLAIN\r\n')
|
|
conn.sendall(f'{tag} OK CAPABILITY completed\r\n'.encode())
|
|
elif cmd == 'AUTHENTICATE':
|
|
# AUTHENTICATE PLAIN <b64token>（一步式）
|
|
token = base64.b64decode(parts[3]).decode('utf-8', 'replace')
|
|
_, user, pw = token.split('\0')
|
|
if user == auth_user and pw == auth_pass:
|
|
conn.sendall(f'{tag} OK AUTHENTICATE completed\r\n'.encode())
|
|
else:
|
|
conn.sendall(f'{tag} NO [AUTHENTICATIONFAILED] Invalid credentials\r\n'.encode())
|
|
elif cmd == 'LOGIN':
|
|
user = parts[2].strip('"')
|
|
pw = parts[3].strip('"')
|
|
if user == auth_user and pw == auth_pass:
|
|
conn.sendall(f'{tag} OK LOGIN completed\r\n'.encode())
|
|
else:
|
|
conn.sendall(f'{tag} NO [AUTHENTICATIONFAILED] Invalid credentials\r\n'.encode())
|
|
elif cmd == 'LIST' or cmd == 'LSUB':
|
|
for name, attrs in MAILBOXES:
|
|
attr_str = ' '.join(attrs) if attrs else r'\HasNoChildren'
|
|
conn.sendall(f'* {cmd} ({attr_str}) "/" "{name}"\r\n'.encode())
|
|
conn.sendall(f'{tag} OK {cmd} completed\r\n'.encode())
|
|
elif cmd == 'SELECT' or cmd == 'EXAMINE':
|
|
conn.sendall(b'* 3 EXISTS\r\n* 0 RECENT\r\n')
|
|
conn.sendall(b'* OK [UIDVALIDITY 123456] UIDs valid\r\n')
|
|
conn.sendall(b'* OK [UIDNEXT 10] Predicted next UID\r\n')
|
|
conn.sendall(f'{tag} OK [READ-WRITE] SELECT completed\r\n'.encode())
|
|
elif cmd == 'SEARCH' or cmd == 'UID_SEARCH':
|
|
if 'ALL' in text.upper():
|
|
conn.sendall(b'* SEARCH 1 2 3\r\n')
|
|
else:
|
|
conn.sendall(b'* SEARCH 2\r\n')
|
|
conn.sendall(f'{tag} OK SEARCH completed\r\n'.encode())
|
|
elif cmd == 'UID_FETCH':
|
|
mset = rest.split(' ')[0]
|
|
uids = parse_uid_set(mset)
|
|
items = rest[rest.find('(') + 1:rest.rfind(')')]
|
|
has_fields = 'HEADER.FIELDS' in items
|
|
has_header = 'BODY.PEEK[HEADER]' in items or 'BODY[HEADER]' in items
|
|
has_text = 'BODY.PEEK[TEXT]' in items or 'BODY[TEXT]' in items
|
|
has_size = 'RFC822.SIZE' in items
|
|
has_flags = 'FLAGS' in items
|
|
has_internal = 'INTERNALDATE' in items
|
|
for uid in uids:
|
|
if uid not in MESSAGES:
|
|
continue
|
|
h, b, flags, size, idate = MESSAGES[uid]
|
|
resp = f'* {uid} FETCH (UID {uid}'
|
|
if has_flags:
|
|
resp += ' FLAGS (' + ' '.join(flags) + ')'
|
|
if has_size:
|
|
resp += f' RFC822.SIZE {size}'
|
|
if has_internal:
|
|
resp += f' INTERNALDATE "{idate}"'
|
|
if has_fields:
|
|
fields = b'From: "Zhang San" <zhangsan@example.com>\r\n'
|
|
fields += b'To: lisi@example.com\r\n'
|
|
fields += b'Subject: =?UTF-8?B?5L2g5aW955qE5L2T5YaF?=\r\n'
|
|
fields += b'Message-ID: <msg-1@example.com>\r\n'
|
|
conn.sendall(
|
|
f'{resp} BODY[HEADER.FIELDS (DATE FROM TO CC SUBJECT MESSAGE-ID IN-REPLY-TO REFERENCES)] {{{len(fields)}}}\r\n'.encode())
|
|
conn.sendall(fields)
|
|
conn.sendall(b')\r\n')
|
|
continue
|
|
if has_header and has_text:
|
|
conn.sendall(f'{resp} BODY[HEADER] {{{len(h)}}}\r\n'.encode())
|
|
conn.sendall(h)
|
|
conn.sendall(f' BODY[TEXT] {{{len(b)}}}\r\n'.encode())
|
|
conn.sendall(b)
|
|
conn.sendall(b')\r\n')
|
|
continue
|
|
if has_header:
|
|
conn.sendall(f'{resp} BODY[HEADER] {{{len(h)}}}\r\n'.encode())
|
|
conn.sendall(h)
|
|
conn.sendall(b')\r\n')
|
|
continue
|
|
conn.sendall(resp.encode() + b')\r\n')
|
|
conn.sendall(f'{tag} OK FETCH completed\r\n'.encode())
|
|
elif cmd == 'UID_STORE':
|
|
mset = rest.split(' ')[0]
|
|
uids = parse_uid_set(mset)
|
|
for uid in uids:
|
|
conn.sendall(f'* {uid} FETCH (UID {uid} FLAGS (\\Seen))\r\n'.encode())
|
|
conn.sendall(f'{tag} OK STORE completed\r\n'.encode())
|
|
elif cmd == 'LOGOUT':
|
|
conn.sendall(b'* BYE Logging out\r\n')
|
|
conn.sendall(f'{tag} OK LOGOUT completed\r\n'.encode())
|
|
return
|
|
else:
|
|
conn.sendall(f'{tag} OK {cmd} completed\r\n'.encode())
|
|
except Exception:
|
|
pass
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('port', type=int)
|
|
parser.add_argument('--auth-user', default='test@example.com')
|
|
parser.add_argument('--auth-pass', default='secret')
|
|
args = parser.parse_args()
|
|
|
|
srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
srv.bind(('127.0.0.1', args.port))
|
|
srv.listen(5)
|
|
print(f'MOCK_IMAP_READY {args.port}', flush=True)
|
|
while True:
|
|
conn, _ = srv.accept()
|
|
t = threading.Thread(target=handle, args=(conn, args.auth_user, args.auth_pass), daemon=True)
|
|
t.start()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|