mitmproxy.connection

  1import dataclasses
  2import time
  3import uuid
  4import warnings
  5from abc import ABCMeta
  6from collections.abc import Sequence
  7from dataclasses import dataclass
  8from dataclasses import field
  9from enum import Flag
 10from typing import Literal
 11
 12from mitmproxy import certs
 13from mitmproxy.coretypes import serializable
 14from mitmproxy.net import server_spec
 15from mitmproxy.proxy import mode_specs
 16from mitmproxy.utils import human
 17
 18
 19class ConnectionState(Flag):
 20    """The current state of the underlying socket."""
 21
 22    CLOSED = 0
 23    CAN_READ = 1
 24    CAN_WRITE = 2
 25    OPEN = CAN_READ | CAN_WRITE
 26
 27
 28TransportProtocol = Literal["tcp", "udp"]
 29
 30# https://docs.openssl.org/master/man3/SSL_get_version/#return-values
 31TlsVersion = 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
 48kw_only = {"kw_only": True}
 49
 50
 51# noinspection PyDataclass
 52@dataclass(**kw_only)
 53class Connection(serializable.SerializableDataclass, metaclass=ABCMeta):
 54    """
 55    Base class for client and server connections.
 56
 57    The connection object only exposes metadata about the connection, but not the underlying socket object.
 58    This is intentional, all I/O should be handled by `mitmproxy.proxy.server` exclusively.
 59    """
 60
 61    peername: Address | None
 62    """The remote's `(ip, port)` tuple for this connection."""
 63    sockname: Address | None
 64    """Our local `(ip, port)` tuple for this connection."""
 65
 66    state: ConnectionState = field(
 67        default=ConnectionState.CLOSED, metadata={"serialize": False}
 68    )
 69    """The current connection state."""
 70
 71    # all connections have a unique id. While
 72    # f.client_conn == f2.client_conn already holds true for live flows (where we have object identity),
 73    # we also want these semantics for recorded flows.
 74    id: str = field(default_factory=lambda: str(uuid.uuid4()))
 75    """A unique UUID to identify the connection."""
 76    transport_protocol: TransportProtocol = field(default="tcp")
 77    """The connection protocol in use."""
 78    error: str | None = None
 79    """
 80    A string describing a general error with connections to this address.
 81
 82    The purpose of this property is to signal that new connections to the particular endpoint should not be attempted,
 83    for example because it uses an untrusted TLS certificate. Regular (unexpected) disconnects do not set the error
 84    property. This property is only reused per client connection.
 85    """
 86
 87    tls: bool = False
 88    """
 89    `True` if TLS should be established, `False` otherwise.
 90    Note that this property only describes if a connection should eventually be protected using TLS.
 91    To check if TLS has already been established, use `Connection.tls_established`.
 92    """
 93    certificate_list: Sequence[certs.Cert] = ()
 94    """
 95    The TLS certificate list as sent by the peer.
 96    The first certificate is the end-entity certificate.
 97
 98    > [RFC 8446] Prior to TLS 1.3, "certificate_list" ordering required each
 99    > certificate to certify the one immediately preceding it; however,
100    > some implementations allowed some flexibility.  Servers sometimes
101    > send both a current and deprecated intermediate for transitional
102    > purposes, and others are simply configured incorrectly, but these
103    > cases can nonetheless be validated properly.  For maximum
104    > compatibility, all implementations SHOULD be prepared to handle
105    > potentially extraneous certificates and arbitrary orderings from any
106    > TLS version, with the exception of the end-entity certificate which
107    > MUST be first.
108    """
109    alpn: bytes | None = None
110    """The application-layer protocol as negotiated using
111    [ALPN](https://en.wikipedia.org/wiki/Application-Layer_Protocol_Negotiation)."""
112    alpn_offers: Sequence[bytes] = ()
113    """The ALPN offers as sent in the ClientHello."""
114    # we may want to add SSL_CIPHER_description here, but that's currently not exposed by cryptography
115    cipher: str | None = None
116    """The active cipher name as returned by OpenSSL's `SSL_CIPHER_get_name`."""
117    cipher_list: Sequence[str] = ()
118    """Ciphers accepted by the proxy server on this connection."""
119    tls_version: TlsVersion | None = None
120    """The active TLS version."""
121    sni: str | None = None
122    """
123    The [Server Name Indication (SNI)](https://en.wikipedia.org/wiki/Server_Name_Indication) sent in the ClientHello.
124    """
125
126    timestamp_start: float | None = None
127    timestamp_end: float | None = None
128    """*Timestamp:* Connection has been closed."""
129    timestamp_tls_setup: float | None = None
130    """*Timestamp:* TLS handshake has been completed successfully."""
131
132    @property
133    def connected(self) -> bool:
134        """*Read-only:* `True` if Connection.state is ConnectionState.OPEN, `False` otherwise."""
135        return self.state is ConnectionState.OPEN
136
137    @property
138    def tls_established(self) -> bool:
139        """*Read-only:* `True` if TLS has been established, `False` otherwise."""
140        return self.timestamp_tls_setup is not None
141
142    def __eq__(self, other):
143        if isinstance(other, Connection):
144            return self.id == other.id
145        return False
146
147    def __hash__(self):
148        return hash(self.id)
149
150    def __repr__(self):
151        attrs = {
152            # ensure these come first.
153            "id": None,
154            "address": None,
155        }
156        for f in dataclasses.fields(self):
157            val = getattr(self, f.name)
158            if val != f.default:
159                if f.name == "cipher_list":
160                    val = f"<{len(val)} ciphers>"
161                elif f.name == "id":
162                    val = f"…{val[-6:]}"
163                attrs[f.name] = val
164        return f"{type(self).__name__}({attrs!r})"
165
166    @property
167    def alpn_proto_negotiated(self) -> bytes | None:  # pragma: no cover
168        """*Deprecated:* An outdated alias for Connection.alpn."""
169        warnings.warn(
170            "Connection.alpn_proto_negotiated is deprecated, use Connection.alpn instead.",
171            DeprecationWarning,
172            stacklevel=2,
173        )
174        return self.alpn
175
176
177# noinspection PyDataclass
178@dataclass(eq=False, repr=False, **kw_only)
179class Client(Connection):
180    """A connection between a client and mitmproxy."""
181
182    peername: Address
183    """The client's address."""
184    sockname: Address
185    """The local address we received this connection on."""
186
187    mitmcert: certs.Cert | None = None
188    """
189    The certificate used by mitmproxy to establish TLS with the client.
190    """
191
192    proxy_mode: mode_specs.ProxyMode = field(
193        default=mode_specs.ProxyMode.parse("regular")
194    )
195    """The proxy server type this client has been connecting to."""
196
197    timestamp_start: float = field(default_factory=time.time)
198    """*Timestamp:* TCP SYN received"""
199
200    def __str__(self):
201        if self.alpn:
202            tls_state = f", alpn={self.alpn.decode(errors='replace')}"
203        elif self.tls_established:
204            tls_state = ", tls"
205        else:
206            tls_state = ""
207        state = self.state.name
208        assert state
209        return f"Client({human.format_address(self.peername)}, state={state.lower()}{tls_state})"
210
211    @property
212    def address(self):  # pragma: no cover
213        """*Deprecated:* An outdated alias for Client.peername."""
214        warnings.warn(
215            "Client.address is deprecated, use Client.peername instead.",
216            DeprecationWarning,
217            stacklevel=2,
218        )
219        return self.peername
220
221    @address.setter
222    def address(self, x):  # pragma: no cover
223        warnings.warn(
224            "Client.address is deprecated, use Client.peername instead.",
225            DeprecationWarning,
226            stacklevel=2,
227        )
228        self.peername = x
229
230    @property
231    def cipher_name(self) -> str | None:  # pragma: no cover
232        """*Deprecated:* An outdated alias for Connection.cipher."""
233        warnings.warn(
234            "Client.cipher_name is deprecated, use Client.cipher instead.",
235            DeprecationWarning,
236            stacklevel=2,
237        )
238        return self.cipher
239
240    @property
241    def clientcert(self) -> certs.Cert | None:  # pragma: no cover
242        """*Deprecated:* An outdated alias for Connection.certificate_list[0]."""
243        warnings.warn(
244            "Client.clientcert is deprecated, use Client.certificate_list instead.",
245            DeprecationWarning,
246            stacklevel=2,
247        )
248        if self.certificate_list:
249            return self.certificate_list[0]
250        else:
251            return None
252
253    @clientcert.setter
254    def clientcert(self, val):  # pragma: no cover
255        warnings.warn(
256            "Client.clientcert is deprecated, use Client.certificate_list instead.",
257            DeprecationWarning,
258            stacklevel=2,
259        )
260        if val:
261            self.certificate_list = [val]
262        else:
263            self.certificate_list = []
264
265
266# noinspection PyDataclass
267@dataclass(eq=False, repr=False, **kw_only)
268class Server(Connection):
269    """A connection between mitmproxy and an upstream server."""
270
271    address: Address | None  # type: ignore
272    """
273    The server's `(host, port)` address tuple.
274
275    The host can either be a domain or a plain IP address.
276    Which of those two will be present depends on the proxy mode and the client.
277    For explicit proxies, this value will reflect what the client instructs mitmproxy to connect to.
278    For example, if the client starts off a connection with `CONNECT example.com HTTP/1.1`, it will be `example.com`.
279    For transparent proxies such as WireGuard mode, this value will be an IP address.
280    """
281
282    peername: Address | None = None
283    """
284    The server's resolved `(ip, port)` tuple. Will be set during connection establishment.
285    May be `None` in upstream proxy mode when the address is resolved by the upstream proxy only.
286    """
287    sockname: Address | None = None
288
289    timestamp_start: float | None = None
290    """
291    *Timestamp:* Connection establishment started.
292
293    For IP addresses, this corresponds to sending a TCP SYN; for domains, this corresponds to starting a DNS lookup.
294    """
295    timestamp_tcp_setup: float | None = None
296    """*Timestamp:* TCP ACK received."""
297
298    via: server_spec.ServerSpec | None = None
299    """An optional proxy server specification via which the connection should be established."""
300
301    def __str__(self):
302        if self.alpn:
303            tls_state = f", alpn={self.alpn.decode(errors='replace')}"
304        elif self.tls_established:
305            tls_state = ", tls"
306        else:
307            tls_state = ""
308        if self.sockname:
309            local_port = f", src_port={self.sockname[1]}"
310        else:
311            local_port = ""
312        state = self.state.name
313        assert state
314        return f"Server({human.format_address(self.address)}, state={state.lower()}{tls_state}{local_port})"
315
316    def __setattr__(self, name, value):
317        if name in ("address", "via"):
318            connection_open = (
319                self.__dict__.get("state", ConnectionState.CLOSED)
320                is ConnectionState.OPEN
321            )
322            # assigning the current value is okay, that may be an artifact of calling .set_state().
323            attr_changed = self.__dict__.get(name) != value
324            if connection_open and attr_changed:
325                raise RuntimeError(f"Cannot change server.{name} on open connection.")
326        return super().__setattr__(name, value)
327
328    @property
329    def ip_address(self) -> Address | None:  # pragma: no cover
330        """*Deprecated:* An outdated alias for `Server.peername`."""
331        warnings.warn(
332            "Server.ip_address is deprecated, use Server.peername instead.",
333            DeprecationWarning,
334            stacklevel=2,
335        )
336        return self.peername
337
338    @property
339    def cert(self) -> certs.Cert | None:  # pragma: no cover
340        """*Deprecated:* An outdated alias for `Connection.certificate_list[0]`."""
341        warnings.warn(
342            "Server.cert is deprecated, use Server.certificate_list instead.",
343            DeprecationWarning,
344            stacklevel=2,
345        )
346        if self.certificate_list:
347            return self.certificate_list[0]
348        else:
349            return None
350
351    @cert.setter
352    def cert(self, val):  # pragma: no cover
353        warnings.warn(
354            "Server.cert is deprecated, use Server.certificate_list instead.",
355            DeprecationWarning,
356            stacklevel=2,
357        )
358        if val:
359            self.certificate_list = [val]
360        else:
361            self.certificate_list = []
362
363
364__all__ = ["Connection", "Client", "Server", "ConnectionState"]