/usr/lib/python3/dist-packages/pika/credentials.py is in python3-pika 0.11.0-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 | """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.
"""
from .compat import as_bytes
import logging
LOGGER = logging.getLogger(__name__)
class PlainCredentials(object):
"""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 str username: The username to authenticate with
:param str password: The password to authenticate with
:param bool erase_on_connect: erase credentials on connect.
"""
TYPE = 'PLAIN'
def __init__(self, username, password, erase_on_connect=False):
"""Create a new instance of PlainCredentials
:param str username: The username to authenticate with
:param str password: The password to authenticate with
:param bool erase_on_connect: erase credentials on connect.
"""
self.username = username
self.password = password
self.erase_on_connect = erase_on_connect
def __eq__(self, other):
return (isinstance(other, PlainCredentials) and
other.username == self.username and
other.password == self.password and
other.erase_on_connect == self.erase_on_connect)
def __ne__(self, other):
return not self == other
def response_for(self, start):
"""Validate that this type of authentication is supported
:param spec.Connection.Start start: Connection.Start method
:rtype: tuple(str|None, str|None)
"""
if as_bytes(PlainCredentials.TYPE) not in\
as_bytes(start.mechanisms).split():
return None, None
return (PlainCredentials.TYPE,
b'\0' + as_bytes(self.username) +
b'\0' + as_bytes(self.password))
def erase_credentials(self):
"""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
class ExternalCredentials(object):
"""The ExternalCredentials class allows the connection to use EXTERNAL
authentication, generally with a client SSL certificate.
"""
TYPE = 'EXTERNAL'
def __init__(self):
"""Create a new instance of ExternalCredentials"""
self.erase_on_connect = False
def __eq__(self, other):
return (isinstance(other, ExternalCredentials) and
other.erase_on_connect == self.erase_on_connect)
def __ne__(self, other):
return not self == other
def response_for(self, start):
"""Validate that this type of authentication is supported
:param spec.Connection.Start start: Connection.Start method
:rtype: tuple(str or None, str or None)
"""
if as_bytes(ExternalCredentials.TYPE) not in\
as_bytes(start.mechanisms).split():
return None, None
return ExternalCredentials.TYPE, b''
def erase_credentials(self):
"""Called by Connection when it no longer needs the credentials"""
LOGGER.debug('Not supported by this Credentials type')
# Append custom credential types to this list for validation support
VALID_TYPES = [PlainCredentials, ExternalCredentials]
|