aboutsummaryrefslogtreecommitdiffstats
path: root/test
diff options
context:
space:
mode:
Diffstat (limited to 'test')
-rw-r--r--test/mitmproxy/addons/test_export.py65
-rw-r--r--test/mitmproxy/addons/test_serverplayback.py34
-rw-r--r--test/mitmproxy/net/http/test_multipart.py37
-rw-r--r--test/mitmproxy/net/http/test_request.py7
4 files changed, 124 insertions, 19 deletions
diff --git a/test/mitmproxy/addons/test_export.py b/test/mitmproxy/addons/test_export.py
index b0fbeb45..b0e5e47e 100644
--- a/test/mitmproxy/addons/test_export.py
+++ b/test/mitmproxy/addons/test_export.py
@@ -1,4 +1,5 @@
import os
+import shlex
import pytest
import pyperclip
@@ -54,43 +55,87 @@ def tcp_flow():
class TestExportCurlCommand:
def test_get(self, get_request):
- result = """curl -H 'header:qvalue' 'http://address:22/path?a=foo&a=bar&b=baz'"""
+ result = """curl -H 'header: qvalue' 'http://address:22/path?a=foo&a=bar&b=baz'"""
assert export.curl_command(get_request) == result
def test_post(self, post_request):
- result = "curl -H 'content-length:256' -X POST 'http://address:22/path' --data-binary '{}'".format(
- str(bytes(range(256)))[2:-1]
- )
+ post_request.request.content = b'nobinarysupport'
+ result = "curl -X POST http://address:22/path -d nobinarysupport"
assert export.curl_command(post_request) == result
+ def test_fails_with_binary_data(self, post_request):
+ # shlex.quote doesn't support a bytes object
+ # see https://github.com/python/cpython/pull/10871
+ post_request.request.headers["Content-Type"] = "application/json; charset=utf-8"
+ with pytest.raises(exceptions.CommandError):
+ export.curl_command(post_request)
+
def test_patch(self, patch_request):
- result = """curl -H 'header:qvalue' -H 'content-length:7' -X PATCH 'http://address:22/path?query=param' --data-binary 'content'"""
+ result = """curl -H 'header: qvalue' -X PATCH 'http://address:22/path?query=param' -d content"""
assert export.curl_command(patch_request) == result
def test_tcp(self, tcp_flow):
with pytest.raises(exceptions.CommandError):
export.curl_command(tcp_flow)
+ def test_escape_single_quotes_in_body(self):
+ request = tflow.tflow(
+ req=tutils.treq(
+ method=b'POST',
+ headers=(),
+ content=b"'&#"
+ )
+ )
+ command = export.curl_command(request)
+ assert shlex.split(command)[-2] == '-d'
+ assert shlex.split(command)[-1] == "'&#"
+
+ def test_strip_unnecessary(self, get_request):
+ get_request.request.headers.clear()
+ get_request.request.headers["host"] = "address"
+ get_request.request.headers[":authority"] = "address"
+ get_request.request.headers["accept-encoding"] = "br"
+ result = """curl --compressed 'http://address:22/path?a=foo&a=bar&b=baz'"""
+ assert export.curl_command(get_request) == result
+
class TestExportHttpieCommand:
def test_get(self, get_request):
- result = """http GET http://address:22/path?a=foo&a=bar&b=baz 'header:qvalue'"""
+ result = """http GET 'http://address:22/path?a=foo&a=bar&b=baz' 'header: qvalue'"""
assert export.httpie_command(get_request) == result
def test_post(self, post_request):
- result = "http POST http://address:22/path 'content-length:256' <<< '{}'".format(
- str(bytes(range(256)))[2:-1]
- )
+ post_request.request.content = b'nobinarysupport'
+ result = "http POST http://address:22/path <<< nobinarysupport"
assert export.httpie_command(post_request) == result
+ def test_fails_with_binary_data(self, post_request):
+ # shlex.quote doesn't support a bytes object
+ # see https://github.com/python/cpython/pull/10871
+ post_request.request.headers["Content-Type"] = "application/json; charset=utf-8"
+ with pytest.raises(exceptions.CommandError):
+ export.httpie_command(post_request)
+
def test_patch(self, patch_request):
- result = """http PATCH http://address:22/path?query=param 'header:qvalue' 'content-length:7' <<< 'content'"""
+ result = """http PATCH 'http://address:22/path?query=param' 'header: qvalue' <<< content"""
assert export.httpie_command(patch_request) == result
def test_tcp(self, tcp_flow):
with pytest.raises(exceptions.CommandError):
export.httpie_command(tcp_flow)
+ def test_escape_single_quotes_in_body(self):
+ request = tflow.tflow(
+ req=tutils.treq(
+ method=b'POST',
+ headers=(),
+ content=b"'&#"
+ )
+ )
+ command = export.httpie_command(request)
+ assert shlex.split(command)[-2] == '<<<'
+ assert shlex.split(command)[-1] == "'&#"
+
class TestRaw:
def test_req_and_resp_present(self, get_flow):
diff --git a/test/mitmproxy/addons/test_serverplayback.py b/test/mitmproxy/addons/test_serverplayback.py
index c6a0c1f4..2e42fa03 100644
--- a/test/mitmproxy/addons/test_serverplayback.py
+++ b/test/mitmproxy/addons/test_serverplayback.py
@@ -1,13 +1,13 @@
import urllib
-import pytest
-from mitmproxy.test import taddons
-from mitmproxy.test import tflow
+import pytest
import mitmproxy.test.tutils
-from mitmproxy.addons import serverplayback
from mitmproxy import exceptions
from mitmproxy import io
+from mitmproxy.addons import serverplayback
+from mitmproxy.test import taddons
+from mitmproxy.test import tflow
def tdump(path, flows):
@@ -321,7 +321,7 @@ def test_server_playback_full():
with taddons.context(s) as tctx:
tctx.configure(
s,
- server_replay_refresh = True,
+ server_replay_refresh=True,
)
f = tflow.tflow()
@@ -345,7 +345,7 @@ def test_server_playback_kill():
with taddons.context(s) as tctx:
tctx.configure(
s,
- server_replay_refresh = True,
+ server_replay_refresh=True,
server_replay_kill_extra=True
)
@@ -357,3 +357,25 @@ def test_server_playback_kill():
f.request.host = "nonexistent"
tctx.cycle(s, f)
assert f.reply.value == exceptions.Kill
+
+
+def test_server_playback_response_deleted():
+ """
+ The server playback addon holds references to flows that can be modified by the user in the meantime.
+ One thing that can happen is that users remove the response object. This happens for example when doing a client
+ replay at the same time.
+ """
+ sp = serverplayback.ServerPlayback()
+ with taddons.context(sp) as tctx:
+ tctx.configure(sp)
+ f1 = tflow.tflow(resp=True)
+ f2 = tflow.tflow(resp=True)
+
+ assert not sp.flowmap
+
+ sp.load_flows([f1, f2])
+ assert sp.flowmap
+
+ f1.response = f2.response = None
+ assert not sp.next_flow(f1)
+ assert not sp.flowmap
diff --git a/test/mitmproxy/net/http/test_multipart.py b/test/mitmproxy/net/http/test_multipart.py
index 68ae6bbd..6d2e5017 100644
--- a/test/mitmproxy/net/http/test_multipart.py
+++ b/test/mitmproxy/net/http/test_multipart.py
@@ -1,5 +1,6 @@
from mitmproxy.net.http import Headers
from mitmproxy.net.http import multipart
+import pytest
def test_decode():
@@ -22,3 +23,39 @@ def test_decode():
assert len(form) == 2
assert form[0] == (b"field1", b"value1")
assert form[1] == (b"field2", b"value2")
+
+ boundary = 'boundary茅莽'
+ headers = Headers(
+ content_type='multipart/form-data; boundary=' + boundary
+ )
+ result = multipart.decode(headers, content)
+ assert result == []
+
+ headers = Headers(
+ content_type=''
+ )
+ assert multipart.decode(headers, content) == []
+
+
+def test_encode():
+ data = [("file".encode('utf-8'), "shell.jpg".encode('utf-8')),
+ ("file_size".encode('utf-8'), "1000".encode('utf-8'))]
+ headers = Headers(
+ content_type='multipart/form-data; boundary=127824672498'
+ )
+ content = multipart.encode(headers, data)
+
+ assert b'Content-Disposition: form-data; name="file"' in content
+ assert b'Content-Type: text/plain; charset=utf-8\r\n\r\nshell.jpg\r\n\r\n--127824672498\r\n' in content
+ assert b'1000\r\n\r\n--127824672498--\r\n'
+ assert len(content) == 252
+
+ with pytest.raises(ValueError, match=r"boundary found in encoded string"):
+ multipart.encode(headers, [("key".encode('utf-8'), "--127824672498".encode('utf-8'))])
+
+ boundary = 'boundary茅莽'
+ headers = Headers(
+ content_type='multipart/form-data; boundary=' + boundary
+ )
+ result = multipart.encode(headers, data)
+ assert result == b''
diff --git a/test/mitmproxy/net/http/test_request.py b/test/mitmproxy/net/http/test_request.py
index ef581a91..71d5c7a1 100644
--- a/test/mitmproxy/net/http/test_request.py
+++ b/test/mitmproxy/net/http/test_request.py
@@ -371,6 +371,7 @@ class TestRequestUtils:
assert list(request.multipart_form.items()) == []
def test_set_multipart_form(self):
- request = treq(content=b"foobar")
- with pytest.raises(NotImplementedError):
- request.multipart_form = "foobar"
+ request = treq()
+ request.multipart_form = [("file", "shell.jpg"), ("file_size", "1000")]
+ assert request.headers["Content-Type"] == 'multipart/form-data'
+ assert request.content is None
generated by cgit v1.2.3 (git 2.25.1) at 2025-04-03 13:22:48 +0000