Function format_flow

mitmproxy/tools/console/common.py:726–863

This functions calls the proper renderer depending on the flow type. We also want to cache the renderer output, so we extract all attributes relevant for display and call the render with only that. This assures that rows are updated if the flow is changed.

(
    f: flow.Flow,
    *,
    render_mode: RenderMode,
    hostheader: bool = False,  # pass options directly if we need more stuff from them
    focused: bool = True,
)

def format_flow(
    f: flow.Flow,
    *,
    render_mode: RenderMode,
    hostheader: bool = False,  # pass options directly if we need more stuff from them
    focused: bool = True,
) -> urwid.Widget:
    """
    This functions calls the proper renderer depending on the flow type.
    We also want to cache the renderer output, so we extract all attributes
    relevant for display and call the render with only that. This assures that rows
    are updated if the flow is changed.
    """
    duration: float | None
    error_message: str | None
    if f.error:
        error_message = f.error.msg
    else:
        error_message = None

    if isinstance(f, (TCPFlow, UDPFlow)):
        total_size = 0
        for message in f.messages:
            total_size += len(message.content)
        if f.messages:
            duration = f.messages[-1].timestamp - f.client_conn.timestamp_start
        else:
            duration = None
        if f.client_conn.tls_version == "QUICv1":
            protocol = "quic"
        else:
            protocol = f.type
        return format_message_flow(
            render_mode=render_mode,
            focused=focused,
            timestamp_start=f.client_conn.timestamp_start,
            marked=f.marked,
            protocol=protocol,
            client_address=f.client_conn.peername,
            server_address=f.server_conn.address,
            total_size=total_size,
            duration=duration,
            error_message=error_message,
        )
    elif isinstance(f, DNSFlow):
        if f.request.timestamp and f.response and f.response.timestamp:
            duration = f.response.timestamp - f.request.timestamp
        else:
            duration = None
        if f.response:
            response_code_str: str | None = dns.response_codes.to_str(
                f.response.response_code
            )
            response_code_http_equiv = dns.response_codes.http_equiv_status_code(
                f.response.response_code
            )
            answer = ", ".join(str(x) for x in f.response.answers)
        else: