mirror of
https://github.com/Grasscutters/mitmproxy.git
synced 2024-11-23 08:11:00 +00:00
310 lines
8.4 KiB
Python
310 lines
8.4 KiB
Python
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,
|
|
**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: ProxyConfig(
|
|
upstream_server= (False, False, "127.0.0.1", port),
|
|
http_form_in = "absolute",
|
|
http_form_out = "absolute"
|
|
)] * n
|
|
@classmethod
|
|
def setupAll(cls):
|
|
super(ChainProxTest, cls).setupAll()
|
|
cls.chain = []
|
|
for i in range(cls.n):
|
|
config = cls.chain_config[i](cls.proxy.port if i == 0 else cls.chain[-1].port)
|
|
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
|