forked from andrewf/pcap2har
-
Notifications
You must be signed in to change notification settings - Fork 0
/
dpkt_http_replacement.py
205 lines (182 loc) · 8.58 KB
/
dpkt_http_replacement.py
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
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
# $Id: http.py 59 2010-03-24 15:31:17Z jon.oberheide $
"""Hypertext Transfer Protocol.
This version is modified by Andrew Fleenor, on 2 October 2010, to temporarily
fix the bug where a body is parsed for a request that shouldn't have a body."""
import cStringIO
import dpkt
def parse_headers(f):
"""Return dict of HTTP headers parsed from a file object."""
d = {}
while 1:
line = f.readline()
if not line:
raise dpkt.NeedData('premature end of headers')
line = line.strip()
if not line:
break
l = line.split(None, 1)
if not l[0].endswith(':'):
raise dpkt.UnpackError('invalid header: %r' % line)
k = l[0][:-1].lower()
v = len(l) != 1 and l[1] or ''
if k in d:
d[k] += ','+v
else:
d[k] = v
return d
def parse_body(f, headers):
"""Return HTTP body parsed from a file object, given HTTP header dict."""
if headers.get('transfer-encoding', '').lower() == 'chunked':
l = []
found_end = False
while 1:
try:
sz = f.readline().split(None, 1)[0]
except IndexError:
raise dpkt.UnpackError('missing chunk size')
n = int(sz, 16)
if n == 0:
found_end = True
buf = f.read(n)
if f.readline().strip():
break
if n and len(buf) == n:
l.append(buf)
else:
break
if not found_end:
raise dpkt.NeedData('premature end of chunked body')
body = ''.join(l)
elif 'content-length' in headers:
n = int(headers['content-length'])
body = f.read(n)
if len(body) != n:
raise dpkt.NeedData('short body (missing %d bytes)' % (n - len(body)))
else:
# XXX - need to handle HTTP/0.9
body = ''
return body
class Message(dpkt.Packet):
"""Hypertext Transfer Protocol headers + body."""
__metaclass__ = type
__hdr_defaults__ = {}
headers = None
body = None
def __init__(self, *args, **kwargs):
if args:
self.unpack(args[0])
else:
self.headers = {}
self.body = ''
for k, v in self.__hdr_defaults__.iteritems():
setattr(self, k, v)
for k, v in kwargs.iteritems():
setattr(self, k, v)
def unpack(self, buf):
f = cStringIO.StringIO(buf)
# Parse headers
self.headers = parse_headers(f)
# Parse body
self.body = parse_body(f, self.headers)
# Save the rest
self.data = f.read()
def pack_hdr(self):
return ''.join([ '%s: %s\r\n' % t for t in self.headers.iteritems() ])
def __len__(self):
return len(str(self))
def __str__(self):
return '%s\r\n%s' % (self.pack_hdr(), self.body)
class Request(Message):
"""Hypertext Transfer Protocol Request."""
__hdr_defaults__ = {
'method':'GET',
'uri':'/',
'version':'1.0',
}
__methods = dict.fromkeys((
'GET', 'PUT', 'ICY',
'COPY', 'HEAD', 'LOCK', 'MOVE', 'POLL', 'POST',
'BCOPY', 'BMOVE', 'MKCOL', 'TRACE', 'LABEL', 'MERGE',
'DELETE', 'SEARCH', 'UNLOCK', 'REPORT', 'UPDATE', 'NOTIFY',
'BDELETE', 'CONNECT', 'OPTIONS', 'CHECKIN',
'PROPFIND', 'CHECKOUT', 'CCM_POST',
'SUBSCRIBE', 'PROPPATCH', 'BPROPFIND',
'BPROPPATCH', 'UNCHECKOUT', 'MKACTIVITY',
'MKWORKSPACE', 'UNSUBSCRIBE', 'RPC_CONNECT',
'VERSION-CONTROL',
'BASELINE-CONTROL'
))
__proto = 'HTTP'
def unpack(self, buf):
f = cStringIO.StringIO(buf)
line = f.readline()
l = line.strip().split()
if len(l) != 3 or l[0] not in self.__methods or \
not l[2].startswith(self.__proto):
raise dpkt.UnpackError('invalid request: %r' % line)
self.method = l[0]
self.uri = l[1]
self.version = l[2][len(self.__proto)+1:]
Message.unpack(self, f.read())
def __str__(self):
return '%s %s %s/%s\r\n' % (self.method, self.uri, self.__proto,
self.version) + Message.__str__(self)
class Response(Message):
"""Hypertext Transfer Protocol Response."""
__hdr_defaults__ = {
'version':'1.0',
'status':'200',
'reason':'OK'
}
__proto = 'HTTP'
def unpack(self, buf):
f = cStringIO.StringIO(buf)
line = f.readline()
l = line.strip().split(None, 2)
if len(l) < 2 or not l[0].startswith(self.__proto) or not l[1].isdigit():
raise dpkt.UnpackError('invalid response: %r' % line)
self.version = l[0][len(self.__proto)+1:]
self.status = l[1]
self.reason = l[2]
Message.unpack(self, f.read())
def __str__(self):
return '%s/%s %s %s\r\n' % (self.__proto, self.version, self.status,
self.reason) + Message.__str__(self)
if __name__ == '__main__':
import unittest
class HTTPTest(unittest.TestCase):
def test_parse_request(self):
s = """POST /main/redirect/ab/1,295,,00.html HTTP/1.0\r\nReferer: http://www.email.com/login/snap/login.jhtml\r\nConnection: Keep-Alive\r\nUser-Agent: Mozilla/4.75 [en] (X11; U; OpenBSD 2.8 i386; Nav)\r\nHost: ltd.snap.com\r\nAccept: image/gif, image/x-xbitmap, image/jpeg, image/pjpeg, image/png, */*\r\nAccept-Encoding: gzip\r\nAccept-Language: en\r\nAccept-Charset: iso-8859-1,*,utf-8\r\nContent-type: application/x-www-form-urlencoded\r\nContent-length: 61\r\n\r\nsn=em&mn=dtest4&pw=this+is+atest&fr=true&login=Sign+in&od=www"""
r = Request(s)
assert r.method == 'POST'
assert r.uri == '/main/redirect/ab/1,295,,00.html'
assert r.body == 'sn=em&mn=dtest4&pw=this+is+atest&fr=true&login=Sign+in&od=www'
assert r.headers['content-type'] == 'application/x-www-form-urlencoded'
try:
r = Request(s[:60])
assert 'invalid headers parsed!'
except dpkt.UnpackError:
pass
def test_format_request(self):
r = Request()
assert str(r) == 'GET / HTTP/1.0\r\n\r\n'
r.method = 'POST'
r.uri = '/foo/bar/baz.html'
r.headers['content-type'] = 'text/plain'
r.headers['content-length'] = '5'
r.body = 'hello'
assert str(r) == 'POST /foo/bar/baz.html HTTP/1.0\r\ncontent-length: 5\r\ncontent-type: text/plain\r\n\r\nhello'
r = Request(str(r))
assert str(r) == 'POST /foo/bar/baz.html HTTP/1.0\r\ncontent-length: 5\r\ncontent-type: text/plain\r\n\r\nhello'
def test_chunked_response(self):
s = """HTTP/1.1 200 OK\r\nCache-control: no-cache\r\nPragma: no-cache\r\nContent-Type: text/javascript; charset=utf-8\r\nContent-Encoding: gzip\r\nTransfer-Encoding: chunked\r\nSet-Cookie: S=gmail=agg:gmail_yj=v2s:gmproxy=JkU; Domain=.google.com; Path=/\r\nServer: GFE/1.3\r\nDate: Mon, 12 Dec 2005 22:33:23 GMT\r\n\r\na\r\n\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\x00\r\n152\r\nm\x91MO\xc4 \x10\x86\xef\xfe\n\x82\xc9\x9eXJK\xe9\xb6\xee\xc1\xe8\x1e6\x9e4\xf1\xe0a5\x86R\xda\x12Yh\x80\xba\xfa\xef\x85\xee\x1a/\xf21\x99\x0c\xef0<\xc3\x81\xa0\xc3\x01\xe6\x10\xc1<\xa7eYT5\xa1\xa4\xac\xe1\xdb\x15:\xa4\x9d\x0c\xfa5K\x00\xf6.\xaa\xeb\x86\xd5y\xcdHY\x954\x8e\xbc*h\x8c\x8e!L7Y\xe6\'\xeb\x82WZ\xcf>8\x1ed\x87\x851X\xd8c\xe6\xbc\x17Z\x89\x8f\xac \x84e\xde\n!]\x96\x17i\xb5\x02{{\xc2z0\x1e\x0f#7\x9cw3v\x992\x9d\xfc\xc2c8\xea[/EP\xd6\xbc\xce\x84\xd0\xce\xab\xf7`\'\x1f\xacS\xd2\xc7\xd2\xfb\x94\x02N\xdc\x04\x0f\xee\xba\x19X\x03TtW\xd7\xb4\xd9\x92\n\xbcX\xa7;\xb0\x9b\'\x10$?F\xfd\xf3CzPt\x8aU\xef\xb8\xc8\x8b-\x18\xed\xec<\xe0\x83\x85\x08!\xf8"[\xb0\xd3j\x82h\x93\xb8\xcf\xd8\x9b\xba\xda\xd0\x92\x14\xa4a\rc\reM\xfd\x87=X;h\xd9j;\xe0db\x17\xc2\x02\xbd\xb0F\xc2in#\xfb:\xb6\xc4x\x15\xd6\x9f\x8a\xaf\xcf)\x0b^\xbc\xe7i\x11\x80\x8b\x00D\x01\xd8/\x82x\xf6\xd8\xf7J(\xae/\x11p\x1f+\xc4p\t:\xfe\xfd\xdf\xa3Y\xfa\xae4\x7f\x00\xc5\xa5\x95\xa1\xe2\x01\x00\x00\r\n0\r\n\r\n"""
r = Response(s)
assert r.version == '1.1'
assert r.status == '200'
assert r.reason == 'OK'
def test_multicookie_response(self):
s = """HTTP/1.x 200 OK\r\nSet-Cookie: first_cookie=cookie1; path=/; domain=.example.com\r\nSet-Cookie: second_cookie=cookie2; path=/; domain=.example.com\r\nContent-Length: 0\r\n\r\n"""
r = Response(s)
assert type(r.headers['set-cookie']) is list
assert len(r.headers['set-cookie']) == 2
unittest.main()