tessl/pypi-pika

Pure Python implementation of the AMQP 0.9.1 protocol including RabbitMQ's extensions

—

Pending

Overview

Eval results

Files

Authentication & Security

Name: tessl/pypi-pika
Author: tessl

Authentication credential management supporting username/password and external authentication methods with SSL/TLS connection security for secure RabbitMQ connections.

Capabilities

Plain Credentials

Username and password authentication for RabbitMQ connections.

class PlainCredentials:
    """Username/password authentication credentials."""
    
    TYPE = 'PLAIN'
    
    def __init__(self, username, password, erase_on_connect=False):
        """
        Create plain credentials.
        
        Parameters:
        - username (str): Authentication username
        - password (str): Authentication password
        - erase_on_connect (bool): If True, erase credentials after connection
        """
    
    def response_for(self, start):
        """
        Generate authentication response for connection start.
        
        Parameters:
        - start: Connection.Start method from server
        
        Returns:
        - tuple: (mechanism, response) or (None, None) if unsupported
        """
    
    def erase_credentials(self):
        """Clear stored credentials from memory."""
    
    @property
    def username(self) -> str:
        """Authentication username."""
    
    @property
    def password(self) -> str:
        """Authentication password."""
    
    @property
    def erase_on_connect(self) -> bool:
        """True if credentials should be erased after connection."""

External Credentials

External authentication using SSL client certificates.

class ExternalCredentials:
    """External authentication credentials (SSL certificates)."""
    
    TYPE = 'EXTERNAL'
    
    def __init__(self):
        """Create external credentials."""
    
    def response_for(self, start):
        """
        Generate authentication response for connection start.
        
        Parameters:
        - start: Connection.Start method from server
        
        Returns:
        - tuple: (mechanism, response) or (None, None) if unsupported
        """
    
    def erase_credentials(self):
        """Clear stored credentials (no-op for external auth)."""
    
    @property
    def erase_on_connect(self) -> bool:
        """Always False for external credentials."""

SSL/TLS Configuration

SSL options for secure connections to RabbitMQ.

class SSLOptions:
    """SSL/TLS connection configuration."""
    
    def __init__(self, context=None, server_hostname=None):
        """
        Create SSL options.
        
        Parameters:
        - context (ssl.SSLContext): SSL context for connection
        - server_hostname (str): Server hostname for SSL verification
        """
    
    @property
    def context(self):
        """SSL context for the connection."""
    
    @property
    def server_hostname(self) -> str:
        """Server hostname for SSL verification."""

Credential Types

Valid credential types supported by pika.

VALID_TYPES = [PlainCredentials, ExternalCredentials]

Usage Examples

Basic Authentication

import pika

# Create credentials
credentials = pika.PlainCredentials('myuser', 'mypassword')

# Use in connection parameters
parameters = pika.ConnectionParameters(
    host='rabbitmq.example.com',
    credentials=credentials
)

connection = pika.BlockingConnection(parameters)

Secure Credentials with Erasure

import pika

# Credentials that are erased after connection
credentials = pika.PlainCredentials(
    'myuser', 
    'mypassword', 
    erase_on_connect=True
)

parameters = pika.ConnectionParameters(
    host='rabbitmq.example.com',
    credentials=credentials
)

connection = pika.BlockingConnection(parameters)
# Credentials are now erased from memory

SSL Connection with Client Certificate

import pika
import ssl

# Create SSL context
context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
context.check_hostname = False
context.verify_mode = ssl.CERT_REQUIRED

# Load client certificate and key
context.load_cert_chain('/path/to/client.crt', '/path/to/client.key')

# Load CA certificate
context.load_verify_locations('/path/to/ca.crt')

# Create SSL options
ssl_options = pika.SSLOptions(context, 'rabbitmq.example.com')

# Use external credentials for certificate-based auth
from pika.credentials import ExternalCredentials
credentials = ExternalCredentials()

parameters = pika.ConnectionParameters(
    host='rabbitmq.example.com',
    port=5671,  # SSL port
    credentials=credentials,
    ssl_options=ssl_options
)

connection = pika.BlockingConnection(parameters)

SSL Connection with Password Auth

import pika
import ssl

# Create SSL context
context = ssl.create_default_context()

# Create SSL options
ssl_options = pika.SSLOptions(context, 'rabbitmq.example.com')

# Use regular credentials over SSL
credentials = pika.PlainCredentials('myuser', 'mypassword')

parameters = pika.ConnectionParameters(
    host='rabbitmq.example.com',
    port=5671,  # SSL port
    credentials=credentials,
    ssl_options=ssl_options
)

connection = pika.BlockingConnection(parameters)

URL with Authentication

import pika

# Username and password in URL
url = 'amqp://myuser:mypassword@rabbitmq.example.com:5672/%2F'
connection = pika.BlockingConnection(pika.URLParameters(url))

# SSL URL with authentication
ssl_url = 'amqps://myuser:mypassword@rabbitmq.example.com:5671/%2F'
connection = pika.BlockingConnection(pika.URLParameters(ssl_url))

Custom SSL Context

import pika
import ssl

# Create custom SSL context
context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
context.minimum_version = ssl.TLSVersion.TLSv1_2
context.set_ciphers('ECDHE+AESGCM:ECDHE+CHACHA20:DHE+AESGCM:DHE+CHACHA20:!aNULL:!MD5:!DSS')

# Disable hostname checking for self-signed certificates
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE

ssl_options = pika.SSLOptions(context)

parameters = pika.ConnectionParameters(
    host='localhost',
    port=5671,
    ssl_options=ssl_options,
    credentials=pika.PlainCredentials('guest', 'guest')
)

connection = pika.BlockingConnection(parameters)

Environment-Based Credentials

import pika
import os

# Get credentials from environment
username = os.getenv('RABBITMQ_USER', 'guest')
password = os.getenv('RABBITMQ_PASS', 'guest')
host = os.getenv('RABBITMQ_HOST', 'localhost')

credentials = pika.PlainCredentials(username, password)
parameters = pika.ConnectionParameters(host=host, credentials=credentials)

connection = pika.BlockingConnection(parameters)

Install with Tessl CLI