Repository: aio-libs/aiosmtpd
Pull request #59 (2017-03-29 03:11:22)

Handle long lines
I've found two test cases that cause the current aiosmtpd to fail with an unhandled exception. The issue is that StreamReader.readline raises a ValueError when its buffer overflows. The issue can arise both when reading a command line and reading a text line in DATA.

This PR sets a lower limit on StreamReader (1000 bytes per RFC 5321 4.5.3.1.6. + 1 byte for a leading transparent dot) to make the buffer overflow more frequent, and then it adds more complicated handling of StreamReader.readuntil in command line reading and DATA line reading. I couldn't get it working with StreamReader.readline, but since its implementation simply calls readuntil and masks some exceptions, it seems like we should use readuntil anyway.

I'm not sure exactly what to do in case of an early EOF -- should the code simply send some low-level SMTP error back to the client and return? Right now my implementation is (too) lenient about EOFs (materializing as asyncio.IncompleteReadError exceptions).

Mortal: After digging into the circumstances surrounding early EOF or connection reset from client I have found that cancelling the `_handle_client` coroutine on connection reset is the sanest thing to do, and then simply catching CancelledError at the appropriate points in `_handle_client`.

Currently the Python 3.4 build fails with stalling on Travis in controller.rst. I don't know why that is.

I've managed to add tests that keep coverage at 100% which was not an easy task. However, most of the tasks merely consist of opening an SMTP session and closing/resetting at odd times without asserting that the output of aiosmtpd is anything useful -- it only ensures that no exception is raised.
Mortal: This PR conflicts with #70 as they get rid of `self._connection_closed` in two different ways.
kozzztik: I dont see conflict here. Your PR completely is remove self._connection_closed usage, but #70 replace it with self.transport. You can still not use it, even it is replaced )
warsaw: In the interest of getting 1.0 before Pycon, and since this branch has conflicts, I'm going to defer it past 1.0
Mortal: Rebased on top of #115.
waynew: @Mortal do you have any interest in continuing this PR? There's apparently a conflict that needs to be resolved.
Mortal: I'll try to have a look at making the limit configurable on Tuesday.
waynew: @Mortal any update on this PR, too?
Mortal: As with the other PR, [help-wanted] is probably the best way forward.
pepoluan: I'm still thinking about the necessity of limiting `StreamReader` to N bytes (1001 in the PR).

Can we be sure that the underlying TCP stack will not buffer things until 1 page (4096 octets) and deliver things in bulk?

Yes, I know of the PSH flag, but IIRC it was never a requirement ("MUST"), just a recommendation ("SHOULD") for TCP stack implementation to actually deliver the data to the app when PSH flag is received.
CLAassistant: [![CLA assistant check](https://cla-assistant.io/pull/badge/not_signed)](https://cla-assistant.io/aio-libs/aiosmtpd?pullRequest=59) <br/>Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our [Contributor License Agreement](https://cla-assistant.io/aio-libs/aiosmtpd?pullRequest=59) before we can accept your contribution.<br/><sub>You have signed the CLA already but the status is still pending? Let us [recheck](https://cla-assistant.io/check/aio-libs/aiosmtpd?pullRequest=59) it.</sub>
Mortal: Hi @pepoluan, thanks for picking up this project! I don't really have the headspace these days to keep working on aiosmtpd, sorry. Can you confirm that the notification from CLAassistant is legit? I tried googling and I couldn't find any news anywhere that aiosmtpd specifically or aio-libs generally have started using CLAs. If you can confirm then I'll gladly sign a CLA.
pepoluan: Hi @Mortal , sorry for not answering sooner. This week had been a bit hectic in my life...

The CLA is legit, set by the maintainers of the aio-libs project, however there seems to be some... opinion about it. It's being discussed in #208 .

I myself am personally neutral about it, and I also personally think there's nothing wrong with the words, but it's still an interesting discussion over there (with links to other comments in other PRs).

I won't stop you from signing the CLA, but at the moment I also am not requiring anyone to sign the CLA. Just do what you think is right 😉

(For the record, I have already signed it.)
pepoluan: Alright, I'm picking up this PR. Here are some changes I'm implementing:

* For `smtp_DATA`, actually returns `500 Line too long` now
* For `smtp_DATA`, delays response of `500` or `552` until `<CR><LF>.<CR><LF>` is received
  (This seems to be implied by [RFC5321 § 4.2.5](https://tools.ietf.org/html/rfc5321#section-4.2.5))
* Line length limit is customizable (defaults to 1001 octets <== might break users?)
* Test cases that trigger `LimitOverrunError` in `smtp_DATA` needs to be change to catch `SMTPDataError`
  (Previously they just pass)

... and other things that I don't recall on the top of my head.

I'll do some testing, then push it as draft.
pepoluan: Ack! Forgot to commit `pragma:nobranch`... pushing again...
codecov[bot]: # [Codecov](https://codecov.io/gh/aio-libs/aiosmtpd/pull/59?src=pr&el=h1) Report
> Merging [#59](https://codecov.io/gh/aio-libs/aiosmtpd/pull/59?src=pr&el=desc) (21764fb) into [master](https://codecov.io/gh/aio-libs/aiosmtpd/commit/51ac147d378d7a11fa54e33de51b7b47201ae1f1?el=desc) (51ac147) will **not change** coverage.
> The diff coverage is `100.00%`.

[![Impacted file tree graph](https://codecov.io/gh/aio-libs/aiosmtpd/pull/59/graphs/tree.svg?width=650&height=150&src=pr&token=k9dc7NVG69)](https://codecov.io/gh/aio-libs/aiosmtpd/pull/59?src=pr&el=tree)

```diff
@@            Coverage Diff            @@
##            master       #59   +/-   ##
=========================================
  Coverage   100.00%   100.00%           
=========================================
  Files            5         5           
  Lines          934       969   +35     
  Branches       168       172    +4     
=========================================
+ Hits           934       969   +35     
```


| [Impacted Files](https://codecov.io/gh/aio-libs/aiosmtpd/pull/59?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [aiosmtpd/smtp.py](https://codecov.io/gh/aio-libs/aiosmtpd/pull/59/diff?src=pr&el=tree#diff-YWlvc210cGQvc210cC5weQ==) | `100.00% <100.00%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/aio-libs/aiosmtpd/pull/59?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/aio-libs/aiosmtpd/pull/59?src=pr&el=footer). Last update [51ac147...21764fb](https://codecov.io/gh/aio-libs/aiosmtpd/pull/59?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

pepoluan: Okay, I've tested on my systems, and they all pass with flying colors.

So I'm un-draft-ing this for review & merge.
pepoluan: Testing progress on my systems:
- [x] Windows 10 (via PyCharm runner): `(ALL)`
- [ ] Windows 10 (via PSCore 7.0.3): `(ALL)`

pepoluan: With your blessings @waynew , we can get this merged :-)
pepoluan: Come Monday I'll merge this if no objection.

Protocol violation (see #9 ) is a Serious Bug™ and should be fixed asap.

Patch:
diff --git a/aiosmtpd/docs/NEWS.rst b/aiosmtpd/docs/NEWS.rst
index 54679df..2acd503 100644
--- a/aiosmtpd/docs/NEWS.rst
+++ b/aiosmtpd/docs/NEWS.rst
@@ -2,12 +2,14 @@
  NEWS for aiosmtpd
 ===================
 
-1.2.3a1 (aiosmtpd-next)
-=======================
+1.2.3 (aiosmtpd-next)
+=====================
 
 Fixed/Improved
 --------------
-* handle_exception() no longer get called when the client disconnected. (Closes #127, #162)
+* ``handle_exception()`` no longer gets called when the client disconnected (Closes #127, #162)
+* Implement & enforce line-length-limit, thus becoming Compliant with RFC 5321 § 4.5.3.1.6
+* Delay all SMTP Status Code replies during ``DATA`` phase until the phase termination (Closes #9)
 
 
 1.2.2 (2020-11-08)
diff --git a/aiosmtpd/docs/smtp.rst b/aiosmtpd/docs/smtp.rst
index 38629eb..e8a0c08 100644
--- a/aiosmtpd/docs/smtp.rst
+++ b/aiosmtpd/docs/smtp.rst
@@ -103,7 +103,11 @@ override to provide additional responses.
 SMTP API
 ========
 
-.. class:: SMTP(handler, *, data_size_limit=33554432, enable_SMTPUTF8=False, decode_data=False, hostname=None, ident=None, tls_context=None, require_starttls=False, timeout=300, auth_required=False, auth_require_tls=True, auth_exclude_mechanism=None, auth_callback=None, loop=None)
+.. class:: SMTP(handler, *, data_size_limit=33554432, enable_SMTPUTF8=False, \
+   decode_data=False, hostname=None, ident=None, tls_context=None, \
+   require_starttls=False, timeout=300, auth_required=False, \
+   auth_require_tls=True, auth_exclude_mechanism=None, auth_callback=None, \
+   loop=None)
 
    *handler* is an instance of a :ref:`handler <handlers>` class.
 
@@ -158,6 +162,22 @@ SMTP API
    *loop* is the asyncio event loop to use.  If not given,
    :meth:`asyncio.new_event_loop()` is called to create the event loop.
 
+   .. py:attribute:: line_length_limit
+
+      The maximum line length, in octets (not characters; one UTF-8 character
+      may result in more than one octet).
+      Defaults to ``1001`` in compliance with
+      :rfc:`RFC 5321 § 4.5.3.1.6 <5321#section-4.5.3.1.6>`
+
+      .. attention::
+
+         This sets the *stream limit* of :meth:`asyncio.StreamReader.readuntil`,
+         thus impacting how the method works.
+         In previous versions of aiosmtpd, the limit is not set.
+         To return to the behavior of the previous versions, set
+         :attr:`line_length_limit` to ``2**16`` *before* instantiating the
+         :class:`SMTP` class.
+
    .. attribute:: event_handler
 
       The *handler* instance passed into the constructor.
diff --git a/aiosmtpd/smtp.py b/aiosmtpd/smtp.py
index 54d6cd0..54c8606 100644
--- a/aiosmtpd/smtp.py
+++ b/aiosmtpd/smtp.py
@@ -1,4 +1,5 @@
 import ssl
+import enum
 import socket
 import asyncio
 import logging
@@ -24,30 +25,44 @@ from typing import (
 from warnings import warn
 
 
-__version__ = '1.2.3a1'
-__ident__ = 'Python SMTP {}'.format(__version__)
-log = logging.getLogger('mail.log')
-
-
-DATA_SIZE_DEFAULT = 33554432
-EMPTYBYTES = b''
-NEWLINE = '\n'
-
+# region #### Custom Data Types #######################################################
 
 class _Missing:
     pass
 
 
-MISSING = _Missing()
+class _AuthMechAttr(NamedTuple):
+    method: "AuthMechanismType"
+    is_builtin: bool
+
+
+class _DataState(enum.Enum):
+    NOMINAL = enum.auto()
+    TOO_LONG = enum.auto()
+    TOO_MUCH = enum.auto()
 
 
 AuthMechanismType = Callable[["SMTP", List[str]], Awaitable[Any]]
 _TriStateType = Union[None, _Missing, bytes]
 
 
-class _AuthMechAttr(NamedTuple):
-    method: AuthMechanismType
-    is_builtin: bool
+# endregion
+
+
+# region #### Constant & Constant-likes ###############################################
+
+__version__ = '1.2.3a2'
+__ident__ = 'Python SMTP {}'.format(__version__)
+log = logging.getLogger('mail.log')
+
+
+DATA_SIZE_DEFAULT = 2**25  # Where does this number come from, I wonder...
+EMPTY_BARR = bytearray()
+EMPTYBYTES = b''
+MISSING = _Missing()
+NEWLINE = '\n'
+
+# endregion
 
 
 @public
@@ -98,29 +113,37 @@ class SMTP(asyncio.StreamReaderProtocol):
     command_size_limit = 512
     command_size_limits = collections.defaultdict(
         lambda x=command_size_limit: x)
-
-    def __init__(self, handler,
-                 *,
-                 data_size_limit=DATA_SIZE_DEFAULT,
-                 enable_SMTPUTF8=False,
-                 decode_data=False,
-                 hostname=None,
-                 ident=None,
-                 tls_context=None,
-                 require_starttls=False,
-                 timeout=300,
-                 auth_required=False,
-                 auth_require_tls=True,
-                 auth_exclude_mechanism: Optional[Iterable[str]] = None,
-                 auth_callback: Callable[[str, bytes, bytes], bool] = None,
-                 loop=None):
+    line_length_limit = 1001
+    """Maximum line length according to RFC 5321 s 4.5.3.1.6"""
+    # The number comes from this calculation:
+    # (RFC 5322 s 2.1.1 + RFC 6532 s 3.4) 998 octets + CRLF = 1000 octets
+    # (RFC 5321 s 4.5.3.1.6) 1000 octets + "transparent dot" = 1001 octets
+
+    def __init__(
+            self, handler,
+            *,
+            data_size_limit=DATA_SIZE_DEFAULT,
+            enable_SMTPUTF8=False,
+            decode_data=False,
+            hostname=None,
+            ident=None,
+            tls_context=None,
+            require_starttls=False,
+            timeout=300,
+            auth_required=False,
+            auth_require_tls=True,
+            auth_exclude_mechanism: Optional[Iterable[str]] = None,
+            auth_callback: Callable[[str, bytes, bytes], bool] = None,
+            loop=None,
+    ):
         self.__ident__ = ident or __ident__
         self.loop = loop if loop else make_loop()
         super().__init__(
-            asyncio.StreamReader(loop=self.loop),
+            asyncio.StreamReader(loop=self.loop, limit=self.line_length_limit),
             client_connected_cb=self._client_connected_cb,
             loop=self.loop)
         self.event_handler = handler
+        assert data_size_limit is None or isinstance(data_size_limit, int)
         self.data_size_limit = data_size_limit
         self.enable_SMTPUTF8 = enable_SMTPUTF8
         self._decode_data = decode_data
@@ -300,9 +323,24 @@ class SMTP(asyncio.StreamReaderProtocol):
         log.info('%r handling connection', self.session.peer)
         await self.push('220 {} {}'.format(self.hostname, self.__ident__))
         while self.transport is not None:   # pragma: nobranch
-            # XXX Put the line limit stuff into the StreamReader?
             try:
-                line: bytes = await self._reader.readline()
+                try:
+                    line = await self._reader.readuntil()
+                except asyncio.LimitOverrunError as error:
+                    # Line too long. Read until end of line before sending 500.
+                    await self._reader.read(error.consumed)
+                    while True:
+                        try:
+                            await self._reader.readuntil()
+                            break
+                        except asyncio.LimitOverrunError as e:
+                            # Line is even longer...
+                            await self._reader.read(e.consumed)
+                            continue
+                    # Now that we have read a full line from the client,
+                    # send error response and read the next command line.
+                    await self.push('500 Command line too long')
+                    continue
                 log.debug('_handle_client readline: %s', line)
                 # XXX this rstrip may not completely preserve old behavior.
                 line = line.rstrip(b'\r\n')
@@ -348,7 +386,7 @@ class SMTP(asyncio.StreamReaderProtocol):
                           if self.session.extended_smtp
                           else self.command_size_limit)
                 if len(line) > max_sz:
-                    await self.push('500 Error: line too long')
+                    await self.push('500 Command line too long')
                     continue
                 if not self._tls_handshake_okay and command != 'QUIT':
                     await self.push(
@@ -869,42 +907,87 @@ class SMTP(asyncio.StreamReaderProtocol):
         if arg:
             await self.push('501 Syntax: DATA')
             return
+
         await self.push('354 End data with <CR><LF>.<CR><LF>')
-        data = []
-        num_bytes = 0
-        size_exceeded = False
+        data: List[bytearray] = []
+
+        num_bytes: int = 0
+        limit: Optional[int] = self.data_size_limit
+        line_fragments: List[bytes] = []
+        state: _DataState = _DataState.NOMINAL
         while self.transport is not None:           # pragma: nobranch
+            # Since eof_received cancels this coroutine,
+            # readuntil() can never raise asyncio.IncompleteReadError.
             try:
-                line = await self._reader.readline()
+                line: bytes = await self._reader.readuntil()
                 log.debug('DATA readline: %s', line)
+                assert line.endswith(b'\n')
             except asyncio.CancelledError:
                 # The connection got reset during the DATA command.
                 log.info('Connection lost during DATA')
                 self._writer.close()
                 raise
-            if line == b'.\r\n':
+            except asyncio.LimitOverrunError as e:
+                # The line exceeds StreamReader's "stream limit".
+                # Delay SMTP Status Code sending until data receive is complete
+                # This seems to be implied in RFC 5321 § 4.2.5
+                if state == _DataState.NOMINAL:
+                    # Transition to TOO_LONG only if we haven't gone TOO_MUCH yet
+                    state = _DataState.TOO_LONG
+                # Discard data immediately to prevent memory pressure
+                data *= 0
+                # Drain the stream anyways
+                line = await self._reader.read(e.consumed)
+                assert not line.endswith(b'\n')
+            # A lone dot in a line signals the end of DATA.
+            if not line_fragments and line == b'.\r\n':
                 break
             num_bytes += len(line)
-            if (not size_exceeded
-                    and self.data_size_limit
-                    and num_bytes > self.data_size_limit):
-                size_exceeded = True
+            if state == _DataState.NOMINAL and limit and num_bytes > limit:
+                # Delay SMTP Status Code sending until data receive is complete
+                # This seems to be implied in RFC 5321 § 4.2.5
+                state = _DataState.TOO_MUCH
+                # Discard data immediately to prevent memory pressure
+                data *= 0
+            line_fragments.append(line)
+            if line.endswith(b'\n'):
+                # Record data only if state is "NOMINAL"
+                if state == _DataState.NOMINAL:
+                    line = EMPTY_BARR.join(line_fragments)
+                    if len(line) > self.line_length_limit:
+                        # Theoretically we shouldn't reach this place. But it's always
+                        # good to practice DEFENSIVE coding.
+                        state = _DataState.TOO_LONG
+                        # Discard data immediately to prevent memory pressure
+                        data *= 0
+                    else:
+                        data.append(EMPTY_BARR.join(line_fragments))
+                line_fragments *= 0
+
+        # Day of reckoning! Let's take care of those out-of-nominal situations
+        if state != _DataState.NOMINAL:
+            if state == _DataState.TOO_LONG:
+                await self.push("500 Line too long (see RFC5321 4.5.3.1.6)")
+            elif state == _DataState.TOO_MUCH:  # pragma: nobranch
                 await self.push('552 Error: Too much mail data')
-            if not size_exceeded:
-                data.append(line)
-        if size_exceeded:
             self._set_post_data_state()
             return
+
+        # If unfinished_line is non-empty, then the connection was closed.
+        assert not line_fragments
+
         # Remove extraneous carriage returns and de-transparency
         # according to RFC 5321, Section 4.5.2.
-        for i, text in enumerate(data):
+        for text in data:
             if text.startswith(b'.'):
-                data[i] = text[1:]
-        content = original_content = EMPTYBYTES.join(data)
+                del text[0]
+        original_content: bytes = EMPTYBYTES.join(data)
+        # Discard data immediately to prevent memory pressure
+        data *= 0
+
         if self._decode_data:
             if self.enable_SMTPUTF8:
-                content = original_content.decode(
-                    'utf-8', errors='surrogateescape')
+                content = original_content.decode('utf-8', errors='surrogateescape')
             else:
                 try:
                     content = original_content.decode('ascii', errors='strict')
@@ -914,8 +997,11 @@ class SMTP(asyncio.StreamReaderProtocol):
                     # but the client ignores that and sends non-ascii anyway.
                     await self.push('500 Error: strict ASCII mode')
                     return
+        else:
+            content = original_content
         self.envelope.content = content
         self.envelope.original_content = original_content
+
         # Call the new API first if it's implemented.
         if "DATA" in self._handle_hooks:
             status = await self._call_handler_hook('DATA')


Test patch:
diff --git a/aiosmtpd/testing/helpers.py b/aiosmtpd/testing/helpers.py
index 6999d32..8b3af86 100644
--- a/aiosmtpd/testing/helpers.py
+++ b/aiosmtpd/testing/helpers.py
@@ -1,6 +1,7 @@
 """Testing helpers."""
 
 import sys
+import select
 import socket
 import struct
 import asyncio
@@ -8,6 +9,7 @@ import logging
 import warnings
 
 from contextlib import ExitStack
+from typing import List
 from unittest import TestCase
 from unittest.mock import patch
 
@@ -82,3 +84,19 @@ SUPPORTED_COMMANDS_TLS: bytes = (
 )
 
 SUPPORTED_COMMANDS_NOTLS = SUPPORTED_COMMANDS_TLS.replace(b" STARTTLS", b"")
+
+
+def send_recv(
+        sock: socket.socket, data: bytes, end: bytes = b"\r\n", timeout=0.1
+) -> bytes:
+    sock.send(data + end)
+    slist = [sock]
+    result: List[bytes] = []
+    while True:
+        read_s, _, _ = select.select(slist, [], [], timeout)
+        if read_s:
+            # We can use sock instead of read_s because slist only contains sock
+            result.append(sock.recv(1024))
+        else:
+            break
+    return b"".join(result)
diff --git a/aiosmtpd/tests/test_main.py b/aiosmtpd/tests/test_main.py
index 0581080..ee544ed 100644
--- a/aiosmtpd/tests/test_main.py
+++ b/aiosmtpd/tests/test_main.py
@@ -72,24 +72,21 @@ class TestMain(unittest.TestCase):
             mock.assert_called_with(pwd.getpwnam('nobody').pw_uid)
 
     @unittest.skipIf(pwd is None, 'No pwd module available')
-    def test_setuid_permission_error(self):
-        mock = self.resources.enter_context(
-            patch('os.setuid', side_effect=PermissionError))
-        stderr = StringIO()
-        self.resources.enter_context(patch('sys.stderr', stderr))
+    @patch('os.setuid', side_effect=PermissionError)
+    @patch('sys.stderr', new_callable=StringIO)
+    def test_setuid_permission_error(self, mock_err, mock_setuid):
         with self.assertRaises(SystemExit) as cm:
             main(args=())
         self.assertEqual(cm.exception.code, 1)
-        mock.assert_called_with(pwd.getpwnam('nobody').pw_uid)
+        mock_setuid.assert_called_with(pwd.getpwnam('nobody').pw_uid)
         self.assertEqual(
-            stderr.getvalue(),
+            mock_err.getvalue(),
             'Cannot setuid "nobody"; try running with -n option.\n')
 
     @unittest.skipIf(pwd is None, 'No pwd module available')
-    def test_setuid_no_pwd_module(self):
-        self.resources.enter_context(patch('aiosmtpd.main.pwd', None))
-        stderr = StringIO()
-        self.resources.enter_context(patch('sys.stderr', stderr))
+    @patch('aiosmtpd.main.pwd', None)  # Returns no object, so needs no arg
+    @patch('sys.stderr', new_callable=StringIO)
+    def test_setuid_no_pwd_module(self, mock_err):
         with self.assertRaises(SystemExit) as cm:
             main(args=())
         self.assertEqual(cm.exception.code, 1)
@@ -100,29 +97,25 @@ class TestMain(unittest.TestCase):
         # the string DOES appear in stderr, just buried.
         self.assertIn(
             'Cannot import module "pwd"; try running with -n option.\n',
-            stderr.getvalue(),
+            mock_err.getvalue(),
         )
 
     @unittest.skipUnless(has_setuid, 'setuid is unvailable')
-    def test_n(self):
-        self.resources.enter_context(patch('aiosmtpd.main.pwd', None))
-        self.resources.enter_context(
-            patch('os.setuid', side_effect=PermissionError))
-        # Just to short-circuit the main() function.
-        self.resources.enter_context(
-            patch('aiosmtpd.main.partial', side_effect=RuntimeError))
+    # Just to short-circuit the main() function.
+    @patch('aiosmtpd.main.partial', side_effect=RuntimeError)
+    @patch('os.setuid', side_effect=PermissionError)
+    @patch('aiosmtpd.main.pwd', None)  # Returns no object, so needs no arg
+    def test_n(self, mock_setuid, mock_partial):
         # Getting the RuntimeError means that a SystemExit was never
         # triggered in the setuid section.
         self.assertRaises(RuntimeError, main, ('-n',))
 
     @unittest.skipUnless(has_setuid, 'setuid is unvailable')
-    def test_nosetuid(self):
-        self.resources.enter_context(patch('aiosmtpd.main.pwd', None))
-        self.resources.enter_context(
-            patch('os.setuid', side_effect=PermissionError))
-        # Just to short-circuit the main() function.
-        self.resources.enter_context(
-            patch('aiosmtpd.main.partial', side_effect=RuntimeError))
+    # Just to short-circuit the main() function.
+    @patch('aiosmtpd.main.partial', side_effect=RuntimeError)
+    @patch('os.setuid', side_effect=PermissionError)
+    @patch('aiosmtpd.main.pwd', None)  # Returns no object, so needs no arg
+    def test_nosetuid(self, mock_setuid, mock_partial):
         # Getting the RuntimeError means that a SystemExit was never
         # triggered in the setuid section.
         self.assertRaises(RuntimeError, main, ('--nosetuid',))
@@ -227,25 +220,21 @@ class TestParseArgs(unittest.TestCase):
         usage_lines = stderr.getvalue().splitlines()
         self.assertEqual(usage_lines[-1][-24:], 'Invalid port number: foo')
 
-    def test_version(self):
-        stdout = StringIO()
-        with ExitStack() as resources:
-            resources.enter_context(patch('sys.stdout', stdout))
-            resources.enter_context(patch('aiosmtpd.main.PROGRAM', 'smtpd'))
-            cm = resources.enter_context(self.assertRaises(SystemExit))
+    @patch('aiosmtpd.main.PROGRAM', 'smtpd')  # does NOT return a mock object! so...
+    @patch('sys.stdout', new_callable=StringIO)
+    def test_version(self, mock_out):  # ...so, only one arg here
+        with self.assertRaises(SystemExit) as cm:
             parseargs(('--version',))
-            self.assertEqual(cm.exception.code, 0)
-        self.assertEqual(stdout.getvalue(), 'smtpd {}\n'.format(__version__))
-
-    def test_v(self):
-        stdout = StringIO()
-        with ExitStack() as resources:
-            resources.enter_context(patch('sys.stdout', stdout))
-            resources.enter_context(patch('aiosmtpd.main.PROGRAM', 'smtpd'))
-            cm = resources.enter_context(self.assertRaises(SystemExit))
+        self.assertEqual(cm.exception.code, 0)
+        self.assertEqual(mock_out.getvalue(), 'smtpd {}\n'.format(__version__))
+
+    @patch('aiosmtpd.main.PROGRAM', 'smtpd')  # does NOT return a mock object! so...
+    @patch('sys.stdout', new_callable=StringIO)
+    def test_v(self, mock_out):  # ...so, only one arg here
+        with self.assertRaises(SystemExit) as cm:
             parseargs(('-v',))
-            self.assertEqual(cm.exception.code, 0)
-        self.assertEqual(stdout.getvalue(), 'smtpd {}\n'.format(__version__))
+        self.assertEqual(cm.exception.code, 0)
+        self.assertEqual(mock_out.getvalue(), 'smtpd {}\n'.format(__version__))
 
 
 class TestSigint(unittest.TestCase):
diff --git a/aiosmtpd/tests/test_smtp.py b/aiosmtpd/tests/test_smtp.py
index 53aa4ed..e29da7c 100644
--- a/aiosmtpd/tests/test_smtp.py
+++ b/aiosmtpd/tests/test_smtp.py
@@ -4,21 +4,26 @@ import time
 import socket
 import asyncio
 import unittest
+import warnings
 
 from aiosmtpd.controller import Controller
 from aiosmtpd.handlers import Sink
-from aiosmtpd.smtp import MISSING, SMTP as Server, __ident__ as GREETING
+from aiosmtpd.smtp import (
+    MISSING, Session as SMTPSess, SMTP as Server, __ident__ as GREETING
+)
 from aiosmtpd.testing.helpers import (
     SUPPORTED_COMMANDS_NOTLS,
     assert_auth_invalid,
     assert_auth_required,
     assert_auth_success,
     reset_connection,
+    send_recv,
 )
 from base64 import b64encode
 from contextlib import ExitStack
 from smtplib import (
     SMTP, SMTPDataError, SMTPResponseException, SMTPServerDisconnected)
+from typing import ContextManager, cast
 from unittest.mock import Mock, PropertyMock, patch
 
 CRLF = '\r\n'
@@ -39,8 +44,7 @@ class DecodingController(Controller):
 
 
 class PeekerHandler:
-    def __init__(self):
-        self.session = None
+    session: SMTPSess = None
 
     async def handle_MAIL(
             self, server, session, envelope, address, mail_options
@@ -771,7 +775,20 @@ class TestSMTP(unittest.TestCase):
         with SMTP(*self.address) as client:
             code, response = client.docmd('a' * 513)
             self.assertEqual(code, 500)
-            self.assertEqual(response, b'Error: line too long')
+            self.assertEqual(response, b'Command line too long')
+
+    def test_way_too_long_command(self):
+        with SMTP(*self.address) as client:
+            # Send a very large string to ensure it is broken
+            # into several packets, which hits the inner
+            # LimitOverrunError code path in _handle_client.
+            client.send('a' * 1000000)
+            code, response = client.docmd('a' * 1001)
+            self.assertEqual(code, 500)
+            self.assertEqual(response, b'Command line too long')
+            code, response = client.docmd('NOOP')
+            self.assertEqual(code, 250)
+            self.assertEqual(response, b'OK')
 
     def test_unknown_command(self):
         with SMTP(*self.address) as client:
@@ -881,7 +898,9 @@ class TestSMTP(unittest.TestCase):
             code, response = client.docmd('AUTH PLAIN')
             self.assertEqual(code, 334)
             self.assertEqual(response, b'')
-            code, response = client.docmd('*')
+            # Suppress log.warning()
+            with patch("logging.Logger.warning"):
+                code, response = client.docmd('*')
             self.assertEqual(code, 501)
             self.assertEqual(response, b'Auth aborted')
 
@@ -967,7 +986,8 @@ class TestSMTPAuth(unittest.TestCase):
             self.assertEqual(auth_peeker.login, None)
             self.assertEqual(auth_peeker.password, None)
             code, response = client.mail("alice@example.com")
-            self.assertEqual(self.handler.session.login_data, b"")
+            sess: SMTPSess = self.handler.session
+            self.assertEqual(sess.login_data, b"")
 
     def test_auth_login_null_credential(self):
         with SMTP(*self.address) as client:
@@ -983,7 +1003,8 @@ class TestSMTPAuth(unittest.TestCase):
             self.assertEqual(auth_peeker.login, None)
             self.assertEqual(auth_peeker.password, None)
             code, response = client.mail("alice@example.com")
-            self.assertEqual(self.handler.session.login_data, b"")
+            sess: SMTPSess = self.handler.session
+            self.assertEqual(sess.login_data, b"")
 
     def test_auth_login_abort_login(self):
         with SMTP(*self.address) as client:
@@ -991,7 +1012,9 @@ class TestSMTPAuth(unittest.TestCase):
             code, response = client.docmd("AUTH LOGIN")
             self.assertEqual(code, 334)
             self.assertEqual(response, b"VXNlciBOYW1lAA==")
-            code, response = client.docmd('*')
+            # Suppress log.warning()
+            with patch("logging.Logger.warning"):
+                code, response = client.docmd('*')
             self.assertEqual(code, 501)
             self.assertEqual(response, b"Auth aborted")
 
@@ -1005,7 +1028,9 @@ class TestSMTPAuth(unittest.TestCase):
             code, response = client.docmd('=')
             self.assertEqual(code, 334)
             self.assertEqual(response, b"UGFzc3dvcmQA")
-            code, response = client.docmd('*')
+            # Suppress log.warning()
+            with patch("logging.Logger.warning"):
+                code, response = client.docmd('*')
             self.assertEqual(code, 501)
             self.assertEqual(response, b"Auth aborted")
 
@@ -1028,10 +1053,16 @@ class TestSMTPAuth(unittest.TestCase):
 class TestRequiredAuthentication(unittest.TestCase):
     def setUp(self):
         controller = RequiredAuthDecodingController(Sink)
-        controller.start()
         self.addCleanup(controller.stop)
+        controller.start()
         self.address = (controller.hostname, controller.port)
 
+        self.resource = ExitStack()
+        self.addCleanup(self.resource.close)
+        # Suppress auth_req_but_no_tls warning
+        self.resource.enter_context(cast(ContextManager, warnings.catch_warnings()))
+        warnings.simplefilter("ignore", category=UserWarning)
+
     def test_help_unauthenticated(self):
         with SMTP(*self.address) as client:
             code, response = client.docmd('HELP')
@@ -1316,6 +1347,138 @@ Testing
             self.assertEqual(cm.exception.smtp_error,
                              b'Error: Too much mail data')
 
+    def test_too_long_body_delay_error(self):
+        size, sock = 20, None
+
+        cont = Controller(Sink(), hostname="localhost",
+                          server_kwargs={"data_size_limit": size})
+        self.addCleanup(cont.stop)
+        cont.start()
+
+        with socket.socket() as sock:
+            sock.connect((cont.hostname, cont.port))
+            rslt = send_recv(sock, b"EHLO example.com")
+            self.assertTrue(rslt.startswith(b"220"))
+            rslt = send_recv(sock, b"MAIL FROM: <anne@example.com>")
+            self.assertTrue(rslt.startswith(b"250"))
+            rslt = send_recv(sock, b"RCPT TO: <bruce@example.com>")
+            self.assertTrue(rslt.startswith(b"250"))
+            rslt = send_recv(sock, b"DATA")
+            self.assertTrue(rslt.startswith(b"354"))
+            rslt = send_recv(sock, b"a" * (size + 3))
+            # Must NOT receive status code here even if data is too much
+            self.assertEqual(b"", rslt)
+            rslt = send_recv(sock, b"\r\n.")
+            # *NOW* we must receive status code
+            self.assertEqual(b"552 Error: Too much mail data\r\n", rslt)
+
+    def test_data_line_too_long(self):
+        handler = ReceivingHandler()
+        controller = NoDecodeController(handler)
+        controller.start()
+        self.addCleanup(controller.stop)
+        with SMTP(controller.hostname, controller.port) as client:
+            client.helo('example.com')
+            mail = b'\r\n'.join([b'a' * 5555] * 3)
+            with self.assertRaises(SMTPDataError) as cm:
+                client.sendmail('anne@example.com', ['bart@example.com'], mail)
+        self.assertEqual(cm.exception.smtp_code, 500)
+        self.assertEqual(cm.exception.smtp_error,
+                         b'Line too long (see RFC5321 4.5.3.1.6)')
+
+    def test_too_long_line_delay_error(self):
+        sock = None
+
+        cont = Controller(Sink(), hostname="localhost")
+        self.addCleanup(cont.stop)
+        cont.start()
+
+        with socket.socket() as sock:
+            sock.connect((cont.hostname, cont.port))
+            rslt = send_recv(sock, b"EHLO example.com")
+            self.assertTrue(rslt.startswith(b"220"))
+            rslt = send_recv(sock, b"MAIL FROM: <anne@example.com>")
+            self.assertTrue(rslt.startswith(b"250"))
+            rslt = send_recv(sock, b"RCPT TO: <bruce@example.com>")
+            self.assertTrue(rslt.startswith(b"250"))
+            rslt = send_recv(sock, b"DATA")
+            self.assertTrue(rslt.startswith(b"354"))
+            rslt = send_recv(sock, b"a" * (Server.line_length_limit + 3))
+            # Must NOT receive status code here even if data is too much
+            self.assertEqual(b"", rslt)
+            rslt = send_recv(sock, b"\r\n.")
+            # *NOW* we must receive status code
+            self.assertEqual(b"500 Line too long (see RFC5321 4.5.3.1.6)\r\n", rslt)
+
+    def test_too_long_lines_then_too_long_body(self):
+        # If "too long line" state was reached before "too much data" happens,
+        # SMTP should respond with '500' instead of '552'
+        size = 2000
+        controller = SizedController(Sink(), size=size)
+        controller.start()
+        self.addCleanup(controller.stop)
+        with SMTP(controller.hostname, controller.port) as client:
+            client.helo('example.com')
+            mail = '\r\n'.join(['z' * (size - 1)] * 2)
+            with self.assertRaises(SMTPResponseException) as cm:
+                client.sendmail('anne@example.com', ['bart@example.com'], mail)
+        self.assertEqual(cm.exception.smtp_code, 500)
+        self.assertEqual(cm.exception.smtp_error,
+                         b'Line too long (see RFC5321 4.5.3.1.6)')
+
+    def test_too_long_body_then_too_long_lines(self):
+        # If "too much mail" state was reached before "too long line" gets received,
+        # SMTP should respond with '552' instead of '500'
+        controller = SizedController(Sink(), size=700)
+        controller.start()
+        self.addCleanup(controller.stop)
+        with SMTP(controller.hostname, controller.port) as client:
+            client.helo('example.com')
+            mail = '\r\n'.join(['z' * 76] * 10 + ["a" * 1100] * 2)
+            with self.assertRaises(SMTPResponseException) as cm:
+                client.sendmail('anne@example.com', ['bart@example.com'], mail)
+            self.assertEqual(cm.exception.smtp_code, 552)
+            self.assertEqual(cm.exception.smtp_error,
+                             b'Error: Too much mail data')
+
+    def test_long_line_double_count(self):
+        controller = SizedController(Sink(), size=10000)
+        # With a read limit of 1001 bytes in aiosmtp.SMTP, asyncio.StreamReader
+        # returns too-long lines of length up to 2002 bytes.
+        # This test ensures that bytes in partial lines are only counted once.
+        # If the implementation has a double-counting bug, then a message of
+        # 9998 bytes + CRLF will raise SMTPResponseException.
+        controller.start()
+        self.addCleanup(controller.stop)
+        with SMTP(controller.hostname, controller.port) as client:
+            client.helo('example.com')
+            mail = 'z' * 9998
+            with self.assertRaises(SMTPDataError) as cm:
+                client.sendmail('anne@example.com', ['bart@example.com'], mail)
+        self.assertEqual(cm.exception.smtp_code, 500)
+        self.assertEqual(cm.exception.smtp_error,
+                         b'Line too long (see RFC5321 4.5.3.1.6)')
+
+    @patch("aiosmtpd.smtp.EMPTY_BARR")
+    def test_long_line_leak(self, mock_ebarr):
+        # Simulates situation where readuntil() does not raise LimitOverrunError,
+        # but somehow the line_fragments when join()ed resulted in a too-long line
+
+        # Hijack EMPTY_BARR.join() to return a bytes object that's definitely too long
+        mock_ebarr.join.return_value = (b"a" * 1010)
+
+        controller = Controller(Sink())
+        self.addCleanup(controller.stop)
+        controller.start()
+        with SMTP(controller.hostname, controller.port) as client:
+            client.helo('example.com')
+            mail = 'z' * 72  # Make sure this is small and definitely within limits
+            with self.assertRaises(SMTPDataError) as cm:
+                client.sendmail('anne@example.com', ['bart@example.com'], mail)
+        self.assertEqual(cm.exception.smtp_code, 500)
+        self.assertEqual(cm.exception.smtp_error,
+                         b'Line too long (see RFC5321 4.5.3.1.6)')
+
     def test_dots_escaped(self):
         handler = ReceivingHandler()
         controller = DecodingController(handler)
@@ -1328,84 +1491,75 @@ Testing
             self.assertEqual(len(handler.box), 1)
             self.assertEqual(handler.box[0].content, 'Test\r\n.\r\nmail\r\n')
 
-    def test_unexpected_errors(self):
+    # Suppress logging to the console during the tests.  Depending on
+    # timing, the exception may or may not be logged.
+    @patch("logging.Logger.exception")
+    def test_unexpected_errors(self, mock_logex):
         handler = ErroringHandler()
         controller = ErrorController(handler)
         controller.start()
         self.addCleanup(controller.stop)
-        with ExitStack() as resources:
-            # Suppress logging to the console during the tests.  Depending on
-            # timing, the exception may or may not be logged.
-            resources.enter_context(patch('aiosmtpd.smtp.log.exception'))
-            client = resources.enter_context(
-                SMTP(controller.hostname, controller.port))
-            code, response = client.helo('example.com')
+        with SMTP(controller.hostname, controller.port) as client:
+            code, mesg = client.helo('example.com')
         self.assertEqual(code, 500)
-        self.assertEqual(response, b'ErroringHandler handling error')
+        self.assertEqual(mesg, b'ErroringHandler handling error')
         self.assertIsInstance(handler.error, ValueError)
 
-    def test_unexpected_errors_unhandled(self):
+    # Suppress logging to the console during the tests.  Depending on
+    # timing, the exception may or may not be logged.
+    @patch("logging.Logger.exception")
+    def test_unexpected_errors_unhandled(self, mock_logex):
         handler = Sink()
         handler.error = None
         controller = ErrorController(handler)
         controller.start()
         self.addCleanup(controller.stop)
-        with ExitStack() as resources:
-            # Suppress logging to the console during the tests.  Depending on
-            # timing, the exception may or may not be logged.
-            resources.enter_context(patch('aiosmtpd.smtp.log.exception'))
-            client = resources.enter_context(
-                SMTP(controller.hostname, controller.port))
-            code, response = client.helo('example.com')
+        with SMTP(controller.hostname, controller.port) as client:
+            code, mesg = client.helo('example.com')
         self.assertEqual(code, 500)
-        self.assertEqual(response, b'Error: (ValueError) test')
+        self.assertEqual(mesg, b'Error: (ValueError) test')
         # handler.error did not change because the handler does not have a
         # handle_exception() method.
         self.assertIsNone(handler.error)
 
-    def test_unexpected_errors_custom_response(self):
+    # Suppress logging to the console during the tests.  Depending on
+    # timing, the exception may or may not be logged.
+    @patch("logging.Logger.exception")
+    def test_unexpected_errors_custom_response(self, mock_logex):
         handler = ErroringHandlerCustomResponse()
         controller = ErrorController(handler)
         controller.start()
         self.addCleanup(controller.stop)
-        with ExitStack() as resources:
-            # Suppress logging to the console during the tests.  Depending on
-            # timing, the exception may or may not be logged.
-            resources.enter_context(patch('aiosmtpd.smtp.log.exception'))
-            client = resources.enter_context(
-                SMTP(controller.hostname, controller.port))
-            code, response = client.helo('example.com')
+        with SMTP(controller.hostname, controller.port) as client:
+            code, mesg = client.helo('example.com')
         self.assertEqual(code, 451)
-        self.assertEqual(response, b'Temporary error: (ValueError) test')
+        self.assertEqual(mesg, b'Temporary error: (ValueError) test')
         self.assertIsInstance(handler.error, ValueError)
 
-    def test_exception_handler_exception(self):
+    # Suppress logging to the console during the tests.  Depending on
+    # timing, the exception may or may not be logged.
+    @patch("logging.Logger.exception")
+    def test_exception_handler_exception(self, mock_logex):
         handler = ErroringErrorHandler()
         controller = ErrorController(handler)
         controller.start()
         self.addCleanup(controller.stop)
-        with ExitStack() as resources:
-            # Suppress logging to the console during the tests.  Depending on
-            # timing, the exception may or may not be logged.
-            resources.enter_context(patch('aiosmtpd.smtp.log.exception'))
-            client = resources.enter_context(
-                SMTP(controller.hostname, controller.port))
-            code, response = client.helo('example.com')
+        with SMTP(controller.hostname, controller.port) as client:
+            code, mesg = client.helo('example.com')
         self.assertEqual(code, 500)
-        self.assertEqual(response,
-                         b'Error: (ValueError) ErroringErrorHandler test')
+        self.assertEqual(mesg, b'Error: (ValueError) ErroringErrorHandler test')
         self.assertIsInstance(handler.error, ValueError)
 
     def test_exception_handler_multiple_connections_lost(self):
         handler = ErroringHandlerConnectionLost()
         controller = Controller(handler)
-        controller.start()
         self.addCleanup(controller.stop)
+        controller.start()
         with SMTP(controller.hostname, controller.port) as client1:
-            code, response = client1.ehlo('example.com')
+            code, mesg = client1.ehlo('example.com')
             self.assertEqual(code, 250)
             with SMTP(controller.hostname, controller.port) as client2:
-                code, response = client2.ehlo('example.com')
+                code, mesg = client2.ehlo('example.com')
                 self.assertEqual(code, 250)
                 with self.assertRaises(SMTPServerDisconnected) as cm:
                     mail = CRLF.join(['Test', '.', 'mail'])
@@ -1418,27 +1572,23 @@ Testing
                 # At this point connection should be down
                 with self.assertRaises(SMTPServerDisconnected) as cm:
                     client2.mail("alice@example.com")
-                self.assertEqual(
-                    "please run connect() first",
-                    str(cm.exception))
+                self.assertEqual("please run connect() first", str(cm.exception))
             # client1 shouldn't be affected.
-            code, response = client1.mail("alice@example.com")
+            code, mesg = client1.mail("alice@example.com")
             self.assertEqual(code, 250)
 
-    def test_exception_handler_undescribable(self):
+    # Suppress logging to the console during the tests.  Depending on
+    # timing, the exception may or may not be logged.
+    @patch("logging.Logger.exception")
+    def test_exception_handler_undescribable(self, mock_logex):
         handler = UndescribableErrorHandler()
         controller = ErrorController(handler)
         controller.start()
         self.addCleanup(controller.stop)
-        with ExitStack() as resources:
-            # Suppress logging to the console during the tests.  Depending on
-            # timing, the exception may or may not be logged.
-            resources.enter_context(patch('aiosmtpd.smtp.log.exception'))
-            client = resources.enter_context(
-                SMTP(controller.hostname, controller.port))
-            code, response = client.helo('example.com')
+        with SMTP(controller.hostname, controller.port) as client:
+            code, mesg = client.helo('example.com')
         self.assertEqual(code, 500)
-        self.assertEqual(response, b'Error: Cannot describe error')
+        self.assertEqual(mesg, b'Error: Cannot describe error')
         self.assertIsInstance(handler.error, ValueError)
 
     def test_bad_encodings(self):
@@ -1524,10 +1674,14 @@ class TestClientCrash(unittest.TestCase):
 
     def test_connection_reset_during_DATA(self):
         with SMTP(*self.address) as client:
-            client.helo('example.com')
-            client.docmd('MAIL FROM: <anne@example.com>')
-            client.docmd('RCPT TO: <bart@example.com>')
-            client.docmd('DATA')
+            code, response = client.helo('example.com')
+            self.assertEqual(code, 250)
+            code, response = client.docmd('MAIL FROM: <anne@example.com>')
+            self.assertEqual(code, 250)
+            code, response = client.docmd('RCPT TO: <anne@example.com>')
+            self.assertEqual(code, 250)
+            code, response = client.docmd('DATA')
+            self.assertEqual(code, 354)
             # Start sending the DATA but reset the connection before that
             # completes, i.e. before the .\r\n
             client.send(b'From: <anne@example.com>')
@@ -1549,12 +1703,22 @@ class TestClientCrash(unittest.TestCase):
             # server just hung.
             self.assertRaises(SMTPServerDisconnected, client.noop)
 
+    def test_connection_reset_in_long_command(self):
+        with SMTP(*self.address) as client:
+            client.send('F' + 5555 * 'O')  # without CRLF
+            reset_connection(client)
+
     def test_close_in_command(self):
         with SMTP(*self.address) as client:
             # Don't include the CRLF.
             client.send('FOO')
             client.close()
 
+    def test_close_in_long_command(self):
+        with SMTP(*self.address) as client:
+            client.send('F' + 5555 * 'O')  # without CRLF
+            client.close()
+
     def test_close_in_data(self):
         with SMTP(*self.address) as client:
             code, response = client.helo('example.com')