auth_middleware

auth_middleware

API key authentication middleware for the OpenJarvis 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/openjarvis/server/auth_middleware.py

def __init__(self, app, api_key: str = "") -> None:  # noqa: ANN001
    super().__init__(app)
    self._api_key = api_key or os.environ.get("OPENJARVIS_API_KEY", "")

Functions

generate_api_key

generate_api_key() -> str

Generate a new API key with oj_sk_ prefix.

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

def generate_api_key() -> str:
    """Generate a new API key with ``oj_sk_`` prefix."""
    return f"oj_sk_{secrets.token_urlsafe(32)}"

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/openjarvis/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 OPENJARVIS_API_KEY to be set. "
            "Run: jarvis auth generate-key",
            host,
        )
        sys.exit(1)

authenticate_websocket

authenticate_websocket(websocket, expected_key: str) -> tuple[bool, str | None]

Authenticate a WebSocket and return its negotiated auth subprotocol.

Programmatic clients can send Authorization: Bearer <key>. Browser clients, which cannot set that header, offer openjarvis.auth.v1 plus a marked, unpadded base64url encoding of the UTF-8 key. The encoding only makes the credential valid subprotocol syntax; it does not make it secret.

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

def authenticate_websocket(
    websocket,
    expected_key: str,  # noqa: ANN001
) -> tuple[bool, str | None]:
    """Authenticate a WebSocket and return its negotiated auth subprotocol.

    Programmatic clients can send ``Authorization: Bearer <key>``. Browser
    clients, which cannot set that header, offer ``openjarvis.auth.v1`` plus a
    marked, unpadded base64url encoding of the UTF-8 key. The encoding only
    makes the credential valid subprotocol syntax; it does not make it secret.
    """
    credential_protocol, selected_protocol = _offered_websocket_auth(websocket)

    # Match AuthMiddleware's local, keyless behavior. If a stale client still
    # offers a well-formed auth protocol, negotiate it so the browser does not
    # fail an otherwise allowed handshake.
    if not expected_key:
        return True, selected_protocol

    auth = websocket.headers.get("authorization", "")
    scheme, _, header_token = auth.partition(" ")
    header_valid = scheme.lower() == "bearer" and _api_keys_match(
        header_token, expected_key
    )

    expected_protocol = _websocket_key_protocol(expected_key)
    protocol_valid = bool(credential_protocol and expected_protocol) and (
        secrets.compare_digest(credential_protocol, expected_protocol)
    )
    return header_valid or protocol_valid, selected_protocol

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. See :func:authenticate_websocket for the supported credential transports. URL query parameters are deliberately not accepted because request targets commonly appear in access logs and browser history.

Source code in src/openjarvis/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. See :func:`authenticate_websocket` for the supported
    credential transports. URL query parameters are deliberately not accepted
    because request targets commonly appear in access logs and browser history.
    """
    return authenticate_websocket(websocket, expected_key)[0]