from test import support from test.support import socket_helper from contextlib import contextmanager import imaplib import os.path import socketserver import time import calendar import threading import socket from test.support import (verbose, run_with_tz, run_with_locale, cpython_only) from test.support import hashlib_helper from test.support import threading_helper import unittest from unittest import mock from datetime import datetime, timezone, timedelta try: import ssl except ImportError: ssl = None CERTFILE = os.path.join(os.path.dirname(__file__) or os.curdir, "keycert3.pem") CAFILE = os.path.join(os.path.dirname(__file__) or os.curdir, "pycacert.pem") class TestImaplib(unittest.TestCase): def test_Internaldate2tuple(self): t0 = calendar.timegm((2000, 1, 1, 0, 0, 0, -1, -1, -1)) tt = imaplib.Internaldate2tuple( b'25 (INTERNALDATE "01-Jan-2000 00:00:00 +0000")') self.assertEqual(time.mktime(tt), t0) tt = imaplib.Internaldate2tuple( b'25 (INTERNALDATE "01-Jan-2000 11:30:00 +1130")') self.assertEqual(time.mktime(tt), t0) tt = imaplib.Internaldate2tuple( b'25 (INTERNALDATE "31-Dec-1999 12:30:00 -1130")') self.assertEqual(time.mktime(tt), t0) @run_with_tz('MST+07MDT,M4.1.0,M10.5.0') def test_Internaldate2tuple_issue10941(self): self.assertNotEqual(imaplib.Internaldate2tuple( b'25 (INTERNALDATE "02-Apr-2000 02:30:00 +0000")'), imaplib.Internaldate2tuple( b'25 (INTERNALDATE "02-Apr-2000 03:30:00 +0000")')) def timevalues(self): return [2000000000, 2000000000.0, time.localtime(2000000000), (2033, 5, 18, 5, 33, 20, -1, -1, -1), (2033, 5, 18, 5, 33, 20, -1, -1, 1), datetime.fromtimestamp(2000000000, timezone(timedelta(0, 2 * 60 * 60))), '"18-May-2033 05:33:20 +0200"'] @run_with_locale('LC_ALL', 'de_DE', 'fr_FR') # DST rules included to work around quirk where the Gnu C library may not # otherwise restore the previous time zone @run_with_tz('STD-1DST,M3.2.0,M11.1.0') def test_Time2Internaldate(self): expected = '"18-May-2033 05:33:20 +0200"' for t in self.timevalues(): internal = imaplib.Time2Internaldate(t) self.assertEqual(internal, expected) def test_that_Time2Internaldate_returns_a_result(self): # Without tzset, we can check only that it successfully # produces a result, not the correctness of the result itself, # since the result depends on the timezone the machine is in. for t in self.timevalues(): imaplib.Time2Internaldate(t) def test_imap4_host_default_value(self): # Check whether the IMAP4_PORT is truly unavailable. with socket.socket() as s: try: s.connect(('', imaplib.IMAP4_PORT)) self.skipTest( "Cannot run the test with local IMAP server running.") except socket.error: pass # This is the exception that should be raised. expected_errnos = socket_helper.get_socket_conn_refused_errs() with self.assertRaises(OSError) as cm: imaplib.IMAP4() self.assertIn(cm.exception.errno, expected_errnos) if ssl: class SecureTCPServer(socketserver.TCPServer): def get_request(self): newsocket, fromaddr = self.socket.accept() context = ssl.SSLContext() context.load_cert_chain(CERTFILE) connstream = context.wrap_socket(newsocket, server_side=True) return connstream, fromaddr IMAP4_SSL = imaplib.IMAP4_SSL else: class SecureTCPServer: pass IMAP4_SSL = None class SimpleIMAPHandler(socketserver.StreamRequestHandler): timeout = support.LOOPBACK_TIMEOUT continuation = None capabilities = '' def setup(self): super().setup() self.server.is_selected = False self.server.logged = None def _send(self, message): if verbose: print("SENT: %r" % message.strip()) self.wfile.write(message) def _send_line(self, message): self._send(message + b'\r\n') def _send_textline(self, message): self._send_line(message.encode('ASCII')) def _send_tagged(self, tag, code, message): self._send_textline(' '.join((tag, code, message))) def handle(self): # Send a welcome message. self._send_textline('* OK IMAP4rev1') while 1: # Gather up input until we receive a line terminator or we timeout. # Accumulate read(1) because it's simpler to handle the differences # between naked sockets and SSL sockets. line = b'' while 1: try: part = self.rfile.read(1) if part == b'': # Naked sockets return empty strings.. return line += part except OSError: # ..but SSLSockets raise exceptions. return if line.endswith(b'\r\n'): break if verbose: print('GOT: %r' % line.strip()) if self.continuation: try: self.continuation.send(line) except StopIteration: self.continuation = None continue splitline = line.decode('ASCII').split() tag = splitline[0] cmd = splitline[1] args = splitline[2:] if hasattr(self, 'cmd_' + cmd): continuation = getattr(self, 'cmd_' + cmd)(tag, args) if continuation: self.continuation = continuation next(continuation) else: self._send_tagged(tag, 'BAD', cmd + ' unknown') def cmd_CAPABILITY(self, tag, args): caps = ('IMAP4rev1 ' + self.capabilities if self.capabilities else 'IMAP4rev1') self._send_textline('* CAPABILITY ' + caps) self._send_tagged(tag, 'OK', 'CAPABILITY completed') def cmd_LOGOUT(self, tag, args): self.server.logged = None self._send_textline('* BYE IMAP4ref1 Server logging out') self._send_tagged(tag, 'OK', 'LOGOUT completed') def cmd_LOGIN(self, tag, args): self.server.logged = args[0] self._send_tagged(tag, 'OK', 'LOGIN completed') def cmd_SELECT(self, tag, args): self.server.is_selected = True self._send_line(b'* 2 EXISTS') self._send_tagged(tag, 'OK', '[READ-WRITE] SELECT completed.') def cmd_UNSELECT(self, tag, args): if self.server.is_selected: self.server.is_selected = False self._send_tagged(tag, 'OK', 'Returned to authenticated state. (Success)') else: self._send_tagged(tag, 'BAD', 'No mailbox selected') class NewIMAPTestsMixin(): client = None def _setup(self, imap_handler, connect=True): """ Sets up imap_handler for tests. imap_handler should inherit from either: - SimpleIMAPHandler - for testing IMAP commands, - socketserver.StreamRequestHandler - if raw access to stream is needed. Returns (client, server). """ class TestTCPServer(self.server_class): def handle_error(self, request, client_address): """ End request and raise the error if one occurs. """ self.close_request(request) self.server_close() raise self.addCleanup(self._cleanup) self.server = self.server_class((socket_helper.HOST, 0), imap_handler) self.thread = threading.Thread( name=self._testMethodName+'-server', target=self.server.serve_forever, # Short poll interval to make the test finish quickly. # Time between requests is short enough that we won't wake # up spuriously too many times. kwargs={'poll_interval': 0.01}) self.thread.daemon = True # In case this function raises. self.thread.start() if connect: self.client = self.imap_class(*self.server.server_address) return self.client, self.server def _cleanup(self): """ Cleans up the test server. This method should not be called manually, it is added to the cleanup queue in the _setup method already. """ # if logout was called already we'd raise an exception trying to # shutdown the client once again if self.client is not None and self.client.state != 'LOGOUT': self.client.shutdown() # cleanup the server self.server.shutdown() self.server.server_close() threading_helper.join_thread(self.thread) # Explicitly clear the attribute to prevent dangling thread self.thread = None def test_EOF_without_complete_welcome_message(self): # http://bugs.python.org/issue5949 class EOFHandler(socketserver.StreamRequestHandler): def handle(self): self.wfile.write(b'* OK') _, server = self._setup(EOFHandler, connect=False) self.assertRaises(imaplib.IMAP4.abort, self.imap_class, *server.server_address) def test_line_termination(self): class BadNewlineHandler(SimpleIMAPHandler): def cmd_CAPABILITY(self, tag, args): self._send(b'* CAPABILITY IMAP4rev1 AUTH\n') self._send_tagged(tag, 'OK', 'CAPABILITY completed') _, server = self._setup(BadNewlineHandler, connect=False) self.assertRaises(imaplib.IMAP4.abort, self.imap_class, *server.server_address) def test_enable_raises_error_if_not_AUTH(self): class EnableHandler(SimpleIMAPHandler): capabilities = 'AUTH ENABLE UTF8=ACCEPT' client, _ = self._setup(EnableHandler) self.assertFalse(client.utf8_enabled) with self.assertRaisesRegex(imaplib.IMAP4.error, 'ENABLE.*NONAUTH'): client.enable('foo') self.assertFalse(client.utf8_enabled) def test_enable_raises_error_if_no_capability(self): client, _ = self._setup(SimpleIMAPHandler) with self.assertRaisesRegex(imaplib.IMAP4.error, 'does not support ENABLE'): client.enable('foo') def test_enable_UTF8_raises_error_if_not_supported(self): client, _ = self._setup(SimpleIMAPHandler) typ, data = client.login('user', 'pass') self.assertEqual(typ, 'OK') with self.assertRaisesRegex(imaplib.IMAP4.error, 'does not support ENABLE'): client.enable('UTF8=ACCEPT') def test_enable_UTF8_True_append(self): class UTF8AppendServer(SimpleIMAPHandler): capabilities = 'ENABLE UTF8=ACCEPT' def cmd_ENABLE(self, tag, args): self._send_tagged(tag, 'OK', 'ENABLE successful') def cmd_AUTHENTICATE(self, tag, args): self._send_textline('+') self.server.response = yield self._send_tagged(tag, 'OK', 'FAKEAUTH successful') def cmd_APPEND(self, tag, args): self._send_textline('+') self.server.response = yield self._send_tagged(tag, 'OK', 'okay') client, server = self._setup(UTF8AppendServer) self.assertEqual(client._encoding, 'ascii') code, _ = client.authenticate('MYAUTH', lambda x: b'fake') self.assertEqual(code, 'OK') self.assertEqual(server.response, b'ZmFrZQ==\r\n') # b64 encoded 'fake' code, _ = client.enable('UTF8=ACCEPT') self.assertEqual(code, 'OK') self.assertEqual(client._encoding, 'utf-8') msg_string = 'Subject: üñí©öðé' typ, data = client.append(None, None, None, msg_string.encode('utf-8')) self.assertEqual(typ, 'OK') self.assertEqual(server.response, ('UTF8 (%s)\r\n' % msg_string).encode('utf-8')) def test_search_disallows_charset_in_utf8_mode(self): class UTF8Server(SimpleIMAPHandler): capabilities = 'AUTH ENABLE UTF8=ACCEPT' def cmd_ENABLE(self, tag, args): self._send_tagged(tag, 'OK', 'ENABLE successful') def cmd_AUTHENTICATE(self, tag, args): self._send_textline('+') self.server.response = yield self._send_tagged(tag, 'OK', 'FAKEAUTH successful') client, _ = self._setup(UTF8Server) typ, _ = client.authenticate('MYAUTH', lambda x: b'fake') self.assertEqual(typ, 'OK') typ, _ = client.enable('UTF8=ACCEPT') self.assertEqual(typ, 'OK') self.assertTrue(client.utf8_enabled) with self.assertRaisesRegex(imaplib.IMAP4.error, 'charset.*UTF8'): client.search('foo', 'bar') def test_bad_auth_name(self): class MyServer(SimpleIMAPHandler): def cmd_AUTHENTICATE(self, tag, args): self._send_tagged(tag, 'NO', 'unrecognized authentication type {}'.format(args[0])) client, _ = self._setup(MyServer) with self.assertRaisesRegex(imaplib.IMAP4.error, 'unrecognized authentication type METHOD'): client.authenticate('METHOD', lambda: 1) def test_invalid_authentication(self): class MyServer(SimpleIMAPHandler): def cmd_AUTHENTICATE(self, tag, args): self._send_textline('+') self.response = yield self._send_tagged(tag, 'NO', '[AUTHENTICATIONFAILED] invalid') client, _ = self._setup(MyServer) with self.assertRaisesRegex(imaplib.IMAP4.error, r'\[AUTHENTICATIONFAILED\] invalid'): client.authenticate('MYAUTH', lambda x: b'fake') def test_valid_authentication_bytes(self): class MyServer(SimpleIMAPHandler): def cmd_AUTHENTICATE(self, tag, args): self._send_textline('+') self.server.response = yield self._send_tagged(tag, 'OK', 'FAKEAUTH successful') client, server = self._setup(MyServer) code, _ = client.authenticate('MYAUTH', lambda x: b'fake') self.assertEqual(code, 'OK') self.assertEqual(server.response, b'ZmFrZQ==\r\n') # b64 encoded 'fake' def test_valid_authentication_plain_text(self): class MyServer(SimpleIMAPHandler): def cmd_AUTHENTICATE(self, tag, args): self._send_textline('+') self.server.response = yield self._send_tagged(tag, 'OK', 'FAKEAUTH successful') client, server = self._setup(MyServer) code, _ = client.authenticate('MYAUTH', lambda x: 'fake') self.assertEqual(code, 'OK') self.assertEqual(server.response, b'ZmFrZQ==\r\n') # b64 encoded 'fake' @hashlib_helper.requires_hashdigest('md5') def test_login_cram_md5_bytes(self): class AuthHandler(SimpleIMAPHandler): capabilities = 'LOGINDISABLED AUTH=CRAM-MD5' def cmd_AUTHENTICATE(self, tag, args): self._send_textline('+ PDE4OTYuNjk3MTcwOTUyQHBvc3RvZmZpY2Uucm' 'VzdG9uLm1jaS5uZXQ=') r = yield if (r == b'dGltIGYxY2E2YmU0NjRiOWVmYT' b'FjY2E2ZmZkNmNmMmQ5ZjMy\r\n'): self._send_tagged(tag, 'OK', 'CRAM-MD5 successful') else: self._send_tagged(tag, 'NO', 'No access') client, _ = self._setup(AuthHandler) self.assertTrue('AUTH=CRAM-MD5' in client.capabilities) ret, _ = client.login_cram_md5("tim", b"tanstaaftanstaaf") self.assertEqual(ret, "OK") @hashlib_helper.requires_hashdigest('md5') def test_login_cram_md5_plain_text(self): class AuthHandler(SimpleIMAPHandler): capabilities = 'LOGINDISABLED AUTH=CRAM-MD5' def cmd_AUTHENTICATE(self, tag, args): self._send_textline('+ PDE4OTYuNjk3MTcwOTUyQHBvc3RvZmZpY2Uucm' 'VzdG9uLm1jaS5uZXQ=') r = yield if (r == b'dGltIGYxY2E2YmU0NjRiOWVmYT' b'FjY2E2ZmZkNmNmMmQ5ZjMy\r\n'): self._send_tagged(tag, 'OK', 'CRAM-MD5 successful') else: self._send_tagged(tag, 'NO', 'No access') client, _ = self._setup(AuthHandler) self.assertTrue('AUTH=CRAM-MD5' in client.capabilities) ret, _ = client.login_cram_md5("tim", "tanstaaftanstaaf") self.assertEqual(ret, "OK") def test_aborted_authentication(self): class MyServer(SimpleIMAPHandler): def cmd_AUTHENTICATE(self, tag, args): self._send_textline('+') self.response = yield if self.response == b'*\r\n': self._send_tagged( tag, 'NO', '[AUTHENTICATIONFAILED] aborted') else: self._send_tagged(tag, 'OK', 'MYAUTH successful') client, _ = self._setup(MyServer) with self.assertRaisesRegex(imaplib.IMAP4.error, r'\[AUTHENTICATIONFAILED\] aborted'): client.authenticate('MYAUTH', lambda x: None) @mock.patch('imaplib._MAXLINE', 10) def test_linetoolong(self): class TooLongHandler(SimpleIMAPHandler): def handle(self): # send response line longer than the limit set in the next line self.wfile.write(b'* OK ' + 11 * b'x' + b'\r\n') _, server = self._setup(TooLongHandler, connect=False) with self.assertRaisesRegex(imaplib.IMAP4.error, 'got more than 10 bytes'): self.imap_class(*server.server_address) def test_simple_with_statement(self): _, server = self._setup(SimpleIMAPHandler, connect=False) with self.imap_class(*server.server_address): pass def test_imaplib_timeout_test(self): _, server = self._setup(SimpleIMAPHandler) addr = server.server_address[1] client = self.imap_class("localhost", addr, timeout=None) self.assertEqual(client.sock.timeout, None) client.shutdown() client = self.imap_class("localhost", addr, timeout=support.LOOPBACK_TIMEOUT) self.assertEqual(client.sock.timeout, support.LOOPBACK_TIMEOUT) client.shutdown() with self.assertRaises(ValueError): client = self.imap_class("localhost", addr, timeout=0) def test_imaplib_timeout_functionality_test(self): class TimeoutHandler(SimpleIMAPHandler): def handle(self): time.sleep(1) SimpleIMAPHandler.handle(self) _, server = self._setup(TimeoutHandler) addr = server.server_address[1] with self.assertRaises(socket.timeout): client = self.imap_class("localhost", addr, timeout=0.001) def test_with_statement(self): _, server = self._setup(SimpleIMAPHandler, connect=False) with self.imap_class(*server.server_address) as imap: imap.login('user', 'pass') self.assertEqual(server.logged, 'user') self.assertIsNone(server.logged) def test_with_statement_logout(self): # It is legal to log out explicitly inside the with block _, server = self._setup(SimpleIMAPHandler, connect=False) with self.imap_class(*server.server_address) as imap: imap.login('user', 'pass') self.assertEqual(server.logged, 'user') imap.logout() self.assertIsNone(server.logged) self.assertIsNone(server.logged) # command tests def test_login(self): client, _ = self._setup(SimpleIMAPHandler) typ, data = client.login('user', 'pass') self.assertEqual(typ, 'OK') self.assertEqual(data[0], b'LOGIN completed') self.assertEqual(client.state, 'AUTH') def test_logout(self): client, _ = self._setup(SimpleIMAPHandler) typ, data = client.login('user', 'pass') self.assertEqual(typ, 'OK') self.assertEqual(data[0], b'LOGIN completed') typ, data = client.logout() self.assertEqual(typ, 'BYE', (typ, data)) self.assertEqual(data[0], b'IMAP4ref1 Server logging out', (typ, data)) self.assertEqual(client.state, 'LOGOUT') def test_lsub(self): class LsubCmd(SimpleIMAPHandler): def cmd_LSUB(self, tag, args): self._send_textline('* LSUB () "." directoryA') return self._send_tagged(tag, 'OK', 'LSUB completed') client, _ = self._setup(LsubCmd) client.login('user', 'pass') typ, data = client.lsub() self.assertEqual(typ, 'OK') self.assertEqual(data[0], b'() "." directoryA') def test_unselect(self): client, _ = self._setup(SimpleIMAPHandler) client.login('user', 'pass') typ, data = client.select() self.assertEqual(typ, 'OK') self.assertEqual(data[0], b'2') typ, data = client.unselect() self.assertEqual(typ, 'OK') self.assertEqual(data[0], b'Returned to authenticated state. (Success)') self.assertEqual(client.state, 'AUTH') class NewIMAPTests(NewIMAPTestsMixin, unittest.TestCase): imap_class = imaplib.IMAP4 server_class = socketserver.TCPServer @unittest.skipUnless(ssl, "SSL not available") class NewIMAPSSLTests(NewIMAPTestsMixin, unittest.TestCase): imap_class = IMAP4_SSL server_class = SecureTCPServer def test_ssl_raises(self): ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) self.assertEqual(ssl_context.verify_mode, ssl.CERT_REQUIRED) self.assertEqual(ssl_context.check_hostname, True) ssl_context.load_verify_locations(CAFILE) with self.assertRaisesRegex(ssl.CertificateError, "IP address mismatch, certificate is not valid for " "'127.0.0.1'"): _, server = self._setup(SimpleIMAPHandler) client = self.imap_class(*server.server_address, ssl_context=ssl_context) client.shutdown() def test_ssl_verified(self): ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) ssl_context.load_verify_locations(CAFILE) _, server = self._setup(SimpleIMAPHandler) client = self.imap_class("localhost", server.server_address[1], ssl_context=ssl_context) client.shutdown() # Mock the private method _connect(), so mark the test as specific # to CPython stdlib @cpython_only def test_certfile_arg_warn(self): with support.check_warnings(('', DeprecationWarning)): with mock.patch.object(self.imap_class, 'open'): with mock.patch.object(self.imap_class, '_connect'): self.imap_class('localhost', 143, certfile=CERTFILE) class ThreadedNetworkedTests(unittest.TestCase): server_class = socketserver.TCPServer imap_class = imaplib.IMAP4 def make_server(self, addr, hdlr): class MyServer(self.server_class): def handle_error(self, request, client_address): self.close_request(request) self.server_close() raise if verbose: print("creating server") server = MyServer(addr, hdlr) self.assertEqual(server.server_address, server.socket.getsockname()) if verbose: print("server created") print("ADDR =", addr) print("CLASS =", self.server_class) print("HDLR =", server.RequestHandlerClass) t = threading.Thread( name='%s serving' % self.server_class, target=server.serve_forever, # Short poll interval to make the test finish quickly. # Time between requests is short enough that we won't wake # up spuriously too many times. kwargs={'poll_interval': 0.01}) t.daemon = True # In case this function raises. t.start() if verbose: print("server running") return server, t def reap_server(self, server, thread): if verbose: print("waiting for server") server.shutdown() server.server_close() thread.join() if verbose: print("done") @contextmanager def reaped_server(self, hdlr): server, thread = self.make_server((socket_helper.HOST, 0), hdlr) try: yield server finally: self.reap_server(server, thread) @contextmanager def reaped_pair(self, hdlr): with self.reaped_server(hdlr) as server: client = self.imap_class(*server.server_address) try: yield server, client finally: client.logout() @threading_helper.reap_threads def test_connect(self): with self.reaped_server(SimpleIMAPHandler) as server: client = self.imap_class(*server.server_address) client.shutdown() @threading_helper.reap_threads def test_bracket_flags(self): # This violates RFC 3501, which disallows ']' characters in tag names, # but imaplib has allowed producing such tags forever, other programs # also produce them (eg: OtherInbox's Organizer app as of 20140716), # and Gmail, for example, accepts them and produces them. So we # support them. See issue #21815. class BracketFlagHandler(SimpleIMAPHandler): def handle(self): self.flags = ['Answered', 'Flagged', 'Deleted', 'Seen', 'Draft'] super().handle() def cmd_AUTHENTICATE(self, tag, args): self._send_textline('+') self.server.response = yield self._send_tagged(tag, 'OK', 'FAKEAUTH successful') def cmd_SELECT(self, tag, args): flag_msg = ' \\'.join(self.flags) self._send_line(('* FLAGS (%s)' % flag_msg).encode('ascii')) self._send_line(b'* 2 EXISTS') self._send_line(b'* 0 RECENT') msg = ('* OK [PERMANENTFLAGS %s \\*)] Flags permitted.' % flag_msg) self._send_line(msg.encode('ascii')) self._send_tagged(tag, 'OK', '[READ-WRITE] SELECT completed.') def cmd_STORE(self, tag, args): new_flags = args[2].strip('(').strip(')').split() self.flags.extend(new_flags) flags_msg = '(FLAGS (%s))' % ' \\'.join(self.flags) msg = '* %s FETCH %s' % (args[0], flags_msg) self._send_line(msg.encode('ascii')) self._send_tagged(tag, 'OK', 'STORE completed.') with self.reaped_pair(BracketFlagHandler) as (server, client): code, data = client.authenticate('MYAUTH', lambda x: b'fake') self.assertEqual(code, 'OK') self.assertEqual(server.response, b'ZmFrZQ==\r\n') client.select('test') typ, [data] = client.store(b'1', "+FLAGS", "[test]") self.assertIn(b'[test]', data) client.select('test') typ, [data] = client.response('PERMANENTFLAGS') self.assertIn(b'[test]', data) @threading_helper.reap_threads def test_issue5949(self): class EOFHandler(socketserver.StreamRequestHandler): def handle(self): # EOF without sending a complete welcome message. self.wfile.write(b'* OK') with self.reaped_server(EOFHandler) as server: self.assertRaises(imaplib.IMAP4.abort, self.imap_class, *server.server_address) @threading_helper.reap_threads def test_line_termination(self): class BadNewlineHandler(SimpleIMAPHandler): def cmd_CAPABILITY(self, tag, args): self._send(b'* CAPABILITY IMAP4rev1 AUTH\n') self._send_tagged(tag, 'OK', 'CAPABILITY completed') with self.reaped_server(BadNewlineHandler) as server: self.assertRaises(imaplib.IMAP4.abort, self.imap_class, *server.server_address) class UTF8Server(SimpleIMAPHandler): capabilities = 'AUTH ENABLE UTF8=ACCEPT' def cmd_ENABLE(self, tag, args): self._send_tagged(tag, 'OK', 'ENABLE successful') def cmd_AUTHENTICATE(self, tag, args): self._send_textline('+') self.server.response = yield self._send_tagged(tag, 'OK', 'FAKEAUTH successful') @threading_helper.reap_threads def test_enable_raises_error_if_not_AUTH(self): with self.reaped_pair(self.UTF8Server) as (server, client): self.assertFalse(client.utf8_enabled) self.assertRaises(imaplib.IMAP4.error, client.enable, 'foo') self.assertFalse(client.utf8_enabled) # XXX Also need a test that enable after SELECT raises an error. @threading_helper.reap_threads def test_enable_raises_error_if_no_capability(self): class NoEnableServer(self.UTF8Server): capabilities = 'AUTH' with self.reaped_pair(NoEnableServer) as (server, client): self.assertRaises(imaplib.IMAP4.error, client.enable, 'foo') @threading_helper.reap_threads def test_enable_UTF8_raises_error_if_not_supported(self): class NonUTF8Server(SimpleIMAPHandler): pass with self.assertRaises(imaplib.IMAP4.error): with self.reaped_pair(NonUTF8Server) as (server, client): typ, data = client.login('user', 'pass') self.assertEqual(typ, 'OK') client.enable('UTF8=ACCEPT') pass @threading_helper.reap_threads def test_enable_UTF8_True_append(self): class UTF8AppendServer(self.UTF8Server): def cmd_APPEND(self, tag, args): self._send_textline('+') self.server.response = yield self._send_tagged(tag, 'OK', 'okay') with self.reaped_pair(UTF8AppendServer) as (server, client): self.assertEqual(client._encoding, 'ascii') code, _ = client.authenticate('MYAUTH', lambda x: b'fake') self.assertEqual(code, 'OK') self.assertEqual(server.response, b'ZmFrZQ==\r\n') # b64 encoded 'fake' code, _ = client.enable('UTF8=ACCEPT') self.assertEqual(code, 'OK') self.assertEqual(client._encoding, 'utf-8') msg_string = 'Subject: üñí©öðé' typ, data = client.append( None, None, None, msg_string.encode('utf-8')) self.assertEqual(typ, 'OK') self.assertEqual( server.response, ('UTF8 (%s)\r\n' % msg_string).encode('utf-8') ) # XXX also need a test that makes sure that the Literal and Untagged_status # regexes uses unicode in UTF8 mode instead of the default ASCII. @threading_helper.reap_threads def test_search_disallows_charset_in_utf8_mode(self): with self.reaped_pair(self.UTF8Server) as (server, client): typ, _ = client.authenticate('MYAUTH', lambda x: b'fake') self.assertEqual(typ, 'OK') typ, _ = client.enable('UTF8=ACCEPT') self.assertEqual(typ, 'OK') self.assertTrue(client.utf8_enabled) self.assertRaises(imaplib.IMAP4.error, client.search, 'foo', 'bar') @threading_helper.reap_threads def test_bad_auth_name(self): class MyServer(SimpleIMAPHandler): def cmd_AUTHENTICATE(self, tag, args): self._send_tagged(tag, 'NO', 'unrecognized authentication ' 'type {}'.format(args[0])) with self.reaped_pair(MyServer) as (server, client): with self.assertRaises(imaplib.IMAP4.error): client.authenticate('METHOD', lambda: 1) @threading_helper.reap_threads def test_invalid_authentication(self): class MyServer(SimpleIMAPHandler): def cmd_AUTHENTICATE(self, tag, args): self._send_textline('+') self.response = yield self._send_tagged(tag, 'NO', '[AUTHENTICATIONFAILED] invalid') with self.reaped_pair(MyServer) as (server, client): with self.assertRaises(imaplib.IMAP4.error): code, data = client.authenticate('MYAUTH', lambda x: b'fake') @threading_helper.reap_threads def test_valid_authentication(self): class MyServer(SimpleIMAPHandler): def cmd_AUTHENTICATE(self, tag, args): self._send_textline('+') self.server.response = yield self._send_tagged(tag, 'OK', 'FAKEAUTH successful') with self.reaped_pair(MyServer) as (server, client): code, data = client.authenticate('MYAUTH', lambda x: b'fake') self.assertEqual(code, 'OK') self.assertEqual(server.response, b'ZmFrZQ==\r\n') # b64 encoded 'fake' with self.reaped_pair(MyServer) as (server, client): code, data = client.authenticate('MYAUTH', lambda x: 'fake') self.assertEqual(code, 'OK') self.assertEqual(server.response, b'ZmFrZQ==\r\n') # b64 encoded 'fake' @threading_helper.reap_threads @hashlib_helper.requires_hashdigest('md5') def test_login_cram_md5(self): class AuthHandler(SimpleIMAPHandler): capabilities = 'LOGINDISABLED AUTH=CRAM-MD5' def cmd_AUTHENTICATE(self, tag, args): self._send_textline('+ PDE4OTYuNjk3MTcwOTUyQHBvc3RvZmZpY2Uucm' 'VzdG9uLm1jaS5uZXQ=') r = yield if (r == b'dGltIGYxY2E2YmU0NjRiOWVmYT' b'FjY2E2ZmZkNmNmMmQ5ZjMy\r\n'): self._send_tagged(tag, 'OK', 'CRAM-MD5 successful') else: self._send_tagged(tag, 'NO', 'No access') with self.reaped_pair(AuthHandler) as (server, client): self.assertTrue('AUTH=CRAM-MD5' in client.capabilities) ret, data = client.login_cram_md5("tim", "tanstaaftanstaaf") self.assertEqual(ret, "OK") with self.reaped_pair(AuthHandler) as (server, client): self.assertTrue('AUTH=CRAM-MD5' in client.capabilities) ret, data = client.login_cram_md5("tim", b"tanstaaftanstaaf") self.assertEqual(ret, "OK") @threading_helper.reap_threads def test_aborted_authentication(self): class MyServer(SimpleIMAPHandler): def cmd_AUTHENTICATE(self, tag, args): self._send_textline('+') self.response = yield if self.response == b'*\r\n': self._send_tagged(tag, 'NO', '[AUTHENTICATIONFAILED] aborted') else: self._send_tagged(tag, 'OK', 'MYAUTH successful') with self.reaped_pair(MyServer) as (server, client): with self.assertRaises(imaplib.IMAP4.error): code, data = client.authenticate('MYAUTH', lambda x: None) def test_linetoolong(self): class TooLongHandler(SimpleIMAPHandler): def handle(self): # Send a very long response line self.wfile.write(b'* OK ' + imaplib._MAXLINE * b'x' + b'\r\n') with self.reaped_server(TooLongHandler) as server: self.assertRaises(imaplib.IMAP4.error, self.imap_class, *server.server_address) @threading_helper.reap_threads def test_simple_with_statement(self): # simplest call with self.reaped_server(SimpleIMAPHandler) as server: with self.imap_class(*server.server_address): pass @threading_helper.reap_threads def test_with_statement(self): with self.reaped_server(SimpleIMAPHandler) as server: with self.imap_class(*server.server_address) as imap: imap.login('user', 'pass') self.assertEqual(server.logged, 'user') self.assertIsNone(server.logged) @threading_helper.reap_threads def test_with_statement_logout(self): # what happens if already logout in the block? with self.reaped_server(SimpleIMAPHandler) as server: with self.imap_class(*server.server_address) as imap: imap.login('user', 'pass') self.assertEqual(server.logged, 'user') imap.logout() self.assertIsNone(server.logged) self.assertIsNone(server.logged) @unittest.skipUnless(ssl, "SSL not available") class ThreadedNetworkedTestsSSL(ThreadedNetworkedTests): server_class = SecureTCPServer imap_class = IMAP4_SSL @threading_helper.reap_threads def test_ssl_verified(self): ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) ssl_context.load_verify_locations(CAFILE) with self.assertRaisesRegex( ssl.CertificateError, "IP address mismatch, certificate is not valid for " "'127.0.0.1'"): with self.reaped_server(SimpleIMAPHandler) as server: client = self.imap_class(*server.server_address, ssl_context=ssl_context) client.shutdown() with self.reaped_server(SimpleIMAPHandler) as server: client = self.imap_class("localhost", server.server_address[1], ssl_context=ssl_context) client.shutdown() @unittest.skipUnless( support.is_resource_enabled('network'), 'network resource disabled') class RemoteIMAPTest(unittest.TestCase): host = 'cyrus.andrew.cmu.edu' port = 143 username = 'anonymous' password = 'pass' imap_class = imaplib.IMAP4 def setUp(self): with socket_helper.transient_internet(self.host): self.server = self.imap_class(self.host, self.port) def tearDown(self): if self.server is not None: with socket_helper.transient_internet(self.host): self.server.logout() def test_logincapa(self): with socket_helper.transient_internet(self.host): for cap in self.server.capabilities: self.assertIsInstance(cap, str) self.assertIn('LOGINDISABLED', self.server.capabilities) self.assertIn('AUTH=ANONYMOUS', self.server.capabilities) rs = self.server.login(self.username, self.password) self.assertEqual(rs[0], 'OK') def test_logout(self): with socket_helper.transient_internet(self.host): rs = self.server.logout() self.server = None self.assertEqual(rs[0], 'BYE', rs) @unittest.skipUnless(ssl, "SSL not available") @unittest.skipUnless( support.is_resource_enabled('network'), 'network resource disabled') class RemoteIMAP_STARTTLSTest(RemoteIMAPTest): def setUp(self): super().setUp() with socket_helper.transient_internet(self.host): rs = self.server.starttls() self.assertEqual(rs[0], 'OK') def test_logincapa(self): for cap in self.server.capabilities: self.assertIsInstance(cap, str) self.assertNotIn('LOGINDISABLED', self.server.capabilities) @unittest.skipUnless(ssl, "SSL not available") class RemoteIMAP_SSLTest(RemoteIMAPTest): port = 993 imap_class = IMAP4_SSL def setUp(self): pass def tearDown(self): pass def create_ssl_context(self): ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) ssl_context.check_hostname = False ssl_context.verify_mode = ssl.CERT_NONE ssl_context.load_cert_chain(CERTFILE) return ssl_context def check_logincapa(self, server): try: for cap in server.capabilities: self.assertIsInstance(cap, str) self.assertNotIn('LOGINDISABLED', server.capabilities) self.assertIn('AUTH=PLAIN', server.capabilities) rs = server.login(self.username, self.password) self.assertEqual(rs[0], 'OK') finally: server.logout() def test_logincapa(self): with socket_helper.transient_internet(self.host): _server = self.imap_class(self.host, self.port) self.check_logincapa(_server) def test_logout(self): with socket_helper.transient_internet(self.host): _server = self.imap_class(self.host, self.port) rs = _server.logout() self.assertEqual(rs[0], 'BYE', rs) def test_ssl_context_certfile_exclusive(self): with socket_helper.transient_internet(self.host): self.assertRaises( ValueError, self.imap_class, self.host, self.port, certfile=CERTFILE, ssl_context=self.create_ssl_context()) def test_ssl_context_keyfile_exclusive(self): with socket_helper.transient_internet(self.host): self.assertRaises( ValueError, self.imap_class, self.host, self.port, keyfile=CERTFILE, ssl_context=self.create_ssl_context()) if __name__ == "__main__": unittest.main()