From e6eeab60946e61047ed858422badbda189a6f9e8 Mon Sep 17 00:00:00 2001 From: Aldo Cortesi Date: Tue, 25 Apr 2017 19:06:24 +1200 Subject: Revamp how addons work - Addons now nest, which means that addons can manage addons. This has a number of salutary effects - the scripts addon no longer has to poke into the global addons list, we no longer have to replace/remove/boot-outof parent addons when we load scripts, and this paves the way for making our top-level tools into addons themselves. - All addon calls are now wrapped in a safe execution environment where exceptions are caught, and output to stdout/stderr are intercepted and turned into logs. - We no longer support script arguments in sys.argv - creating an option properly is the only way to pass arguments. This means that all scripts are always directly controllable from interctive tooling, and that arguments are type-checked. For now, I've disabled testing of the har dump example - it needs to be moved to the new argument handling, and become a class addon. I'll address that in a separate patch. --- test/examples/_test_har_dump.py | 114 ++++++++++++++++++++++++++++++++++++++++ test/examples/test_examples.py | 31 +++++------ test/examples/test_har_dump.py | 114 ---------------------------------------- 3 files changed, 128 insertions(+), 131 deletions(-) create mode 100644 test/examples/_test_har_dump.py delete mode 100644 test/examples/test_har_dump.py (limited to 'test/examples') diff --git a/test/examples/_test_har_dump.py b/test/examples/_test_har_dump.py new file mode 100644 index 00000000..e5cfd2e1 --- /dev/null +++ b/test/examples/_test_har_dump.py @@ -0,0 +1,114 @@ +import json +import shlex +import pytest + +from mitmproxy import options +from mitmproxy import proxy +from mitmproxy import master +from mitmproxy.addons import script + +from mitmproxy.test import tflow +from mitmproxy.test import tutils +from mitmproxy.net.http import cookies + +example_dir = tutils.test_data.push("../examples") + + +class ScriptError(Exception): + pass + + +class RaiseMaster(master.Master): + def add_log(self, e, level): + if level in ("warn", "error"): + raise ScriptError(e) + + +def tscript(cmd, args=""): + o = options.Options() + cmd = example_dir.path(cmd) + " " + args + m = RaiseMaster(o, proxy.DummyServer()) + sc = script.Script(cmd) + m.addons.add(sc) + return m, sc + + +class TestHARDump: + + def flow(self, resp_content=b'message'): + times = dict( + timestamp_start=746203272, + timestamp_end=746203272, + ) + + # Create a dummy flow for testing + return tflow.tflow( + req=tutils.treq(method=b'GET', **times), + resp=tutils.tresp(content=resp_content, **times) + ) + + def test_no_file_arg(self): + with pytest.raises(ScriptError): + tscript("complex/har_dump.py") + + def test_simple(self, tmpdir): + path = str(tmpdir.join("somefile")) + + m, sc = tscript("complex/har_dump.py", shlex.quote(path)) + m.addons.trigger("response", self.flow()) + m.addons.remove(sc) + + with open(path, "r") as inp: + har = json.load(inp) + assert len(har["log"]["entries"]) == 1 + + def test_base64(self, tmpdir): + path = str(tmpdir.join("somefile")) + + m, sc = tscript("complex/har_dump.py", shlex.quote(path)) + m.addons.trigger( + "response", self.flow(resp_content=b"foo" + b"\xFF" * 10) + ) + m.addons.remove(sc) + + with open(path, "r") as inp: + har = json.load(inp) + assert har["log"]["entries"][0]["response"]["content"]["encoding"] == "base64" + + def test_format_cookies(self): + m, sc = tscript("complex/har_dump.py", "-") + format_cookies = sc.ns.format_cookies + + CA = cookies.CookieAttrs + + f = format_cookies([("n", "v", CA([("k", "v")]))])[0] + assert f['name'] == "n" + assert f['value'] == "v" + assert not f['httpOnly'] + assert not f['secure'] + + f = format_cookies([("n", "v", CA([("httponly", None), ("secure", None)]))])[0] + assert f['httpOnly'] + assert f['secure'] + + f = format_cookies([("n", "v", CA([("expires", "Mon, 24-Aug-2037 00:00:00 GMT")]))])[0] + assert f['expires'] + + def test_binary(self, tmpdir): + + f = self.flow() + f.request.method = "POST" + f.request.headers["content-type"] = "application/x-www-form-urlencoded" + f.request.content = b"foo=bar&baz=s%c3%bc%c3%9f" + f.response.headers["random-junk"] = bytes(range(256)) + f.response.content = bytes(range(256)) + + path = str(tmpdir.join("somefile")) + + m, sc = tscript("complex/har_dump.py", shlex.quote(path)) + m.addons.trigger("response", f) + m.addons.remove(sc) + + with open(path, "r") as inp: + har = json.load(inp) + assert len(har["log"]["entries"]) == 1 diff --git a/test/examples/test_examples.py b/test/examples/test_examples.py index 46fdcd36..4b691df2 100644 --- a/test/examples/test_examples.py +++ b/test/examples/test_examples.py @@ -1,5 +1,3 @@ -import pytest - from mitmproxy import options from mitmproxy import contentviews from mitmproxy import proxy @@ -8,6 +6,7 @@ from mitmproxy.addons import script from mitmproxy.test import tflow from mitmproxy.test import tutils +from mitmproxy.test import taddons from mitmproxy.net.http import Headers from ..mitmproxy import tservers @@ -27,7 +26,7 @@ class RaiseMaster(master.Master): def tscript(cmd, args=""): o = options.Options() - cmd = example_dir.path(cmd) + " " + args + cmd = example_dir.path(cmd) m = RaiseMaster(o, proxy.DummyServer()) sc = script.Script(cmd) m.addons.add(sc) @@ -48,14 +47,18 @@ class TestScripts(tservers.MasterTest): assert any(b'tEST!' in val[0][1] for val in fmt) def test_iframe_injector(self): - with pytest.raises(ScriptError): - tscript("simple/modify_body_inject_iframe.py") - - m, sc = tscript("simple/modify_body_inject_iframe.py", "http://example.org/evil_iframe") - f = tflow.tflow(resp=tutils.tresp(content=b"mitmproxy")) - m.addons.handle_lifecycle("response", f) - content = f.response.content - assert b'iframe' in content and b'evil_iframe' in content + with taddons.context() as tctx: + sc = tctx.script(example_dir.path("simple/modify_body_inject_iframe.py")) + tctx.configure( + sc, + iframe = "http://example.org/evil_iframe" + ) + f = tflow.tflow( + resp=tutils.tresp(content=b"mitmproxy") + ) + tctx.master.addons.invoke_addon(sc, "response", f) + content = f.response.content + assert b'iframe' in content and b'evil_iframe' in content def test_modify_form(self): m, sc = tscript("simple/modify_form.py") @@ -81,12 +84,6 @@ class TestScripts(tservers.MasterTest): m.addons.handle_lifecycle("request", f) assert f.request.query["mitmproxy"] == "rocks" - def test_arguments(self): - m, sc = tscript("simple/script_arguments.py", "mitmproxy rocks") - f = tflow.tflow(resp=tutils.tresp(content=b"I <3 mitmproxy")) - m.addons.handle_lifecycle("response", f) - assert f.response.content == b"I <3 rocks" - def test_redirect_requests(self): m, sc = tscript("simple/redirect_requests.py") f = tflow.tflow(req=tutils.treq(host="example.org")) diff --git a/test/examples/test_har_dump.py b/test/examples/test_har_dump.py deleted file mode 100644 index e5cfd2e1..00000000 --- a/test/examples/test_har_dump.py +++ /dev/null @@ -1,114 +0,0 @@ -import json -import shlex -import pytest - -from mitmproxy import options -from mitmproxy import proxy -from mitmproxy import master -from mitmproxy.addons import script - -from mitmproxy.test import tflow -from mitmproxy.test import tutils -from mitmproxy.net.http import cookies - -example_dir = tutils.test_data.push("../examples") - - -class ScriptError(Exception): - pass - - -class RaiseMaster(master.Master): - def add_log(self, e, level): - if level in ("warn", "error"): - raise ScriptError(e) - - -def tscript(cmd, args=""): - o = options.Options() - cmd = example_dir.path(cmd) + " " + args - m = RaiseMaster(o, proxy.DummyServer()) - sc = script.Script(cmd) - m.addons.add(sc) - return m, sc - - -class TestHARDump: - - def flow(self, resp_content=b'message'): - times = dict( - timestamp_start=746203272, - timestamp_end=746203272, - ) - - # Create a dummy flow for testing - return tflow.tflow( - req=tutils.treq(method=b'GET', **times), - resp=tutils.tresp(content=resp_content, **times) - ) - - def test_no_file_arg(self): - with pytest.raises(ScriptError): - tscript("complex/har_dump.py") - - def test_simple(self, tmpdir): - path = str(tmpdir.join("somefile")) - - m, sc = tscript("complex/har_dump.py", shlex.quote(path)) - m.addons.trigger("response", self.flow()) - m.addons.remove(sc) - - with open(path, "r") as inp: - har = json.load(inp) - assert len(har["log"]["entries"]) == 1 - - def test_base64(self, tmpdir): - path = str(tmpdir.join("somefile")) - - m, sc = tscript("complex/har_dump.py", shlex.quote(path)) - m.addons.trigger( - "response", self.flow(resp_content=b"foo" + b"\xFF" * 10) - ) - m.addons.remove(sc) - - with open(path, "r") as inp: - har = json.load(inp) - assert har["log"]["entries"][0]["response"]["content"]["encoding"] == "base64" - - def test_format_cookies(self): - m, sc = tscript("complex/har_dump.py", "-") - format_cookies = sc.ns.format_cookies - - CA = cookies.CookieAttrs - - f = format_cookies([("n", "v", CA([("k", "v")]))])[0] - assert f['name'] == "n" - assert f['value'] == "v" - assert not f['httpOnly'] - assert not f['secure'] - - f = format_cookies([("n", "v", CA([("httponly", None), ("secure", None)]))])[0] - assert f['httpOnly'] - assert f['secure'] - - f = format_cookies([("n", "v", CA([("expires", "Mon, 24-Aug-2037 00:00:00 GMT")]))])[0] - assert f['expires'] - - def test_binary(self, tmpdir): - - f = self.flow() - f.request.method = "POST" - f.request.headers["content-type"] = "application/x-www-form-urlencoded" - f.request.content = b"foo=bar&baz=s%c3%bc%c3%9f" - f.response.headers["random-junk"] = bytes(range(256)) - f.response.content = bytes(range(256)) - - path = str(tmpdir.join("somefile")) - - m, sc = tscript("complex/har_dump.py", shlex.quote(path)) - m.addons.trigger("response", f) - m.addons.remove(sc) - - with open(path, "r") as inp: - har = json.load(inp) - assert len(har["log"]["entries"]) == 1 -- cgit v1.2.3