Class Server

mitmproxy/connection.py:263–356

A connection between mitmproxy and an upstream server.

@dataclass(eq=False, repr=False, kw_only=True)
class Server(Connection):
    """A connection between mitmproxy and an upstream server."""

    address: Address | None  # type: ignore
    """
    The server's `(host, port)` address tuple.

    The host can either be a domain or a plain IP address.
    Which of those two will be present depends on the proxy mode and the client.
    For explicit proxies, this value will reflect what the client instructs mitmproxy to connect to.
    For example, if the client starts off a connection with `CONNECT example.com HTTP/1.1`, it will be `example.com`.
    For transparent proxies such as WireGuard mode, this value will be an IP address.
    """

    peername: Address | None = None
    """
    The server's resolved `(ip, port)` tuple. Will be set during connection establishment.
    May be `None` in upstream proxy mode when the address is resolved by the upstream proxy only.
    """
    sockname: Address | None = None

    timestamp_start: float | None = None
    """
    *Timestamp:* Connection establishment started.

    For IP addresses, this corresponds to sending a TCP SYN; for domains, this corresponds to starting a DNS lookup.
    """
    timestamp_tcp_setup: float | None = None
    """*Timestamp:* TCP ACK received."""

    via: server_spec.ServerSpec | None = None
    """An optional proxy server specification via which the connection should be established."""

    def __str__(self):
        if self.alpn:
            tls_state = f", alpn={self.alpn.decode(errors='replace')}"
        elif self.tls_established:
            tls_state = ", tls"
        else:
            tls_state = ""
        if self.sockname:
            local_port = f", src_port={self.sockname[1]}"
        else:
            local_port = ""
        state = self.state.name
        assert state
        return f"Server({human.format_address(self.address)}, state={state.lower()}{tls_state}{local_port})"

    def __setattr__(self, name, value):
        if name in ("address", "via"):
            connection_open = (
                self.__dict__.get("state", ConnectionState.CLOSED)
                is ConnectionState.OPEN
            )
            # assigning the current value is okay, that may be an artifact of calling .set_state().
            attr_changed = self.__dict__.get(name) != value
            if connection_open and attr_changed:
                raise RuntimeError(f"Cannot change server.{name} on open connection.")