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
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
|
from libmproxy.protocol.http import *
from cStringIO import StringIO
import tutils, tservers
def test_HttpAuthenticationError():
x = HttpAuthenticationError({"foo": "bar"})
assert str(x)
assert "foo" in x.headers
def test_stripped_chunked_encoding_no_content():
"""
https://github.com/mitmproxy/mitmproxy/issues/186
"""
r = tutils.tresp(content="")
r.headers["Transfer-Encoding"] = ["chunked"]
assert "Content-Length" in r._assemble_headers()
r = tutils.treq(content="")
r.headers["Transfer-Encoding"] = ["chunked"]
assert "Content-Length" in r._assemble_headers()
class TestHTTPRequest:
def test_asterisk_form(self):
s = StringIO("OPTIONS * HTTP/1.1")
f = tutils.tflow(req=None)
f.request = HTTPRequest.from_stream(s)
assert f.request.form_in == "relative"
f.request.host = f.server_conn.address.host
f.request.port = f.server_conn.address.port
f.request.scheme = "http"
assert f.request.assemble() == "OPTIONS * HTTP/1.1\r\nHost: address:22\r\n\r\n"
def test_origin_form(self):
s = StringIO("GET /foo\xff HTTP/1.1")
tutils.raises("Bad HTTP request line", HTTPRequest.from_stream, s)
s = StringIO("GET /foo HTTP/1.1\r\nConnection: Upgrade\r\nUpgrade: h2c")
r = HTTPRequest.from_stream(s)
assert r.headers["Upgrade"] == ["h2c"]
raw = r._assemble_headers()
assert "Upgrade" not in raw
assert "Host" not in raw
r.url = "http://example.com/foo"
raw = r._assemble_headers()
assert "Host" in raw
assert not "Host" in r.headers
r.update_host_header()
assert "Host" in r.headers
def test_authority_form(self):
s = StringIO("CONNECT oops-no-port.com HTTP/1.1")
tutils.raises("Bad HTTP request line", HTTPRequest.from_stream, s)
s = StringIO("CONNECT address:22 HTTP/1.1")
r = HTTPRequest.from_stream(s)
r.scheme, r.host, r.port = "http", "address", 22
assert r.assemble() == "CONNECT address:22 HTTP/1.1\r\nHost: address:22\r\n\r\n"
assert r.pretty_url(False) == "address:22"
def test_absolute_form(self):
s = StringIO("GET oops-no-protocol.com HTTP/1.1")
tutils.raises("Bad HTTP request line", HTTPRequest.from_stream, s)
s = StringIO("GET http://address:22/ HTTP/1.1")
r = HTTPRequest.from_stream(s)
assert r.assemble() == "GET http://address:22/ HTTP/1.1\r\nHost: address:22\r\n\r\n"
def test_assemble_unknown_form(self):
r = tutils.treq()
tutils.raises("Invalid request form", r.assemble, "antiauthority")
def test_set_url(self):
r = tutils.treq_absolute()
r.url = "https://otheraddress:42/ORLY"
assert r.scheme == "https"
assert r.host == "otheraddress"
assert r.port == 42
assert r.path == "/ORLY"
def test_repr(self):
r = tutils.treq()
assert repr(r)
class TestHTTPResponse:
def test_read_from_stringio(self):
_s = "HTTP/1.1 200 OK\r\n" \
"Content-Length: 7\r\n" \
"\r\n"\
"content\r\n" \
"HTTP/1.1 204 OK\r\n" \
"\r\n"
s = StringIO(_s)
r = HTTPResponse.from_stream(s, "GET")
assert r.code == 200
assert r.content == "content"
assert HTTPResponse.from_stream(s, "GET").code == 204
s = StringIO(_s)
r = HTTPResponse.from_stream(s, "HEAD") # HEAD must not have content by spec. We should leave it on the pipe.
assert r.code == 200
assert r.content == ""
tutils.raises("Invalid server response: 'content", HTTPResponse.from_stream, s, "GET")
def test_repr(self):
r = tutils.tresp()
assert "unknown content type" in repr(r)
r.headers["content-type"] = ["foo"]
assert "foo" in repr(r)
assert repr(tutils.tresp(content=CONTENT_MISSING))
class TestHTTPFlow(object):
def test_repr(self):
f = tutils.tflow(resp=True, err=True)
assert repr(f)
class TestInvalidRequests(tservers.HTTPProxTest):
ssl = True
def test_double_connect(self):
p = self.pathoc()
r = p.request("connect:'%s:%s'" % ("127.0.0.1", self.server2.port))
assert r.status_code == 400
assert "Must not CONNECT on already encrypted connection" in r.content
def test_relative_request(self):
p = self.pathoc_raw()
p.connect()
r = p.request("get:/p/200")
assert r.status_code == 400
assert "Invalid HTTP request form" in r.content
|
