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