Authentication Support

def add_basic_auth(agent, username, password):
    """
    Add HTTP Basic Authentication to an agent.
    
    Creates a wrapper agent that automatically includes Basic Auth
    headers for all requests made through it.
    
    Parameters:
    - agent: IAgent - Twisted agent to wrap
    - username: str or bytes - Username for authentication
    - password: str or bytes - Password for authentication
    
    Returns:
    IAgent - Wrapped agent with Basic Auth support
    """

def add_auth(agent, auth_config):
    """
    Add authentication to an agent based on configuration.
    
    Currently supports HTTP Basic Authentication via tuple format.
    
    Parameters:
    - agent: IAgent - Twisted agent to wrap  
    - auth_config: tuple - Authentication configuration (username, password)
    
    Returns:
    IAgent - Agent with authentication configured
    
    Raises:
    UnknownAuthConfig - If auth_config format is not recognized
    """

class _RequestHeaderSetterAgent:
    """
    Wrap an agent to set request headers on all requests.
    
    This is used internally by authentication functions but can be
    used directly for custom header requirements.
    """
    
    def __init__(self, agent, headers):
        """
        Initialize header-setting agent wrapper.
        
        Parameters:
        - agent: IAgent - Agent to wrap
        - headers: Headers - Headers to add to each request
        """
    
    def request(self, method, uri, headers=None, bodyProducer=None):
        """
        Make request with automatic header injection.
        
        Merges the configured headers with any headers provided
        in the request, with request headers taking precedence.
        """

class UnknownAuthConfig(Exception):
    """
    Exception raised when authentication config cannot be interpreted.
    
    Raised by add_auth() when the auth_config parameter is not
    in a recognized format.
    """

import treq
from twisted.internet import defer

@defer.inlineCallbacks
def basic_auth_requests():
    # Using auth parameter (recommended for simple cases)
    response = yield treq.get(
        'https://httpbin.org/basic-auth/user/pass',
        auth=('user', 'pass')
    )
    
    # Verify authentication worked
    data = yield response.json()
    print("Authenticated user:", data['user'])
    
    # POST with authentication
    response = yield treq.post(
        'https://httpbin.org/basic-auth/api/key',
        auth=('api', 'key'),
        json={'data': 'sensitive information'}
    )

from twisted.web.client import Agent
from treq.auth import add_basic_auth
from treq.client import HTTPClient
from twisted.internet import reactor

@defer.inlineCallbacks
def persistent_auth():
    # Create base agent
    base_agent = Agent(reactor)
    
    # Wrap with authentication
    auth_agent = add_basic_auth(base_agent, 'username', 'password')
    
    # Create client using authenticated agent
    client = HTTPClient(auth_agent)
    
    # All requests through this client will be authenticated
    response1 = yield client.get('https://api.example.com/profile')
    response2 = yield client.post('https://api.example.com/data', json={'key': 'value'})
    response3 = yield client.put('https://api.example.com/resource/123')
    
    # No need to specify auth for each request

from treq.auth import add_basic_auth, add_auth
from twisted.web.client import Agent

@defer.inlineCallbacks
def multiple_auth_schemes():
    base_agent = Agent(reactor)
    
    # API with Basic Auth
    api_agent = add_basic_auth(base_agent, 'api_user', 'api_key')
    api_client = HTTPClient(api_agent)
    
    # Admin interface with different credentials
    admin_agent = add_auth(base_agent, ('admin', 'admin_password'))
    admin_client = HTTPClient(admin_agent)
    
    # Use appropriate client for each service
    api_data = yield api_client.get('https://api.service.com/data')
    admin_data = yield admin_client.get('https://admin.service.com/users')

from treq.auth import _RequestHeaderSetterAgent
from twisted.web.http_headers import Headers
from twisted.web.client import Agent

@defer.inlineCallbacks
def custom_header_auth():
    # Create headers for API key authentication
    auth_headers = Headers({
        'Authorization': ['Bearer your-api-token'],
        'X-API-Key': ['your-api-key']
    })
    
    # Wrap agent with custom headers
    base_agent = Agent(reactor)
    auth_agent = _RequestHeaderSetterAgent(base_agent, auth_headers)
    
    # Create client
    client = HTTPClient(auth_agent)
    
    # All requests include authentication headers
    response = yield client.get('https://api.example.com/protected-resource')

from treq.auth import UnknownAuthConfig

@defer.inlineCallbacks
def handle_auth_errors():
    try:
        # This will raise UnknownAuthConfig
        agent = add_auth(base_agent, "invalid-config")
    except UnknownAuthConfig as e:
        print(f"Authentication configuration error: {e}")
        # Fall back to no authentication or different config
        agent = base_agent
    
    # Handle authentication failures
    try:
        response = yield treq.get(
            'https://httpbin.org/basic-auth/user/pass',
            auth=('wrong', 'credentials')
        )
        if response.code == 401:
            print("Authentication failed - invalid credentials")
        elif response.code == 403:
            print("Authentication succeeded but access forbidden")
    except Exception as e:
        print(f"Request failed: {e}")

from http.cookiejar import CookieJar

@defer.inlineCallbacks
def session_auth():
    # Client with cookie support
    cookiejar = CookieJar()
    client = HTTPClient(Agent(reactor), cookiejar=cookiejar)
    
    # Login with credentials
    login_response = yield client.post(
        'https://example.com/login',
        data={'username': 'user', 'password': 'pass'}
    )
    
    if login_response.code == 200:
        # Session cookie automatically stored
        print("Login successful")
        
        # Subsequent requests use session cookie
        profile = yield client.get('https://example.com/profile')
        data = yield profile.json()
        
        # Logout
        yield client.post('https://example.com/logout')
    else:
        print("Login failed")

@defer.inlineCallbacks
def advanced_auth_patterns():
    # Token refresh pattern
    class TokenRefreshAgent:
        def __init__(self, base_agent, get_token_func):
            self.base_agent = base_agent
            self.get_token = get_token_func
            self.current_token = None
        
        @defer.inlineCallbacks
        def request(self, method, uri, headers=None, bodyProducer=None):
            # Refresh token if needed
            if not self.current_token:
                self.current_token = yield self.get_token()
            
            # Add token to headers
            if headers is None:
                headers = Headers()
            else:
                headers = headers.copy()
            
            headers.addRawHeader('Authorization', f'Bearer {self.current_token}')
            
            response = yield self.base_agent.request(method, uri, headers, bodyProducer)
            
            # Handle token expiration
            if response.code == 401:
                self.current_token = yield self.get_token()
                headers.setRawHeaders('Authorization', [f'Bearer {self.current_token}'])
                response = yield self.base_agent.request(method, uri, headers, bodyProducer)
            
            defer.returnValue(response)

# Basic auth configuration
AuthConfig = Tuple[Union[str, bytes], Union[str, bytes]]

# Agent interface
from twisted.web.iweb import IAgent

# Headers type
from twisted.web.http_headers import Headers

# Username/password types
Username = Union[str, bytes]
Password = Union[str, bytes]