Skip to content

NatsJWT

faststream.nats.security.NatsJWT #

NatsJWT(
    jwt_cb: Callable[[], bytes | bytearray],
    signature_cb: Callable[[str], bytes],
    *,
    ssl_context: SSLContext | None = None,
    use_ssl: bool | None = None,
    tls_hostname: str | None = None,
    tls_handshake_first: bool = False,
)

Bases: NatsSecurity

NATS user JWT authentication with NKey challenge signing.

Source code in faststream/nats/security.py 
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
def __init__(
    self,
    jwt_cb: Callable[[], bytes | bytearray],
    signature_cb: Callable[[str], bytes],
    *,
    ssl_context: "SSLContext | None" = None,
    use_ssl: bool | None = None,
    tls_hostname: str | None = None,
    tls_handshake_first: bool = False,
) -> None:
    super().__init__(
        ssl_context=ssl_context,
        use_ssl=use_ssl,
        tls_hostname=tls_hostname,
        tls_handshake_first=tls_handshake_first,
    )
    self.jwt_cb = jwt_cb
    self.signature_cb = signature_cb

jwt_cb instance-attribute #

jwt_cb = jwt_cb

signature_cb instance-attribute #

signature_cb = signature_cb

ssl_context instance-attribute #

ssl_context: Optional[SSLContext] = ssl_context

use_ssl instance-attribute #

use_ssl: bool = use_ssl

tls_hostname instance-attribute #

tls_hostname = tls_hostname

tls_handshake_first instance-attribute #

tls_handshake_first = tls_handshake_first

get_requirement #

get_requirement() -> list[dict[str, Any]]

Get the AsyncAPI requirement for NATS JWT authentication.

Source code in faststream/nats/security.py 
359
360
361
362
@override
def get_requirement(self) -> list[dict[str, Any]]:
    """Get the AsyncAPI requirement for NATS JWT authentication."""
    return [{"nats-jwt": []}]

get_schema #

get_schema() -> dict[str, dict[str, Any]]

Get the AsyncAPI schema for NATS JWT authentication.

Source code in faststream/nats/security.py 
364
365
366
367
368
369
370
371
372
373
374
375
376
@override
def get_schema(self) -> dict[str, dict[str, Any]]:
    """Get the AsyncAPI schema for NATS JWT authentication."""
    return {
        "nats-jwt": {
            "type": "asymmetricEncryption",
            "description": (
                "NATS user JWT authentication with NKey challenge signing."
            ),
            "x-nats-auth": "jwt-nkey",
            "x-nats-credential-source": "callbacks",
        },
    }