mitmproxy.connection

  1import uuid
  2import warnings
  3from abc import ABCMeta
  4from collections.abc import Sequence
  5from enum import Flag
  6from typing import Literal, Optional
  7
  8from mitmproxy import certs
  9from mitmproxy.coretypes import serializable
 10from mitmproxy.proxy import mode_specs
 11from mitmproxy.net import server_spec
 12from mitmproxy.utils import human
 13
 14
 15class ConnectionState(Flag):
 16    """The current state of the underlying socket."""
 17
 18    CLOSED = 0
 19    CAN_READ = 1
 20    CAN_WRITE = 2
 21    OPEN = CAN_READ | CAN_WRITE
 22
 23
 24TransportProtocol = Literal["tcp", "udp"]
 25
 26
 27# practically speaking we may have IPv6 addresses with flowinfo and scope_id,
 28# but type checking isn't good enough to properly handle tuple unions.
 29# this version at least provides useful type checking messages.
 30Address = tuple[str, int]
 31
 32
 33class Connection(serializable.Serializable, metaclass=ABCMeta):
 34    """
 35    Base class for client and server connections.
 36
 37    The connection object only exposes metadata about the connection, but not the underlying socket object.
 38    This is intentional, all I/O should be handled by `mitmproxy.proxy.server` exclusively.
 39    """
 40
 41    # all connections have a unique id. While
 42    # f.client_conn == f2.client_conn already holds true for live flows (where we have object identity),
 43    # we also want these semantics for recorded flows.
 44    id: str
 45    """A unique UUID to identify the connection."""
 46    state: ConnectionState
 47    """The current connection state."""
 48    transport_protocol: TransportProtocol
 49    """The connection protocol in use."""
 50    peername: Optional[Address]
 51    """The remote's `(ip, port)` tuple for this connection."""
 52    sockname: Optional[Address]
 53    """Our local `(ip, port)` tuple for this connection."""
 54    error: Optional[str] = None
 55    """
 56    A string describing a general error with connections to this address.
 57
 58    The purpose of this property is to signal that new connections to the particular endpoint should not be attempted,
 59    for example because it uses an untrusted TLS certificate. Regular (unexpected) disconnects do not set the error
 60    property. This property is only reused per client connection.
 61    """
 62
 63    tls: bool = False
 64    """
 65    `True` if TLS should be established, `False` otherwise.
 66    Note that this property only describes if a connection should eventually be protected using TLS.
 67    To check if TLS has already been established, use `Connection.tls_established`.
 68    """
 69    certificate_list: Sequence[certs.Cert] = ()
 70    """
 71    The TLS certificate list as sent by the peer.
 72    The first certificate is the end-entity certificate.
 73
 74    > [RFC 8446] Prior to TLS 1.3, "certificate_list" ordering required each
 75    > certificate to certify the one immediately preceding it; however,
 76    > some implementations allowed some flexibility.  Servers sometimes
 77    > send both a current and deprecated intermediate for transitional
 78    > purposes, and others are simply configured incorrectly, but these
 79    > cases can nonetheless be validated properly.  For maximum
 80    > compatibility, all implementations SHOULD be prepared to handle
 81    > potentially extraneous certificates and arbitrary orderings from any
 82    > TLS version, with the exception of the end-entity certificate which
 83    > MUST be first.
 84    """
 85    alpn: Optional[bytes] = None
 86    """The application-layer protocol as negotiated using
 87    [ALPN](https://en.wikipedia.org/wiki/Application-Layer_Protocol_Negotiation)."""
 88    alpn_offers: Sequence[bytes] = ()
 89    """The ALPN offers as sent in the ClientHello."""
 90    # we may want to add SSL_CIPHER_description here, but that's currently not exposed by cryptography
 91    cipher: Optional[str] = None
 92    """The active cipher name as returned by OpenSSL's `SSL_CIPHER_get_name`."""
 93    cipher_list: Sequence[str] = ()
 94    """Ciphers accepted by the proxy server on this connection."""
 95    tls_version: Optional[str] = None
 96    """The active TLS version."""
 97    sni: Optional[str] = None
 98    """
 99    The [Server Name Indication (SNI)](https://en.wikipedia.org/wiki/Server_Name_Indication) sent in the ClientHello.
100    """
101
102    timestamp_start: Optional[float]
103    timestamp_end: Optional[float] = None
104    """*Timestamp:* Connection has been closed."""
105    timestamp_tls_setup: Optional[float] = None
106    """*Timestamp:* TLS handshake has been completed successfully."""
107
108    @property
109    def connected(self) -> bool:
110        """*Read-only:* `True` if Connection.state is ConnectionState.OPEN, `False` otherwise."""
111        return self.state is ConnectionState.OPEN
112
113    @property
114    def tls_established(self) -> bool:
115        """*Read-only:* `True` if TLS has been established, `False` otherwise."""
116        return self.timestamp_tls_setup is not None
117
118    def __eq__(self, other):
119        if isinstance(other, Connection):
120            return self.id == other.id
121        return False
122
123    def __hash__(self):
124        return hash(self.id)
125
126    def __repr__(self):
127        attrs = repr(
128            {
129                k: {
130                    "cipher_list": lambda: f"<{len(v)} ciphers>",
131                    "id": lambda: f"…{v[-6:]}",
132                }.get(k, lambda: v)()
133                for k, v in self.__dict__.items()
134            }
135        )
136        return f"{type(self).__name__}({attrs})"
137
138    @property
139    def alpn_proto_negotiated(self) -> Optional[bytes]:  # pragma: no cover
140        """*Deprecated:* An outdated alias for Connection.alpn."""
141        warnings.warn(
142            "Connection.alpn_proto_negotiated is deprecated, use Connection.alpn instead.",
143            DeprecationWarning,
144            stacklevel=2,
145        )
146        return self.alpn
147
148
149class Client(Connection):
150    """A connection between a client and mitmproxy."""
151
152    peername: Address
153    """The client's address."""
154    sockname: Address
155    """The local address we received this connection on."""
156
157    mitmcert: Optional[certs.Cert] = None
158    """
159    The certificate used by mitmproxy to establish TLS with the client.
160    """
161
162    proxy_mode: mode_specs.ProxyMode
163    """The proxy server type this client has been connecting to."""
164
165    timestamp_start: float
166    """*Timestamp:* TCP SYN received"""
167
168    def __init__(
169        self,
170        peername: Address,
171        sockname: Address,
172        timestamp_start: float,
173        *,
174        transport_protocol: TransportProtocol = "tcp",
175        proxy_mode: mode_specs.ProxyMode = mode_specs.ProxyMode.parse("regular"),
176    ):
177        self.id = str(uuid.uuid4())
178        self.peername = peername
179        self.sockname = sockname
180        self.timestamp_start = timestamp_start
181        self.state = ConnectionState.OPEN
182        self.transport_protocol = transport_protocol
183        self.proxy_mode = proxy_mode
184
185    def __str__(self):
186        if self.alpn:
187            tls_state = f", alpn={self.alpn.decode(errors='replace')}"
188        elif self.tls_established:
189            tls_state = ", tls"
190        else:
191            tls_state = ""
192        return f"Client({human.format_address(self.peername)}, state={self.state.name.lower()}{tls_state})"
193
194    def get_state(self):
195        # Important: Retain full compatibility with old proxy core for now!
196        # This means we need to add all new fields to the old implementation.
197        return {
198            "address": self.peername,
199            "alpn": self.alpn,
200            "cipher_name": self.cipher,
201            "id": self.id,
202            "mitmcert": self.mitmcert.get_state()
203            if self.mitmcert is not None
204            else None,
205            "sni": self.sni,
206            "timestamp_end": self.timestamp_end,
207            "timestamp_start": self.timestamp_start,
208            "timestamp_tls_setup": self.timestamp_tls_setup,
209            "tls_established": self.tls_established,
210            "tls_extensions": [],
211            "tls_version": self.tls_version,
212            # only used in sans-io
213            "state": self.state.value,
214            "sockname": self.sockname,
215            "error": self.error,
216            "tls": self.tls,
217            "certificate_list": [x.get_state() for x in self.certificate_list],
218            "alpn_offers": self.alpn_offers,
219            "cipher_list": self.cipher_list,
220            "proxy_mode": self.proxy_mode.get_state(),
221        }
222
223    @classmethod
224    def from_state(cls, state) -> "Client":
225        client = Client(state["address"], ("mitmproxy", 8080), state["timestamp_start"])
226        client.set_state(state)
227        return client
228
229    def set_state(self, state):
230        self.peername = tuple(state["address"]) if state["address"] else None
231        self.alpn = state["alpn"]
232        self.cipher = state["cipher_name"]
233        self.id = state["id"]
234        self.sni = state["sni"]
235        self.timestamp_end = state["timestamp_end"]
236        self.timestamp_start = state["timestamp_start"]
237        self.timestamp_tls_setup = state["timestamp_tls_setup"]
238        self.tls_version = state["tls_version"]
239        # only used in sans-io
240        self.state = ConnectionState(state["state"])
241        self.sockname = tuple(state["sockname"]) if state["sockname"] else None
242        self.error = state["error"]
243        self.tls = state["tls"]
244        self.certificate_list = [
245            certs.Cert.from_state(x) for x in state["certificate_list"]
246        ]
247        self.mitmcert = (
248            certs.Cert.from_state(state["mitmcert"])
249            if state["mitmcert"] is not None
250            else None
251        )
252        self.alpn_offers = state["alpn_offers"]
253        self.cipher_list = state["cipher_list"]
254        self.proxy_mode = mode_specs.ProxyMode.from_state(state["proxy_mode"])
255
256    @property
257    def address(self):  # pragma: no cover
258        """*Deprecated:* An outdated alias for Client.peername."""
259        warnings.warn(
260            "Client.address is deprecated, use Client.peername instead.",
261            DeprecationWarning,
262            stacklevel=2,
263        )
264        return self.peername
265
266    @address.setter
267    def address(self, x):  # pragma: no cover
268        warnings.warn(
269            "Client.address is deprecated, use Client.peername instead.",
270            DeprecationWarning,
271            stacklevel=2,
272        )
273        self.peername = x
274
275    @property
276    def cipher_name(self) -> Optional[str]:  # pragma: no cover
277        """*Deprecated:* An outdated alias for Connection.cipher."""
278        warnings.warn(
279            "Client.cipher_name is deprecated, use Client.cipher instead.",
280            DeprecationWarning,
281            stacklevel=2,
282        )
283        return self.cipher
284
285    @property
286    def clientcert(self) -> Optional[certs.Cert]:  # pragma: no cover
287        """*Deprecated:* An outdated alias for Connection.certificate_list[0]."""
288        warnings.warn(
289            "Client.clientcert is deprecated, use Client.certificate_list instead.",
290            DeprecationWarning,
291            stacklevel=2,
292        )
293        if self.certificate_list:
294            return self.certificate_list[0]
295        else:
296            return None
297
298    @clientcert.setter
299    def clientcert(self, val):  # pragma: no cover
300        warnings.warn(
301            "Client.clientcert is deprecated, use Client.certificate_list instead.",
302            DeprecationWarning,
303            stacklevel=2,
304        )
305        if val:
306            self.certificate_list = [val]
307        else:
308            self.certificate_list = []
309
310
311class Server(Connection):
312    """A connection between mitmproxy and an upstream server."""
313
314    peername: Optional[Address] = None
315    """The server's resolved `(ip, port)` tuple. Will be set during connection establishment."""
316    sockname: Optional[Address] = None
317    address: Optional[Address]
318    """The server's `(host, port)` address tuple. The host can either be a domain or a plain IP address."""
319
320    timestamp_start: Optional[float] = None
321    """*Timestamp:* TCP SYN sent."""
322    timestamp_tcp_setup: Optional[float] = None
323    """*Timestamp:* TCP ACK received."""
324
325    via: Optional[server_spec.ServerSpec] = None
326    """An optional proxy server specification via which the connection should be established."""
327
328    def __init__(
329        self,
330        address: Optional[Address],
331        *,
332        transport_protocol: TransportProtocol = "tcp",
333    ):
334        self.id = str(uuid.uuid4())
335        self.address = address
336        self.state = ConnectionState.CLOSED
337        self.transport_protocol = transport_protocol
338
339    def __str__(self):
340        if self.alpn:
341            tls_state = f", alpn={self.alpn.decode(errors='replace')}"
342        elif self.tls_established:
343            tls_state = ", tls"
344        else:
345            tls_state = ""
346        if self.sockname:
347            local_port = f", src_port={self.sockname[1]}"
348        else:
349            local_port = ""
350        return f"Server({human.format_address(self.address)}, state={self.state.name.lower()}{tls_state}{local_port})"
351
352    def __setattr__(self, name, value):
353        if name in ("address", "via"):
354            connection_open = (
355                self.__dict__.get("state", ConnectionState.CLOSED)
356                is ConnectionState.OPEN
357            )
358            # assigning the current value is okay, that may be an artifact of calling .set_state().
359            attr_changed = self.__dict__.get(name) != value
360            if connection_open and attr_changed:
361                raise RuntimeError(f"Cannot change server.{name} on open connection.")
362        return super().__setattr__(name, value)
363
364    def get_state(self):
365        return {
366            "address": self.address,
367            "alpn": self.alpn,
368            "id": self.id,
369            "ip_address": self.peername,
370            "sni": self.sni,
371            "source_address": self.sockname,
372            "timestamp_end": self.timestamp_end,
373            "timestamp_start": self.timestamp_start,
374            "timestamp_tcp_setup": self.timestamp_tcp_setup,
375            "timestamp_tls_setup": self.timestamp_tls_setup,
376            "tls_established": self.tls_established,
377            "tls_version": self.tls_version,
378            "via": None,
379            # only used in sans-io
380            "state": self.state.value,
381            "error": self.error,
382            "tls": self.tls,
383            "certificate_list": [x.get_state() for x in self.certificate_list],
384            "alpn_offers": self.alpn_offers,
385            "cipher_name": self.cipher,
386            "cipher_list": self.cipher_list,
387            "via2": self.via,
388        }
389
390    @classmethod
391    def from_state(cls, state) -> "Server":
392        server = Server(None)
393        server.set_state(state)
394        return server
395
396    def set_state(self, state):
397        self.address = tuple(state["address"]) if state["address"] else None
398        self.alpn = state["alpn"]
399        self.id = state["id"]
400        self.peername = tuple(state["ip_address"]) if state["ip_address"] else None
401        self.sni = state["sni"]
402        self.sockname = (
403            tuple(state["source_address"]) if state["source_address"] else None
404        )
405        self.timestamp_end = state["timestamp_end"]
406        self.timestamp_start = state["timestamp_start"]
407        self.timestamp_tcp_setup = state["timestamp_tcp_setup"]
408        self.timestamp_tls_setup = state["timestamp_tls_setup"]
409        self.tls_version = state["tls_version"]
410        self.state = ConnectionState(state["state"])
411        self.error = state["error"]
412        self.tls = state["tls"]
413        self.certificate_list = [
414            certs.Cert.from_state(x) for x in state["certificate_list"]
415        ]
416        self.alpn_offers = state["alpn_offers"]
417        self.cipher = state["cipher_name"]
418        self.cipher_list = state["cipher_list"]
419        self.via = state["via2"]
420
421    @property
422    def ip_address(self) -> Optional[Address]:  # pragma: no cover
423        """*Deprecated:* An outdated alias for `Server.peername`."""
424        warnings.warn(
425            "Server.ip_address is deprecated, use Server.peername instead.",
426            DeprecationWarning,
427            stacklevel=2,
428        )
429        return self.peername
430
431    @property
432    def cert(self) -> Optional[certs.Cert]:  # pragma: no cover
433        """*Deprecated:* An outdated alias for `Connection.certificate_list[0]`."""
434        warnings.warn(
435            "Server.cert is deprecated, use Server.certificate_list instead.",
436            DeprecationWarning,
437            stacklevel=2,
438        )
439        if self.certificate_list:
440            return self.certificate_list[0]
441        else:
442            return None
443
444    @cert.setter
445    def cert(self, val):  # pragma: no cover
446        warnings.warn(
447            "Server.cert is deprecated, use Server.certificate_list instead.",
448            DeprecationWarning,
449            stacklevel=2,
450        )
451        if val:
452            self.certificate_list = [val]
453        else:
454            self.certificate_list = []
455
456
457__all__ = ["Connection", "Client", "Server", "ConnectionState"]