Merge pull request #46930 from dwoz/crptodomekeyfix

Clean up bad public key headers

salt/crypt.py

@@ -28,7 +28,7 @@ from salt.ext.six.moves import zip  # pylint: disable=import-error,redefined-bui
 from salt.ext import six
 
 try:
-    from M2Crypto import RSA, EVP
+    from M2Crypto import RSA, EVP, BIO
     HAS_M2 = True
 except ImportError:
     HAS_M2 = False
@@ -206,7 +206,10 @@ def get_rsa_pub_key(path):
     '''
     log.debug('salt.crypt.get_rsa_pub_key: Loading public key')
     if HAS_M2:
-        key = RSA.load_pub_key(path)
+        with salt.utils.files.fopen(path) as f:
+            data = f.read().replace(b'RSA ', '')
+        bio = BIO.MemoryBuffer(data)
+        key = RSA.load_pub_key_bio(bio)
     else:
         with salt.utils.files.fopen(path) as f:
             key = RSA.importKey(f.read())

tests/unit/test_crypt.py

@@ -3,6 +3,8 @@
 # python libs
 from __future__ import absolute_import
 import os
+import tempfile
+import shutil
 
 # salt testing libs
 from tests.support.unit import TestCase, skipIf
@@ -196,8 +198,7 @@ class M2CryptTestCase(TestCase):
             self.assertEqual(SIG, crypt.sign_message('/keydir/keyname.pem', MSG, passphrase='password'))
 
     def test_verify_signature(self):
-        key = M2Crypto.RSA.load_pub_key_bio(M2Crypto.BIO.MemoryBuffer(six.b(PUBKEY_DATA)))
-        with patch('M2Crypto.RSA.load_pub_key', return_value=key):
+        with patch('salt.utils.files.fopen', mock_open(read_data=PUBKEY_DATA)):
             self.assertTrue(crypt.verify_signature('/keydir/keyname.pub', MSG, SIG))
 
     def test_encrypt_decrypt_bin(self):
@@ -206,3 +207,46 @@ class M2CryptTestCase(TestCase):
         encrypted = salt.crypt.private_encrypt(priv_key, b'salt')
         decrypted = salt.crypt.public_decrypt(pub_key, encrypted)
         self.assertEqual(b'salt', decrypted)
+
+
+class TestBadCryptodomePubKey(TestCase):
+    '''
+    Test that we can load public keys exported by pycrpytodome<=3.4.6
+    '''
+
+    TEST_KEY = (
+        '-----BEGIN RSA PUBLIC KEY-----\n'
+        'MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAzLtFhsvfbFDFaUgulSEX\n'
+        'Gl12XriL1DT78Ef2/u8HHaSMmPie37BLWas/zaHwI6066bIyYQJ/nUCahTaoHM7L\n'
+        'GlWc0wOU6zyfpihCRQHil05Y6F+olFBoZuYbFPvtp7/hJx/D7I/0n2o/c7M5i3Y2\n'
+        '3sBxAYNooIQHXHUmPQW6C9iu95ylZDW8JQzYy/EI4vCC8yQMdTK8jK1FQV0Sbwny\n'
+        'qcMxSyAWDoFbnhh2P2TnO8HOWuUOaXR8ZHOJzVcDl+a6ew+medW090x3K5O1f80D\n'
+        '+WjgnG6b2HG7VQpOCfM2GALD/FrxicPilvZ38X1aLhJuwjmVE4LAAv8DVNJXohaO\n'
+        'WQIDAQAB\n'
+        '-----END RSA PUBLIC KEY-----\n'
+    )
+
+    def setUp(self):
+        self.test_dir = tempfile.mkdtemp()
+        self.key_path = os.path.join(self.test_dir, 'cryptodom-3.4.6.pub')
+        with salt.utils.files.fopen(self.key_path, 'wb') as fd:
+            fd.write(self.TEST_KEY.encode())
+
+    def tearDown(self):
+        shutil.rmtree(self.test_dir)
+
+    @skipIf(not HAS_M2, "Skip when m2crypto is not installed")
+    def test_m2_bad_key(self):
+        '''
+        Load public key with an invalid header using m2crypto and validate it
+        '''
+        key = salt.crypt.get_rsa_pub_key(self.key_path)
+        assert key.check_key() == 1
+
+    @skipIf(HAS_M2, "Skip when m2crypto is installed")
+    def test_crypto_bad_key(self):
+        '''
+        Load public key with an invalid header and validate it without m2crypto
+        '''
+        key = salt.crypt.get_rsa_pub_key(self.key_path)
+        assert key.can_encrypt()