test/http-server

   1 #!/usr/bin/env python3
   2
   3 # This is a simple HTTP server based on the HTTPServer and
   4 # SimpleHTTPRequestHandler. It has been extended with PUT
   5 # and DELETE functionality to store or delete results.
   6 #
   7 # See: https://github.com/python/cpython/blob/main/Lib/http/server.py
   8
   9 from functools import partial
  10 from http import HTTPStatus
  11 from http.server import HTTPServer, SimpleHTTPRequestHandler
  12 import os
  13 import signal
  14 import socket
  15 import sys
  16
  17
  18 class AuthenticationError(Exception):
  19     pass
  20
  21
  22 class PUTEnabledHTTPRequestHandler(SimpleHTTPRequestHandler):
  23     def __init__(self, *args, basic_auth=None, **kwargs):
  24         self.basic_auth = None
  25         if basic_auth:
  26             import base64
  27
  28             self.basic_auth = base64.b64encode(
  29                 basic_auth.encode("ascii")
  30             ).decode("ascii")
  31         super().__init__(*args, **kwargs)
  32
  33     def do_GET(self):
  34         try:
  35             self._handle_auth()
  36             super().do_GET()
  37         except AuthenticationError:
  38             self.send_error(HTTPStatus.UNAUTHORIZED, "Need Authentication")
  39
  40     def do_HEAD(self):
  41         try:
  42             self._handle_auth()
  43             super().do_HEAD()
  44         except AuthenticationError:
  45             self.send_error(HTTPStatus.UNAUTHORIZED, "Need Authentication")
  46
  47     def do_PUT(self):
  48         path = self.translate_path(self.path)
  49         os.makedirs(os.path.dirname(path), exist_ok=True)
  50         try:
  51             self._handle_auth()
  52             file_length = int(self.headers["Content-Length"])
  53             with open(path, "wb") as output_file:
  54                 output_file.write(self.rfile.read(file_length))
  55             self.send_response(HTTPStatus.CREATED)
  56             self.send_header("Content-Length", "0")
  57             self.end_headers()
  58         except AuthenticationError:
  59             self.send_error(HTTPStatus.UNAUTHORIZED, "Need Authentication")
  60         except OSError:
  61             self.send_error(
  62                 HTTPStatus.INTERNAL_SERVER_ERROR, "Cannot open file for writing"
  63             )
  64
  65     def do_DELETE(self):
  66         path = self.translate_path(self.path)
  67         try:
  68             self._handle_auth()
  69             os.remove(path)
  70             self.send_response(HTTPStatus.OK)
  71             self.send_header("Content-Length", "0")
  72             self.end_headers()
  73         except AuthenticationError:
  74             self.send_error(HTTPStatus.UNAUTHORIZED, "Need Authentication")
  75         except OSError:
  76             self.send_error(
  77                 HTTPStatus.INTERNAL_SERVER_ERROR, "Cannot delete file"
  78             )
  79
  80     def _handle_auth(self):
  81         if not self.basic_auth:
  82             return
  83         authorization = self.headers.get("authorization")
  84         if authorization:
  85             authorization = authorization.split()
  86             if len(authorization) == 2:
  87                 if (
  88                     authorization[0] == "Basic"
  89                     and authorization[1] == self.basic_auth
  90                 ):
  91                     return
  92         raise AuthenticationError("Authentication required")
  93
  94
  95 def _get_best_family(*address):
  96     infos = socket.getaddrinfo(
  97         *address,
  98         type=socket.SOCK_STREAM,
  99         flags=socket.AI_PASSIVE,
 100     )
 101     family, type, proto, canonname, sockaddr = next(iter(infos))
 102     return family, sockaddr
 103
 104
 105 def run(HandlerClass, ServerClass, port, bind):
 106     HandlerClass.protocol_version = "HTTP/1.1"
 107     ServerClass.address_family, addr = _get_best_family(bind, port)
 108
 109     with ServerClass(addr, HandlerClass) as httpd:
 110         host, port = httpd.socket.getsockname()[:2]
 111         url_host = f"[{host}]" if ":" in host else host
 112         print(
 113             f"Serving HTTP on {host} port {port} "
 114             f"(http://{url_host}:{port}/) ..."
 115         )
 116         try:
 117             httpd.serve_forever()
 118         except KeyboardInterrupt:
 119             print("\nKeyboard interrupt received, exiting.")
 120             sys.exit(0)
 121
 122
 123 def on_terminate(signum, frame):
 124     sys.stdout.flush()
 125     sys.stderr.flush()
 126     sys.exit(128 + signum)
 127
 128
 129 if __name__ == "__main__":
 130     import argparse
 131
 132     parser = argparse.ArgumentParser()
 133     parser.add_argument(
 134         "--basic-auth", "-B", help="Basic auth tuple like user:pass"
 135     )
 136     parser.add_argument(
 137         "--bind",
 138         "-b",
 139         metavar="ADDRESS",
 140         help="Specify alternate bind address " "[default: all interfaces]",
 141     )
 142     parser.add_argument(
 143         "--directory",
 144         "-d",
 145         default=os.getcwd(),
 146         help="Specify alternative directory " "[default:current directory]",
 147     )
 148     parser.add_argument(
 149         "port",
 150         action="store",
 151         default=8080,
 152         type=int,
 153         nargs="?",
 154         help="Specify alternate port [default: 8080]",
 155     )
 156     args = parser.parse_args()
 157
 158     handler_class = partial(
 159         PUTEnabledHTTPRequestHandler, basic_auth=args.basic_auth
 160     )
 161
 162     os.chdir(args.directory)
 163
 164     signal.signal(signal.SIGINT, on_terminate)
 165     signal.signal(signal.SIGTERM, on_terminate)
 166
 167     run(
 168         HandlerClass=handler_class,
 169         ServerClass=HTTPServer,
 170         port=args.port,
 171         bind=args.bind,
 172     )