codekingpro/portable-devtools

1import io2from dataclasses import dataclass3 4from kaitaistruct import KaitaiStream5from OpenSSL import SSL6 7from mitmproxy import connection8from mitmproxy.contrib.kaitaistruct import dtls_client_hello9from mitmproxy.contrib.kaitaistruct import tls_client_hello10from mitmproxy.net import check11from mitmproxy.proxy import context12 13 14class ClientHello:15    """16    A TLS ClientHello is the first message sent by the client when initiating TLS.17    """18 19    _raw_bytes: bytes20 21    def __init__(self, raw_client_hello: bytes, dtls: bool = False):22        """Create a TLS ClientHello object from raw bytes."""23        self._raw_bytes = raw_client_hello24        if dtls:25            self._client_hello = dtls_client_hello.DtlsClientHello(26                KaitaiStream(io.BytesIO(raw_client_hello))27            )28        else:29            self._client_hello = tls_client_hello.TlsClientHello(30                KaitaiStream(io.BytesIO(raw_client_hello))31            )32 33    def raw_bytes(self, wrap_in_record: bool = True) -> bytes:34        """35        The raw ClientHello bytes as seen on the wire.36 37        If `wrap_in_record` is True, the ClientHello will be wrapped in a synthetic TLS record38        (`0x160303 + len(chm) + 0x01 + len(ch)`), which is the format expected by some tools.39        The synthetic record assumes TLS version (`0x0303`), which may be different from what has been sent over the40        wire. JA3 hashes are unaffected by this as they only use the TLS version from the ClientHello data structure.41 42        A future implementation may return not just the exact ClientHello, but also the exact record(s) as seen on the43        wire.44        """45        if isinstance(self._client_hello, dtls_client_hello.DtlsClientHello):46            raise NotImplementedError47 48        if wrap_in_record:49            return (50                # record layer51                b"\x16\x03\x03"52                + (len(self._raw_bytes) + 4).to_bytes(2, byteorder="big")53                +54                # handshake header55                b"\x01"56                + len(self._raw_bytes).to_bytes(3, byteorder="big")57                +58                # ClientHello as defined in https://datatracker.ietf.org/doc/html/rfc8446#section-4.1.2.59                self._raw_bytes60            )61        else:62            return self._raw_bytes63 64    @property65    def cipher_suites(self) -> list[int]:66        """The cipher suites offered by the client (as raw ints)."""67        return self._client_hello.cipher_suites.cipher_suites68 69    @property70    def sni(self) -> str | None:71        """72        The [Server Name Indication](https://en.wikipedia.org/wiki/Server_Name_Indication),73        which indicates which hostname the client wants to connect to.74        """75        if ext := getattr(self._client_hello, "extensions", None):76            for extension in ext.extensions:77                is_valid_sni_extension = (78                    extension.type == 0x0079                    and len(extension.body.server_names) == 180                    and extension.body.server_names[0].name_type == 081                    and check.is_valid_host(extension.body.server_names[0].host_name)82                )83                if is_valid_sni_extension:84                    return extension.body.server_names[0].host_name.decode("ascii")85        return None86 87    @property88    def alpn_protocols(self) -> list[bytes]:89        """90        The application layer protocols offered by the client as part of the91        [ALPN](https://en.wikipedia.org/wiki/Application-Layer_Protocol_Negotiation) TLS extension.92        """93        if ext := getattr(self._client_hello, "extensions", None):94            for extension in ext.extensions:95                if extension.type == 0x10:96                    return list(x.name for x in extension.body.alpn_protocols)97        return []98 99    @property100    def extensions(self) -> list[tuple[int, bytes]]:101        """The raw list of extensions in the form of `(extension_type, raw_bytes)` tuples."""102        ret = []103        if ext := getattr(self._client_hello, "extensions", None):104            for extension in ext.extensions:105                body = getattr(extension, "_raw_body", extension.body)106                ret.append((extension.type, body))107        return ret108 109    def __repr__(self):110        return f"ClientHello(sni: {self.sni}, alpn_protocols: {self.alpn_protocols})"111 112 113@dataclass114class ClientHelloData:115    """116    Event data for `tls_clienthello` event hooks.117    """118 119    context: context.Context120    """The context object for this connection."""121    client_hello: ClientHello122    """The entire parsed TLS ClientHello."""123    ignore_connection: bool = False124    """125    If set to `True`, do not intercept this connection and forward encrypted contents unmodified.126    """127    establish_server_tls_first: bool = False128    """129    If set to `True`, pause this handshake and establish TLS with an upstream server first.130    This makes it possible to process the server certificate when generating an interception certificate.131    """132 133 134@dataclass135class TlsData:136    """137    Event data for `tls_start_client`, `tls_start_server`, and `tls_handshake` event hooks.138    """139 140    conn: connection.Connection141    """The affected connection."""142    context: context.Context143    """The context object for this connection."""144    ssl_conn: SSL.Connection | None = None145    """146    The associated pyOpenSSL `SSL.Connection` object.147    This will be set by an addon in the `tls_start_*` event hooks.148    """149    is_dtls: bool = False150    """151    If set to `True`, indicates that it is a DTLS event.152    """153