tessl/pypi-pycognito

Python class to integrate Boto3's Cognito client so it is easy to login users with SRP support.

—

Pending

Overview

Eval results

Files

HTTP Integration

Name: tessl/pypi-pycognito
Author: tessl

Requests authentication plugin for seamless integration with HTTP clients, automatically handling Cognito token management and authentication headers. The RequestsSrpAuth class provides automatic authentication for HTTP requests using Cognito tokens.

Capabilities

RequestsSrpAuth Class

A Requests authentication plugin that automatically populates Authorization headers with Cognito tokens and handles token renewal.

class RequestsSrpAuth(requests.auth.AuthBase):
    """
    Requests Auth Plugin for automatic Cognito token authentication.
    
    Automatically handles:
    - Initial authentication if no tokens present
    - Token expiration checking and renewal
    - Authorization header population
    - Token type selection (ID or Access token)
    """
    
    def __init__(self, username: str = None, password: str = None, 
                 user_pool_id: str = None, user_pool_region: str = None,
                 client_id: str = None, cognito: Cognito = None,
                 http_header: str = "Authorization", http_header_prefix: str = "Bearer ",
                 auth_token_type: TokenType = TokenType.ACCESS_TOKEN,
                 boto3_client_kwargs: dict = None):
        """
        Initialize RequestsSrpAuth plugin.
        
        Args:
            username (str, optional): Cognito username (required if cognito not provided)
            password (str, optional): User password (required if cognito not provided)
            user_pool_id (str, optional): User pool ID (required if cognito not provided)
            user_pool_region (str, optional): AWS region (required if cognito not provided)
            client_id (str, optional): Client ID (required if cognito not provided)
            cognito (Cognito, optional): Pre-configured Cognito instance
            http_header (str): HTTP header name for token (default: "Authorization")  
            http_header_prefix (str): Token prefix (default: "Bearer ")
            auth_token_type (TokenType): Token type to use (ACCESS_TOKEN or ID_TOKEN)
            boto3_client_kwargs (dict, optional): Additional boto3 client arguments
            
        Usage Modes:
            1. Individual parameters: Provide username, password, user_pool_id, etc.
            2. Pre-configured Cognito: Provide existing Cognito instance
            
        Note:
            Either provide individual authentication parameters OR a Cognito instance,
            not both.
        """

Usage Example:

import requests
from pycognito.utils import RequestsSrpAuth, TokenType

# Method 1: Using individual parameters
auth = RequestsSrpAuth(
    username='user@example.com',
    password='password',
    user_pool_id='us-east-1_example123',
    client_id='your-client-id',
    user_pool_region='us-east-1'
)

# Method 2: Using pre-configured Cognito instance
from pycognito import Cognito

cognito_instance = Cognito(
    user_pool_id='us-east-1_example123',
    client_id='your-client-id',
    username='user@example.com'
)
cognito_instance.authenticate(password='password')

auth = RequestsSrpAuth(cognito=cognito_instance)

# Method 3: Custom configuration 
auth = RequestsSrpAuth(
    username='user@example.com',
    password='password',
    user_pool_id='us-east-1_example123',
    client_id='your-client-id',
    user_pool_region='us-east-1',
    http_header='X-Auth-Token',
    http_header_prefix='Token ',
    auth_token_type=TokenType.ID_TOKEN
)

Automatic Authentication

The auth plugin automatically handles authentication and token management during HTTP requests.

def __call__(self, request: requests.Request) -> requests.Request:
    """
    Process request and add authentication header.
    
    Args:
        request (requests.Request): HTTP request to authenticate
        
    Returns:
        requests.Request: Request with authentication header added
        
    Automatic Actions:
        1. Check if user is authenticated (has access token)
        2. If not authenticated, perform initial authentication
        3. Check if current token is expired
        4. If expired, automatically renew using refresh token
        5. Add appropriate token to specified HTTP header
        6. Return modified request
        
    Header Format:
        {http_header}: {http_header_prefix}{token}
        
    Default:
        Authorization: Bearer {access_token}
    """

Usage Example:

import requests
from pycognito.utils import RequestsSrpAuth

# Set up authentication
auth = RequestsSrpAuth(
    username='user@example.com',
    password='password',
    user_pool_id='us-east-1_example123',
    client_id='your-client-id',
    user_pool_region='us-east-1'
)

# Make authenticated requests - token handling is automatic
response = requests.get('https://api.example.com/protected', auth=auth)

if response.status_code == 200:
    print("Request successful!")
    print(response.json())
else:
    print(f"Request failed: {response.status_code}")

Token Type Selection

Choose between ID tokens and access tokens for different use cases.

class TokenType(str, Enum):
    """Token types for authentication headers."""
    ID_TOKEN = "id_token"
    ACCESS_TOKEN = "access_token"

Usage Example:

from pycognito.utils import RequestsSrpAuth, TokenType

# Use access token (default - for API authorization)
api_auth = RequestsSrpAuth(
    username='user@example.com',
    password='password',
    user_pool_id='us-east-1_example123',
    client_id='your-client-id',
    auth_token_type=TokenType.ACCESS_TOKEN
)

# Use ID token (for user identity/profile information)
identity_auth = RequestsSrpAuth(
    username='user@example.com',
    password='password',
    user_pool_id='us-east-1_example123',
    client_id='your-client-id',
    auth_token_type=TokenType.ID_TOKEN
)

# API calls
api_response = requests.get('https://api.example.com/data', auth=api_auth)

# Identity/profile calls  
profile_response = requests.get('https://api.example.com/profile', auth=identity_auth)

Usage Patterns

Basic API Client with Authentication

import requests
from pycognito.utils import RequestsSrpAuth

class APIClient:
    """API client with automatic Cognito authentication."""
    
    def __init__(self, base_url, username, password, user_pool_id, client_id, region):
        self.base_url = base_url
        
        # Set up authentication
        self.auth = RequestsSrpAuth(
            username=username,
            password=password,
            user_pool_id=user_pool_id,
            client_id=client_id,
            user_pool_region=region
        )
    
    def get(self, endpoint, **kwargs):
        """Make authenticated GET request."""
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        return requests.get(url, auth=self.auth, **kwargs)
    
    def post(self, endpoint, data=None, json=None, **kwargs):
        """Make authenticated POST request."""
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        return requests.post(url, data=data, json=json, auth=self.auth, **kwargs)
    
    def put(self, endpoint, data=None, json=None, **kwargs):
        """Make authenticated PUT request."""
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        return requests.put(url, data=data, json=json, auth=self.auth, **kwargs)
    
    def delete(self, endpoint, **kwargs):
        """Make authenticated DELETE request."""
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        return requests.delete(url, auth=self.auth, **kwargs)

# Usage
client = APIClient(
    base_url='https://api.example.com',
    username='user@example.com',
    password='password',
    user_pool_id='us-east-1_example123',
    client_id='your-client-id',
    region='us-east-1'
)

# All requests automatically include authentication
users = client.get('/users')
user_data = client.post('/users', json={'name': 'John Doe'})

Session-Based HTTP Client

import requests
from pycognito.utils import RequestsSrpAuth

def create_authenticated_session(username, password, user_pool_id, client_id, region):
    """Create requests Session with automatic Cognito authentication."""
    
    # Create session
    session = requests.Session()
    
    # Set up authentication
    session.auth = RequestsSrpAuth(
        username=username,
        password=password,
        user_pool_id=user_pool_id,
        client_id=client_id,
        user_pool_region=region
    )
    
    # Optional: Set common headers
    session.headers.update({
        'Content-Type': 'application/json',
        'User-Agent': 'MyApp/1.0'
    })
    
    return session

# Usage
session = create_authenticated_session(
    username='user@example.com',
    password='password',
    user_pool_id='us-east-1_example123',
    client_id='your-client-id',
    region='us-east-1'
)

# All requests in this session are automatically authenticated
response1 = session.get('https://api.example.com/endpoint1')
response2 = session.post('https://api.example.com/endpoint2', json={'data': 'value'})
response3 = session.put('https://api.example.com/endpoint3', json={'update': 'value'})

Custom Header Authentication

from pycognito.utils import RequestsSrpAuth, TokenType

# Custom header for legacy APIs
custom_auth = RequestsSrpAuth(
    username='user@example.com',
    password='password',
    user_pool_id='us-east-1_example123',
    client_id='your-client-id',
    user_pool_region='us-east-1',
    http_header='X-API-Key',
    http_header_prefix='',  # No prefix
    auth_token_type=TokenType.ACCESS_TOKEN
)

# API expects: X-API-Key: {access_token}
response = requests.get('https://legacy-api.example.com/data', auth=custom_auth)

# JWT Bearer token for modern APIs
jwt_auth = RequestsSrpAuth(
    username='user@example.com', 
    password='password',
    user_pool_id='us-east-1_example123',
    client_id='your-client-id',
    user_pool_region='us-east-1',
    http_header='Authorization',
    http_header_prefix='JWT ',
    auth_token_type=TokenType.ID_TOKEN
)

# API expects: Authorization: JWT {id_token}
response = requests.get('https://jwt-api.example.com/profile', auth=jwt_auth)

Error Handling and Retry Logic

import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
from pycognito.utils import RequestsSrpAuth

class RobustAPIClient:
    """API client with retry logic and error handling."""
    
    def __init__(self, base_url, username, password, user_pool_id, client_id, region):
        self.base_url = base_url
        
        # Create session with retry strategy
        self.session = requests.Session()
        
        # Configure retry strategy
        retry_strategy = Retry(
            total=3,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504],
        )
        
        adapter = HTTPAdapter(max_retries=retry_strategy)
        self.session.mount("http://", adapter)
        self.session.mount("https://", adapter)
        
        # Set up authentication
        self.session.auth = RequestsSrpAuth(
            username=username,
            password=password,
            user_pool_id=user_pool_id,
            client_id=client_id,
            user_pool_region=region
        )
    
    def make_request(self, method, endpoint, **kwargs):
        """Make request with comprehensive error handling."""
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        
        try:
            response = self.session.request(method, url, **kwargs)
            response.raise_for_status()
            return response
            
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 401:
                print("Authentication failed - check credentials")
            elif e.response.status_code == 403:
                print("Access forbidden - check permissions")
            else:
                print(f"HTTP error: {e}")
            raise
            
        except requests.exceptions.ConnectionError:
            print("Connection error - check network")
            raise
            
        except requests.exceptions.Timeout:
            print("Request timeout - server may be overloaded")
            raise
            
        except requests.exceptions.RequestException as e:
            print(f"Request failed: {e}")
            raise

# Usage
client = RobustAPIClient(
    base_url='https://api.example.com',
    username='user@example.com',
    password='password',
    user_pool_id='us-east-1_example123',
    client_id='your-client-id',
    region='us-east-1'
)

try:
    response = client.make_request('GET', '/protected-data')
    data = response.json()
    print(f"Success: {data}")
except Exception as e:
    print(f"Request failed: {e}")

Multi-Environment Configuration

from pycognito.utils import RequestsSrpAuth
import os

class MultiEnvironmentAuth:
    """Authentication manager for multiple environments."""
    
    ENVIRONMENTS = {
        'dev': {
            'user_pool_id': 'us-east-1_dev123',
            'client_id': 'dev-client-id',
            'region': 'us-east-1',
            'base_url': 'https://dev-api.example.com'
        },
        'staging': {
            'user_pool_id': 'us-east-1_staging456',
            'client_id': 'staging-client-id', 
            'region': 'us-east-1',
            'base_url': 'https://staging-api.example.com'
        },
        'prod': {
            'user_pool_id': 'us-east-1_prod789',
            'client_id': 'prod-client-id',
            'region': 'us-east-1',
            'base_url': 'https://api.example.com'
        }
    }
    
    def __init__(self, environment='dev'):
        self.env = environment
        self.config = self.ENVIRONMENTS[environment]
    
    def create_auth(self, username, password):
        """Create auth for specified environment."""
        return RequestsSrpAuth(
            username=username,
            password=password,
            user_pool_id=self.config['user_pool_id'],
            client_id=self.config['client_id'],
            user_pool_region=self.config['region']
        )
    
    def make_request(self, method, endpoint, username, password, **kwargs):
        """Make authenticated request to environment."""
        auth = self.create_auth(username, password)
        url = f"{self.config['base_url']}/{endpoint.lstrip('/')}"
        
        return requests.request(method, url, auth=auth, **kwargs)

# Usage
# Development environment
dev_auth = MultiEnvironmentAuth('dev')
dev_response = dev_auth.make_request(
    'GET', '/users',
    username='dev-user@example.com',
    password='dev-password'
)

# Production environment  
prod_auth = MultiEnvironmentAuth('prod')
prod_response = prod_auth.make_request(
    'GET', '/users',
    username='prod-user@example.com', 
    password='prod-password'
)

# Environment from environment variable
env = os.getenv('API_ENVIRONMENT', 'dev')
auth_manager = MultiEnvironmentAuth(env)

Async HTTP Integration

import asyncio
import aiohttp
from pycognito.utils import RequestsSrpAuth, TokenType

class AsyncCognitoAuth:
    """Async-compatible Cognito authentication."""
    
    def __init__(self, username, password, user_pool_id, client_id, region):
        # Initialize sync auth to get tokens
        self.sync_auth = RequestsSrpAuth(
            username=username,
            password=password,
            user_pool_id=user_pool_id,
            client_id=client_id,
            user_pool_region=region,
            auth_token_type=TokenType.ACCESS_TOKEN
        )
        
        # Perform initial authentication
        import requests
        dummy_request = requests.Request('GET', 'http://example.com')
        self.sync_auth(dummy_request)  # This triggers authentication
    
    def get_headers(self):
        """Get authentication headers for async requests."""
        token = getattr(self.sync_auth.cognito_client, self.sync_auth.token_type.value)
        return {
            self.sync_auth.http_header: f"{self.sync_auth.http_header_prefix}{token}"
        }
    
    def refresh_if_needed(self):
        """Refresh token if needed (sync operation)."""
        self.sync_auth.cognito_client.check_token(renew=True)

async def async_api_client():
    """Example async API client with Cognito authentication."""
    
    # Set up authentication
    auth = AsyncCognitoAuth(
        username='user@example.com',
        password='password',
        user_pool_id='us-east-1_example123',
        client_id='your-client-id',
        region='us-east-1'
    )
    
    # Make async requests
    async with aiohttp.ClientSession() as session:
        
        # Refresh token if needed before making requests
        auth.refresh_if_needed()
        headers = auth.get_headers()
        
        # Make multiple concurrent requests
        tasks = []
        for i in range(5):
            task = session.get(
                f'https://api.example.com/data/{i}',
                headers=headers
            )
            tasks.append(task)
        
        responses = await asyncio.gather(*tasks)
        
        # Process responses
        for i, response in enumerate(responses):
            if response.status == 200:
                data = await response.json()
                print(f"Request {i} successful: {data}")
            else:
                print(f"Request {i} failed: {response.status}")

# Usage
asyncio.run(async_api_client())

Install with Tessl CLI