tessl/pypi-h11

A pure-Python, bring-your-own-I/O implementation of HTTP/1.1

—

Pending

Overview

Eval results

Files

Exception Handling

Name: tessl/pypi-h11
Author: tessl

Protocol error hierarchy for handling HTTP/1.1 violations and connection errors. h11 provides structured exceptions to help applications distinguish between local programming errors and remote protocol violations.

Capabilities

Base Protocol Error

Abstract base class for all HTTP/1.1 protocol violations.

class ProtocolError(Exception):
    """
    Base class for HTTP/1.1 protocol violations.
    
    Attributes:
        error_status_hint (int): Suggested HTTP status code for the error (default: 400)
    
    Note:
        Cannot be instantiated directly. Use LocalProtocolError or RemoteProtocolError.
    """
    def __init__(self, msg, error_status_hint=400):
        """
        Initialize protocol error.
        
        Args:
            msg (str): Error description
            error_status_hint (int): Suggested HTTP status code (default: 400)
        """

Local Protocol Errors

Errors caused by the local application violating HTTP/1.1 protocol rules.

class LocalProtocolError(ProtocolError):
    """
    Indicates you tried to do something HTTP/1.1 says is illegal.
    
    Raised when:
    - Sending events in wrong connection state
    - Invalid event parameters or construction
    - Calling methods with invalid arguments
    - Violating HTTP/1.1 message format rules
    """
    def _reraise_as_remote_protocol_error(self):
        """
        Internal method to convert to RemoteProtocolError.
        
        Note:
            Used internally by h11 for symmetric error handling.
            Do not call directly in application code.
        """

Common Causes:

import h11

conn = h11.Connection(h11.CLIENT)

# 1. Wrong state transitions
try:
    # Trying to send Response from CLIENT role
    resp = h11.Response(status_code=200)
    conn.send(resp)
except h11.LocalProtocolError as e:
    print(f"Invalid state transition: {e}")

# 2. Invalid event construction  
try:
    # Invalid status code
    resp = h11.Response(status_code=99)  # Must be 100-999
except h11.LocalProtocolError as e:
    print(f"Invalid response: {e}")

# 3. Malformed headers
try:
    req = h11.Request(
        method=b'GET',
        target=b'/',
        headers=[(b'invalid\x00header', b'value')]  # Null bytes not allowed
    )
except h11.LocalProtocolError as e:
    print(f"Invalid headers: {e}")

Remote Protocol Errors

Errors caused by the remote peer violating HTTP/1.1 protocol rules.

class RemoteProtocolError(ProtocolError):
    """
    Indicates the remote peer violated HTTP/1.1 protocol.
    
    Raised when:
    - Peer sends malformed HTTP messages
    - Invalid header formats or values
    - Protocol state violations by peer
    - Message parsing failures due to invalid input
    """

Common Causes:

import h11

conn = h11.Connection(h11.SERVER)

# Examples of data that would cause RemoteProtocolError:

try:
    # Malformed request line
    conn.receive_data(b'INVALID REQUEST LINE\r\n\r\n')
    event = conn.next_event()
except h11.RemoteProtocolError as e:
    print(f"Peer sent invalid data: {e}")

try:
    # Invalid headers
    conn.receive_data(b'GET / HTTP/1.1\r\nInvalid Header Line\r\n\r\n')
    event = conn.next_event()
except h11.RemoteProtocolError as e:
    print(f"Peer sent malformed headers: {e}")

try:
    # Invalid chunk encoding
    conn.receive_data(b'GET / HTTP/1.1\r\nTransfer-Encoding: chunked\r\n\r\nINVALID_HEX\r\n')
    event = conn.next_event()
except h11.RemoteProtocolError as e:
    print(f"Peer sent invalid chunked encoding: {e}")

Error Handling Patterns

Basic Exception Handling

import h11

def safe_connection_handling(conn, raw_data):
    """Handle connection with proper error handling."""
    try:
        conn.receive_data(raw_data)
        return conn.next_event()
        
    except h11.LocalProtocolError as e:
        # Our code did something wrong
        print(f"Local error - fix the code: {e}")
        print(f"Suggested status: {e.error_status_hint}")
        # Connection is now in ERROR state
        return None
        
    except h11.RemoteProtocolError as e:
        # Peer sent invalid data  
        print(f"Remote error - peer violated protocol: {e}")
        print(f"Suggested status: {e.error_status_hint}")
        # Send error response to peer if possible
        return None

Server Error Response Generation

def handle_client_errors(conn, socket):
    """Handle client protocol errors in server context."""
    try:
        # Process client data
        raw_data = socket.recv(4096)
        conn.receive_data(raw_data)
        event = conn.next_event()
        return event
        
    except h11.RemoteProtocolError as e:
        # Client sent invalid HTTP
        try:
            # Try to send error response
            error_response = h11.Response(
                status_code=e.error_status_hint,
                headers=[
                    (b'content-type', b'text/plain'),
                    (b'connection', b'close')
                ]
            )
            error_data = conn.send(error_response)
            socket.send(error_data)
            
            # Send error body
            body = h11.Data(data=f"Bad Request: {e}".encode())
            body_data = conn.send(body)
            socket.send(body_data)
            
            # End message
            eom = h11.EndOfMessage()
            eom_data = conn.send(eom)
            socket.send(eom_data)
            
        except h11.LocalProtocolError:
            # Connection too broken to send response
            pass
        finally:
            socket.close()
        
        return None

Connection State After Errors

def check_connection_after_error(conn):
    """Check connection state after handling errors."""
    
    if conn.our_state is h11.ERROR:
        print("Connection in ERROR state - cannot continue")
        return "error"
    elif conn.our_state is h11.MUST_CLOSE:
        print("Connection must be closed")
        return "must_close"  
    elif conn.our_state is h11.CLOSED:
        print("Connection already closed")
        return "closed"
    else:
        print(f"Connection state: {conn.our_state}")
        return "ok"

# Usage after exception handling
try:
    event = conn.next_event()
except (h11.LocalProtocolError, h11.RemoteProtocolError) as e:
    status = check_connection_after_error(conn)
    if status in ("error", "must_close", "closed"):
        # Close socket and clean up
        pass

Client Error Handling

def robust_client_request(conn, socket, request):
    """Send request with comprehensive error handling."""
    try:
        # Send request
        data = conn.send(request)
        socket.send(data)
        
        # Send end of message
        eom = h11.EndOfMessage()
        data = conn.send(eom)
        socket.send(data)
        
        return True
        
    except h11.LocalProtocolError as e:
        # Our request was invalid
        print(f"Invalid request format: {e}")
        return False
        
def robust_client_receive(conn, socket):
    """Receive response with error handling."""
    while True:
        try:
            raw_data = socket.recv(4096)
            if not raw_data:
                break
                
            conn.receive_data(raw_data)
            
            while True:
                event = conn.next_event()
                
                if event is h11.NEED_DATA:
                    break
                elif isinstance(event, h11.Response):
                    return event
                elif isinstance(event, h11.ConnectionClosed):
                    return None
                    
        except h11.RemoteProtocolError as e:
            # Server sent invalid response
            print(f"Server protocol error: {e}")
            return None

Error Recovery Strategies

def connection_recovery_strategy(conn, error):
    """Determine recovery strategy based on error type and connection state."""
    
    if isinstance(error, h11.LocalProtocolError):
        # Our code error - fix the bug
        strategy = "fix_code"
    elif isinstance(error, h11.RemoteProtocolError):
        # Peer error - different strategies based on severity
        if "header" in str(error).lower():
            strategy = "send_400_bad_request"
        elif "chunk" in str(error).lower():
            strategy = "send_400_bad_request"  
        else:
            strategy = "close_connection"
    else:
        strategy = "close_connection"
    
    # Check if connection allows recovery
    if conn.our_state is h11.ERROR:
        strategy = "close_connection"
    elif conn.our_state is h11.MUST_CLOSE:
        strategy = "close_after_response"
        
    return strategy

# Usage
try:
    event = conn.next_event()
except (h11.LocalProtocolError, h11.RemoteProtocolError) as e:
    strategy = connection_recovery_strategy(conn, e)
    
    if strategy == "fix_code":
        # Log error for developer attention
        logger.error(f"h11 usage error: {e}")
    elif strategy == "send_400_bad_request":
        # Send error response to client
        send_error_response(conn, socket, 400, str(e))
    elif strategy in ("close_connection", "close_after_response"):
        # Clean shutdown
        socket.close()

Install with Tessl CLI