Reject NUL, CR and LF in _raw_command arguments before sending (#658)

CR and LF end the command line on the wire, so an argument carrying
them, for example a search criterion built from a sender-controlled
header, is run by the server as extra commands. _raw_command bypasses
imaplib._command, so CPython's check for the same characters does not
apply here. Check every argument before the first send, so a bad one
never leaves a half-sent command; 8-bit values still go as literals,
which may carry CR and LF. Fixes #657

`literal` is now exported so callers can wrap a value that carries CR or LF,
as the new error message suggests. _literal stays as an alias for code that
imported the old name. 

search() quoted any value with a space, so a literal holding CR or LF came
out as a _quoted string and the new check rejected it, even though the
error message says to wrap the value in literal.

Author: HardMax71

imapclient/imapclient.py

@@ -42,6 +42,7 @@
     "FLAGGED",
     "DRAFT",
     "RECENT",
+    "literal",
 ]
 
 
@@ -1461,9 +1462,9 @@ def chunks():
                         yield to_bytes(seq_to_parenstr(m["flags"]))
                     if "date" in m:
                         yield to_bytes('"%s"' % datetime_to_INTERNALDATE(m["date"]))
-                    yield _literal(to_bytes(m["msg"]))
+                    yield literal(to_bytes(m["msg"]))
                 else:
-                    yield _literal(to_bytes(m))
+                    yield literal(to_bytes(m))
 
         msgs = list(chunks())
 
@@ -1709,11 +1710,25 @@ def _raw_command(self, command, args, uid=True):
             prefix.append(b"UID")
         prefix.append(command)
 
-        line = []
-        for item, is_last in _iter_with_last(prefix + args):
+        # Check every argument before anything goes on the wire: a bad
+        # argument found after a literal was sent would leave the server
+        # waiting inside a half-sent command.
+        for item in itertools.chain(prefix, args):
             if not isinstance(item, bytes):
                 raise ValueError("command args must be passed as bytes")
+            if b"\x00" in item:
+                raise ValueError("NUL is not allowed in command arguments")
+            if not _is8bit(item) and (b"\r" in item or b"\n" in item):
+                # CR and LF end the command line on the wire, so an argument
+                # carrying them would be run as extra commands. A literal can
+                # hold them, which is what the 8-bit path already sends.
+                raise ValueError(
+                    "CR and LF are not allowed in command arguments; "
+                    "wrap the value in imapclient.literal to send it as a literal"
+                )
 
+        line = []
+        for item, is_last in _iter_with_last(prefix + args):
             if _is8bit(item):
                 # If a line was already started send it
                 if line:
@@ -1872,6 +1887,8 @@ def _normalise_search_criteria(criteria, charset=None):
             inner[0] = b"(" + inner[0]
             inner[-1] = inner[-1] + b")"
             out.extend(inner)  # flatten
+        elif isinstance(item, literal):
+            out.append(item)
         else:
             out.append(_quoted.maybe(to_bytes(item, charset)))
     return out
@@ -1883,10 +1900,13 @@ def _normalise_sort_criteria(criteria, charset=None):
     return b"(" + b" ".join(to_bytes(item).upper() for item in criteria) + b")"
 
 
-class _literal(bytes):
+class literal(bytes):
     """Hold message data that should always be sent as a literal."""
 
 
+_literal = literal  # old private name, kept for callers that imported it
+
+
 class _quoted(bytes):
     """
     This class holds a quoted bytes value which provides access to the
@@ -1979,7 +1999,7 @@ def as_triplets(items):
 
 
 def _is8bit(data):
-    return isinstance(data, _literal) or any(b > 127 for b in data)
+    return isinstance(data, literal) or any(b > 127 for b in data)
 
 
 def _iter_with_last(items):

tests/test_imapclient.py

@@ -15,9 +15,9 @@
 from imapclient.exceptions import CapabilityError, IMAPClientError, ProtocolError
 from imapclient.fixed_offset import FixedOffset
 from imapclient.imapclient import (
-    _literal,
     _parse_quota,
     IMAPlibLoggerAdapter,
+    literal,
     MailboxQuotaRoots,
     Quota,
     require_capability,
@@ -416,11 +416,11 @@ def test_multiappend_with_flags_and_internaldate(self):
                 b'"foobar"',
                 b"(FLAG WAVE)",
                 b'"05-Apr-2009 11:00:05 +0200"',
-                _literal(b"msg1"),
+                literal(b"msg1"),
                 b"(FLAG WAVE)",
-                _literal(b"msg2"),
+                literal(b"msg2"),
                 b'"05-Apr-2009 11:00:05 +0200"',
-                _literal(b"msg3"),
+                literal(b"msg3"),
             ],
             uid=False,
         )
@@ -749,6 +749,40 @@ def fake_get_response():
         self.assertListEqual([(99, b"EXISTS")], responses)
 
 
+class TestRawCommandRejectsControlCharacters(IMAPClientTest):
+    def setUp(self):
+        super().setUp()
+        self.client._imap.send = Mock()
+        self.client._cached_capabilities = (b"IMAP4REV1",)
+
+    def test_crlf_in_a_line_argument_is_rejected_before_sending(self):
+        criterion = b'x"\r\nX1 STORE 1:* +FLAGS (\\Deleted)\r\nX2 EXPUNGE'
+
+        with self.assertRaises(ValueError):
+            self.client._raw_command(b"SEARCH", [b"HEADER", b"Message-ID", criterion])
+
+        self.client._imap.send.assert_not_called()
+
+    def test_crlf_via_search_criteria_is_rejected_before_sending(self):
+        with self.assertRaises(ValueError):
+            self.client.search(["HEADER", "Message-ID", 'x"\r\nX1 NOOP'])
+
+        self.client._imap.send.assert_not_called()
+
+    def test_nul_is_rejected_even_inside_a_literal(self):
+        with self.assertRaises(ValueError):
+            self.client._raw_command(b"SEARCH", [b"TEXT", literal(b"a\x00b")])
+
+        self.client._imap.send.assert_not_called()
+
+    def test_crlf_inside_a_literal_is_allowed(self):
+        self.client._send_literal = Mock()
+
+        self.client._raw_command(b"SEARCH", [b"TEXT", literal(b"line one\r\nline two")])
+
+        self.client._send_literal.assert_called_once()
+
+
 class TestDebugLogging(IMAPClientTest):
     def test_IMAP_is_patched(self):
         # Remove all logging handlers so that the order of tests does not
@@ -1006,7 +1040,7 @@ def test_literal_plus(self):
         self.client._cached_capabilities = (b"LITERAL+",)
 
         typ, data = self.client._raw_command(
-            b"APPEND", [b"\xff", _literal(b"hello")], uid=False
+            b"APPEND", [b"\xff", literal(b"hello")], uid=False
         )
         self.assertEqual(typ, "OK")
         self.assertEqual(data, ["done"])
@@ -1020,7 +1054,7 @@ def test_literal_plus_multiple_literals(self):
 
         typ, data = self.client._raw_command(
             b"APPEND",
-            [b"\xff", _literal(b"hello"), b"TEXT", _literal(b"test")],
+            [b"\xff", literal(b"hello"), b"TEXT", literal(b"test")],
             uid=False,
         )
         self.assertEqual(typ, "OK")

tests/test_util_functions.py

@@ -9,6 +9,7 @@
     _normalise_search_criteria,
     _quoted,
     join_message_ids,
+    literal,
     normalise_text_list,
     seq_to_parenstr,
     seq_to_parenstr_upper,
@@ -139,6 +140,10 @@ def test_mixed_list(self):
     def test_quoting(self):
         self.check(["foo bar"], None, [_quoted(b'"foo bar"')])
 
+    def test_literal_is_not_quoted(self):
+        value = literal(b"line one\r\nline two")
+        self.check(["TEXT", value], None, [b"TEXT", value])
+
     def test_ints(self):
         self.check(["modseq", 500], None, [b"modseq", b"500"])