IMAP Relay for gmail (to inspect/debug imap), prints unencrypted data (without TLS) and disables compression

gmail_imap_relay.py

import tornado
from tornado import gen
from tornado.ioloop import IOLoop
from tornado.iostream import IOStream
from tornado.tcpclient import TCPClient
from tornado.tcpserver import TCPServer

COLOR_WHITE = 7
COLOR_RECEIVED = 3
COLOR_SENT = 2
COLOR_RED = 1


def print_line(con_number, color, line):
    print('[{}] \x1b[3{}m{}\x1b[0m'.format(con_number, color, line))


class GmailImapRelayServer(TCPServer):
    def __init__(self, relay_host, relay_port, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.relay_host = relay_host
        self.relay_port = relay_port
        self.connection_number = 0
        self.client = TCPClient(io_loop=kwargs.get('io_loop'))

    @gen.coroutine
    def relay_stream(self, cid, color, direction: str, in_stream: IOStream, out_stream: IOStream):
        while True:
            try:
                result = yield in_stream.read_until(b'\n')
            except tornado.iostream.StreamClosedError:
                print_line(cid, COLOR_RED, '{} stream closed, closing other end'.format(direction))
                out_stream.close()
                break
            print_line(cid, color, result.decode('utf-8', errors='replace').strip())
            if direction == 'received' and result.startswith(b'* CAPABILITY'):
                result = result.replace(B'COMPRESS=DEFLATE ', B'')
                print_line(cid, color, result.decode('utf-8', errors='replace').strip())
            try:
                yield out_stream.write(result)
            except tornado.iostream.StreamClosedError:
                print_line(cid, COLOR_RED, 'Broken pipe in {}>> stream, closing other end'.format(direction))
                in_stream.close()
                break

    @gen.coroutine
    def handle_stream(self, stream: IOStream, address):
        self.connection_number += 1
        cid = self.connection_number
        print_line(cid, COLOR_WHITE, 'New connection received')
        upstream = yield self.client.connect(self.relay_host, self.relay_port, ssl_options={})
        print_line(cid, COLOR_WHITE, 'Upstream connected, starting relay')
        worker1 = self.relay_stream(cid, COLOR_SENT, 'sent', stream, upstream)
        worker2 = self.relay_stream(cid, COLOR_RECEIVED, 'received', upstream, stream)
        # wait for streams to close
        yield worker1
        yield worker2
        print_line(cid, COLOR_WHITE, 'Connection terminated')
        stream.close()
        upstream.close()


def run():
    server = GmailImapRelayServer('imap.gmail.com', 993)
    server.listen(1993, '0.0.0.0')
    print("IMAP Relay is running on %s:%s" % ('0.0.0.0', 1993))
    IOLoop.current().start()


if __name__ == '__main__':
    run()