import os from http.server import BaseHTTPRequestHandler import socketserver import threading import pyqrcode import socket import struct import base64 import inspect from SuperBeam import RequestHandlers class SuperBeamServer(BaseHTTPRequestHandler): """SuperBeam Server Handle incoming requests to serve information to the client. Class is built on top of :mod:`http.server.BaseHTTPRequestHandler` as simple http server. During initialization, the RequestHandlers are registered as routable paths. Call :func:`serve_forever()` to start the HTTP server. The mainloop will serve a list of files given as argument to :func:`serve_forever()`. Args: debug (bool): Enable debug output. Defaults to `False`. *args: Variable length argument list. **kwargs: Arbitrary keyword arguments. Attributes: debug (bool): enable debug output. files (list): list of files served by the `SuperBeamServer`. """ def __init__(self, request, client_address, server, debug=False): self.debug = debug self.routes = {} for _, cls in inspect.getmembers(RequestHandlers, inspect.isclass): if hasattr(cls, 'URI'): for route in cls.URI: self.routes[route] = cls super(SuperBeamServer, self).__init__(request, client_address, server) def do_GET(self): """GET request handler The :class:`SuperBeamServer` supports specifically defined requests URIs. Each requested URI in handled by a different handler class. The classes are defined in :mod:`SuperBeam.RequestHandlers` and registered in the ``path`` dictionary of this method. The requested path is checked against the registered handlers and - if exists - forwarded to :func:`respond()`, where the request gets answered. """ if self.debug: print(threading.current_thread()) # TODO Add globbing if self.path in self.routes: # TODO startswith key check self.respond(self.routes[self.path]) def respond(self, handler): """Respond client with request with registered handler. Method is called by :func:`do_GET()`. Currently supported handlers are described in :mod:`SuperBeam.RequestHandlers`. Args: class: Handler class implemeting a static ``handle()`` method. """ if self.debug: print('='*30 + ' DEBUG ' + '='*30) print('CONNECTION:', self.connection) print('HEADERS :', self.headers) print('REQUEST :', self.request) print('REQUEST* :', self.requestline) print('PATH :', self.path) print('COMMAND :', self.command) print('HANDLER :', handler) print('='*30 + ' DEBUG ' + '='*30) handler.handle(self) def get_primary_ip() -> str: """Get primary IP address. Obtain the primary interface IP address. If an error occurs, the default loopback adddress '127.0.0.1' is returned. Returns: str: IP Address """ s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) try: # doesn't even have to be reachable s.connect(('10.255.255.255', 1)) IP = s.getsockname()[0] except Exception as exc: print(exc) IP = '127.0.0.1' finally: s.close() return IP def get_qrcode() -> tuple: """Generate QR Code to be scanned by the SuperBeam Client. Obtains the IP address of by calling :func:`get_primary_ip()` and creates the ``QRCode`` object with :mod:`pyqrcode`. Returns: tuple(`pyqrcode.QRCode`, str): QRCode object, IP address """ primary_ip = get_primary_ip() octets = [1] octets.extend([int(_) for _ in primary_ip.split('.')[::-1]]) url = 'http://superbe.am/q?' + base64.b64encode( struct.pack('BBBBB', *octets) ).decode('utf-8') return pyqrcode.create(url), primary_ip def show_qrcode(): """Display QR Code with link for SuperBeam Client. The QR code is generated in function :func:`get_qrcode()`. The QR code is rendered in the terminal window. """ qrcode, primary_ip = get_qrcode() print(qrcode.terminal()) print(f'SuperBeam Server: http:://{primary_ip}:8080/\n') def build_filelist(paths, recursive=True) -> list: """List of files or directories to be served to SuperBeam clients. Args: paths (list(str)): list of directories to serve. recursive (bool, optional): serve subdirectories of `paths`. Defaults to `True` Returns: list: list of files with fullpath to be served. """ filelist = [] for path in paths: if os.path.isdir(path) and recursive: for root, dirs, files in os.walk(path): for file_ in files: filelist.append(os.path.join(root, file_)) else: filelist.append(path) return filelist def serve_forever(files): """Start the SuperBeam server and serve `files`. Infinite main loop serving `files` via SuperBeam server. Consider calling :func:`build_filelist()` to get a list of files to serve. The server tries to obtain the primary IP address of the system and binds the server to TCP port 8080. Args: files (list(str)): a list of files with fullpath information to serve. """ Handler = SuperBeamServer Handler.files = files with socketserver.ThreadingTCPServer(('', 8080), Handler) as httpd: show_qrcode() httpd.serve_forever()