Authentication Credentials

credentials

The credentials classes are used to encapsulate all authentication information for the :class:~pika.connection.ConnectionParameters class.

The :class:~pika.credentials.PlainCredentials class returns the properly formatted username and password to the :class:~pika.connection.Connection.

To authenticate with Pika, create a :class:~pika.credentials.PlainCredentials object passing in the username and password and pass it as the credentials argument value to the :class:~pika.connection.ConnectionParameters object.

If you are using :class:~pika.connection.URLParameters you do not need a credentials object, one will automatically be created for you.

If you are looking to implement SSL certificate style authentication, you would extend the :class:~pika.credentials.ExternalCredentials class implementing the required behavior.

ExternalCredentials

The ExternalCredentials class allows the connection to use EXTERNAL authentication, generally with a client SSL certificate.

Source code in pika/credentials.py

class ExternalCredentials:
    """The ExternalCredentials class allows the connection to use EXTERNAL authentication, generally
    with a client SSL certificate.
    """

    TYPE = 'EXTERNAL'

    def __init__(self) -> None:
        """Create a new instance of ExternalCredentials."""
        self.erase_on_connect = False

    @override
    def __eq__(self, other) -> bool:
        if isinstance(other, ExternalCredentials):
            return self.erase_on_connect == other.erase_on_connect
        return NotImplemented

    @override
    def __ne__(self, other) -> bool:
        result = self.__eq__(other)
        if result is not NotImplemented:
            return not result
        return NotImplemented

    def response_for(
            self, start: Connection.Start) -> tuple[str | None, bytes | None]:
        """
        Validate that this type of authentication is supported.

        :param start: Connection.Start method
        """
        if as_bytes(ExternalCredentials.TYPE) not in as_bytes(
                start.mechanisms).split():
            return None, None
        return ExternalCredentials.TYPE, b''

    def erase_credentials(self) -> None:
        """Called by Connection when it no longer needs the credentials."""
        LOGGER.debug('Not supported by this Credentials type')

__init__

__init__() -> None

Create a new instance of ExternalCredentials.

Source code in pika/credentials.py

def __init__(self) -> None:
    """Create a new instance of ExternalCredentials."""
    self.erase_on_connect = False

erase_credentials

erase_credentials() -> None

Called by Connection when it no longer needs the credentials.

Source code in pika/credentials.py

def erase_credentials(self) -> None:
    """Called by Connection when it no longer needs the credentials."""
    LOGGER.debug('Not supported by this Credentials type')

response_for

response_for(
    start: Start,
) -> tuple[str | None, bytes | None]

Validate that this type of authentication is supported.

PARAMETER	DESCRIPTION
start	Connection.Start method TYPE: Start

Source code in pika/credentials.py

def response_for(
        self, start: Connection.Start) -> tuple[str | None, bytes | None]:
    """
    Validate that this type of authentication is supported.

    :param start: Connection.Start method
    """
    if as_bytes(ExternalCredentials.TYPE) not in as_bytes(
            start.mechanisms).split():
        return None, None
    return ExternalCredentials.TYPE, b''

PlainCredentials

A credentials object for the default authentication methodology with RabbitMQ.

If you do not pass in credentials to the ConnectionParameters object, it will create credentials for 'guest' with the password of 'guest'.

If you pass True to erase_on_connect the credentials will not be stored in memory after the Connection attempt has been made.

PARAMETER	DESCRIPTION
username	The username to authenticate with TYPE: str
password	The password to authenticate with TYPE: str
erase_on_connect	erase credentials on connect. TYPE: bool DEFAULT: False

Source code in pika/credentials.py

class PlainCredentials:
    """
    A credentials object for the default authentication methodology with RabbitMQ.

    If you do not pass in credentials to the ConnectionParameters object, it will create credentials
    for 'guest' with the password of 'guest'.

    If you pass True to erase_on_connect the credentials will not be stored in memory after the
    Connection attempt has been made.

    :param username: The username to authenticate with
    :param password: The password to authenticate with
    :param erase_on_connect: erase credentials on connect.
    """

    TYPE = 'PLAIN'

    def __init__(self,
                 username: str,
                 password: str,
                 erase_on_connect: bool = False) -> None:
        """
        Create a new instance of PlainCredentials.

        :param username: The username to authenticate with
        :param password: The password to authenticate with
        :param erase_on_connect: erase credentials on connect.
        """
        self.username: str | None = username
        self.password: str | None = password
        self.erase_on_connect: bool = erase_on_connect

    @override
    def __eq__(self, other) -> bool:
        if isinstance(other, PlainCredentials):
            return (self.username == other.username and
                    self.password == other.password and
                    self.erase_on_connect == other.erase_on_connect)
        return NotImplemented

    @override
    def __ne__(self, other) -> bool:
        result = self.__eq__(other)
        if result is not NotImplemented:
            return not result
        return NotImplemented

    def response_for(
            self, start: Connection.Start) -> tuple[str | None, bytes | None]:
        """
        Validate that this type of authentication is supported.

        :param start: Connection.Start method
        """
        if as_bytes(PlainCredentials.TYPE) not in as_bytes(
                start.mechanisms).split():
            return None, None
        return (PlainCredentials.TYPE, b'\0' + as_bytes(self.username or '') +
                b'\0' + as_bytes(self.password or ''))

    def erase_credentials(self) -> None:
        """Called by Connection when it no longer needs the credentials."""
        if self.erase_on_connect:
            LOGGER.info('Erasing stored credential values')
            self.username = None
            self.password = None

__init__

__init__(
    username: str,
    password: str,
    erase_on_connect: bool = False,
) -> None

Create a new instance of PlainCredentials.

PARAMETER	DESCRIPTION
username	The username to authenticate with TYPE: str
password	The password to authenticate with TYPE: str
erase_on_connect	erase credentials on connect. TYPE: bool DEFAULT: False

Source code in pika/credentials.py

def __init__(self,
             username: str,
             password: str,
             erase_on_connect: bool = False) -> None:
    """
    Create a new instance of PlainCredentials.

    :param username: The username to authenticate with
    :param password: The password to authenticate with
    :param erase_on_connect: erase credentials on connect.
    """
    self.username: str | None = username
    self.password: str | None = password
    self.erase_on_connect: bool = erase_on_connect

erase_credentials

erase_credentials() -> None

Called by Connection when it no longer needs the credentials.

Source code in pika/credentials.py

def erase_credentials(self) -> None:
    """Called by Connection when it no longer needs the credentials."""
    if self.erase_on_connect:
        LOGGER.info('Erasing stored credential values')
        self.username = None
        self.password = None

response_for

response_for(
    start: Start,
) -> tuple[str | None, bytes | None]

Validate that this type of authentication is supported.

PARAMETER	DESCRIPTION
start	Connection.Start method TYPE: Start

Source code in pika/credentials.py

def response_for(
        self, start: Connection.Start) -> tuple[str | None, bytes | None]:
    """
    Validate that this type of authentication is supported.

    :param start: Connection.Start method
    """
    if as_bytes(PlainCredentials.TYPE) not in as_bytes(
            start.mechanisms).split():
        return None, None
    return (PlainCredentials.TYPE, b'\0' + as_bytes(self.username or '') +
            b'\0' + as_bytes(self.password or ''))

PlainCredentials

PlainCredentials

A credentials object for the default authentication methodology with RabbitMQ.

If you do not pass in credentials to the ConnectionParameters object, it will create credentials for 'guest' with the password of 'guest'.

If you pass True to erase_on_connect the credentials will not be stored in memory after the Connection attempt has been made.

PARAMETER	DESCRIPTION
username	The username to authenticate with TYPE: str
password	The password to authenticate with TYPE: str
erase_on_connect	erase credentials on connect. TYPE: bool DEFAULT: False

Source code in pika/credentials.py

class PlainCredentials:
    """
    A credentials object for the default authentication methodology with RabbitMQ.

    If you do not pass in credentials to the ConnectionParameters object, it will create credentials
    for 'guest' with the password of 'guest'.

    If you pass True to erase_on_connect the credentials will not be stored in memory after the
    Connection attempt has been made.

    :param username: The username to authenticate with
    :param password: The password to authenticate with
    :param erase_on_connect: erase credentials on connect.
    """

    TYPE = 'PLAIN'

    def __init__(self,
                 username: str,
                 password: str,
                 erase_on_connect: bool = False) -> None:
        """
        Create a new instance of PlainCredentials.

        :param username: The username to authenticate with
        :param password: The password to authenticate with
        :param erase_on_connect: erase credentials on connect.
        """
        self.username: str | None = username
        self.password: str | None = password
        self.erase_on_connect: bool = erase_on_connect

    @override
    def __eq__(self, other) -> bool:
        if isinstance(other, PlainCredentials):
            return (self.username == other.username and
                    self.password == other.password and
                    self.erase_on_connect == other.erase_on_connect)
        return NotImplemented

    @override
    def __ne__(self, other) -> bool:
        result = self.__eq__(other)
        if result is not NotImplemented:
            return not result
        return NotImplemented

    def response_for(
            self, start: Connection.Start) -> tuple[str | None, bytes | None]:
        """
        Validate that this type of authentication is supported.

        :param start: Connection.Start method
        """
        if as_bytes(PlainCredentials.TYPE) not in as_bytes(
                start.mechanisms).split():
            return None, None
        return (PlainCredentials.TYPE, b'\0' + as_bytes(self.username or '') +
                b'\0' + as_bytes(self.password or ''))

    def erase_credentials(self) -> None:
        """Called by Connection when it no longer needs the credentials."""
        if self.erase_on_connect:
            LOGGER.info('Erasing stored credential values')
            self.username = None
            self.password = None

TYPE class-attribute instance-attribute

TYPE = 'PLAIN'

erase_on_connect instance-attribute

erase_on_connect: bool = erase_on_connect

password instance-attribute

password: str | None = password

username instance-attribute

username: str | None = username

__eq__

__eq__(other) -> bool

Source code in pika/credentials.py

@override
def __eq__(self, other) -> bool:
    if isinstance(other, PlainCredentials):
        return (self.username == other.username and
                self.password == other.password and
                self.erase_on_connect == other.erase_on_connect)
    return NotImplemented

__init__

__init__(
    username: str,
    password: str,
    erase_on_connect: bool = False,
) -> None

Create a new instance of PlainCredentials.

PARAMETER	DESCRIPTION
username	The username to authenticate with TYPE: str
password	The password to authenticate with TYPE: str
erase_on_connect	erase credentials on connect. TYPE: bool DEFAULT: False

Source code in pika/credentials.py

def __init__(self,
             username: str,
             password: str,
             erase_on_connect: bool = False) -> None:
    """
    Create a new instance of PlainCredentials.

    :param username: The username to authenticate with
    :param password: The password to authenticate with
    :param erase_on_connect: erase credentials on connect.
    """
    self.username: str | None = username
    self.password: str | None = password
    self.erase_on_connect: bool = erase_on_connect

__ne__

__ne__(other) -> bool

Source code in pika/credentials.py

@override
def __ne__(self, other) -> bool:
    result = self.__eq__(other)
    if result is not NotImplemented:
        return not result
    return NotImplemented

erase_credentials

erase_credentials() -> None

Called by Connection when it no longer needs the credentials.

Source code in pika/credentials.py

def erase_credentials(self) -> None:
    """Called by Connection when it no longer needs the credentials."""
    if self.erase_on_connect:
        LOGGER.info('Erasing stored credential values')
        self.username = None
        self.password = None

response_for

response_for(
    start: Start,
) -> tuple[str | None, bytes | None]

Validate that this type of authentication is supported.

PARAMETER	DESCRIPTION
start	Connection.Start method TYPE: Start

Source code in pika/credentials.py

def response_for(
        self, start: Connection.Start) -> tuple[str | None, bytes | None]:
    """
    Validate that this type of authentication is supported.

    :param start: Connection.Start method
    """
    if as_bytes(PlainCredentials.TYPE) not in as_bytes(
            start.mechanisms).split():
        return None, None
    return (PlainCredentials.TYPE, b'\0' + as_bytes(self.username or '') +
            b'\0' + as_bytes(self.password or ''))

ExternalCredentials

ExternalCredentials

The ExternalCredentials class allows the connection to use EXTERNAL authentication, generally with a client SSL certificate.

Source code in pika/credentials.py

class ExternalCredentials:
    """The ExternalCredentials class allows the connection to use EXTERNAL authentication, generally
    with a client SSL certificate.
    """

    TYPE = 'EXTERNAL'

    def __init__(self) -> None:
        """Create a new instance of ExternalCredentials."""
        self.erase_on_connect = False

    @override
    def __eq__(self, other) -> bool:
        if isinstance(other, ExternalCredentials):
            return self.erase_on_connect == other.erase_on_connect
        return NotImplemented

    @override
    def __ne__(self, other) -> bool:
        result = self.__eq__(other)
        if result is not NotImplemented:
            return not result
        return NotImplemented

    def response_for(
            self, start: Connection.Start) -> tuple[str | None, bytes | None]:
        """
        Validate that this type of authentication is supported.

        :param start: Connection.Start method
        """
        if as_bytes(ExternalCredentials.TYPE) not in as_bytes(
                start.mechanisms).split():
            return None, None
        return ExternalCredentials.TYPE, b''

    def erase_credentials(self) -> None:
        """Called by Connection when it no longer needs the credentials."""
        LOGGER.debug('Not supported by this Credentials type')

TYPE class-attribute instance-attribute

TYPE = 'EXTERNAL'

erase_on_connect instance-attribute

erase_on_connect = False

__eq__

__eq__(other) -> bool

Source code in pika/credentials.py

@override
def __eq__(self, other) -> bool:
    if isinstance(other, ExternalCredentials):
        return self.erase_on_connect == other.erase_on_connect
    return NotImplemented

__init__

__init__() -> None

Create a new instance of ExternalCredentials.

Source code in pika/credentials.py

def __init__(self) -> None:
    """Create a new instance of ExternalCredentials."""
    self.erase_on_connect = False

__ne__

__ne__(other) -> bool

Source code in pika/credentials.py

@override
def __ne__(self, other) -> bool:
    result = self.__eq__(other)
    if result is not NotImplemented:
        return not result
    return NotImplemented

erase_credentials

erase_credentials() -> None

Called by Connection when it no longer needs the credentials.

Source code in pika/credentials.py

def erase_credentials(self) -> None:
    """Called by Connection when it no longer needs the credentials."""
    LOGGER.debug('Not supported by this Credentials type')

response_for

response_for(
    start: Start,
) -> tuple[str | None, bytes | None]

Validate that this type of authentication is supported.

PARAMETER	DESCRIPTION
start	Connection.Start method TYPE: Start

Source code in pika/credentials.py

def response_for(
        self, start: Connection.Start) -> tuple[str | None, bytes | None]:
    """
    Validate that this type of authentication is supported.

    :param start: Connection.Start method
    """
    if as_bytes(ExternalCredentials.TYPE) not in as_bytes(
            start.mechanisms).split():
        return None, None
    return ExternalCredentials.TYPE, b''