Middleware Components

from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
from uvicorn.middleware.asgi2 import ASGI2Middleware
from uvicorn.middleware.wsgi import WSGIMiddleware
from uvicorn.middleware.message_logger import MessageLoggerMiddleware

class ProxyHeadersMiddleware:
    """
    Middleware for handling proxy headers.

    Parses X-Forwarded-Proto and X-Forwarded-For headers from trusted
    proxies and updates the ASGI scope with the client's real IP address
    and protocol scheme.

    This is essential when running behind reverse proxies like Nginx,
    Apache, or cloud load balancers.
    """

    def __init__(
        self,
        app: ASGI3Application,
        trusted_hosts: list[str] | str = "127.0.0.1",
    ) -> None:
        """
        Initialize proxy headers middleware.

        Args:
            app: ASGI application to wrap
            trusted_hosts: Trusted proxy hosts/networks as comma-separated string or list
                          Supports:
                          - "*" to trust all proxies (not recommended in production)
                          - IP addresses: "127.0.0.1", "192.168.1.1"
                          - CIDR networks: "10.0.0.0/8", "172.16.0.0/12"
                          - Multiple values: "127.0.0.1,10.0.0.0/8"
                          Default: "127.0.0.1"

        The middleware only processes headers from trusted proxy addresses.
        Headers from untrusted sources are ignored for security.
        """

    async def __call__(
        self,
        scope: Scope,
        receive: ASGIReceiveCallable,
        send: ASGISendCallable,
    ) -> None:
        """
        Process ASGI connection with proxy header handling.

        Args:
            scope: ASGI connection scope
            receive: Receive callable for incoming messages
            send: Send callable for outgoing messages

        For HTTP and WebSocket scopes, this middleware:
        1. Checks if the connection is from a trusted proxy
        2. Parses X-Forwarded-Proto header to update scope["scheme"]
        3. Parses X-Forwarded-For header to update scope["client"]
        4. Calls the wrapped application with updated scope
        """

class ASGI2Middleware:
    """
    Adapter to run ASGI2 applications as ASGI3.

    ASGI2 uses a class-based interface where the application is instantiated
    with the scope and then called with receive/send. ASGI3 uses a single
    callable that accepts all three arguments.

    This middleware bridges the two interfaces, allowing ASGI2 applications
    to run on ASGI3 servers like uvicorn.
    """

    def __init__(self, app: ASGI2Application) -> None:
        """
        Initialize ASGI2 middleware.

        Args:
            app: ASGI2 application class (not instance)
                 The app should have __init__(scope) and async __call__(receive, send)
        """

    async def __call__(
        self,
        scope: Scope,
        receive: ASGIReceiveCallable,
        send: ASGISendCallable,
    ) -> None:
        """
        Run ASGI2 application as ASGI3.

        Args:
            scope: ASGI connection scope
            receive: Receive callable for incoming messages
            send: Send callable for outgoing messages

        This method:
        1. Instantiates the ASGI2 app with the scope
        2. Calls the instance with receive and send
        """

class WSGIMiddleware:
    """
    Adapter to run WSGI applications in ASGI.

    Bridges the synchronous WSGI interface with the asynchronous ASGI
    interface by running WSGI applications in a thread pool executor.

    Note: This is uvicorn's implementation. The a2wsgi package provides
    a more feature-complete WSGI adapter and is recommended for production use.
    """

    def __init__(self, app: WSGIApp, workers: int = 10) -> None:
        """
        Initialize WSGI middleware.

        Args:
            app: WSGI application callable
                 Should have signature: app(environ, start_response)
            workers: Number of worker threads for handling WSGI calls (default: 10)
                    Each WSGI request runs in a thread from this pool since
                    WSGI is synchronous while ASGI is asynchronous.
        """

    async def __call__(
        self,
        scope: Scope,
        receive: ASGIReceiveCallable,
        send: ASGISendCallable,
    ) -> None:
        """
        Run WSGI application in ASGI context.

        Args:
            scope: ASGI connection scope (only HTTP supported)
            receive: Receive callable for incoming messages
            send: Send callable for outgoing messages

        This method:
        1. Receives the HTTP request body
        2. Builds WSGI environ from ASGI scope
        3. Runs WSGI app in thread pool
        4. Sends WSGI response back through ASGI send
        """

class MessageLoggerMiddleware:
    """
    Middleware that logs all ASGI messages at TRACE level.

    Useful for debugging ASGI applications by showing the exact message
    flow between server and application. Messages with large bodies are
    logged with size placeholders instead of full content.
    """

    def __init__(self, app: ASGI3Application) -> None:
        """
        Initialize message logger middleware.

        Args:
            app: ASGI application to wrap

        Attributes:
            task_counter: Counter for assigning unique task IDs
            app: Wrapped ASGI application
            logger: Logger instance for message logging
        """

    async def __call__(
        self,
        scope: Scope,
        receive: ASGIReceiveCallable,
        send: ASGISendCallable,
    ) -> None:
        """
        Run application with message logging.

        Args:
            scope: ASGI connection scope
            receive: Receive callable for incoming messages
            send: Send callable for outgoing messages

        This method wraps receive and send callables to log all messages
        at TRACE level. Each connection is assigned a unique task ID for
        tracking messages across the connection lifecycle.
        """

# WSGI environ dictionary
Environ = MutableMapping[str, Any]

# WSGI exception info tuple
ExcInfo = tuple[type[BaseException], BaseException, Optional[types.TracebackType]]

# WSGI start_response callable
StartResponse = Callable[[str, Iterable[tuple[str, str]], Optional[ExcInfo]], None]

# WSGI application callable
WSGIApp = Callable[[Environ, StartResponse], Union[Iterable[bytes], BaseException]]

def build_environ(
    scope: HTTPScope,
    message: ASGIReceiveEvent,
    body: io.BytesIO,
) -> Environ:
    """
    Build WSGI environ dictionary from ASGI scope.

    Args:
        scope: ASGI HTTP scope
        message: ASGI receive event
        body: Request body as BytesIO

    Returns:
        WSGI environ dictionary with all required and optional keys

    The environ includes:
    - CGI variables (REQUEST_METHOD, PATH_INFO, QUERY_STRING, etc.)
    - Server variables (SERVER_NAME, SERVER_PORT, SERVER_PROTOCOL)
    - WSGI variables (wsgi.version, wsgi.url_scheme, wsgi.input, wsgi.errors)
    - HTTP headers (converted from ASGI format)
    """

from uvicorn import run
from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware

async def app(scope, receive, send):
    # Your ASGI application
    ...

# Wrap with proxy headers middleware
app_with_proxy = ProxyHeadersMiddleware(
    app,
    trusted_hosts="127.0.0.1,10.0.0.0/8",
)

# Run with middleware
run(app_with_proxy, host="0.0.0.0", port=8000)

import uvicorn

# Proxy headers middleware is automatically applied when proxy_headers=True
uvicorn.run(
    "myapp:app",
    host="0.0.0.0",
    port=8000,
    proxy_headers=True,  # Enable proxy headers (default: True)
    forwarded_allow_ips="127.0.0.1,192.168.0.0/16",  # Trusted proxies
)

from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware

# Trust all proxies - NOT RECOMMENDED IN PRODUCTION
app = ProxyHeadersMiddleware(
    app,
    trusted_hosts="*",  # Trust all sources
)

from uvicorn.middleware.asgi2 import ASGI2Middleware

# ASGI2 application (class-based)
class MyASGI2App:
    def __init__(self, scope):
        self.scope = scope

    async def __call__(self, receive, send):
        assert self.scope['type'] == 'http'
        await send({
            'type': 'http.response.start',
            'status': 200,
            'headers': [[b'content-type', b'text/plain']],
        })
        await send({
            'type': 'http.response.body',
            'body': b'Hello from ASGI2!',
        })

# Wrap with ASGI2 middleware
app = ASGI2Middleware(MyASGI2App)

# Run with uvicorn
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=8000)

import uvicorn

# Uvicorn automatically detects ASGI2 applications
# No need to manually wrap with ASGI2Middleware

class MyASGI2App:
    def __init__(self, scope):
        self.scope = scope

    async def __call__(self, receive, send):
        ...

uvicorn.run(
    MyASGI2App,
    host="127.0.0.1",
    port=8000,
    interface="asgi2",  # Or use "auto" for automatic detection
)

from uvicorn.middleware.wsgi import WSGIMiddleware

# WSGI application (e.g., Flask)
def wsgi_app(environ, start_response):
    status = '200 OK'
    headers = [('Content-Type', 'text/plain')]
    start_response(status, headers)
    return [b'Hello from WSGI!']

# Wrap with WSGI middleware
app = WSGIMiddleware(wsgi_app, workers=10)

# Run with uvicorn
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=8000)

from flask import Flask
from uvicorn.middleware.wsgi import WSGIMiddleware
import uvicorn

# Create Flask app
flask_app = Flask(__name__)

@flask_app.route("/")
def hello():
    return "Hello from Flask on Uvicorn!"

# Wrap Flask with WSGI middleware
app = WSGIMiddleware(flask_app)

# Run with uvicorn
uvicorn.run(app, host="127.0.0.1", port=8000)

import uvicorn
from flask import Flask

flask_app = Flask(__name__)

@flask_app.route("/")
def hello():
    return "Hello, World!"

# Uvicorn automatically detects WSGI applications
uvicorn.run(
    flask_app,
    host="127.0.0.1",
    port=8000,
    interface="wsgi",  # Or use "auto" for automatic detection
)

from uvicorn.middleware.message_logger import MessageLoggerMiddleware
from uvicorn.logging import TRACE_LOG_LEVEL
import logging

# Enable TRACE logging
logging.addLevelName(TRACE_LOG_LEVEL, "TRACE")
logging.basicConfig(level=TRACE_LOG_LEVEL)

async def app(scope, receive, send):
    # Your ASGI application
    ...

# Wrap with message logger
app_with_logging = MessageLoggerMiddleware(app)

# Run with uvicorn
import uvicorn
uvicorn.run(
    app_with_logging,
    host="127.0.0.1",
    port=8000,
    log_level="trace",  # Enable trace logging
)

from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
from uvicorn.middleware.message_logger import MessageLoggerMiddleware

async def app(scope, receive, send):
    # Your ASGI application
    ...

# Apply middleware in order (innermost first)
app = MessageLoggerMiddleware(app)  # Applied last
app = ProxyHeadersMiddleware(app, trusted_hosts="127.0.0.1")  # Applied first

# Run with uvicorn
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)

from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware

async def app(scope, receive, send):
    ...

# Trust specific IP addresses
app = ProxyHeadersMiddleware(
    app,
    trusted_hosts=["127.0.0.1", "192.168.1.1", "192.168.1.2"],
)

# Trust CIDR networks
app = ProxyHeadersMiddleware(
    app,
    trusted_hosts=[
        "10.0.0.0/8",      # Private network
        "172.16.0.0/12",   # Private network
        "192.168.0.0/16",  # Private network
    ],
)

# Mix IPs and networks
app = ProxyHeadersMiddleware(
    app,
    trusted_hosts="127.0.0.1,10.0.0.0/8,192.168.1.100",
)

from uvicorn.middleware.wsgi import WSGIMiddleware

def wsgi_app(environ, start_response):
    # Blocking WSGI application
    import time
    time.sleep(0.1)  # Simulate blocking I/O

    status = '200 OK'
    headers = [('Content-Type', 'text/plain')]
    start_response(status, headers)
    return [b'Done!']

# Increase workers for blocking applications
app = WSGIMiddleware(
    wsgi_app,
    workers=50,  # More threads for handling blocking calls
)

import uvicorn
uvicorn.run(app, host="127.0.0.1", port=8000)

from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware

async def app(scope, receive, send):
    # Access the modified scope
    client_host, client_port = scope['client']
    scheme = scope['scheme']

    body = f"Client: {client_host}:{client_port}\nScheme: {scheme}".encode()

    await send({
        'type': 'http.response.start',
        'status': 200,
        'headers': [[b'content-type', b'text/plain']],
    })
    await send({
        'type': 'http.response.body',
        'body': body,
    })

# Wrap with proxy headers
app = ProxyHeadersMiddleware(app, trusted_hosts="*")

# Test with curl:
# curl -H "X-Forwarded-For: 203.0.113.1" -H "X-Forwarded-Proto: https" http://localhost:8000

from uvicorn.middleware.message_logger import MessageLoggerMiddleware
from uvicorn.logging import TRACE_LOG_LEVEL
import logging

# Configure trace logging
logging.addLevelName(TRACE_LOG_LEVEL, "TRACE")
logger = logging.getLogger("uvicorn")
logger.setLevel(TRACE_LOG_LEVEL)

handler = logging.StreamHandler()
handler.setLevel(TRACE_LOG_LEVEL)
logger.addHandler(handler)

async def app(scope, receive, send):
    message = await receive()  # Logged
    await send({  # Logged
        'type': 'http.response.start',
        'status': 200,
        'headers': [[b'content-type', b'text/plain']],
    })
    await send({  # Logged
        'type': 'http.response.body',
        'body': b'Hello!',
    })

# Wrap with message logger
app = MessageLoggerMiddleware(app)

import uvicorn
uvicorn.run(app, host="127.0.0.1", port=8000, log_level=TRACE_LOG_LEVEL)