TLS sessions: multiple improvements and cleanups

Author: gpotter2

doc/scapy/usage.rst

@@ -802,7 +802,6 @@ Those sessions can be used using the ``session=`` parameter of ``sniff()``. Exam
    To implement your own Session class, in order to support another flow-based protocol, start by copying a sample from `scapy/sessions.py <https://github.com/secdev/scapy/blob/master/scapy/sessions.py>`_
    Your custom ``Session`` class only needs to extend the :py:class:`~scapy.sessions.DefaultSession` class, and implement a ``on_packet_received`` function, such as in the example.
 
-.. note:: Would you need it, you can use: ``class TLS_over_TCP(TLSSession, TCPSession): pass`` to sniff TLS packets that are defragmented.
 
 How to use TCPSession to defragment TCP packets
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

scapy/layers/tls/record.py

@@ -40,14 +40,6 @@
 if conf.crypto_valid_advanced:
     from scapy.layers.tls.crypto.cipher_aead import Cipher_CHACHA20_POLY1305
 
-# Util
-
-
-def _tls_version_check(version, min):
-    """Returns if version >= min, or False if version == None"""
-    if version is None:
-        return False
-    return version >= min
 
 ###############################################################################
 #   TLS Record Protocol                                                       #
@@ -216,7 +208,7 @@ def addfield(self, pkt, s, val):
         # Add TLS13ClientHello in case of HelloRetryRequest
         # Add ChangeCipherSpec for middlebox compatibility
         if (isinstance(pkt, _GenericTLSSessionInheritance) and
-                _tls_version_check(pkt.tls_session.tls_version, 0x0304) and
+                pkt.tls_session.tls_version == 0x0304 and
                 not isinstance(pkt.msg[0], TLS13ServerHello) and
                 not isinstance(pkt.msg[0], TLS13ClientHello) and
                 not isinstance(pkt.msg[0], TLSChangeCipherSpec)):
@@ -337,7 +329,7 @@ def dispatch_hook(cls, _pkt=None, *args, **kargs):
                     # Not SSLv2: continuation
                     return _TLSEncryptedContent
                 # Check TLS 1.3
-                if s and _tls_version_check(s.tls_version, 0x0304):
+                if s and s.tls_version == 0x0304:
                     _has_cipher = lambda x: (
                         x and not isinstance(x.cipher, Cipher_NULL)
                     )
@@ -734,11 +726,11 @@ def post_build(self, pkt, pay):
         return hdr + efrag + pay
 
     def mysummary(self):
-        s = super(TLS, self).mysummary()
+        s, n = super(TLS, self).mysummary()
         if self.msg:
             s += " / "
             s += " / ".join(getattr(x, "_name", x.name) for x in self.msg)
-        return s
+        return s, n
 
 ###############################################################################
 #   TLS ChangeCipherSpec                                                      #

scapy/layers/tls/record_tls13.py

@@ -223,3 +223,10 @@ def post_build(self, pkt, pay):
             self.tls_session.triggered_pwcs_commit = False
 
         return hdr + frag + pay
+
+    def mysummary(self):
+        s, n = super(TLS13, self).mysummary()
+        if self.inner and self.inner.msg:
+            s += " / "
+            s += " / ".join(getattr(x, "_name", x.name) for x in self.inner.msg)
+        return s, n

scapy/layers/tls/session.py

@@ -18,7 +18,7 @@
 from scapy.error import log_runtime, warning
 from scapy.packet import Packet
 from scapy.pton_ntop import inet_pton
-from scapy.sessions import DefaultSession
+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
@@ -903,13 +903,20 @@ def eq(self, other):
 
         return False
 
-    def __repr__(self):
+    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                                                         #
 ###############################################################################
@@ -1079,25 +1086,41 @@ def show2(self):
         s.rcs = rcs_snap
         s.wcs = wcs_snap
 
-    def mysummary(self):
-        return "TLS %s / %s" % (repr(self.tls_session),
-                                getattr(self, "_name", self.name))
+    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 TLSSession
+        # 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
+            length = 0
+            curdata = data
+            while len(curdata) > 5:
+                clength = struct.unpack("!H", curdata[3:5])[0] + 5
+                curdata = curdata[clength:]
+                length += clength
             if len(data) == length:
-                return cls(data)
-            elif len(data) > length:
-                pkt = cls(data)
-                if hasattr(pkt.payload, "tcp_reassemble"):
-                    return pkt.payload.tcp_reassemble(data[length:], metadata, session)
-                else:
-                    return pkt
+                # get the underlayer as it is used to populate tls_session
+                underlayer = metadata["original"][TCP].copy()
+                underlayer.remove_payload()
+                return cls(data, _underlayer=underlayer)
         else:
             return cls(data)
 
@@ -1162,8 +1185,13 @@ def __repr__(self):
         return "\n".join(map(lambda x: fmt % x, res))
 
 
-class TLSSession(DefaultSession):
+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

scapy/sessions.py

@@ -355,6 +355,7 @@ def _process_packet(self, pkt):
         packet = None  # type: Optional[Packet]
         if data.full():
             # Reassemble using all previous packets
+            metadata["original"] = pkt
             packet = tcp_reassemble(
                 bytes(data),
                 metadata,

test/pcaps/tls_tcp_frag_withnss.pcap.gz

@@ -1590,3 +1590,28 @@ if shutil.which("editcap"):
     assert b"GET /secret.txt HTTP/1.0\n" in packets[11].msg[0].data
     assert b"z2|gxarIKOxt,G1d>.Q2MzGY[k@" in packets[13].msg[0].data
     conf = bck_conf
+
+= pcap file & external TLS Key Log file with TCPSession
+* GH3722
+
+load_layer("tls")
+from scapy.tools.UTscapy import scapy_path
+
+# Write SSLKEYLOGFILE
+temp_sslkeylog = get_temp_file()
+with open(temp_sslkeylog, "w") as fd:
+    fd.write("CLIENT_RANDOM 09F91DA01B1FEB50B691C932959111E5E1D676437F7A42DE47EA881F6295D4E7 EE119869B732F0F9561FFDD95E50A2ACBF268EE0C7C33B409E68C1972E0B280944F7345E845E82F909CCFEB61C456E1F\n")
+
+bck_conf = conf
+conf.tls_session_enable = True
+conf.tls_nss_filename = temp_sslkeylog
+
+packets = sniff(offline=scapy_path("test/pcaps/tls_tcp_frag_withnss.pcap.gz"), session=TCPSession)
+
+# XXXXXXXXXXXXXXXXXX
+# WIP WIP WIP
+# DO NOT MERGE
+# WIP WIP WIP
+# XXXXXXXXXXXXXXXXXX
+
+conf = bck_conf

test/tls.uts

@@ -1214,6 +1214,8 @@ assert len(ch.client_shares[0].key_exchange) == 97
 
 = Parse TLS 1.3 Client Hello with non-rfc 5077 ticket
 
+from scapy.layers.tls.keyexchange_tls13 import TLS_Ext_PreSharedKey_CH
+
 ch = TLS(b'\x16\x03\x01\x01\x1a\x01\x00\x01\x16\x03\x03\xec\x9c>\xb2\x9e|B\x05\x17f\x86\xc8\x18\x0421\x87\x87\x12\xf6\xec\xa2J\x95\x84[\xf8\xab\xe9gK> \xc6%\xff&wn)\xb2\xf5\xe8_x\x96\xe9\nEsK\xda\x86o\x82f\xa5\xbadk\xf4Ar~}\x00\x08\x13\x02\x13\x03\x13\x01\x00\xff\x01\x00\x00\xc5\x00\x0b\x00\x04\x03\x00\x01\x02\x00\n\x00\x16\x00\x14\x00\x1d\x00\x17\x00\x1e\x00\x19\x00\x18\x01\x00\x01\x01\x01\x02\x01\x03\x01\x04\x00#\x00\x00\x00\x16\x00\x00\x00\x17\x00\x00\x00\r\x00\x1e\x00\x1c\x04\x03\x05\x03\x06\x03\x08\x07\x08\x08\x08\t\x08\n\x08\x0b\x08\x04\x08\x05\x08\x06\x04\x01\x05\x01\x06\x01\x00+\x00\x03\x02\x03\x04\x00-\x00\x02\x01\x01\x003\x00&\x00$\x00\x1d\x00 l\x19\xe1f1 )6\xbf\x91\x9e\xab\xd2\x06\x16\x0b|\x88\xf7,\xf1\x88\x99Z\xb6\xb3\x93\xe4\x08z\x8a\t\x00)\x00:\x00\x15\x00\x0fClient_identity\x00\x00\x00\x00\x00! m\xf3^\xc1l\xac5\xf2\xe3=\xeb\xe3\x81\xd3\xb3\xdd\xbd\xbd\x01\xc9\xdd\x01i\x8c1\xa0ye\xcd\x04\x9e\x9c')
 
 assert isinstance(ch.msg[0].ext[9], TLS_Ext_PreSharedKey_CH)