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

make options explicit

Browse Source
This commit is contained in:
Maximilian Hils 2016-07-13 21:40:13 -07:00
parent b8a23eeaa3
commit ec6fbe9eb6
13 changed files with 190 additions and 133 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

58
mitmproxy/console/master.py
View File

@ -14,6 +14,7 @@ import traceback
import weakref
import urwid
from typing import Optional # noqa
from mitmproxy import builtins
from mitmproxy import contentviews
@ -21,7 +22,6 @@ from mitmproxy import controller
from mitmproxy import exceptions
from mitmproxy import flow
from mitmproxy import script
import mitmproxy.options
from mitmproxy.console import flowlist
from mitmproxy.console import flowview
from mitmproxy.console import grideditor
@ -177,40 +177,26 @@ class ConsoleState(flow.State):
self.add_flow_setting(flow, "marked", marked)
class Options(mitmproxy.options.Options):
attributes = [
"app",
"app_domain",
"app_ip",
"anticache",
"anticomp",
"client_replay",
"eventlog",
"follow",
"keepserving",
"kill",
"intercept",
"limit",
"no_server",
"refresh_server_playback",
"rfile",
"scripts",
"showhost",
"replacements",
"rheaders",
"setheaders",
"server_replay",
"stickycookie",
"stickyauth",
"stream_large_bodies",
"verbosity",
"wfile",
"nopop",
"palette",
"palette_transparent",
"no_mouse",
"outfile",
]
class Options(flow.options.Options):
def __init__(
self,
eventlog=False, # type: bool
follow=False, # type: bool
intercept=False, # type: bool
limit=None, # type: Optional[str]
palette=None, # type: Optional[str]
palette_transparent=False, # type: bool
no_mouse=False, # type: bool
**kwargs
):
self.eventlog = eventlog
self.follow = follow
self.intercept = intercept
self.limit = limit
self.palette = palette
self.palette_transparent = palette_transparent
self.no_mouse = no_mouse
super(Options, self).__init__(**kwargs)
class ConsoleMaster(flow.FlowMaster):
@ -221,6 +207,8 @@ class ConsoleMaster(flow.FlowMaster):
self.addons.add(*builtins.default_addons())
self.stream_path = None
# This line is just for type hinting
self.options = self.options # type: Options
self.options.errored.connect(self.options_error)
if options.replacements:

54
mitmproxy/dump.py
View File

@ -6,12 +6,14 @@ import traceback
import click
from typing import Optional # noqa
import typing
from mitmproxy import contentviews
from mitmproxy import controller
from mitmproxy import exceptions
from mitmproxy import filt
from mitmproxy import flow
from mitmproxy import options
from mitmproxy import builtins
from netlib import human
from netlib import tcp
@ -22,40 +24,20 @@ class DumpError(Exception):
pass
class Options(options.Options):
attributes = [
"app",
"app_host",
"app_port",
"anticache",
"anticomp",
"client_replay",
"filtstr",
"flow_detail",
"keepserving",
"kill",
"no_server",
"nopop",
"refresh_server_playback",
"replacements",
"rfile",
"rheaders",
"setheaders",
"server_replay",
"scripts",
"showhost",
"stickycookie",
"stickyauth",
"stream_large_bodies",
"verbosity",
"outfile",
"replay_ignore_content",
"replay_ignore_params",
"replay_ignore_payload_params",
"replay_ignore_host",
"tfile"
]
class Options(flow.options.Options):
def __init__(
self,
filtstr=None, # type: Optional[str]
flow_detail=1, # type: int
keepserving=False, # type: bool
tfile=None, # type: Optional[typing.io.TextIO]
**kwargs
):
self.filtstr = filtstr
self.flow_detail = flow_detail
self.keepserving = keepserving
self.tfile = tfile
super(Options, self).__init__(**kwargs)
class DumpMaster(flow.FlowMaster):
@ -63,6 +45,8 @@ class DumpMaster(flow.FlowMaster):
def __init__(self, server, options):
flow.FlowMaster.__init__(self, options, server, flow.State())
self.addons.add(*builtins.default_addons())
# This line is just for type hinting
self.options = self.options # type: Options
self.o = options
self.showhost = options.showhost
self.replay_ignore_params = options.replay_ignore_params

2
mitmproxy/flow/__init__.py
View File

@ -8,6 +8,7 @@ from mitmproxy.flow.modules import (
ServerPlaybackState
)
from mitmproxy.flow.state import State, FlowView
from mitmproxy.flow import options
# TODO: We may want to remove the imports from .modules and just expose "modules"
@ -17,4 +18,5 @@ __all__ = [
"FlowMaster",
"AppRegistry", "ReplaceHooks", "SetHeaders", "StreamLargeBodies", "ClientPlaybackState",
"ServerPlaybackState", "State", "FlowView",
"options",
]

69
mitmproxy/flow/options.py Normal file
View File

@ -0,0 +1,69 @@
from __future__ import absolute_import, print_function, division
from mitmproxy import options
from typing import Tuple, Optional, Sequence # noqa
APP_HOST = "mitm.it"
APP_PORT = 80
class Options(options.Options):
def __init__(
self,
# TODO: rename to onboarding_app_*
app=True, # type: bool
app_host=APP_HOST, # type: str
app_port=APP_PORT, # type: int
anticache=False, # type: bool
anticomp=False, # type: bool
client_replay=None, # type: Optional[str]
kill=False, # type: bool
no_server=False, # type: bool
nopop=False, # type: bool
refresh_server_playback=False, # type: bool
rfile=None, # type: Optional[str]
scripts=(), # type: Sequence[str]
showhost=False, # type: bool
replacements=(), # type: Sequence[Tuple[str, str, str]]
rheaders=(), # type: Sequence[str]
setheaders=(), # type: Sequence[Tuple[str, str, str]]
server_replay=None, # type: Optional[str]
stickycookie=None, # type: Optional[str]
stickyauth=None, # type: Optional[str]
stream_large_bodies=None, # type: Optional[str]
verbosity=1, # type: int
outfile=None, # type: Optional[str]
replay_ignore_content=False, # type: bool
replay_ignore_params=(), # type: Sequence[str]
replay_ignore_payload_params=(), # type: Sequence[str]
replay_ignore_host=False, # type: bool
):
# We could replace all assignments with clever metaprogramming,
# but type hints are a much more valueable asset.
self.app = app
self.app_host = app_host
self.app_port = app_port
self.anticache = anticache
self.anticomp = anticomp
self.client_replay = client_replay
self.kill = kill
self.no_server = no_server
self.nopop = nopop
self.refresh_server_playback = refresh_server_playback
self.rfile = rfile
self.scripts = scripts
self.showhost = showhost
self.replacements = replacements
self.rheaders = rheaders
self.setheaders = setheaders
self.server_replay = server_replay
self.stickycookie = stickycookie
self.stickyauth = stickyauth
self.stream_large_bodies = stream_large_bodies
self.verbosity = verbosity
self.outfile = outfile
self.replay_ignore_content = replay_ignore_content
self.replay_ignore_params = replay_ignore_params
self.replay_ignore_payload_params = replay_ignore_payload_params
self.replay_ignore_host = replay_ignore_host
super(Options, self).__init__()

27
mitmproxy/options.py
View File

@ -14,14 +14,21 @@ class Options(object):
exception, all changes are rolled back, the exception is suppressed,
and the .errored signal is notified.
"""
_initialized = False
attributes = []
def __init__(self, **kwargs):
def __new__(cls, *args, **kwargs):
# Initialize instance._opts before __init__ is called.
# This allows us to call super().__init__() last, which then sets
# ._initialized = True as the final operation.
instance = super(Options, cls).__new__(cls)
instance.__dict__["_opts"] = {}
return instance
def __init__(self):
self.__dict__["changed"] = blinker.Signal()
self.__dict__["errored"] = blinker.Signal()
self.__dict__["_opts"] = dict([(i, None) for i in self.attributes])
for k, v in kwargs.items():
self._opts[k] = v
self.__dict__["_initialized"] = True
@contextlib.contextmanager
def rollback(self):
@ -48,6 +55,9 @@ class Options(object):
raise AttributeError()
def __setattr__(self, attr, value):
if not self._initialized:
self._opts[attr] = value
return
if attr not in self._opts:
raise KeyError("No such option: %s" % attr)
with self.rollback():
@ -71,4 +81,11 @@ class Options(object):
return lambda x: self.__setattr__(attr, x)
def __repr__(self):
return pprint.pformat(self._opts)
options = pprint.pformat(self._opts, indent=4).strip(" {}")
if "\n" in options:
options = "\n " + options + "\n"
return "{mod}.{cls}({{{options}}})".format(
mod=type(self).__module__,
cls=type(self).__name__,
options=options
)

60
mitmproxy/web/master.py
View File

@ -6,11 +6,12 @@ import collections
import tornado.httpserver
import tornado.ioloop
from typing import Optional # noqa
from mitmproxy import builtins
from mitmproxy import controller
from mitmproxy import exceptions
from mitmproxy import flow
from mitmproxy import options
from mitmproxy.web import app
from netlib.http import authentication
@ -90,43 +91,26 @@ class WebState(flow.State):
)
class Options(options.Options):
attributes = [
"app",
"app_domain",
"app_ip",
"anticache",
"anticomp",
"client_replay",
"eventlog",
"keepserving",
"kill",
"intercept",
"no_server",
"outfile",
"refresh_server_playback",
"rfile",
"scripts",
"showhost",
"replacements",
"rheaders",
"setheaders",
"server_replay",
"stickycookie",
"stickyauth",
"stream_large_bodies",
"verbosity",
"wfile",
"nopop",
"wdebug",
"wport",
"wiface",
"wauthenticator",
"wsingleuser",
"whtpasswd",
]
class Options(flow.options.Options):
def __init__(
self,
wdebug=bool, # type: bool
wport=8081, # type: int
wiface="127.0.0.1", # type: str
wauthenticator=None, # type: Optional[authentication.PassMan]
wsingleuser=None, # type: Optional[str]
whtpasswd=None, # type: Optional[str]
**kwargs
):
self.wdebug = wdebug
self.wport = wport
self.wiface = wiface
self.wauthenticator = wauthenticator
self.wsingleuser = wsingleuser
self.whtpasswd = whtpasswd
super(Options, self).__init__(**kwargs)
# TODO: This doesn't belong here.
def process_web_options(self, parser):
if self.wsingleuser or self.whtpasswd:
if self.wsingleuser:
@ -153,6 +137,8 @@ class WebMaster(flow.FlowMaster):
self.app = app.Application(
self, self.options.wdebug, self.options.wauthenticator
)
# This line is just for type hinting
self.options = self.options # type: Options
if options.rfile:
try:
self.load_flows_file(options.rfile)

2
test/mitmproxy/builtins/test_anticache.py
View File

@ -2,7 +2,7 @@ from .. import tutils, mastertest
from mitmproxy.builtins import anticache
from mitmproxy.flow import master
from mitmproxy.flow import state
from mitmproxy import options
from mitmproxy.flow import options
class TestAntiCache(mastertest.MasterTest):

2
test/mitmproxy/builtins/test_anticomp.py
View File

@ -2,7 +2,7 @@ from .. import tutils, mastertest
from mitmproxy.builtins import anticomp
from mitmproxy.flow import master
from mitmproxy.flow import state
from mitmproxy import options
from mitmproxy.flow import options
class TestAntiComp(mastertest.MasterTest):

2
test/mitmproxy/builtins/test_stickyauth.py
View File

@ -2,7 +2,7 @@ from .. import tutils, mastertest
from mitmproxy.builtins import stickyauth
from mitmproxy.flow import master
from mitmproxy.flow import state
from mitmproxy import options
from mitmproxy.flow import options
class TestStickyAuth(mastertest.MasterTest):

2
test/mitmproxy/builtins/test_stickycookie.py
View File

@ -2,7 +2,7 @@ from .. import tutils, mastertest
from mitmproxy.builtins import stickycookie
from mitmproxy.flow import master
from mitmproxy.flow import state
from mitmproxy import options
from mitmproxy.flow import options
from netlib import tutils as ntutils

6
test/mitmproxy/console/test_master.py
View File

@ -111,14 +111,14 @@ def test_options():
class TestMaster(mastertest.MasterTest):
def mkmaster(self, filt, **options):
def mkmaster(self, **options):
if "verbosity" not in options:
options["verbosity"] = 0
o = console.master.Options(filtstr=filt, **options)
o = console.master.Options(**options)
return console.master.ConsoleMaster(None, o)
def test_basic(self):
m = self.mkmaster(None)
m = self.mkmaster()
for i in (1, 2, 3):
self.dummy_cycle(m, 1, b"")
assert len(m.state.flows) == i

30
test/mitmproxy/test_options.py
View File

@ -7,10 +7,10 @@ from netlib import tutils
class TO(options.Options):
attributes = [
"one",
"two"
]
def __init__(self, one=None, two=None):
self.one = one
self.two = two
super(TO, self).__init__()
def test_options():
@ -19,8 +19,13 @@ def test_options():
assert o.two == "three"
o.one = "one"
assert o.one == "one"
tutils.raises("no such option", setattr, o, "nonexistent", "value")
tutils.raises("no such option", o.update, nonexistent = "value")
with tutils.raises(TypeError):
TO(nonexistent = "value")
with tutils.raises("no such option"):
o.nonexistent = "value"
with tutils.raises("no such option"):
o.update(nonexistent = "value")
rec = []
@ -43,7 +48,8 @@ def test_setter():
f = o.setter("two")
f("xxx")
assert o.two == "xxx"
tutils.raises("no such option", o.setter, "nonexistent")
with tutils.raises("no such option"):
o.setter("nonexistent")
def test_rollback():
@ -61,7 +67,7 @@ def test_rollback():
def err(opts):
if opts.one == "ten":
raise exceptions.OptionsError
raise exceptions.OptionsError()
o.changed.connect(sub)
o.changed.connect(err)
@ -73,3 +79,11 @@ def test_rollback():
assert len(rec) == 2
assert rec[0].one == "ten"
assert rec[1].one == "two"
def test_repr():
assert repr(TO()) == "test.mitmproxy.test_options.TO({'one': None, 'two': None})"
assert repr(TO(one='x' * 60)) == """test.mitmproxy.test_options.TO({
'one': 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx',
'two': None
})"""

9
test/mitmproxy/test_web_master.py
View File

@ -3,15 +3,12 @@ from . import mastertest
class TestWebMaster(mastertest.MasterTest):
def mkmaster(self, filt, **options):
o = master.Options(
filtstr=filt,
**options
)
def mkmaster(self, **options):
o = master.Options(**options)
return master.WebMaster(None, o)
def test_basic(self):
m = self.mkmaster(None)
m = self.mkmaster()
for i in (1, 2, 3):
self.dummy_cycle(m, 1, b"")
assert len(m.state.flows) == i