From 89785c2625016ad76d5dac6c1cfa298f9bdaacb5 Mon Sep 17 00:00:00 2001
From: "Daniel A. Wozniak" <dwozniak@saltstack.com>
Date: Fri, 6 Apr 2018 10:56:50 -0700
Subject: [PATCH] Clean up bad public key headers

Remove the "RSA" delaration from public key headers generated by
pycryptodome<=3.4.6

Add test for bad public key without m2crypto

Fix verify signature test

We load the key to memory first now so mocking salt.utils.fopen fixes
the signature verification test

Fix corrupt public key with m2crypto python3

m2crypto open file in rb mode for pub key
---
 salt/crypt.py            |  7 ++-
 tests/unit/test_crypt.py | 94 +++++++++++++++++++++++++++++++++++++++-
 2 files changed, 97 insertions(+), 4 deletions(-)

diff --git a/salt/crypt.py b/salt/crypt.py
index bd82d71ff5..7a95d3190a 100644
--- a/salt/crypt.py
+++ b/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
@@ -207,7 +207,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, 'rb') as f:
+            data = f.read().replace(b'RSA ', b'')
+        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())
diff --git a/tests/unit/test_crypt.py b/tests/unit/test_crypt.py
index 63f1e4cf48..fb8591fcda 100644
--- a/tests/unit/test_crypt.py
+++ b/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=six.b(PUBKEY_DATA))):
             self.assertTrue(crypt.verify_signature('/keydir/keyname.pub', MSG, SIG))
 
     def test_encrypt_decrypt_bin(self):
@@ -206,3 +207,92 @@ 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()
+
+
+class TestM2CryptoRegression47124(TestCase):
+
+    SIGNATURE = (
+        b'w\xac\xfe18o\xeb\xfb\x14+\x9e\xd1\xb7\x7fe}\xec\xd6\xe1P\x9e\xab'
+        b'\xb5\x07\xe0\xc1\xfd\xda#\x04Z\x8d\x7f\x0b\x1f}:~\xb2s\x860u\x02N'
+        b'\xd4q"\xb7\x86*\x8f\x1f\xd0\x9d\x11\x92\xc5~\xa68\xac>\x12H\xc2%y,'
+        b'\xe6\xceU\x1e\xa3?\x0c,\xf0u\xbb\xd0[g_\xdd\x8b\xb0\x95:Y\x18\xa5*'
+        b'\x99\xfd\xf3K\x92\x92 ({\xd1\xff\xd9F\xc8\xd6K\x86e\xf9\xa8\xad\xb0z'
+        b'\xe3\x9dD\xf5k\x8b_<\xe7\xe7\xec\xf3"\'\xd5\xd2M\xb4\xce\x1a\xe3$'
+        b'\x9c\x81\xad\xf9\x11\xf6\xf5>)\xc7\xdd\x03&\xf7\x86@ks\xa6\x05\xc2'
+        b'\xd0\xbd\x1a7\xfc\xde\xe6\xb0\xad!\x12#\xc86Y\xea\xc5\xe3\xe2\xb3'
+        b'\xc9\xaf\xfa\x0c\xf2?\xbf\x93w\x18\x9e\x0b\xa2a\x10:M\x05\x89\xe2W.Q'
+        b'\xe8;yGT\xb1\xf2\xc6A\xd2\xc4\xbeN\xb3\xcfS\xaf\x03f\xe2\xb4)\xe7\xf6'
+        b'\xdbs\xd0Z}8\xa4\xd2\x1fW*\xe6\x1c"\x8b\xd0\x18w\xb9\x7f\x9e\x96\xa3'
+        b'\xd9v\xf7\x833\x8e\x01'
+    )
+
+    @skipIf(not HAS_M2, "Skip when m2crypto is not installed")
+    def test_m2crypto_verify_bytes(self):
+        message = salt.utils.stringutils.to_unicode('meh')
+        with patch('salt.utils.files.fopen', mock_open(read_data=six.b(PUBKEY_DATA))):
+            salt.crypt.verify_signature('/keydir/keyname.pub', message, self.SIGNATURE)
+
+    @skipIf(not HAS_M2, "Skip when m2crypto is not installed")
+    def test_m2crypto_verify_unicode(self):
+        message = salt.utils.stringutils.to_bytes('meh')
+        with patch('salt.utils.files.fopen', mock_open(read_data=six.b(PUBKEY_DATA))):
+            salt.crypt.verify_signature('/keydir/keyname.pub', message, self.SIGNATURE)
+
+    @skipIf(not HAS_M2, "Skip when m2crypto is not installed")
+    def test_m2crypto_sign_bytes(self):
+        message = salt.utils.stringutils.to_unicode('meh')
+        key = M2Crypto.RSA.load_key_string(six.b(PRIVKEY_DATA))
+        with patch('salt.crypt.get_rsa_key', return_value=key):
+            signature = salt.crypt.sign_message('/keydir/keyname.pem', message, passphrase='password')
+        self.assertEqual(signature, self.SIGNATURE)
+
+    @skipIf(not HAS_M2, "Skip when m2crypto is not installed")
+    def test_m2crypto_sign_unicode(self):
+        message = salt.utils.stringutils.to_bytes('meh')
+        key = M2Crypto.RSA.load_key_string(six.b(PRIVKEY_DATA))
+        with patch('salt.crypt.get_rsa_key', return_value=key):
+            signature = salt.crypt.sign_message('/keydir/keyname.pem', message, passphrase='password')
+        self.assertEqual(signature, self.SIGNATURE)
-- 
2.17.1


