nat.runtime.user_manager#

Runtime credential resolver that auto-detects identity source and creates UserInfo.

Attributes#

logger

Classes#

UserManager

Stateless resolver that creates UserInfo from HTTP/WebSocket connections.

Module Contents#

logger#
class UserManager#

Stateless resolver that creates UserInfo from HTTP/WebSocket connections.

classmethod extract_user_from_connection(
connection: starlette.requests.Request | fastapi.WebSocket,
) → nat.data_models.user_info.UserInfo | None#

Resolve an HTTP/WebSocket connection into a UserInfo.

Args:

connection: The incoming Starlette Request or WebSocket.

Returns:

A fully populated UserInfo, or None if no credential is present on the connection.

Raises:
ValueError: If a credential is found but cannot be resolved

to a valid user identity.

classmethod _resolve_from_auth_header(
auth_header: str,
) → nat.data_models.user_info.UserInfo | None#

Parse an Authorization header and resolve identity by scheme.

Args:

auth_header: Raw header value (e.g. Bearer <token> or Basic <b64>).

Returns:

A UserInfo if the header contains a recognised scheme with a non-empty credential, or None if the header is malformed or uses an unsupported scheme.

Raises:
ValueError: If a credential is present but cannot be decoded

(e.g. invalid JWT structure, malformed base64).

static _from_auth_payload(
payload: nat.data_models.api_server.AuthPayload,
) → nat.data_models.user_info.UserInfo#

Resolve a UserInfo from a WebSocket auth message payload.

This is an identity resolver, not an authenticator. JWTs are decoded with verify_signature=False to extract identity claims; API keys and basic credentials are mapped directly. Clients should verify and authenticate credentials (e.g. via JWKS, OAuth flows, or other auth middleware) before sending them over a WebSocket auth message.

Args:

payload: Discriminated union of JWT, API key, or basic auth credentials.

Returns:

A UserInfo with a deterministic user ID.

Raises:

ValueError: If the payload cannot be resolved to a valid user identity.

Extract the nat-session cookie value from a Request or WebSocket.

static _get_api_key_header(
connection: starlette.requests.Request | fastapi.WebSocket,
) → str | None#

Extract the X-API-Key header value from a connection.

static _get_auth_header(
connection: starlette.requests.Request | fastapi.WebSocket,
) → str | None#

Extract the raw Authorization header value from a connection.

Build a UserInfo from a session cookie value.

static _user_info_from_jwt(
claims: dict[str, Any],
) → nat.data_models.user_info.UserInfo#

Build a UserInfo from decoded JWT claims.

Registered claims (sub, iss, aud, exp, iat) follow RFC 7519. Identity claims (email, preferred_username, name) follow OpenID Connect Core 1.0 Section 5.1. sub is preferred as the stable identifier per RFC 7519 Section 4.1.2.

Raises:

ValueError: If the JWT contains no usable identity claim.

static _user_info_from_basic_auth(
b64_credential: str,
) → nat.data_models.user_info.UserInfo#

Build a UserInfo from a base64-encoded Basic Auth credential.

Args:

b64_credential: The base64-encoded username:password string.

Raises:

ValueError: If the credential cannot be decoded or is malformed.