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
 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 | None
 59    """The remote's `(ip, port)` tuple for this connection."""
 60    sockname: Address | None
 61    """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. While
 69    # 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 = None
 76    """
 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 error
 81    property. This property is only reused per client connection.
 82    """
 83
 84    tls: bool = False
 85    """
 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 each
 96    > certificate to certify the one immediately preceding it; however,
 97    > some implementations allowed some flexibility.  Servers sometimes
 98    > send both a current and deprecated intermediate for transitional
 99    > purposes, and others are simply configured incorrectly, but these
100    > cases can nonetheless be validated properly.  For maximum
101    > compatibility, all implementations SHOULD be prepared to handle
102    > potentially extraneous certificates and arbitrary orderings from any
103    > TLS version, with the exception of the end-entity certificate which
104    > MUST be first.
105    """
106    alpn: bytes | None = None
107    """The application-layer protocol as negotiated using
108    [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 cryptography
112    cipher: str | None = None
113    """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 = None
117    """The active TLS version."""
118    sni: str | None = None
119    """
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 = None
124    timestamp_end: float | None = None
125    """*Timestamp:* Connection has been closed."""
126    timestamp_tls_setup: float | None = None
127    """*Timestamp:* TLS handshake has been completed successfully."""
128
129    @property
130    def connected(self) -> bool:
131        """*Read-only:* `True` if Connection.state is ConnectionState.OPEN, `False` otherwise."""
132        return self.state is ConnectionState.OPEN
133
134    @property
135    def tls_established(self) -> bool:
136        """*Read-only:* `True` if TLS has been established, `False` otherwise."""
137        return self.timestamp_tls_setup is not None
138
139    def __eq__(self, other):
140        if isinstance(other, Connection):
141            return self.id == other.id
142        return False
143
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] = val
161        return f"{type(self).__name__}({attrs!r})"
162
163    @property
164    def alpn_proto_negotiated(self) -> bytes | None:  # pragma: no cover
165        """*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.alpn
172
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: Address
179    """The client's address."""
180    sockname: Address
181    """The local address we received this connection on."""
182
183    mitmcert: certs.Cert | None = None
184    """
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.name
204        assert state
205        return f"Client({human.format_address(self.peername)}, state={state.lower()}{tls_state})"
206
207    @property
208    def address(self):  # pragma: no cover
209        """*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.peername
216
217    @address.setter
218    def address(self, x):  # pragma: no cover
219        warnings.warn(
220            "Client.address is deprecated, use Client.peername instead.",
221            DeprecationWarning,
222            stacklevel=2,
223        )
224        self.peername = x
225
226    @property
227    def cipher_name(self) -> str | None:  # pragma: no cover
228        """*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.cipher
235
236    @property
237    def clientcert(self) -> certs.Cert | None:  # pragma: no cover
238        """*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 None
248
249    @clientcert.setter
250    def clientcert(self, val):  # pragma: no cover
251        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: ignore
267    """
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 = None
278    """
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 = None
283
284    timestamp_start: float | None = None
285    """
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 = None
291    """*Timestamp:* TCP ACK received."""
292
293    via: server_spec.ServerSpec | None = None
294    """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.name
308        assert state
309        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.OPEN
316            )
317            # assigning the current value is okay, that may be an artifact of calling .set_state().
318            attr_changed = self.__dict__.get(name) != value
319            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    @property
324    def ip_address(self) -> Address | None:  # pragma: no cover
325        """*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.peername
332
333    @property
334    def cert(self) -> certs.Cert | None:  # pragma: no cover
335        """*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 None
345
346    @cert.setter
347    def cert(self, val):  # pragma: no cover
348        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"]