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
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
|
import asynchat
import asyncore
import socket
import sys
from collections import defaultdict
from typing import Any
from typing_extensions import TypeAlias
if sys.version_info >= (3, 11):
__all__ = ["SMTPChannel", "SMTPServer", "DebuggingServer", "PureProxy"]
else:
__all__ = ["SMTPChannel", "SMTPServer", "DebuggingServer", "PureProxy", "MailmanProxy"]
_Address: TypeAlias = tuple[str, int] # (host, port)
class SMTPChannel(asynchat.async_chat):
COMMAND: int
DATA: int
command_size_limits: defaultdict[str, int]
smtp_server: SMTPServer
conn: socket.socket
addr: Any
received_lines: list[str]
smtp_state: int
seen_greeting: str
mailfrom: str
rcpttos: list[str]
received_data: str
fqdn: str
peer: str
command_size_limit: int
data_size_limit: int
enable_SMTPUTF8: bool
@property
def max_command_size_limit(self) -> int: ...
def __init__(
self,
server: SMTPServer,
conn: socket.socket,
addr: Any,
data_size_limit: int = 33554432,
map: asyncore._MapType | None = None,
enable_SMTPUTF8: bool = False,
decode_data: bool = False,
) -> None: ...
# base asynchat.async_chat.push() accepts bytes
def push(self, msg: str) -> None: ... # type: ignore[override]
def collect_incoming_data(self, data: bytes) -> None: ...
def found_terminator(self) -> None: ...
def smtp_HELO(self, arg: str) -> None: ...
def smtp_NOOP(self, arg: str) -> None: ...
def smtp_QUIT(self, arg: str) -> None: ...
def smtp_MAIL(self, arg: str) -> None: ...
def smtp_RCPT(self, arg: str) -> None: ...
def smtp_RSET(self, arg: str) -> None: ...
def smtp_DATA(self, arg: str) -> None: ...
def smtp_EHLO(self, arg: str) -> None: ...
def smtp_HELP(self, arg: str) -> None: ...
def smtp_VRFY(self, arg: str) -> None: ...
def smtp_EXPN(self, arg: str) -> None: ...
class SMTPServer(asyncore.dispatcher):
channel_class: type[SMTPChannel]
data_size_limit: int
enable_SMTPUTF8: bool
def __init__(
self,
localaddr: _Address,
remoteaddr: _Address,
data_size_limit: int = 33554432,
map: asyncore._MapType | None = None,
enable_SMTPUTF8: bool = False,
decode_data: bool = False,
) -> None: ...
def handle_accepted(self, conn: socket.socket, addr: Any) -> None: ...
def process_message(
self, peer: _Address, mailfrom: str, rcpttos: list[str], data: bytes | str, **kwargs: Any
) -> str | None: ...
class DebuggingServer(SMTPServer): ...
class PureProxy(SMTPServer):
def process_message(self, peer: _Address, mailfrom: str, rcpttos: list[str], data: bytes | str) -> str | None: ... # type: ignore[override]
if sys.version_info < (3, 11):
class MailmanProxy(PureProxy):
def process_message(self, peer: _Address, mailfrom: str, rcpttos: list[str], data: bytes | str) -> str | None: ... # type: ignore[override]
|
