codekingpro/portable-devtools

1import dataclasses2import time3import uuid4import warnings5from abc import ABCMeta6from collections.abc import Sequence7from dataclasses import dataclass8from dataclasses import field9from enum import Flag10from typing import Literal11 12from mitmproxy import certs13from mitmproxy.coretypes import serializable14from mitmproxy.net import server_spec15from mitmproxy.proxy import mode_specs16from mitmproxy.utils import human17 18 19class ConnectionState(Flag):20    """The current state of the underlying socket."""21 22    CLOSED = 023    CAN_READ = 124    CAN_WRITE = 225    OPEN = CAN_READ | CAN_WRITE26 27 28TransportProtocol = Literal["tcp", "udp"]29 30# https://docs.openssl.org/master/man3/SSL_get_version/#return-values31TlsVersion = Literal[32    "SSLv3",33    "TLSv1",34    "TLSv1.1",35    "TLSv1.2",36    "TLSv1.3",37    "DTLSv0.9",38    "DTLSv1",39    "DTLSv1.2",40    "QUICv1",41]42 43# practically speaking we may have IPv6 addresses with flowinfo and scope_id,44# but type checking isn't good enough to properly handle tuple unions.45# this version at least provides useful type checking messages.46Address = tuple[str, int]47 48 49@dataclass(kw_only=True)50class Connection(serializable.SerializableDataclass, metaclass=ABCMeta):51    """52    Base class for client and server connections.53 54    The connection object only exposes metadata about the connection, but not the underlying socket object.55    This is intentional, all I/O should be handled by `mitmproxy.proxy.server` exclusively.56    """57 58    peername: Address | None59    """The remote's `(ip, port)` tuple for this connection."""60    sockname: Address | None61    """Our local `(ip, port)` tuple for this connection."""62 63    state: ConnectionState = field(64        default=ConnectionState.CLOSED, metadata={"serialize": False}65    )66    """The current connection state."""67 68    # all connections have a unique id. While69    # f.client_conn == f2.client_conn already holds true for live flows (where we have object identity),70    # we also want these semantics for recorded flows.71    id: str = field(default_factory=lambda: str(uuid.uuid4()))72    """A unique UUID to identify the connection."""73    transport_protocol: TransportProtocol = field(default="tcp")74    """The connection protocol in use."""75    error: str | None = None76    """77    A string describing a general error with connections to this address.78 79    The purpose of this property is to signal that new connections to the particular endpoint should not be attempted,80    for example because it uses an untrusted TLS certificate. Regular (unexpected) disconnects do not set the error81    property. This property is only reused per client connection.82    """83 84    tls: bool = False85    """86    `True` if TLS should be established, `False` otherwise.87    Note that this property only describes if a connection should eventually be protected using TLS.88    To check if TLS has already been established, use `Connection.tls_established`.89    """90    certificate_list: Sequence[certs.Cert] = ()91    """92    The TLS certificate list as sent by the peer.93    The first certificate is the end-entity certificate.94 95    > [RFC 8446] Prior to TLS 1.3, "certificate_list" ordering required each96    > certificate to certify the one immediately preceding it; however,97    > some implementations allowed some flexibility.  Servers sometimes98    > send both a current and deprecated intermediate for transitional99    > purposes, and others are simply configured incorrectly, but these100    > cases can nonetheless be validated properly.  For maximum101    > compatibility, all implementations SHOULD be prepared to handle102    > potentially extraneous certificates and arbitrary orderings from any103    > TLS version, with the exception of the end-entity certificate which104    > MUST be first.105    """106    alpn: bytes | None = None107    """The application-layer protocol as negotiated using108    [ALPN](https://en.wikipedia.org/wiki/Application-Layer_Protocol_Negotiation)."""109    alpn_offers: Sequence[bytes] = ()110    """The ALPN offers as sent in the ClientHello."""111    # we may want to add SSL_CIPHER_description here, but that's currently not exposed by cryptography112    cipher: str | None = None113    """The active cipher name as returned by OpenSSL's `SSL_CIPHER_get_name`."""114    cipher_list: Sequence[str] = ()115    """Ciphers accepted by the proxy server on this connection."""116    tls_version: TlsVersion | None = None117    """The active TLS version."""118    sni: str | None = None119    """120    The [Server Name Indication (SNI)](https://en.wikipedia.org/wiki/Server_Name_Indication) sent in the ClientHello.121    """122 123    timestamp_start: float | None = None124    timestamp_end: float | None = None125    """*Timestamp:* Connection has been closed."""126    timestamp_tls_setup: float | None = None127    """*Timestamp:* TLS handshake has been completed successfully."""128 129    @property130    def connected(self) -> bool:131        """*Read-only:* `True` if Connection.state is ConnectionState.OPEN, `False` otherwise."""132        return self.state is ConnectionState.OPEN133 134    @property135    def tls_established(self) -> bool:136        """*Read-only:* `True` if TLS has been established, `False` otherwise."""137        return self.timestamp_tls_setup is not None138 139    def __eq__(self, other):140        if isinstance(other, Connection):141            return self.id == other.id142        return False143 144    def __hash__(self):145        return hash(self.id)146 147    def __repr__(self):148        attrs = {149            # ensure these come first.150            "id": None,151            "address": None,152        }153        for f in dataclasses.fields(self):154            val = getattr(self, f.name)155            if val != f.default:156                if f.name == "cipher_list":157                    val = f"<{len(val)} ciphers>"158                elif f.name == "id":159                    val = f"…{val[-6:]}"160                attrs[f.name] = val161        return f"{type(self).__name__}({attrs!r})"162 163    @property164    def alpn_proto_negotiated(self) -> bytes | None:  # pragma: no cover165        """*Deprecated:* An outdated alias for Connection.alpn."""166        warnings.warn(167            "Connection.alpn_proto_negotiated is deprecated, use Connection.alpn instead.",168            DeprecationWarning,169            stacklevel=2,170        )171        return self.alpn172 173 174@dataclass(eq=False, repr=False, kw_only=True)175class Client(Connection):  # type: ignore[override]176    """A connection between a client and mitmproxy."""177 178    peername: Address179    """The client's address."""180    sockname: Address181    """The local address we received this connection on."""182 183    mitmcert: certs.Cert | None = None184    """185    The certificate used by mitmproxy to establish TLS with the client.186    """187 188    proxy_mode: mode_specs.ProxyMode = field(189        default=mode_specs.ProxyMode.parse("regular")190    )191    """The proxy server type this client has been connecting to."""192 193    timestamp_start: float = field(default_factory=time.time)194    """*Timestamp:* TCP SYN received"""195 196    def __str__(self):197        if self.alpn:198            tls_state = f", alpn={self.alpn.decode(errors='replace')}"199        elif self.tls_established:200            tls_state = ", tls"201        else:202            tls_state = ""203        state = self.state.name204        assert state205        return f"Client({human.format_address(self.peername)}, state={state.lower()}{tls_state})"206 207    @property208    def address(self):  # pragma: no cover209        """*Deprecated:* An outdated alias for Client.peername."""210        warnings.warn(211            "Client.address is deprecated, use Client.peername instead.",212            DeprecationWarning,213            stacklevel=2,214        )215        return self.peername216 217    @address.setter218    def address(self, x):  # pragma: no cover219        warnings.warn(220            "Client.address is deprecated, use Client.peername instead.",221            DeprecationWarning,222            stacklevel=2,223        )224        self.peername = x225 226    @property227    def cipher_name(self) -> str | None:  # pragma: no cover228        """*Deprecated:* An outdated alias for Connection.cipher."""229        warnings.warn(230            "Client.cipher_name is deprecated, use Client.cipher instead.",231            DeprecationWarning,232            stacklevel=2,233        )234        return self.cipher235 236    @property237    def clientcert(self) -> certs.Cert | None:  # pragma: no cover238        """*Deprecated:* An outdated alias for Connection.certificate_list[0]."""239        warnings.warn(240            "Client.clientcert is deprecated, use Client.certificate_list instead.",241            DeprecationWarning,242            stacklevel=2,243        )244        if self.certificate_list:245            return self.certificate_list[0]246        else:247            return None248 249    @clientcert.setter250    def clientcert(self, val):  # pragma: no cover251        warnings.warn(252            "Client.clientcert is deprecated, use Client.certificate_list instead.",253            DeprecationWarning,254            stacklevel=2,255        )256        if val:257            self.certificate_list = [val]258        else:259            self.certificate_list = []260 261 262@dataclass(eq=False, repr=False, kw_only=True)263class Server(Connection):264    """A connection between mitmproxy and an upstream server."""265 266    address: Address | None  # type: ignore267    """268    The server's `(host, port)` address tuple.269 270    The host can either be a domain or a plain IP address.271    Which of those two will be present depends on the proxy mode and the client.272    For explicit proxies, this value will reflect what the client instructs mitmproxy to connect to.273    For example, if the client starts off a connection with `CONNECT example.com HTTP/1.1`, it will be `example.com`.274    For transparent proxies such as WireGuard mode, this value will be an IP address.275    """276 277    peername: Address | None = None278    """279    The server's resolved `(ip, port)` tuple. Will be set during connection establishment.280    May be `None` in upstream proxy mode when the address is resolved by the upstream proxy only.281    """282    sockname: Address | None = None283 284    timestamp_start: float | None = None285    """286    *Timestamp:* Connection establishment started.287 288    For IP addresses, this corresponds to sending a TCP SYN; for domains, this corresponds to starting a DNS lookup.289    """290    timestamp_tcp_setup: float | None = None291    """*Timestamp:* TCP ACK received."""292 293    via: server_spec.ServerSpec | None = None294    """An optional proxy server specification via which the connection should be established."""295 296    def __str__(self):297        if self.alpn:298            tls_state = f", alpn={self.alpn.decode(errors='replace')}"299        elif self.tls_established:300            tls_state = ", tls"301        else:302            tls_state = ""303        if self.sockname:304            local_port = f", src_port={self.sockname[1]}"305        else:306            local_port = ""307        state = self.state.name308        assert state309        return f"Server({human.format_address(self.address)}, state={state.lower()}{tls_state}{local_port})"310 311    def __setattr__(self, name, value):312        if name in ("address", "via"):313            connection_open = (314                self.__dict__.get("state", ConnectionState.CLOSED)315                is ConnectionState.OPEN316            )317            # assigning the current value is okay, that may be an artifact of calling .set_state().318            attr_changed = self.__dict__.get(name) != value319            if connection_open and attr_changed:320                raise RuntimeError(f"Cannot change server.{name} on open connection.")321        return super().__setattr__(name, value)322 323    @property324    def ip_address(self) -> Address | None:  # pragma: no cover325        """*Deprecated:* An outdated alias for `Server.peername`."""326        warnings.warn(327            "Server.ip_address is deprecated, use Server.peername instead.",328            DeprecationWarning,329            stacklevel=2,330        )331        return self.peername332 333    @property334    def cert(self) -> certs.Cert | None:  # pragma: no cover335        """*Deprecated:* An outdated alias for `Connection.certificate_list[0]`."""336        warnings.warn(337            "Server.cert is deprecated, use Server.certificate_list instead.",338            DeprecationWarning,339            stacklevel=2,340        )341        if self.certificate_list:342            return self.certificate_list[0]343        else:344            return None345 346    @cert.setter347    def cert(self, val):  # pragma: no cover348        warnings.warn(349            "Server.cert is deprecated, use Server.certificate_list instead.",350            DeprecationWarning,351            stacklevel=2,352        )353        if val:354            self.certificate_list = [val]355        else:356            self.certificate_list = []357 358 359__all__ = ["Connection", "Client", "Server", "ConnectionState"]360