nat.runtime.user_manager#

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

Attributes#

logger

Exceptions#

`IdentityCredentialNotAcceptedError`	Raised when a supplied identity credential method is disabled by policy.
`IdentityHeaderError`	Raised when a configured trusted identity header is missing or ambiguous.
`JwtVerificationError`	Raised when an enabled JWT verification policy rejects a token.

Classes#

UserManager

Stateless resolver that creates UserInfo from HTTP/WebSocket connections.

Module Contents#

logger#

exception IdentityCredentialNotAcceptedError#

Bases: ValueError

Raised when a supplied identity credential method is disabled by policy.

Initialize self. See help(type(self)) for accurate signature.

exception IdentityHeaderError#

Bases: ValueError

Raised when a configured trusted identity header is missing or ambiguous.

Initialize self. See help(type(self)) for accurate signature.

exception JwtVerificationError#

Bases: ValueError

Raised when an enabled JWT verification policy rejects a token.

Initialize self. See help(type(self)) for accurate signature.

class UserManager#

Stateless resolver that creates UserInfo from HTTP/WebSocket connections.

classmethod extract_user_from_connection( connection: starlette.requests.Request | fastapi.WebSocket, accepted_identity_credentials: Collection[nat.data_models.api_server.IdentityCredentialType] | None = None, identity_header: str | None = None, ) → 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 extract_user_from_connection_with_verification( connection: starlette.requests.Request | fastapi.WebSocket, accepted_identity_credentials: Collection[nat.data_models.api_server.IdentityCredentialType] | None = None, jwt_validators: Mapping[str, nat.authentication.credential_validator.bearer_token_validator.BearerTokenValidator] | None = None, identity_header: str | None = None, ) → nat.data_models.user_info.UserInfo | None#

Async:

classmethod _resolve_from_auth_header_with_verification( auth_header: str, accepted_identity_credentials: Collection[nat.data_models.api_server.IdentityCredentialType] | None = None, jwt_validators: Mapping[str, nat.authentication.credential_validator.bearer_token_validator.BearerTokenValidator] | None = None, ) → nat.data_models.user_info.UserInfo | None#

Async:

classmethod _resolve_from_auth_header( auth_header: str, accepted_identity_credentials: Collection[nat.data_models.api_server.IdentityCredentialType] | None = None, ) → 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, accepted_identity_credentials: Collection[nat.data_models.api_server.IdentityCredentialType] | None = None, ) → 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.

static from_auth_payload_with_verification( payload: nat.data_models.api_server.AuthPayload, accepted_identity_credentials: Collection[nat.data_models.api_server.IdentityCredentialType] | None = None, jwt_validators: Mapping[str, nat.authentication.credential_validator.bearer_token_validator.BearerTokenValidator] | None = None, ) → nat.data_models.user_info.UserInfo#

Async:

Resolve an auth payload, verifying JWTs when a verifier is configured.

static _decode_verified_jwt( token: str, jwt_validators: Mapping[str, nat.authentication.credential_validator.bearer_token_validator.BearerTokenValidator] | None, ) → dict[str, Any]#

Async:

static _ensure_identity_credential_accepted( credential_type: nat.data_models.api_server.IdentityCredentialType, accepted_identity_credentials: Collection[nat.data_models.api_server.IdentityCredentialType] | None, ) → None#: Reject a recognized credential type when it is disabled by policy.

static _get_identity_header( connection: starlette.requests.Request | fastapi.WebSocket, header_name: str, ) → str#: Read one non-empty value for a trusted identity header, failing closed otherwise.

static _get_session_cookie( connection: starlette.requests.Request | fastapi.WebSocket, ) → str | None#: 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.

static _user_info_from_session_cookie( cookie_value: str, ) → nat.data_models.user_info.UserInfo#: Build a UserInfo from a session cookie value.

static _user_info_from_jwt( claims: dict[str, Any], *, issuer_scoped: bool = False, ) → 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.