openssl_privatekey: Standardize implementaton of the module

The OpenSSLObject class has been merged[1]. This commit makes the
openssl_privatekey rely on this class and standardize the way openssl
module should be written.

Co-Authored-By: Christian Pointner <cpointner@mgit.at>

[1] https://github.com/ansible/ansible/pull/26945

lib/ansible/module_utils/crypto.py

@@ -95,11 +95,20 @@ class OpenSSLObject(object):
         self.changed = False
         self.check_mode = check_mode
 
-    @abc.abstractmethod
-    def check(self):
+    def check(self, module, perms_required=True):
         """Ensure the resource is in its desired state."""
 
-        pass
+        def _check_state():
+            return os.path.exists(self.path)
+
+        def _check_perms(module):
+            file_args = module.load_file_common_arguments(module.params)
+            return not module.set_fs_attributes_if_different(file_args, False)
+
+        if not perms_required:
+            return _check_state()
+
+        return _check_state() and _check_perms(module)
 
     @abc.abstractmethod
     def dump(self):

lib/ansible/modules/crypto/openssl_privatekey.py

@@ -132,9 +132,11 @@ fingerprint:
       sha512: "fd:ed:5e:39:48:5f:9f:fe:7f:25:06:3f:79:08:cd:ee:a5:e7:b3:3d:13:82:87:1f:84:e1:f5:c7:28:77:53:94:86:56:38:69:f0:d9:35:22:01:1e:a6:60:...:0f:9b"
 '''
 
-import errno
+import os
+
+from ansible.module_utils import crypto as crypto_utils
+from ansible.module_utils._text import to_native
 from ansible.module_utils.basic import AnsibleModule
-from ansible.module_utils.crypto import get_fingerprint
 from ansible.module_utils.pycompat24 import get_exception
 
 try:
@@ -144,44 +146,42 @@ except ImportError:
 else:
     pyopenssl_found = True
 
-import os
-from ansible.module_utils._text import to_native
-
 
-class PrivateKeyError(Exception):
+class PrivateKeyError(crypto_utils.OpenSSLObjectError):
     pass
 
 
-class PrivateKey(object):
+class PrivateKey(crypto_utils.OpenSSLObject):
 
     def __init__(self, module):
+        super(PrivateKey, self).__init__(
+            module.params['path'],
+            module.params['state'],
+            module.params['force'],
+            module.check_mode
+        )
         self.size = module.params['size']
-        self.state = module.params['state']
-        self.name = os.path.basename(module.params['path'])
-        self.type = module.params['type']
-        self.force = module.params['force']
-        self.path = module.params['path']
         self.passphrase = module.params['passphrase']
         self.cipher = module.params['cipher']
-        self.mode = module.params['mode']
-        self.changed = True
         self.privatekey = None
         self.fingerprint = {}
-        self.check_mode = module.check_mode
+
+        self.mode = module.params['mode']
+        if not self.mode:
+            self.mode = int('0600', 8)
+
+        self.type = crypto.TYPE_RSA
+        if module.params['type'] == 'DSA':
+            self.type = crypto.TYPE_DSA
 
     def generate(self, module):
         """Generate a keypair."""
 
-        if not os.path.exists(self.path) or self.force:
+        if not self.check(module, perms_required=False) or self.force:
             self.privatekey = crypto.PKey()
 
-            if self.type == 'RSA':
-                crypto_type = crypto.TYPE_RSA
-            else:
-                crypto_type = crypto.TYPE_DSA
-
             try:
-                self.privatekey.generate_key(crypto_type, self.size)
+                self.privatekey.generate_key(self.type, self.size)
             except (TypeError, ValueError) as exc:
                 raise PrivateKeyError(exc)
 
@@ -196,39 +196,56 @@ class PrivateKey(object):
 
                 os.write(privatekey_file, crypto.dump_privatekey(crypto.FILETYPE_PEM, self.privatekey, **extras))
                 os.close(privatekey_file)
+                self.changed = True
             except IOError as exc:
                 self.remove()
                 raise PrivateKeyError(exc)
-        else:
-            self.changed = False
 
-        self.fingerprint = get_fingerprint(self.path, self.passphrase)
+        self.fingerprint = crypto_utils.get_fingerprint(self.path, self.passphrase)
         file_args = module.load_file_common_arguments(module.params)
         if module.set_fs_attributes_if_different(file_args, False):
             self.changed = True
 
-    def remove(self):
-        """Remove the private key from the filesystem."""
+    def check(self, module, perms_required=True):
+        """Ensure the resource is in its desired state."""
+
+        state_and_perms = super(PrivateKey, self).check(module, perms_required)
+
+        def _check_size(privatekey):
+            return self.size == privatekey.bits()
+
+        def _check_type(privatekey):
+            return self.type == privatekey.type()
 
+        def _check_passphrase():
             try:
-            os.remove(self.path)
-        except OSError as exc:
-            if exc.errno != errno.ENOENT:
-                raise PrivateKeyError(exc)
-            else:
-                self.changed = False
+                crypto_utils.load_privatekey(self.path, self.passphrase)
+                return True
+            except crypto.Error:
+                return False
+
+        if not state_and_perms or not _check_passphrase():
+            return False
+
+        privatekey = crypto_utils.load_privatekey(self.path, self.passphrase)
+
+        return _check_size(privatekey) and _check_type(privatekey)
 
     def dump(self):
         """Serialize the object into a dictionary."""
 
         result = {
             'size': self.size,
-            'type': self.type,
             'filename': self.path,
             'changed': self.changed,
             'fingerprint': self.fingerprint,
         }
 
+        if self.type == crypto.TYPE_RSA:
+            result['type'] = 'RSA'
+        else:
+            result['type'] = 'DSA'
+
         return result
 
 
@@ -252,21 +269,20 @@ def main():
     if not pyopenssl_found:
         module.fail_json(msg='the python pyOpenSSL module is required')
 
-    path = module.params['path']
     base_dir = os.path.dirname(module.params['path'])
-
     if not os.path.isdir(base_dir):
-        module.fail_json(name=base_dir, msg='The directory %s does not exist or the file is not a directory' % base_dir)
-
-    if not module.params['mode']:
-        module.params['mode'] = int('0600', 8)
+        module.fail_json(
+            name=base_dir,
+            msg='The directory %s does not exist or the file is not a directory' % base_dir
+        )
 
     private_key = PrivateKey(module)
+
     if private_key.state == 'present':
 
         if module.check_mode:
             result = private_key.dump()
-            result['changed'] = module.params['force'] or not os.path.exists(path)
+            result['changed'] = module.params['force'] or not private_key.check(module)
             module.exit_json(**result)
 
         try:
@@ -277,7 +293,7 @@ def main():
 
         if module.check_mode:
             result = private_key.dump()
-            result['changed'] = os.path.exists(path)
+            result['changed'] = os.path.exists(module.params['path'])
             module.exit_json(**result)
 
         try:

test/sanity/pep8/legacy-files.txt

@@ -209,7 +209,6 @@ lib/ansible/modules/clustering/consul_session.py
 lib/ansible/modules/clustering/kubernetes.py
 lib/ansible/modules/clustering/pacemaker_cluster.py
 lib/ansible/modules/commands/command.py
-lib/ansible/modules/crypto/openssl_privatekey.py
 lib/ansible/modules/crypto/openssl_publickey.py
 lib/ansible/modules/database/misc/elasticsearch_plugin.py
 lib/ansible/modules/database/misc/kibana_plugin.py