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
|
#!/usr/bin/python3
# Tiny HTTP Proxy. Based on the work of SUZUKI Hisao.
#
# Ported to py3 and modified to remove the bits we don't need
# and modernized.
import os
import http.server
import select
import socket
import socketserver
import sys
import urllib.parse
import argparse
import time
class ProxyHandler(http.server.BaseHTTPRequestHandler):
server_version = "testsproxy/1.0"
block_all = False
stall_for_seconds = 0
def log_request(self, m=""):
super().log_request(m)
sys.stdout.flush()
sys.stderr.flush()
def handle(self):
(ip, port) = self.client_address
super().handle()
def _connect_to(self, netloc, soc):
i = netloc.find(":")
if i >= 0:
host_port = netloc[:i], int(netloc[i + 1 :])
else:
host_port = netloc, 80
try:
soc.connect(host_port)
except socket.error as arg:
try:
msg = arg[1]
except:
msg = arg
self.send_error(404, msg)
return False
return True
def do_CONNECT(self):
if self.block_all:
self.log_error("blocking all traffic")
self.send_error(400, "All traffic blocked")
return
if self.stall_for_seconds > 0:
self.log_message("stalling for %d seconds", self.stall_for_seconds)
time.sleep(self.stall_for_seconds)
soc = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
if self._connect_to(self.path, soc):
self.log_request(200)
s = self.protocol_version + " 200 Connection established\r\n"
self.wfile.write(s.encode())
s = "Proxy-agent: {}\r\n".format(self.version_string())
self.wfile.write(s.encode())
self.wfile.write("\r\n".encode())
self._read_write(soc, 300)
finally:
soc.close()
self.connection.close()
def do_GET(self):
(scm, netloc, path, params, query, fragment) = urllib.parse.urlparse(
self.path, "http"
)
if scm != "http" or fragment or not netloc:
s = "bad url {}".format(self.path)
self.send_error(400, s.encode())
return
soc = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
if self._connect_to(netloc, soc):
self.log_request()
s = "{} {} {}\r\n".format(
self.command,
urllib.parse.urlunparse(("", "", path, params, query, "")),
self.request_version,
)
soc.send(s.encode())
self.headers["Connection"] = "close"
del self.headers["Proxy-Connection"]
for key, val in self.headers.items():
s = "{}: {}\r\n".format(key, val)
soc.send(s.encode())
soc.send("\r\n".encode())
self._read_write(soc)
finally:
soc.close()
self.connection.close()
def _read_write(self, soc, max_idling=20):
iw = [self.connection, soc]
ow = []
count = 0
while True:
count += 1
(ins, _, exs) = select.select(iw, ow, iw, 3)
if exs:
break
if ins:
for i in ins:
if i is soc:
out = self.connection
else:
out = soc
data = i.recv(8192)
if data:
out.send(data)
count = 0
if count == max_idling:
break
do_HEAD = do_GET
do_POST = do_GET
do_PUT = do_GET
do_DELETE = do_GET
def maybe_sd_notify(s: str) -> None:
addr = os.getenv("NOTIFY_SOCKET")
if not addr:
return
soc = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
soc.connect(addr)
soc.sendall(s.encode())
class ThreadingHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
def __init__(self, *args):
super().__init__(*args)
maybe_sd_notify("READY=1")
def parse_arguments():
parser = argparse.ArgumentParser(description="tiny proxy")
parser.add_argument("--block-all", help="block all requests", action="store_true")
parser.add_argument(
"--stall-for-seconds", help="block all requests", type=int, default=0
)
return parser.parse_args()
if __name__ == "__main__":
opts = parse_arguments()
port = 3128
if opts.block_all:
print("blocking all traffic")
ProxyHandler.block_all = True
if opts.stall_for_seconds > 0:
print("stalling for", opts.stall_for_seconds, "seconds")
ProxyHandler.stall_for_seconds = opts.stall_for_seconds
print("starting tinyproxy on port {}".format(port))
http.server.test(ProxyHandler, ThreadingHTTPServer, port=port)
|
