mitmproxy/test/tservers.py

import os.path
import threading, Queue
import shutil, tempfile
import flask
import mock

from libmproxy.proxy.config import ProxyConfig
from libmproxy.proxy.server import ProxyServer
from libmproxy.proxy.primitives import TransparentUpstreamServerResolver
import libpathod.test, libpathod.pathoc
from libmproxy import flow, controller
from libmproxy.cmdline import APP_HOST, APP_PORT
import tutils

testapp = flask.Flask(__name__)

@testapp.route("/")
def hello():
    return "testapp"

@testapp.route("/error")
def error():
    raise ValueError("An exception...")


def errapp(environ, start_response):
    raise ValueError("errapp")


class TestMaster(flow.FlowMaster):
    def __init__(self, config):
        s = ProxyServer(config, 0)
        state = flow.State()
        flow.FlowMaster.__init__(self, s, state)
        self.apps.add(testapp, "testapp", 80)
        self.apps.add(errapp, "errapp", 80)
        self.clear_log()

    def handle_request(self, m):
        flow.FlowMaster.handle_request(self, m)
        m.reply()

    def handle_response(self, m):
        flow.FlowMaster.handle_response(self, m)
        m.reply()

    def clear_log(self):
        self.log = []

    def handle_log(self, l):
        self.log.append(l.msg)
        l.reply()


class ProxyThread(threading.Thread):
    def __init__(self, tmaster):
        threading.Thread.__init__(self)
        self.tmaster = tmaster
        self.name = "ProxyThread (%s:%s)" % (tmaster.server.address.host, tmaster.server.address.port)
        controller.should_exit = False

    @property
    def port(self):
        return self.tmaster.server.address.port

    @property
    def log(self):
        return self.tmaster.log

    def run(self):
        self.tmaster.run()

    def shutdown(self):
        self.tmaster.shutdown()


class ProxTestBase(object):
    # Test Configuration
    ssl = None
    ssloptions = False
    clientcerts = False
    no_upstream_cert = False
    authenticator = None
    masterclass = TestMaster
    externalapp = False
    certforward = False
    @classmethod
    def setupAll(cls):
        cls.server = libpathod.test.Daemon(ssl=cls.ssl, ssloptions=cls.ssloptions)
        cls.server2 = libpathod.test.Daemon(ssl=cls.ssl, ssloptions=cls.ssloptions)
        pconf = cls.get_proxy_config()
        cls.confdir = os.path.join(tempfile.gettempdir(), "mitmproxy")
        cls.config = ProxyConfig(
            no_upstream_cert = cls.no_upstream_cert,
            confdir = cls.confdir,
            authenticator = cls.authenticator,
            certforward = cls.certforward,
            ssl_ports=([cls.server.port, cls.server2.port] if cls.ssl else []),
            **pconf
        )
        tmaster = cls.masterclass(cls.config)
        tmaster.start_app(APP_HOST, APP_PORT, cls.externalapp)
        cls.proxy = ProxyThread(tmaster)
        cls.proxy.start()

    @property
    def master(cls):
        return cls.proxy.tmaster

    @classmethod
    def teardownAll(cls):
        shutil.rmtree(cls.confdir)
        cls.proxy.shutdown()
        cls.server.shutdown()
        cls.server2.shutdown()

    def setUp(self):
        self.master.clear_log()
        self.master.state.clear()
        self.server.clear_log()
        self.server2.clear_log()

    @property
    def scheme(self):
        return "https" if self.ssl else "http"

    @property
    def proxies(self):
        """
            The URL base for the server instance.
        """
        return (
            (self.scheme, ("127.0.0.1", self.proxy.port))
        )

    @classmethod
    def get_proxy_config(cls):
        d = dict()
        if cls.clientcerts:
            d["clientcerts"] = tutils.test_data.path("data/clientcert")
        return d


class HTTPProxTest(ProxTestBase):
    def pathoc_raw(self):
        return libpathod.pathoc.Pathoc(("127.0.0.1", self.proxy.port))

    def pathoc(self, sni=None):
        """
            Returns a connected Pathoc instance.
        """
        p = libpathod.pathoc.Pathoc(("localhost", self.proxy.port), ssl=self.ssl, sni=sni)
        if self.ssl:
            p.connect(("127.0.0.1", self.server.port))
        else:
            p.connect()
        return p

    def pathod(self, spec, sni=None):
        """
            Constructs a pathod GET request, with the appropriate base and proxy.
        """
        p = self.pathoc(sni=sni)
        spec = spec.encode("string_escape")
        if self.ssl:
            q = "get:'/p/%s'"%spec
        else:
            q = "get:'%s/p/%s'"%(self.server.urlbase, spec)
        return p.request(q)

    def app(self, page):
        if self.ssl:
            p = libpathod.pathoc.Pathoc(("127.0.0.1", self.proxy.port), True)
            p.connect((APP_HOST, APP_PORT))
            return p.request("get:'/%s'"%page)
        else:
            p = self.pathoc()
            return p.request("get:'http://%s/%s'"%(APP_HOST, page))


class TResolver:
    def __init__(self, port):
        self.port = port

    def original_addr(self, sock):
        return ("127.0.0.1", self.port)


class TransparentProxTest(ProxTestBase):
    ssl = None
    resolver = TResolver

    @classmethod
    @mock.patch("libmproxy.platform.resolver")
    def setupAll(cls, _):
        super(TransparentProxTest, cls).setupAll()
        if cls.ssl:
            ports = [cls.server.port, cls.server2.port]
        else:
            ports = []
        cls.config.get_upstream_server = TransparentUpstreamServerResolver(cls.resolver(cls.server.port), ports)

    @classmethod
    def get_proxy_config(cls):
        d = ProxTestBase.get_proxy_config()
        d["mode"] = "transparent"
        return d

    def pathod(self, spec, sni=None):
        """
            Constructs a pathod GET request, with the appropriate base and proxy.
        """
        if self.ssl:
            p = self.pathoc(sni=sni)
            q = "get:'/p/%s'"%spec
        else:
            p = self.pathoc()
            q = "get:'/p/%s'"%spec
        return p.request(q)

    def pathoc(self, sni=None):
        """
            Returns a connected Pathoc instance.
        """
        p = libpathod.pathoc.Pathoc(("localhost", self.proxy.port), ssl=self.ssl, sni=sni)
        p.connect()
        return p


class ReverseProxTest(ProxTestBase):
    ssl = None
    @classmethod
    def get_proxy_config(cls):
        d = ProxTestBase.get_proxy_config()
        d["upstream_server"] = (
            True if cls.ssl else False,
            True if cls.ssl else False,
            "127.0.0.1",
            cls.server.port
        )
        d["http_form_in"] = "relative"
        d["http_form_out"] = "relative"
        return d

    def pathoc(self, sni=None):
        """
            Returns a connected Pathoc instance.
        """
        p = libpathod.pathoc.Pathoc(("localhost", self.proxy.port), ssl=self.ssl, sni=sni)
        p.connect()
        return p

    def pathod(self, spec, sni=None):
        """
            Constructs a pathod GET request, with the appropriate base and proxy.
        """
        if self.ssl:
            p = self.pathoc(sni=sni)
            q = "get:'/p/%s'"%spec
        else:
            p = self.pathoc()
            q = "get:'/p/%s'"%spec
        return p.request(q)


class ChainProxTest(ProxTestBase):
    """
    Chain n instances of mitmproxy in a row - because we can.
    """
    n = 2
    chain_config = [lambda port, sslports: ProxyConfig(
        upstream_server= (False, False, "127.0.0.1", port),
        http_form_in = "absolute",
        http_form_out = "absolute",
        ssl_ports=sslports
    )] * n
    @classmethod
    def setupAll(cls):
        super(ChainProxTest, cls).setupAll()
        cls.chain = []
        for i in range(cls.n):
            sslports = [cls.server.port, cls.server2.port]
            config = cls.chain_config[i](cls.proxy.port if i == 0 else cls.chain[-1].port,
                                         sslports)
            tmaster = cls.masterclass(config)
            tmaster.start_app(APP_HOST, APP_PORT, cls.externalapp)
            cls.chain.append(ProxyThread(tmaster))
            cls.chain[-1].start()

    @classmethod
    def teardownAll(cls):
        super(ChainProxTest, cls).teardownAll()
        for p in cls.chain:
            p.tmaster.shutdown()

    def setUp(self):
        super(ChainProxTest, self).setUp()
        for p in self.chain:
            p.tmaster.clear_log()
            p.tmaster.state.clear()


class HTTPChainProxyTest(ChainProxTest):
    def pathoc(self, sni=None):
        """
            Returns a connected Pathoc instance.
        """
        p = libpathod.pathoc.Pathoc(("localhost", self.chain[-1].port), ssl=self.ssl, sni=sni)
        if self.ssl:
            p.connect(("127.0.0.1", self.server.port))
        else:
            p.connect()
        return p