Simple Python 3 HTTP server for logging all GET and POST requests

server.py

#!/usr/bin/env python3
"""
License: MIT License
Copyright (c) 2023 Miel Donkers

Very simple HTTP server in python for logging requests
Usage::
    ./server.py [<port>]
"""
from http.server import BaseHTTPRequestHandler, HTTPServer
import logging

class S(BaseHTTPRequestHandler):
    def _set_response(self):
        self.send_response(200)
        self.send_header('Content-type', 'text/html')
        self.end_headers()

    def do_GET(self):
        logging.info("GET request,\nPath: %s\nHeaders:\n%s\n", str(self.path), str(self.headers))
        self._set_response()
        self.wfile.write("GET request for {}".format(self.path).encode('utf-8'))

    def do_POST(self):
        content_length = int(self.headers['Content-Length']) # <--- Gets the size of data
        post_data = self.rfile.read(content_length) # <--- Gets the data itself
        logging.info("POST request,\nPath: %s\nHeaders:\n%s\n\nBody:\n%s\n",
                str(self.path), str(self.headers), post_data.decode('utf-8'))

        self._set_response()
        self.wfile.write("POST request for {}".format(self.path).encode('utf-8'))

def run(server_class=HTTPServer, handler_class=S, port=8080):
    logging.basicConfig(level=logging.INFO)
    server_address = ('', port)
    httpd = server_class(server_address, handler_class)
    logging.info('Starting httpd...\n')
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        pass
    httpd.server_close()
    logging.info('Stopping httpd...\n')

if __name__ == '__main__':
    from sys import argv

    if len(argv) == 2:
        run(port=int(argv[1]))
    else:
        run()

What about do_DELETE

#! /usr/bin/env python3

#The MIT License (MIT)
#
#Copyright <2022> <Alexandre Maia Godoi>
#
#Permission is hereby granted, free of charge, to any person obtaining a copy of
#this software and associated documentation files (the “Software”), to deal in
#the Software without restriction, including without limitation the rights to
#use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
#of the Software, and to permit persons to whom the Software is furnished to do so.

#THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
#IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
#FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
#AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
#LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
#FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
#DEALINGS IN THE SOFTWARE.


from http.server import BaseHTTPRequestHandler, HTTPServer
import logging
import sys

COLOR = "\033[1;32m"
RESET_COLOR = "\033[00m"

class S(BaseHTTPRequestHandler):
    def _set_response(self):
        self.send_response(200)
        self.send_header('Content-type', 'text/html')
        self.end_headers()

    def do_log(self, method):
        content_length = self.headers['Content-Length']
        content_length = 0 if (content_length is None) else int(content_length)
        post_data = self.rfile.read(content_length)
        logging.info(COLOR + method + " request,\n" + RESET_COLOR + "Path: %s\nHeaders:\n%sBody:\n%s\n",
                str(self.path), str(self.headers), post_data.decode('utf-8'))
        self._set_response()
        self.wfile.write((method + " request for {}".format(self.path)).encode('utf-8'))

    def do_GET(self):
        self.do_log("GET")

    def do_POST(self):
        self.do_log("POST")

    def do_PUT(self):
        self.do_log("PUT")

    def do_DELETE(self):
        self.do_log("DELETE")

def run(address, port, server_class=HTTPServer, handler_class=S):
    logging.basicConfig(level=logging.INFO)
    server_address = (address, port)
    httpd = server_class(server_address, handler_class)
    logging.info('Starting httpd...\n')
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        pass
    httpd.server_close()
    logging.info('Stopping httpd...\n')

if __name__ == '__main__':
    if len(sys.argv) != 3:
        print("Usage:\n" + sys.argv[0] + " [address] [port]")
        sys.exit(1)

    run(sys.argv[1], int(sys.argv[2]))

I am trying to migrate the same code to AWS Lambda and need guidance, any help would be greatly appreciated.

@mdonkers @amgodoi would you have an objection to set a FOSS license for your gist's code?

— to give more context, I'm assoc. professor in CS and if possible, I'd like to set a small assignment, reusing your PoC 😊

no objection @erikmd

@mdonkers @amgodoi would you have an objection to set a FOSS license for your gist's code?

— to give more context, I'm assoc. professor in CS and if possible, I'd like to set a small assignment, reusing your PoC blush

@erikmd I wonder to what extend such simple code is licensable in the first place. But no objections. Hereby granting you (and everyone else) all the freedom to do with the code however you like.

OK thanks for your reply!

Awesome, thanks! Used in gist 36583dde8e0eb8e97e2cff2e7d9d2836

port

What host and port are you trying to run it on?

error when receiving the request result: headers responsible for the end of the data were not found

Thank you! This can be easily extended to handle https requests by adding
httpd.socket = ssl.wrap_socket(httpd.socket, server_side=True, keyfile="key.pem", certfile='cert.pem')
after line 36. However, ssl.wrap_socket() is deprecated. Does anyone know how to do this with SSLContext.wrap_socket()?

I tried with

context = ssl.create_default_context()
context.load_cert_chain(certfile='cert.pem', keyfile="key.pem")

but then context.wrap_socket(httpd.socket, server_side=True) complains about "check_hostname requires server_hostname". Adding a hostname, it complains that hostnames can only be used in client mode.

@maiphi any luck in getting the https to work ?

Seeing your question, I just looked into it again. This should do the job (note that I did only a very quick test):

context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
context.load_cert_chain('cert.pem', 'key.pem')
httpd.socket = context.wrap_socket(httpd.socket, server_side=True)

Note that except for the deprecation warning the version using ssl.wrap_socket() works fine, too.

wonderful! many, many, many thanks 👌