orchestrata_ultimate.gateway.handlers.auth

Reference › orchestrata_ultimate › gateway › handlers › auth

AuthHandler serves the auth.authenticate RPC (SPEC-014 US-006), the call a client makes over an open socket to learn who it is and what the gateway offers. It is built by rpc and uses TokenAuthenticator from auth with settings from config; it also takes the gateway version from the gateway handler. handle_authenticate reads the token parameter and validates it with the same code path the connect-time flow uses. On failure it returns authenticated: false with the error text. On success it returns the client id, the read_write scope, the gateway version and a list of capability names (cxp, volk, session, governance, knowledge, gateway, state). See gateway.

At a glance

Class diagram

Dependency diagram

API

Classes

AuthHandler

Decorators: traces(spec='014', story='US-006', reqs=['REQ-014-086']) · Line 16

Token authentication handler.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/gateway/handlers/auth.py

"""Auth RPC handler -- authenticate.

Implements SPEC-014 auth.authenticate method.
"""

from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.gateway.auth import TokenAuthenticator
from orchestrata_ultimate.gateway.config import GatewayConfig
from orchestrata_ultimate.gateway.exceptions import AuthenticationError
from orchestrata_ultimate.gateway.handlers.gateway import GATEWAY_VERSION


@traces(spec="014", story="US-006", reqs=["REQ-014-086"])
class AuthHandler:
    """Token authentication handler.

    Validates the caller's JWT with the same TokenAuthenticator the connect
    path uses. This handler previously returned authenticated=True without
    reading params at all (CRITICAL bypass, audit 2026-07-06); anonymous
    access is only granted when the deployment explicitly opts in via
    ``allow_anonymous``.
    """

    def __init__(self, config: GatewayConfig | None = None) -> None:
        # GatewayConfig() self-populates from env in __post_init__.
        self._config = config or GatewayConfig()
        self._auth = TokenAuthenticator(self._config)

    @traces(
        spec="014",
        story="US-006",
        reqs=["REQ-014-086", "REQ-014-093"],
    )
    async def handle_authenticate(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle auth.authenticate -- validate token.

        Args:
            connection_id: Caller connection ID.
            params: token and optional client metadata.

        Returns:
            Authentication result with scope.
        """
        token = (params or {}).get("token", "")
        if token:
            try:
                claims = self._auth.validate_token(token)
            except AuthenticationError as exc:
                return {"authenticated": False, "error": str(exc)}
            client_id = str(claims["sub"])
        elif self._config.allow_anonymous:
            client_id = f"vizint-{connection_id[:8]}"
        else:
            return {"authenticated": False, "error": "Missing token"}
        return {
            "authenticated": True,
            "client_id": client_id,
            "scope": "read_write",
            "gateway_version": GATEWAY_VERSION,
            "capabilities": [
                "cxp",
                "volk",
                "session",
                "governance",
                "knowledge",
                "gateway",
                "state",
            ],
            "expires_at": None,
        }