Skip to content

RabbitExternalAuth

faststream.rabbit.RabbitExternalAuth #

RabbitExternalAuth(
    ssl_context: Optional[SSLContext] = None,
    use_ssl: bool = True,
)

Bases: BaseSecurity

RabbitMQ SASL EXTERNAL authentication using a client TLS certificate.

Source code in faststream/rabbit/security.py 
14
15
16
17
18
19
20
21
22
def __init__(
    self,
    ssl_context: Optional["SSLContext"] = None,
    use_ssl: bool = True,
) -> None:
    super().__init__(
        ssl_context=ssl_context,
        use_ssl=use_ssl,
    )

ssl_context instance-attribute #

ssl_context: Optional[SSLContext] = ssl_context

use_ssl instance-attribute #

use_ssl: bool = use_ssl

get_requirement #

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

Get the security requirements for X.509 authentication.

Source code in faststream/rabbit/security.py 
24
25
26
def get_requirement(self) -> list[dict[str, Any]]:
    """Get the security requirements for X.509 authentication."""
    return [{"rabbitmq-external": []}]

get_schema #

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

Get the security schema for X.509 authentication.

Source code in faststream/rabbit/security.py 
28
29
30
def get_schema(self) -> dict[str, dict[str, str]]:
    """Get the security schema for X.509 authentication."""
    return {"rabbitmq-external": {"type": "X509"}}