Brief explanation of WSGI middleware

wsgi_middleware.py

"""
A brief explanation of WSGI middleware.
Nabla Zero Labs
"""

# The main WSGI application returns the string "Hello, {ip_address}!", where
# `ip_address` is the IP address of the host making the request.

def application(environ, start_response):
    print("Main application")
    ip_address = environ.get("REMOTE_ADDR", "<unknown>")
    body = f"Hello, {ip_address}!".encode("utf-8")
    headers = [("content-type", "text/plain; charset=utf-8"),
               ("content-length", f"{len(body)}")]
    start_response("200 OK", headers)
    return [body]

# A middleware can sit "in front" of the application, meaning that it is called
# before the application. For example, the following middleware will reject any
# requests that do not provide an AUTHORIZATION header (for simplicity's sake,
# any non-nil value will be accepted, but in a real deployment you would use an
# actual authorization scheme).

def authentication_middleware(app):
    def wrapped(environ, start_response):
        print("Authentication middleware")
        if environ.get("HTTP_AUTHORIZATION", None):
            return app(environ, start_response)
        else:
            body = "Please provide an AUTHORIZATION header.".encode("utf-8")
            headers = [("content-type", "text/plain; charset=utf-8"),
                       ("content-length", f"{len(body)}")]
            start_response("401 Unauthorized", headers)
            return [body]
    return wrapped

# A middleware can also sit "behind" the application, meaning that it is called
# after the application sends a response. For example, the following middleware
# will add a CONTENT-SECURITY-POLICY header to all responses.

def adding_csp_headers(start_response):
    def wrapped(status, headers):
        print("Adding CSP headers")
        headers.append(("content-security-policy", "default-src 'none'"))
        return start_response(status, headers)
    return wrapped

def csp_middleware(app):
    def wrapped(environ, start_response):
        print("CSP Middleware")
        return app(environ, adding_csp_headers(start_response))
    return wrapped

# This is how we would wrap our original application to use these two
# middlewares (the order of wrapping does matter; that's a feature).

wrapped_application = authentication_middleware(csp_middleware(application))

Example:

Request:

$ curl localhost:8000
Please provide an AUTHORIZATION header.

Logs:

Authentication middleware

Request:

$ curl -v -H "Authorization: " localhost:8000
> GET / HTTP/1.1
> Host: localhost:8000
> User-Agent: curl/7.54.0
> Accept: */*
> Authorization: x
> 
< HTTP/1.1 200 OK
< Server: gunicorn/19.7.1
< Date: Fri, 02 Mar 2018 20:56:50 GMT
< Connection: close
< content-type: text/plain; charset=utf-8
< content-length: 17
< content-security-policy: default-src 'none'
Hello, 127.0.0.1!

Logs:

Authentication middleware
CSP Middleware
Main application
Adding CSP headers

The WSGI architecture enables middleware to be used as decorator, which makes the whole thing very clean and pretty.

ENCODING = "utf-8"
def requires_authentication(app):
    def wrapped(environ, start_response):
        print("Authentication middleware")
        if environ.get("HTTP_AUTHORIZATION", None):
            return app(environ, start_response)
        else:
            body = "Please provide an AUTHORIZATION header.".encode(ENCODING)
            headers = [("content-type", f"text/plain; charset={ENCODING}"),
                       ("content-length", f"{len(body)}")]
            start_response("401 Unauthorized", headers)
            return [body]
    return wrapped

@requires_authentication
def application(environ, start_response):
    print("Main application")
    ip_address = environ.get("REMOTE_ADDR", "<unknown>")
    body = f"Hello, {ip_address}!".encode(ENCODING)
    headers = [("content-type", f"text/plain; charset={ENCODING}"),
               ("content-length", f"{len(body)}")]
    start_response("200 OK", headers)
    return [body]

A decorator enforcing a very obvious content security policy

ENCODING = "utf-8"

class CSPWriter:
    def __init__(self, app, policy):
        self._app    = app
        self._policy = ("content-security-policy", policy)
    def __call__(self, environ, start_response):
        def secured(status, headers):
            headers.append(self._policy)
            return start_response(status, headers)
        return self._app(environ, secured)

def CSP(*args):
    policy = "; ".join(args)
    def wrapped(app):
        return CSPWriter(app, policy)
    return wrapped

@CSP("default-src: 'none'", "script-src: 'none'")
def application(environ, start_response):
    ip_address = environ.get("REMOTE_ADDR", "<unknown>")
    body = f"Hello, {ip_address}!".encode(ENCODING)
    headers = [("content-type", f"text/plain; charset={ENCODING}"),
               ("content-length", f"{len(body)}")]
    start_response("200 OK", headers)
    return [body]