Spoolman/spoolman/security.py

"""Origin trust decisions, shared by the CSRF and websocket guards.

Spoolman has no authentication by design; the boundary that protects a user's data is the
network. This module answers the one question those guards need: "is this browser origin
someone we trust?".

The answer is driven entirely by ``SPOOLMAN_CORS_ORIGIN``, which is already an operator-declared
origin allowlist. An operator who allowlists an origin for CORS has declared it trusted to make
credentialed cross-origin requests, so refusing that same origin here would break the very
deployment CORS was configured for. Deliberately no second environment variable: two variables
answering one question would drift apart, and there is nothing a second one could express that
this one cannot.

Three cases, in the operator's terms:

* Unset -- the case for nearly every deployment -- means "same-origin only". Non-browser clients
  such as Moonraker and OctoPrint send no ``Origin`` at all and are unaffected.
* A list of origins means "these, plus my own". This is the same declaration CORS already needs.
* ``*`` means "I do not want origin checks", and is honoured as such: it is what an operator
  reaches for to opt out, so quietly reinterpreting it as something narrower would only leave
  them with no way to say what they meant. Debug mode implies the same thing. Both are logged
  loudly at startup.

Deployments behind a reverse proxy that rewrites ``Host`` are handled without any configuration
by also accepting ``X-Forwarded-Host``; see :func:`is_trusted_request`.
"""

import logging
from functools import cache
from typing import TYPE_CHECKING
from urllib.parse import urlsplit

from starlette.requests import HTTPConnection
from starlette.responses import JSONResponse

from spoolman import env

if TYPE_CHECKING:
    from starlette.types import ASGIApp, Receive, Scope, Send

logger = logging.getLogger(__name__)

WILDCARD_ORIGIN = "*"

# Methods that can change state. Safe methods are left alone: the same-origin policy already
# stops a cross-origin page from reading their responses, and OPTIONS must stay reachable for
# CORS preflight to work at all.
UNSAFE_METHODS = frozenset({"POST", "PUT", "PATCH", "DELETE"})

# Websocket close code used when a handshake is refused. In the 4000-4999 range reserved for
# application use, and deliberately echoes HTTP 403.
WS_CLOSE_FORBIDDEN = 4403

# An origin of "null" is what a browser sends from a sandboxed iframe, a data: URL or a
# file:// page. It identifies nobody, so it can never be trusted.
_OPAQUE_ORIGIN = "null"

_DEFAULT_PORTS = {"http": ":80", "https": ":443"}


@cache
def _warn_all_origins_trusted(reason: str) -> None:
    """Warn that origin checks are disabled. Cached so it is logged only once per process."""
    logger.warning(
        "%s, so origin checks are disabled. Any website you visit in a browser that can reach this "
        "Spoolman instance is able to read and modify your data. Only use this on a trusted network; "
        "otherwise list the origins you actually need instead.",
        reason,
    )


def _authority(origin: str) -> str:
    """Extract the ``host[:port]`` part of an already-normalized origin.

    A port that is the default for the scheme is dropped, so that ``https://spoolman.local:443``
    and ``https://spoolman.local`` compare equal to a ``Host`` of ``spoolman.local``.

    Args:
        origin: A normalized origin, e.g. ``https://spoolman.local:7912``.

    Returns:
        str: The authority, or an empty string if the origin has no host part.

    """
    parts = urlsplit(origin)
    default_port = _DEFAULT_PORTS.get(parts.scheme)
    if default_port is not None:
        return parts.netloc.removesuffix(default_port)
    return parts.netloc


def _normalize_host(host: str) -> str:
    """Normalize a ``Host`` header for comparison against an origin's authority.

    Args:
        host: The raw ``Host`` header value, e.g. ``Spoolman.local:7912``.

    Returns:
        str: The normalized host.

    """
    host = host.strip().lower()
    # The scheme is unknown here, so drop whichever default port is present.
    return host.removesuffix(":80").removesuffix(":443")


def _own_hosts(host: str, forwarded_host: str | None) -> set[str]:
    """Collect the hostnames that identify this instance, as the browser could have addressed it.

    Args:
        host: The request's ``Host`` header.
        forwarded_host: The request's ``X-Forwarded-Host`` header, if any. A chain of proxies
            appends to this header, so only the first entry -- the host the browser actually
            asked for -- is considered.

    Returns:
        set[str]: The normalized hostnames.

    """
    hosts = {_normalize_host(host)}
    if forwarded_host:
        hosts.add(_normalize_host(forwarded_host.split(",")[0]))
    hosts.discard("")
    return hosts


def trusts_all_origins() -> bool:
    """Determine whether the operator has opted out of origin checks entirely.

    That is either ``SPOOLMAN_CORS_ORIGIN=*`` or debug mode, which already declares that all
    origins are allowed. Both are logged once, loudly.

    Returns:
        bool: Whether every origin should be trusted.

    """
    if env.is_debug_mode():
        _warn_all_origins_trusted("Running in debug mode")
        return True
    if WILDCARD_ORIGIN in (env.get_cors_origin() or []):
        _warn_all_origins_trusted(f"SPOOLMAN_CORS_ORIGIN is set to '{WILDCARD_ORIGIN}'")
        return True
    return False


def get_trusted_origins() -> set[str]:
    """Get the set of normalized origins the operator has explicitly declared trusted.

    Does not account for the wildcard; see :func:`trusts_all_origins` for that.

    Returns:
        set[str]: The trusted origins. Empty when nothing was configured.

    """
    return {origin for origin in env.get_cors_origin() or [] if origin != WILDCARD_ORIGIN}


def is_trusted_origin(origin: str | None, host: str, forwarded_host: str | None = None) -> bool:
    """Determine whether a request's browser origin is trusted.

    Returns True for every origin when the operator has opted out via ``SPOOLMAN_CORS_ORIGIN=*``
    or debug mode.

    Args:
        origin: The request's ``Origin`` header, if any. An absent header means either a
            same-origin navigation or a non-browser client such as Moonraker or OctoPrint,
            neither of which the same-origin policy applies to, so both are trusted.
        host: The request's own ``Host`` header. An origin pointing at the host the request
            was made to is by definition same-origin.
        forwarded_host: The request's ``X-Forwarded-Host`` header, if any. See
            :func:`is_trusted_request` for why this is trusted.

    Returns:
        bool: Whether the origin is trusted.

    """
    if origin is None or not origin.strip():
        return True

    if trusts_all_origins():
        return True

    normalized = env.normalize_origin(origin)
    if normalized == _OPAQUE_ORIGIN:
        return False

    if normalized in get_trusted_origins():
        return True

    authority = _authority(normalized)
    return bool(authority) and authority in _own_hosts(host, forwarded_host)


def is_trusted_request(connection: HTTPConnection) -> bool:
    """Determine whether a request or websocket handshake comes from a trusted origin.

    This is the entry point for the CSRF and websocket guards; it accepts both a ``Request``
    and a ``WebSocket``.

    ``X-Forwarded-Host`` is honoured because the reverse proxies that need it are exactly the
    ones that break the ``Origin``/``Host`` comparison: nginx without ``proxy_set_header Host
    $host`` and Apache without ``ProxyPreserveHost On`` replace ``Host`` with the backend's
    address, but set ``X-Forwarded-Host`` to what the browser asked for. Without it those
    deployments would see the guard reject their own web UI.

    Honouring it does not weaken the guard against the attacks it exists to stop, because a
    malicious page cannot set the header on any of them: an HTML form cannot set request headers
    at all, the browser websocket API cannot either, and a ``fetch`` that adds one stops being a
    simple request, so it needs a CORS preflight that an untrusted origin will not get. A
    non-browser client can forge it freely, but it can equally forge ``Origin`` and ``Host``, and
    the guard never defended against those -- an absent ``Origin`` is trusted by design so that
    Moonraker and OctoPrint keep working.

    Args:
        connection: The incoming HTTP request or websocket connection.

    Returns:
        bool: Whether the connection is trusted.

    """
    headers = connection.headers
    return is_trusted_origin(
        headers.get("origin"),
        headers.get("host", ""),
        headers.get("x-forwarded-host"),
    )


class TrustedOriginMiddleware:
    """Refuse state-changing requests and websocket handshakes from untrusted browser origins.

    Spoolman has no authentication, so without this a page on any website the user happens to
    visit can make their browser write to a Spoolman instance it can reach (a form post needs no
    CORS preflight), and can open a websocket to it (websockets are exempt from CORS entirely).

    A single middleware rather than a per-endpoint dependency, so that an endpoint added later
    cannot forget to opt in.
    """

    def __init__(self, app: "ASGIApp") -> None:
        """Wrap the given ASGI application."""
        self.app = app

    async def __call__(self, scope: "Scope", receive: "Receive", send: "Send") -> None:
        """Reject the connection if its origin is not trusted, otherwise pass it through."""
        # Every websocket handshake is guarded, since websockets are exempt from CORS and are
        # readable by the page that opens them. Of plain HTTP only the state-changing methods are.
        guarded = scope["type"] == "websocket" or (scope["type"] == "http" and scope["method"] in UNSAFE_METHODS)
        if guarded:
            connection = HTTPConnection(scope)
            if not is_trusted_request(connection):
                await self._refuse(scope, receive, send, connection)
                return
        await self.app(scope, receive, send)

    async def _refuse(
        self,
        scope: "Scope",
        receive: "Receive",
        send: "Send",
        connection: HTTPConnection,
    ) -> None:
        """Close the connection, and log enough for an operator to see why."""
        logger.warning(
            "Refused a %s from an untrusted origin. Origin: %r, Host: %r, X-Forwarded-Host: %r, path: %r. "
            "If this was you, add the origin to SPOOLMAN_CORS_ORIGIN.",
            "websocket handshake" if scope["type"] == "websocket" else scope["method"] + " request",
            connection.headers.get("origin"),
            connection.headers.get("host"),
            connection.headers.get("x-forwarded-host"),
            scope.get("path"),
        )
        if scope["type"] == "websocket":
            # The ASGI server expects the connect message to be consumed before a close is sent.
            await receive()
            await send({"type": "websocket.close", "code": WS_CLOSE_FORBIDDEN})
            return
        response = JSONResponse(
            status_code=403,
            content={
                "message": (
                    "Request refused: it came from an origin this Spoolman instance does not trust. "
                    "Set SPOOLMAN_CORS_ORIGIN if this origin should be allowed."
                ),
            },
        )
        await response(scope, receive, send)