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 115 116 117 118 119 120 121 122 123 124
|
#!/usr/bin/env python
"""dummy SMTP server for use in tests"""
import optparse
import ssl
import sys
import traceback
import warnings
with warnings.catch_warnings():
warnings.filterwarnings('ignore', 'The (asyncore|smtpd) module is deprecated', DeprecationWarning)
import asyncore
import smtpd
from mercurial import (
pycompat,
server,
sslutil,
ui as uimod,
)
def log(msg):
sys.stdout.write(msg)
sys.stdout.flush()
class dummysmtpserver(smtpd.SMTPServer):
def __init__(self, localaddr):
smtpd.SMTPServer.__init__(self, localaddr, remoteaddr=None)
def process_message(self, peer, mailfrom, rcpttos, data, **kwargs):
log('%s from=%s to=%s\n' % (peer[0], mailfrom, ', '.join(rcpttos)))
def handle_error(self):
# On Windows, a bad SSL connection sometimes generates a WSAECONNRESET.
# The default handler will shutdown this server, and then both the
# current connection and subsequent ones fail on the client side with
# "No connection could be made because the target machine actively
# refused it". If we eat the error, then the client properly aborts in
# the expected way, and the server is available for subsequent requests.
traceback.print_exc()
class dummysmtpsecureserver(dummysmtpserver):
def __init__(self, localaddr, certfile):
dummysmtpserver.__init__(self, localaddr)
self._certfile = certfile
def handle_accept(self):
pair = self.accept()
if not pair:
return
conn, addr = pair
ui = uimod.ui.load()
try:
# wrap_socket() would block, but we don't care
conn = sslutil.wrapserversocket(conn, ui, certfile=self._certfile)
except ssl.SSLError:
log('%s ssl error\n' % addr[0])
conn.close()
return
smtpd.SMTPChannel(self, conn, addr)
def run():
try:
asyncore.loop()
except KeyboardInterrupt:
pass
def _encodestrsonly(v):
if isinstance(v, type(u'')):
return v.encode('ascii')
return v
def bytesvars(obj):
unidict = vars(obj)
bd = {k.encode('ascii'): _encodestrsonly(v) for k, v in unidict.items()}
if bd[b'daemon_postexec'] is not None:
bd[b'daemon_postexec'] = [
_encodestrsonly(v) for v in bd[b'daemon_postexec']
]
return bd
def main():
op = optparse.OptionParser()
op.add_option('-d', '--daemon', action='store_true')
op.add_option('--daemon-postexec', action='append')
op.add_option('-p', '--port', type=int, default=8025)
op.add_option('-a', '--address', default='localhost')
op.add_option('--pid-file', metavar='FILE')
op.add_option('--tls', choices=['none', 'smtps'], default='none')
op.add_option('--certificate', metavar='FILE')
opts, args = op.parse_args()
if opts.tls == 'smtps' and not opts.certificate:
op.error('--certificate must be specified')
addr = (opts.address, opts.port)
def init():
if opts.tls == 'none':
dummysmtpserver(addr)
else:
dummysmtpsecureserver(addr, opts.certificate)
log('listening at %s:%d\n' % addr)
server.runservice(
bytesvars(opts),
initfn=init,
runfn=run,
runargs=[pycompat.sysexecutable, pycompat.fsencode(__file__)]
+ pycompat.sysargv[1:],
)
if __name__ == '__main__':
main()
|