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

fix #2257

Browse Source
This commit is contained in:
Maximilian Hils 2017-04-26 12:08:16 +02:00
parent 07cb83597b
commit 1aa6d9d324
4 changed files with 110 additions and 79 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
mitmproxy/export.py
View File

@ -6,19 +6,7 @@ import textwrap
from typing import Any
from mitmproxy import http
def _native(s):
if isinstance(s, bytes):
return s.decode()
return s
def dictstr(items, indent: str) -> str:
lines = []
for k, v in items:
lines.append(indent + "%s: %s,\n" % (repr(_native(k)), repr(_native(v))))
return "{\n%s}\n" % "".join(lines)
from mitmproxy.utils import strutils
def curl_command(flow: http.HTTPFlow) -> str:
@ -36,7 +24,10 @@ def curl_command(flow: http.HTTPFlow) -> str:
data += "'%s'" % request.url
if request.content:
data += " --data-binary '%s'" % _native(request.content)
data += " --data-binary '%s'" % strutils.bytes_to_escaped_str(
request.content,
escape_single_quotes=True
)
return data
@ -127,10 +118,14 @@ def locust_code(flow):
args = ""
headers = ""
def conv(x):
return strutils.bytes_to_escaped_str(x, escape_single_quotes=True)
if flow.request.headers:
lines = [
(_native(k), _native(v)) for k, v in flow.request.headers.fields
if _native(k).lower() not in [":authority", "host", "cookie"]
(conv(k), conv(v)) for k, v in flow.request.headers.fields
if conv(k).lower() not in [":authority", "host", "cookie"]
]
lines = [" '%s': '%s',\n" % (k, v) for k, v in lines]
headers += "\n headers = {\n%s }\n" % "".join(lines)
@ -148,7 +143,7 @@ def locust_code(flow):
data = ""
if flow.request.content:
data = "\n data = '''%s'''\n" % _native(flow.request.content)
data = "\n data = '''%s'''\n" % conv(flow.request.content)
args += "\n data=data,"
code = code.format(

2
test/mitmproxy/data/test_flow_export/locust_task_post.py
View File

@ -2,7 +2,7 @@
def path(self):
url = self.locust.host + '/path'
data = '''content'''
data = '''\x00\x01\x02\x03\x04\x05\x06\x07\x08\t\n\x0b\x0c\r\x0e\x0f\x10\x11\x12\x13\x14\x15\x16\x17\x18\x19\x1a\x1b\x1c\x1d\x1e\x1f !"#$%&\'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefghijklmnopqrstuvwxyz{|}~\x7f\x80\x81\x82\x83\x84\x85\x86\x87\x88\x89\x8a\x8b\x8c\x8d\x8e\x8f\x90\x91\x92\x93\x94\x95\x96\x97\x98\x99\x9a\x9b\x9c\x9d\x9e\x9f\xa0\xa1\xa2\xa3\xa4\xa5\xa6\xa7\xa8\xa9\xaa\xab\xac\xad\xae\xaf\xb0\xb1\xb2\xb3\xb4\xb5\xb6\xb7\xb8\xb9\xba\xbb\xbc\xbd\xbe\xbf\xc0\xc1\xc2\xc3\xc4\xc5\xc6\xc7\xc8\xc9\xca\xcb\xcc\xcd\xce\xcf\xd0\xd1\xd2\xd3\xd4\xd5\xd6\xd7\xd8\xd9\xda\xdb\xdc\xdd\xde\xdf\xe0\xe1\xe2\xe3\xe4\xe5\xe6\xe7\xe8\xe9\xea\xeb\xec\xed\xee\xef\xf0\xf1\xf2\xf3\xf4\xf5\xf6\xf7\xf8\xf9\xfa\xfb\xfc\xfd\xfe\xff'''
self.response = self.client.request(
method='POST',

13
test/mitmproxy/data/test_flow_export/python_post.py
View File

@ -2,7 +2,16 @@ import requests
response = requests.post(
'http://address:22/path',
data=b'content'
data=(b'\x00\x01\x02\x03\x04\x05\x06\x07\x08\t\n\x0b\x0c\r\x0e\x0f\x10\x11\x12\x13'
b'\x14\x15\x16\x17\x18\x19\x1a\x1b\x1c\x1d\x1e\x1f !"#$%&\'()*+,-./01234567'
b'89:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefghijklmnopqrstuvwxyz{|}~\x7f'
b'\x80\x81\x82\x83\x84\x85\x86\x87\x88\x89\x8a\x8b\x8c\x8d\x8e\x8f'
b'\x90\x91\x92\x93\x94\x95\x96\x97\x98\x99\x9a\x9b\x9c\x9d\x9e\x9f'
b'\xa0\xa1\xa2\xa3\xa4\xa5\xa6\xa7\xa8\xa9\xaa\xab\xac\xad\xae\xaf'
b'\xb0\xb1\xb2\xb3\xb4\xb5\xb6\xb7\xb8\xb9\xba\xbb\xbc\xbd\xbe\xbf'
b'\xc0\xc1\xc2\xc3\xc4\xc5\xc6\xc7\xc8\xc9\xca\xcb\xcc\xcd\xce\xcf'
b'\xd0\xd1\xd2\xd3\xd4\xd5\xd6\xd7\xd8\xd9\xda\xdb\xdc\xdd\xde\xdf'
b'\xe0\xe1\xe2\xe3\xe4\xe5\xe6\xe7\xe8\xe9\xea\xeb\xec\xed\xee\xef'
b'\xf0\xf1\xf2\xf3\xf4\xf5\xf6\xf7\xf8\xf9\xfa\xfb\xfc\xfd\xfe\xff')
)
print(response.text)

145
test/mitmproxy/test_export.py
View File

@ -1,13 +1,15 @@
from mitmproxy.test import tflow
import re
from mitmproxy.net.http import Headers
import pytest
from mitmproxy import export # heh
from mitmproxy.net.http import Headers
from mitmproxy.test import tflow
from mitmproxy.test import tutils
def clean_blanks(s):
return re.sub(r"^(\s+)$", "", s, flags=re.MULTILINE)
return re.sub(r"^\s+", "", s, flags=re.MULTILINE)
def python_equals(testdata, text):
@ -19,85 +21,110 @@ def python_equals(testdata, text):
assert clean_blanks(text).rstrip() == clean_blanks(d).rstrip()
def req_get():
return tutils.treq(method=b'GET', content=b'', path=b"/path?a=foo&a=bar&b=baz")
@pytest.fixture
def get_request():
return tflow.tflow(
req=tutils.treq(
method=b'GET',
content=b'',
path=b"/path?a=foo&a=bar&b=baz"
)
)
def req_post():
return tutils.treq(method=b'POST', headers=())
@pytest.fixture
def post_request():
return tflow.tflow(
req=tutils.treq(
method=b'POST',
headers=(),
content=bytes(range(256))
)
)
def req_patch():
return tutils.treq(method=b'PATCH', path=b"/path?query=param")
@pytest.fixture
def patch_request():
return tflow.tflow(
req=tutils.treq(method=b'PATCH', path=b"/path?query=param")
)
class TestExportCurlCommand:
def test_get(self):
flow = tflow.tflow(req=req_get())
class TExport:
def test_get(self, get_request):
raise NotImplementedError()
def test_post(self, post_request):
raise NotImplementedError()
def test_patch(self, patch_request):
raise NotImplementedError()
class TestExportCurlCommand(TExport):
def test_get(self, get_request):
result = """curl -H 'header:qvalue' -H 'content-length:7' 'http://address:22/path?a=foo&a=bar&b=baz'"""
assert export.curl_command(flow) == result
assert export.curl_command(get_request) == result
def test_post(self):
flow = tflow.tflow(req=req_post())
result = """curl -X POST 'http://address:22/path' --data-binary 'content'"""
assert export.curl_command(flow) == result
def test_post(self, post_request):
result = "curl -X POST 'http://address:22/path' --data-binary '{}'".format(
str(bytes(range(256)))[2:-1]
)
assert export.curl_command(post_request) == result
def test_patch(self):
flow = tflow.tflow(req=req_patch())
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'"""
assert export.curl_command(flow) == result
assert export.curl_command(patch_request) == result
class TestExportPythonCode:
def test_get(self):
flow = tflow.tflow(req=req_get())
python_equals("mitmproxy/data/test_flow_export/python_get.py", export.python_code(flow))
class TestExportPythonCode(TExport):
def test_get(self, get_request):
python_equals("mitmproxy/data/test_flow_export/python_get.py",
export.python_code(get_request))
def test_post(self):
flow = tflow.tflow(req=req_post())
python_equals("mitmproxy/data/test_flow_export/python_post.py", export.python_code(flow))
def test_post(self, post_request):
python_equals("mitmproxy/data/test_flow_export/python_post.py",
export.python_code(post_request))
def test_post_json(self):
p = req_post()
p.content = b'{"name": "example", "email": "example@example.com"}'
p.headers = Headers(content_type="application/json")
flow = tflow.tflow(req=p)
python_equals("mitmproxy/data/test_flow_export/python_post_json.py", export.python_code(flow))
def test_post_json(self, post_request):
post_request.request.content = b'{"name": "example", "email": "example@example.com"}'
post_request.request.headers = Headers(content_type="application/json")
python_equals("mitmproxy/data/test_flow_export/python_post_json.py",
export.python_code(post_request))
def test_patch(self):
flow = tflow.tflow(req=req_patch())
python_equals("mitmproxy/data/test_flow_export/python_patch.py", export.python_code(flow))
def test_patch(self, patch_request):
python_equals("mitmproxy/data/test_flow_export/python_patch.py",
export.python_code(patch_request))
class TestExportLocustCode:
def test_get(self):
flow = tflow.tflow(req=req_get())
python_equals("mitmproxy/data/test_flow_export/locust_get.py", export.locust_code(flow))
class TestExportLocustCode(TExport):
def test_get(self, get_request):
python_equals("mitmproxy/data/test_flow_export/locust_get.py",
export.locust_code(get_request))
def test_post(self):
p = req_post()
p.content = b'content'
p.headers = ''
flow = tflow.tflow(req=p)
python_equals("mitmproxy/data/test_flow_export/locust_post.py", export.locust_code(flow))
def test_post(self, post_request):
post_request.request.content = b'content'
post_request.request.headers.clear()
python_equals("mitmproxy/data/test_flow_export/locust_post.py",
export.locust_code(post_request))
def test_patch(self):
flow = tflow.tflow(req=req_patch())
python_equals("mitmproxy/data/test_flow_export/locust_patch.py", export.locust_code(flow))
def test_patch(self, patch_request):
python_equals("mitmproxy/data/test_flow_export/locust_patch.py",
export.locust_code(patch_request))
class TestExportLocustTask:
def test_get(self):
flow = tflow.tflow(req=req_get())
python_equals("mitmproxy/data/test_flow_export/locust_task_get.py", export.locust_task(flow))
class TestExportLocustTask(TExport):
def test_get(self, get_request):
python_equals("mitmproxy/data/test_flow_export/locust_task_get.py",
export.locust_task(get_request))
def test_post(self):
flow = tflow.tflow(req=req_post())
python_equals("mitmproxy/data/test_flow_export/locust_task_post.py", export.locust_task(flow))
def test_post(self, post_request):
python_equals("mitmproxy/data/test_flow_export/locust_task_post.py",
export.locust_task(post_request))
def test_patch(self):
flow = tflow.tflow(req=req_patch())
python_equals("mitmproxy/data/test_flow_export/locust_task_patch.py", export.locust_task(flow))
def test_patch(self, patch_request):
python_equals("mitmproxy/data/test_flow_export/locust_task_patch.py",
export.locust_task(patch_request))
class TestURL: