WIP: Add key logging for TLS 1.3 and TLS 1.2 (#523)

Author: nickrabbott

tlslite/session.py

@@ -98,6 +98,8 @@ def __init__(self):
         self.tickets = None
         self.tls_1_0_tickets = None
         self.ec_point_format = 0
+        self.cl_handshake_traffic_secret = None
+        self.sr_handshake_traffic_secret = None
 
     def create(self, masterSecret, sessionID, cipherSuite,
                srpUsername, clientCertChain, serverCertChain,
@@ -106,7 +108,9 @@ def create(self, masterSecret, sessionID, cipherSuite,
                appProto=bytearray(0), cl_app_secret=bytearray(0),
                sr_app_secret=bytearray(0), exporterMasterSecret=bytearray(0),
                resumptionMasterSecret=bytearray(0), tickets=None,
-               tls_1_0_tickets=None, ec_point_format=None):
+               tls_1_0_tickets=None, ec_point_format=None,
+               cl_handshake_traffic_secret=bytearray(0),
+               sr_handshake_traffic_secret=bytearray(0)):
         self.masterSecret = masterSecret
         self.sessionID = sessionID
         self.cipherSuite = cipherSuite
@@ -124,6 +128,8 @@ def create(self, masterSecret, sessionID, cipherSuite,
         self.sr_app_secret = sr_app_secret
         self.exporterMasterSecret = exporterMasterSecret
         self.resumptionMasterSecret = resumptionMasterSecret
+        self.cl_handshake_traffic_secret = cl_handshake_traffic_secret
+        self.sr_handshake_traffic_secret = sr_handshake_traffic_secret
         # NOTE we need a reference copy not a copy of object here!
         self.tickets = tickets
         self.tls_1_0_tickets = tls_1_0_tickets

tlslite/tlsconnection.py

@@ -101,6 +101,7 @@ def __init__(self, sock):
         self._pha_supported = False
         self.client_cert_compression_algo = None
         self.server_cert_compression_algo = None
+        self.sslkeylogfile = os.environ.get('SSLKEYLOGFILE')
 
     def keyingMaterialExporter(self, label, length=20):
         """Return keying material as described in RFC 5705
@@ -414,7 +415,6 @@ def _handshakeClientAsync(self, srpParams=(), certParams=(), anonParams=(),
                               session=None, settings=None, checker=None,
                               nextProtos=None, serverName=None, reqTack=True,
                               alpn=None):
-
         handshaker = self._handshakeClientAsyncHelper(srpParams=srpParams,
                 certParams=certParams,
                 anonParams=anonParams,
@@ -427,6 +427,10 @@ def _handshakeClientAsync(self, srpParams=(), certParams=(), anonParams=(),
         for result in self._handshakeWrapperAsync(handshaker, checker):
             yield result
 
+        # Log client random and master secret for version < TLS1.3
+        if self.sslkeylogfile and self.version < (3, 4):
+            self._log_session_keys('CLIENT_RANDOM', self._clientRandom, self.session.masterSecret)
+
 
     def _handshakeClientAsyncHelper(self, srpParams, certParams, anonParams,
                                session, settings, serverName, nextProtos,
@@ -1323,6 +1327,13 @@ def _clientTLS13Handshake(self, settings, session, clientHello,
                                                     self._handshake_hash,
                                                     prfName)
 
+        # TLS1.3 log Client and Server traffic secrets for SSLKEYLOGFILE
+        if self.sslkeylogfile:
+            self._log_session_keys('CLIENT_HANDSHAKE_TRAFFIC_SECRET',
+                                   clientHello.random,
+                                   cl_handshake_traffic_secret)
+            self._log_session_keys('SERVER_HANDSHAKE_TRAFFIC_SECRET', clientHello.random, sr_handshake_traffic_secret)
+
         # prepare for reading encrypted messages
         self._recordLayer.calcTLS1_3PendingState(
             serverHello.cipher_suite,
@@ -1613,6 +1624,13 @@ def _clientTLS13Handshake(self, settings, session, clientHello,
                                                bytearray(b'exp master'),
                                                self._handshake_hash, prfName)
 
+
+        # Now that we have all the TLS1.3 secrets during the handshake, log them if necessary
+        if self.sslkeylogfile:
+            self._log_session_keys('EXPORTER_SECRET', clientHello.random, exporter_master_secret)
+            self._log_session_keys('CLIENT_TRAFFIC_SECRET_0', clientHello.random, cl_app_traffic)
+            self._log_session_keys('SERVER_TRAFFIC_SECRET_0', clientHello.random, sr_app_traffic)
+
         self._recordLayer.calcTLS1_3PendingState(
             serverHello.cipher_suite,
             cl_app_traffic,
@@ -1708,7 +1726,9 @@ def _clientTLS13Handshake(self, settings, session, clientHello,
                             exporterMasterSecret=exporter_master_secret,
                             resumptionMasterSecret=resumption_master_secret,
                             # NOTE it must be a reference, not a copy!
-                            tickets=self.tickets)
+                            tickets=self.tickets,
+                            cl_handshake_traffic_secret=cl_handshake_traffic_secret,
+                            sr_handshake_traffic_secret=sr_handshake_traffic_secret)
 
         yield "finished" if not resuming else "resumed_and_finished"
 
@@ -2250,6 +2270,10 @@ def handshakeServerAsync(self, verifierDB=None,
         for result in self._handshakeWrapperAsync(handshaker, checker):
             yield result
 
+        # Log client random and master secret for version < TLS1.3
+        if self.sslkeylogfile and self.version < (3, 4):
+            self._log_session_keys('CLIENT_RANDOM', self._clientRandom, self.session.masterSecret)
+
 
     def _handshakeServerAsyncHelper(self, verifierDB,
                                     cert_chain, privateKey, reqCert,
@@ -2956,6 +2980,14 @@ def _serverTLS13Handshake(self, settings, clientHello, cipherSuite,
                                                     bytearray(b'c hs traffic'),
                                                     self._handshake_hash,
                                                     prf_name)
+
+        # TLS1.3 log Client and Server traffic secrets for SSLKEYLOGFILE
+        if self.sslkeylogfile:
+            self._log_session_keys('CLIENT_HANDSHAKE_TRAFFIC_SECRET',
+                                   clientHello.random,
+                                   cl_handshake_traffic_secret)
+            self._log_session_keys('SERVER_HANDSHAKE_TRAFFIC_SECRET', clientHello.random, sr_handshake_traffic_secret)
+
         self.version = version
         self._recordLayer.calcTLS1_3PendingState(
             cipherSuite,
@@ -3222,6 +3254,14 @@ def _serverTLS13Handshake(self, settings, clientHello, cipherSuite,
                                                self._handshake_hash,
                                                prf_name)
 
+
+        # Now that we have all the TLS1.3 secrets during the handshake, log them if necessary
+        if self.sslkeylogfile:
+            self._log_session_keys('EXPORTER_SECRET', clientHello.random, exporter_master_secret)
+            self._log_session_keys('CLIENT_TRAFFIC_SECRET_0', clientHello.random, cl_app_traffic)
+            self._log_session_keys('SERVER_TRAFFIC_SECRET_0', clientHello.random, sr_app_traffic)
+
+
         # verify Finished of client
         cl_finished_key = HKDF_expand_label(cl_handshake_traffic_secret,
                                             b"finished", b'',
@@ -3285,7 +3325,9 @@ def _serverTLS13Handshake(self, settings, clientHello, cipherSuite,
                             exporterMasterSecret=exporter_master_secret,
                             resumptionMasterSecret=resumption_master_secret,
                             # NOTE it must be a reference, not a copy
-                            tickets=self.tickets)
+                            tickets=self.tickets,
+                            cl_handshake_traffic_secret=cl_handshake_traffic_secret,
+                            sr_handshake_traffic_secret=sr_handshake_traffic_secret)
 
         # switch to application_traffic_secret for client packets
         self._changeReadState()
@@ -4738,6 +4780,11 @@ def _serverFinished(self, premasterSecret, clientRandom, serverRandom,
                                 clientRandom, serverRandom,
                                 cipherImplementations)
 
+        #Log client random and master secret if SSLKEYLOGFILE is set
+        if self.sslkeylogfile and self.version < (3, 3):
+            print('Logging session keys in serverFinished')
+            self._log_session_keys("CLIENT_RANDOM", clientRandom, masterSecret)
+
         #Exchange ChangeCipherSpec and Finished messages
         for result in self._getFinished(masterSecret,
                                         cipherSuite,
@@ -4934,6 +4981,10 @@ def _calculate_master_secret(self, premaster_secret, cipher_suite,
                               output_length=48)
         return secret
 
+    def _log_session_keys(self, secret_label, client_random, secret):
+        with open(self.sslkeylogfile, 'a') as ssl_key_log_file:
+            ssl_key_log_file.write(f"{secret_label} {client_random.hex()} {secret.hex()}\n")
+
     @staticmethod
     def _pickServerKeyExchangeSig(settings, clientHello, certList=None,
                                   private_key=None,

unit_tests/test_tlslite_ssl_key_log_file.py

@@ -0,0 +1,112 @@
+try:
+    import unittest2 as unittest
+except ImportError:
+    import unittest
+
+import tempfile
+from socket import socket, AF_INET, SOCK_STREAM
+import os
+from typing import Dict, List
+
+import unittest
+
+from tlslite.api import TLSConnection, HandshakeSettings
+from tlslite.errors import TLSRemoteAlert
+
+
+def create_connection(hostname='www.example.com', port=443, settings=HandshakeSettings()):
+    raw_socket = socket(AF_INET, SOCK_STREAM)
+    raw_socket.connect((hostname, port))
+    connection = TLSConnection(raw_socket)
+    connection.handshakeClientCert(settings=settings)
+    return connection
+
+
+def validate_log_file(log_file_name, labels: Dict[str, List[str]]):
+    """
+    TLS1.3 labels
+    ------------------
+    SERVER_HANDSHAKE_TRAFFIC_SECRET <client_random> <secret>
+    EXPORTER_SECRET <client_random> <secret>
+    SERVER_TRAFFIC_SECRET_0  <client_random> <secret>
+    CLIENT_HANDSHAKE_TRAFFIC_SECRET <client_random> <secret>
+    CLIENT_TRAFFIC_SECRET_0 <client_random> <secret>
+
+    TLS1.2 (and below)
+    ----------------
+    CLIENT_RANDOM <client_random> <secret>
+    """
+    with open(log_file_name, 'r') as log_file:
+        for line in log_file.readlines():
+            entry = line.split()
+            label, client_random, secret = entry[0], entry[1], entry[2]
+            if label in labels and labels[label][1] == client_random and labels[label][2] != secret:
+                return False
+
+        return True
+
+
+class TestSslKeyLogFile(unittest.TestCase):
+    def setUp(self):
+        self.temp_log_file = tempfile.NamedTemporaryFile(delete_on_close=False)
+        os.environ['SSLKEYLOGFILE'] = self.temp_log_file.name
+
+    def tearDown(self):
+        # Clean up the temporary file
+        os.remove(self.temp_log_file.name)
+        if os.environ.get('SSLKEYLOGFILE'):
+            del os.environ['SSLKEYLOGFILE']
+
+    def test_tlsv1_1(self):
+        settings = HandshakeSettings()
+        settings.minVersion = (3,2)
+        settings.maxVersion = (3,2)
+
+        try:
+            connection = create_connection('www.example.com', 443, settings)
+            expected_labels = {
+                'CLIENT_RANDOM': connection._clientRandom
+            }
+            self.assertTrue(validate_log_file(self.temp_log_file.name, expected_labels))
+            connection.close()
+        except TLSRemoteAlert as alert:
+            print(f"TLS Remote Alert: {alert}")
+
+
+    def test_tlsv1_2(self):
+        settings = HandshakeSettings()
+        settings.minVersion = (3,3)
+        settings.maxVersion = (3,3)
+
+        try:
+            connection = create_connection('www.example.com', 443, settings)
+            expected_labels = {
+                'CLIENT_RANDOM': connection._clientRandom
+            }
+            self.assertTrue(validate_log_file(self.temp_log_file.name, expected_labels))
+            connection.close()
+        except TLSRemoteAlert as alert:
+            print(f"TLS Remote Alert: {alert}")
+
+    def test_tlsv1_3(self):
+        settings = HandshakeSettings()
+        settings.minVersion = (3, 4)
+        settings.maxVersion = (3, 4)
+
+        try:
+            connection = create_connection('www.example.com', 443, settings)
+            expected_labels = {
+                "SERVER_HANDSHAKE_TRAFFIC_SECRET": connection.session.sr_handshake_traffic_secret,
+                "EXPORTER_SECRET": connection.session.exporterMasterSecret,
+                "SERVER_TRAFFIC_SECRET_0": connection.session.sr_app_secret,
+                "CLIENT_HANDSHAKE_TRAFFIC_SECRET": connection.session.cl_handshake_traffic_secret,
+                "CLIENT_TRAFFIC_SECRET_0": connection.session.cl_app_secret
+            }
+            self.assertTrue(validate_log_file(self.temp_log_file.name, expected_labels))
+            connection.close()
+        except TLSRemoteAlert as alert:
+            print(f"TLS Remote Alert: {alert}")
+
+
+if __name__ == "__main__":
+    unittest.main()