#!/usr/bin/env python3 import argparse import http.server import socketserver import os import logging # setting up logging for this script _level = logging.INFO _format = "%(asctime)-15s [%(levelname)-8s] %(lineno)d : %(funcName)s : %(message)s" logging.basicConfig(format=_format, level=_level) log = logging.getLogger() # set up web directory WEB_DIR = '/tmp/web' class SimpleHandler(http.server.BaseHTTPRequestHandler): def do_GET(self): log.info(f"server version: {self.server_version}") log.info(f"client address: {self.client_address}") self.send_response(200) self.send_header('Content-type', 'text/html') self.end_headers() self.wfile.write(b"
1.1.0
") return class SuperSimpleHandler(http.server.SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): super().__init__(*args, directory=WEB_DIR, **kwargs) def do_GET(self): log.info(f"server version: {self.server_version}") log.info(f"client address: {self.client_address}") self.send_response(200) self.send_header('Content-type', 'text/html') self.end_headers() self.wfile.write(b"1.1.0
") return class SockHandler(socketserver.BaseRequestHandler): def handle(self): self.data = self.request.recv(1024).strip() log.info(f"client address: {self.client_address}") log.info(f"received data:\n{self.data}") self.request.sendall(b"Hello, World!\n") def parse_args(): """ Parse args """ argp = argparse.ArgumentParser() argp.add_argument('-i', '--ip-address', default="", help="IP Address to bind the server") argp.add_argument('-p', '--port', type=int, default=8080, help="Port on which to listen for traffic") argp.add_argument('-d', '--debug', help="Run in debug mode") return argp.parse_args() def create_file(size=1073741824, web_dir=WEB_DIR): """ Create 1G file for 1G download test """ with open(web_dir, 'wb') as fp: fp.seek(size - 1) fp.write(b'\0') return def remove_file(web_dir=WEB_DIR): """ Remove the 1G file since we no longer need it """ os.remove(web_dir) return if __name__ == "__main__": args = parse_args() if args.debug: log.setLevel(logging.DEBUG) try: if not os.path.isdir(WEB_DIR): os.mkdir(WEB_DIR) log.info("Creating 1 gigabyte file...") create_file(web_dir=os.path.join(WEB_DIR, 'speedtest')) simple_handler = SuperSimpleHandler() # http.server.SimpleHTTPRequestHandler with socketserver.TCPServer((args.ip_address, args.port), SockHandler) as tcpd: log.info(f"Server started on {args.ip_address}:{args.port}") tcpd.serve_forever() except KeyboardInterrupt as control_c: log.info("Server stopped") finally: log.info("Deleting 1 gigabyte file...") remove_file(web_dir=os.path.join(WEB_DIR, 'speedtest'))