Skip to content

NatsCredentials

faststream.nats.security.NatsCredentials #

NatsCredentials(
    credentials: Credentials,
    *,
    ssl_context: SSLContext | None = None,
    use_ssl: bool | None = None,
    tls_hostname: str | None = None,
    tls_handshake_first: bool = False,
)

Bases: NatsSecurity

NATS JWT credentials authentication.

Source code in faststream/nats/security.py
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def __init__(
    self,
    credentials: Credentials,
    *,
    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)

    super().__init__(
        ssl_context=ssl_context,
        use_ssl=use_ssl,
        tls_hostname=tls_hostname,
        tls_handshake_first=tls_handshake_first,
    )
    self.credentials = credentials

credentials instance-attribute #

credentials = credentials

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

Source code in faststream/nats/security.py
157
158
159
160
@override
def get_requirement(self) -> list[dict[str, Any]]:
    """Get the AsyncAPI requirement for NATS credentials authentication."""
    return [{"nats-jwt": []}]

get_schema #

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

Get the AsyncAPI schema for NATS credentials authentication.

Source code in faststream/nats/security.py
162
163
164
165
166
167
168
169
170
171
172
173
174
@override
def get_schema(self) -> dict[str, dict[str, Any]]:
    """Get the AsyncAPI schema for NATS credentials authentication."""
    return {
        "nats-jwt": {
            "type": "asymmetricEncryption",
            "description": (
                "NATS user JWT authentication with NKey challenge signing."
            ),
            "x-nats-auth": "jwt-nkey",
            "x-nats-credential-source": "credentials-file",
        },
    }

from_file classmethod #

from_file(
    credentials_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 combined NATS credentials file.

Source code in faststream/nats/security.py
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
@classmethod
def from_file(
    cls,
    credentials_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 combined NATS credentials file."""
    return cls(
        credentials_file,
        ssl_context=ssl_context,
        use_ssl=use_ssl,
        tls_hostname=tls_hostname,
        tls_handshake_first=tls_handshake_first,
    )

from_files classmethod #

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

Load separate NATS user JWT and seed files.

Source code in faststream/nats/security.py
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
@classmethod
def from_files(
    cls,
    *,
    jwt: str | Path,
    seed: str | Path,
    ssl_context: "SSLContext | None" = None,
    use_ssl: bool | None = None,
    tls_hostname: str | None = None,
    tls_handshake_first: bool = False,
) -> Self:
    """Load separate NATS user JWT and seed files."""
    return cls(
        (str(jwt), str(seed)),
        ssl_context=ssl_context,
        use_ssl=use_ssl,
        tls_hostname=tls_hostname,
        tls_handshake_first=tls_handshake_first,
    )

from_raw classmethod #

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

Use raw combined NATS credentials.

Source code in faststream/nats/security.py
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
@classmethod
def from_raw(
    cls,
    credentials: str,
    *,
    ssl_context: "SSLContext | None" = None,
    use_ssl: bool | None = None,
    tls_hostname: str | None = None,
    tls_handshake_first: bool = False,
) -> Self:
    """Use raw combined NATS credentials."""
    markers = ("BEGIN NATS USER JWT", "BEGIN USER NKEY SEED")
    if not credentials or any(marker not in credentials for marker in markers):
        msg = "Raw NATS credentials do not contain the required markers."
        raise ValueError(msg)

    return cls(
        RawCredentials(credentials),
        ssl_context=ssl_context,
        use_ssl=use_ssl,
        tls_hostname=tls_hostname,
        tls_handshake_first=tls_handshake_first,
    )