Skip to content

Instantly share code, notes, and snippets.

@santiagobasulto
Created April 4, 2021 13:21
Show Gist options
  • Star 1 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save santiagobasulto/16e5ee311af735ddcd43235dcda4c46b to your computer and use it in GitHub Desktop.
Save santiagobasulto/16e5ee311af735ddcd43235dcda4c46b to your computer and use it in GitHub Desktop.
A Python 3 debugging server based on http.server that logs every request. Use it to inspect incoming connections.
import time
from functools import partialmethod
from http.server import BaseHTTPRequestHandler, HTTPServer
import utils
class Request:
def __init__(self, method, path, headers, stream):
self.path = path
self.method = method.upper()
self.headers = headers
self.content_length = int(self.headers.get('Content-Length') or 0)
self.content_type = self.headers.get('Content-Type')
self.encoding = utils.get_encoding_from_headers(self.content_type)
self._body = stream.read(self.content_length)
@property
def body(self):
encoding = self.encoding or 'utf-8'
return self._body.decode(encoding)
class DebugRequestHandler(BaseHTTPRequestHandler):
def _set_response(self):
self.send_response(200)
self.send_header('Content-type', 'text/html')
self.end_headers()
def do_X(self, method):
request = Request(method, self.path, self.headers, self.rfile)
self._set_response()
print(f"{request.method} - {request.path} received")
if request.body:
print("Body follows:")
print(request.body)
self.wfile.write(f"{request.method} - {request.path} received".encode('utf-8'))
do_GET = partialmethod(do_X, 'GET')
do_POST = partialmethod(do_X, 'POST')
do_PUT = partialmethod(do_X, 'PUT')
do_PATCH = partialmethod(do_X, 'PATCH')
do_DELETE = partialmethod(do_X, 'DELETE')
do_HEAD = partialmethod(do_X, 'HEAD')
do_OPTIONS = partialmethod(do_X, 'OPTIONS')
class DebugServer(HTTPServer):
DEFAULT_HANDLER = DebugRequestHandler
def __init__(self, address='', port=5555, handler=None, response=200, log_to_console=True, log_to_disk=False, disk_log_location=None):
super().__init__((address, port), handler or self.DEFAULT_HANDLER)
self.debug_response = response
self.debug_log_to_console = log_to_console
self.debug_log_to_disk = log_to_disk
if __name__ == '__main__':
httpd = DebugServer()
try:
print('Serving...')
httpd.serve_forever()
except KeyboardInterrupt:
pass
httpd.server_close()
"""
This code was stolen from Requests:
https://github.com/psf/requests/blob/2c2138e811487b13020eb331482fb991fd399d4e/requests/utils.py
"""
def _parse_content_type_header(content_type):
"""Returns content type and parameters from given header
:param header: string
:return: tuple containing content type and dictionary of
parameters
"""
tokens = content_type.split(';')
content_type, params = tokens[0].strip(), tokens[1:]
params_dict = {}
items_to_strip = "\"' "
for param in params:
param = param.strip()
if param:
key, value = param, True
index_of_equals = param.find("=")
if index_of_equals != -1:
key = param[:index_of_equals].strip(items_to_strip)
value = param[index_of_equals + 1:].strip(items_to_strip)
params_dict[key.lower()] = value
return content_type, params_dict
def get_encoding_from_headers(content_type):
"""Returns encodings from given HTTP Header Dict.
:param headers: dictionary to extract encoding from.
:rtype: str
"""
if not content_type:
return None
content_type, params = _parse_content_type_header(content_type)
if 'charset' in params:
return params['charset'].strip("'\"")
if 'text' in content_type:
return 'ISO-8859-1'
if 'application/json' in content_type:
# Assume UTF-8 based on RFC 4627: https://www.ietf.org/rfc/rfc4627.txt since the charset was unset
return 'utf-8'
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment