1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
|
from unittest import mock
import pytest
from mitmproxy import exceptions
from mitmproxy.test import tflow
from mitmproxy.net.http import http1
from mitmproxy.net.tcp import TCPClient
from mitmproxy.test.tutils import treq
from ... import tservers
class TestHTTPFlow:
def test_repr(self):
f = tflow.tflow(resp=True, err=True)
assert repr(f)
class TestInvalidRequests(tservers.HTTPProxyTest):
ssl = True
def test_double_connect(self):
p = self.pathoc()
with p.connect():
r = p.request("connect:'%s:%s'" % ("127.0.0.1", self.server2.port))
assert r.status_code == 400
assert b"Unexpected CONNECT" in r.content
def test_relative_request(self):
p = self.pathoc_raw()
with p.connect():
r = p.request("get:/p/200")
assert r.status_code == 400
assert b"Invalid HTTP request form" in r.content
class TestProxyMisconfiguration(tservers.TransparentProxyTest):
def test_absolute_request(self):
p = self.pathoc()
with p.connect():
r = p.request("get:'http://localhost:%d/p/200'" % self.server.port)
assert r.status_code == 400
assert b"misconfiguration" in r.content
class TestExpectHeader(tservers.HTTPProxyTest):
def test_simple(self):
client = TCPClient(("127.0.0.1", self.proxy.port))
client.connect()
# call pathod server, wait a second to complete the request
client.wfile.write(
b"POST http://localhost:%d/p/200 HTTP/1.1\r\n"
b"Expect: 100-continue\r\n"
b"Content-Length: 16\r\n"
b"\r\n" % self.server.port
)
client.wfile.flush()
assert client.rfile.readline() == b"HTTP/1.1 100 Continue\r\n"
assert client.rfile.readline() == b"\r\n"
client.wfile.write(b"0123456789abcdef\r\n")
client.wfile.flush()
resp = http1.read_response(client.rfile, treq())
assert resp.status_code == 200
client.finish()
client.close()
class TestHeadContentLength(tservers.HTTPProxyTest):
def test_head_content_length(self):
p = self.pathoc()
with p.connect():
resp = p.request(
"""head:'%s/p/200:h"Content-Length"="42"'""" % self.server.urlbase
)
assert resp.headers["Content-Length"] == "42"
class TestStreaming(tservers.HTTPProxyTest):
@pytest.mark.parametrize('streaming', [True, False])
def test_streaming(self, streaming):
class Stream:
def requestheaders(self, f):
f.request.stream = streaming
def responseheaders(self, f):
f.response.stream = streaming
def assert_write(self, v):
if streaming:
assert len(v) <= 4096
return self.o.write(v)
self.master.addons.add(Stream())
p = self.pathoc()
with p.connect():
with mock.patch("mitmproxy.net.tcp.Writer.write", side_effect=assert_write, autospec=True):
# response with 10000 bytes
r = p.request("post:'%s/p/200:b@10000'" % self.server.urlbase)
assert len(r.content) == 10000
if streaming:
with pytest.raises(exceptions.HttpReadDisconnect): # as the assertion in assert_write fails
# request with 10000 bytes
p.request("post:'%s/p/200':b@10000" % self.server.urlbase)
else:
assert p.request("post:'%s/p/200':b@10000" % self.server.urlbase)
|
