Skip to content

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 
23
24
25
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 
50
51
52
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 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
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)

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 — browsers cannot set headers on a WebSocket handshake — or via an Authorization: Bearer <key> header for programmatic clients.

Source code in src/openjarvis/server/auth_middleware.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
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 — browsers cannot set headers on a WebSocket handshake — or via
    an ``Authorization: Bearer <key>`` header for programmatic clients.
    """
    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:
        return False
    return secrets.compare_digest(token, expected_key)