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
|
#!/usr/bin/env python3
# Copyright (c) 2022 Egor Tensin <Egor.Tensin@gmail.com>
# This file is part of the "void" project.
# For details, see https://github.com/egor-tensin/void.
# Distributed under the MIT License.
import argparse
from contextlib import contextmanager
import http.server
import os
import signal
import sys
from threading import Condition, Lock, Thread
import traceback
from app import Request, Response, Void
DEFAULT_PORT = 23666
EXITING = False
def script_dir():
return os.path.dirname(os.path.realpath(__file__))
def set_exiting():
global EXITING
with SignalHandler.LOCK:
EXITING = True
SignalHandler.CV.notify()
class SignalHandler:
LOCK = Lock()
CV = Condition(LOCK)
def __init__(self, httpd):
self.httpd = httpd
self.thread = Thread(target=self.run)
def __enter__(self):
self.thread.start()
return self
def __exit__(self, type, value, traceback):
self.thread.join()
def run(self):
with SignalHandler.CV:
while not EXITING:
SignalHandler.CV.wait()
self.httpd.shutdown()
def handle_sigterm(signum, frame):
print('\nSIGTERM received, exiting...')
set_exiting()
def handle_sigint():
print('\nKeyboard interrupt received, exiting...')
set_exiting()
@contextmanager
def setup_signal_handlers(httpd):
handler_thread = SignalHandler(httpd)
with handler_thread:
signal.signal(signal.SIGTERM, handle_sigterm)
yield
# TODO: cleanup signal handlers?
class RequestHandler(http.server.SimpleHTTPRequestHandler):
VOID = None
def address_string(self):
if 'x-forwarded-for' in self.headers:
return self.headers['x-forwarded-for'].split(',')[0].strip()
if 'x-real-ip' in self.headers:
return self.headers['x-real-ip']
return super().address_string()
def do_GET(self):
try:
request = Request.from_http_path(self.path)
except ValueError:
return super().do_GET()
try:
response = request.process(RequestHandler.VOID)
response.write_to_request_handler(self)
except:
status = http.server.HTTPStatus.INTERNAL_SERVER_ERROR
response = Response(traceback.format_exc(), status)
response.write_to_request_handler(self)
return
def make_server(port):
addr = ('', port)
server = http.server.HTTPServer
if sys.version_info >= (3, 7):
server = http.server.ThreadingHTTPServer
return server(addr, RequestHandler)
def parse_args(args=None):
if args is None:
args = sys.argv[1:]
parser = argparse.ArgumentParser()
parser.add_argument('-p', '--port', metavar='PORT',
type=int, default=DEFAULT_PORT,
help='set port number')
parser.add_argument('-v', '--void', metavar='PATH', dest='backup',
help='set path to the void')
return parser.parse_args(args)
def main(args=None):
# It's a failsafe; this script is only allowed to serve the directory it
# resides in.
os.chdir(script_dir())
args = parse_args(args)
with Void(args.backup) as void:
RequestHandler.VOID = void
httpd = make_server(args.port)
with setup_signal_handlers(httpd):
try:
httpd.serve_forever()
except KeyboardInterrupt:
handle_sigint()
if __name__ == '__main__':
main()
|
