# SPDX-License-Identifier: GPL-2.0-only # This file is part of Scapy # See https://scapy.net/ for more information # Copyright (C) 2017 Maxence Tury # 2019 Romain Perez """ Common TLS 1.3 fields & bindings. This module covers the record layer, along with the ChangeCipherSpec, Alert and ApplicationData submessages. For the Handshake type, see tls_handshake.py. See the TLS class documentation for more information. """ import struct from scapy.config import conf from scapy.error import log_runtime, warning from scapy.compat import raw, orb from scapy.fields import ByteEnumField, PacketField, XStrField from scapy.layers.tls.session import _GenericTLSSessionInheritance from scapy.layers.tls.basefields import _TLSVersionField, _tls_version, \ _TLSMACField, _TLSLengthField, _tls_type from scapy.layers.tls.record import _TLSMsgListField, TLS from scapy.layers.tls.crypto.cipher_aead import AEADTagError from scapy.layers.tls.crypto.cipher_stream import Cipher_NULL from scapy.layers.tls.crypto.common import CipherError from scapy.layers.tls.crypto.pkcs1 import pkcs_i2osp ############################################################################### # TLS Record Protocol # ############################################################################### class TLSInnerPlaintext(_GenericTLSSessionInheritance): name = "TLS Inner Plaintext" fields_desc = [_TLSMsgListField("msg", []), ByteEnumField("type", None, _tls_type), XStrField("pad", "")] def pre_dissect(self, s): """ We need to parse the padding and type as soon as possible, else we won't be able to parse the message list... """ if len(s) < 1: raise Exception("Invalid InnerPlaintext (too short).") tmp_len = len(s) - 1 if s[-1] != b"\x00": msg_len = tmp_len else: n = 1 while s[-n] != b"\x00" and n < tmp_len: n += 1 msg_len = tmp_len - n self.fields_desc[0].length_from = lambda pkt: msg_len self.type = struct.unpack("B", s[msg_len:msg_len + 1])[0] return s class _TLSInnerPlaintextField(PacketField): def __init__(self, name, default, *args, **kargs): super(_TLSInnerPlaintextField, self).__init__(name, default, TLSInnerPlaintext) def m2i(self, pkt, m): return self.cls(m, tls_session=pkt.tls_session) def getfield(self, pkt, s): tag_len = pkt.tls_session.rcs.mac_len frag_len = pkt.len - tag_len if frag_len < 1: warning("InnerPlaintext should at least contain a byte type!") return s, None remain, i = super(_TLSInnerPlaintextField, self).getfield(pkt, s[:frag_len]) # noqa: E501 # remain should be empty here return remain + s[frag_len:], i def i2m(self, pkt, p): if isinstance(p, _GenericTLSSessionInheritance): p.tls_session = pkt.tls_session if not pkt.tls_session.frozen: return p.raw_stateful() return raw(p) class TLS13(_GenericTLSSessionInheritance): __slots__ = ["deciphered_len"] name = "TLS 1.3" fields_desc = [ByteEnumField("type", 0x17, _tls_type), _TLSVersionField("version", 0x0303, _tls_version), _TLSLengthField("len", None), _TLSInnerPlaintextField("inner", TLSInnerPlaintext()), _TLSMACField("auth_tag", None)] def __init__(self, *args, **kargs): self.deciphered_len = kargs.get("deciphered_len", None) super(TLS13, self).__init__(*args, **kargs) # Parsing methods def _tls_auth_decrypt(self, s): """ Provided with the record header and AEAD-ciphered data, return the sliced and clear tuple (TLSInnerPlaintext, tag). Note that we still return the slicing of the original input in case of decryption failure. Also, if the integrity check fails, a warning will be issued, but we still return the sliced (unauthenticated) plaintext. """ rcs = self.tls_session.rcs read_seq_num = struct.pack("!Q", rcs.seq_num) rcs.seq_num += 1 add_data = (pkcs_i2osp(self.type, 1) + pkcs_i2osp(self.version, 2) + pkcs_i2osp(len(s), 2)) try: return rcs.cipher.auth_decrypt(add_data, s, read_seq_num) except CipherError as e: return e.args except AEADTagError as e: pkt_info = self.firstlayer().summary() log_runtime.info("TLS: record integrity check failed [%s]", pkt_info) # noqa: E501 return e.args def pre_dissect(self, s): """ Decrypt, verify and decompress the message. """ # We commit the pending read state if it has been triggered. if self.tls_session.triggered_prcs_commit: if self.tls_session.prcs is not None: self.tls_session.rcs = self.tls_session.prcs self.tls_session.prcs = None self.tls_session.triggered_prcs_commit = False if len(s) < 5: raise Exception("Invalid record: header is too short.") self.type = orb(s[0]) if (isinstance(self.tls_session.rcs.cipher, Cipher_NULL) or self.type == 0x14): self.deciphered_len = None return s else: msglen = struct.unpack('!H', s[3:5])[0] hdr, efrag, r = s[:5], s[5:5 + msglen], s[msglen + 5:] frag, auth_tag = self._tls_auth_decrypt(efrag) self.deciphered_len = len(frag) return hdr + frag + auth_tag + r def post_dissect(self, s): """ Commit the pending read state if it has been triggered. We update nothing if the prcs was not set, as this probably means that we're working out-of-context (and we need to keep the default rcs). """ if self.tls_session.triggered_prcs_commit: if self.tls_session.prcs is not None: self.tls_session.rcs = self.tls_session.prcs self.tls_session.prcs = None self.tls_session.triggered_prcs_commit = False return s def do_dissect_payload(self, s): """ Try to dissect the following data as a TLS message. Note that overloading .guess_payload_class() would not be enough, as the TLS session to be used would get lost. """ if s: try: p = TLS(s, _internal=1, _underlayer=self, tls_session=self.tls_session) except KeyboardInterrupt: raise except Exception: p = conf.raw_layer(s, _internal=1, _underlayer=self) self.add_payload(p) # Building methods def _tls_auth_encrypt(self, s): """ Return the TLSCiphertext.encrypted_record for AEAD ciphers. """ wcs = self.tls_session.wcs write_seq_num = struct.pack("!Q", wcs.seq_num) wcs.seq_num += 1 add_data = (pkcs_i2osp(self.type, 1) + pkcs_i2osp(self.version, 2) + pkcs_i2osp(len(s) + wcs.cipher.tag_len, 2)) return wcs.cipher.auth_encrypt(s, add_data, write_seq_num) def post_build(self, pkt, pay): """ Apply the previous methods according to the writing cipher type. """ # Compute the length of TLSPlaintext fragment hdr, frag = pkt[:5], pkt[5:] if not isinstance(self.tls_session.wcs.cipher, Cipher_NULL): frag = self._tls_auth_encrypt(frag) if self.len is not None: # The user gave us a 'len', let's respect this ultimately hdr = hdr[:3] + struct.pack("!H", self.len) else: # Update header with the length of TLSCiphertext.inner hdr = hdr[:3] + struct.pack("!H", len(frag)) # Now we commit the pending write state if it has been triggered. We # update nothing if the pwcs was not set. This probably means that # we're working out-of-context (and we need to keep the default wcs). if self.tls_session.triggered_pwcs_commit: if self.tls_session.pwcs is not None: self.tls_session.wcs = self.tls_session.pwcs self.tls_session.pwcs = None self.tls_session.triggered_pwcs_commit = False return hdr + frag + pay