Skip to content

NatsNKey

faststream.nats.security.NatsNKey #

NatsNKey(
    seed_file: str | Path,
    *,
    ssl_context: SSLContext | None = None,
    use_ssl: bool | None = None,
    tls_hostname: str | None = None,
    tls_handshake_first: bool = False,
)

Bases: NatsSecurity

NATS NKey challenge-response authentication.

Source code in faststream/nats/security.py
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
def __init__(
    self,
    seed_file: str | Path,
    *,
    ssl_context: "SSLContext | None" = None,
    use_ssl: bool | None = None,
    tls_hostname: str | None = None,
    tls_handshake_first: bool = False,
) -> None:
    if _nkeys is None:
        raise ImportError(INSTALL_NATS_NKEYS)

    if not str(seed_file):
        msg = "NATS NKey seed file cannot be empty."
        raise ValueError(msg)

    super().__init__(
        ssl_context=ssl_context,
        use_ssl=use_ssl,
        tls_hostname=tls_hostname,
        tls_handshake_first=tls_handshake_first,
    )
    self.seed = str(seed_file)
    self.seed_is_file = True

seed instance-attribute #

seed = str(seed_file)

seed_is_file instance-attribute #

seed_is_file = True

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 NKey authentication.

Source code in faststream/nats/security.py
270
271
272
273
@override
def get_requirement(self) -> list[dict[str, Any]]:
    """Get the AsyncAPI requirement for NATS NKey authentication."""
    return [{"nats-nkey": []}]

get_schema #

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

Get the AsyncAPI schema for NATS NKey authentication.

Source code in faststream/nats/security.py
275
276
277
278
279
280
281
282
283
284
285
286
287
288
@override
def get_schema(self) -> dict[str, dict[str, Any]]:
    """Get the AsyncAPI schema for NATS NKey authentication."""
    return {
        "nats-nkey": {
            "type": "asymmetricEncryption",
            "description": (
                "NATS NKey challenge-response authentication using an "
                "Ed25519 signature."
            ),
            "x-nats-auth": "nkey",
            "x-nats-algorithm": "ed25519",
        },
    }

from_file classmethod #

from_file(
    seed_file: str | Path,
    *,
    ssl_context: SSLContext | None = None,
    use_ssl: bool | None = None,
    tls_hostname: str | None = None,
    tls_handshake_first: bool = False,
) -> Self

Load a NATS NKey seed file.

Source code in faststream/nats/security.py
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
@classmethod
def from_file(
    cls,
    seed_file: str | Path,
    *,
    ssl_context: "SSLContext | None" = None,
    use_ssl: bool | None = None,
    tls_hostname: str | None = None,
    tls_handshake_first: bool = False,
) -> Self:
    """Load a NATS NKey seed file."""
    return cls(
        seed_file,
        ssl_context=ssl_context,
        use_ssl=use_ssl,
        tls_hostname=tls_hostname,
        tls_handshake_first=tls_handshake_first,
    )

from_seed classmethod #

from_seed(
    seed: str,
    *,
    ssl_context: SSLContext | None = None,
    use_ssl: bool | None = None,
    tls_hostname: str | None = None,
    tls_handshake_first: bool = False,
) -> Self

Use a raw NATS NKey seed.

Source code in faststream/nats/security.py
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
@classmethod
def from_seed(
    cls,
    seed: str,
    *,
    ssl_context: "SSLContext | None" = None,
    use_ssl: bool | None = None,
    tls_hostname: str | None = None,
    tls_handshake_first: bool = False,
) -> Self:
    """Use a raw NATS NKey seed."""
    if not seed:
        msg = "NATS NKey seed cannot be empty."
        raise ValueError(msg)

    security = cls(
        seed,
        ssl_context=ssl_context,
        use_ssl=use_ssl,
        tls_hostname=tls_hostname,
        tls_handshake_first=tls_handshake_first,
    )
    security.seed_is_file = False
    return security