forked from ietf-tools/datatracker
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathaiosmtpd.py
More file actions
73 lines (58 loc) · 2.89 KB
/
aiosmtpd.py
File metadata and controls
73 lines (58 loc) · 2.89 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
# Copyright The IETF Trust 2014-2025, All Rights Reserved
"""aiosmtpd-related utilities
These are for testing / dev use. If you're using this for production code, think very
hard about the choices you're making...
"""
from aiosmtpd import handlers
from aiosmtpd.controller import Controller
from aiosmtpd.smtp import SMTP
from email.utils import parseaddr
from typing import Optional, TextIO
class SMTPTestHandler:
def __init__(self, inbox: list):
self.inbox = inbox
async def handle_DATA(self, server, session, envelope):
"""Handle the DATA command and 'deliver' the message"""
self.inbox.append(envelope.content)
# Per RFC2033: https://datatracker.ietf.org/doc/html/rfc2033.html#section-4.2
# ...after the final ".", the server returns one reply
# for each previously successful RCPT command in the mail transaction,
# in the order that the RCPT commands were issued. Even if there were
# multiple successful RCPT commands giving the same forward-path, there
# must be one reply for each successful RCPT command.
return "\n".join("250 OK" for _ in envelope.rcpt_tos)
async def handle_RCPT(self, server, session, envelope, address, rcpt_options):
"""Handle an RCPT command and add the address to the envelope if it is acceptable"""
_, address = parseaddr(address)
if address == "":
return "501 Syntax: RCPT TO: <address>"
if "poison" in address:
return "550 Error: Not touching that"
# At this point the address is acceptable
envelope.rcpt_tos.append(address)
return "250 OK"
class SMTPTestServerDriver:
def __init__(self, address: str, port: int, inbox: Optional[list] = None):
# Allow longer lines than the 1001 that RFC 5321 requires. As of 2025-04-16 the
# datatracker emits some non-compliant messages.
# See https://aiosmtpd.aio-libs.org/en/latest/smtp.html
SMTP.line_length_limit = 4000 # tests start failing between 3000 and 4000
self.controller = Controller(
hostname=address,
port=port,
handler=SMTPTestHandler(inbox=[] if inbox is None else inbox),
)
def start(self):
self.controller.start()
def stop(self):
self.controller.stop()
class DevDebuggingHandler(handlers.Debugging):
"""Debugging handler for use in dev ONLY"""
def __init__(self, stream: Optional[TextIO] = None):
# Allow longer lines than the 1001 that RFC 5321 requires. As of 2025-04-16 the
# datatracker emits some non-compliant messages.
# See https://aiosmtpd.aio-libs.org/en/latest/smtp.html
# Doing this in a handler class is a huge hack. Tests all pass with this set
# to 4000, but make the limit longer for dev just in case.
SMTP.line_length_limit = 10000
super().__init__(stream)