diff options
| author | Thomas Kriechbaumer <Kriechi@users.noreply.github.com> | 2017-03-22 12:02:18 +0100 |
|---|---|---|
| committer | GitHub <noreply@github.com> | 2017-03-22 12:02:18 +0100 |
| commit | 907825714bf434efdee3ea99707aa509bb0f0c58 (patch) | |
| tree | 5652ffbfe0ec94f853ebbe12d1f40e0470142423 /test/examples/test_examples.py | |
| parent | ab2fcbef8dceb5abc97f7e73638629ded606daa6 (diff) | |
| download | mitmproxy-907825714bf434efdee3ea99707aa509bb0f0c58.tar.gz mitmproxy-907825714bf434efdee3ea99707aa509bb0f0c58.tar.bz2 mitmproxy-907825714bf434efdee3ea99707aa509bb0f0c58.zip | |
move examples tests (#2199)
Diffstat (limited to 'test/examples/test_examples.py')
| -rw-r--r-- | test/examples/test_examples.py | 120 |
1 files changed, 120 insertions, 0 deletions
diff --git a/test/examples/test_examples.py b/test/examples/test_examples.py new file mode 100644 index 00000000..46fdcd36 --- /dev/null +++ b/test/examples/test_examples.py @@ -0,0 +1,120 @@ +import pytest + +from mitmproxy import options +from mitmproxy import contentviews +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 Headers + +from ..mitmproxy import tservers + +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 TestScripts(tservers.MasterTest): + def test_add_header(self): + m, _ = tscript("simple/add_header.py") + f = tflow.tflow(resp=tutils.tresp()) + m.addons.handle_lifecycle("response", f) + assert f.response.headers["newheader"] == "foo" + + def test_custom_contentviews(self): + m, sc = tscript("simple/custom_contentview.py") + swapcase = contentviews.get("swapcase") + _, fmt = swapcase(b"<html>Test!</html>") + 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"<html><body>mitmproxy</body></html>")) + m.addons.handle_lifecycle("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") + + form_header = Headers(content_type="application/x-www-form-urlencoded") + f = tflow.tflow(req=tutils.treq(headers=form_header)) + m.addons.handle_lifecycle("request", f) + + assert f.request.urlencoded_form["mitmproxy"] == "rocks" + + f.request.headers["content-type"] = "" + m.addons.handle_lifecycle("request", f) + assert list(f.request.urlencoded_form.items()) == [("foo", "bar")] + + def test_modify_querystring(self): + m, sc = tscript("simple/modify_querystring.py") + f = tflow.tflow(req=tutils.treq(path="/search?q=term")) + + m.addons.handle_lifecycle("request", f) + assert f.request.query["mitmproxy"] == "rocks" + + f.request.path = "/" + 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")) + m.addons.handle_lifecycle("request", f) + assert f.request.host == "mitmproxy.org" + + def test_send_reply_from_proxy(self): + m, sc = tscript("simple/send_reply_from_proxy.py") + f = tflow.tflow(req=tutils.treq(host="example.com", port=80)) + m.addons.handle_lifecycle("request", f) + assert f.response.content == b"Hello World" + + def test_dns_spoofing(self): + m, sc = tscript("complex/dns_spoofing.py") + original_host = "example.com" + + host_header = Headers(host=original_host) + f = tflow.tflow(req=tutils.treq(headers=host_header, port=80)) + + m.addons.handle_lifecycle("requestheaders", f) + + # Rewrite by reverse proxy mode + f.request.scheme = "https" + f.request.port = 443 + + m.addons.handle_lifecycle("request", f) + + assert f.request.scheme == "http" + assert f.request.port == 80 + + assert f.request.headers["Host"] == original_host |