# SPDX-License-Identifier: GPL-2.0-only # This file is part of Scapy # See https://scapy.net/ for more information # Copyright (C) 2007, 2008, 2009 Arnaud Ebalard # 2015, 2016, 2017 Maxence Tury # 2019 Romain Perez """ TLS session handler. """ import binascii import collections import socket import struct from scapy.config import conf from scapy.compat import raw from scapy.error import log_runtime, warning from scapy.packet import Packet from scapy.pton_ntop import inet_pton from scapy.sessions import TCPSession from scapy.utils import repr_hex, strxor from scapy.layers.inet import TCP from scapy.layers.tls.crypto.compression import Comp_NULL from scapy.layers.tls.crypto.hkdf import TLS13_HKDF from scapy.layers.tls.crypto.prf import PRF # Typing imports from typing import Dict def load_nss_keys(filename): # type: (str) -> Dict[str, bytes] """ Parses a NSS Keys log and returns unpacked keys in a dictionary. """ # http://udn.realityripple.com/docs/Mozilla/Projects/NSS/Key_Log_Format keys = collections.defaultdict(dict) try: fd = open(filename) fd.close() except FileNotFoundError: warning("Cannot open NSS Key Log: %s", filename) return {} try: with open(filename) as fd: for line in fd: if line.startswith("#"): continue data = line.strip().split(" ") if len(data) != 3 or data[0] != data[0].upper(): warning("Invalid NSS Key Log Entry: %s", line.strip()) return {} try: client_random = binascii.unhexlify(data[1]) except ValueError: warning("Invalid ClientRandom: %s", data[1]) return {} try: secret = binascii.unhexlify(data[2]) except ValueError: warning("Invalid Secret: %s", data[2]) return {} # Warn that a duplicated entry was detected. The latest one # will be kept in the resulting dictionary. if client_random in keys[data[0]]: warning("Duplicated entry for %s !", data[0]) keys[data[0]][client_random] = secret return keys except UnicodeDecodeError as ex: warning("Cannot read NSS Key Log: %s %s", filename, str(ex)) return {} # Note the following import may happen inside connState.__init__() # in order to avoid to avoid cyclical dependencies. # from scapy.layers.tls.crypto.suites import TLS_NULL_WITH_NULL_NULL ############################################################################### # Connection states # ############################################################################### class connState(object): """ From RFC 5246, section 6.1: A TLS connection state is the operating environment of the TLS Record Protocol. It specifies a compression algorithm, an encryption algorithm, and a MAC algorithm. In addition, the parameters for these algorithms are known: the MAC key and the bulk encryption keys for the connection in both the read and the write directions. Logically, there are always four connection states outstanding: the current read and write states, and the pending read and write states. All records are processed under the current read and write states. The security parameters for the pending states can be set by the TLS Handshake Protocol, and the ChangeCipherSpec can selectively make either of the pending states current, in which case the appropriate current state is disposed of and replaced with the pending state; the pending state is then reinitialized to an empty state. It is illegal to make a state that has not been initialized with security parameters a current state. The initial current state always specifies that no encryption, compression, or MAC will be used. (For practical reasons, Scapy scraps these two last lines, through the implementation of dummy ciphers and MAC with TLS_NULL_WITH_NULL_NULL.) These attributes and behaviours are mostly mapped in this class. Also, note that Scapy may make a current state out of a pending state which has been initialized with dummy security parameters. We need this in order to know when the content of a TLS message is encrypted, whether we possess the right keys to decipher/verify it or not. For instance, when Scapy parses a CKE without knowledge of any secret, and then a CCS, it needs to know that the following Finished is encrypted and signed according to a new cipher suite, even though it cannot decipher the message nor verify its integrity. """ def __init__(self, connection_end="server", read_or_write="read", seq_num=0, compression_alg=Comp_NULL, ciphersuite=None, tls_version=0x0303): self.tls_version = tls_version # It is the user's responsibility to keep the record seq_num # under 2**64-1. If this value gets maxed out, the TLS class in # record.py will crash when trying to encode it with struct.pack(). self.seq_num = seq_num self.connection_end = connection_end self.row = read_or_write if ciphersuite is None: from scapy.layers.tls.crypto.suites import TLS_NULL_WITH_NULL_NULL ciphersuite = TLS_NULL_WITH_NULL_NULL self.ciphersuite = ciphersuite(tls_version=tls_version) if not self.ciphersuite.usable: warning("TLS cipher suite not usable. " "Is the cryptography Python module installed?") return self.compression = compression_alg() self.key_exchange = ciphersuite.kx_alg() self.cipher = ciphersuite.cipher_alg() self.hash = ciphersuite.hash_alg() if tls_version > 0x0200: if ciphersuite.cipher_alg.type == "aead": self.hmac = None self.mac_len = self.cipher.tag_len else: self.hmac = ciphersuite.hmac_alg() self.mac_len = self.hmac.hmac_len else: self.hmac = ciphersuite.hmac_alg() # should be Hmac_NULL self.mac_len = self.hash.hash_len if tls_version >= 0x0304: self.hkdf = TLS13_HKDF(self.hash.name.lower()) else: self.prf = PRF(ciphersuite.hash_alg.name, tls_version) def debug_repr(self, name, secret): if conf.debug_tls and secret: log_runtime.debug("TLS: %s %s %s: %s", self.connection_end, self.row, name, repr_hex(secret)) def derive_keys(self, client_random=b"", server_random=b"", master_secret=b""): # XXX Can this be called over a non-usable suite? What happens then? cs = self.ciphersuite # Derive the keys according to the cipher type and protocol version key_block = self.prf.derive_key_block(master_secret, server_random, client_random, cs.key_block_len) # When slicing the key_block, keep the right half of the material skip_first = False if ((self.connection_end == "client" and self.row == "read") or (self.connection_end == "server" and self.row == "write")): skip_first = True pos = 0 cipher_alg = cs.cipher_alg # MAC secret (for block and stream ciphers) if (cipher_alg.type == "stream") or (cipher_alg.type == "block"): start = pos if skip_first: start += cs.hmac_alg.key_len end = start + cs.hmac_alg.key_len mac_secret = key_block[start:end] self.debug_repr("mac_secret", mac_secret) pos += 2 * cs.hmac_alg.key_len else: mac_secret = None # Cipher secret start = pos if skip_first: start += cipher_alg.key_len end = start + cipher_alg.key_len cipher_secret = key_block[start:end] if cs.kx_alg.export: reqLen = cipher_alg.expanded_key_len cipher_secret = self.prf.postprocess_key_for_export(cipher_secret, client_random, server_random, self.connection_end, # noqa: E501 self.row, reqLen) self.debug_repr("cipher_secret", cipher_secret) pos += 2 * cipher_alg.key_len # Implicit IV (for block and AEAD ciphers) start = pos if cipher_alg.type == "block": if skip_first: start += cipher_alg.block_size end = start + cipher_alg.block_size elif cipher_alg.type == "aead": if skip_first: start += cipher_alg.fixed_iv_len end = start + cipher_alg.fixed_iv_len # Now we have the secrets, we can instantiate the algorithms if cs.hmac_alg is None: # AEAD self.hmac = None self.mac_len = cipher_alg.tag_len else: self.hmac = cs.hmac_alg(mac_secret) self.mac_len = self.hmac.hmac_len if cipher_alg.type == "stream": cipher = cipher_alg(cipher_secret) elif cipher_alg.type == "block": # We set an IV every time, even though it does not matter for # TLS 1.1+ as it requires an explicit IV. Indeed the cipher.iv # would get updated in TLS.post_build() or TLS.pre_dissect(). iv = key_block[start:end] if cs.kx_alg.export: reqLen = cipher_alg.block_size iv = self.prf.generate_iv_for_export(client_random, server_random, self.connection_end, self.row, reqLen) cipher = cipher_alg(cipher_secret, iv) self.debug_repr("block iv", iv) elif cipher_alg.type == "aead": fixed_iv = key_block[start:end] nonce_explicit_init = 0 # If you ever wanted to set a random nonce_explicit, use this: # exp_bit_len = cipher_alg.nonce_explicit_len * 8 # nonce_explicit_init = random.randint(0, 2**exp_bit_len - 1) cipher = cipher_alg(cipher_secret, fixed_iv, nonce_explicit_init) self.debug_repr("aead fixed iv", fixed_iv) self.cipher = cipher def sslv2_derive_keys(self, key_material): """ There is actually only one key, the CLIENT-READ-KEY or -WRITE-KEY. Note that skip_first is opposite from the one with SSLv3 derivation. Also, if needed, the IV should be set elsewhere. """ skip_first = True if ((self.connection_end == "client" and self.row == "read") or (self.connection_end == "server" and self.row == "write")): skip_first = False cipher_alg = self.ciphersuite.cipher_alg start = 0 if skip_first: start += cipher_alg.key_len end = start + cipher_alg.key_len cipher_secret = key_material[start:end] self.cipher = cipher_alg(cipher_secret) self.debug_repr("cipher_secret", cipher_secret) def tls13_derive_keys(self, key_material): cipher_alg = self.ciphersuite.cipher_alg key_len = cipher_alg.key_len iv_len = cipher_alg.fixed_iv_len write_key = self.hkdf.expand_label(key_material, b"key", b"", key_len) write_iv = self.hkdf.expand_label(key_material, b"iv", b"", iv_len) self.cipher = cipher_alg(write_key, write_iv) def snapshot(self): """ This is used mostly as a way to keep the cipher state and the seq_num. """ snap = connState(connection_end=self.connection_end, read_or_write=self.row, seq_num=self.seq_num, compression_alg=type(self.compression), ciphersuite=type(self.ciphersuite), tls_version=self.tls_version) snap.cipher = self.cipher.snapshot() if self.hmac: snap.hmac.key = self.hmac.key return snap def __repr__(self): res = "Connection end : %s\n" % self.connection_end.upper() res += "Cipher suite : %s (0x%04x)\n" % (self.ciphersuite.name, self.ciphersuite.val) res += "Compression : %s (0x%02x)\n" % (self.compression.name, self.compression.val) tabsize = 4 return res.expandtabs(tabsize) class readConnState(connState): def __init__(self, **kargs): connState.__init__(self, read_or_write="read", **kargs) class writeConnState(connState): def __init__(self, **kargs): connState.__init__(self, read_or_write="write", **kargs) ############################################################################### # TLS session # ############################################################################### class tlsSession(object): """ This is our TLS context, which gathers information from both sides of the TLS connection. These sides are represented by a readConnState instance and a writeConnState instance. Along with overarching network attributes, a tlsSession object also holds negotiated, shared information, such as the key exchange parameters and the master secret (when available). The default connection_end is "server". This corresponds to the expected behaviour for static exchange analysis (with a ClientHello parsed first). """ def __init__(self, ipsrc=None, ipdst=None, sport=None, dport=None, sid=None, connection_end="server", wcs=None, rcs=None): # Use this switch to prevent additions to the 'handshake_messages'. self.frozen = False # Network settings self.ipsrc = ipsrc self.ipdst = ipdst self.sport = sport self.dport = dport self.sid = sid # Identify duplicate sessions self.firsttcp = None # Our TCP socket. None until we send (or receive) a packet. self.sock = None # Connection states self.connection_end = connection_end if wcs is None: # Instantiate wcs with dummy values. self.wcs = writeConnState(connection_end=connection_end) self.wcs.derive_keys() else: self.wcs = wcs if rcs is None: # Instantiate rcs with dummy values. self.rcs = readConnState(connection_end=connection_end) self.rcs.derive_keys() else: self.rcs = rcs # The pending write/read states are updated by the building/parsing # of various TLS packets. They get committed to self.wcs/self.rcs # once Scapy builds/parses a ChangeCipherSpec message, or for certain # other messages in case of TLS 1.3. self.pwcs = None self.triggered_pwcs_commit = False self.prcs = None self.triggered_prcs_commit = False # Certificates and private keys # The server certificate chain, as a list of Cert instances. # Either we act as server and it has to be provided, or it is expected # to be sent by the server through a Certificate message. # The server certificate should be self.server_certs[0]. self.server_certs = [] # The server private key, as a PrivKey instance, when acting as server. # XXX It would be nice to be able to provide both an RSA and an ECDSA # key in order for the same Scapy server to support both families of # cipher suites. See INIT_TLS_SESSION() in automaton_srv.py. # (For now server_key holds either one of both types for DHE # authentication, while server_rsa_key is used only for RSAkx.) self.server_key = None self.server_rsa_key = None # self.server_ecdsa_key = None # A dictionary containing keys extracted from a NSS Keys Log using # the load_nss_keys() function. self.nss_keys = None # Back in the dreadful EXPORT days, US servers were forbidden to use # RSA keys longer than 512 bits for RSAkx. When their usual RSA key # was longer than this, they had to create a new key and send it via # a ServerRSAParams message. When receiving such a message, # Scapy stores this key in server_tmp_rsa_key as a PubKey instance. self.server_tmp_rsa_key = None # When client authentication is performed, we need at least a # client certificate chain. If we act as client, we also have # to provide the key associated with the first certificate. self.client_certs = [] self.client_key = None # Ephemeral key exchange parameters # The agreed-upon ephemeral key group self.kx_group = None # These are the group/curve parameters, needed to hold the information # e.g. from receiving an SKE to sending a CKE. Usually, only one of # these attributes will be different from None. self.client_kx_ffdh_params = None self.client_kx_ecdh_params = None # These are PrivateKeys and PublicKeys from the appropriate FFDH/ECDH # cryptography module, i.e. these are not raw bytes. Usually, only one # in two will be different from None, e.g. when being a TLS client you # will need the client_kx_privkey (the serialized public key is not # actually registered) and you will receive a server_kx_pubkey. self.client_kx_privkey = None self.client_kx_pubkey = None self.server_kx_privkey = None self.server_kx_pubkey = None # When using TLS 1.3, the tls13_client_pubshares will contain every # potential key share (equate the 'client_kx_pubkey' before) the client # offered, indexed by the id of the FFDH/ECDH group. These dicts # effectively replace the four previous attributes. self.tls13_client_privshares = {} self.tls13_client_pubshares = {} self.tls13_server_privshare = {} self.tls13_server_pubshare = {} # Negotiated session parameters # The advertised TLS version found in the ClientHello (and # EncryptedPreMasterSecret if used). If acting as server, it is set to # the value advertised by the client in its ClientHello. # The default value corresponds to TLS 1.2 (and TLS 1.3, incidentally). self.advertised_tls_version = 0x0303 # The agreed-upon TLS version found in the ServerHello. self.tls_version = None # These attributes should eventually be known to both sides (SSLv3-TLS 1.2). # noqa: E501 self.client_random = None self.server_random = None self.pre_master_secret = None self.master_secret = None # The advertised supported signature algorithms found in the ClientHello # extension. (for TLS 1.2-TLS 1.3 only) self.advertised_sig_algs = [] # The agreed-upon signature algorithm (for TLS 1.2-TLS 1.3 only) self.selected_sig_alg = None # A session ticket received by the client. self.client_session_ticket = None # These attributes should only be used with SSLv2 connections. # We need to keep the KEY-MATERIAL here because it may be reused. self.sslv2_common_cs = [] self.sslv2_connection_id = None self.sslv2_challenge = None self.sslv2_challenge_clientcert = None self.sslv2_key_material = None # These attributes should only be used with TLS 1.3 connections. self.tls13_psk_secret = None self.tls13_early_secret = None self.tls13_dhe_secret = None self.tls13_handshake_secret = None self.tls13_master_secret = None self.tls13_derived_secrets = {} self.tls13_cert_req_ctxt = False self.post_handshake = False # whether handshake is done self.post_handshake_auth = False # whether "Post-Handshake Auth" is used self.tls13_ticket_ciphersuite = None self.tls13_retry = False self.middlebox_compatibility = False # Handshake messages needed for Finished computation/validation. # No record layer headers, no HelloRequests, no ChangeCipherSpecs. self.handshake_messages = [] self.handshake_messages_parsed = [] # Post-handshake, handshake messages for post-handshake client authentication self.post_handshake_messages = [] # Flag, whether we derive the secret as Extended MS or not self.extms = False self.session_hash = None self.encrypt_then_mac = False # All exchanged TLS packets. # XXX no support for now # self.exchanged_pkts = [] def __setattr__(self, name, val): if name == "connection_end": if hasattr(self, "rcs") and self.rcs: self.rcs.connection_end = val if hasattr(self, "wcs") and self.wcs: self.wcs.connection_end = val if hasattr(self, "prcs") and self.prcs: self.prcs.connection_end = val if hasattr(self, "pwcs") and self.pwcs: self.pwcs.connection_end = val super(tlsSession, self).__setattr__(name, val) # Get infos from underlayer def set_underlayer(self, _underlayer): if isinstance(_underlayer, TCP): tcp = _underlayer self.sport = tcp.sport self.dport = tcp.dport try: self.ipsrc = tcp.underlayer.src self.ipdst = tcp.underlayer.dst except AttributeError: pass if self.firsttcp is None: self.firsttcp = tcp.seq # Mirroring def mirror(self): """ This function takes a tlsSession object and swaps the IP addresses, ports, connection ends and connection states. The triggered_commit are also swapped (though it is probably overkill, it is cleaner this way). It is useful for static analysis of a series of messages from both the client and the server. In such a situation, it should be used every time the message being read comes from a different side than the one read right before, as the reading state becomes the writing state, and vice versa. For instance you could do:: client_hello = open('client_hello.raw').read() m1 = TLS(client_hello) m2 = TLS(server_hello, tls_session=m1.tls_session.mirror()) m3 = TLS(server_cert, tls_session=m2.tls_session) m4 = TLS(client_keyexchange, tls_session=m3.tls_session.mirror()) """ self.ipdst, self.ipsrc = self.ipsrc, self.ipdst self.dport, self.sport = self.sport, self.dport self.rcs, self.wcs = self.wcs, self.rcs if self.rcs: self.rcs.row = "read" if self.wcs: self.wcs.row = "write" self.prcs, self.pwcs = self.pwcs, self.prcs if self.prcs: self.prcs.row = "read" if self.pwcs: self.pwcs.row = "write" self.triggered_prcs_commit, self.triggered_pwcs_commit = \ self.triggered_pwcs_commit, self.triggered_prcs_commit if self.connection_end == "client": self.connection_end = "server" elif self.connection_end == "server": self.connection_end = "client" return self # Secrets management for SSLv3 to TLS 1.2 def compute_master_secret(self): if self.pre_master_secret is None: warning("Missing pre_master_secret while computing master_secret!") if self.client_random is None: warning("Missing client_random while computing master_secret!") if self.server_random is None: warning("Missing server_random while computing master_secret!") if self.extms and self.session_hash is None: warning("Missing session hash while computing master secret!") ms = self.pwcs.prf.compute_master_secret(self.pre_master_secret, self.client_random, self.server_random, self.extms, self.session_hash) self.master_secret = ms if conf.debug_tls: log_runtime.debug("TLS: master secret: %s", repr_hex(ms)) def use_nss_master_secret_if_present(self) -> bool: # Load the master secret from an NSS Key dictionary if not self.nss_keys or "CLIENT_RANDOM" not in self.nss_keys: return False if self.client_random in self.nss_keys["CLIENT_RANDOM"]: self.master_secret = self.nss_keys["CLIENT_RANDOM"][self.client_random] return True return False def compute_ms_and_derive_keys(self): if not self.master_secret: self.compute_master_secret() self.prcs.derive_keys(client_random=self.client_random, server_random=self.server_random, master_secret=self.master_secret) self.pwcs.derive_keys(client_random=self.client_random, server_random=self.server_random, master_secret=self.master_secret) # Secrets management for SSLv2 def compute_sslv2_key_material(self): if self.master_secret is None: warning("Missing master_secret while computing key_material!") if self.sslv2_challenge is None: warning("Missing challenge while computing key_material!") if self.sslv2_connection_id is None: warning("Missing connection_id while computing key_material!") km = self.pwcs.prf.derive_key_block(self.master_secret, self.sslv2_challenge, self.sslv2_connection_id, 2 * self.pwcs.cipher.key_len) self.sslv2_key_material = km if conf.debug_tls: log_runtime.debug("TLS: master secret: %s", repr_hex(self.master_secret)) # noqa: E501 log_runtime.debug("TLS: key material: %s", repr_hex(km)) def compute_sslv2_km_and_derive_keys(self): self.compute_sslv2_key_material() self.prcs.sslv2_derive_keys(key_material=self.sslv2_key_material) self.pwcs.sslv2_derive_keys(key_material=self.sslv2_key_material) # Secrets management for TLS 1.3 def compute_tls13_early_secrets(self, external=False): """ This function computes the Early Secret, the binder_key, the client_early_traffic_secret and the early_exporter_master_secret (See RFC8446, section 7.1). The parameter external is used for the computation of the binder_key: - For external PSK provisioned outside out of TLS, the parameter external must be True. - For resumption PSK, the parameter external must be False. If no argument is specified, the label "res binder" will be used by default. Ciphers key and IV are updated accordingly for 0-RTT data. self.handshake_messages should be ClientHello only. """ # if no hash algorithm is set, default to SHA-256 if self.prcs and self.prcs.hkdf: hkdf = self.prcs.hkdf elif self.pwcs and self.pwcs.hkdf: hkdf = self.pwcs.hkdf else: hkdf = TLS13_HKDF("sha256") if self.tls13_early_secret is None: self.tls13_early_secret = hkdf.extract(None, self.tls13_psk_secret) if "binder_key" not in self.tls13_derived_secrets: if external: bk = hkdf.derive_secret(self.tls13_early_secret, b"ext binder", b"") else: bk = hkdf.derive_secret(self.tls13_early_secret, b"res binder", b"") self.tls13_derived_secrets["binder_key"] = bk cets = hkdf.derive_secret(self.tls13_early_secret, b"c e traffic", b"".join(self.handshake_messages)) self.tls13_derived_secrets["client_early_traffic_secret"] = cets ees = hkdf.derive_secret(self.tls13_early_secret, b"e exp master", b"".join(self.handshake_messages)) self.tls13_derived_secrets["early_exporter_secret"] = ees if self.connection_end == "server": if self.prcs: self.prcs.tls13_derive_keys(cets) elif self.connection_end == "client": if self.pwcs: self.pwcs.tls13_derive_keys(cets) def compute_tls13_handshake_secrets(self): """ Ciphers key and IV are updated accordingly for Handshake data. self.handshake_messages should be ClientHello...ServerHello. """ if self.prcs: hkdf = self.prcs.hkdf elif self.pwcs: hkdf = self.pwcs.hkdf else: warning("No HKDF. This is abnormal.") return if self.tls13_early_secret is None: self.tls13_early_secret = hkdf.extract(None, self.tls13_psk_secret) secret = hkdf.derive_secret(self.tls13_early_secret, b"derived", b"") self.tls13_handshake_secret = hkdf.extract(secret, self.tls13_dhe_secret) # noqa: E501 chts = hkdf.derive_secret(self.tls13_handshake_secret, b"c hs traffic", b"".join(self.handshake_messages)) self.tls13_derived_secrets["client_handshake_traffic_secret"] = chts shts = hkdf.derive_secret(self.tls13_handshake_secret, b"s hs traffic", b"".join(self.handshake_messages)) self.tls13_derived_secrets["server_handshake_traffic_secret"] = shts def compute_tls13_traffic_secrets(self): """ Ciphers key and IV are updated accordingly for Application data. self.handshake_messages should be ClientHello...ServerFinished. """ if self.prcs and self.prcs.hkdf: hkdf = self.prcs.hkdf elif self.pwcs and self.pwcs.hkdf: hkdf = self.pwcs.hkdf else: warning("No HKDF. This is abnormal.") return tmp = hkdf.derive_secret(self.tls13_handshake_secret, b"derived", b"") self.tls13_master_secret = hkdf.extract(tmp, None) cts0 = hkdf.derive_secret(self.tls13_master_secret, b"c ap traffic", b"".join(self.handshake_messages)) self.tls13_derived_secrets["client_traffic_secrets"] = [cts0] sts0 = hkdf.derive_secret(self.tls13_master_secret, b"s ap traffic", b"".join(self.handshake_messages)) self.tls13_derived_secrets["server_traffic_secrets"] = [sts0] es = hkdf.derive_secret(self.tls13_master_secret, b"exp master", b"".join(self.handshake_messages)) self.tls13_derived_secrets["exporter_secret"] = es if self.connection_end == "server": # self.prcs.tls13_derive_keys(cts0) self.pwcs.tls13_derive_keys(sts0) elif self.connection_end == "client": # self.pwcs.tls13_derive_keys(cts0) self.prcs.tls13_derive_keys(sts0) def compute_tls13_traffic_secrets_end(self): cts0 = self.tls13_derived_secrets["client_traffic_secrets"][0] if self.connection_end == "server": self.prcs.tls13_derive_keys(cts0) elif self.connection_end == "client": self.pwcs.tls13_derive_keys(cts0) def compute_tls13_verify_data(self, connection_end, read_or_write, handshake_context): # RFC8446 - 4.4 # +-----------+-------------------------+-----------------------------+ # | Mode | Handshake Context | Base Key | # +-----------+-------------------------+-----------------------------+ # | Server | ClientHello ... later | server_handshake_traffic_ | # | | of EncryptedExtensions/ | secret | # | | CertificateRequest | | # | | | | # | Client | ClientHello ... later | client_handshake_traffic_ | # | | of server | secret | # | | Finished/EndOfEarlyData | | # | | | | # | Post- | ClientHello ... client | client_application_traffic_ | # | Handshake | Finished + | secret_N | # | | CertificateRequest | | # +-----------+-------------------------+-----------------------------+ if self.post_handshake: # RFC8446 - 4.6 # TLS also allows other messages to be sent after the main handshake. # These messages use a handshake content type and are encrypted under # the appropriate application traffic key. shts = self.tls13_derived_secrets["server_traffic_secrets"][-1] chts = self.tls13_derived_secrets["client_traffic_secrets"][-1] else: shts = self.tls13_derived_secrets["server_handshake_traffic_secret"] chts = self.tls13_derived_secrets["client_handshake_traffic_secret"] if read_or_write == "read": hkdf = self.rcs.hkdf if connection_end == "client": basekey = shts elif connection_end == "server": basekey = chts elif read_or_write == "write": hkdf = self.wcs.hkdf if connection_end == "client": basekey = chts elif connection_end == "server": basekey = shts if not hkdf or not basekey: warning("Missing arguments for verify_data computation!") return None return hkdf.compute_verify_data(basekey, handshake_context) def compute_tls13_resumption_secret(self): """ self.handshake_messages should be ClientHello...ClientFinished. """ if self.connection_end == "server": hkdf = self.prcs.hkdf elif self.connection_end == "client": hkdf = self.pwcs.hkdf rs = hkdf.derive_secret(self.tls13_master_secret, b"res master", b"".join(self.handshake_messages)) self.tls13_derived_secrets["resumption_secret"] = rs def compute_tls13_next_traffic_secrets(self, connection_end, read_or_write): # noqa : E501 """ Ciphers key and IV are updated accordingly. """ if self.rcs.hkdf: hkdf = self.rcs.hkdf hl = hkdf.hash.digest_size elif self.wcs.hkdf: hkdf = self.wcs.hkdf hl = hkdf.hash.digest_size if read_or_write == "read": if connection_end == "client": cts = self.tls13_derived_secrets["client_traffic_secrets"] ctsN = cts[-1] ctsN_1 = hkdf.expand_label(ctsN, b"traffic upd", b"", hl) cts.append(ctsN_1) self.prcs.tls13_derive_keys(ctsN_1) elif connection_end == "server": sts = self.tls13_derived_secrets["server_traffic_secrets"] stsN = sts[-1] stsN_1 = hkdf.expand_label(stsN, b"traffic upd", b"", hl) sts.append(stsN_1) self.prcs.tls13_derive_keys(stsN_1) elif read_or_write == "write": if connection_end == "client": cts = self.tls13_derived_secrets["client_traffic_secrets"] ctsN = cts[-1] ctsN_1 = hkdf.expand_label(ctsN, b"traffic upd", b"", hl) cts.append(ctsN_1) self.pwcs.tls13_derive_keys(ctsN_1) elif connection_end == "server": sts = self.tls13_derived_secrets["server_traffic_secrets"] stsN = sts[-1] stsN_1 = hkdf.expand_label(stsN, b"traffic upd", b"", hl) sts.append(stsN_1) self.pwcs.tls13_derive_keys(stsN_1) # Tests for record building/parsing def consider_read_padding(self): # Return True if padding is needed. Used by TLSPadField. return (self.rcs.cipher.type == "block" and not (False in self.rcs.cipher.ready.values())) def consider_write_padding(self): # Return True if padding is needed. Used by TLSPadField. return self.wcs.cipher.type == "block" def use_explicit_iv(self, version, cipher_type): # Return True if an explicit IV is needed. Required for TLS 1.1+ # when either a block or an AEAD cipher is used. if cipher_type == "stream": return False return version >= 0x0302 # Python object management def hash(self): s1 = struct.pack("!H", self.sport) s2 = struct.pack("!H", self.dport) family = socket.AF_INET if ':' in self.ipsrc: family = socket.AF_INET6 s1 += inet_pton(family, self.ipsrc) s2 += inet_pton(family, self.ipdst) return strxor(s1, s2) def eq(self, other): ok = False if (self.sport == other.sport and self.dport == other.dport and self.ipsrc == other.ipsrc and self.ipdst == other.ipdst): ok = True if (not ok and self.dport == other.sport and self.sport == other.dport and self.ipdst == other.ipsrc and self.ipsrc == other.ipdst): ok = True if ok: if self.sid and other.sid: return self.sid == other.sid return True return False def repr(self, _underlayer=None): sid = repr(self.sid) if len(sid) > 12: sid = sid[:11] + "..." if _underlayer and _underlayer.dport != self.dport: return "%s:%s > %s:%s" % (self.ipdst, str(self.dport), self.ipsrc, str(self.sport)) return "%s:%s > %s:%s" % (self.ipsrc, str(self.sport), self.ipdst, str(self.dport)) def __repr__(self): return self.repr() ############################################################################### # Session singleton # ############################################################################### class _GenericTLSSessionInheritance(Packet): """ Many classes inside the TLS module need to get access to session-related information. For instance, an encrypted TLS record cannot be parsed without some knowledge of the cipher suite being used and the secrets which have been negotiated. Passing information is also essential to the handshake. To this end, various TLS objects inherit from the present class. """ __slots__ = ["tls_session", "rcs_snap_init", "wcs_snap_init"] name = "Dummy Generic TLS Packet" fields_desc = [] def __init__(self, _pkt="", post_transform=None, _internal=0, _underlayer=None, tls_session=None, **fields): try: setme = self.tls_session is None except Exception: setme = True newses = False if setme: if tls_session is None: newses = True self.tls_session = tlsSession() else: self.tls_session = tls_session self.rcs_snap_init = self.tls_session.rcs.snapshot() self.wcs_snap_init = self.tls_session.wcs.snapshot() if isinstance(_underlayer, TCP): # Get information from _underlayer self.tls_session.set_underlayer(_underlayer) # Load a NSS Key Log file if conf.tls_nss_filename is not None: if conf.tls_nss_keys is None: conf.tls_nss_keys = load_nss_keys(conf.tls_nss_filename) if conf.tls_session_enable: if newses: s = conf.tls_sessions.find(self.tls_session) if s: if conf.tls_nss_keys is not None: s.nss_keys = conf.tls_nss_keys if s.dport == self.tls_session.dport: self.tls_session = s else: self.tls_session = s.mirror() else: if conf.tls_nss_keys is not None: self.tls_session.nss_keys = conf.tls_nss_keys conf.tls_sessions.add(self.tls_session) if self.tls_session.connection_end == "server": srk = conf.tls_sessions.server_rsa_key if not self.tls_session.server_rsa_key and \ srk: self.tls_session.server_rsa_key = srk Packet.__init__(self, _pkt=_pkt, post_transform=post_transform, _internal=_internal, _underlayer=_underlayer, **fields) def __getattr__(self, attr): """ The tls_session should be found only through the normal mechanism. """ if attr == "tls_session": return None return super(_GenericTLSSessionInheritance, self).__getattr__(attr) def tls_session_update(self, msg_str): """ post_{build, dissection}_tls_session_update() are used to update the tlsSession context. The default definitions below, along with tls_session_update(), may prevent code duplication in some cases. """ pass def post_build_tls_session_update(self, msg_str): self.tls_session_update(msg_str) def post_dissection_tls_session_update(self, msg_str): self.tls_session_update(msg_str) def copy(self): pkt = Packet.copy(self) pkt.tls_session = self.tls_session return pkt def clone_with(self, payload=None, **kargs): pkt = Packet.clone_with(self, payload=payload, **kargs) pkt.tls_session = self.tls_session return pkt def raw_stateful(self): return super(_GenericTLSSessionInheritance, self).__bytes__() def str_stateful(self): return self.raw_stateful() def __bytes__(self): """ The __bytes__ call has to leave the connection states unchanged. We also have to delete raw_packet_cache in order to access post_build. For performance, the pending connStates are not snapshotted. This should not be an issue, but maybe pay attention to this. The previous_freeze_state prevents issues with calling a raw() calling in turn another raw(), which would unfreeze the session too soon. """ s = self.tls_session rcs_snap = s.rcs.snapshot() wcs_snap = s.wcs.snapshot() rpc_snap = self.raw_packet_cache rpcf_snap = self.raw_packet_cache_fields s.wcs = self.rcs_snap_init self.raw_packet_cache = None previous_freeze_state = s.frozen s.frozen = True built_packet = super(_GenericTLSSessionInheritance, self).__bytes__() s.frozen = previous_freeze_state s.rcs = rcs_snap s.wcs = wcs_snap self.raw_packet_cache = rpc_snap self.raw_packet_cache_fields = rpcf_snap return built_packet __str__ = __bytes__ def show2(self): """ Rebuild the TLS packet with the same context, and then .show() it. We need self.__class__ to call the subclass in a dynamic way. Howether we do not want the tls_session.{r,w}cs.seq_num to be updated. We have to bring back the init states (it's possible the cipher context has been updated because of parsing) but also to keep the current state and restore it afterwards (the raw() call may also update the states). """ s = self.tls_session rcs_snap = s.rcs.snapshot() wcs_snap = s.wcs.snapshot() s.rcs = self.rcs_snap_init built_packet = raw(self) s.frozen = True self.__class__(built_packet, tls_session=s).show() s.frozen = False s.rcs = rcs_snap s.wcs = wcs_snap def mysummary(self, first=True): from scapy.layers.tls.record import TLS from scapy.layers.tls.record_tls13 import TLS13 if ( self.underlayer and isinstance(self.underlayer, _GenericTLSSessionInheritance) ): summary = getattr(self, "_name", self.name) else: _underlayer = None if self.underlayer and isinstance(self.underlayer, TCP): _underlayer = self.underlayer summary = "TLS %s / %s" % ( self.tls_session.repr(_underlayer=_underlayer), getattr(self, "_name", self.name) ) return summary, [TLS, TLS13] @classmethod def tcp_reassemble(cls, data, metadata, session): # Used with TCPSession from scapy.layers.tls.record import TLS from scapy.layers.tls.record_tls13 import TLS13 if cls in (TLS, TLS13): length = struct.unpack("!H", data[3:5])[0] + 5 if len(data) >= length: # get the underlayer as it is used to populate tls_session if "original" not in metadata: return cls(data) underlayer = metadata["original"][TCP].copy() underlayer.remove_payload() # eventually get the tls_session now for TLS.dispatch_hook tls_session = None if conf.tls_session_enable: s = tlsSession() s.set_underlayer(underlayer) tls_session = conf.tls_sessions.find(s) if tls_session: if tls_session.dport != underlayer.dport: tls_session = tls_session.mirror() if tls_session.firsttcp == underlayer.seq: log_runtime.info( "TLS: session %s is a duplicate of a previous " "dissection. Discard it" % repr(tls_session) ) conf.tls_sessions.rem(tls_session, force=True) tls_session = None return cls(data, _underlayer=underlayer, tls_session=tls_session) else: return cls(data) ############################################################################### # Multiple TLS sessions # ############################################################################### class _tls_sessions(object): def __init__(self): self.sessions = {} self.server_rsa_key = None def add(self, session): s = self.find(session) if s: log_runtime.info("TLS: previous session shall not be overwritten") return h = session.hash() if h in self.sessions: self.sessions[h].append(session) else: self.sessions[h] = [session] def rem(self, session, force=False): if not force: s = self.find(session) if s: log_runtime.info("TLS: previous session shall not be overwritten") return h = session.hash() self.sessions[h].remove(session) def find(self, session): try: h = session.hash() except Exception: return None if h in self.sessions: for k in self.sessions[h]: if k.eq(session): if conf.debug_tls: log_runtime.info("TLS: found session matching %s", k) return k if conf.debug_tls: log_runtime.info("TLS: did not find session matching %s", session) return None def __repr__(self): res = [("First endpoint", "Second endpoint", "Session ID")] for li in self.sessions.values(): for s in li: src = "%s[%d]" % (s.ipsrc, s.sport) dst = "%s[%d]" % (s.ipdst, s.dport) sid = repr(s.sid) if len(sid) > 12: sid = sid[:11] + "..." res.append((src, dst, sid)) colwidth = (max(len(y) for y in x) for x in zip(*res)) fmt = " ".join(map(lambda x: "%%-%ds" % x, colwidth)) return "\n".join(map(lambda x: fmt % x, res)) class TLSSession(TCPSession): def __init__(self, *args, **kwargs): # XXX this doesn't bring any value. warning( "TLSSession is deprecated and will be removed in a future version. " "Please use TCPSession instead with conf.tls_session_enable=True" ) server_rsa_key = kwargs.pop("server_rsa_key", None) super(TLSSession, self).__init__(*args, **kwargs) self._old_conf_status = conf.tls_session_enable conf.tls_session_enable = True if server_rsa_key: conf.tls_sessions.server_rsa_key = server_rsa_key def toPacketList(self): conf.tls_session_enable = self._old_conf_status return super(TLSSession, self).toPacketList() # Instantiate the TLS sessions holder conf.tls_sessions = _tls_sessions()