+++ /dev/null
-:mod:`smtpd` --- SMTP Server
-============================
-
-.. module:: smtpd
- :synopsis: A SMTP server implementation in Python.
- :deprecated:
-
-.. moduleauthor:: Barry Warsaw <barry@python.org>
-.. sectionauthor:: Moshe Zadka <moshez@moshez.org>
-
-**Source code:** :source:`Lib/smtpd.py`
-
---------------
-
-This module offers several classes to implement SMTP (email) servers.
-
-.. deprecated-removed:: 3.6 3.12
- The :mod:`smtpd` module is deprecated
- (see :pep:`PEP 594 <594#smtpd>` for details).
- The `aiosmtpd <https://aiosmtpd.readthedocs.io/>`_ package is a recommended
- replacement for this module. It is based on :mod:`asyncio` and provides a
- more straightforward API.
-
-Several server implementations are present; one is a generic
-do-nothing implementation, which can be overridden, while the other two offer
-specific mail-sending strategies.
-
-Additionally the SMTPChannel may be extended to implement very specific
-interaction behaviour with SMTP clients.
-
-The code supports :RFC:`5321`, plus the :rfc:`1870` SIZE and :rfc:`6531`
-SMTPUTF8 extensions.
-
-.. include:: ../includes/wasm-notavail.rst
-
-SMTPServer Objects
-------------------
-
-
-.. class:: SMTPServer(localaddr, remoteaddr, data_size_limit=33554432,\
- map=None, enable_SMTPUTF8=False, decode_data=False)
-
- Create a new :class:`SMTPServer` object, which binds to local address
- *localaddr*. It will treat *remoteaddr* as an upstream SMTP relayer. Both
- *localaddr* and *remoteaddr* should be a :ref:`(host, port) <host_port>`
- tuple. The object inherits from :class:`asyncore.dispatcher`, and so will
- insert itself into :mod:`asyncore`'s event loop on instantiation.
-
- *data_size_limit* specifies the maximum number of bytes that will be
- accepted in a ``DATA`` command. A value of ``None`` or ``0`` means no
- limit.
-
- *map* is the socket map to use for connections (an initially empty
- dictionary is a suitable value). If not specified the :mod:`asyncore`
- global socket map is used.
-
- *enable_SMTPUTF8* determines whether the ``SMTPUTF8`` extension (as defined
- in :RFC:`6531`) should be enabled. The default is ``False``.
- When ``True``, ``SMTPUTF8`` is accepted as a parameter to the ``MAIL``
- command and when present is passed to :meth:`process_message` in the
- ``kwargs['mail_options']`` list. *decode_data* and *enable_SMTPUTF8*
- cannot be set to ``True`` at the same time.
-
- *decode_data* specifies whether the data portion of the SMTP transaction
- should be decoded using UTF-8. When *decode_data* is ``False`` (the
- default), the server advertises the ``8BITMIME``
- extension (:rfc:`6152`), accepts the ``BODY=8BITMIME`` parameter to
- the ``MAIL`` command, and when present passes it to :meth:`process_message`
- in the ``kwargs['mail_options']`` list. *decode_data* and *enable_SMTPUTF8*
- cannot be set to ``True`` at the same time.
-
- .. method:: process_message(peer, mailfrom, rcpttos, data, **kwargs)
-
- Raise a :exc:`NotImplementedError` exception. Override this in subclasses to
- do something useful with this message. Whatever was passed in the
- constructor as *remoteaddr* will be available as the :attr:`_remoteaddr`
- attribute. *peer* is the remote host's address, *mailfrom* is the envelope
- originator, *rcpttos* are the envelope recipients and *data* is a string
- containing the contents of the e-mail (which should be in :rfc:`5321`
- format).
-
- If the *decode_data* constructor keyword is set to ``True``, the *data*
- argument will be a unicode string. If it is set to ``False``, it
- will be a bytes object.
-
- *kwargs* is a dictionary containing additional information. It is empty
- if ``decode_data=True`` was given as an init argument, otherwise
- it contains the following keys:
-
- *mail_options*:
- a list of all received parameters to the ``MAIL``
- command (the elements are uppercase strings; example:
- ``['BODY=8BITMIME', 'SMTPUTF8']``).
-
- *rcpt_options*:
- same as *mail_options* but for the ``RCPT`` command.
- Currently no ``RCPT TO`` options are supported, so for now
- this will always be an empty list.
-
- Implementations of ``process_message`` should use the ``**kwargs``
- signature to accept arbitrary keyword arguments, since future feature
- enhancements may add keys to the kwargs dictionary.
-
- Return ``None`` to request a normal ``250 Ok`` response; otherwise
- return the desired response string in :RFC:`5321` format.
-
- .. attribute:: channel_class
-
- Override this in subclasses to use a custom :class:`SMTPChannel` for
- managing SMTP clients.
-
- .. versionadded:: 3.4
- The *map* constructor argument.
-
- .. versionchanged:: 3.5
- *localaddr* and *remoteaddr* may now contain IPv6 addresses.
-
- .. versionadded:: 3.5
- The *decode_data* and *enable_SMTPUTF8* constructor parameters, and the
- *kwargs* parameter to :meth:`process_message` when *decode_data* is
- ``False``.
-
- .. versionchanged:: 3.6
- *decode_data* is now ``False`` by default.
-
-
-DebuggingServer Objects
------------------------
-
-
-.. class:: DebuggingServer(localaddr, remoteaddr)
-
- Create a new debugging server. Arguments are as per :class:`SMTPServer`.
- Messages will be discarded, and printed on stdout.
-
-
-PureProxy Objects
------------------
-
-
-.. class:: PureProxy(localaddr, remoteaddr)
-
- Create a new pure proxy server. Arguments are as per :class:`SMTPServer`.
- Everything will be relayed to *remoteaddr*. Note that running this has a good
- chance to make you into an open relay, so please be careful.
-
-
-SMTPChannel Objects
--------------------
-
-.. class:: SMTPChannel(server, conn, addr, data_size_limit=33554432,\
- map=None, enable_SMTPUTF8=False, decode_data=False)
-
- Create a new :class:`SMTPChannel` object which manages the communication
- between the server and a single SMTP client.
-
- *conn* and *addr* are as per the instance variables described below.
-
- *data_size_limit* specifies the maximum number of bytes that will be
- accepted in a ``DATA`` command. A value of ``None`` or ``0`` means no
- limit.
-
- *enable_SMTPUTF8* determines whether the ``SMTPUTF8`` extension (as defined
- in :RFC:`6531`) should be enabled. The default is ``False``.
- *decode_data* and *enable_SMTPUTF8* cannot be set to ``True`` at the same
- time.
-
- A dictionary can be specified in *map* to avoid using a global socket map.
-
- *decode_data* specifies whether the data portion of the SMTP transaction
- should be decoded using UTF-8. The default is ``False``.
- *decode_data* and *enable_SMTPUTF8* cannot be set to ``True`` at the same
- time.
-
- To use a custom SMTPChannel implementation you need to override the
- :attr:`SMTPServer.channel_class` of your :class:`SMTPServer`.
-
- .. versionchanged:: 3.5
- The *decode_data* and *enable_SMTPUTF8* parameters were added.
-
- .. versionchanged:: 3.6
- *decode_data* is now ``False`` by default.
-
- The :class:`SMTPChannel` has the following instance variables:
-
- .. attribute:: smtp_server
-
- Holds the :class:`SMTPServer` that spawned this channel.
-
- .. attribute:: conn
-
- Holds the socket object connecting to the client.
-
- .. attribute:: addr
-
- Holds the address of the client, the second value returned by
- :func:`socket.accept <socket.socket.accept>`
-
- .. attribute:: received_lines
-
- Holds a list of the line strings (decoded using UTF-8) received from
- the client. The lines have their ``"\r\n"`` line ending translated to
- ``"\n"``.
-
- .. attribute:: smtp_state
-
- Holds the current state of the channel. This will be either
- :attr:`COMMAND` initially and then :attr:`DATA` after the client sends
- a "DATA" line.
-
- .. attribute:: seen_greeting
-
- Holds a string containing the greeting sent by the client in its "HELO".
-
- .. attribute:: mailfrom
-
- Holds a string containing the address identified in the "MAIL FROM:" line
- from the client.
-
- .. attribute:: rcpttos
-
- Holds a list of strings containing the addresses identified in the
- "RCPT TO:" lines from the client.
-
- .. attribute:: received_data
-
- Holds a string containing all of the data sent by the client during the
- DATA state, up to but not including the terminating ``"\r\n.\r\n"``.
-
- .. attribute:: fqdn
-
- Holds the fully qualified domain name of the server as returned by
- :func:`socket.getfqdn`.
-
- .. attribute:: peer
-
- Holds the name of the client peer as returned by ``conn.getpeername()``
- where ``conn`` is :attr:`conn`.
-
- The :class:`SMTPChannel` operates by invoking methods named ``smtp_<command>``
- upon reception of a command line from the client. Built into the base
- :class:`SMTPChannel` class are methods for handling the following commands
- (and responding to them appropriately):
-
- ======== ===================================================================
- Command Action taken
- ======== ===================================================================
- HELO Accepts the greeting from the client and stores it in
- :attr:`seen_greeting`. Sets server to base command mode.
- EHLO Accepts the greeting from the client and stores it in
- :attr:`seen_greeting`. Sets server to extended command mode.
- NOOP Takes no action.
- QUIT Closes the connection cleanly.
- MAIL Accepts the "MAIL FROM:" syntax and stores the supplied address as
- :attr:`mailfrom`. In extended command mode, accepts the
- :rfc:`1870` SIZE attribute and responds appropriately based on the
- value of *data_size_limit*.
- RCPT Accepts the "RCPT TO:" syntax and stores the supplied addresses in
- the :attr:`rcpttos` list.
- RSET Resets the :attr:`mailfrom`, :attr:`rcpttos`, and
- :attr:`received_data`, but not the greeting.
- DATA Sets the internal state to :attr:`DATA` and stores remaining lines
- from the client in :attr:`received_data` until the terminator
- ``"\r\n.\r\n"`` is received.
- HELP Returns minimal information on command syntax
- VRFY Returns code 252 (the server doesn't know if the address is valid)
- EXPN Reports that the command is not implemented.
- ======== ===================================================================
+++ /dev/null
-import unittest
-import textwrap
-from test import support, mock_socket
-from test.support import socket_helper
-from test.support import warnings_helper
-import socket
-import io
-
-
-smtpd = warnings_helper.import_deprecated('smtpd')
-asyncore = warnings_helper.import_deprecated('asyncore')
-
-if not socket_helper.has_gethostname:
- raise unittest.SkipTest("test requires gethostname()")
-
-
-class DummyServer(smtpd.SMTPServer):
- def __init__(self, *args, **kwargs):
- smtpd.SMTPServer.__init__(self, *args, **kwargs)
- self.messages = []
- if self._decode_data:
- self.return_status = 'return status'
- else:
- self.return_status = b'return status'
-
- def process_message(self, peer, mailfrom, rcpttos, data, **kw):
- self.messages.append((peer, mailfrom, rcpttos, data))
- if data == self.return_status:
- return '250 Okish'
- if 'mail_options' in kw and 'SMTPUTF8' in kw['mail_options']:
- return '250 SMTPUTF8 message okish'
-
-
-class DummyDispatcherBroken(Exception):
- pass
-
-
-class BrokenDummyServer(DummyServer):
- def listen(self, num):
- raise DummyDispatcherBroken()
-
-
-class SMTPDServerTest(unittest.TestCase):
- def setUp(self):
- smtpd.socket = asyncore.socket = mock_socket
-
- def test_process_message_unimplemented(self):
- server = smtpd.SMTPServer((socket_helper.HOST, 0), ('b', 0),
- decode_data=True)
- conn, addr = server.accept()
- channel = smtpd.SMTPChannel(server, conn, addr, decode_data=True)
-
- def write_line(line):
- channel.socket.queue_recv(line)
- channel.handle_read()
-
- write_line(b'HELO example')
- write_line(b'MAIL From:eggs@example')
- write_line(b'RCPT To:spam@example')
- write_line(b'DATA')
- self.assertRaises(NotImplementedError, write_line, b'spam\r\n.\r\n')
-
- def test_decode_data_and_enable_SMTPUTF8_raises(self):
- self.assertRaises(
- ValueError,
- smtpd.SMTPServer,
- (socket_helper.HOST, 0),
- ('b', 0),
- enable_SMTPUTF8=True,
- decode_data=True)
-
- def tearDown(self):
- asyncore.close_all()
- asyncore.socket = smtpd.socket = socket
-
-
-class DebuggingServerTest(unittest.TestCase):
-
- def setUp(self):
- smtpd.socket = asyncore.socket = mock_socket
-
- def send_data(self, channel, data, enable_SMTPUTF8=False):
- def write_line(line):
- channel.socket.queue_recv(line)
- channel.handle_read()
- write_line(b'EHLO example')
- if enable_SMTPUTF8:
- write_line(b'MAIL From:eggs@example BODY=8BITMIME SMTPUTF8')
- else:
- write_line(b'MAIL From:eggs@example')
- write_line(b'RCPT To:spam@example')
- write_line(b'DATA')
- write_line(data)
- write_line(b'.')
-
- def test_process_message_with_decode_data_true(self):
- server = smtpd.DebuggingServer((socket_helper.HOST, 0), ('b', 0),
- decode_data=True)
- conn, addr = server.accept()
- channel = smtpd.SMTPChannel(server, conn, addr, decode_data=True)
- with support.captured_stdout() as s:
- self.send_data(channel, b'From: test\n\nhello\n')
- stdout = s.getvalue()
- self.assertEqual(stdout, textwrap.dedent("""\
- ---------- MESSAGE FOLLOWS ----------
- From: test
- X-Peer: peer-address
-
- hello
- ------------ END MESSAGE ------------
- """))
-
- def test_process_message_with_decode_data_false(self):
- server = smtpd.DebuggingServer((socket_helper.HOST, 0), ('b', 0))
- conn, addr = server.accept()
- channel = smtpd.SMTPChannel(server, conn, addr)
- with support.captured_stdout() as s:
- self.send_data(channel, b'From: test\n\nh\xc3\xa9llo\xff\n')
- stdout = s.getvalue()
- self.assertEqual(stdout, textwrap.dedent("""\
- ---------- MESSAGE FOLLOWS ----------
- b'From: test'
- b'X-Peer: peer-address'
- b''
- b'h\\xc3\\xa9llo\\xff'
- ------------ END MESSAGE ------------
- """))
-
- def test_process_message_with_enable_SMTPUTF8_true(self):
- server = smtpd.DebuggingServer((socket_helper.HOST, 0), ('b', 0),
- enable_SMTPUTF8=True)
- conn, addr = server.accept()
- channel = smtpd.SMTPChannel(server, conn, addr, enable_SMTPUTF8=True)
- with support.captured_stdout() as s:
- self.send_data(channel, b'From: test\n\nh\xc3\xa9llo\xff\n')
- stdout = s.getvalue()
- self.assertEqual(stdout, textwrap.dedent("""\
- ---------- MESSAGE FOLLOWS ----------
- b'From: test'
- b'X-Peer: peer-address'
- b''
- b'h\\xc3\\xa9llo\\xff'
- ------------ END MESSAGE ------------
- """))
-
- def test_process_SMTPUTF8_message_with_enable_SMTPUTF8_true(self):
- server = smtpd.DebuggingServer((socket_helper.HOST, 0), ('b', 0),
- enable_SMTPUTF8=True)
- conn, addr = server.accept()
- channel = smtpd.SMTPChannel(server, conn, addr, enable_SMTPUTF8=True)
- with support.captured_stdout() as s:
- self.send_data(channel, b'From: test\n\nh\xc3\xa9llo\xff\n',
- enable_SMTPUTF8=True)
- stdout = s.getvalue()
- self.assertEqual(stdout, textwrap.dedent("""\
- ---------- MESSAGE FOLLOWS ----------
- mail options: ['BODY=8BITMIME', 'SMTPUTF8']
- b'From: test'
- b'X-Peer: peer-address'
- b''
- b'h\\xc3\\xa9llo\\xff'
- ------------ END MESSAGE ------------
- """))
-
- def tearDown(self):
- asyncore.close_all()
- asyncore.socket = smtpd.socket = socket
-
-
-class TestFamilyDetection(unittest.TestCase):
- def setUp(self):
- smtpd.socket = asyncore.socket = mock_socket
-
- def tearDown(self):
- asyncore.close_all()
- asyncore.socket = smtpd.socket = socket
-
- @unittest.skipUnless(socket_helper.IPV6_ENABLED, "IPv6 not enabled")
- def test_socket_uses_IPv6(self):
- server = smtpd.SMTPServer((socket_helper.HOSTv6, 0), (socket_helper.HOSTv4, 0))
- self.assertEqual(server.socket.family, socket.AF_INET6)
-
- def test_socket_uses_IPv4(self):
- server = smtpd.SMTPServer((socket_helper.HOSTv4, 0), (socket_helper.HOSTv6, 0))
- self.assertEqual(server.socket.family, socket.AF_INET)
-
-
-class TestRcptOptionParsing(unittest.TestCase):
- error_response = (b'555 RCPT TO parameters not recognized or not '
- b'implemented\r\n')
-
- def setUp(self):
- smtpd.socket = asyncore.socket = mock_socket
- self.old_debugstream = smtpd.DEBUGSTREAM
- self.debug = smtpd.DEBUGSTREAM = io.StringIO()
-
- def tearDown(self):
- asyncore.close_all()
- asyncore.socket = smtpd.socket = socket
- smtpd.DEBUGSTREAM = self.old_debugstream
-
- def write_line(self, channel, line):
- channel.socket.queue_recv(line)
- channel.handle_read()
-
- def test_params_rejected(self):
- server = DummyServer((socket_helper.HOST, 0), ('b', 0))
- conn, addr = server.accept()
- channel = smtpd.SMTPChannel(server, conn, addr)
- self.write_line(channel, b'EHLO example')
- self.write_line(channel, b'MAIL from: <foo@example.com> size=20')
- self.write_line(channel, b'RCPT to: <foo@example.com> foo=bar')
- self.assertEqual(channel.socket.last, self.error_response)
-
- def test_nothing_accepted(self):
- server = DummyServer((socket_helper.HOST, 0), ('b', 0))
- conn, addr = server.accept()
- channel = smtpd.SMTPChannel(server, conn, addr)
- self.write_line(channel, b'EHLO example')
- self.write_line(channel, b'MAIL from: <foo@example.com> size=20')
- self.write_line(channel, b'RCPT to: <foo@example.com>')
- self.assertEqual(channel.socket.last, b'250 OK\r\n')
-
-
-class TestMailOptionParsing(unittest.TestCase):
- error_response = (b'555 MAIL FROM parameters not recognized or not '
- b'implemented\r\n')
-
- def setUp(self):
- smtpd.socket = asyncore.socket = mock_socket
- self.old_debugstream = smtpd.DEBUGSTREAM
- self.debug = smtpd.DEBUGSTREAM = io.StringIO()
-
- def tearDown(self):
- asyncore.close_all()
- asyncore.socket = smtpd.socket = socket
- smtpd.DEBUGSTREAM = self.old_debugstream
-
- def write_line(self, channel, line):
- channel.socket.queue_recv(line)
- channel.handle_read()
-
- def test_with_decode_data_true(self):
- server = DummyServer((socket_helper.HOST, 0), ('b', 0), decode_data=True)
- conn, addr = server.accept()
- channel = smtpd.SMTPChannel(server, conn, addr, decode_data=True)
- self.write_line(channel, b'EHLO example')
- for line in [
- b'MAIL from: <foo@example.com> size=20 SMTPUTF8',
- b'MAIL from: <foo@example.com> size=20 SMTPUTF8 BODY=8BITMIME',
- b'MAIL from: <foo@example.com> size=20 BODY=UNKNOWN',
- b'MAIL from: <foo@example.com> size=20 body=8bitmime',
- ]:
- self.write_line(channel, line)
- self.assertEqual(channel.socket.last, self.error_response)
- self.write_line(channel, b'MAIL from: <foo@example.com> size=20')
- self.assertEqual(channel.socket.last, b'250 OK\r\n')
-
- def test_with_decode_data_false(self):
- server = DummyServer((socket_helper.HOST, 0), ('b', 0))
- conn, addr = server.accept()
- channel = smtpd.SMTPChannel(server, conn, addr)
- self.write_line(channel, b'EHLO example')
- for line in [
- b'MAIL from: <foo@example.com> size=20 SMTPUTF8',
- b'MAIL from: <foo@example.com> size=20 SMTPUTF8 BODY=8BITMIME',
- ]:
- self.write_line(channel, line)
- self.assertEqual(channel.socket.last, self.error_response)
- self.write_line(
- channel,
- b'MAIL from: <foo@example.com> size=20 SMTPUTF8 BODY=UNKNOWN')
- self.assertEqual(
- channel.socket.last,
- b'501 Error: BODY can only be one of 7BIT, 8BITMIME\r\n')
- self.write_line(
- channel, b'MAIL from: <foo@example.com> size=20 body=8bitmime')
- self.assertEqual(channel.socket.last, b'250 OK\r\n')
-
- def test_with_enable_smtputf8_true(self):
- server = DummyServer((socket_helper.HOST, 0), ('b', 0), enable_SMTPUTF8=True)
- conn, addr = server.accept()
- channel = smtpd.SMTPChannel(server, conn, addr, enable_SMTPUTF8=True)
- self.write_line(channel, b'EHLO example')
- self.write_line(
- channel,
- b'MAIL from: <foo@example.com> size=20 body=8bitmime smtputf8')
- self.assertEqual(channel.socket.last, b'250 OK\r\n')
-
-
-class SMTPDChannelTest(unittest.TestCase):
- def setUp(self):
- smtpd.socket = asyncore.socket = mock_socket
- self.old_debugstream = smtpd.DEBUGSTREAM
- self.debug = smtpd.DEBUGSTREAM = io.StringIO()
- self.server = DummyServer((socket_helper.HOST, 0), ('b', 0),
- decode_data=True)
- conn, addr = self.server.accept()
- self.channel = smtpd.SMTPChannel(self.server, conn, addr,
- decode_data=True)
-
- def tearDown(self):
- asyncore.close_all()
- asyncore.socket = smtpd.socket = socket
- smtpd.DEBUGSTREAM = self.old_debugstream
-
- def write_line(self, line):
- self.channel.socket.queue_recv(line)
- self.channel.handle_read()
-
- def test_broken_connect(self):
- self.assertRaises(
- DummyDispatcherBroken, BrokenDummyServer,
- (socket_helper.HOST, 0), ('b', 0), decode_data=True)
-
- def test_decode_data_and_enable_SMTPUTF8_raises(self):
- self.assertRaises(
- ValueError, smtpd.SMTPChannel,
- self.server, self.channel.conn, self.channel.addr,
- enable_SMTPUTF8=True, decode_data=True)
-
- def test_server_accept(self):
- self.server.handle_accept()
-
- def test_missing_data(self):
- self.write_line(b'')
- self.assertEqual(self.channel.socket.last,
- b'500 Error: bad syntax\r\n')
-
- def test_EHLO(self):
- self.write_line(b'EHLO example')
- self.assertEqual(self.channel.socket.last, b'250 HELP\r\n')
-
- def test_EHLO_bad_syntax(self):
- self.write_line(b'EHLO')
- self.assertEqual(self.channel.socket.last,
- b'501 Syntax: EHLO hostname\r\n')
-
- def test_EHLO_duplicate(self):
- self.write_line(b'EHLO example')
- self.write_line(b'EHLO example')
- self.assertEqual(self.channel.socket.last,
- b'503 Duplicate HELO/EHLO\r\n')
-
- def test_EHLO_HELO_duplicate(self):
- self.write_line(b'EHLO example')
- self.write_line(b'HELO example')
- self.assertEqual(self.channel.socket.last,
- b'503 Duplicate HELO/EHLO\r\n')
-
- def test_HELO(self):
- name = smtpd.socket.getfqdn()
- self.write_line(b'HELO example')
- self.assertEqual(self.channel.socket.last,
- '250 {}\r\n'.format(name).encode('ascii'))
-
- def test_HELO_EHLO_duplicate(self):
- self.write_line(b'HELO example')
- self.write_line(b'EHLO example')
- self.assertEqual(self.channel.socket.last,
- b'503 Duplicate HELO/EHLO\r\n')
-
- def test_HELP(self):
- self.write_line(b'HELP')
- self.assertEqual(self.channel.socket.last,
- b'250 Supported commands: EHLO HELO MAIL RCPT ' + \
- b'DATA RSET NOOP QUIT VRFY\r\n')
-
- def test_HELP_command(self):
- self.write_line(b'HELP MAIL')
- self.assertEqual(self.channel.socket.last,
- b'250 Syntax: MAIL FROM: <address>\r\n')
-
- def test_HELP_command_unknown(self):
- self.write_line(b'HELP SPAM')
- self.assertEqual(self.channel.socket.last,
- b'501 Supported commands: EHLO HELO MAIL RCPT ' + \
- b'DATA RSET NOOP QUIT VRFY\r\n')
-
- def test_HELO_bad_syntax(self):
- self.write_line(b'HELO')
- self.assertEqual(self.channel.socket.last,
- b'501 Syntax: HELO hostname\r\n')
-
- def test_HELO_duplicate(self):
- self.write_line(b'HELO example')
- self.write_line(b'HELO example')
- self.assertEqual(self.channel.socket.last,
- b'503 Duplicate HELO/EHLO\r\n')
-
- def test_HELO_parameter_rejected_when_extensions_not_enabled(self):
- self.extended_smtp = False
- self.write_line(b'HELO example')
- self.write_line(b'MAIL from:<foo@example.com> SIZE=1234')
- self.assertEqual(self.channel.socket.last,
- b'501 Syntax: MAIL FROM: <address>\r\n')
-
- def test_MAIL_allows_space_after_colon(self):
- self.write_line(b'HELO example')
- self.write_line(b'MAIL from: <foo@example.com>')
- self.assertEqual(self.channel.socket.last,
- b'250 OK\r\n')
-
- def test_extended_MAIL_allows_space_after_colon(self):
- self.write_line(b'EHLO example')
- self.write_line(b'MAIL from: <foo@example.com> size=20')
- self.assertEqual(self.channel.socket.last,
- b'250 OK\r\n')
-
- def test_NOOP(self):
- self.write_line(b'NOOP')
- self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
-
- def test_HELO_NOOP(self):
- self.write_line(b'HELO example')
- self.write_line(b'NOOP')
- self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
-
- def test_NOOP_bad_syntax(self):
- self.write_line(b'NOOP hi')
- self.assertEqual(self.channel.socket.last,
- b'501 Syntax: NOOP\r\n')
-
- def test_QUIT(self):
- self.write_line(b'QUIT')
- self.assertEqual(self.channel.socket.last, b'221 Bye\r\n')
-
- def test_HELO_QUIT(self):
- self.write_line(b'HELO example')
- self.write_line(b'QUIT')
- self.assertEqual(self.channel.socket.last, b'221 Bye\r\n')
-
- def test_QUIT_arg_ignored(self):
- self.write_line(b'QUIT bye bye')
- self.assertEqual(self.channel.socket.last, b'221 Bye\r\n')
-
- def test_bad_state(self):
- self.channel.smtp_state = 'BAD STATE'
- self.write_line(b'HELO example')
- self.assertEqual(self.channel.socket.last,
- b'451 Internal confusion\r\n')
-
- def test_command_too_long(self):
- self.write_line(b'HELO example')
- self.write_line(b'MAIL from: ' +
- b'a' * self.channel.command_size_limit +
- b'@example')
- self.assertEqual(self.channel.socket.last,
- b'500 Error: line too long\r\n')
-
- def test_MAIL_command_limit_extended_with_SIZE(self):
- self.write_line(b'EHLO example')
- fill_len = self.channel.command_size_limit - len('MAIL from:<@example>')
- self.write_line(b'MAIL from:<' +
- b'a' * fill_len +
- b'@example> SIZE=1234')
- self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
-
- self.write_line(b'MAIL from:<' +
- b'a' * (fill_len + 26) +
- b'@example> SIZE=1234')
- self.assertEqual(self.channel.socket.last,
- b'500 Error: line too long\r\n')
-
- def test_MAIL_command_rejects_SMTPUTF8_by_default(self):
- self.write_line(b'EHLO example')
- self.write_line(
- b'MAIL from: <naive@example.com> BODY=8BITMIME SMTPUTF8')
- self.assertEqual(self.channel.socket.last[0:1], b'5')
-
- def test_data_longer_than_default_data_size_limit(self):
- # Hack the default so we don't have to generate so much data.
- self.channel.data_size_limit = 1048
- self.write_line(b'HELO example')
- self.write_line(b'MAIL From:eggs@example')
- self.write_line(b'RCPT To:spam@example')
- self.write_line(b'DATA')
- self.write_line(b'A' * self.channel.data_size_limit +
- b'A\r\n.')
- self.assertEqual(self.channel.socket.last,
- b'552 Error: Too much mail data\r\n')
-
- def test_MAIL_size_parameter(self):
- self.write_line(b'EHLO example')
- self.write_line(b'MAIL FROM:<eggs@example> SIZE=512')
- self.assertEqual(self.channel.socket.last,
- b'250 OK\r\n')
-
- def test_MAIL_invalid_size_parameter(self):
- self.write_line(b'EHLO example')
- self.write_line(b'MAIL FROM:<eggs@example> SIZE=invalid')
- self.assertEqual(self.channel.socket.last,
- b'501 Syntax: MAIL FROM: <address> [SP <mail-parameters>]\r\n')
-
- def test_MAIL_RCPT_unknown_parameters(self):
- self.write_line(b'EHLO example')
- self.write_line(b'MAIL FROM:<eggs@example> ham=green')
- self.assertEqual(self.channel.socket.last,
- b'555 MAIL FROM parameters not recognized or not implemented\r\n')
-
- self.write_line(b'MAIL FROM:<eggs@example>')
- self.write_line(b'RCPT TO:<eggs@example> ham=green')
- self.assertEqual(self.channel.socket.last,
- b'555 RCPT TO parameters not recognized or not implemented\r\n')
-
- def test_MAIL_size_parameter_larger_than_default_data_size_limit(self):
- self.channel.data_size_limit = 1048
- self.write_line(b'EHLO example')
- self.write_line(b'MAIL FROM:<eggs@example> SIZE=2096')
- self.assertEqual(self.channel.socket.last,
- b'552 Error: message size exceeds fixed maximum message size\r\n')
-
- def test_need_MAIL(self):
- self.write_line(b'HELO example')
- self.write_line(b'RCPT to:spam@example')
- self.assertEqual(self.channel.socket.last,
- b'503 Error: need MAIL command\r\n')
-
- def test_MAIL_syntax_HELO(self):
- self.write_line(b'HELO example')
- self.write_line(b'MAIL from eggs@example')
- self.assertEqual(self.channel.socket.last,
- b'501 Syntax: MAIL FROM: <address>\r\n')
-
- def test_MAIL_syntax_EHLO(self):
- self.write_line(b'EHLO example')
- self.write_line(b'MAIL from eggs@example')
- self.assertEqual(self.channel.socket.last,
- b'501 Syntax: MAIL FROM: <address> [SP <mail-parameters>]\r\n')
-
- def test_MAIL_missing_address(self):
- self.write_line(b'HELO example')
- self.write_line(b'MAIL from:')
- self.assertEqual(self.channel.socket.last,
- b'501 Syntax: MAIL FROM: <address>\r\n')
-
- def test_MAIL_chevrons(self):
- self.write_line(b'HELO example')
- self.write_line(b'MAIL from:<eggs@example>')
- self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
-
- def test_MAIL_empty_chevrons(self):
- self.write_line(b'EHLO example')
- self.write_line(b'MAIL from:<>')
- self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
-
- def test_MAIL_quoted_localpart(self):
- self.write_line(b'EHLO example')
- self.write_line(b'MAIL from: <"Fred Blogs"@example.com>')
- self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
- self.assertEqual(self.channel.mailfrom, '"Fred Blogs"@example.com')
-
- def test_MAIL_quoted_localpart_no_angles(self):
- self.write_line(b'EHLO example')
- self.write_line(b'MAIL from: "Fred Blogs"@example.com')
- self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
- self.assertEqual(self.channel.mailfrom, '"Fred Blogs"@example.com')
-
- def test_MAIL_quoted_localpart_with_size(self):
- self.write_line(b'EHLO example')
- self.write_line(b'MAIL from: <"Fred Blogs"@example.com> SIZE=1000')
- self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
- self.assertEqual(self.channel.mailfrom, '"Fred Blogs"@example.com')
-
- def test_MAIL_quoted_localpart_with_size_no_angles(self):
- self.write_line(b'EHLO example')
- self.write_line(b'MAIL from: "Fred Blogs"@example.com SIZE=1000')
- self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
- self.assertEqual(self.channel.mailfrom, '"Fred Blogs"@example.com')
-
- def test_nested_MAIL(self):
- self.write_line(b'HELO example')
- self.write_line(b'MAIL from:eggs@example')
- self.write_line(b'MAIL from:spam@example')
- self.assertEqual(self.channel.socket.last,
- b'503 Error: nested MAIL command\r\n')
-
- def test_VRFY(self):
- self.write_line(b'VRFY eggs@example')
- self.assertEqual(self.channel.socket.last,
- b'252 Cannot VRFY user, but will accept message and attempt ' + \
- b'delivery\r\n')
-
- def test_VRFY_syntax(self):
- self.write_line(b'VRFY')
- self.assertEqual(self.channel.socket.last,
- b'501 Syntax: VRFY <address>\r\n')
-
- def test_EXPN_not_implemented(self):
- self.write_line(b'EXPN')
- self.assertEqual(self.channel.socket.last,
- b'502 EXPN not implemented\r\n')
-
- def test_no_HELO_MAIL(self):
- self.write_line(b'MAIL from:<foo@example.com>')
- self.assertEqual(self.channel.socket.last,
- b'503 Error: send HELO first\r\n')
-
- def test_need_RCPT(self):
- self.write_line(b'HELO example')
- self.write_line(b'MAIL From:eggs@example')
- self.write_line(b'DATA')
- self.assertEqual(self.channel.socket.last,
- b'503 Error: need RCPT command\r\n')
-
- def test_RCPT_syntax_HELO(self):
- self.write_line(b'HELO example')
- self.write_line(b'MAIL From: eggs@example')
- self.write_line(b'RCPT to eggs@example')
- self.assertEqual(self.channel.socket.last,
- b'501 Syntax: RCPT TO: <address>\r\n')
-
- def test_RCPT_syntax_EHLO(self):
- self.write_line(b'EHLO example')
- self.write_line(b'MAIL From: eggs@example')
- self.write_line(b'RCPT to eggs@example')
- self.assertEqual(self.channel.socket.last,
- b'501 Syntax: RCPT TO: <address> [SP <mail-parameters>]\r\n')
-
- def test_RCPT_lowercase_to_OK(self):
- self.write_line(b'HELO example')
- self.write_line(b'MAIL From: eggs@example')
- self.write_line(b'RCPT to: <eggs@example>')
- self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
-
- def test_no_HELO_RCPT(self):
- self.write_line(b'RCPT to eggs@example')
- self.assertEqual(self.channel.socket.last,
- b'503 Error: send HELO first\r\n')
-
- def test_data_dialog(self):
- self.write_line(b'HELO example')
- self.write_line(b'MAIL From:eggs@example')
- self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
- self.write_line(b'RCPT To:spam@example')
- self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
-
- self.write_line(b'DATA')
- self.assertEqual(self.channel.socket.last,
- b'354 End data with <CR><LF>.<CR><LF>\r\n')
- self.write_line(b'data\r\nmore\r\n.')
- self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
- self.assertEqual(self.server.messages,
- [(('peer-address', 'peer-port'),
- 'eggs@example',
- ['spam@example'],
- 'data\nmore')])
-
- def test_DATA_syntax(self):
- self.write_line(b'HELO example')
- self.write_line(b'MAIL From:eggs@example')
- self.write_line(b'RCPT To:spam@example')
- self.write_line(b'DATA spam')
- self.assertEqual(self.channel.socket.last, b'501 Syntax: DATA\r\n')
-
- def test_no_HELO_DATA(self):
- self.write_line(b'DATA spam')
- self.assertEqual(self.channel.socket.last,
- b'503 Error: send HELO first\r\n')
-
- def test_data_transparency_section_4_5_2(self):
- self.write_line(b'HELO example')
- self.write_line(b'MAIL From:eggs@example')
- self.write_line(b'RCPT To:spam@example')
- self.write_line(b'DATA')
- self.write_line(b'..\r\n.\r\n')
- self.assertEqual(self.channel.received_data, '.')
-
- def test_multiple_RCPT(self):
- self.write_line(b'HELO example')
- self.write_line(b'MAIL From:eggs@example')
- self.write_line(b'RCPT To:spam@example')
- self.write_line(b'RCPT To:ham@example')
- self.write_line(b'DATA')
- self.write_line(b'data\r\n.')
- self.assertEqual(self.server.messages,
- [(('peer-address', 'peer-port'),
- 'eggs@example',
- ['spam@example','ham@example'],
- 'data')])
-
- def test_manual_status(self):
- # checks that the Channel is able to return a custom status message
- self.write_line(b'HELO example')
- self.write_line(b'MAIL From:eggs@example')
- self.write_line(b'RCPT To:spam@example')
- self.write_line(b'DATA')
- self.write_line(b'return status\r\n.')
- self.assertEqual(self.channel.socket.last, b'250 Okish\r\n')
-
- def test_RSET(self):
- self.write_line(b'HELO example')
- self.write_line(b'MAIL From:eggs@example')
- self.write_line(b'RCPT To:spam@example')
- self.write_line(b'RSET')
- self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
- self.write_line(b'MAIL From:foo@example')
- self.write_line(b'RCPT To:eggs@example')
- self.write_line(b'DATA')
- self.write_line(b'data\r\n.')
- self.assertEqual(self.server.messages,
- [(('peer-address', 'peer-port'),
- 'foo@example',
- ['eggs@example'],
- 'data')])
-
- def test_HELO_RSET(self):
- self.write_line(b'HELO example')
- self.write_line(b'RSET')
- self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
-
- def test_RSET_syntax(self):
- self.write_line(b'RSET hi')
- self.assertEqual(self.channel.socket.last, b'501 Syntax: RSET\r\n')
-
- def test_unknown_command(self):
- self.write_line(b'UNKNOWN_CMD')
- self.assertEqual(self.channel.socket.last,
- b'500 Error: command "UNKNOWN_CMD" not ' + \
- b'recognized\r\n')
-
- def test_attribute_deprecations(self):
- with warnings_helper.check_warnings(('', DeprecationWarning)):
- spam = self.channel._SMTPChannel__server
- with warnings_helper.check_warnings(('', DeprecationWarning)):
- self.channel._SMTPChannel__server = 'spam'
- with warnings_helper.check_warnings(('', DeprecationWarning)):
- spam = self.channel._SMTPChannel__line
- with warnings_helper.check_warnings(('', DeprecationWarning)):
- self.channel._SMTPChannel__line = 'spam'
- with warnings_helper.check_warnings(('', DeprecationWarning)):
- spam = self.channel._SMTPChannel__state
- with warnings_helper.check_warnings(('', DeprecationWarning)):
- self.channel._SMTPChannel__state = 'spam'
- with warnings_helper.check_warnings(('', DeprecationWarning)):
- spam = self.channel._SMTPChannel__greeting
- with warnings_helper.check_warnings(('', DeprecationWarning)):
- self.channel._SMTPChannel__greeting = 'spam'
- with warnings_helper.check_warnings(('', DeprecationWarning)):
- spam = self.channel._SMTPChannel__mailfrom
- with warnings_helper.check_warnings(('', DeprecationWarning)):
- self.channel._SMTPChannel__mailfrom = 'spam'
- with warnings_helper.check_warnings(('', DeprecationWarning)):
- spam = self.channel._SMTPChannel__rcpttos
- with warnings_helper.check_warnings(('', DeprecationWarning)):
- self.channel._SMTPChannel__rcpttos = 'spam'
- with warnings_helper.check_warnings(('', DeprecationWarning)):
- spam = self.channel._SMTPChannel__data
- with warnings_helper.check_warnings(('', DeprecationWarning)):
- self.channel._SMTPChannel__data = 'spam'
- with warnings_helper.check_warnings(('', DeprecationWarning)):
- spam = self.channel._SMTPChannel__fqdn
- with warnings_helper.check_warnings(('', DeprecationWarning)):
- self.channel._SMTPChannel__fqdn = 'spam'
- with warnings_helper.check_warnings(('', DeprecationWarning)):
- spam = self.channel._SMTPChannel__peer
- with warnings_helper.check_warnings(('', DeprecationWarning)):
- self.channel._SMTPChannel__peer = 'spam'
- with warnings_helper.check_warnings(('', DeprecationWarning)):
- spam = self.channel._SMTPChannel__conn
- with warnings_helper.check_warnings(('', DeprecationWarning)):
- self.channel._SMTPChannel__conn = 'spam'
- with warnings_helper.check_warnings(('', DeprecationWarning)):
- spam = self.channel._SMTPChannel__addr
- with warnings_helper.check_warnings(('', DeprecationWarning)):
- self.channel._SMTPChannel__addr = 'spam'
-
-@unittest.skipUnless(socket_helper.IPV6_ENABLED, "IPv6 not enabled")
-class SMTPDChannelIPv6Test(SMTPDChannelTest):
- def setUp(self):
- smtpd.socket = asyncore.socket = mock_socket
- self.old_debugstream = smtpd.DEBUGSTREAM
- self.debug = smtpd.DEBUGSTREAM = io.StringIO()
- self.server = DummyServer((socket_helper.HOSTv6, 0), ('b', 0),
- decode_data=True)
- conn, addr = self.server.accept()
- self.channel = smtpd.SMTPChannel(self.server, conn, addr,
- decode_data=True)
-
-class SMTPDChannelWithDataSizeLimitTest(unittest.TestCase):
-
- def setUp(self):
- smtpd.socket = asyncore.socket = mock_socket
- self.old_debugstream = smtpd.DEBUGSTREAM
- self.debug = smtpd.DEBUGSTREAM = io.StringIO()
- self.server = DummyServer((socket_helper.HOST, 0), ('b', 0),
- decode_data=True)
- conn, addr = self.server.accept()
- # Set DATA size limit to 32 bytes for easy testing
- self.channel = smtpd.SMTPChannel(self.server, conn, addr, 32,
- decode_data=True)
-
- def tearDown(self):
- asyncore.close_all()
- asyncore.socket = smtpd.socket = socket
- smtpd.DEBUGSTREAM = self.old_debugstream
-
- def write_line(self, line):
- self.channel.socket.queue_recv(line)
- self.channel.handle_read()
-
- def test_data_limit_dialog(self):
- self.write_line(b'HELO example')
- self.write_line(b'MAIL From:eggs@example')
- self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
- self.write_line(b'RCPT To:spam@example')
- self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
-
- self.write_line(b'DATA')
- self.assertEqual(self.channel.socket.last,
- b'354 End data with <CR><LF>.<CR><LF>\r\n')
- self.write_line(b'data\r\nmore\r\n.')
- self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
- self.assertEqual(self.server.messages,
- [(('peer-address', 'peer-port'),
- 'eggs@example',
- ['spam@example'],
- 'data\nmore')])
-
- def test_data_limit_dialog_too_much_data(self):
- self.write_line(b'HELO example')
- self.write_line(b'MAIL From:eggs@example')
- self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
- self.write_line(b'RCPT To:spam@example')
- self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
-
- self.write_line(b'DATA')
- self.assertEqual(self.channel.socket.last,
- b'354 End data with <CR><LF>.<CR><LF>\r\n')
- self.write_line(b'This message is longer than 32 bytes\r\n.')
- self.assertEqual(self.channel.socket.last,
- b'552 Error: Too much mail data\r\n')
-
-
-class SMTPDChannelWithDecodeDataFalse(unittest.TestCase):
-
- def setUp(self):
- smtpd.socket = asyncore.socket = mock_socket
- self.old_debugstream = smtpd.DEBUGSTREAM
- self.debug = smtpd.DEBUGSTREAM = io.StringIO()
- self.server = DummyServer((socket_helper.HOST, 0), ('b', 0))
- conn, addr = self.server.accept()
- self.channel = smtpd.SMTPChannel(self.server, conn, addr)
-
- def tearDown(self):
- asyncore.close_all()
- asyncore.socket = smtpd.socket = socket
- smtpd.DEBUGSTREAM = self.old_debugstream
-
- def write_line(self, line):
- self.channel.socket.queue_recv(line)
- self.channel.handle_read()
-
- def test_ascii_data(self):
- self.write_line(b'HELO example')
- self.write_line(b'MAIL From:eggs@example')
- self.write_line(b'RCPT To:spam@example')
- self.write_line(b'DATA')
- self.write_line(b'plain ascii text')
- self.write_line(b'.')
- self.assertEqual(self.channel.received_data, b'plain ascii text')
-
- def test_utf8_data(self):
- self.write_line(b'HELO example')
- self.write_line(b'MAIL From:eggs@example')
- self.write_line(b'RCPT To:spam@example')
- self.write_line(b'DATA')
- self.write_line(b'utf8 enriched text: \xc5\xbc\xc5\xba\xc4\x87')
- self.write_line(b'and some plain ascii')
- self.write_line(b'.')
- self.assertEqual(
- self.channel.received_data,
- b'utf8 enriched text: \xc5\xbc\xc5\xba\xc4\x87\n'
- b'and some plain ascii')
-
-
-class SMTPDChannelWithDecodeDataTrue(unittest.TestCase):
-
- def setUp(self):
- smtpd.socket = asyncore.socket = mock_socket
- self.old_debugstream = smtpd.DEBUGSTREAM
- self.debug = smtpd.DEBUGSTREAM = io.StringIO()
- self.server = DummyServer((socket_helper.HOST, 0), ('b', 0),
- decode_data=True)
- conn, addr = self.server.accept()
- # Set decode_data to True
- self.channel = smtpd.SMTPChannel(self.server, conn, addr,
- decode_data=True)
-
- def tearDown(self):
- asyncore.close_all()
- asyncore.socket = smtpd.socket = socket
- smtpd.DEBUGSTREAM = self.old_debugstream
-
- def write_line(self, line):
- self.channel.socket.queue_recv(line)
- self.channel.handle_read()
-
- def test_ascii_data(self):
- self.write_line(b'HELO example')
- self.write_line(b'MAIL From:eggs@example')
- self.write_line(b'RCPT To:spam@example')
- self.write_line(b'DATA')
- self.write_line(b'plain ascii text')
- self.write_line(b'.')
- self.assertEqual(self.channel.received_data, 'plain ascii text')
-
- def test_utf8_data(self):
- self.write_line(b'HELO example')
- self.write_line(b'MAIL From:eggs@example')
- self.write_line(b'RCPT To:spam@example')
- self.write_line(b'DATA')
- self.write_line(b'utf8 enriched text: \xc5\xbc\xc5\xba\xc4\x87')
- self.write_line(b'and some plain ascii')
- self.write_line(b'.')
- self.assertEqual(
- self.channel.received_data,
- 'utf8 enriched text: żźć\nand some plain ascii')
-
-
-class SMTPDChannelTestWithEnableSMTPUTF8True(unittest.TestCase):
- def setUp(self):
- smtpd.socket = asyncore.socket = mock_socket
- self.old_debugstream = smtpd.DEBUGSTREAM
- self.debug = smtpd.DEBUGSTREAM = io.StringIO()
- self.server = DummyServer((socket_helper.HOST, 0), ('b', 0),
- enable_SMTPUTF8=True)
- conn, addr = self.server.accept()
- self.channel = smtpd.SMTPChannel(self.server, conn, addr,
- enable_SMTPUTF8=True)
-
- def tearDown(self):
- asyncore.close_all()
- asyncore.socket = smtpd.socket = socket
- smtpd.DEBUGSTREAM = self.old_debugstream
-
- def write_line(self, line):
- self.channel.socket.queue_recv(line)
- self.channel.handle_read()
-
- def test_MAIL_command_accepts_SMTPUTF8_when_announced(self):
- self.write_line(b'EHLO example')
- self.write_line(
- 'MAIL from: <naïve@example.com> BODY=8BITMIME SMTPUTF8'.encode(
- 'utf-8')
- )
- self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
-
- def test_process_smtputf8_message(self):
- self.write_line(b'EHLO example')
- for mail_parameters in [b'', b'BODY=8BITMIME SMTPUTF8']:
- self.write_line(b'MAIL from: <a@example> ' + mail_parameters)
- self.assertEqual(self.channel.socket.last[0:3], b'250')
- self.write_line(b'rcpt to:<b@example.com>')
- self.assertEqual(self.channel.socket.last[0:3], b'250')
- self.write_line(b'data')
- self.assertEqual(self.channel.socket.last[0:3], b'354')
- self.write_line(b'c\r\n.')
- if mail_parameters == b'':
- self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
- else:
- self.assertEqual(self.channel.socket.last,
- b'250 SMTPUTF8 message okish\r\n')
-
- def test_utf8_data(self):
- self.write_line(b'EHLO example')
- self.write_line(
- 'MAIL From: naïve@examplé BODY=8BITMIME SMTPUTF8'.encode('utf-8'))
- self.assertEqual(self.channel.socket.last[0:3], b'250')
- self.write_line('RCPT To:späm@examplé'.encode('utf-8'))
- self.assertEqual(self.channel.socket.last[0:3], b'250')
- self.write_line(b'DATA')
- self.assertEqual(self.channel.socket.last[0:3], b'354')
- self.write_line(b'utf8 enriched text: \xc5\xbc\xc5\xba\xc4\x87')
- self.write_line(b'.')
- self.assertEqual(
- self.channel.received_data,
- b'utf8 enriched text: \xc5\xbc\xc5\xba\xc4\x87')
-
- def test_MAIL_command_limit_extended_with_SIZE_and_SMTPUTF8(self):
- self.write_line(b'ehlo example')
- fill_len = (512 + 26 + 10) - len('mail from:<@example>')
- self.write_line(b'MAIL from:<' +
- b'a' * (fill_len + 1) +
- b'@example>')
- self.assertEqual(self.channel.socket.last,
- b'500 Error: line too long\r\n')
- self.write_line(b'MAIL from:<' +
- b'a' * fill_len +
- b'@example>')
- self.assertEqual(self.channel.socket.last, b'250 OK\r\n')
-
- def test_multiple_emails_with_extended_command_length(self):
- self.write_line(b'ehlo example')
- fill_len = (512 + 26 + 10) - len('mail from:<@example>')
- for char in [b'a', b'b', b'c']:
- self.write_line(b'MAIL from:<' + char * fill_len + b'a@example>')
- self.assertEqual(self.channel.socket.last[0:3], b'500')
- self.write_line(b'MAIL from:<' + char * fill_len + b'@example>')
- self.assertEqual(self.channel.socket.last[0:3], b'250')
- self.write_line(b'rcpt to:<hans@example.com>')
- self.assertEqual(self.channel.socket.last[0:3], b'250')
- self.write_line(b'data')
- self.assertEqual(self.channel.socket.last[0:3], b'354')
- self.write_line(b'test\r\n.')
- self.assertEqual(self.channel.socket.last[0:3], b'250')
-
-
-class MiscTestCase(unittest.TestCase):
- def test__all__(self):
- not_exported = {
- "program", "Devnull", "DEBUGSTREAM", "NEWLINE", "COMMASPACE",
- "DATA_SIZE_DEFAULT", "usage", "Options", "parseargs",
- }
- support.check__all__(self, smtpd, not_exported=not_exported)
-
-
-if __name__ == "__main__":
- unittest.main()
projects / thirdparty / Python / cpython.git / commitdiff
