97 lines
3.6 KiB
Python
Executable File
97 lines
3.6 KiB
Python
Executable File
from http.server import ThreadingHTTPServer, SimpleHTTPRequestHandler
|
|
import json
|
|
import sys
|
|
import os
|
|
import socket
|
|
import rsgdnd.utils as utils
|
|
|
|
def _get_best_family(*address):
|
|
infos = socket.getaddrinfo(
|
|
*address,
|
|
type=socket.SOCK_STREAM,
|
|
flags=socket.AI_PASSIVE,
|
|
)
|
|
family, type, proto, canonname, sockaddr = next(iter(infos))
|
|
return family, sockaddr
|
|
|
|
|
|
def run_whitelist(protocol="HTTP/1.0", port=80, bind=None, get_callback=None, post_callback=None, whitelist=[]):
|
|
"""Test the HTTP request handler class.
|
|
|
|
This runs an HTTP server on port 80 (or the port argument).
|
|
|
|
"""
|
|
import contextlib
|
|
class WhiltelistHandler(SimpleHTTPRequestHandler):
|
|
whitelist = []
|
|
get_callback = None
|
|
post_callback = None
|
|
def do_GET(self):
|
|
# tried doing this transform in the __init__, wouldn't work for some reason
|
|
self.whitelist = [self.translate_path(x) for x in self.whitelist]
|
|
self.whitelist = [os.path.relpath(x, self.directory) for x in self.whitelist]
|
|
response = {}
|
|
in_path = self.translate_path(self.path)
|
|
in_path = os.path.relpath(in_path, self.directory)
|
|
if self.get_callback and self.get_callback(in_path, response):
|
|
if response:
|
|
self.send_response(200)
|
|
self.end_headers()
|
|
self.wfile.write(bytes(json.dumps(response), "utf-8"))
|
|
else:
|
|
SimpleHTTPRequestHandler.do_GET(self)
|
|
elif in_path in self.whitelist:
|
|
SimpleHTTPRequestHandler.do_GET(self)
|
|
else:
|
|
self.send_response(404)
|
|
self.end_headers()
|
|
pass
|
|
|
|
def do_POST(self):
|
|
response = {}
|
|
in_path = self.translate_path(self.path)
|
|
in_path = os.path.relpath(in_path, self.directory)
|
|
data_string = self.rfile.read(int(self.headers['Content-Length']))
|
|
parsed = data_string.decode("utf-8")
|
|
parsed = json.loads(parsed)
|
|
if self.post_callback and self.post_callback(in_path, parsed, response):
|
|
self.send_response(200)
|
|
self.end_headers()
|
|
self.wfile.write(bytes(json.dumps(response), "utf-8"))
|
|
else:
|
|
self.send_response(404)
|
|
self.end_headers()
|
|
|
|
|
|
|
|
HandlerClass = WhiltelistHandler
|
|
|
|
class DualStackServer(ThreadingHTTPServer):
|
|
def server_bind(self):
|
|
# suppress exception when protocol is IPv4
|
|
with contextlib.suppress(Exception):
|
|
self.socket.setsockopt(
|
|
socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0)
|
|
return super().server_bind()
|
|
|
|
ServerClass = DualStackServer
|
|
|
|
ServerClass.address_family, addr = _get_best_family(bind, port)
|
|
HandlerClass.protocol_version = protocol
|
|
HandlerClass.whitelist = whitelist
|
|
HandlerClass.get_callback = get_callback
|
|
HandlerClass.post_callback = post_callback
|
|
with ServerClass(addr, HandlerClass) as httpd:
|
|
host, port = httpd.socket.getsockname()[:2]
|
|
url_host = f'[{host}]' if ':' in host else host
|
|
utils.log(
|
|
f"Serving HTTP on {host} port {port} "
|
|
f"(http://{url_host}:{port}/) ..."
|
|
f"\nWhitelist: {str(whitelist)}"
|
|
)
|
|
try:
|
|
httpd.serve_forever()
|
|
except KeyboardInterrupt:
|
|
utils.log("\nKeyboard interrupt received, exiting.")
|
|
sys.exit(0)
|