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 #1532 from cortesi/playback

Browse Source 
Playback and fix construct breakage
This commit is contained in:
Aldo Cortesi 2016-09-07 12:59:11 +12:00 committed by GitHub
commit ea49b8a2e2
11 changed files with 433 additions and 491 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
mitmproxy/builtins/__init__.py
View File

@ -8,6 +8,7 @@ from mitmproxy.builtins import stickycookie
from mitmproxy.builtins import script
from mitmproxy.builtins import replace
from mitmproxy.builtins import setheaders
from mitmproxy.builtins import serverplayback
def default_addons():
@ -20,4 +21,5 @@ def default_addons():
filestreamer.FileStreamer(),
replace.Replace(),
setheaders.SetHeaders(),
serverplayback.ServerPlayback()
]

133
mitmproxy/builtins/serverplayback.py Normal file
View File

@ -0,0 +1,133 @@
from __future__ import absolute_import, print_function, division
from six.moves import urllib
import hashlib
from netlib import strutils
from mitmproxy import exceptions, flow, ctx
class ServerPlayback(object):
def __init__(self):
self.options = None
self.flowmap = {}
self.stop = False
self.final_flow = None
def load(self, flows):
for i in flows:
if i.response:
l = self.flowmap.setdefault(self._hash(i), [])
l.append(i)
def clear(self):
self.flowmap = {}
def count(self):
return sum([len(i) for i in self.flowmap.values()])
def _hash(self, flow):
"""
Calculates a loose hash of the flow request.
"""
r = flow.request
_, _, path, _, query, _ = urllib.parse.urlparse(r.url)
queriesArray = urllib.parse.parse_qsl(query, keep_blank_values=True)
key = [str(r.port), str(r.scheme), str(r.method), str(path)]
if not self.options.replay_ignore_content:
form_contents = r.urlencoded_form or r.multipart_form
if self.options.replay_ignore_payload_params and form_contents:
params = [
strutils.always_bytes(i)
for i in self.options.replay_ignore_payload_params
]
for p in form_contents.items(multi=True):
if p[0] not in params:
key.append(p)
else:
key.append(str(r.raw_content))
if not self.options.replay_ignore_host:
key.append(r.host)
filtered = []
ignore_params = self.options.replay_ignore_params or []
for p in queriesArray:
if p[0] not in ignore_params:
filtered.append(p)
for p in filtered:
key.append(p[0])
key.append(p[1])
if self.options.rheaders:
headers = []
for i in self.options.rheaders:
v = r.headers.get(i)
headers.append((i, v))
key.append(headers)
return hashlib.sha256(
repr(key).encode("utf8", "surrogateescape")
).digest()
def next_flow(self, request):
"""
Returns the next flow object, or None if no matching flow was
found.
"""
hsh = self._hash(request)
if hsh in self.flowmap:
if self.options.nopop:
return self.flowmap[hsh][0]
else:
ret = self.flowmap[hsh].pop(0)
if not self.flowmap[hsh]:
del self.flowmap[hsh]
return ret
def configure(self, options, updated):
self.options = options
if options.server_replay and "server_replay" in updated:
try:
flows = flow.read_flows_from_paths(options.server_replay)
except exceptions.FlowReadException as e:
raise exceptions.OptionsError(str(e))
self.clear()
self.load(flows)
# FIXME: These options have to be renamed to something more sensible -
# prefixed with serverplayback_ where appropriate, and playback_ where
# they're shared with client playback.
#
# options.kill
# options.rheaders,
# options.nopop,
# options.replay_ignore_params,
# options.replay_ignore_content,
# options.replay_ignore_payload_params,
# options.replay_ignore_host
def tick(self):
if self.stop and not self.final_flow.live:
ctx.master.shutdown()
def request(self, f):
if self.flowmap:
rflow = self.next_flow(f)
if rflow:
response = rflow.response.copy()
response.is_replay = True
if self.options.refresh_server_playback:
response.refresh()
f.response = response
if not self.flowmap and not self.options.keepserving:
self.final_flow = f
self.stop = True
elif self.options.kill:
ctx.log.warn(
"server_playback: killed non-replay request {}".format(
f.request.url
)
)
f.reply.kill()

12
mitmproxy/dump.py
View File

@ -59,18 +59,6 @@ class DumpMaster(flow.FlowMaster):
"HTTP/2 is disabled. Use --no-http2 to silence this warning.",
file=sys.stderr)
if options.server_replay:
self.start_server_playback(
self._readflow(options.server_replay),
options.kill, options.rheaders,
not options.keepserving,
options.nopop,
options.replay_ignore_params,
options.replay_ignore_content,
options.replay_ignore_payload_params,
options.replay_ignore_host
)
if options.client_replay:
self.start_client_playback(
self._readflow(options.client_replay),

6
mitmproxy/flow/__init__.py
View File

@ -4,16 +4,14 @@ from mitmproxy.flow import export, modules
from mitmproxy.flow.io import FlowWriter, FilteredFlowWriter, FlowReader, read_flows_from_paths
from mitmproxy.flow.master import FlowMaster
from mitmproxy.flow.modules import (
AppRegistry, StreamLargeBodies, ClientPlaybackState, ServerPlaybackState
AppRegistry, StreamLargeBodies, ClientPlaybackState
)
from mitmproxy.flow.state import State, FlowView
# TODO: We may want to remove the imports from .modules and just expose "modules"
__all__ = [
"export", "modules",
"FlowWriter", "FilteredFlowWriter", "FlowReader", "read_flows_from_paths",
"FlowMaster",
"AppRegistry", "StreamLargeBodies", "ClientPlaybackState",
"ServerPlaybackState", "State", "FlowView",
"State", "FlowView",
]

79
mitmproxy/flow/master.py
View File

@ -29,15 +29,8 @@ class FlowMaster(controller.Master):
if server:
self.add_server(server)
self.state = state
self.server_playback = None # type: Optional[modules.ServerPlaybackState]
self.client_playback = None # type: Optional[modules.ClientPlaybackState]
self.kill_nonreplay = False
self.stream_large_bodies = None # type: Optional[modules.StreamLargeBodies]
self.replay_ignore_params = False
self.replay_ignore_content = None
self.replay_ignore_host = False
self.apps = modules.AppRegistry()
def start_app(self, host, port):
@ -62,56 +55,6 @@ class FlowMaster(controller.Master):
def stop_client_playback(self):
self.client_playback = None
def start_server_playback(
self,
flows,
kill,
headers,
exit,
nopop,
ignore_params,
ignore_content,
ignore_payload_params,
ignore_host):
"""
flows: List of flows.
kill: Boolean, should we kill requests not part of the replay?
ignore_params: list of parameters to ignore in server replay
ignore_content: true if request content should be ignored in server replay
ignore_payload_params: list of content params to ignore in server replay
ignore_host: true if request host should be ignored in server replay
"""
self.server_playback = modules.ServerPlaybackState(
headers,
flows,
exit,
nopop,
ignore_params,
ignore_content,
ignore_payload_params,
ignore_host)
self.kill_nonreplay = kill
def stop_server_playback(self):
self.server_playback = None
def do_server_playback(self, flow):
"""
This method should be called by child classes in the request
handler. Returns True if playback has taken place, None if not.
"""
if self.server_playback:
rflow = self.server_playback.next_flow(flow)
if not rflow:
return None
response = rflow.response.copy()
response.is_replay = True
if self.options.refresh_server_playback:
response.refresh()
flow.response = response
return True
return None
def tick(self, timeout):
if self.client_playback:
stop = (
@ -126,17 +69,6 @@ class FlowMaster(controller.Master):
else:
self.client_playback.tick(self)
if self.server_playback:
stop = (
self.server_playback.count() == 0 and
self.state.active_flow_count() == 0 and
not self.kill_nonreplay
)
exit = self.server_playback.exit
if stop:
self.stop_server_playback()
if exit:
self.shutdown()
return super(FlowMaster, self).tick(timeout)
def duplicate_flow(self, f):
@ -229,13 +161,6 @@ class FlowMaster(controller.Master):
except IOError as v:
raise exceptions.FlowReadException(v.strerror)
def process_new_request(self, f):
if self.server_playback:
pb = self.do_server_playback(f)
if not pb and self.kill_nonreplay:
self.add_log("Killed {}".format(f.request.url), "info")
f.reply.kill()
def replay_request(self, f, block=False):
"""
Returns None if successful, or error message if not.
@ -256,7 +181,8 @@ class FlowMaster(controller.Master):
f.response = None
f.error = None
self.process_new_request(f)
# FIXME: process through all addons?
# self.process_new_request(f)
rt = http_replay.RequestReplayThread(
self.server.config,
f,
@ -314,7 +240,6 @@ class FlowMaster(controller.Master):
return
if f not in self.state.flows: # don't add again on replay
self.state.add_flow(f)
self.process_new_request(f)
return f
@controller.handler

97
mitmproxy/flow/modules.py
View File

@ -1,13 +1,8 @@
from __future__ import absolute_import, print_function, division
import hashlib
from six.moves import urllib
from mitmproxy import controller
from netlib import wsgi
from netlib import version
from netlib import strutils
from netlib.http import http1
@ -84,95 +79,3 @@ class ClientPlaybackState:
master.request(self.current)
if self.current.response:
master.response(self.current)
class ServerPlaybackState:
def __init__(
self,
headers,
flows,
exit,
nopop,
ignore_params,
ignore_content,
ignore_payload_params,
ignore_host):
"""
headers: Case-insensitive list of request headers that should be
included in request-response matching.
"""
self.headers = headers
self.exit = exit
self.nopop = nopop
self.ignore_params = ignore_params
self.ignore_content = ignore_content
self.ignore_payload_params = [strutils.always_bytes(x) for x in (ignore_payload_params or ())]
self.ignore_host = ignore_host
self.fmap = {}
for i in flows:
if i.response:
l = self.fmap.setdefault(self._hash(i), [])
l.append(i)
def count(self):
return sum(len(i) for i in self.fmap.values())
def _hash(self, flow):
"""
Calculates a loose hash of the flow request.
"""
r = flow.request
_, _, path, _, query, _ = urllib.parse.urlparse(r.url)
queriesArray = urllib.parse.parse_qsl(query, keep_blank_values=True)
key = [
str(r.port),
str(r.scheme),
str(r.method),
str(path),
]
if not self.ignore_content:
form_contents = r.urlencoded_form or r.multipart_form
if self.ignore_payload_params and form_contents:
key.extend(
p for p in form_contents.items(multi=True)
if p[0] not in self.ignore_payload_params
)
else:
key.append(str(r.raw_content))
if not self.ignore_host:
key.append(r.host)
filtered = []
ignore_params = self.ignore_params or []
for p in queriesArray:
if p[0] not in ignore_params:
filtered.append(p)
for p in filtered:
key.append(p[0])
key.append(p[1])
if self.headers:
headers = []
for i in self.headers:
v = r.headers.get(i)
headers.append((i, v))
key.append(headers)
return hashlib.sha256(repr(key).encode("utf8", "surrogateescape")).digest()
def next_flow(self, request):
"""
Returns the next flow object, or None if no matching flow was
found.
"""
l = self.fmap.get(self._hash(request))
if not l:
return None
if self.nopop:
return l[0]
else:
return l.pop(0)

2
mitmproxy/options.py
View File

@ -31,6 +31,7 @@ class Options(optmanager.OptManager):
anticomp=False, # type: bool
client_replay=None, # type: Optional[str]
kill=False, # type: bool
keepserving=True, # type: bool
no_server=False, # type: bool
nopop=False, # type: bool
refresh_server_playback=False, # type: bool
@ -87,6 +88,7 @@ class Options(optmanager.OptManager):
self.anticache = anticache
self.anticomp = anticomp
self.client_replay = client_replay
self.keepserving = keepserving
self.kill = kill
self.no_server = no_server
self.nopop = nopop

284
test/mitmproxy/builtins/test_serverplayback.py Normal file
View File

@ -0,0 +1,284 @@
from .. import tutils, mastertest
import netlib.tutils
from mitmproxy.builtins import serverplayback
from mitmproxy import options
from mitmproxy import exceptions
from mitmproxy import flow
class TestServerPlayback:
def test_server_playback(self):
sp = serverplayback.ServerPlayback()
sp.configure(options.Options(), [])
f = tutils.tflow(resp=True)
assert not sp.flowmap
sp.load([f])
assert sp.flowmap
assert sp.next_flow(f)
assert not sp.flowmap
def test_ignore_host(self):
sp = serverplayback.ServerPlayback()
sp.configure(options.Options(replay_ignore_host=True), [])
r = tutils.tflow(resp=True)
r2 = tutils.tflow(resp=True)
r.request.host = "address"
r2.request.host = "address"
assert sp._hash(r) == sp._hash(r2)
r2.request.host = "wrong_address"
assert sp._hash(r) == sp._hash(r2)
def test_ignore_content(self):
s = serverplayback.ServerPlayback()
s.configure(options.Options(replay_ignore_content=False), [])
r = tutils.tflow(resp=True)
r2 = tutils.tflow(resp=True)
r.request.content = b"foo"
r2.request.content = b"foo"
assert s._hash(r) == s._hash(r2)
r2.request.content = b"bar"
assert not s._hash(r) == s._hash(r2)
s.configure(options.Options(replay_ignore_content=True), [])
r = tutils.tflow(resp=True)
r2 = tutils.tflow(resp=True)
r.request.content = b"foo"
r2.request.content = b"foo"
assert s._hash(r) == s._hash(r2)
r2.request.content = b"bar"
assert s._hash(r) == s._hash(r2)
r2.request.content = b""
assert s._hash(r) == s._hash(r2)
r2.request.content = None
assert s._hash(r) == s._hash(r2)
def test_ignore_content_wins_over_params(self):
s = serverplayback.ServerPlayback()
s.configure(
options.Options(
replay_ignore_content=True,
replay_ignore_payload_params=[
"param1", "param2"
]
),
[]
)
# NOTE: parameters are mutually exclusive in options
r = tutils.tflow(resp=True)
r.request.headers["Content-Type"] = "application/x-www-form-urlencoded"
r.request.content = b"paramx=y"
r2 = tutils.tflow(resp=True)
r2.request.headers["Content-Type"] = "application/x-www-form-urlencoded"
r2.request.content = b"paramx=x"
# same parameters
assert s._hash(r) == s._hash(r2)
def test_ignore_payload_params_other_content_type(self):
s = serverplayback.ServerPlayback()
s.configure(
options.Options(
replay_ignore_content=False,
replay_ignore_payload_params=[
"param1", "param2"
]
),
[]
)
r = tutils.tflow(resp=True)
r.request.headers["Content-Type"] = "application/json"
r.request.content = b'{"param1":"1"}'
r2 = tutils.tflow(resp=True)
r2.request.headers["Content-Type"] = "application/json"
r2.request.content = b'{"param1":"1"}'
# same content
assert s._hash(r) == s._hash(r2)
# distint content (note only x-www-form-urlencoded payload is analysed)
r2.request.content = b'{"param1":"2"}'
assert not s._hash(r) == s._hash(r2)
def test_hash(self):
s = serverplayback.ServerPlayback()
s.configure(options.Options(), [])
r = tutils.tflow()
r2 = tutils.tflow()
assert s._hash(r)
assert s._hash(r) == s._hash(r2)
r.request.headers["foo"] = "bar"
assert s._hash(r) == s._hash(r2)
r.request.path = "voing"
assert s._hash(r) != s._hash(r2)
r.request.path = "path?blank_value"
r2.request.path = "path?"
assert s._hash(r) != s._hash(r2)
def test_headers(self):
s = serverplayback.ServerPlayback()
s.configure(options.Options(rheaders=["foo"]), [])
r = tutils.tflow(resp=True)
r.request.headers["foo"] = "bar"
r2 = tutils.tflow(resp=True)
assert not s._hash(r) == s._hash(r2)
r2.request.headers["foo"] = "bar"
assert s._hash(r) == s._hash(r2)
r2.request.headers["oink"] = "bar"
assert s._hash(r) == s._hash(r2)
r = tutils.tflow(resp=True)
r2 = tutils.tflow(resp=True)
assert s._hash(r) == s._hash(r2)
def test_load(self):
s = serverplayback.ServerPlayback()
s.configure(options.Options(), [])
r = tutils.tflow(resp=True)
r.request.headers["key"] = "one"
r2 = tutils.tflow(resp=True)
r2.request.headers["key"] = "two"
s.load([r, r2])
assert s.count() == 2
n = s.next_flow(r)
assert n.request.headers["key"] == "one"
assert s.count() == 1
n = s.next_flow(r)
assert n.request.headers["key"] == "two"
assert not s.flowmap
assert s.count() == 0
assert not s.next_flow(r)
def test_load_with_nopop(self):
s = serverplayback.ServerPlayback()
s.configure(options.Options(nopop=True), [])
r = tutils.tflow(resp=True)
r.request.headers["key"] = "one"
r2 = tutils.tflow(resp=True)
r2.request.headers["key"] = "two"
s.load([r, r2])
assert s.count() == 2
s.next_flow(r)
assert s.count() == 2
def test_ignore_params(self):
s = serverplayback.ServerPlayback()
s.configure(
options.Options(
replay_ignore_params=["param1", "param2"]
),
[]
)
r = tutils.tflow(resp=True)
r.request.path = "/test?param1=1"
r2 = tutils.tflow(resp=True)
r2.request.path = "/test"
assert s._hash(r) == s._hash(r2)
r2.request.path = "/test?param1=2"
assert s._hash(r) == s._hash(r2)
r2.request.path = "/test?param2=1"
assert s._hash(r) == s._hash(r2)
r2.request.path = "/test?param3=2"
assert not s._hash(r) == s._hash(r2)
def test_ignore_payload_params(self):
s = serverplayback.ServerPlayback()
s.configure(
options.Options(
replay_ignore_payload_params=["param1", "param2"]
),
[]
)
r = tutils.tflow(resp=True)
r.request.headers["Content-Type"] = "application/x-www-form-urlencoded"
r.request.content = b"paramx=x&param1=1"
r2 = tutils.tflow(resp=True)
r2.request.headers["Content-Type"] = "application/x-www-form-urlencoded"
r2.request.content = b"paramx=x&param1=1"
# same parameters
assert s._hash(r) == s._hash(r2)
# ignored parameters !=
r2.request.content = b"paramx=x&param1=2"
assert s._hash(r) == s._hash(r2)
# missing parameter
r2.request.content = b"paramx=x"
assert s._hash(r) == s._hash(r2)
# ignorable parameter added
r2.request.content = b"paramx=x&param1=2"
assert s._hash(r) == s._hash(r2)
# not ignorable parameter changed
r2.request.content = b"paramx=y&param1=1"
assert not s._hash(r) == s._hash(r2)
# not ignorable parameter missing
r2.request.content = b"param1=1"
assert not s._hash(r) == s._hash(r2)
def test_server_playback_full(self):
state = flow.State()
s = serverplayback.ServerPlayback()
o = options.Options(refresh_server_playback = True, keepserving=False)
m = mastertest.RecordingMaster(o, None, state)
m.addons.add(o, s)
f = tutils.tflow()
f.response = netlib.tutils.tresp(content=f.request.content)
s.load([f, f])
tf = tutils.tflow()
assert not tf.response
m.request(tf)
assert tf.response == f.response
tf = tutils.tflow()
tf.request.content = b"gibble"
assert not tf.response
m.request(tf)
assert not tf.response
assert not s.stop
s.tick()
assert not s.stop
tf = tutils.tflow()
m.request(tutils.tflow())
assert s.stop
def test_server_playback_kill(self):
state = flow.State()
s = serverplayback.ServerPlayback()
o = options.Options(refresh_server_playback = True, kill=True)
m = mastertest.RecordingMaster(o, None, state)
m.addons.add(o, s)
f = tutils.tflow()
f.response = netlib.tutils.tresp(content=f.request.content)
s.load([f])
f = tutils.tflow()
f.request.host = "nonexistent"
m.request(f)
assert f.reply.value == exceptions.Kill

8
test/mitmproxy/mastertest.py
View File

@ -5,6 +5,10 @@ from mitmproxy.flow import master
from mitmproxy import flow, proxy, models, controller
class TestMaster:
pass
class MasterTest:
def cycle(self, master, content):
@ -16,7 +20,9 @@ class MasterTest:
master.serverconnect(f.server_conn)
master.request(f)
if not f.error:
f.response = models.HTTPResponse.wrap(netlib.tutils.tresp(content=content))
f.response = models.HTTPResponse.wrap(
netlib.tutils.tresp(content=content)
)
master.response(f)
master.clientdisconnect(f)
return f

2
test/mitmproxy/test_dump.py
View File

@ -50,7 +50,7 @@ class TestDumpMaster(mastertest.MasterTest):
def test_replay(self):
o = dump.Options(server_replay=["nonexistent"], kill=True)
tutils.raises(dump.DumpError, dump.DumpMaster, None, o)
tutils.raises(exceptions.OptionsError, dump.DumpMaster, None, o)
with tutils.tmpdir() as t:
p = os.path.join(t, "rep")

299
test/mitmproxy/test_flow.py
View File

@ -70,228 +70,6 @@ class TestClientPlaybackState:
assert not fm.client_playback
class TestServerPlaybackState:
def test_hash(self):
s = flow.ServerPlaybackState(
None,
[],
False,
False,
None,
False,
None,
False)
r = tutils.tflow()
r2 = tutils.tflow()
assert s._hash(r)
assert s._hash(r) == s._hash(r2)
r.request.headers["foo"] = "bar"
assert s._hash(r) == s._hash(r2)
r.request.path = "voing"
assert s._hash(r) != s._hash(r2)
r.request.path = "path?blank_value"
r2.request.path = "path?"
assert s._hash(r) != s._hash(r2)
def test_headers(self):
s = flow.ServerPlaybackState(
["foo"],
[],
False,
False,
None,
False,
None,
False)
r = tutils.tflow(resp=True)
r.request.headers["foo"] = "bar"
r2 = tutils.tflow(resp=True)
assert not s._hash(r) == s._hash(r2)
r2.request.headers["foo"] = "bar"
assert s._hash(r) == s._hash(r2)
r2.request.headers["oink"] = "bar"
assert s._hash(r) == s._hash(r2)
r = tutils.tflow(resp=True)
r2 = tutils.tflow(resp=True)
assert s._hash(r) == s._hash(r2)
def test_load(self):
r = tutils.tflow(resp=True)
r.request.headers["key"] = "one"
r2 = tutils.tflow(resp=True)
r2.request.headers["key"] = "two"
s = flow.ServerPlaybackState(
None, [
r, r2], False, False, None, False, None, False)
assert s.count() == 2
assert len(s.fmap.keys()) == 1
n = s.next_flow(r)
assert n.request.headers["key"] == "one"
assert s.count() == 1
n = s.next_flow(r)
assert n.request.headers["key"] == "two"
assert s.count() == 0
assert not s.next_flow(r)
def test_load_with_nopop(self):
r = tutils.tflow(resp=True)
r.request.headers["key"] = "one"
r2 = tutils.tflow(resp=True)
r2.request.headers["key"] = "two"
s = flow.ServerPlaybackState(
None, [
r, r2], False, True, None, False, None, False)
assert s.count() == 2
s.next_flow(r)
assert s.count() == 2
def test_ignore_params(self):
s = flow.ServerPlaybackState(
None, [], False, False, [
"param1", "param2"], False, None, False)
r = tutils.tflow(resp=True)
r.request.path = "/test?param1=1"
r2 = tutils.tflow(resp=True)
r2.request.path = "/test"
assert s._hash(r) == s._hash(r2)
r2.request.path = "/test?param1=2"
assert s._hash(r) == s._hash(r2)
r2.request.path = "/test?param2=1"
assert s._hash(r) == s._hash(r2)
r2.request.path = "/test?param3=2"
assert not s._hash(r) == s._hash(r2)
def test_ignore_payload_params(self):
s = flow.ServerPlaybackState(
None, [], False, False, None, False, [
"param1", "param2"], False)
r = tutils.tflow(resp=True)
r.request.headers["Content-Type"] = "application/x-www-form-urlencoded"
r.request.content = b"paramx=x&param1=1"
r2 = tutils.tflow(resp=True)
r2.request.headers["Content-Type"] = "application/x-www-form-urlencoded"
r2.request.content = b"paramx=x&param1=1"
# same parameters
assert s._hash(r) == s._hash(r2)
# ignored parameters !=
r2.request.content = b"paramx=x&param1=2"
assert s._hash(r) == s._hash(r2)
# missing parameter
r2.request.content = b"paramx=x"
assert s._hash(r) == s._hash(r2)
# ignorable parameter added
r2.request.content = b"paramx=x&param1=2"
assert s._hash(r) == s._hash(r2)
# not ignorable parameter changed
r2.request.content = b"paramx=y&param1=1"
assert not s._hash(r) == s._hash(r2)
# not ignorable parameter missing
r2.request.content = b"param1=1"
assert not s._hash(r) == s._hash(r2)
def test_ignore_payload_params_other_content_type(self):
s = flow.ServerPlaybackState(
None, [], False, False, None, False, [
"param1", "param2"], False)
r = tutils.tflow(resp=True)
r.request.headers["Content-Type"] = "application/json"
r.request.content = b'{"param1":"1"}'
r2 = tutils.tflow(resp=True)
r2.request.headers["Content-Type"] = "application/json"
r2.request.content = b'{"param1":"1"}'
# same content
assert s._hash(r) == s._hash(r2)
# distint content (note only x-www-form-urlencoded payload is analysed)
r2.request.content = b'{"param1":"2"}'
assert not s._hash(r) == s._hash(r2)
def test_ignore_payload_wins_over_params(self):
# NOTE: parameters are mutually exclusive in options
s = flow.ServerPlaybackState(
None, [], False, False, None, True, [
"param1", "param2"], False)
r = tutils.tflow(resp=True)
r.request.headers["Content-Type"] = "application/x-www-form-urlencoded"
r.request.content = b"paramx=y"
r2 = tutils.tflow(resp=True)
r2.request.headers["Content-Type"] = "application/x-www-form-urlencoded"
r2.request.content = b"paramx=x"
# same parameters
assert s._hash(r) == s._hash(r2)
def test_ignore_content(self):
s = flow.ServerPlaybackState(
None,
[],
False,
False,
None,
False,
None,
False)
r = tutils.tflow(resp=True)
r2 = tutils.tflow(resp=True)
r.request.content = b"foo"
r2.request.content = b"foo"
assert s._hash(r) == s._hash(r2)
r2.request.content = b"bar"
assert not s._hash(r) == s._hash(r2)
# now ignoring content
s = flow.ServerPlaybackState(
None,
[],
False,
False,
None,
True,
None,
False)
r = tutils.tflow(resp=True)
r2 = tutils.tflow(resp=True)
r.request.content = b"foo"
r2.request.content = b"foo"
assert s._hash(r) == s._hash(r2)
r2.request.content = b"bar"
assert s._hash(r) == s._hash(r2)
r2.request.content = b""
assert s._hash(r) == s._hash(r2)
r2.request.content = None
assert s._hash(r) == s._hash(r2)
def test_ignore_host(self):
s = flow.ServerPlaybackState(
None,
[],
False,
False,
None,
False,
None,
True)
r = tutils.tflow(resp=True)
r2 = tutils.tflow(resp=True)
r.request.host = "address"
r2.request.host = "address"
assert s._hash(r) == s._hash(r2)
r2.request.host = "wrong_address"
assert s._hash(r) == s._hash(r2)
class TestHTTPFlow(object):
def test_copy(self):
@ -753,16 +531,6 @@ class TestFlowMaster:
DummyServer(ProxyConfig(options.Options())),
s
)
assert not fm.start_server_playback(
pb,
False,
[],
False,
False,
None,
False,
None,
False)
assert not fm.start_client_playback(pb, False)
fm.client_playback.testing = True
@ -773,73 +541,6 @@ class TestFlowMaster:
f.error = Error("error")
fm.error(f)
def test_server_playback(self):
s = flow.State()
f = tutils.tflow()
f.response = HTTPResponse.wrap(netlib.tutils.tresp(content=f.request))
pb = [f]
fm = flow.FlowMaster(options.Options(), None, s)
fm.refresh_server_playback = True
assert not fm.do_server_playback(tutils.tflow())
fm.start_server_playback(
pb,
False,
[],
False,
False,
None,
False,
None,
False)
assert fm.do_server_playback(tutils.tflow())
fm.start_server_playback(
pb,
False,
[],
True,
False,
None,
False,
None,
False)
r = tutils.tflow()
r.request.content = b"gibble"
assert not fm.do_server_playback(r)
assert fm.do_server_playback(tutils.tflow())
fm.tick(0)
assert fm.should_exit.is_set()
fm.stop_server_playback()
assert not fm.server_playback
def test_server_playback_kill(self):
s = flow.State()
f = tutils.tflow()
f.response = HTTPResponse.wrap(netlib.tutils.tresp(content=f.request))
pb = [f]
fm = flow.FlowMaster(None, None, s)
fm.refresh_server_playback = True
fm.start_server_playback(
pb,
True,
[],
False,
False,
None,
False,
None,
False)
f = tutils.tflow()
f.request.host = "nonexistent"
fm.request(f)
assert f.reply.value == Kill
class TestRequest: