dlmp-middleware 1.3.6

ASGI/WSGI middleware for Python web applications

webapp-middleware

[[TOC]]

Overview

This package currently implements the following middlwares for both FastAPI/ASGI and Flask/WSGI web application frameworks:

• AuthMiddleware: authentication middleware with support for ELB/Cognito, ALB/mTLS, and SAML/trusted header schemes, with optional support for enforcement of AUP acceptance.
• LogMiddleware: intercept-based logging configuration providing colorization/formatting, JSON serialization, and more
• TimingMiddleware: profiles timings for operations and optionally publishes metrics logs and/or Server-Timing header

Installation

This package is published to PyPI as dlmp-middleware. Install it into your virtualenv with the options for the middlewares you intend to use:

pip install "dlmp-middleware[auth,log]"

Available options:

• auth
• log
• timing

Note: the distribution name is dlmp-middleware, but the import name remains middleware (e.g. from middleware import AuthMiddleware).

It can also be installed directly from source (ensure SSH credentials for your Git host are configured):

pip install "dlmp-middleware[auth,log] @ git+ssh://<git-host>/<namespace>/webapp-middleware.git@<version>"

Middleware Packages

LogMiddleware

Log Format

Logs can be configured to be serialized to JSON (recommended if the destination is Cloudwatch Logs):

is_production = APP_ENV == "production"

LogMiddleware(json=is_production)

Exempt Paths

Specific paths can be exempted from logging, either by exact match or by path prefix followed by an asterisk:

log_exempt_paths = (
  "/health",
  "/static/*",
)

LogMiddleware(exempt=log_exempt_paths)

Usage Outside of Middleware

The intercept logger can be invoked outside of the context of middleware, useful in cases where an application contains non-web services and logging configuration should be normalized

Invoke configure_logging at application start:

from middleware import configure_logging

configure_logging(json=True)

In application code, use the logging module as usual:

import logging

logging.getLogger(__name__).warn("...")

TimingMiddleware

Defaults

In a default configuration, the FastAPI or Flask middleware publishes a timing configuration with the following keys:

• route: Total time taken to process all code for a given route (inclusive of more granular timings below)
• encode: Time taken for the web framework to encode the response
• render: Time taken for the web framework to render / produce the response

Configuring Application-specific Timings

To add application-specific timings, the timings argument can be used.

Below are some examples of common external data sources and APIs.

SQLAlchemy

import sqlalchemy

timings={
    "db_exec": (
        sqlalchemy.engine.base.Engine.execute,
    ),
    "db_fetch": (
        sqlalchemy.engine.ResultProxy.fetchone,
        sqlalchemy.engine.ResultProxy.fetchmany,
        sqlalchemy.engine.ResultProxy.fetchall,
    ),
}

AWS (boto3)

import botocore

timings={
    "aws": (
        botocore.client.BaseClient._make_api_call,
    ),
}

Redis (redis-py)

import redis

timings={
    "redis": (
        redis.connection.Connection.send_command,
        redis.connection.Connection.read_response,
    ),
}

AuthMiddleware

ELB / Cognito

In the default configuration with no middleware options provided, if the x-amzn-oidc-data HTTP header is present, Cognito authentication provided via ELB will automatically be used. An alternate header name can be optionally provided:

AuthMiddleware(
    elb_header_name="custom-http-header-name",  # HTTP header name containing Cognito JWT (default: x-amzn-oidc-data)
)

Mock User

For development/testing purposes, a mock user can be statically provided:

test_user = {
    "username": "testuser",
    "groups": ["group-1", "group-2"],
}

AuthMiddleware(mock_user=test_user)

API Gateway / mTLS Authentication

User context can be read from connections proxied through API Gateway with mTLS authentication configured.

Note: If mTLS authentication is configured and an HTTP connection is made with the relevant header present but a user session cookie is not present, an exception will NOT be raised, access will be granted, and the user context will be left unset. The trusted TLS header asserted by API Gateway only attests that a connecting device bears a valid TLS certificate. Applications implementing separate device + user level authentication in this manner are required to assert the presence of a user session in application logic accordingly.

AuthMiddleware(
    tls_cookie_name="my-app-user-cookie",     # HTTP cookie name bearing user session credentials
    tls_cookie_secret="user-cookie-secret",   # signing secret for TLS cookie
    tls_header_name="x-amzn-mtls-identity",   # HTTP header name to use for detection of TLS authentication
)

SAML / Trusted Proxy

For applications deployed behind a trusted authenticating proxy, an HTTP header can be specified

AuthMiddleware(
    saml_user_header_name="x-saml-user",      # HTTP header name containing SAML username
    saml_groups_header_name="x-saml-groups",  # HTTP header name containing SAML groups
)

Exempt Paths

Specific HTTP paths can be exempted from requiring authentication, either by exact match or by HTTP path prefix followed by an asterisk. Note that user context will not be present on the request context for exempted routes:

auth_exempt_paths = (
  "/health",
  "/version",
  "/public/*",
)

AuthMiddleware(exempt=auth_exempt_paths)

AUP Enforcement

An aup argument can be provided to AuthMiddleware which verifies that users have accepted an acceptable use policy prior to allowing access to the application. It detects the presence of a JSON web token (JWT) signed by service at url with HMAC passphrase secret which attests that a user has read and signed the AUP document identified by aup_id. If the cookie does not exist or the signature is invalid, the user is redirected to the AUP service to obtain a valid cookie.

The debug and exempt options configured for AuthMiddleware also apply to AUP enforcement.

The JWT payload is a JSON object with the following values:

• user: username as authenticated by AuthMiddleware; must match the current authenticated user (required)
• signatures: an object with required AUP ID as a key and document revision as value (required)
• exp: expiration timestamp in epoch seconds (optional)

Configuration

AupService(
    aup_id="my-aup-v1",                     # ID associated with the application's policy (required)
    cookie_name="clickthrough_signatures",  # Cookie name (default: clickthrough_signatures)
    exempt=("/path", "/prefix/*"),          # HTTP paths or prefixes to exempt from AUP enforcement (default: None)
    mock_verified=False,                    # Whether to mock successful verification (default: false)
    path="/verify",                         # AUP service HTTP path for verification requests (default: /verify)
    secret="secret-key",                    # AUP service HMAC secret used to sign JWT (required)
    url="https://aup.example.com",          # AUP service base URL (required)
)

Mock Acceptance

AUP verification can be bypassed for development/testing purposes:

aup_service = AupService(mock_verified=True, …)
AuthMiddleware(aup=aup_service, …)

Examples

FastAPI

from fastapi import FastAPI

from middleware import AuthMiddleware

app = FastAPI()

auth_options = dict(
    aup=dict(
        aup_id="app-name-aup",
        secret="service-secret-key",
        url="https://aup.example.com",
    ),
    exempt=("/healthcheck", "/version"),
)

app.add_middleware(AuthMiddleware, **auth_options)

Flask

from flask import Flask

from middleware import AuthMiddleware

app = Flask(__name__)

auth_options = dict(
    aup=dict(
        aup_id="app-name-aup",
        secret="service-secret-key",
        url="https://aup.example.com",
    ),
    exempt=("/healthcheck", "/version"),
)

app.wsgi_app = AuthMiddleware(app.wsgi_app, **auth_options)