## This file is part of Scapy
|
## Copyright (C) 2017 Maxence Tury
|
## This program is published under a GPLv2 license
|
|
"""
|
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
|
from scapy.fields import *
|
from scapy.packet import *
|
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
|
from scapy.layers.tls.crypto.cipher_aead import AEADTagError
|
from scapy.layers.tls.crypto.cipher_stream import Cipher_NULL
|
from scapy.layers.tls.crypto.ciphers import CipherError
|
|
|
###############################################################################
|
### 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).")
|
|
l = len(s) - 1
|
if s[-1] != b"\x00":
|
msg_len = l
|
else:
|
n = 1
|
while s[-n] != b"\x00" and n < l:
|
n += 1
|
msg_len = l - 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])
|
# 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", 0x0301, _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
|
try:
|
return rcs.cipher.auth_decrypt(b"", 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)
|
return e.args
|
|
def pre_dissect(self, s):
|
"""
|
Decrypt, verify and decompress the message.
|
"""
|
if len(s) < 5:
|
raise Exception("Invalid record: header is too short.")
|
|
if isinstance(self.tls_session.rcs.cipher, Cipher_NULL):
|
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:
|
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
|
return wcs.cipher.auth_encrypt(s, b"", 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.rcs.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
|
