openssl_*: proper mode support

changelogs/fragments/54085-openssl-mode-writing.yaml

@@ -0,0 +1,5 @@
+minor_changes:
+- "openssl_pkcs12, openssl_privatekey, openssl_publickey - These modules no longer delete the output file before starting to regenerate the output, or when generating the output failed."
+bugfixes:
+- "openssl_pkcs12, openssl_privatekey - These modules now accept the output file mode in symbolic form or as a octal string (https://github.com/ansible/ansible/issues/53476)."
+- "openssl_certificate, openssl_csr, openssl_pkcs12, openssl_privatekey, openssl_publickey - The modules are now able to overwrite write-protected files (https://github.com/ansible/ansible/issues/48656)."

lib/ansible/module_utils/crypto.py

@@ -39,6 +39,7 @@
 import hashlib
 import os
 import re
+import tempfile
 
 from ansible.module_utils import six
 from ansible.module_utils._text import to_bytes
@@ -234,6 +235,49 @@ def select_message_digest(digest_string):
     return digest
 
 
+def write_file(module, content, default_mode=None):
+    '''
+    Writes content into destination file as securely as possible.
+    Uses file arguments from module.
+    '''
+    # Find out parameters for file
+    file_args = module.load_file_common_arguments(module.params)
+    if file_args['mode'] is None:
+        file_args['mode'] = default_mode
+    # Create tempfile name
+    tmp_fd, tmp_name = tempfile.mkstemp(prefix=b'.ansible_tmp')
+    try:
+        os.close(tmp_fd)
+    except Exception as dummy:
+        pass
+    module.add_cleanup_file(tmp_name)  # if we fail, let Ansible try to remove the file
+    try:
+        try:
+            # Create tempfile
+            file = os.open(tmp_name, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
+            os.write(file, content)
+            os.close(file)
+        except Exception as e:
+            try:
+                os.remove(tmp_name)
+            except Exception as dummy:
+                pass
+            module.fail_json(msg='Error while writing result into temporary file: {0}'.format(e))
+        # Update destination to wanted permissions
+        if os.path.exists(file_args['path']):
+            module.set_fs_attributes_if_different(file_args, False)
+        # Move tempfile to final destination
+        module.atomic_move(tmp_name, file_args['path'])
+        # Try to update permissions again
+        module.set_fs_attributes_if_different(file_args, False)
+    except Exception as e:
+        try:
+            os.remove(tmp_name)
+        except Exception as dummy:
+            pass
+        module.fail_json(msg='Error while writing result: {0}'.format(e))
+
+
 @six.add_metaclass(abc.ABCMeta)
 class OpenSSLObject(object):
 

lib/ansible/modules/crypto/openssl_certificate.py

@@ -746,12 +746,7 @@ def generate(self, module):
 
             self.cert = certificate
 
-            try:
-                with open(self.path, 'wb') as cert_file:
-                    cert_file.write(certificate.public_bytes(Encoding.PEM))
-            except Exception as exc:
-                raise CertificateError(exc)
-
+            crypto_utils.write_file(module, certificate.public_bytes(Encoding.PEM))
             self.changed = True
         else:
             self.cert = crypto_utils.load_certificate(self.path, backend=self.backend)
@@ -829,12 +824,7 @@ def generate(self, module):
             cert.sign(self.privatekey, self.digest)
             self.cert = cert
 
-            try:
-                with open(self.path, 'wb') as cert_file:
-                    cert_file.write(crypto.dump_certificate(crypto.FILETYPE_PEM, self.cert))
-            except EnvironmentError as exc:
-                raise CertificateError(exc)
-
+            crypto_utils.write_file(module, crypto.dump_certificate(crypto.FILETYPE_PEM, self.cert))
             self.changed = True
 
         file_args = module.load_file_common_arguments(module.params)
@@ -922,12 +912,7 @@ def generate(self, module):
 
             self.cert = certificate
 
-            try:
-                with open(self.path, 'wb') as cert_file:
-                    cert_file.write(certificate.public_bytes(Encoding.PEM))
-            except Exception as exc:
-                raise CertificateError(exc)
-
+            crypto_utils.write_file(module, certificate.public_bytes(Encoding.PEM))
             self.changed = True
         else:
             self.cert = crypto_utils.load_certificate(self.path, backend=self.backend)
@@ -1016,12 +1001,7 @@ def generate(self, module):
             cert.sign(self.ca_privatekey, self.digest)
             self.cert = cert
 
-            try:
-                with open(self.path, 'wb') as cert_file:
-                    cert_file.write(crypto.dump_certificate(crypto.FILETYPE_PEM, self.cert))
-            except EnvironmentError as exc:
-                raise CertificateError(exc)
-
+            crypto_utils.write_file(module, crypto.dump_certificate(crypto.FILETYPE_PEM, self.cert))
             self.changed = True
 
         file_args = module.load_file_common_arguments(module.params)
@@ -1760,8 +1740,7 @@ def generate(self, module):
                                                             self.csr_path,
                                                             self.challenge_path),
                                          check_rc=True)[1]
-                with open(self.path, 'wb') as certfile:
-                    certfile.write(to_bytes(crt))
+                crypto_utils.write_file(module, to_bytes(crt))
                 self.changed = True
             except OSError as exc:
                 raise CertificateError(exc)

lib/ansible/modules/crypto/openssl_csr.py

@@ -422,14 +422,7 @@ def generate(self, module):
         '''Generate the certificate signing request.'''
         if not self.check(module, perms_required=False) or self.force:
             result = self._generate_csr()
-
-            try:
-                csr_file = open(self.path, 'wb')
-                csr_file.write(result)
-                csr_file.close()
-            except (IOError, OSError) as exc:
-                raise CertificateSigningRequestError(exc)
-
+            crypto_utils.write_file(module, result)
             self.changed = True
 
         file_args = module.load_file_common_arguments(module.params)

lib/ansible/modules/crypto/openssl_pkcs12.py

@@ -199,9 +199,9 @@ def __init__(self, module):
         self.privatekey_passphrase = module.params['privatekey_passphrase']
         self.privatekey_path = module.params['privatekey_path']
         self.src = module.params['src']
-        self.mode = module.params['mode']
-        if not self.mode:
-            self.mode = 0o400
+
+        if module.params['mode'] is None:
+            module.params['mode'] = '0400'
 
     def check(self, module, perms_required=True):
         """Ensure the resource is in its desired state."""
@@ -240,11 +240,6 @@ def generate(self, module):
 
         self.pkcs12 = crypto.PKCS12()
 
-        try:
-            self.remove(module)
-        except PkcsError as exc:
-            module.fail_json(msg=to_native(exc))
-
         if self.ca_certificates:
             ca_certs = [crypto_utils.load_certificate(ca_cert) for ca_cert
                         in self.ca_certificates]
@@ -266,22 +261,16 @@ def generate(self, module):
             except crypto_utils.OpenSSLBadPassphraseError as exc:
                 raise PkcsError(exc)
 
-        try:
-            pkcs12_file = os.open(self.path,
-                                  os.O_WRONLY | os.O_CREAT | os.O_TRUNC,
-                                  self.mode)
-            os.write(pkcs12_file, self.pkcs12.export(self.passphrase,
-                                                     self.iter_size, self.maciter_size))
-            os.close(pkcs12_file)
-        except (IOError, OSError) as exc:
-            self.remove(module)
-            raise PkcsError(exc)
+        crypto_utils.write_file(
+            module,
+            self.pkcs12.export(self.passphrase, self.iter_size, self.maciter_size),
+            0o600
+        )
 
     def parse(self, module):
         """Read PKCS#12 file."""
 
         try:
-            self.remove(module)
             with open(self.src, 'rb') as pkcs12_fh:
                 pkcs12_content = pkcs12_fh.read()
             p12 = crypto.load_pkcs12(pkcs12_content,
@@ -291,14 +280,9 @@ def parse(self, module):
             crt = crypto.dump_certificate(crypto.FILETYPE_PEM,
                                           p12.get_certificate())
 
-            pkcs12_file = os.open(self.path,
-                                  os.O_WRONLY | os.O_CREAT | os.O_TRUNC,
-                                  self.mode)
-            os.write(pkcs12_file, b'%s%s' % (pkey, crt))
-            os.close(pkcs12_file)
+            crypto_utils.write_file(module, b'%s%s' % (pkey, crt))
 
         except IOError as exc:
-            self.remove(module)
             raise PkcsError(exc)
 
 

lib/ansible/modules/crypto/openssl_privatekey.py

@@ -275,9 +275,8 @@ def __init__(self, module):
         self.backup = module.params['backup']
         self.backup_path = None
 
-        self.mode = module.params.get('mode', None)
-        if self.mode is None:
-            self.mode = 0o600
+        if module.params['mode'] is None:
+            module.params['mode'] = '0600'
 
     @abc.abstractmethod
     def _generate_private_key_data(self):
@@ -294,26 +293,8 @@ def generate(self, module):
             if self.backup:
                 self.backup_file = module.backup_local(self.path)
             privatekey_data = self._generate_private_key_data()
-            try:
-                privatekey_file = os.open(self.path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC)
-                os.close(privatekey_file)
-                if isinstance(self.mode, string_types):
-                    try:
-                        self.mode = int(self.mode, 8)
-                    except ValueError as e:
-                        try:
-                            st = os.lstat(self.path)
-                            self.mode = AnsibleModule._symbolic_mode_to_octal(st, self.mode)
-                        except ValueError as e:
-                            module.fail_json(msg="%s" % to_native(e), exception=traceback.format_exc())
-                os.chmod(self.path, self.mode)
-                privatekey_file = os.open(self.path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, self.mode)
-                os.write(privatekey_file, privatekey_data)
-                os.close(privatekey_file)
-                self.changed = True
-            except IOError as exc:
-                self.remove()
-                raise PrivateKeyError(exc)
+            crypto_utils.write_file(module, privatekey_data, 0o600)
+            self.changed = True
 
         self.fingerprint = self._get_fingerprint()
         file_args = module.load_file_common_arguments(module.params)

lib/ansible/modules/crypto/openssl_publickey.py

@@ -197,16 +197,14 @@ def generate(self, module):
                     )
                     publickey_content = crypto.dump_publickey(crypto.FILETYPE_PEM, self.privatekey)
 
-                with open(self.path, 'wb') as publickey_file:
-                    publickey_file.write(publickey_content)
+                crypto_utils.write_file(module, publickey_content)
 
                 self.changed = True
             except crypto_utils.OpenSSLBadPassphraseError as exc:
                 raise PublicKeyError(exc)
             except (IOError, OSError) as exc:
                 raise PublicKeyError(exc)
             except AttributeError as exc:
-                self.remove(module)
                 raise PublicKeyError('You need to have PyOpenSSL>=16.0.0 to generate public keys')
 
         self.fingerprint = crypto_utils.get_fingerprint(

test/integration/targets/openssl_privatekey/tasks/impl.yml

@@ -203,3 +203,33 @@
     backup: yes
     state: absent
   register: remove_2
+
+- name: Generate privatekey_mode (mode 0400)
+  openssl_privatekey:
+    path: '{{ output_dir }}/privatekey_mode.pem'
+    mode: '0400'
+    select_crypto_backend: '{{ select_crypto_backend }}'
+  register: privatekey_mode_1
+- name: Stat for privatekey_mode
+  stat:
+    path: '{{ output_dir }}/privatekey_mode.pem'
+  register: privatekey_mode_1_stat
+
+- name: Generate privatekey_mode (mode 0400, idempotency)
+  openssl_privatekey:
+    path: '{{ output_dir }}/privatekey_mode.pem'
+    mode: '0400'
+    select_crypto_backend: '{{ select_crypto_backend }}'
+  register: privatekey_mode_2
+
+- name: Generate privatekey_mode (mode 0400, force)
+  openssl_privatekey:
+    path: '{{ output_dir }}/privatekey_mode.pem'
+    mode: '0400'
+    force: yes
+    select_crypto_backend: '{{ select_crypto_backend }}'
+  register: privatekey_mode_3
+- name: Stat for privatekey_mode
+  stat:
+    path: '{{ output_dir }}/privatekey_mode.pem'
+  register: privatekey_mode_3_stat

test/integration/targets/openssl_privatekey/tests/validate.yml

@@ -126,3 +126,13 @@
       - remove_2 is not changed
       - remove_1.backup_file is not none
       - remove_2.backup_file is undefined
+
+- name: Validate mode
+  assert:
+    that:
+      - privatekey_mode_1 is changed
+      - privatekey_mode_1_stat.stat.mode == '0400'
+      - privatekey_mode_2 is not changed
+      - privatekey_mode_3 is changed
+      - privatekey_mode_3_stat.stat.mode == '0400'
+      - privatekey_mode_1_stat.stat.mtime != privatekey_mode_3_stat.stat.mtime