diff options
| -rw-r--r-- | Doc/library/email.rst | 3 | ||||
| -rw-r--r-- | Doc/library/smtpd.rst | 268 | ||||
| -rw-r--r-- | Doc/library/superseded.rst | 1 | ||||
| -rw-r--r-- | Doc/whatsnew/3.12.rst | 12 | ||||
| -rw-r--r-- | Lib/test/mock_socket.py | 4 | ||||
| -rwxr-xr-x | Lib/test/smtpd.py (renamed from Lib/smtpd.py) | 17 | ||||
| -rw-r--r-- | Lib/test/test_logging.py | 3 | ||||
| -rw-r--r-- | Lib/test/test_smtpd.py | 1019 | ||||
| -rw-r--r-- | Lib/test/test_smtplib.py | 2 | ||||
| -rw-r--r-- | Misc/NEWS.d/next/Library/2022-05-26-08-41-34.gh-issue-93243.uw6x5z.rst | 1 | ||||
| -rw-r--r-- | PCbuild/lib.pyproj | 3 | ||||
| -rw-r--r-- | Python/stdlib_module_names.h | 1 | ||||
| -rwxr-xr-x | Tools/wasm/wasm_assets.py | 1 |
13 files changed, 23 insertions, 1312 deletions
diff --git a/Doc/library/email.rst b/Doc/library/email.rst index 5eebcd9..816fae9 100644 --- a/Doc/library/email.rst +++ b/Doc/library/email.rst @@ -147,6 +147,3 @@ Legacy API: Module :mod:`mailbox` Tools for creating, reading, and managing collections of messages on disk using a variety standard formats. - - Module :mod:`smtpd` - SMTP server framework (primarily useful for testing) diff --git a/Doc/library/smtpd.rst b/Doc/library/smtpd.rst deleted file mode 100644 index ac0c9ae..0000000 --- a/Doc/library/smtpd.rst +++ /dev/null @@ -1,268 +0,0 @@ -: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. - ======== =================================================================== diff --git a/Doc/library/superseded.rst b/Doc/library/superseded.rst index b38f166..57ef963 100644 --- a/Doc/library/superseded.rst +++ b/Doc/library/superseded.rst @@ -27,7 +27,6 @@ backwards compatibility. They have been superseded by other modules. optparse.rst ossaudiodev.rst pipes.rst - smtpd.rst sndhdr.rst spwd.rst sunau.rst diff --git a/Doc/whatsnew/3.12.rst b/Doc/whatsnew/3.12.rst index 9ea4d03..ddf9e1f 100644 --- a/Doc/whatsnew/3.12.rst +++ b/Doc/whatsnew/3.12.rst @@ -299,6 +299,14 @@ Removed (and corresponding ``EXPERIMENTAL_ISOLATED_SUBINTERPRETERS``) have been removed. +* ``smtpd`` has been removed according to the schedule in :pep:`594`, + having been deprecated in Python 3.4.7 and 3.5.4. + Use aiosmtpd_ PyPI module or any other + :mod:`asyncio`-based server instead. + (Contributed by Oleg Iarygin in :gh:`93243`.) + +.. _aiosmtpd: https://pypi.org/project/aiosmtpd/ + * Remove ``io.OpenWrapper`` and ``_pyio.OpenWrapper``, deprecated in Python 3.10: just use :func:`open` instead. The :func:`open` (:func:`io.open`) function is a built-in function. Since Python 3.10, :func:`_pyio.open` is @@ -382,6 +390,10 @@ Changes in the Python API to :term:`filesystem encoding and error handler`. Argument files should be encoded in UTF-8 instead of ANSI Codepage on Windows. +* Removed the ``asyncore``-based ``smtpd`` module deprecated in Python 3.4.7 + and 3.5.4. A recommended replacement is the + :mod:`asyncio`-based aiosmtpd_ PyPI module. + * :func:`shlex.split`: Passing ``None`` for *s* argument now raises an exception, rather than reading :data:`sys.stdin`. The feature was deprecated in Python 3.9. diff --git a/Lib/test/mock_socket.py b/Lib/test/mock_socket.py index c7abddc..b85b955 100644 --- a/Lib/test/mock_socket.py +++ b/Lib/test/mock_socket.py @@ -1,4 +1,4 @@ -"""Mock socket module used by the smtpd and smtplib tests. +"""Mock socket module used by the smtplib tests. """ # imported for _GLOBAL_DEFAULT_TIMEOUT @@ -33,7 +33,7 @@ class MockFile: class MockSocket: - """Mock socket object used by smtpd and smtplib tests. + """Mock socket object used by the smtplib tests. """ def __init__(self, family=None): global _reply_data diff --git a/Lib/smtpd.py b/Lib/test/smtpd.py index b23579f..f9d4b04 100755 --- a/Lib/smtpd.py +++ b/Lib/test/smtpd.py @@ -77,25 +77,16 @@ import getopt import time import socket import collections -from warnings import _deprecated, warn +from test.support.import_helper import import_module +from warnings import warn from email._header_value_parser import get_addr_spec, get_angle_addr __all__ = [ "SMTPChannel", "SMTPServer", "DebuggingServer", "PureProxy", ] -_DEPRECATION_MSG = ('The {name} module is deprecated and unmaintained and will ' - 'be removed in Python {remove}. Please see aiosmtpd ' - '(https://aiosmtpd.readthedocs.io/) for the recommended ' - 'replacement.') -_deprecated(__name__, _DEPRECATION_MSG, remove=(3, 12)) - - -# These are imported after the above warning so that users get the correct -# deprecation warning. -import asyncore -import asynchat - +asyncore = import_module('asyncore', deprecated=True) +asynchat = import_module('asynchat', deprecated=True) program = sys.argv[0] __version__ = 'Python SMTP proxy version 0.3' diff --git a/Lib/test/test_logging.py b/Lib/test/test_logging.py index a900dab..a505e80 100644 --- a/Lib/test/test_logging.py +++ b/Lib/test/test_logging.py @@ -61,9 +61,10 @@ from urllib.parse import urlparse, parse_qs from socketserver import (ThreadingUDPServer, DatagramRequestHandler, ThreadingTCPServer, StreamRequestHandler) +with warnings.catch_warnings(): + from . import smtpd asyncore = warnings_helper.import_deprecated('asyncore') -smtpd = warnings_helper.import_deprecated('smtpd') try: diff --git a/Lib/test/test_smtpd.py b/Lib/test/test_smtpd.py deleted file mode 100644 index 39ff879..0000000 --- a/Lib/test/test_smtpd.py +++ /dev/null @@ -1,1019 +0,0 @@ -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() diff --git a/Lib/test/test_smtplib.py b/Lib/test/test_smtplib.py index a4074c0..3210ef2 100644 --- a/Lib/test/test_smtplib.py +++ b/Lib/test/test_smtplib.py @@ -24,9 +24,9 @@ from test.support import threading_helper from test.support import warnings_helper from unittest.mock import Mock +from . import smtpd asyncore = warnings_helper.import_deprecated('asyncore') -smtpd = warnings_helper.import_deprecated('smtpd') support.requires_working_socket(module=True) diff --git a/Misc/NEWS.d/next/Library/2022-05-26-08-41-34.gh-issue-93243.uw6x5z.rst b/Misc/NEWS.d/next/Library/2022-05-26-08-41-34.gh-issue-93243.uw6x5z.rst new file mode 100644 index 0000000..f03ed7b --- /dev/null +++ b/Misc/NEWS.d/next/Library/2022-05-26-08-41-34.gh-issue-93243.uw6x5z.rst @@ -0,0 +1 @@ +The :mod:`smtpd` module was removed per the schedule in :pep:`594`. diff --git a/PCbuild/lib.pyproj b/PCbuild/lib.pyproj index 0556efe..e8c99f6 100644 --- a/PCbuild/lib.pyproj +++ b/PCbuild/lib.pyproj @@ -642,7 +642,6 @@ <Compile Include="shutil.py" /> <Compile Include="signal.py" /> <Compile Include="site.py" /> - <Compile Include="smtpd.py" /> <Compile Include="smtplib.py" /> <Compile Include="sndhdr.py" /> <Compile Include="socket.py" /> @@ -760,6 +759,7 @@ <Compile Include="test\sample_doctest_no_doctests.py" /> <Compile Include="test\seq_tests.py" /> <Compile Include="test\signalinterproctester.py" /> + <Compile Include="test\smtpd.py" /> <Compile Include="test\sortperf.py" /> <Compile Include="test\ssltests.py" /> <Compile Include="test\ssl_servers.py" /> @@ -1261,7 +1261,6 @@ <Compile Include="test\test_signal.py" /> <Compile Include="test\test_site.py" /> <Compile Include="test\test_slice.py" /> - <Compile Include="test\test_smtpd.py" /> <Compile Include="test\test_smtplib.py" /> <Compile Include="test\test_smtpnet.py" /> <Compile Include="test\test_sndhdr.py" /> diff --git a/Python/stdlib_module_names.h b/Python/stdlib_module_names.h index ff422e7..b281566 100644 --- a/Python/stdlib_module_names.h +++ b/Python/stdlib_module_names.h @@ -241,7 +241,6 @@ static const char* _Py_stdlib_module_names[] = { "shutil", "signal", "site", -"smtpd", "smtplib", "sndhdr", "socket", diff --git a/Tools/wasm/wasm_assets.py b/Tools/wasm/wasm_assets.py index 07450ac..695e6ff 100755 --- a/Tools/wasm/wasm_assets.py +++ b/Tools/wasm/wasm_assets.py @@ -77,7 +77,6 @@ OMIT_NETWORKING_FILES = ( "mailcap.py", "nntplib.py", "poplib.py", - "smtpd.py", "smtplib.py", "socketserver.py", "telnetlib.py", |