auth_middleware

auth_middleware

API key authentication middleware for the Diapason server.

Classes

AuthMiddleware

AuthMiddleware(app, api_key: str = '')

Bases: BaseHTTPMiddleware

Validates Authorization: Bearer <key> on /v1/* and /api/* routes.

Webhook routes and health checks are exempt — they use per-channel signature verification instead.

Source code in src/diapason/server/auth_middleware.py

def __init__(self, app, api_key: str = "") -> None:  # noqa: ANN001
    super().__init__(app)
    self._api_key = api_key or (_env_get("API_KEY") or "")

RateLimitMiddleware

RateLimitMiddleware(
    app,
    *,
    requests_per_minute: int = 60,
    burst_size: int = 10,
    enabled: bool = True,
)

Bases: BaseHTTPMiddleware

Throttle authenticated API traffic per credential and client address.

Source code in src/diapason/server/auth_middleware.py

def __init__(
    self,
    app,  # noqa: ANN001
    *,
    requests_per_minute: int = 60,
    burst_size: int = 10,
    enabled: bool = True,
) -> None:
    super().__init__(app)
    self._limiter = RateLimiter(
        RateLimitConfig(
            requests_per_minute=requests_per_minute,
            burst_size=burst_size,
            enabled=enabled,
        )
    )
    # A separate bucket for the mesh routes that carry no API key.
    #
    # The main limiter only ever ran for paths that require the key, so
    # every route deliberately opened to unauthenticated devices was also
    # opened to unlimited traffic — including the pairing front door,
    # whose own module claimed the opposite. Keyed on client address,
    # since there is no credential to key on before the body is parsed,
    # and parsing the body is the cost we are trying to bound.
    #
    # Roomier than the authenticated bucket on purpose: a paired device
    # legitimately polls every couple of seconds and beacons besides, and
    # throttling that would break the very devices this is protecting.
    self._open_limiter = RateLimiter(
        RateLimitConfig(
            requests_per_minute=max(120, requests_per_minute * 2),
            burst_size=max(20, burst_size * 2),
            enabled=enabled,
        )
    )

    # Le seau du TRANSFERT, à part (25 août 2026). Un morceau vaut un
    # mégaoctet : un fichier de 500 Mio, c'est cinq cents requêtes en
    # rafale, parfaitement légitimes. Les compter dans le seau du
    # maillage aurait fait tomber présence et relèves. Ce seau est donc
    # large en NOMBRE — le vrai plafond est en octets, et il vit dans le
    # routeur, là où l'on sait ce qu'une session a déjà reçu.
    # Le seau des GESTES. Douze images par seconde pendant dix minutes,
    # c'est sept mille deux cents requêtes — toutes légitimes. Ce seau
    # est large en nombre parce que le vrai garde-fou est ailleurs : la
    # session se désarme seule après quatre-vingt-dix secondes de
    # silence et dix minutes au total.
    self._gesture_limiter = RateLimiter(
        RateLimitConfig(
            requests_per_minute=max(1200, requests_per_minute * 20),
            burst_size=max(60, burst_size * 6),
            enabled=enabled,
        )
    )

    self._transfer_limiter = RateLimiter(
        RateLimitConfig(
            requests_per_minute=max(1200, requests_per_minute * 20),
            burst_size=max(200, burst_size * 20),
            enabled=enabled,
        )
    )

Functions:

est_route_de_gestes

est_route_de_gestes(path: str) -> bool

Une route du mode gestes, dont le débit normal est élevé.

Source code in src/diapason/server/auth_middleware.py

def est_route_de_gestes(path: str) -> bool:
    """Une route du mode gestes, dont le débit normal est élevé."""
    return bool(_GESTES_RE.match(path or ""))

est_route_de_transfert

est_route_de_transfert(path: str) -> bool

Une route de transfert de fichiers, authentifiée par session.

Source code in src/diapason/server/auth_middleware.py

def est_route_de_transfert(path: str) -> bool:
    """Une route de transfert de fichiers, authentifiée par session."""
    return bool(_TRANSFERT_RE.match(path or ""))

generate_api_key

generate_api_key() -> str

Generate a 256-bit Diapason API key.

Source code in src/diapason/server/auth_middleware.py

def generate_api_key() -> str:
    """Generate a 256-bit Diapason API key."""
    return f"diapason_sk_{secrets.token_urlsafe(32)}"

ensure_local_api_key

ensure_local_api_key(
    explicit_key: str = "",
) -> tuple[str, Path | None]

Return an API key, creating a permission-restricted local key if needed.

Precedence is an explicit configured value, DIAPASON_API_KEY (with legacy environment aliases), then <DIAPASON_HOME>/auth/local_api_key. The generated file and its parent are owner-only and creation is atomic.

Source code in src/diapason/server/auth_middleware.py

def ensure_local_api_key(explicit_key: str = "") -> tuple[str, Path | None]:
    """Return an API key, creating a permission-restricted local key if needed.

    Precedence is an explicit configured value, ``DIAPASON_API_KEY`` (with
    legacy environment aliases), then ``<DIAPASON_HOME>/auth/local_api_key``.
    The generated file and its parent are owner-only and creation is atomic.
    """
    configured = explicit_key or (_env_get("API_KEY") or "")
    if configured:
        if len(configured.encode("utf-8")) < 32:
            raise ValueError(
                "DIAPASON_API_KEY must contain at least 32 bytes; "
                "generate one with 'diapason auth generate-key'."
            )
        return configured, None

    auth_dir = get_config_dir() / "auth"
    key_path = auth_dir / "local_api_key"
    auth_dir.mkdir(mode=0o700, parents=True, exist_ok=True)
    try:
        auth_dir.chmod(0o700)
    except OSError:
        pass

    try:
        existing = _read_local_api_key(key_path)
    except FileNotFoundError:
        existing = ""
    if existing:
        if len(existing.encode("utf-8")) < 32:
            raise RuntimeError(
                "Local API key is too short; rotate it with "
                f"'diapason auth create-key': {key_path}"
            )
        return existing, key_path

    generated = generate_api_key()
    try:
        flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
        if hasattr(os, "O_NOFOLLOW"):
            flags |= os.O_NOFOLLOW
        descriptor = os.open(key_path, flags, 0o600)
    except FileExistsError:
        existing = _read_local_api_key(key_path)
        if len(existing.encode("utf-8")) < 32:
            raise RuntimeError(
                f"Local API key file is missing or too short: {key_path}"
            )
        return existing, key_path
    with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
        handle.write(generated + "\n")
    return generated, key_path

check_bind_safety

check_bind_safety(host: str, *, api_key: str) -> None

Refuse to bind non-loopback without an API key.

Raises SystemExit if host is not a loopback address and api_key is empty.

Source code in src/diapason/server/auth_middleware.py

def check_bind_safety(host: str, *, api_key: str) -> None:
    """Refuse to bind non-loopback without an API key.

    Raises ``SystemExit`` if *host* is not a loopback address and
    *api_key* is empty.
    """
    import ipaddress
    import sys

    try:
        is_loop = ipaddress.ip_address(host).is_loopback
    except ValueError:
        is_loop = host in ("localhost", "")

    if not is_loop and not api_key:
        logger.error(
            "Binding to %s requires DIAPASON_API_KEY to be set. "
            "Run: diapason auth generate-key",
            host,
        )
        sys.exit(1)

check_cors_safety

check_cors_safety(host: str, origins: list[str]) -> None

Reject credentialed wildcard CORS on a non-loopback listener.

Source code in src/diapason/server/auth_middleware.py

def check_cors_safety(host: str, origins: list[str]) -> None:
    """Reject credentialed wildcard CORS on a non-loopback listener."""
    import ipaddress

    try:
        is_loop = ipaddress.ip_address(host).is_loopback
    except ValueError:
        is_loop = host in ("localhost", "")
    if not is_loop and "*" in origins:
        raise ValueError(
            "Wildcard CORS is forbidden on a non-loopback Diapason server. "
            "Configure explicit trusted origins in server.cors_origins."
        )

websocket_authorized

websocket_authorized(websocket, expected_key: str) -> bool

Return True if a WebSocket connection presents the expected key.

AuthMiddleware is a BaseHTTPMiddleware and never sees WebSocket upgrade requests, so streaming endpoints must check the token themselves in the handshake before calling websocket.accept().

When expected_key is empty, authentication is disabled (the loopback / local-only default, matching :class:AuthMiddleware) and all connections are allowed. The token may be supplied either as a ?token= query parameter for backwards compatibility, via an Authorization: Bearer header for programmatic clients, or as a diapason-auth.<key> offered subprotocol. The desktop uses the last form so access logs never contain its credential in the request URL.

Source code in src/diapason/server/auth_middleware.py

def websocket_authorized(websocket, expected_key: str) -> bool:  # noqa: ANN001
    """Return ``True`` if a WebSocket connection presents the expected key.

    ``AuthMiddleware`` is a ``BaseHTTPMiddleware`` and never sees WebSocket
    upgrade requests, so streaming endpoints must check the token themselves
    in the handshake before calling ``websocket.accept()``.

    When *expected_key* is empty, authentication is disabled (the loopback /
    local-only default, matching :class:`AuthMiddleware`) and all connections
    are allowed. The token may be supplied either as a ``?token=`` query
    parameter for backwards compatibility, via an ``Authorization: Bearer``
    header for programmatic clients, or as a ``diapason-auth.<key>`` offered
    subprotocol. The desktop uses the last form so access logs never contain
    its credential in the request URL.
    """
    if not expected_key:
        return True
    token = websocket.query_params.get("token", "")
    if not token:
        auth = websocket.headers.get("authorization", "")
        scheme, _, value = auth.partition(" ")
        if scheme.lower() == "bearer":
            token = value
    if not token:
        offered = websocket.headers.get("sec-websocket-protocol", "")
        for protocol in (item.strip() for item in offered.split(",")):
            if protocol.startswith("diapason-auth."):
                token = protocol.removeprefix("diapason-auth.")
                break
    if not token:
        return False
    return secrets.compare_digest(token, expected_key)

websocket_response_subprotocol

websocket_response_subprotocol(websocket) -> str | None

Select the non-secret Diapason protocol when a browser offers it.

Source code in src/diapason/server/auth_middleware.py

def websocket_response_subprotocol(websocket) -> str | None:  # noqa: ANN001
    """Select the non-secret Diapason protocol when a browser offers it."""
    offered = websocket.headers.get("sec-websocket-protocol", "")
    protocols = {item.strip() for item in offered.split(",")}
    return "diapason" if "diapason" in protocols else None