abstracted encryption support from Properties/CfgPrivateKeyCreator to StructFile

3 files changed, 108 insertions, 216 deletions

diff --git a/testsuite/Testsrc/Testlib/TestServer/TestPlugin/Testhelpers.py b/testsuite/Testsrc/Testlib/TestServer/TestPlugin/Testhelpers.py
index 75cc41a34..3d4df3e0b 100644
--- a/testsuite/Testsrc/Testlib/TestServer/TestPlugin/Testhelpers.py
+++ b/testsuite/Testsrc/Testlib/TestServer/TestPlugin/Testhelpers.py
@@ -1,5 +1,4 @@
 import os
-import re
 import sys
 import copy
 import lxml.etree
@@ -21,6 +20,12 @@ from common import *
 from TestServer.TestPlugin.Testbase import TestPlugin, TestDebuggable
 from TestServer.TestPlugin.Testinterfaces import TestGenerator
 
+try:
+    from Bcfg2.Encryption import EVPError
+    HAS_CRYPTO = True
+except:
+    HAS_CRYPTO = False
+
 
 def tostring(el):
     return lxml.etree.tostring(el, xml_declaration=False).decode('UTF-8')
@@ -674,6 +679,108 @@ class TestStructFile(TestXMLFileBacked):
         children[4] = standalone
         return (xdata, groups, subgroups, children, subchildren, standalone)
 
+    def test_Index(self):
+        has_crypto = Bcfg2.Server.Plugin.helpers.HAS_CRYPTO
+        Bcfg2.Server.Plugin.helpers.HAS_CRYPTO = False
+        TestXMLFileBacked.test_Index(self)
+        Bcfg2.Server.Plugin.helpers.HAS_CRYPTO = has_crypto
+
+    @skipUnless(HAS_CRYPTO, "No crypto libraries found, skipping")
+    def test_Index_crypto(self):
+        if not self.test_obj.encryption:
+            return skip("Encryption disabled on %s objects, skipping" %
+                        self.test_obj.__name__)
+
+        sf = self.get_obj()
+        sf.setup = Mock()
+        sf.setup.cfp.get.return_value = "strict"
+        sf._decrypt = Mock()
+        sf._decrypt.return_value = 'plaintext'
+        sf.data = '''
+<EncryptedData>
+  <Group name="test">
+    <Datum encrypted="foo">crypted</Datum>
+  </Group>
+  <Group name="test" negate="true">
+    <Datum>plain</Datum>
+  </Group>
+</EncryptedData>'''
+
+        # test successful decryption
+        sf.Index()
+        self.assertItemsEqual(
+            sf._decrypt.call_args_list,
+            [call(el) for el in sf.xdata.xpath("//*[@encrypted]")])
+        for el in sf.xdata.xpath("//*[@encrypted]"):
+            self.assertEqual(el.text, sf._decrypt.return_value)
+
+        # test failed decryption, strict
+        sf._decrypt.reset_mock()
+        sf._decrypt.side_effect = EVPError
+        self.assertRaises(PluginExecutionError, sf.Index)
+
+        # test failed decryption, lax
+        sf.setup.cfp.get.return_value = "lax"
+        sf._decrypt.reset_mock()
+        sf.Index()
+        self.assertItemsEqual(
+            sf._decrypt.call_args_list,
+            [call(el) for el in sf.xdata.xpath("//*[@encrypted]")])
+
+    @skipUnless(HAS_CRYPTO, "No crypto libraries found, skipping")
+    @patchIf(HAS_CRYPTO, "Bcfg2.Encryption.ssl_decrypt")
+    @patchIf(HAS_CRYPTO, "Bcfg2.Encryption.get_passphrases")
+    @patchIf(HAS_CRYPTO, "Bcfg2.Encryption.bruteforce_decrypt")
+    def test_decrypt(self, mock_bruteforce, mock_get_passphrases, mock_ssl):
+        sf = self.get_obj()
+
+        def reset():
+            mock_bruteforce.reset_mock()
+            mock_get_passphrases.reset_mock()
+            mock_ssl.reset_mock()
+
+        # test element without text contents
+        self.assertIsNone(sf._decrypt(lxml.etree.Element("Test")))
+        self.assertFalse(mock_bruteforce.called)
+        self.assertFalse(mock_get_passphrases.called)
+        self.assertFalse(mock_ssl.called)
+
+        # test element with a passphrase in the config file
+        reset()
+        el = lxml.etree.Element("Test", encrypted="foo")
+        el.text = "crypted"
+        mock_get_passphrases.return_value = dict(foo="foopass", bar="barpass")
+        mock_ssl.return_value = "decrypted with ssl"
+        self.assertEqual(sf._decrypt(el), mock_ssl.return_value)
+        mock_get_passphrases.assert_called_with()
+        mock_ssl.assert_called_with(el.text, "foopass")
+        self.assertFalse(mock_bruteforce.called)
+
+        # test failure to decrypt element with a passphrase in the config
+        reset()
+        mock_ssl.side_effect = EVPError
+        self.assertRaises(EVPError, sf._decrypt, el)
+        mock_get_passphrases.assert_called_with()
+        mock_ssl.assert_called_with(el.text, "foopass")
+        self.assertFalse(mock_bruteforce.called)
+
+        # test element without valid passphrase
+        reset()
+        el.set("encrypted", "true")
+        mock_bruteforce.return_value = "decrypted with bruteforce"
+        self.assertEqual(sf._decrypt(el), mock_bruteforce.return_value)
+        mock_get_passphrases.assert_called_with()
+        mock_bruteforce.assert_called_with(el.text)
+        self.assertFalse(mock_ssl.called)
+
+        # test failure to decrypt element without valid passphrase
+        reset()
+        mock_bruteforce.side_effect = EVPError
+        self.assertRaises(EVPError, sf._decrypt, el)
+        mock_get_passphrases.assert_called_with()
+        mock_bruteforce.assert_called_with(el.text)
+        self.assertFalse(mock_ssl.called)
+
     def test_include_element(self):
         sf = self.get_obj()
         metadata = Mock()
diff --git a/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgPrivateKeyCreator.py b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgPrivateKeyCreator.py
index 4c8ab8b43..bf34d4c3c 100644
--- a/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgPrivateKeyCreator.py
+++ b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgPrivateKeyCreator.py
@@ -306,102 +306,3 @@ class TestCfgPrivateKeyCreator(TestCfgCreator, TestStructFile):
 
             inner2()
 
-    def test_Index(self):
-        has_crypto = Bcfg2.Server.Plugins.Cfg.CfgPrivateKeyCreator.HAS_CRYPTO
-        Bcfg2.Server.Plugins.Cfg.CfgPrivateKeyCreator.HAS_CRYPTO = False
-        TestStructFile.test_Index(self)
-        Bcfg2.Server.Plugins.Cfg.CfgPrivateKeyCreator.HAS_CRYPTO = has_crypto
-
-    @skipUnless(HAS_CRYPTO, "No crypto libraries found, skipping")
-    def test_Index_crypto(self):
-        pkc = self.get_obj()
-        pkc.setup = Mock()
-        pkc.setup.cfp.get.return_value = "strict"
-        pkc._decrypt = Mock()
-        pkc._decrypt.return_value = 'plaintext'
-        pkc.data = '''
-<PrivateKey>
-  <Group name="test">
-    <Passphrase encrypted="foo">crypted</Passphrase>
-  </Group>
-  <Group name="test" negate="true">
-    <Passphrase>plain</Passphrase>
-  </Group>
-</PrivateKey>'''
-
-        # test successful decryption
-        pkc.Index()
-        self.assertItemsEqual(
-            pkc._decrypt.call_args_list,
-            [call(el)
-             for el in pkc.xdata.xpath("//Passphrase[@encrypted]")])
-        for el in pkc.xdata.xpath("//Crypted"):
-            self.assertEqual(el.text, pkc._decrypt.return_value)
-
-        # test failed decryption, strict
-        pkc._decrypt.reset_mock()
-        pkc._decrypt.side_effect = EVPError
-        self.assertRaises(PluginExecutionError, pkc.Index)
-
-        # test failed decryption, lax
-        pkc.setup.cfp.get.return_value = "lax"
-        pkc._decrypt.reset_mock()
-        pkc.Index()
-        self.assertItemsEqual(
-            pkc._decrypt.call_args_list,
-            [call(el)
-             for el in pkc.xdata.xpath("//Passphrase[@encrypted]")])
-
-    @skipUnless(HAS_CRYPTO, "No crypto libraries found, skipping")
-    @patchIf(HAS_CRYPTO, "Bcfg2.Encryption.ssl_decrypt")
-    @patchIf(HAS_CRYPTO, "Bcfg2.Encryption.get_passphrases")
-    @patchIf(HAS_CRYPTO, "Bcfg2.Encryption.bruteforce_decrypt")
-    def test_decrypt(self, mock_bruteforce, mock_get_passphrases, mock_ssl):
-        pkc = self.get_obj()
-
-        def reset():
-            mock_bruteforce.reset_mock()
-            mock_get_passphrases.reset_mock()
-            mock_ssl.reset_mock()
-
-        # test element without text contents
-        self.assertIsNone(pkc._decrypt(lxml.etree.Element("Test")))
-        self.assertFalse(mock_bruteforce.called)
-        self.assertFalse(mock_get_passphrases.called)
-        self.assertFalse(mock_ssl.called)
-
-        # test element with a passphrase in the config file
-        reset()
-        el = lxml.etree.Element("Test", encrypted="foo")
-        el.text = "crypted"
-        mock_get_passphrases.return_value = dict(foo="foopass", bar="barpass")
-        mock_ssl.return_value = "decrypted with ssl"
-        self.assertEqual(pkc._decrypt(el), mock_ssl.return_value)
-        mock_get_passphrases.assert_called_with()
-        mock_ssl.assert_called_with(el.text, "foopass")
-        self.assertFalse(mock_bruteforce.called)
-
-        # test failure to decrypt element with a passphrase in the config
-        reset()
-        mock_ssl.side_effect = EVPError
-        self.assertRaises(EVPError, pkc._decrypt, el)
-        mock_get_passphrases.assert_called_with()
-        mock_ssl.assert_called_with(el.text, "foopass")
-        self.assertFalse(mock_bruteforce.called)
-
-        # test element without valid passphrase
-        reset()
-        el.set("encrypted", "true")
-        mock_bruteforce.return_value = "decrypted with bruteforce"
-        self.assertEqual(pkc._decrypt(el), mock_bruteforce.return_value)
-        mock_get_passphrases.assert_called_with()
-        mock_bruteforce.assert_called_with(el.text)
-        self.assertFalse(mock_ssl.called)
-
-        # test failure to decrypt element without valid passphrase
-        reset()
-        mock_bruteforce.side_effect = EVPError
-        self.assertRaises(EVPError, pkc._decrypt, el)
-        mock_get_passphrases.assert_called_with()
-        mock_bruteforce.assert_called_with(el.text)
-        self.assertFalse(mock_ssl.called)
diff --git a/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestProperties.py b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestProperties.py
index b0ca77a78..b63d08524 100644
--- a/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestProperties.py
+++ b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestProperties.py
@@ -19,12 +19,6 @@ from TestPlugin import TestStructFile, TestFileBacked, TestConnector, \
     TestPlugin, TestDirectoryBacked
 
 try:
-    from Bcfg2.Encryption import EVPError
-    HAS_CRYPTO = True
-except:
-    HAS_CRYPTO = False
-
-try:
     import json
     JSON = "json"
 except ImportError:
@@ -243,116 +237,6 @@ class TestXMLPropertyFile(TestPropertyFile, TestStructFile):
         mock_exists.assert_called_with(schemafile)
         mock_XMLSchema.assert_called_with(file=schemafile)
 
-    def test_Index(self):
-        TestStructFile.test_Index(self)
-
-        pf = self.get_obj()
-        pf.xdata = lxml.etree.Element("Properties")
-        lxml.etree.SubElement(pf.xdata, "Crypted", encrypted="foo")
-        pf.data = lxml.etree.tostring(pf.xdata)
-        # extra test: crypto is not available, but properties file is
-        # encrypted
-        has_crypto = Bcfg2.Server.Plugins.Properties.HAS_CRYPTO
-        Bcfg2.Server.Plugins.Properties.HAS_CRYPTO = False
-        try:
-            self.assertRaises(PluginExecutionError, pf.Index)
-        finally:
-            Bcfg2.Server.Plugins.Properties.HAS_CRYPTO = has_crypto
-
-    @skipUnless(HAS_CRYPTO, "No crypto libraries found, skipping")
-    def test_Index_crypto(self):
-        pf = self.get_obj()
-        pf._decrypt = Mock()
-        pf._decrypt.return_value = 'plaintext'
-        pf.data = '''
-<Properties decrypt="strict">
-  <Crypted encrypted="foo">
-    crypted
-    <Plain foo="bar">plain</Plain>
-  </Crypted>
-  <Crypted encrypted="bar">crypted</Crypted>
-  <Plain bar="baz">plain</Plain>
-  <Plain>
-    <Crypted encrypted="foo">crypted</Crypted>
-  </Plain>
-</Properties>'''
-
-        print "HAS_CRYPTO: %s" % HAS_CRYPTO
-        print "Properties HAS_CRYPTO: %s" % Bcfg2.Server.Plugins.Properties.HAS_CRYPTO
-
-        # test successful decryption
-        pf.Index()
-        self.assertItemsEqual(pf._decrypt.call_args_list,
-                              [call(el) for el in pf.xdata.xpath("//Crypted")])
-        for el in pf.xdata.xpath("//Crypted"):
-            self.assertEqual(el.text, pf._decrypt.return_value)
-
-        # test failed decryption, strict
-        pf._decrypt.reset_mock()
-        pf._decrypt.side_effect = EVPError
-        self.assertRaises(PluginExecutionError, pf.Index)
-
-        # test failed decryption, lax
-        pf.data = pf.data.replace("strict", "lax")
-        pf._decrypt.reset_mock()
-        pf.Index()
-        self.assertItemsEqual(pf._decrypt.call_args_list,
-                              [call(el) for el in pf.xdata.xpath("//Crypted")])
-
-    @skipUnless(HAS_CRYPTO, "No crypto libraries found, skipping")
-    @patchIf(HAS_CRYPTO, "Bcfg2.Encryption.ssl_decrypt")
-    @patchIf(HAS_CRYPTO, "Bcfg2.Encryption.get_passphrases")
-    @patchIf(HAS_CRYPTO, "Bcfg2.Encryption.bruteforce_decrypt")
-    def test_decrypt(self, mock_bruteforce, mock_get_passphrases, mock_ssl):
-        pf = self.get_obj()
-
-        def reset():
-            mock_bruteforce.reset_mock()
-            mock_get_passphrases.reset_mock()
-            mock_ssl.reset_mock()
-
-        # test element without text contents
-        self.assertIsNone(pf._decrypt(lxml.etree.Element("Test")))
-        self.assertFalse(mock_bruteforce.called)
-        self.assertFalse(mock_get_passphrases.called)
-        self.assertFalse(mock_ssl.called)
-
-        # test element with a passphrase in the config file
-        reset()
-        el = lxml.etree.Element("Test", encrypted="foo")
-        el.text = "crypted"
-        mock_get_passphrases.return_value = dict(foo="foopass", bar="barpass")
-        mock_ssl.return_value = "decrypted with ssl"
-        self.assertEqual(pf._decrypt(el), mock_ssl.return_value)
-        mock_get_passphrases.assert_called_with()
-        mock_ssl.assert_called_with(el.text, "foopass")
-        self.assertFalse(mock_bruteforce.called)
-
-        # test failure to decrypt element with a passphrase in the config
-        reset()
-        mock_ssl.side_effect = EVPError
-        self.assertRaises(EVPError, pf._decrypt, el)
-        mock_get_passphrases.assert_called_with()
-        mock_ssl.assert_called_with(el.text, "foopass")
-        self.assertFalse(mock_bruteforce.called)
-
-        # test element without valid passphrase
-        reset()
-        el.set("encrypted", "true")
-        mock_bruteforce.return_value = "decrypted with bruteforce"
-        self.assertEqual(pf._decrypt(el), mock_bruteforce.return_value)
-        mock_get_passphrases.assert_called_with()
-        mock_bruteforce.assert_called_with(el.text)
-        self.assertFalse(mock_ssl.called)
-
-        # test failure to decrypt element without valid passphrase
-        reset()
-        mock_bruteforce.side_effect = EVPError
-        self.assertRaises(EVPError, pf._decrypt, el)
-        mock_get_passphrases.assert_called_with()
-        mock_bruteforce.assert_called_with(el.text)
-        self.assertFalse(mock_ssl.called)
-
     @patch("copy.copy")
     def test_get_additional_data(self, mock_copy):
         pf = self.get_obj()