-
-
Notifications
You must be signed in to change notification settings - Fork 6.4k
Expand file tree
/
Copy pathtest_har.py
More file actions
201 lines (168 loc) · 8.15 KB
/
Copy pathtest_har.py
File metadata and controls
201 lines (168 loc) · 8.15 KB
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
#!/usr/bin/env python
"""
Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
See the file 'LICENSE' for copying permission
Tests for lib/utils/har.py -- HAR (HTTP Archive) collector and HTTP
request/response parsing used by sqlmap's --har-file feature.
"""
import os
import sys
import unittest
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from _testutils import bootstrap
bootstrap()
from lib.utils import har as H
class TestFakeSocket(unittest.TestCase):
def test_makefile_returns_bytesio(self):
sock = H.FakeSocket(b"hello\r\n")
f = sock.makefile()
self.assertEqual(f.read(), b"hello\r\n")
class TestRawPair(unittest.TestCase):
def test_stores_fields(self):
pair = H.RawPair(b"GET / HTTP/1.0\r\n\r\n",
b"HTTP/1.0 200 OK\r\n\r\n",
startTime=1000, endTime=2000)
self.assertEqual(pair.request, b"GET / HTTP/1.0\r\n\r\n")
self.assertEqual(pair.response, b"HTTP/1.0 200 OK\r\n\r\n")
self.assertEqual(pair.startTime, 1000)
self.assertEqual(pair.endTime, 2000)
class TestHTTPCollector(unittest.TestCase):
def test_collect_and_obtain(self):
c = H.HTTPCollector()
c.collectRequest(b"GET / HTTP/1.0\r\nHost: example.com\r\n\r\n",
b"HTTP/1.0 200 OK\r\nContent-Type: text/html\r\n\r\nbody",
startTime=1000, endTime=2000)
result = c.obtain()
log = result["log"]
self.assertEqual(log["version"], "1.2")
self.assertEqual(log["creator"]["name"], "sqlmap")
entries = log["entries"]
self.assertEqual(len(entries), 1)
self.assertEqual(entries[0]["request"]["method"], "GET")
self.assertEqual(entries[0]["response"]["status"], 200)
class TestHTTPCollectorFactory(unittest.TestCase):
def test_create_returns_collector(self):
f = H.HTTPCollectorFactory(harFile=True)
c = f.create()
self.assertIsInstance(c, H.HTTPCollector)
class TestEntry(unittest.TestCase):
def test_toDict(self):
req = H.Request("GET", "/path", "HTTP/1.1",
{"Host": "example.com"})
resp = H.Response("HTTP/1.1", 200, "OK",
{"Content-Type": "text/html"}, b"body")
entry = H.Entry(req, resp, startTime=1000, endTime=2000,
extendedArguments={})
d = entry.toDict()
self.assertEqual(d["request"]["method"], "GET")
self.assertEqual(d["response"]["status"], 200)
self.assertEqual(d["time"], 1000000)
self.assertIn("startedDateTime", d)
class TestRequest(unittest.TestCase):
def test_parse_simple_get(self):
raw = b"GET /path HTTP/1.1\r\nHost: example.com\r\n\r\n"
req = H.Request.parse(raw)
self.assertEqual(req.method, "GET")
self.assertEqual(req.path, "/path")
self.assertEqual(req.httpVersion, "HTTP/1.1")
self.assertEqual(req.headers.get("Host"), "example.com")
def test_parse_with_comment(self):
raw = (b"HTTP request [#1]:\r\n"
b"POST /submit HTTP/1.0\r\n"
b"Host: example.com\r\n"
b"Content-Type: text/plain\r\n"
b"Content-Length: 4\r\n"
b"\r\n"
b"body")
req = H.Request.parse(raw)
self.assertEqual(req.method, "POST")
self.assertEqual(req.path, "/submit")
self.assertEqual(req.comment, b"HTTP request [#1]:")
self.assertIn(b"body", req.postBody)
def test_toDict(self):
req = H.Request("GET", "/", "HTTP/1.0",
{"Host": "test.com", "Accept": "*/*"})
d = req.toDict()
self.assertEqual(d["method"], "GET")
self.assertEqual(d["url"], "http://test.com/")
self.assertEqual(len(d["headers"]), 2)
def test_toDict_with_postbody(self):
req = H.Request("POST", "/", "HTTP/1.1",
{"Host": "test.com", "Content-Type": "application/json"},
postBody=b'{"a":1}')
d = req.toDict()
self.assertEqual(d["postData"]["mimeType"], "application/json")
self.assertIn('{"a":1}', d["postData"]["text"])
def test_toDict_binary_postbody_base64(self):
# Regression: a non-UTF-8 (binary) POST body - e.g. a raw file upload - must be base64-encoded
# losslessly, not mangled through a lossy text decode, so the exported HAR reproduces the request.
# Mirrors the Response.toDict() contract (see TestResponse.test_toDict_binary_content_encoded).
import base64 as _b64
payload = b"\xff\xd8\xff\xe0\x00\x10JFIF" # JPEG header: invalid UTF-8, contains a NUL
req = H.Request("POST", "/upload", "HTTP/1.1",
{"Host": "test.com", "Content-Type": "application/octet-stream"},
postBody=payload)
d = req.toDict()
self.assertEqual(d["postData"]["encoding"], "base64")
self.assertEqual(_b64.b64decode(d["postData"]["text"]), payload) # losslessly reconstructable
def test_url_property(self):
req = H.Request("GET", "/path?q=1", "HTTP/1.0",
{"Host": "example.com"})
self.assertEqual(req.url, "http://example.com/path?q=1")
def test_url_no_host_header(self):
req = H.Request("GET", "/", "HTTP/1.0", {})
self.assertIn("unknown", req.url)
class TestResponse(unittest.TestCase):
def test_parse_simple(self):
raw = b"HTTP/1.1 200 OK\r\nContent-Type: text/html\r\nContent-Length: 4\r\n\r\nbody"
resp = H.Response.parse(raw)
self.assertEqual(resp.status, 200)
self.assertEqual(resp.statusText, "OK")
self.assertEqual(resp.headers.get("Content-Type"), "text/html")
self.assertEqual(resp.content, b"body")
def test_parse_with_comment(self):
raw = (b"HTTP response [#1] (200 Fine):\r\n"
b"HTTP/1.0 200 Fine\r\n"
b"Content-Type: text/plain\r\n"
b"\r\n"
b"response body")
resp = H.Response.parse(raw)
self.assertEqual(resp.status, 200)
self.assertEqual(resp.statusText, "Fine")
self.assertIn(b"HTTP response", resp.comment)
def test_toDict(self):
resp = H.Response("HTTP/1.1", 404, "Not Found",
{"Content-Type": "text/html"}, b"not found")
d = resp.toDict()
self.assertEqual(d["status"], 404)
self.assertEqual(d["statusText"], "Not Found")
self.assertEqual(d["content"]["text"], "not found")
self.assertEqual(d["content"]["size"], 9)
def test_toDict_binary_content_encoded(self):
resp = H.Response("HTTP/1.1", 200, "OK",
{"Content-Type": "application/octet-stream"},
b"\x00\x01\xff")
d = resp.toDict()
self.assertEqual(d["content"]["encoding"], "base64")
def test_toDict_binary_without_null_still_base64(self):
# regression: binary content lacking NUL/0x01 (e.g. a JPEG header) must still be
# base64-encoded, not mangled into the "text" field as a lossy decode
import base64 as _b64
jpeg = b"\xff\xd8\xff\xe0\x10JFIF\x02\x03"
d = H.Response("HTTP/1.1", 200, "OK", {"Content-Type": "image/jpeg"}, jpeg).toDict()
self.assertEqual(d["content"]["encoding"], "base64")
self.assertEqual(_b64.b64decode(d["content"]["text"]), jpeg) # lossless
def test_toDict_utf8_multibyte_stays_text(self):
# valid UTF-8 (incl. multibyte) is human-readable text, not base64
original = b"caf\xc3\xa9 \xe4\xbd\xa0\xe5\xa5\xbd".decode("utf-8") # cafe+e-acute+two CJK, pure-ASCII source
d = H.Response("HTTP/1.1", 200, "OK", {"Content-Type": "text/plain; charset=utf-8"},
original.encode("utf-8")).toDict()
self.assertNotIn("encoding", d["content"])
self.assertEqual(d["content"]["text"], original)
def test_toDict_non_text_content(self):
resp = H.Response("HTTP/1.1", 200, "OK",
{"Content-Type": "text/plain"}, b"plain text")
d = resp.toDict()
self.assertEqual(d["content"]["text"], "plain text")
if __name__ == "__main__":
unittest.main(verbosity=2)