-
Notifications
You must be signed in to change notification settings - Fork 0
/
smtp_sync.py
114 lines (81 loc) · 4.38 KB
/
smtp_sync.py
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
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
# system imports:
from abc import abstractmethod
import logging
import sys
# email_proto imports:
from event_handling import SyncClient, SyncServer
import smtp_proto as proto
from transport import SyncTransport
from util import b2s
logger = logging.getLogger ( __name__ )
class Client ( SyncClient ):
protocls = proto.Client
def greeting ( self ) -> proto.SuccessResponse:
return self._request ( proto.GreetingRequest() )
def helo ( self, local_hostname: str ) -> proto.SuccessResponse:
return self._request ( proto.HeloRequest ( local_hostname ) )
def ehlo ( self, local_hostname: str ) -> proto.EhloResponse:
return self._request ( proto.EhloRequest ( local_hostname ) )
def starttls ( self ) -> proto.SuccessResponse:
return self._request ( proto.StartTlsRequest() )
def auth_plain1 ( self, uid: str, pwd: str ) -> proto.SuccessResponse:
return self._request ( proto.AuthPlain1Request ( uid, pwd ) )
def auth_plain2 ( self, uid: str, pwd: str ) -> proto.SuccessResponse:
return self._request ( proto.AuthPlain2Request ( uid, pwd ) )
def auth_login ( self, uid: str, pwd: str ) -> proto.SuccessResponse:
return self._request ( proto.AuthLoginRequest ( uid, pwd ) )
def expn ( self, mailbox: str ) -> proto.ExpnResponse:
return self._request ( proto.ExpnRequest ( mailbox ) )
def vrfy ( self, mailbox: str ) -> proto.VrfyResponse:
return self._request ( proto.VrfyRequest ( mailbox ) )
def mail_from ( self, email: str ) -> proto.SuccessResponse:
return self._request ( proto.MailFromRequest ( email ) )
def rcpt_to ( self, email: str ) -> proto.SuccessResponse:
return self._request ( proto.RcptToRequest ( email ) )
def data ( self, content: bytes ) -> proto.SuccessResponse:
return self._request ( proto.DataRequest ( content ) )
def rset ( self ) -> proto.SuccessResponse:
return self._request ( proto.RsetRequest() )
def noop ( self ) -> proto.SuccessResponse:
return self._request ( proto.NoOpRequest() )
def quit ( self ) -> proto.SuccessResponse:
return self._request ( proto.QuitRequest() )
def on_StartTlsBeginEvent ( self, event: proto.StartTlsBeginEvent ) -> None:
self.transport.starttls_client ( self.server_hostname )
class Server ( SyncServer ):
protocls = proto.Server
def on_GreetingAcceptEvent ( self, event: proto.GreetingAcceptEvent ) -> None:
# implementations only need to override this if they want to change the behavior
event.accept()
def on_HeloAcceptEvent ( self, event: proto.HeloAcceptEvent ) -> None:
# implementations only need to override this if they want to change the behavior
event.accept()
def on_EhloAcceptEvent ( self, event: proto.EhloAcceptEvent ) -> None:
# implementations only need to override this if they want to change the behavior
event.accept()
@abstractmethod
def on_StartTlsAcceptEvent ( self, event: proto.StartTlsAcceptEvent ) -> None: # pragma: no cover
cls = type ( self )
raise NotImplementedError ( f'{cls.__module__}.{cls.__name__}.on_StartTlsAcceptEvent()' )
def on_StartTlsBeginEvent ( self, event: proto.StartTlsBeginEvent ) -> None: # pragma: no cover
self.transport.starttls_server()
@abstractmethod
def on_AuthEvent ( self, event: proto.AuthEvent ) -> None: # pragma: no cover
cls = type ( self )
raise NotImplementedError ( f'{cls.__module__}.{cls.__name__}.on_AuthEvent()' )
def on_ExpnEvent ( self, event: proto.ExpnEvent ) -> None:
event.reject() # NOTE: it isn't required to implement this. The default behavior is to report '550 Access Denied!'
def on_VrfyEvent ( self, event: proto.VrfyEvent ) -> None:
event.reject() # NOTE: it isn't required to implement this. The default behavior is to report '550 Access Denied!'
@abstractmethod
def on_MailFromEvent ( self, event: proto.MailFromEvent ) -> None: # pragma: no cover
cls = type ( self )
raise NotImplementedError ( f'{cls.__module__}.{cls.__name__}.on_MailFromEvent()' )
@abstractmethod
def on_RcptToEvent ( self, event: proto.RcptToEvent ) -> None: # pragma: no cover
cls = type ( self )
raise NotImplementedError ( f'{cls.__module__}.{cls.__name__}.on_RcptToEvent()' )
@abstractmethod
def on_CompleteEvent ( self, event: proto.CompleteEvent ) -> None: # pragma: no cover
cls = type ( self )
raise NotImplementedError ( f'{cls.__module__}.{cls.__name__}.on_CompleteEvent()' )