Reject NUL, CR and LF in _raw_command arguments before sending

imapclient/imapclient.py

@@ -42,6 +42,7 @@
     "FLAGGED",
     "DRAFT",
     "RECENT",
+    "literal",
 ]
 
 
@@ -1457,9 +1458,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())
 
@@ -1705,11 +1706,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:
@@ -1868,6 +1883,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
@@ -1879,10 +1896,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
@@ -1975,7 +1995,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"])