#!/usr/bin/env python3 """Mock IMAP 服务器 —— 用于协议栈集成测试 支持: LOGIN/AUTHENTICATE PLAIN, CAPABILITY, LIST, SELECT, UID SEARCH, UID FETCH(含多段字面量), UID STORE, LOGOUT 用法: python3 mock_imap_server.py [--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" \r\n' 'To: lisi@example.com\r\n' 'Subject: =?UTF-8?B?5L2g5aW955qE5L2T5YaF?=\r\n' 'Message-ID: \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" \r\n' 'To: zhangsan@example.com\r\n' 'Subject: Re: 测试邮件\r\n' 'Message-ID: \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 （一步式） 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" \r\n' fields += b'To: lisi@example.com\r\n' fields += b'Subject: =?UTF-8?B?5L2g5aW955qE5L2T5YaF?=\r\n' fields += b'Message-ID: \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()