Grasscutters/mitmproxy
3
0
Fork 0
mirror of https://github.com/Grasscutters/mitmproxy.git synced 2025-02-02 00:05:27 +00:00
Code Issues Projects Releases Wiki Activity

addon options: shift server_replay options

Browse Source
This commit is contained in:
Aldo Cortesi 2018-02-24 15:00:40 +13:00
parent 74a0230788
commit 52c8d7e0f8
3 changed files with 70 additions and 79 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

56
mitmproxy/addons/serverplayback.py
View File

@ -1,8 +1,6 @@
import hashlib
import urllib
import typing
from typing import Any # noqa
from typing import List # noqa
from mitmproxy import ctx
from mitmproxy import flow
@ -19,6 +17,60 @@ class ServerPlayback:
self.final_flow = None
self.configured = False
def load(self, loader):
loader.add_option(
"server_replay_kill_extra", bool, False,
"Kill extra requests during replay."
)
loader.add_option(
"server_replay_nopop", bool, False,
"""
Don't remove flows from server replay state after use. This makes it
possible to replay same response multiple times.
"""
)
loader.add_option(
"server_replay_refresh", bool, True,
"""
Refresh server replay responses by adjusting date, expires and
last-modified headers, as well as adjusting cookie expiration.
"""
)
loader.add_option(
"server_replay_use_headers", typing.Sequence[str], [],
"Request headers to be considered during replay."
)
loader.add_option(
"server_replay", typing.Sequence[str], [],
"Replay server responses from a saved file."
)
loader.add_option(
"server_replay_ignore_content", bool, False,
"Ignore request's content while searching for a saved flow to replay."
)
loader.add_option(
"server_replay_ignore_params", typing.Sequence[str], [],
"""
Request's parameters to be ignored while searching for a saved flow
to replay.
"""
)
loader.add_option(
"server_replay_ignore_payload_params", typing.Sequence[str], [],
"""
Request's payload parameters (application/x-www-form-urlencoded or
multipart/form-data) to be ignored while searching for a saved flow
to replay.
"""
)
loader.add_option(
"server_replay_ignore_host", bool, False,
"""
Ignore request's destination host while searching for a saved flow
to replay.
"""
)
@command.command("replay.server")
def load_flows(self, flows: typing.Sequence[flow.Flow]) -> None:
"""

61
mitmproxy/options.py
View File

@ -67,20 +67,11 @@ class Options(optmanager.OptManager):
view_filter = None # type: Optional[str]
# FIXME: Options that should be uncomplicated to migrate to addons
server_replay_refresh = None # type: bool
replacements = None # type: Sequence[str]
rfile = None # type: Optional[str]
save_stream_file = None # type: Optional[str]
save_stream_filter = None # type: Optional[str]
scripts = None # type: Sequence[str]
server_replay = None # type: Sequence[str]
server_replay_ignore_content = None # type: bool
server_replay_ignore_host = None # type: bool
server_replay_ignore_params = None # type: Sequence[str]
server_replay_ignore_payload_params = None # type: Sequence[str]
server_replay_kill_extra = None # type: bool
server_replay_nopop = None # type: bool
server_replay_use_headers = None # type: Sequence[str]
setheaders = None # type: Sequence[str]
stickyauth = None # type: Optional[str]
stickycookie = None # type: Optional[str]
@ -96,28 +87,10 @@ class Options(optmanager.OptManager):
def __init__(self, **kwargs) -> None:
super().__init__()
self.add_option(
"server_replay_kill_extra", bool, False,
"Kill extra requests during replay."
)
self.add_option(
"server", bool, True,
"Start a proxy server. Enabled by default."
)
self.add_option(
"server_replay_nopop", bool, False,
"""
Don't remove flows from server replay state after use. This makes it
possible to replay same response multiple times.
"""
)
self.add_option(
"server_replay_refresh", bool, True,
"""
Refresh server replay responses by adjusting date, expires and
last-modified headers, as well as adjusting cookie expiration.
"""
)
self.add_option(
"rfile", Optional[str], None,
"Read flows from file."
@ -139,10 +112,6 @@ class Options(optmanager.OptManager):
the separator can be any character.
"""
)
self.add_option(
"server_replay_use_headers", Sequence[str], [],
"Request headers to be considered during replay."
)
self.add_option(
"setheaders", Sequence[str], [],
"""
@ -150,10 +119,6 @@ class Options(optmanager.OptManager):
separator can be any character.
"""
)
self.add_option(
"server_replay", Sequence[str], [],
"Replay server responses from a saved file."
)
self.add_option(
"stickycookie", Optional[str], None,
"Set sticky cookie filter. Matched against requests."
@ -195,32 +160,6 @@ class Options(optmanager.OptManager):
"save_stream_filter", Optional[str], None,
"Filter which flows are written to file."
)
self.add_option(
"server_replay_ignore_content", bool, False,
"Ignore request's content while searching for a saved flow to replay."
)
self.add_option(
"server_replay_ignore_params", Sequence[str], [],
"""
Request's parameters to be ignored while searching for a saved flow
to replay.
"""
)
self.add_option(
"server_replay_ignore_payload_params", Sequence[str], [],
"""
Request's payload parameters (application/x-www-form-urlencoded or
multipart/form-data) to be ignored while searching for a saved flow
to replay.
"""
)
self.add_option(
"server_replay_ignore_host", bool, False,
"""
Ignore request's destination host while searching for a saved flow
to replay.
"""
)
# Proxy options
self.add_option(

32
test/mitmproxy/addons/test_serverplayback.py
View File

@ -19,7 +19,7 @@ def tdump(path, flows):
def test_load_file(tmpdir):
s = serverplayback.ServerPlayback()
with taddons.context():
with taddons.context(s):
fpath = str(tmpdir.join("flows"))
tdump(fpath, [tflow.tflow(resp=True)])
s.load_file(fpath)
@ -30,7 +30,7 @@ def test_load_file(tmpdir):
def test_config(tmpdir):
s = serverplayback.ServerPlayback()
with taddons.context() as tctx:
with taddons.context(s) as tctx:
fpath = str(tmpdir.join("flows"))
tdump(fpath, [tflow.tflow(resp=True)])
tctx.configure(s, server_replay=[fpath])
@ -41,7 +41,7 @@ def test_config(tmpdir):
def test_tick():
s = serverplayback.ServerPlayback()
with taddons.context() as tctx:
with taddons.context(s) as tctx:
s.stop = True
s.final_flow = tflow.tflow()
s.final_flow.live = False
@ -51,7 +51,7 @@ def test_tick():
def test_server_playback():
sp = serverplayback.ServerPlayback()
with taddons.context() as tctx:
with taddons.context(sp) as tctx:
tctx.configure(sp)
f = tflow.tflow(resp=True)
@ -70,7 +70,7 @@ def test_server_playback():
def test_ignore_host():
sp = serverplayback.ServerPlayback()
with taddons.context() as tctx:
with taddons.context(sp) as tctx:
tctx.configure(sp, server_replay_ignore_host=True)
r = tflow.tflow(resp=True)
@ -85,7 +85,7 @@ def test_ignore_host():
def test_ignore_content():
s = serverplayback.ServerPlayback()
with taddons.context() as tctx:
with taddons.context(s) as tctx:
tctx.configure(s, server_replay_ignore_content=False)
r = tflow.tflow(resp=True)
@ -113,7 +113,7 @@ def test_ignore_content():
def test_ignore_content_wins_over_params():
s = serverplayback.ServerPlayback()
with taddons.context() as tctx:
with taddons.context(s) as tctx:
tctx.configure(
s,
server_replay_ignore_content=True,
@ -137,7 +137,7 @@ def test_ignore_content_wins_over_params():
def test_ignore_payload_params_other_content_type():
s = serverplayback.ServerPlayback()
with taddons.context() as tctx:
with taddons.context(s) as tctx:
tctx.configure(
s,
server_replay_ignore_content=False,
@ -161,7 +161,7 @@ def test_ignore_payload_params_other_content_type():
def test_hash():
s = serverplayback.ServerPlayback()
with taddons.context() as tctx:
with taddons.context(s) as tctx:
tctx.configure(s)
r = tflow.tflow()
@ -181,7 +181,7 @@ def test_hash():
def test_headers():
s = serverplayback.ServerPlayback()
with taddons.context() as tctx:
with taddons.context(s) as tctx:
tctx.configure(s, server_replay_use_headers=["foo"])
r = tflow.tflow(resp=True)
@ -200,7 +200,7 @@ def test_headers():
def test_load():
s = serverplayback.ServerPlayback()
with taddons.context() as tctx:
with taddons.context(s) as tctx:
tctx.configure(s)
r = tflow.tflow(resp=True)
@ -227,7 +227,7 @@ def test_load():
def test_load_with_server_replay_nopop():
s = serverplayback.ServerPlayback()
with taddons.context() as tctx:
with taddons.context(s) as tctx:
tctx.configure(s, server_replay_nopop=True)
r = tflow.tflow(resp=True)
@ -245,7 +245,7 @@ def test_load_with_server_replay_nopop():
def test_ignore_params():
s = serverplayback.ServerPlayback()
with taddons.context() as tctx:
with taddons.context(s) as tctx:
tctx.configure(
s,
server_replay_ignore_params=["param1", "param2"]
@ -266,7 +266,7 @@ def test_ignore_params():
def thash(r, r2, setter):
s = serverplayback.ServerPlayback()
with taddons.context() as tctx:
with taddons.context(s) as tctx:
s = serverplayback.ServerPlayback()
tctx.configure(
s,
@ -328,7 +328,7 @@ def test_ignore_payload_params():
def test_server_playback_full():
s = serverplayback.ServerPlayback()
with taddons.context() as tctx:
with taddons.context(s) as tctx:
tctx.configure(
s,
server_replay_refresh = True,
@ -360,7 +360,7 @@ def test_server_playback_full():
def test_server_playback_kill():
s = serverplayback.ServerPlayback()
with taddons.context() as tctx:
with taddons.context(s) as tctx:
tctx.configure(
s,
server_replay_refresh = True,