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
|
from io import BytesIO
import tempfile
import os
import time
import shutil
from contextlib import contextmanager
from mitmproxy.utils import data
from mitmproxy.net import tcp
from mitmproxy.net import http
test_data = data.Data(__name__).push("../../test/")
@contextmanager
def tmpdir(*args, **kwargs):
orig_workdir = os.getcwd()
temp_workdir = tempfile.mkdtemp(*args, **kwargs)
os.chdir(temp_workdir)
yield temp_workdir
os.chdir(orig_workdir)
shutil.rmtree(temp_workdir)
def treader(bytes):
"""
Construct a tcp.Read object from bytes.
"""
fp = BytesIO(bytes)
return tcp.Reader(fp)
def treq(**kwargs):
"""
Returns:
mitmproxy.net.http.Request
"""
default = dict(
first_line_format="relative",
method=b"GET",
scheme=b"http",
host=b"address",
port=22,
path=b"/path",
http_version=b"HTTP/1.1",
headers=http.Headers(((b"header", b"qvalue"), (b"content-length", b"7"))),
content=b"content"
)
default.update(kwargs)
return http.Request(**default)
def tresp(**kwargs):
"""
Returns:
mitmproxy.net.http.Response
"""
default = dict(
http_version=b"HTTP/1.1",
status_code=200,
reason=b"OK",
headers=http.Headers(((b"header-response", b"svalue"), (b"content-length", b"7"))),
content=b"message",
timestamp_start=time.time(),
timestamp_end=time.time(),
)
default.update(kwargs)
return http.Response(**default)
|
