Grasscutters/mitmproxy
3
0
Fork 0
mirror of https://github.com/Grasscutters/mitmproxy.git synced 2024-11-23 00:01:36 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #233 from mitmproxy/untangle_circular_dependencies

Browse Source 
Untangle circular dependencies
This commit is contained in:
Aldo Cortesi 2014-03-10 17:24:05 +13:00
commit 554deee222
23 changed files with 690 additions and 668 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
libmproxy/app.py
View File

@ -17,12 +17,12 @@ def index():
@mapp.route("/cert/pem")
def certs_pem():
p = os.path.join(master().server.config.confdir, proxy.CONF_BASENAME + "-ca-cert.pem")
p = os.path.join(master().server.config.confdir, proxy.config.CONF_BASENAME + "-ca-cert.pem")
return flask.Response(open(p, "rb").read(), mimetype='application/x-x509-ca-cert')
@mapp.route("/cert/p12")
def certs_p12():
p = os.path.join(master().server.config.confdir, proxy.CONF_BASENAME + "-ca-cert.p12")
p = os.path.join(master().server.config.confdir, proxy.config.CONF_BASENAME + "-ca-cert.p12")
return flask.Response(open(p, "rb").read(), mimetype='application/x-pkcs12')

4
libmproxy/cmdline.py
View File

@ -1,4 +1,4 @@
import proxy
from . import proxy
import re, filt
import argparse
@ -387,4 +387,4 @@ def common_options(parser):
help="Allow access to users specified in an Apache htpasswd file."
)
proxy.ssl_option_group(parser)
proxy.config.ssl_option_group(parser)

5
libmproxy/console/common.py
View File

@ -1,6 +1,7 @@
import urwid
import urwid.util
from .. import utils, flow
from .. import utils
from ..protocol.http import CONTENT_MISSING
VIEW_LIST = 0
@ -183,7 +184,7 @@ def format_flow(f, focus, extended=False, hostheader=False, padding=2):
if f.response:
if f.response.content:
contentdesc = utils.pretty_size(len(f.response.content))
elif f.response.content == flow.CONTENT_MISSING:
elif f.response.content == CONTENT_MISSING:
contentdesc = "[content missing]"
else:
contentdesc = "[no content]"

7
libmproxy/console/flowview.py
View File

@ -2,6 +2,7 @@ import os, sys, copy
import urwid
import common, grideditor, contentview
from .. import utils, flow, controller
from ..protocol.http import CONTENT_MISSING
class SearchError(Exception): pass
@ -150,7 +151,7 @@ class FlowView(common.WWrap):
return (description, text_objects)
def cont_view_handle_missing(self, conn, viewmode):
if conn.content == flow.CONTENT_MISSING:
if conn.content == CONTENT_MISSING:
msg, body = "", [urwid.Text([("error", "[content missing]")])], 0
else:
msg, body = self.content_view(viewmode, conn)
@ -178,7 +179,7 @@ class FlowView(common.WWrap):
override = self.override_get()
viewmode = self.viewmode_get(override)
msg, body = self.cont_view_handle_missing(conn, viewmode)
elif conn.content == flow.CONTENT_MISSING:
elif conn.content == CONTENT_MISSING:
pass
return headers, msg, body
@ -643,7 +644,7 @@ class FlowView(common.WWrap):
def delete_body(self, t):
if t == "m":
val = flow.CONTENT_MISSING
val = CONTENT_MISSING
else:
val = None
if self.state.view_flow_mode == common.VIEW_FLOW_REQUEST:

41
libmproxy/flow.py
View File

@ -7,14 +7,15 @@ import hashlib, Cookie, cookielib, re, threading
import os
import flask
import requests
from . import controller, protocol
from .protocol import http
from .proxy.connection import ServerConnection
from .proxy.primitives import ProxyError
import tnetstring, filt, script
from netlib import odict, wsgi
from .proxy import ClientConnection, ServerConnection # FIXME: remove circular dependency
import controller, version, protocol
from netlib import odict, wsgi, tcp
import netlib.http
import version
import app
from .protocol import KILL
from .protocol.http import HTTPResponse, CONTENT_MISSING
from .proxy import RequestReplayThread
ODict = odict.ODict
ODictCaseless = odict.ODictCaseless
@ -565,7 +566,7 @@ class FlowMaster(controller.Master):
rflow = self.server_playback.next_flow(flow)
if not rflow:
return None
response = HTTPResponse._from_state(rflow.response._get_state())
response = http.HTTPResponse._from_state(rflow.response._get_state())
response.is_replay = True
if self.refresh_server_playback:
response.refresh()
@ -642,7 +643,7 @@ class FlowMaster(controller.Master):
"""
if f.intercepting:
return "Can't replay while intercepting..."
if f.request.content == CONTENT_MISSING:
if f.request.content == http.CONTENT_MISSING:
return "Can't replay request with missing content..."
if f.request:
f.request.is_replay = True
@ -693,7 +694,7 @@ class FlowMaster(controller.Master):
err = app.serve(r, r.flow.client_conn.wfile, **{"mitmproxy.master": self})
if err:
self.add_event("Error in wsgi app. %s"%err, "error")
r.reply(KILL)
r.reply(protocol.KILL)
return
f = self.state.add_request(r)
self.replacehooks.run(f)
@ -785,3 +786,25 @@ class FilteredFlowWriter:
d = f._get_state()
tnetstring.dump(d, self.fo)
class RequestReplayThread(threading.Thread):
name="RequestReplayThread"
def __init__(self, config, flow, masterq):
self.config, self.flow, self.channel = config, flow, controller.Channel(masterq)
threading.Thread.__init__(self)
def run(self):
try:
r = self.flow.request
server = ServerConnection(self.flow.server_conn.address(), None)
server.connect()
if self.flow.server_conn.ssl_established:
server.establish_ssl(self.config.clientcerts,
self.flow.server_conn.sni)
server.send(r._assemble())
self.flow.response = http.HTTPResponse.from_stream(server.rfile, r.method, body_size_limit=self.config.body_size_limit)
self.channel.ask("response", self.flow.response)
except (ProxyError, netlib.http.HttpError, tcp.NetLibError), v:
self.flow.error = protocol.primitives.Error(str(v))
self.channel.ask("error", self.flow.error)

9
libmproxy/protocol/__init__.py
View File

@ -1,14 +1,7 @@
from ..proxy import ServerConnection, AddressPriority
from libmproxy.proxy.primitives import AddressPriority
KILL = 0 # const for killed requests
class ConnectionTypeChange(Exception):
"""
Gets raised if the connetion type has been changed (e.g. after HTTP/1.1 101 Switching Protocols).
It's up to the raising ProtocolHandler to specify the new conntype before raising the exception.
"""
pass
class ProtocolHandler(object):
def __init__(self, c):

10
libmproxy/protocol/http.py
View File

@ -1,11 +1,13 @@
import Cookie, urllib, urlparse, time, copy
from email.utils import parsedate_tz, formatdate, mktime_tz
from libmproxy.proxy.primitives import AddressPriority
from ..proxy.connection import ServerConnection
from ..proxy.primitives import ProxyError, ConnectionTypeChange
import netlib.utils
from netlib import http, tcp, http_status, odict
from netlib import http, tcp, http_status
from netlib.odict import ODict, ODictCaseless
from . import ProtocolHandler, ConnectionTypeChange, KILL, TemporaryServerChangeMixin
from .. import encoding, utils, version, filt, controller, stateobject
from ..proxy import ProxyError, AddressPriority, ServerConnection
from . import ProtocolHandler, KILL, TemporaryServerChangeMixin
from .. import encoding, utils, filt, controller, stateobject
from .primitives import Flow, Error

2
libmproxy/protocol/primitives.py
View File

@ -1,5 +1,5 @@
from .. import stateobject, utils, version
from ..proxy import ServerConnection, ClientConnection
from ..proxy.connection import ClientConnection, ServerConnection
import copy

602
libmproxy/proxy.py
View File

@ -1,602 +0,0 @@
import os, socket, time, threading, copy
from OpenSSL import SSL
from netlib import tcp, http, certutils, http_auth
import utils, version, platform, controller, stateobject
TRANSPARENT_SSL_PORTS = [443, 8443]
CONF_BASENAME = "mitmproxy"
CONF_DIR = "~/.mitmproxy"
CA_CERT_NAME = "mitmproxy-ca.pem"
class AddressPriority(object):
"""
Enum that signifies the priority of the given address when choosing the destination host.
Higher is better (None < i)
"""
FORCE = 5
"""forward mode"""
MANUALLY_CHANGED = 4
"""user changed the target address in the ui"""
FROM_SETTINGS = 3
"""reverse proxy mode"""
FROM_CONNECTION = 2
"""derived from transparent resolver"""
FROM_PROTOCOL = 1
"""derived from protocol (e.g. absolute-form http requests)"""
class ProxyError(Exception):
def __init__(self, code, msg, headers=None):
self.code, self.msg, self.headers = code, msg, headers
def __str__(self):
return "ProxyError(%s, %s)" % (self.code, self.msg)
class Log:
def __init__(self, msg):
self.msg = msg
class ProxyConfig:
def __init__(self, confdir=CONF_DIR, clientcerts=None,
no_upstream_cert=False, body_size_limit=None, reverse_proxy=None,
forward_proxy=None, transparent_proxy=None, authenticator=None,
ciphers=None, certs=None
):
self.ciphers = ciphers
self.clientcerts = clientcerts
self.no_upstream_cert = no_upstream_cert
self.body_size_limit = body_size_limit
self.reverse_proxy = reverse_proxy
self.forward_proxy = forward_proxy
self.transparent_proxy = transparent_proxy
self.authenticator = authenticator
self.confdir = os.path.expanduser(confdir)
self.certstore = certutils.CertStore.from_store(self.confdir, CONF_BASENAME)
class ClientConnection(tcp.BaseHandler, stateobject.SimpleStateObject):
def __init__(self, client_connection, address, server):
if client_connection: # Eventually, this object is restored from state. We don't have a connection then.
tcp.BaseHandler.__init__(self, client_connection, address, server)
else:
self.connection = None
self.server = None
self.wfile = None
self.rfile = None
self.address = None
self.clientcert = None
self.timestamp_start = utils.timestamp()
self.timestamp_end = None
self.timestamp_ssl_setup = None
_stateobject_attributes = dict(
timestamp_start=float,
timestamp_end=float,
timestamp_ssl_setup=float
)
def _get_state(self):
d = super(ClientConnection, self)._get_state()
d.update(
address={"address": self.address(), "use_ipv6": self.address.use_ipv6},
clientcert=self.cert.to_pem() if self.clientcert else None
)
return d
def _load_state(self, state):
super(ClientConnection, self)._load_state(state)
self.address = tcp.Address(**state["address"]) if state["address"] else None
self.clientcert = certutils.SSLCert.from_pem(state["clientcert"]) if state["clientcert"] else None
def copy(self):
return copy.copy(self)
def send(self, message):
self.wfile.write(message)
self.wfile.flush()
@classmethod
def _from_state(cls, state):
f = cls(None, tuple(), None)
f._load_state(state)
return f
def convert_to_ssl(self, *args, **kwargs):
tcp.BaseHandler.convert_to_ssl(self, *args, **kwargs)
self.timestamp_ssl_setup = utils.timestamp()
def finish(self):
tcp.BaseHandler.finish(self)
self.timestamp_end = utils.timestamp()
class ServerConnection(tcp.TCPClient, stateobject.SimpleStateObject):
def __init__(self, address, priority):
tcp.TCPClient.__init__(self, address)
self.priority = priority
self.peername = None
self.timestamp_start = None
self.timestamp_end = None
self.timestamp_tcp_setup = None
self.timestamp_ssl_setup = None
_stateobject_attributes = dict(
peername=tuple,
timestamp_start=float,
timestamp_end=float,
timestamp_tcp_setup=float,
timestamp_ssl_setup=float,
address=tcp.Address,
source_address=tcp.Address,
cert=certutils.SSLCert,
ssl_established=bool,
sni=str
)
def _get_state(self):
d = super(ServerConnection, self)._get_state()
d.update(
address={"address": self.address(), "use_ipv6": self.address.use_ipv6},
source_address= {"address": self.source_address(),
"use_ipv6": self.source_address.use_ipv6} if self.source_address else None,
cert=self.cert.to_pem() if self.cert else None
)
return d
def _load_state(self, state):
super(ServerConnection, self)._load_state(state)
self.address = tcp.Address(**state["address"]) if state["address"] else None
self.source_address = tcp.Address(**state["source_address"]) if state["source_address"] else None
self.cert = certutils.SSLCert.from_pem(state["cert"]) if state["cert"] else None
@classmethod
def _from_state(cls, state):
f = cls(tuple(), None)
f._load_state(state)
return f
def copy(self):
return copy.copy(self)
def connect(self):
self.timestamp_start = utils.timestamp()
tcp.TCPClient.connect(self)
self.peername = self.connection.getpeername()
self.timestamp_tcp_setup = utils.timestamp()
def send(self, message):
self.wfile.write(message)
self.wfile.flush()
def establish_ssl(self, clientcerts, sni):
clientcert = None
if clientcerts:
path = os.path.join(clientcerts, self.address.host.encode("idna")) + ".pem"
if os.path.exists(path):
clientcert = path
try:
self.convert_to_ssl(cert=clientcert, sni=sni)
self.timestamp_ssl_setup = utils.timestamp()
except tcp.NetLibError, v:
raise ProxyError(400, str(v))
def finish(self):
tcp.TCPClient.finish(self)
self.timestamp_end = utils.timestamp()
from . import protocol
from .protocol.http import HTTPResponse
class RequestReplayThread(threading.Thread):
name="RequestReplayThread"
def __init__(self, config, flow, masterq):
self.config, self.flow, self.channel = config, flow, controller.Channel(masterq)
threading.Thread.__init__(self)
def run(self):
try:
r = self.flow.request
server = ServerConnection(self.flow.server_conn.address(), None)
server.connect()
if self.flow.server_conn.ssl_established:
server.establish_ssl(self.config.clientcerts,
self.flow.server_conn.sni)
server.send(r._assemble())
self.flow.response = HTTPResponse.from_stream(server.rfile, r.method, body_size_limit=self.config.body_size_limit)
self.channel.ask("response", self.flow.response)
except (ProxyError, http.HttpError, tcp.NetLibError), v:
self.flow.error = protocol.primitives.Error(str(v))
self.channel.ask("error", self.flow.error)
class ConnectionHandler:
def __init__(self, config, client_connection, client_address, server, channel, server_version):
self.config = config
self.client_conn = ClientConnection(client_connection, client_address, server)
self.server_conn = None
self.channel, self.server_version = channel, server_version
self.close = False
self.conntype = None
self.sni = None
self.mode = "regular"
if self.config.reverse_proxy:
self.mode = "reverse"
if self.config.transparent_proxy:
self.mode = "transparent"
def handle(self):
self.log("clientconnect")
self.channel.ask("clientconnect", self)
self.determine_conntype()
try:
try:
# Can we already identify the target server and connect to it?
server_address = None
address_priority = None
if self.config.forward_proxy:
server_address = self.config.forward_proxy[1:]
address_priority = AddressPriority.FORCE
elif self.config.reverse_proxy:
server_address = self.config.reverse_proxy[1:]
address_priority = AddressPriority.FROM_SETTINGS
elif self.config.transparent_proxy:
server_address = self.config.transparent_proxy["resolver"].original_addr(
self.client_conn.connection)
if not server_address:
raise ProxyError(502, "Transparent mode failure: could not resolve original destination.")
address_priority = AddressPriority.FROM_CONNECTION
self.log("transparent to %s:%s" % server_address)
if server_address:
self.set_server_address(server_address, address_priority)
self._handle_ssl()
while not self.close:
try:
protocol.handle_messages(self.conntype, self)
except protocol.ConnectionTypeChange:
self.log("Connection Type Changed: %s" % self.conntype)
continue
# FIXME: Do we want to persist errors?
except (ProxyError, tcp.NetLibError), e:
protocol.handle_error(self.conntype, self, e)
except Exception, e:
self.log(e.__class__)
import traceback
self.log(traceback.format_exc())
self.log(str(e))
self.del_server_connection()
self.log("clientdisconnect")
self.channel.tell("clientdisconnect", self)
def _handle_ssl(self):
"""
Helper function of .handle()
Check if we can already identify SSL connections.
If so, connect to the server and establish an SSL connection
"""
client_ssl = False
server_ssl = False
if self.config.transparent_proxy:
client_ssl = server_ssl = (self.server_conn.address.port in self.config.transparent_proxy["sslports"])
elif self.config.reverse_proxy:
client_ssl = server_ssl = (self.config.reverse_proxy[0] == "https")
# TODO: Make protocol generic (as with transparent proxies)
# TODO: Add SSL-terminating capatbility (SSL -> mitmproxy -> plain and vice versa)
if client_ssl or server_ssl:
self.establish_server_connection()
self.establish_ssl(client=client_ssl, server=server_ssl)
def del_server_connection(self):
"""
Deletes an existing server connection.
"""
if self.server_conn and self.server_conn.connection:
self.server_conn.finish()
self.log("serverdisconnect", ["%s:%s" % (self.server_conn.address.host, self.server_conn.address.port)])
self.channel.tell("serverdisconnect", self)
self.server_conn = None
self.sni = None
def determine_conntype(self):
#TODO: Add ruleset to select correct protocol depending on mode/target port etc.
self.conntype = "http"
def set_server_address(self, address, priority):
"""
Sets a new server address with the given priority.
Does not re-establish either connection or SSL handshake.
@type priority: AddressPriority
"""
address = tcp.Address.wrap(address)
if self.server_conn:
if self.server_conn.priority > priority:
self.log("Attempt to change server address, "
"but priority is too low (is: %s, got: %s)" % (self.server_conn.priority, priority))
return
if self.server_conn.address == address:
self.server_conn.priority = priority # Possibly increase priority
return
self.del_server_connection()
self.log("Set new server address: %s:%s" % (address.host, address.port))
self.server_conn = ServerConnection(address, priority)
def establish_server_connection(self):
"""
Establishes a new server connection.
If there is already an existing server connection, the function returns immediately.
"""
if self.server_conn.connection:
return
self.log("serverconnect", ["%s:%s" % self.server_conn.address()[:2]])
self.channel.tell("serverconnect", self)
try:
self.server_conn.connect()
except tcp.NetLibError, v:
raise ProxyError(502, v)
def establish_ssl(self, client=False, server=False):
"""
Establishes SSL on the existing connection(s) to the server or the client,
as specified by the parameters. If the target server is on the pass-through list,
the conntype attribute will be changed and the SSL connection won't be wrapped.
A protocol handler must raise a ConnTypeChanged exception if it detects that this is happening
"""
# TODO: Implement SSL pass-through handling and change conntype
passthrough = [
# "echo.websocket.org",
# "174.129.224.73" # echo.websocket.org, transparent mode
]
if self.server_conn.address.host in passthrough or self.sni in passthrough:
self.conntype = "tcp"
return
# Logging
if client or server:
subs = []
if client:
subs.append("with client")
if server:
subs.append("with server (sni: %s)" % self.sni)
self.log("Establish SSL", subs)
if server:
if self.server_conn.ssl_established:
raise ProxyError(502, "SSL to Server already established.")
self.establish_server_connection() # make sure there is a server connection.
self.server_conn.establish_ssl(self.config.clientcerts, self.sni)
if client:
if self.client_conn.ssl_established:
raise ProxyError(502, "SSL to Client already established.")
cert, key = self.find_cert()
self.client_conn.convert_to_ssl(
cert, key,
handle_sni = self.handle_sni,
cipher_list = self.config.ciphers
)
def server_reconnect(self, no_ssl=False):
address = self.server_conn.address
had_ssl = self.server_conn.ssl_established
priority = self.server_conn.priority
sni = self.sni
self.log("(server reconnect follows)")
self.del_server_connection()
self.set_server_address(address, priority)
self.establish_server_connection()
if had_ssl and not no_ssl:
self.sni = sni
self.establish_ssl(server=True)
def finish(self):
self.client_conn.finish()
def log(self, msg, subs=()):
msg = [
"%s:%s: %s" % (self.client_conn.address.host, self.client_conn.address.port, msg)
]
for i in subs:
msg.append(" -> " + i)
msg = "\n".join(msg)
self.channel.tell("log", Log(msg))
def find_cert(self):
host = self.server_conn.address.host
sans = []
if not self.config.no_upstream_cert or not self.server_conn.ssl_established:
upstream_cert = self.server_conn.cert
if upstream_cert.cn:
host = upstream_cert.cn.decode("utf8").encode("idna")
sans = upstream_cert.altnames
ret = self.config.certstore.get_cert(host, sans)
if not ret:
raise ProxyError(502, "Unable to generate dummy cert.")
return ret
def handle_sni(self, connection):
"""
This callback gets called during the SSL handshake with the client.
The client has just sent the Sever Name Indication (SNI). We now connect upstream to
figure out which certificate needs to be served.
"""
try:
sn = connection.get_servername()
if sn and sn != self.sni:
self.sni = sn.decode("utf8").encode("idna")
self.log("SNI received: %s" % self.sni)
self.server_reconnect() # reconnect to upstream server with SNI
# Now, change client context to reflect changed certificate:
new_context = SSL.Context(SSL.TLSv1_METHOD)
cert, key = self.find_cert()
new_context.use_privatekey_file(key)
new_context.use_certificate(cert.X509)
connection.set_context(new_context)
# An unhandled exception in this method will core dump PyOpenSSL, so
# make dang sure it doesn't happen.
except Exception, e: # pragma: no cover
pass
class ProxyServerError(Exception):
pass
class ProxyServer(tcp.TCPServer):
allow_reuse_address = True
bound = True
def __init__(self, config, port, host='', server_version=version.NAMEVERSION):
"""
Raises ProxyServerError if there's a startup problem.
"""
self.config = config
self.server_version = server_version
try:
tcp.TCPServer.__init__(self, (host, port))
except socket.error, v:
raise ProxyServerError('Error starting proxy server: ' + v.strerror)
self.channel = None
def start_slave(self, klass, channel):
slave = klass(channel, self)
slave.start()
def set_channel(self, channel):
self.channel = channel
def handle_client_connection(self, conn, client_address):
h = ConnectionHandler(self.config, conn, client_address, self, self.channel, self.server_version)
h.handle()
h.finish()
class DummyServer:
bound = False
def __init__(self, config):
self.config = config
def start_slave(self, *args):
pass
def shutdown(self):
pass
# Command-line utils
def ssl_option_group(parser):
group = parser.add_argument_group("SSL")
group.add_argument(
"--cert", dest='certs', default=[], type=str,
metavar = "SPEC", action="append",
help='Add an SSL certificate. SPEC is of the form "[domain=]path". '\
'The domain may include a wildcard, and is equal to "*" if not specified. '\
'The file at path is a certificate in PEM format. If a private key is included in the PEM, '\
'it is used, else the default key in the conf dir is used. Can be passed multiple times.'
)
group.add_argument(
"--client-certs", action="store",
type=str, dest="clientcerts", default=None,
help="Client certificate directory."
)
group.add_argument(
"--ciphers", action="store",
type=str, dest="ciphers", default=None,
help="SSL cipher specification."
)
def process_proxy_options(parser, options):
body_size_limit = utils.parse_size(options.body_size_limit)
if options.reverse_proxy and options.transparent_proxy:
return parser.error("Can't set both reverse proxy and transparent proxy.")
if options.transparent_proxy:
if not platform.resolver:
return parser.error("Transparent mode not supported on this platform.")
trans = dict(
resolver=platform.resolver(),
sslports=TRANSPARENT_SSL_PORTS
)
else:
trans = None
if options.reverse_proxy:
rp = utils.parse_proxy_spec(options.reverse_proxy)
if not rp:
return parser.error("Invalid reverse proxy specification: %s" % options.reverse_proxy)
else:
rp = None
if options.forward_proxy:
fp = utils.parse_proxy_spec(options.forward_proxy)
if not fp:
return parser.error("Invalid forward proxy specification: %s" % options.forward_proxy)
else:
fp = None
if options.clientcerts:
options.clientcerts = os.path.expanduser(options.clientcerts)
if not os.path.exists(options.clientcerts) or not os.path.isdir(options.clientcerts):
return parser.error(
"Client certificate directory does not exist or is not a directory: %s" % options.clientcerts
)
if (options.auth_nonanonymous or options.auth_singleuser or options.auth_htpasswd):
if options.auth_singleuser:
if len(options.auth_singleuser.split(':')) != 2:
return parser.error("Invalid single-user specification. Please use the format username:password")
username, password = options.auth_singleuser.split(':')
password_manager = http_auth.PassManSingleUser(username, password)
elif options.auth_nonanonymous:
password_manager = http_auth.PassManNonAnon()
elif options.auth_htpasswd:
try:
password_manager = http_auth.PassManHtpasswd(options.auth_htpasswd)
except ValueError, v:
return parser.error(v.message)
authenticator = http_auth.BasicProxyAuth(password_manager, "mitmproxy")
else:
authenticator = http_auth.NullProxyAuth(None)
certs = []
for i in options.certs:
parts = i.split("=", 1)
if len(parts) == 1:
parts = ["*", parts[0]]
parts[1] = os.path.expanduser(parts[1])
if not os.path.exists(parts[1]):
parser.error("Certificate file does not exist: %s"%parts[1])
certs.append(parts)
return ProxyConfig(
clientcerts=options.clientcerts,
body_size_limit=body_size_limit,
no_upstream_cert=options.no_upstream_cert,
reverse_proxy=rp,
forward_proxy=fp,
transparent_proxy=trans,
authenticator=authenticator,
ciphers=options.ciphers,
certs = certs,
)

0
libmproxy/proxy/__init__.py Normal file
View File

124
libmproxy/proxy/config.py Normal file
View File

@ -0,0 +1,124 @@
import os
from .. import utils, platform
from netlib import http_auth, certutils
TRANSPARENT_SSL_PORTS = [443, 8443]
CONF_BASENAME = "mitmproxy"
CONF_DIR = "~/.mitmproxy"
class ProxyConfig:
def __init__(self, confdir=CONF_DIR, clientcerts=None,
no_upstream_cert=False, body_size_limit=None, reverse_proxy=None,
forward_proxy=None, transparent_proxy=None, authenticator=None,
ciphers=None, certs=None
):
self.ciphers = ciphers
self.clientcerts = clientcerts
self.no_upstream_cert = no_upstream_cert
self.body_size_limit = body_size_limit
self.reverse_proxy = reverse_proxy
self.forward_proxy = forward_proxy
self.transparent_proxy = transparent_proxy
self.authenticator = authenticator
self.confdir = os.path.expanduser(confdir)
self.certstore = certutils.CertStore.from_store(self.confdir, CONF_BASENAME)
def process_proxy_options(parser, options):
body_size_limit = utils.parse_size(options.body_size_limit)
if options.reverse_proxy and options.transparent_proxy:
return parser.error("Can't set both reverse proxy and transparent proxy.")
if options.transparent_proxy:
if not platform.resolver:
return parser.error("Transparent mode not supported on this platform.")
trans = dict(
resolver=platform.resolver(),
sslports=TRANSPARENT_SSL_PORTS
)
else:
trans = None
if options.reverse_proxy:
rp = utils.parse_proxy_spec(options.reverse_proxy)
if not rp:
return parser.error("Invalid reverse proxy specification: %s" % options.reverse_proxy)
else:
rp = None
if options.forward_proxy:
fp = utils.parse_proxy_spec(options.forward_proxy)
if not fp:
return parser.error("Invalid forward proxy specification: %s" % options.forward_proxy)
else:
fp = None
if options.clientcerts:
options.clientcerts = os.path.expanduser(options.clientcerts)
if not os.path.exists(options.clientcerts) or not os.path.isdir(options.clientcerts):
return parser.error(
"Client certificate directory does not exist or is not a directory: %s" % options.clientcerts
)
if (options.auth_nonanonymous or options.auth_singleuser or options.auth_htpasswd):
if options.auth_singleuser:
if len(options.auth_singleuser.split(':')) != 2:
return parser.error("Invalid single-user specification. Please use the format username:password")
username, password = options.auth_singleuser.split(':')
password_manager = http_auth.PassManSingleUser(username, password)
elif options.auth_nonanonymous:
password_manager = http_auth.PassManNonAnon()
elif options.auth_htpasswd:
try:
password_manager = http_auth.PassManHtpasswd(options.auth_htpasswd)
except ValueError, v:
return parser.error(v.message)
authenticator = http_auth.BasicProxyAuth(password_manager, "mitmproxy")
else:
authenticator = http_auth.NullProxyAuth(None)
certs = []
for i in options.certs:
parts = i.split("=", 1)
if len(parts) == 1:
parts = ["*", parts[0]]
parts[1] = os.path.expanduser(parts[1])
if not os.path.exists(parts[1]):
parser.error("Certificate file does not exist: %s"%parts[1])
certs.append(parts)
return ProxyConfig(
clientcerts=options.clientcerts,
body_size_limit=body_size_limit,
no_upstream_cert=options.no_upstream_cert,
reverse_proxy=rp,
forward_proxy=fp,
transparent_proxy=trans,
authenticator=authenticator,
ciphers=options.ciphers,
certs = certs,
)
def ssl_option_group(parser):
group = parser.add_argument_group("SSL")
group.add_argument(
"--cert", dest='certs', default=[], type=str,
metavar = "SPEC", action="append",
help='Add an SSL certificate. SPEC is of the form "[domain=]path". '\
'The domain may include a wildcard, and is equal to "*" if not specified. '\
'The file at path is a certificate in PEM format. If a private key is included in the PEM, '\
'it is used, else the default key in the conf dir is used. Can be passed multiple times.'
)
group.add_argument(
"--client-certs", action="store",
type=str, dest="clientcerts", default=None,
help="Client certificate directory."
)
group.add_argument(
"--ciphers", action="store",
type=str, dest="ciphers", default=None,
help="SSL cipher specification."
)

139
libmproxy/proxy/connection.py Normal file
View File

@ -0,0 +1,139 @@
import copy
import os
from .. import stateobject, utils
from .primitives import ProxyError
from netlib import tcp, certutils
class ClientConnection(tcp.BaseHandler, stateobject.SimpleStateObject):
def __init__(self, client_connection, address, server):
if client_connection: # Eventually, this object is restored from state. We don't have a connection then.
tcp.BaseHandler.__init__(self, client_connection, address, server)
else:
self.connection = None
self.server = None
self.wfile = None
self.rfile = None
self.address = None
self.clientcert = None
self.timestamp_start = utils.timestamp()
self.timestamp_end = None
self.timestamp_ssl_setup = None
_stateobject_attributes = dict(
timestamp_start=float,
timestamp_end=float,
timestamp_ssl_setup=float
)
def _get_state(self):
d = super(ClientConnection, self)._get_state()
d.update(
address={"address": self.address(), "use_ipv6": self.address.use_ipv6},
clientcert=self.cert.to_pem() if self.clientcert else None
)
return d
def _load_state(self, state):
super(ClientConnection, self)._load_state(state)
self.address = tcp.Address(**state["address"]) if state["address"] else None
self.clientcert = certutils.SSLCert.from_pem(state["clientcert"]) if state["clientcert"] else None
def copy(self):
return copy.copy(self)
def send(self, message):
self.wfile.write(message)
self.wfile.flush()
@classmethod
def _from_state(cls, state):
f = cls(None, tuple(), None)
f._load_state(state)
return f
def convert_to_ssl(self, *args, **kwargs):
tcp.BaseHandler.convert_to_ssl(self, *args, **kwargs)
self.timestamp_ssl_setup = utils.timestamp()
def finish(self):
tcp.BaseHandler.finish(self)
self.timestamp_end = utils.timestamp()
class ServerConnection(tcp.TCPClient, stateobject.SimpleStateObject):
def __init__(self, address, priority):
tcp.TCPClient.__init__(self, address)
self.priority = priority
self.peername = None
self.timestamp_start = None
self.timestamp_end = None
self.timestamp_tcp_setup = None
self.timestamp_ssl_setup = None
_stateobject_attributes = dict(
peername=tuple,
timestamp_start=float,
timestamp_end=float,
timestamp_tcp_setup=float,
timestamp_ssl_setup=float,
address=tcp.Address,
source_address=tcp.Address,
cert=certutils.SSLCert,
ssl_established=bool,
sni=str
)
def _get_state(self):
d = super(ServerConnection, self)._get_state()
d.update(
address={"address": self.address(), "use_ipv6": self.address.use_ipv6},
source_address= {"address": self.source_address(),
"use_ipv6": self.source_address.use_ipv6} if self.source_address else None,
cert=self.cert.to_pem() if self.cert else None
)
return d
def _load_state(self, state):
super(ServerConnection, self)._load_state(state)
self.address = tcp.Address(**state["address"]) if state["address"] else None
self.source_address = tcp.Address(**state["source_address"]) if state["source_address"] else None
self.cert = certutils.SSLCert.from_pem(state["cert"]) if state["cert"] else None
@classmethod
def _from_state(cls, state):
f = cls(tuple(), None)
f._load_state(state)
return f
def copy(self):
return copy.copy(self)
def connect(self):
self.timestamp_start = utils.timestamp()
tcp.TCPClient.connect(self)
self.peername = self.connection.getpeername()
self.timestamp_tcp_setup = utils.timestamp()
def send(self, message):
self.wfile.write(message)
self.wfile.flush()
def establish_ssl(self, clientcerts, sni):
clientcert = None
if clientcerts:
path = os.path.join(clientcerts, self.address.host.encode("idna")) + ".pem"
if os.path.exists(path):
clientcert = path
try:
self.convert_to_ssl(cert=clientcert, sni=sni)
self.timestamp_ssl_setup = utils.timestamp()
except tcp.NetLibError, v:
raise ProxyError(400, str(v))
def finish(self):
tcp.TCPClient.finish(self)
self.timestamp_end = utils.timestamp()

40
libmproxy/proxy/primitives.py Normal file
View File

@ -0,0 +1,40 @@
class ProxyError(Exception):
def __init__(self, code, msg, headers=None):
self.code, self.msg, self.headers = code, msg, headers
def __str__(self):
return "ProxyError(%s, %s)" % (self.code, self.msg)
class ConnectionTypeChange(Exception):
"""
Gets raised if the connection type has been changed (e.g. after HTTP/1.1 101 Switching Protocols).
It's up to the raising ProtocolHandler to specify the new conntype before raising the exception.
"""
pass
class ProxyServerError(Exception):
pass
class AddressPriority(object):
"""
Enum that signifies the priority of the given address when choosing the destination host.
Higher is better (None < i)
"""
FORCE = 5
"""forward mode"""
MANUALLY_CHANGED = 4
"""user changed the target address in the ui"""
FROM_SETTINGS = 3
"""reverse proxy mode"""
FROM_CONNECTION = 2
"""derived from transparent resolver"""
FROM_PROTOCOL = 1
"""derived from protocol (e.g. absolute-form http requests)"""
class Log:
def __init__(self, msg):
self.msg = msg

287
libmproxy/proxy/server.py Normal file
View File

@ -0,0 +1,287 @@
import socket
from .. import version, protocol
from libmproxy.proxy.primitives import Log
from .primitives import ProxyServerError
from .connection import ClientConnection, ServerConnection
from .primitives import ProxyError, ConnectionTypeChange, AddressPriority
from netlib import tcp
class DummyServer:
bound = False
def __init__(self, config):
self.config = config
def start_slave(self, *args):
pass
def shutdown(self):
pass
class ProxyServer(tcp.TCPServer):
allow_reuse_address = True
bound = True
def __init__(self, config, port, host='', server_version=version.NAMEVERSION):
"""
Raises ProxyServerError if there's a startup problem.
"""
self.config = config
self.server_version = server_version
try:
tcp.TCPServer.__init__(self, (host, port))
except socket.error, v:
raise ProxyServerError('Error starting proxy server: ' + v.strerror)
self.channel = None
def start_slave(self, klass, channel):
slave = klass(channel, self)
slave.start()
def set_channel(self, channel):
self.channel = channel
def handle_client_connection(self, conn, client_address):
h = ConnectionHandler(self.config, conn, client_address, self, self.channel, self.server_version)
h.handle()
h.finish()
class ConnectionHandler:
def __init__(self, config, client_connection, client_address, server, channel, server_version):
self.config = config
self.client_conn = ClientConnection(client_connection, client_address, server)
self.server_conn = None
self.channel, self.server_version = channel, server_version
self.close = False
self.conntype = None
self.sni = None
self.mode = "regular"
if self.config.reverse_proxy:
self.mode = "reverse"
if self.config.transparent_proxy:
self.mode = "transparent"
def handle(self):
self.log("clientconnect")
self.channel.ask("clientconnect", self)
self.determine_conntype()
try:
try:
# Can we already identify the target server and connect to it?
server_address = None
address_priority = None
if self.config.forward_proxy:
server_address = self.config.forward_proxy[1:]
address_priority = AddressPriority.FORCE
elif self.config.reverse_proxy:
server_address = self.config.reverse_proxy[1:]
address_priority = AddressPriority.FROM_SETTINGS
elif self.config.transparent_proxy:
server_address = self.config.transparent_proxy["resolver"].original_addr(
self.client_conn.connection)
if not server_address:
raise ProxyError(502, "Transparent mode failure: could not resolve original destination.")
address_priority = AddressPriority.FROM_CONNECTION
self.log("transparent to %s:%s" % server_address)
if server_address:
self.set_server_address(server_address, address_priority)
self._handle_ssl()
while not self.close:
try:
protocol.handle_messages(self.conntype, self)
except ConnectionTypeChange:
self.log("Connection Type Changed: %s" % self.conntype)
continue
# FIXME: Do we want to persist errors?
except (ProxyError, tcp.NetLibError), e:
protocol.handle_error(self.conntype, self, e)
except Exception, e:
self.log(e.__class__)
import traceback
self.log(traceback.format_exc())
self.log(str(e))
self.del_server_connection()
self.log("clientdisconnect")
self.channel.tell("clientdisconnect", self)
def _handle_ssl(self):
"""
Helper function of .handle()
Check if we can already identify SSL connections.
If so, connect to the server and establish an SSL connection
"""
client_ssl = False
server_ssl = False
if self.config.transparent_proxy:
client_ssl = server_ssl = (self.server_conn.address.port in self.config.transparent_proxy["sslports"])
elif self.config.reverse_proxy:
client_ssl = server_ssl = (self.config.reverse_proxy[0] == "https")
# TODO: Make protocol generic (as with transparent proxies)
# TODO: Add SSL-terminating capatbility (SSL -> mitmproxy -> plain and vice versa)
if client_ssl or server_ssl:
self.establish_server_connection()
self.establish_ssl(client=client_ssl, server=server_ssl)
def del_server_connection(self):
"""
Deletes an existing server connection.
"""
if self.server_conn and self.server_conn.connection:
self.server_conn.finish()
self.log("serverdisconnect", ["%s:%s" % (self.server_conn.address.host, self.server_conn.address.port)])
self.channel.tell("serverdisconnect", self)
self.server_conn = None
self.sni = None
def determine_conntype(self):
#TODO: Add ruleset to select correct protocol depending on mode/target port etc.
self.conntype = "http"
def set_server_address(self, address, priority):
"""
Sets a new server address with the given priority.
Does not re-establish either connection or SSL handshake.
@type priority: libmproxy.proxy.primitives.AddressPriority
"""
address = tcp.Address.wrap(address)
if self.server_conn:
if self.server_conn.priority > priority:
self.log("Attempt to change server address, "
"but priority is too low (is: %s, got: %s)" % (self.server_conn.priority, priority))
return
if self.server_conn.address == address:
self.server_conn.priority = priority # Possibly increase priority
return
self.del_server_connection()
self.log("Set new server address: %s:%s" % (address.host, address.port))
self.server_conn = ServerConnection(address, priority)
def establish_server_connection(self):
"""
Establishes a new server connection.
If there is already an existing server connection, the function returns immediately.
"""
if self.server_conn.connection:
return
self.log("serverconnect", ["%s:%s" % self.server_conn.address()[:2]])
self.channel.tell("serverconnect", self)
try:
self.server_conn.connect()
except tcp.NetLibError, v:
raise ProxyError(502, v)
def establish_ssl(self, client=False, server=False):
"""
Establishes SSL on the existing connection(s) to the server or the client,
as specified by the parameters. If the target server is on the pass-through list,
the conntype attribute will be changed and the SSL connection won't be wrapped.
A protocol handler must raise a ConnTypeChanged exception if it detects that this is happening
"""
# TODO: Implement SSL pass-through handling and change conntype
passthrough = [
# "echo.websocket.org",
# "174.129.224.73" # echo.websocket.org, transparent mode
]
if self.server_conn.address.host in passthrough or self.sni in passthrough:
self.conntype = "tcp"
return
# Logging
if client or server:
subs = []
if client:
subs.append("with client")
if server:
subs.append("with server (sni: %s)" % self.sni)
self.log("Establish SSL", subs)
if server:
if self.server_conn.ssl_established:
raise ProxyError(502, "SSL to Server already established.")
self.establish_server_connection() # make sure there is a server connection.
self.server_conn.establish_ssl(self.config.clientcerts, self.sni)
if client:
if self.client_conn.ssl_established:
raise ProxyError(502, "SSL to Client already established.")
cert, key = self.find_cert()
self.client_conn.convert_to_ssl(
cert, key,
handle_sni = self.handle_sni,
cipher_list = self.config.ciphers
)
def server_reconnect(self, no_ssl=False):
address = self.server_conn.address
had_ssl = self.server_conn.ssl_established
priority = self.server_conn.priority
sni = self.sni
self.log("(server reconnect follows)")
self.del_server_connection()
self.set_server_address(address, priority)
self.establish_server_connection()
if had_ssl and not no_ssl:
self.sni = sni
self.establish_ssl(server=True)
def finish(self):
self.client_conn.finish()
def log(self, msg, subs=()):
msg = [
"%s:%s: %s" % (self.client_conn.address.host, self.client_conn.address.port, msg)
]
for i in subs:
msg.append(" -> " + i)
msg = "\n".join(msg)
self.channel.tell("log", Log(msg))
def find_cert(self):
host = self.server_conn.address.host
sans = []
if not self.config.no_upstream_cert or not self.server_conn.ssl_established:
upstream_cert = self.server_conn.cert
if upstream_cert.cn:
host = upstream_cert.cn.decode("utf8").encode("idna")
sans = upstream_cert.altnames
ret = self.config.certstore.get_cert(host, sans)
if not ret:
raise ProxyError(502, "Unable to generate dummy cert.")
return ret
def handle_sni(self, connection):
"""
This callback gets called during the SSL handshake with the client.
The client has just sent the Sever Name Indication (SNI). We now connect upstream to
figure out which certificate needs to be served.
"""
try:
sn = connection.get_servername()
if sn and sn != self.sni:
self.sni = sn.decode("utf8").encode("idna")
self.log("SNI received: %s" % self.sni)
self.server_reconnect() # reconnect to upstream server with SNI
# Now, change client context to reflect changed certificate:
new_context = SSL.Context(SSL.TLSv1_METHOD)
cert, key = self.find_cert()
new_context.use_privatekey_file(key)
new_context.use_certificate(cert.X509)
connection.set_context(new_context)
# An unhandled exception in this method will core dump PyOpenSSL, so
# make dang sure it doesn't happen.
except Exception, e: # pragma: no cover
pass

11
mitmdump
View File

@ -1,6 +1,9 @@
#!/usr/bin/env python
import sys, signal
from libmproxy import proxy, dump, cmdline
from libmproxy.proxy.config import process_proxy_options
from libmproxy.proxy.primitives import ProxyServerError
from libmproxy.proxy.server import DummyServer, ProxyServer
import libmproxy.version, netlib.version
import argparse
@ -25,13 +28,13 @@ if __name__ == '__main__':
if options.quiet:
options.verbose = 0
proxyconfig = proxy.process_proxy_options(parser, options)
proxyconfig = process_proxy_options(parser, options)
if options.no_server:
server = proxy.DummyServer(proxyconfig)
server = DummyServer(proxyconfig)
else:
try:
server = proxy.ProxyServer(proxyconfig, options.port, options.addr)
except proxy.ProxyServerError, v:
server = ProxyServer(proxyconfig, options.port, options.addr)
except ProxyServerError, v:
print >> sys.stderr, "mitmdump:", v.args[0]
sys.exit(1)

11
mitmproxy
View File

@ -1,6 +1,9 @@
#!/usr/bin/env python
import sys, argparse, os
from libmproxy import proxy, console, cmdline
from libmproxy.proxy.config import process_proxy_options
from libmproxy.proxy.primitives import ProxyServerError
from libmproxy.proxy.server import DummyServer, ProxyServer
import libmproxy.version, netlib.version
from libmproxy.console import palettes
@ -33,14 +36,14 @@ if __name__ == '__main__':
)
options = parser.parse_args()
config = proxy.process_proxy_options(parser, options)
config = process_proxy_options(parser, options)
if options.no_server:
server = proxy.DummyServer(config)
server = DummyServer(config)
else:
try:
server = proxy.ProxyServer(config, options.port, options.addr)
except proxy.ProxyServerError, v:
server = ProxyServer(config, options.port, options.addr)
except ProxyServerError, v:
print >> sys.stderr, "mitmproxy:", v.args[0]
sys.exit(1)

5
test/test_dump.py
View File

@ -1,6 +1,7 @@
import os
from cStringIO import StringIO
from libmproxy import dump, flow, proxy
from libmproxy.proxy.primitives import Log
import tutils
import mock
@ -21,13 +22,13 @@ def test_strfuncs():
class TestDumpMaster:
def _cycle(self, m, content):
req = tutils.treq(content=content)
l = proxy.Log("connect")
l = Log("connect")
l.reply = mock.MagicMock()
m.handle_log(l)
cc = req.flow.client_conn
cc.reply = mock.MagicMock()
m.handle_clientconnect(cc)
sc = proxy.ServerConnection((req.get_host(), req.get_port()), None)
sc = proxy.connection.ServerConnection((req.get_host(), req.get_port()), None)
sc.reply = mock.MagicMock()
m.handle_serverconnection(sc)
m.handle_request(req)

15
test/test_flow.py
View File

@ -1,9 +1,10 @@
import Queue, time, os.path
from cStringIO import StringIO
import email.utils
from libmproxy import filt, protocol, controller, utils, tnetstring, proxy, flow
from libmproxy import filt, protocol, controller, utils, tnetstring, flow
from libmproxy.protocol.primitives import Error, Flow
from libmproxy.protocol.http import decoded
from libmproxy.protocol.http import decoded, CONTENT_MISSING
from libmproxy.proxy.connection import ClientConnection, ServerConnection
from netlib import tcp
import tutils
@ -565,7 +566,7 @@ class TestFlowMaster:
s = flow.State()
fm = flow.FlowMaster(None, s)
f = tutils.tflow_full()
f.request.content = flow.CONTENT_MISSING
f.request.content = CONTENT_MISSING
assert "missing" in fm.replay_request(f)
f.intercepting = True
@ -586,7 +587,7 @@ class TestFlowMaster:
req = tutils.treq()
fm.handle_clientconnect(req.flow.client_conn)
assert fm.scripts[0].ns["log"][-1] == "clientconnect"
sc = proxy.ServerConnection((req.get_host(), req.get_port()), None)
sc = ServerConnection((req.get_host(), req.get_port()), None)
sc.reply = controller.DummyReply()
fm.handle_serverconnection(sc)
assert fm.scripts[0].ns["log"][-1] == "serverconnect"
@ -795,7 +796,7 @@ class TestRequest:
assert r._assemble()
assert r.size() == len(r._assemble())
r.content = flow.CONTENT_MISSING
r.content = CONTENT_MISSING
tutils.raises("Cannot assemble flow with CONTENT_MISSING", r._assemble)
def test_get_url(self):
@ -1003,7 +1004,7 @@ class TestResponse:
assert resp._assemble()
assert resp.size() == len(resp._assemble())
resp.content = flow.CONTENT_MISSING
resp.content = CONTENT_MISSING
tutils.raises("Cannot assemble flow with CONTENT_MISSING", resp._assemble)
def test_refresh(self):
@ -1159,7 +1160,7 @@ class TestClientConnection:
def test_state(self):
c = tutils.tclient_conn()
assert proxy.ClientConnection._from_state(c._get_state()) == c
assert ClientConnection._from_state(c._get_state()) == c
c2 = tutils.tclient_conn()
c2.address.address = (c2.address.host, 4242)

1
test/test_protocol_http.py
View File

@ -1,4 +1,3 @@
from libmproxy import proxy # FIXME: Remove
from libmproxy.protocol.http import *
from libmproxy.protocol import KILL
from cStringIO import StringIO

18
test/test_proxy.py
View File

@ -1,5 +1,9 @@
import argparse
from libmproxy import proxy, flow, cmdline
from libmproxy import cmdline
from libmproxy.proxy.config import process_proxy_options
from libmproxy.proxy.connection import ServerConnection
from libmproxy.proxy.primitives import ProxyError
from libmproxy.proxy.server import DummyServer, ProxyServer
import tutils
from libpathod import test
from netlib import http, tcp
@ -7,7 +11,7 @@ import mock
def test_proxy_error():
p = proxy.ProxyError(111, "msg")
p = ProxyError(111, "msg")
assert str(p)
@ -19,7 +23,7 @@ class TestServerConnection:
self.d.shutdown()
def test_simple(self):
sc = proxy.ServerConnection((self.d.IFACE, self.d.port), None)
sc = ServerConnection((self.d.IFACE, self.d.port), None)
sc.connect()
r = tutils.treq()
r.flow.server_conn = sc
@ -31,7 +35,7 @@ class TestServerConnection:
sc.finish()
def test_terminate_error(self):
sc = proxy.ServerConnection((self.d.IFACE, self.d.port), None)
sc = ServerConnection((self.d.IFACE, self.d.port), None)
sc.connect()
sc.connection = mock.Mock()
sc.connection.recv = mock.Mock(return_value=False)
@ -56,7 +60,7 @@ class TestProcessProxyOptions:
cmdline.common_options(parser)
opts = parser.parse_args(args=args)
m = MockParser()
return m, proxy.process_proxy_options(m, opts)
return m, process_proxy_options(m, opts)
def assert_err(self, err, *args):
m, p = self.p(*args)
@ -115,12 +119,12 @@ class TestProxyServer:
parser = argparse.ArgumentParser()
cmdline.common_options(parser)
opts = parser.parse_args(args=[])
tutils.raises("error starting proxy server", proxy.ProxyServer, opts, 1)
tutils.raises("error starting proxy server", ProxyServer, opts, 1)
class TestDummyServer:
def test_simple(self):
d = proxy.DummyServer(None)
d = DummyServer(None)
d.start_slave()
d.shutdown()

4
test/test_server.py
View File

@ -3,8 +3,8 @@ import mock
from netlib import tcp, http_auth, http
from libpathod import pathoc, pathod
import tutils, tservers
from libmproxy import flow, proxy
from libmproxy.protocol import KILL
from libmproxy.protocol.http import CONTENT_MISSING
"""
Note that the choice of response code in these tests matters more than you
@ -381,7 +381,7 @@ class TestTransparentResolveError(tservers.TransparentProxTest):
class MasterIncomplete(tservers.TestMaster):
def handle_request(self, m):
resp = tutils.tresp()
resp.content = flow.CONTENT_MISSING
resp.content = CONTENT_MISSING
m.reply(resp)

10
test/tservers.py
View File

@ -2,8 +2,10 @@ import os.path
import threading, Queue
import shutil, tempfile
import flask
from libmproxy.proxy.config import ProxyConfig
from libmproxy.proxy.server import ProxyServer
import libpathod.test, libpathod.pathoc
from libmproxy import proxy, flow, controller
from libmproxy import flow, controller
from libmproxy.cmdline import APP_HOST, APP_PORT
import tutils
@ -24,7 +26,7 @@ def errapp(environ, start_response):
class TestMaster(flow.FlowMaster):
def __init__(self, config):
s = proxy.ProxyServer(config, 0)
s = ProxyServer(config, 0)
state = flow.State()
flow.FlowMaster.__init__(self, s, state)
self.apps.add(testapp, "testapp", 80)
@ -84,7 +86,7 @@ class ProxTestBase(object):
cls.server2 = libpathod.test.Daemon(ssl=cls.ssl, ssloptions=cls.ssloptions)
pconf = cls.get_proxy_config()
cls.confdir = os.path.join(tempfile.gettempdir(), "mitmproxy")
config = proxy.ProxyConfig(
config = ProxyConfig(
no_upstream_cert = cls.no_upstream_cert,
confdir = cls.confdir,
authenticator = cls.authenticator,
@ -256,7 +258,7 @@ class ChainProxTest(ProxTestBase):
Chain n instances of mitmproxy in a row - because we can.
"""
n = 2
chain_config = [lambda: proxy.ProxyConfig(
chain_config = [lambda: ProxyConfig(
)] * n
@classmethod
def setupAll(cls):

9
test/tutils.py
View File

@ -1,7 +1,8 @@
import os, shutil, tempfile
from contextlib import contextmanager
from libmproxy import flow, utils, controller, proxy
from libmproxy import flow, utils, controller
from libmproxy.protocol import http
from libmproxy.proxy.connection import ClientConnection, ServerConnection
import mock_urwid
from libmproxy.console.flowview import FlowView
from libmproxy.console import ConsoleState
@ -21,7 +22,7 @@ def SkipWindows(fn):
def tclient_conn():
c = proxy.ClientConnection._from_state(dict(
c = ClientConnection._from_state(dict(
address=dict(address=("address", 22), use_ipv6=True),
clientcert=None
))
@ -30,7 +31,7 @@ def tclient_conn():
def tserver_conn():
c = proxy.ServerConnection._from_state(dict(
c = ServerConnection._from_state(dict(
address=dict(address=("address", 22), use_ipv6=True),
source_address=dict(address=("address", 22), use_ipv6=True),
cert=None
@ -69,7 +70,7 @@ def tresp(req=None, content="message"):
headers = flow.ODictCaseless()
headers["header_response"] = ["svalue"]
cert = certutils.SSLCert.from_der(file(test_data.path("data/dercert"), "rb").read())
f.server_conn = proxy.ServerConnection._from_state(dict(
f.server_conn = ServerConnection._from_state(dict(
address=dict(address=("address", 22), use_ipv6=True),
source_address=None,
cert=cert.to_pem()))