simple http server for testing kong api gw

Dockerfile

FROM python:3.9.5-slim-buster

ENV PYTHONUNBUFFERED 1

RUN adduser \
    --disabled-password \
    --gecos "" \
    --home /app \
    app

USER app

WORKDIR /app

COPY ./main.py /app/main.py

ENTRYPOINT ["python", "main.py"]

main.py

"""
Very simple HTTP server in python for logging requests and testing API GW
based on mdonkers' gist.
Usage::
    ./server.py [<port>]
    Run with kong on same network:
       docker run --network=kong-ee-net --name=minirepo BUILT-IMAGE-ID
"""
import logging
import json
from http.server import BaseHTTPRequestHandler, HTTPServer

logger = logging.getLogger("main")
metadata = {
    "version": "v1",
    "licenses": [],
}


class S(BaseHTTPRequestHandler):
    def _set_response(self, m: bytes = None):
        self.send_response(200)
        self.send_header('Content-type', 'application/json')
        self.send_header('Access-Control-Allow-Credentials', 'true')
        self.send_header('Access-Control-Allow-Origin', '*')
        self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS')
        self.send_header("Access-Control-Allow-Headers", "X-Requested-With, Content-type")
        if m:
            self.send_header('Content-Length', str(len(m)))
        self.end_headers()

        if m:
            self.wfile.write(m)

    def do_GET(self):
        logging.info("GET request,\nPath: %s\nHeaders:\n%s\n", str(self.path), str(self.headers))
        self._set_response(json.dumps(metadata).encode('utf-8'))

    def do_DELETE(self):
        logging.info("DELETE request,\nPath: %s\nHeaders:\n%s\n", str(self.path), str(self.headers))
        lic_to_delete = self.path.replace("/", "")
        metadata["licenses"] = [lic for lic in metadata["licenses"] if lic != lic_to_delete]
        self._set_response(json.dumps(metadata).encode('utf-8'))

    def do_OPTIONS(self):
        logging.info("OPTIONS request")
        self.send_response(200, "ok")
        self.send_header('Access-Control-Allow-Credentials', 'true')
        self.send_header('Access-Control-Allow-Origin', '*')
        self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS')
        self.send_header("Access-Control-Allow-Headers", "X-Requested-With, Content-type")

    def do_POST(self):
        content_length = int(self.headers['Content-Length']) if self.headers['Content-Length'] else 0
        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'))
        d = json.loads(post_data)
        if "license" in d:
            metadata["licenses"].append(d["license"])

        if "version" in d:
            metadata["version"] = d["version"]

        self._set_response()
        self.wfile.write(json.dumps(metadata).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()