tessl/pypi-mitmproxy

An interactive, SSL/TLS-capable intercepting proxy for HTTP/1, HTTP/2, and WebSockets.

—

Pending

Overview

Eval results

Files

HTTP Flow Management

Name: tessl/pypi-mitmproxy
Author: tessl

Core HTTP proxy functionality for intercepting, analyzing, and modifying HTTP/1, HTTP/2, and HTTP/3 traffic. HTTP flows represent complete request/response conversations with comprehensive metadata and manipulation capabilities.

Capabilities

HTTPFlow Class

The main container for HTTP conversations, tracking requests, responses, errors, and metadata throughout the proxy lifecycle.

class HTTPFlow(Flow):
    """
    An HTTP conversation between client and server.
    
    Attributes:
    - request: The HTTP request
    - response: The HTTP response (None if not yet received)
    - error: Error information if the flow failed
    - server_conn: Server connection details
    - client_conn: Client connection details
    - intercepted: Whether the flow is currently intercepted
    - marked: User-defined marker for flow identification
    - metadata: Additional flow metadata
    """
    request: Request
    response: Optional[Response]
    error: Optional[Error]
    server_conn: Optional[Server]
    client_conn: Optional[Client]
    intercepted: bool
    marked: str
    metadata: Dict[str, Any]
    
    def copy(self) -> "HTTPFlow": ...
    def kill(self) -> None: ...
    def resume(self) -> None: ...
    def intercept(self) -> None: ...

Request Class

HTTP request representation with comprehensive access to method, URL, headers, and content.

class Request(Message):
    """
    HTTP request with method, URL, headers, and content.
    
    Parameters:
    - host: Target server hostname
    - port: Target server port
    - method: HTTP method as bytes (b'GET', b'POST', etc.)
    - scheme: URL scheme as bytes (b'http', b'https')
    - authority: Host authority as bytes
    - path: Request path as bytes
    - http_version: HTTP version as bytes (b'HTTP/1.1', b'HTTP/2')
    - headers: HTTP headers container
    - content: Request body content as bytes
    - trailers: HTTP trailers (for chunked encoding)
    - timestamp_start: Request start timestamp
    - timestamp_end: Request completion timestamp
    """
    def __init__(
        self,
        host: str,
        port: int,
        method: bytes,
        scheme: bytes,
        authority: bytes,
        path: bytes,
        http_version: bytes,
        headers: Headers,
        content: bytes,
        trailers: Optional[Headers] = None,
        timestamp_start: float = 0,
        timestamp_end: Optional[float] = None
    ) -> None: ...
    
    # Properties
    @property
    def method(self) -> str: ...
    @method.setter
    def method(self, value: str) -> None: ...
    
    @property
    def url(self) -> str: ...
    @url.setter
    def url(self, value: str) -> None: ...
    
    @property
    def scheme(self) -> str: ...
    @scheme.setter
    def scheme(self, value: str) -> None: ...
    
    @property
    def host(self) -> str: ...
    @host.setter
    def host(self, value: str) -> None: ...
    
    @property
    def port(self) -> int: ...
    @port.setter
    def port(self, value: int) -> None: ...
    
    @property
    def path(self) -> str: ...
    @path.setter
    def path(self, value: str) -> None: ...
    
    @property
    def query(self) -> MultiDict: ...
    @query.setter
    def query(self, value: MultiDict) -> None: ...
    
    # Content methods
    def get_text(self, strict: bool = True) -> str: ...
    def set_text(self, text: str) -> None: ...
    def get_content(self, strict: bool = True) -> bytes: ...
    def set_content(self, content: Optional[bytes]) -> None: ...
    
    # Convenience methods
    def json(self, **kwargs) -> Any: ...
    def urlencoded_form(self) -> MultiDict: ...
    def multipart_form(self) -> MultiDict: ...
    
    # Static constructors
    @classmethod
    def make(
        cls,
        method: str,
        url: str,
        content: Union[bytes, str] = b"",
        headers: Optional[Union[Headers, Dict]] = None,
        **kwargs
    ) -> "Request": ...

Response Class

HTTP response representation with status code, headers, and content manipulation capabilities.

class Response(Message):
    """
    HTTP response with status code, headers, and content.
    
    Parameters:
    - http_version: HTTP version as bytes
    - status_code: HTTP status code (200, 404, etc.)
    - reason: HTTP reason phrase as bytes
    - headers: HTTP response headers
    - content: Response body content as bytes
    - trailers: HTTP trailers (for chunked encoding)
    - timestamp_start: Response start timestamp
    - timestamp_end: Response completion timestamp
    """
    def __init__(
        self,
        http_version: bytes,
        status_code: int,
        reason: bytes,
        headers: Headers,
        content: bytes,
        trailers: Optional[Headers] = None,
        timestamp_start: float = 0,
        timestamp_end: float = 0
    ) -> None: ...
    
    # Properties
    @property
    def status_code(self) -> int: ...
    @status_code.setter
    def status_code(self, value: int) -> None: ...
    
    @property
    def reason(self) -> str: ...
    @reason.setter
    def reason(self, value: str) -> None: ...
    
    @property
    def http_version(self) -> str: ...
    @http_version.setter
    def http_version(self, value: str) -> None: ...
    
    # Content methods (inherited from Message)
    def get_text(self, strict: bool = True) -> str: ...
    def set_text(self, text: str) -> None: ...
    def get_content(self, strict: bool = True) -> bytes: ...
    def set_content(self, content: Optional[bytes]) -> None: ...
    
    # Convenience methods
    def json(self, **kwargs) -> Any: ...
    
    # Static constructors
    @classmethod
    def make(
        cls,
        status_code: int = 200,
        content: Union[bytes, str] = b"",
        headers: Optional[Union[Headers, Dict]] = None,
        **kwargs
    ) -> "Response": ...

Headers Class

Case-insensitive HTTP headers container with multi-value support and comprehensive manipulation methods.

class Headers(multidict.MultiDict):
    """
    Case-insensitive HTTP headers container supporting multiple values per key.
    
    Extends multidict.MultiDict with HTTP-specific functionality.
    """
    def __init__(self, data: Union[Dict, List[Tuple[str, str]], Headers] = None) -> None: ...
    
    # Basic access (case-insensitive)
    def __getitem__(self, key: str) -> str: ...
    def __setitem__(self, key: str, value: str) -> None: ...
    def __delitem__(self, key: str) -> None: ...
    def __contains__(self, key: str) -> bool: ...
    
    # Multi-value methods
    def get(self, key: str, default: Optional[str] = None) -> Optional[str]: ...
    def get_all(self, key: str) -> List[str]: ...
    def add(self, key: str, value: str) -> None: ...
    def extend(self, other: Union[Dict, Headers]) -> None: ...
    
    # HTTP-specific methods
    def set_all(self, key: str, values: List[str]) -> None: ...
    def items(self, multi: bool = False) -> Iterator[Tuple[str, str]]: ...
    def keys(self) -> KeysView: ...
    def values(self) -> ValuesView: ...
    
    # String representation
    def __str__(self) -> str: ...
    def __repr__(self) -> str: ...

Message Base Class

Abstract base class for HTTP requests and responses providing common functionality.

class Message:
    """
    Base class for HTTP requests and responses.
    
    Provides common functionality for header and content manipulation.
    """
    headers: Headers
    content: bytes
    trailers: Optional[Headers]
    timestamp_start: float
    timestamp_end: Optional[float]
    
    # Content handling
    def get_text(self, strict: bool = True) -> str: ...
    def set_text(self, text: str) -> None: ...
    def get_content(self, strict: bool = True) -> bytes: ...
    def set_content(self, content: Optional[bytes]) -> None: ...
    
    # Content decoding/encoding
    def decode(self, strict: bool = True) -> None: ...
    def encode(self, encoding: str) -> None: ...
    
    # JSON handling
    def json(self, **kwargs) -> Any: ...
    
    # Raw data access
    @property
    def raw_content(self) -> bytes: ...
    @raw_content.setter
    def raw_content(self, content: bytes) -> None: ...

Usage Examples

Basic Request/Response Modification

from mitmproxy import http

def request(flow: http.HTTPFlow) -> None:
    """Modify outgoing requests."""
    # Add authentication header
    if "api.example.com" in flow.request.url:
        flow.request.headers["Authorization"] = "Bearer token123"
    
    # Modify request body for POST requests
    if flow.request.method == "POST":
        if flow.request.headers.get("content-type", "").startswith("application/json"):
            data = flow.request.json()
            data["modified"] = True
            flow.request.set_text(json.dumps(data))

def response(flow: http.HTTPFlow) -> None:
    """Modify incoming responses."""
    # Add CORS headers
    flow.response.headers["Access-Control-Allow-Origin"] = "*"
    flow.response.headers["Access-Control-Allow-Methods"] = "GET,POST,PUT,DELETE"
    
    # Modify JSON responses
    if flow.response.headers.get("content-type", "").startswith("application/json"):
        try:
            data = flow.response.json()
            data["proxy_processed"] = True
            flow.response.set_text(json.dumps(data))
        except ValueError:
            pass  # Not valid JSON

Creating Responses Programmatically

from mitmproxy import http

def request(flow: http.HTTPFlow) -> None:
    """Create custom responses."""
    # Block specific requests
    if "/blocked" in flow.request.path:
        flow.response = http.Response.make(
            status_code=403,
            content="Access Denied",
            headers={"Content-Type": "text/plain"}
        )
    
    # Redirect requests
    elif "/old-path" in flow.request.path:
        flow.response = http.Response.make(
            status_code=301,
            headers={"Location": "https://example.com/new-path"}
        )
    
    # Mock API responses
    elif "/api/users" in flow.request.path and flow.request.method == "GET":
        mock_data = {"users": [{"id": 1, "name": "John"}, {"id": 2, "name": "Jane"}]}
        flow.response = http.Response.make(
            status_code=200,
            content=json.dumps(mock_data),
            headers={"Content-Type": "application/json"}
        )

Advanced Flow Manipulation

import time
from mitmproxy import http

def request(flow: http.HTTPFlow) -> None:
    """Advanced request handling."""
    # Log detailed request information
    print(f"Request: {flow.request.method} {flow.request.url}")
    print(f"Headers: {dict(flow.request.headers)}")
    print(f"Content-Length: {len(flow.request.content)}")
    
    # Modify URL parameters
    if flow.request.query:
        # Add tracking parameter
        flow.request.query["proxy_time"] = str(int(time.time()))
        
        # Remove sensitive parameters
        if "password" in flow.request.query:
            del flow.request.query["password"]
    
    # Handle different content types
    content_type = flow.request.headers.get("content-type", "")
    
    if "application/x-www-form-urlencoded" in content_type:
        form_data = flow.request.urlencoded_form()
        form_data["proxy_processed"] = "true"
        # Note: Setting form data requires manual encoding
    
    elif "multipart/form-data" in content_type:
        form_data = flow.request.multipart_form()
        print(f"Form fields: {list(form_data.keys())}")

def response(flow: http.HTTPFlow) -> None:
    """Advanced response handling."""
    # Calculate response time
    if flow.request.timestamp_start and flow.response.timestamp_end:
        response_time = flow.response.timestamp_end - flow.request.timestamp_start
        flow.response.headers["X-Response-Time"] = f"{response_time:.3f}s"
    
    # Compress responses
    if len(flow.response.content) > 1024:
        if "gzip" not in flow.response.headers.get("content-encoding", ""):
            # Note: Actual compression would require additional logic
            flow.response.headers["X-Original-Size"] = str(len(flow.response.content))

Install with Tessl CLI