This document is for py-amqp's development version, which can be significantly different from previous releases. Get the stable docs here: 5.3.

# Source code for amqp.sasl

    """SASL mechanisms for AMQP authentication."""

    import socket
    import warnings
    from io import BytesIO

    from amqp.serialization import _write_table

    class SASL:
        """The base class for all amqp SASL authentication mechanisms.

        You should sub-class this if you're implementing your own authentication.
        """

        @property
        def mechanism(self):
            """Return a bytes containing the SASL mechanism name."""
            raise NotImplementedError

        def start(self, connection):
            """Return the first response to a SASL challenge as a bytes object."""
            raise NotImplementedError

    class PLAIN(SASL):
        """PLAIN SASL authentication mechanism.

        See https://tools.ietf.org/html/rfc4616 for details
        """

        mechanism = b'PLAIN'

        def __init__(self, username, password):
            self.username, self.password = username, password

        __slots__ = (
            "username",
            "password",
            )

        def start(self, connection):
            if self.username is None or self.password is None:
                return NotImplemented
            login_response = BytesIO()
            login_response.write(b'\0')
            login_response.write(self.username.encode('utf-8'))
            login_response.write(b'\0')
            login_response.write(self.password.encode('utf-8'))
            return login_response.getvalue()

    class AMQPLAIN(SASL):
        """AMQPLAIN SASL authentication mechanism.

        This is a non-standard mechanism used by AMQP servers.
        """

        mechanism = b'AMQPLAIN'

        def __init__(self, username, password):
            self.username, self.password = username, password

        __slots__ = (
            "username",
            "password",
            )

        def start(self, connection):
            if self.username is None or self.password is None:
                return NotImplemented
            login_response = BytesIO()
            _write_table({b'LOGIN': self.username, b'PASSWORD': self.password},
                         login_response.write, [])
            # Skip the length at the beginning
            return login_response.getvalue()[4:]

    def _get_gssapi_mechanism():
        try:
            import gssapi
            import gssapi.raw.misc  # Fail if the old python-gssapi is installed
        except ImportError:
            class FakeGSSAPI(SASL):
                """A no-op SASL mechanism for when gssapi isn't available."""

                mechanism = None

                def __init__(self, client_name=None, service=b'amqp',
                             rdns=False, fail_soft=False):
                    if not fail_soft:
                        raise NotImplementedError(
                            "You need to install the `gssapi` module for GSSAPI "
                            "SASL support")

                def start(self):  # pragma: no cover
                    return NotImplemented
            return FakeGSSAPI
        else:
            class GSSAPI(SASL):
                """GSSAPI SASL authentication mechanism.

                See https://tools.ietf.org/html/rfc4752 for details
                """

                mechanism = b'GSSAPI'

                def __init__(self, client_name=None, service=b'amqp',
                             rdns=False, fail_soft=False):
                    if client_name and not isinstance(client_name, bytes):
                        client_name = client_name.encode('ascii')
                    self.client_name = client_name
                    self.fail_soft = fail_soft
                    self.service = service
                    self.rdns = rdns

                __slots__ = (
                    "client_name",
                    "fail_soft",
                    "service",
                    "rdns"
                    )

                def get_hostname(self, connection):
                    sock = connection.transport.sock
                    if self.rdns and sock.family in (socket.AF_INET,
                                                     socket.AF_INET6):
                        peer = sock.getpeername()
                        hostname, _, _ = socket.gethostbyaddr(peer[0])
                    else:
                        hostname = connection.transport.host
                    if not isinstance(hostname, bytes):
                        hostname = hostname.encode('ascii')
                    return hostname

                def start(self, connection):
                    try:
                        if self.client_name:
                            creds = gssapi.Credentials(
                                name=gssapi.Name(self.client_name))
                        else:
                            creds = None
                        hostname = self.get_hostname(connection)
                        name = gssapi.Name(b'@'.join([self.service, hostname]),
                                           gssapi.NameType.hostbased_service)
                        context = gssapi.SecurityContext(name=name, creds=creds)
                        return context.step(None)
                    except gssapi.raw.misc.GSSError:
                        if self.fail_soft:
                            return NotImplemented
                        else:
                            raise
            return GSSAPI

    GSSAPI = _get_gssapi_mechanism()

    class EXTERNAL(SASL):
        """EXTERNAL SASL mechanism.

        Enables external authentication, i.e. not handled through this protocol.
        Only passes 'EXTERNAL' as authentication mechanism, but no further
        authentication data.
        """

        mechanism = b'EXTERNAL'

        def start(self, connection):
            return b''

    class RAW(SASL):
        """A generic custom SASL mechanism.

        This mechanism takes a mechanism name and response to send to the server,
        so can be used for simple custom authentication schemes.
        """

        mechanism = None

        def __init__(self, mechanism, response):
            assert isinstance(mechanism, bytes)
            assert isinstance(response, bytes)
            self.mechanism, self.response = mechanism, response
            warnings.warn("Passing login_method and login_response to Connection "
                          "is deprecated. Please implement a SASL subclass "
                          "instead.", DeprecationWarning)

        def start(self, connection):
            return self.response