Keystore used by default for RSA

- Removed all references to rsa.pubkey and rsa.privkey. Everything
  uses the RSA keystore
- Moved TLSSessionCtx string templating to the RSAKeystore class.
  Each sub element can be responsible for it's formating. Looks
  weird right now, until I have completed the refactoring of other
  elements

scapy_ssl_tls/ssl_tls_automata.py

@@ -386,8 +386,6 @@ def do_bind(self):
         pemo = pem_get_objects(self.pemkey)
         for key_pk in (k for k in pemo.keys() if "PRIVATE" in k.upper()):
             self.srv_tls_sock.tls_ctx.crypto.server.keystore = tlsk.RSAKeystore.from_private(pemo[key_pk].get("full"))
-            self.srv_tls_sock.tls_ctx.crypto.server.rsa.privkey, self.srv_tls_sock.tls_ctx.crypto.server.rsa.pubkey = \
-                self.srv_tls_sock.tls_ctx.crypto.server.keystore.keys
             break
 
     @hookable

scapy_ssl_tls/ssl_tls_crypto.py

@@ -99,35 +99,25 @@ def __init__(self, client=True):
         self.compression = namedtuple("compression", ["method"])
         self.compression.method = None
         self.crypto = namedtuple('crypto', ['client','server'])
-        self.crypto.client = namedtuple('client', ['enc', 'dec', "hmac"])
+        self.crypto.client = namedtuple('client', ['enc', 'dec', "hmac", "keystore"])
         self.crypto.client.enc = None
         self.crypto.client.dec = None
         self.crypto.client.hmac = None
-        self.crypto.client.keystore = None
-        self.crypto.client.rsa = namedtuple("rsa", ["pubkey","privkey"])
-        self.crypto.client.rsa.pubkey = None
-        self.crypto.client.rsa.privkey = None
-        self.crypto.client.dsa = namedtuple("dsa", ["pubkey","privkey"])
-        self.crypto.client.dsa.pubkey = None
-        self.crypto.client.dsa.privkey = None
+        self.crypto.client.keystore = tlsk.EmptyAsymKeystore()
+
         self.crypto.client.dh = namedtuple("dh", ["x", "y_c"])
         self.crypto.client.dh.x = None
         self.crypto.client.dh.y_c = None
         self.crypto.client.ecdh = namedtuple("ecdh", ["curve_name", "priv", "pub"])
         self.crypto.client.ecdh.curve_name = None
         self.crypto.client.ecdh.priv = None
         self.crypto.client.ecdh.pub = None
-        self.crypto.server = namedtuple('server', ['enc','dec','rsa', "hmac"])
+        self.crypto.server = namedtuple('server', ['enc','dec','rsa', "hmac", "keystore"])
         self.crypto.server.enc = None
         self.crypto.server.dec = None
         self.crypto.server.hmac = None
-        self.crypto.server.keystore = None
-        self.crypto.server.rsa = namedtuple("rsa", ["pubkey","privkey"])
-        self.crypto.server.rsa.pubkey = None
-        self.crypto.server.rsa.privkey = None
-        self.crypto.server.dsa = namedtuple("dsa", ["pubkey","privkey"])
-        self.crypto.server.dsa.pubkey = None
-        self.crypto.server.dsa.privkey = None
+        self.crypto.server.keystore = tlsk.EmptyAsymKeystore()
+
         self.crypto.server.dh = namedtuple("dh", ["p", "g", "x", "y_s"])
         self.crypto.server.dh.p = None
         self.crypto.server.dh.g = None
@@ -184,15 +174,8 @@ def __repr__(self):
                   'crypto-server-enc':repr(self.crypto.server.enc),
                   'crypto-server-dec':repr(self.crypto.server.dec),
 
-                  "crypto-client-rsa-pubkey": repr(self.crypto.client.rsa.pubkey),
-                  "crypto-client-rsa-privkey": repr(self.crypto.client.rsa.privkey),
-                  "crypto-server-rsa-pubkey": repr(self.crypto.server.rsa.pubkey),
-                  "crypto-server-rsa-privkey": repr(self.crypto.server.rsa.privkey),
-
-                  "crypto-client-dsa-pubkey": repr(self.crypto.client.dsa.pubkey),
-                  "crypto-client-dsa-privkey": repr(self.crypto.client.dsa.privkey),
-                  "crypto-server-dsa-pubkey": repr(self.crypto.server.dsa.pubkey),
-                  "crypto-server-dsa-privkey": repr(self.crypto.server.dsa.privkey),
+                  "crypto-client-keystore": self.crypto.client.keystore,
+                  "crypto-server-keystore": self.crypto.server.keystore,
 
                   "crypto-client-dh-x": repr(self.crypto.client.dh.x),
                   "crypto-client-dh-y_c": repr(self.crypto.client.dh.y_c),
@@ -245,17 +228,8 @@ def __repr__(self):
         str_ +="\n\t crypto.server.enc=%(crypto-server-enc)s"
         str_ +="\n\t crypto.server.dec=%(crypto-server-dec)s"
 
-        str_ +="\n\t crypto.client.rsa.privkey=%(crypto-client-rsa-privkey)s"
-        str_ +="\n\t crypto.client.rsa.pubkey=%(crypto-client-rsa-pubkey)s"
-
-        str_ +="\n\t crypto.server.rsa.privkey=%(crypto-server-rsa-privkey)s"
-        str_ +="\n\t crypto.server.rsa.pubkey=%(crypto-server-rsa-pubkey)s"
-
-        str_ +="\n\t crypto.client.dsa.privkey=%(crypto-client-dsa-privkey)s"
-        str_ +="\n\t crypto.client.dsa.pubkey=%(crypto-client-dsa-pubkey)s"
-
-        str_ +="\n\t crypto.server.dsa.privkey=%(crypto-server-dsa-privkey)s"
-        str_ +="\n\t crypto.server.dsa.pubkey=%(crypto-server-dsa-pubkey)s"
+        str_ += "\n\t crypto.client.keystore=%(crypto-client-keystore)s"
+        str_ += "\n\t crypto.server.keystore=%(crypto-server-keystore)s"
 
         str_ += "\n\t crypto.client.dh.x=%(crypto-client-dh-x)s"
         str_ += "\n\t crypto.client.dh.y_c=%(crypto-client-dh-y_c)s"
@@ -359,8 +333,12 @@ def _process(self,p):
                 if self.params.negotiated.key_exchange is not None and (self.params.negotiated.key_exchange == tls.TLSKexNames.RSA or self.params.negotiated.sig == RSA):
                     # fetch server pubkey // PKCS1_v1_5
                     cert = p[tls.TLSCertificateList].certificates[0].data
-                    self.crypto.server.keystore = tlsk.RSAKeystore.from_der_certificate(str(cert))
-                    self.crypto.server.rsa.pubkey = self.crypto.server.keystore.public
+                    # If we have a default keystore, create an RSA keystore and populate it from data on the wire
+                    if isinstance(self.crypto.server.keystore, tlsk.EmptyAsymKeystore):
+                        self.crypto.server.keystore = tlsk.RSAKeystore.from_der_certificate(str(cert))
+                    # Else keystore was assigned by user. Just add cert from the wire to the store
+                    else:
+                        self.crypto.server.keystore.certificate = str(cert)
                     # TODO: In the future also handle kex = DH and extract static DH params from cert
                 elif self.params.negotiated.key_exchange is not None and self.params.negotiated.sig == DSA:
                     # TODO: Handle DSA sig key loading here to allow sig checks
@@ -400,9 +378,10 @@ def _process(self,p):
                 if p.haslayer(tls.TLSClientRSAParams):
                     self.crypto.session.encrypted_premaster_secret = p[tls.TLSClientRSAParams].data
                     # If we have the private key, let's decrypt the PMS
-                    if self.crypto.server.rsa.privkey is not None:
-                        self.crypto.session.premaster_secret = PKCS1_v1_5.new(self.crypto.server.rsa.privkey).decrypt(
-                            self.crypto.session.encrypted_premaster_secret, None)
+                    private = self.crypto.server.keystore.private
+                    if private is not None:
+                        self.crypto.session.premaster_secret = PKCS1_v1_5.new(
+                            private).decrypt(self.crypto.session.encrypted_premaster_secret, None)
                 elif p.haslayer(tls.TLSClientDHParams):
                     self.crypto.client.dh.y_c = p[tls.TLSClientDHParams].data
                 elif p.haslayer(tls.TLSClientECDHParams):
@@ -449,10 +428,8 @@ def rsa_load_keys_from_file(self, priv_key_file, client=False):
                 try:
                     if client:
                         self.crypto.client.keystore = tlsk.RSAKeystore.from_private(pemo[key_pk].get("full"))
-                        self.crypto.client.rsa.privkey, self.crypto.client.rsa.pubkey = self.crypto.client.keystore.keys
                     else:
                         self.crypto.server.keystore = tlsk.RSAKeystore.from_private(pemo[key_pk].get("full"))
-                        self.crypto.server.rsa.privkey, self.crypto.server.rsa.pubkey = self.crypto.server.keystore.keys
                     return
                 except ValueError:
                     pass
@@ -461,19 +438,17 @@ def rsa_load_keys_from_file(self, priv_key_file, client=False):
     def rsa_load_keys(self, private, client=False):
         if client:
             self.crypto.client.keystore = tlsk.RSAKeystore.from_private(private)
-            self.crypto.client.rsa.privkey, self.crypto.client.rsa.pubkey = self.crypto.client.keystore.keys
         else:
             self.crypto.server.keystore = tlsk.RSAKeystore.from_private(private)
-            self.crypto.server.rsa.privkey, self.crypto.server.rsa.pubkey = self.crypto.server.keystore.keys
 
     def _generate_random_pms(self, version):
         return "%s%s" % (struct.pack("!H", version), os.urandom(46))
 
     def get_encrypted_pms(self, pms=None):
         cleartext = pms or self.crypto.session.premaster_secret
-        if self.crypto.server.rsa.pubkey is not None:
-            self.crypto.session.encrypted_premaster_secret = PKCS1_v1_5.new(self.crypto.server.rsa.pubkey).\
-                encrypt(cleartext)
+        public = self.crypto.server.keystore.public
+        if public is not None:
+            self.crypto.session.encrypted_premaster_secret = PKCS1_v1_5.new(public).encrypt(cleartext)
         else:
             raise ValueError("Cannot calculate encrypted MS. No server certificate found in connection")
         return self.crypto.session.encrypted_premaster_secret
@@ -565,13 +540,13 @@ def get_handshake_hash(self, hash_):
         return hash_
 
     def get_client_signed_handshake_hash(self, hash_=SHA256.new(), pre_sign_hook=None):
-        if self.crypto.client.rsa.privkey is None:
+        if self.crypto.client.keystore.private is None:
             raise RuntimeError("Missing client private key. Can't sign")
         msg_hash = self.get_handshake_hash(hash_)
         if pre_sign_hook is not None:
             msg_hash = pre_sign_hook(msg_hash)
         # Will throw exception if we can't sign or if data is larger the modulus
-        return Sig_PKCS1_v1_5.new(self.crypto.client.rsa.privkey).sign(msg_hash)
+        return Sig_PKCS1_v1_5.new(self.crypto.client.keystore.private).sign(msg_hash)
 
     def set_mode(self, client=None, server=None):
         self.client = client if client else not server

scapy_ssl_tls/ssl_tls_keystore.py

@@ -76,10 +76,24 @@ def __init__(self, name, public, private=None):
     def from_private(cls, private):
         raise NotImplementedError()
 
+    def __str__(self):
+        template = """
+{name}:
+    certificate: {certificate}
+    public: {public}
+    private: {private}"""
+        return template.format(name=self.name, certificate=repr(self.certificate), public=self.public,
+                               private=self.private)
+
+
+class EmptyAsymKeystore(AsymKeyStore):
+    def __init__(self):
+        super(EmptyAsymKeystore, self).__init__("Empty Keystore", None, None)
+
 
 class RSAKeystore(AsymKeyStore):
     def __init__(self, public, private=None):
-        super(RSAKeystore, self).__init__("RSA", public, private)
+        super(RSAKeystore, self).__init__("RSA Keystore", public, private)
 
     @classmethod
     def from_der_certificate(cls, certificate):
@@ -104,4 +118,4 @@ def from_private(cls, private):
 
 class DSAKeystore(AsymKeyStore):
     def __init__(self, public, private=None):
-        super(DSAKeystore, self).__init__("DSA", public, private)
+        super(DSAKeystore, self).__init__("DSA Keystore", public, private)

tests/test_ssl_tls_crypto.py

@@ -129,15 +129,15 @@ def test_keys_are_set_in_context_when_loaded(self):
         pkt = tls.TLSRecord() / tls.TLSHandshake() / tls.TLSClientHello(version=0x0301)
         tls_ctx.insert(pkt)
         tls_ctx.rsa_load_keys(self.pem_priv_key)
-        self.assertIsNotNone(tls_ctx.crypto.server.rsa.privkey)
-        self.assertIsNotNone(tls_ctx.crypto.server.rsa.pubkey)
+        self.assertIsNotNone(tls_ctx.crypto.server.keystore.private)
+        self.assertIsNotNone(tls_ctx.crypto.server.keystore.public)
         # Broken due to pycrypto bug: https://github.com/dlitz/pycrypto/issues/114
         # Uncomment when fixed upstream
-        # self.assertTrue(tls_ctx.crypto.server.rsa.privkey.can_decrypt())
-        # self.assertTrue(tls_ctx.crypto.server.rsa.pubkey.can_decrypt())
-        self.assertTrue(tls_ctx.crypto.server.rsa.privkey.can_encrypt())
+        # self.assertTrue(tls_ctx.crypto.server.keystore.private.can_decrypt())
+        # self.assertTrue(tls_ctx.crypto.server.keystore.public.can_decrypt())
+        self.assertTrue(tls_ctx.crypto.server.keystore.private.can_encrypt())
         # TODO: Invertigate further: broken also in pycrypto. Should return False for public keys.
-        # self.assertFalse(tls_ctx.crypto.server.rsa.pubkey.can_encrypt())
+        # self.assertFalse(tls_ctx.crypto.server.keystore.public.can_encrypt())
 
     def test_decrypted_pms_matches_generated_pms(self):
         tls_ctx = tlsc.TLSSessionCtx()
@@ -296,8 +296,8 @@ def test_load_rsa_privkey_from_pem_file(self):
         pem_file = env_local_file("openssl_1_0_1_f_server.pem")
         tls_ctx = tlsc.TLSSessionCtx()
         tls_ctx.rsa_load_keys_from_file(pem_file)
-        self.assertTrue(tls_ctx.crypto.server.rsa.privkey)
-        self.assertTrue(tls_ctx.crypto.server.rsa.pubkey)
+        self.assertTrue(tls_ctx.crypto.server.keystore.private)
+        self.assertTrue(tls_ctx.crypto.server.keystore.public)
 
 
 class TestNullCompression(unittest.TestCase):

tests/test_ssl_tls_keystore.py

@@ -9,7 +9,7 @@ class TestAsymKeyStore(unittest.TestCase):
 
     def test_when_rsa_keystore_is_initialized_then_name_is_set(self):
         rsa_keystore = tlsk.RSAKeystore(b"5678")
-        self.assertEqual(rsa_keystore.name, "RSA")
-        self.assertEqual(rsa_keystore.private, None)
-        self.assertEqual(rsa_keystore.public, b"5678")
-        self.assertEqual(rsa_keystore.certificate, None)
+        self.assertEqual("RSA Keystore", rsa_keystore.name)
+        self.assertEqual(None, rsa_keystore.private)
+        self.assertEqual(b"5678", rsa_keystore.public)
+        self.assertEqual(None, rsa_keystore.certificate)