Small script to answer any query and dump headers/body of requests

interceptor.py

#!/bin/env python3

import argparse
import http.server
import logging
import socketserver


class RequestHandler(http.server.BaseHTTPRequestHandler):
    def _set_response(self):
        self.send_response(200)
        self.send_header("Content-type", "text/plain")
        self.end_headers()

    def _log_request(self):
        content_length = int(self.headers.get("Content-Length", 0))
        body = self.rfile.read(content_length) if content_length > 0 else b""

        logging.info("========== New Request ==========")
        logging.info(f"Client: {self.client_address[0]}")
        logging.info(f"Method: {self.command}")
        logging.info(f"Path: {self.path}")
        logging.info("Headers:")
        for key, value in self.headers.items():
            logging.info(f"  {key}: {value}")
        if body:
            logging.info(f"Payload: {body.decode('utf-8', errors='replace')}")
        else:
            logging.info("Payload: (empty)")
        logging.info("=================================")

    def do_GET(self):
        self._log_request()
        self._set_response()
        self.wfile.write(b"GET received")

    def do_POST(self):
        self._log_request()
        self._set_response()
        self.wfile.write(b"POST received")

    def do_PUT(self):
        self._log_request()
        self._set_response()
        self.wfile.write(b"PUT received")


def main():
    parser = argparse.ArgumentParser(
        description="Simple HTTP server for logging requests"
    )
    parser.add_argument(
        "-p", "--port", help="Port to listen on", default=8282, type=int
    )
    parser.add_argument(
        "-l", "--listen", help="Address to listen on", default="0.0.0.0"
    )
    args = parser.parse_args()

    logging.basicConfig(level=logging.INFO, format="[%(asctime)s] %(message)s")

    with socketserver.TCPServer(
        (args.listen, args.port), RequestHandler
    ) as httpd:
        logging.info(f"Server started at port {args.port}")
        try:
            httpd.serve_forever()
        except KeyboardInterrupt:
            logging.info("Shutting down server.")
            httpd.server_close()


if __name__ == "__main__":
    main()