crypto: Add new module openssl_pkcs12 (#27320)

[module] Create new module to handle PKCS#12 file.

lib/ansible/modules/crypto/openssl_pkcs12.py

@@ -0,0 +1,371 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+#
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+# Copyright (c) 2017 Guillaume Delpierre <gde@llew.me>
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+ANSIBLE_METADATA = {'metadata_version': '1.1',
+                    'status': ['preview'],
+                    'supported_by': 'community'}
+
+DOCUMENTATION = '''
+---
+module: openssl_pkcs12
+author: "Guillaume Delpierre (@gdelpierre)"
+version_added: "2.7"
+short_description: Generate OpenSSL PKCS#12 archive.
+description:
+    - This module allows one to (re-)generate PKCS#12.
+requirements:
+    - python-pyOpenSSL
+options:
+    action:
+        default: export
+        choices: ['parse', 'export']
+        description:
+            - C(export) or C(parse) a PKCS#12.
+    ca_certificates:
+        description:
+            - List of CA certificate to include.
+    certificate_path:
+        description:
+            - The path to read certificates and private keys from.  Must be in PEM format.
+    force:
+        default: False
+        type: bool
+        description:
+            - Should the file be regenerated even if it already exists.
+    friendly_name:
+        aliases: ['name']
+        description:
+            - Specifies the friendly name for the certificate and private key.
+    iter_size:
+        default: 2048
+        description:
+            - Number of times to repeat the encryption step.
+    maciter_size:
+        default: 1
+        description:
+            - Number of times to repeat the MAC step.
+    passphrase:
+        description:
+            - The PKCS#12 password.
+    path:
+        required: True
+        description:
+            - Filename to write the PKCS#12 file to.
+    privatekey_passphrase:
+        description:
+            - Passphrase source to decrypt any input private keys with.
+    privatekey_path:
+        description:
+            - File to read private key from.
+    state:
+        default: 'present'
+        choices: ['present', 'absent']
+        description:
+            - Whether the file should exist or not.
+              All parameters except C(path) are ignored when state is C(absent).
+    src:
+        description:
+            - PKCS#12 file path to parse.
+
+extends_documentation_fragment:
+    - files
+'''
+
+EXAMPLES = '''
+- name: 'Generate PKCS#12 file'
+  openssl_pkcs12:
+    action: export
+    path: '/opt/certs/ansible.p12'
+    friendly_name: 'raclette'
+    privatekey_path: '/opt/certs/keys/key.pem'
+    certificate_path: '/opt/certs/cert.pem'
+    ca_certificates: '/opt/certs/ca.pem'
+    state: present
+
+- name: 'Change PKCS#12 file permission'
+  openssl_pkcs12:
+    action: export
+    path: '/opt/certs/ansible.p12'
+    friendly_name: 'raclette'
+    privatekey_path: '/opt/certs/keys/key.pem'
+    certificate_path: '/opt/certs/cert.pem'
+    ca_certificates: '/opt/certs/ca.pem'
+    state: present
+    mode: 0600
+
+- name: 'Regen PKCS#12 file'
+  openssl_pkcs12:
+    action: export
+    src: '/opt/certs/ansible.p12'
+    path: '/opt/certs/ansible.p12'
+    friendly_name: 'raclette'
+    privatekey_path: '/opt/certs/keys/key.pem'
+    certificate_path: '/opt/certs/cert.pem'
+    ca_certificates: '/opt/certs/ca.pem'
+    state: present
+    mode: 0600
+    force: True
+
+- name: 'Dump/Parse PKCS#12 file'
+  openssl_pkcs12:
+    action: parse
+    src: '/opt/certs/ansible.p12'
+    path: '/opt/certs/ansible.pem'
+    state: present
+
+- name: 'Remove PKCS#12 file'
+  openssl_pkcs12:
+    path: '/opt/certs/ansible.p12'
+    state: absent
+'''
+
+RETURN = '''
+filename:
+    description: Path to the generate PKCS#12 file.
+    returned: changed or success
+    type: string
+    sample: /opt/certs/ansible.p12
+privatekey:
+    description: Path to the TLS/SSL private key the public key was generated from
+    returned: changed or success
+    type: string
+    sample: /etc/ssl/private/ansible.com.pem
+'''
+
+import stat
+import os
+
+try:
+    from OpenSSL import crypto
+except ImportError:
+    pyopenssl_found = False
+else:
+    pyopenssl_found = True
+
+from ansible.module_utils.basic import AnsibleModule
+from ansible.module_utils import crypto as crypto_utils
+from ansible.module_utils._text import to_native
+
+
+class PkcsError(crypto_utils.OpenSSLObjectError):
+    pass
+
+
+class Pkcs(crypto_utils.OpenSSLObject):
+
+    def __init__(self, module):
+        super(Pkcs, self).__init__(
+            module.params['path'],
+            module.params['state'],
+            module.params['force'],
+            module.check_mode
+        )
+        self.action = module.params['action']
+        self.ca_certificates = module.params['ca_certificates']
+        self.certificate_path = module.params['certificate_path']
+        self.friendly_name = module.params['friendly_name']
+        self.iter_size = module.params['iter_size']
+        self.maciter_size = module.params['maciter_size']
+        self.passphrase = module.params['passphrase']
+        self.pkcs12 = None
+        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
+
+    def check(self, module, perms_required=True):
+        """Ensure the resource is in its desired state."""
+
+        state_and_perms = super(Pkcs, self).check(module, perms_required)
+
+        def _check_pkey_passphrase():
+            if self.privatekey_passphrase:
+                try:
+                    crypto_utils.load_privatekey(self.path,
+                                                 self.privatekey_passphrase)
+                except crypto.Error:
+                    return False
+            return True
+
+        if not state_and_perms:
+            return state_and_perms
+
+        return _check_pkey_passphrase
+
+    def dump(self):
+        """Serialize the object into a dictionary."""
+
+        result = {
+            'filename': self.path,
+        }
+        if self.privatekey_path:
+            result['privatekey_path'] = self.privatekey_path
+
+        return result
+
+    def generate(self, module):
+        """Generate PKCS#12 file archive."""
+
+        self.pkcs12 = crypto.PKCS12()
+
+        try:
+            self.remove()
+        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]
+            self.pkcs12.set_ca_certificates(ca_certs)
+
+        if self.certificate_path:
+            self.pkcs12.set_certificate(crypto_utils.load_certificate(
+                                        self.certificate_path))
+
+        if self.friendly_name:
+            self.pkcs12.set_friendlyname(self.friendly_name)
+
+        if self.privatekey_path:
+            self.pkcs12.set_privatekey(crypto_utils.load_privatekey(
+                                       self.privatekey_path,
+                                       self.privatekey_passphrase)
+                                       )
+
+        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()
+            raise PkcsError(exc)
+
+    def parse(self, module):
+        """Read PKCS#12 file."""
+
+        try:
+            self.remove()
+
+            p12 = crypto.load_pkcs12(open(self.src, 'rb').read(),
+                                     self.passphrase)
+            pkey = crypto.dump_privatekey(crypto.FILETYPE_PEM,
+                                          p12.get_privatekey())
+            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, '%s%s' % (pkey, crt))
+            os.close(pkcs12_file)
+
+        except IOError as exc:
+            self.remove()
+            raise PkcsError(exc)
+
+
+def main():
+    argument_spec = dict(
+        action=dict(type='str', default='export',
+                    choices=['parse', 'export']),
+        ca_certificates=dict(type='list'),
+        certificate_path=dict(type='path'),
+        force=dict(type='bool', default=False),
+        friendly_name=dict(type='str', aliases=['name']),
+        iter_size=dict(type='int', default=2048),
+        maciter_size=dict(type='int', default=1),
+        passphrase=dict(type='str', no_log=True),
+        path=dict(type='path', required=True),
+        privatekey_passphrase=dict(type='str', no_log=True),
+        privatekey_path=dict(type='path'),
+        state=dict(type='str', default='present',
+                   choices=['present', 'absent']),
+        src=dict(type='path'),
+    )
+
+    required_if = [
+        ['action', 'parse', ['src']],
+    ]
+
+    required_together = [
+        ['privatekey_path', 'friendly_name'],
+    ]
+
+    module = AnsibleModule(
+        add_file_common_args=True,
+        argument_spec=argument_spec,
+        required_if=required_if,
+        required_together=required_together,
+        supports_check_mode=True,
+    )
+
+    if not pyopenssl_found:
+        module.fail_json(msg='The python pyOpenSSL library is required')
+
+    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 path is not a directory' % base_dir
+        )
+
+    pkcs12 = Pkcs(module)
+    changed = False
+
+    if module.params['state'] == 'present':
+        if module.check_mode:
+            result = pkcs12.dump()
+            result['changed'] = module.params['force'] or not pkcs12.check(module)
+            module.exit_json(**result)
+
+        try:
+            if not pkcs12.check(module, perms_required=False) or module.params['force']:
+                if module.params['action'] == 'export':
+                    if not module.params['friendly_name']:
+                        module.fail_json(msg='Friendly_name is required')
+                    pkcs12.generate(module)
+                    changed = True
+                else:
+                    pkcs12.parse(module)
+
+            file_args = module.load_file_common_arguments(module.params)
+            if module.set_fs_attributes_if_different(file_args, changed):
+                changed = True
+
+        except PkcsError as exc:
+            module.fail_json(msg=to_native(exc))
+    else:
+        if module.check_mode:
+            result = pkcs12.dump()
+            result['changed'] = os.path.exists(module.params['path'])
+            module.exit_json(**result)
+
+        if os.path.exists(module.params['path']):
+            try:
+                pkcs12.remove()
+                changed = True
+            except PkcsError as exc:
+                module.fail_json(msg=to_native(exc))
+
+    result = pkcs12.dump()
+    result['changed'] = changed
+    if os.path.exists(module.params['path']):
+        file_mode = "%04o" % stat.S_IMODE(os.stat(module.params['path']).st_mode)
+        result['mode'] = file_mode
+
+    module.exit_json(**result)
+
+
+if __name__ == '__main__':
+    main()

test/integration/targets/openssl_pkcs12/aliases

@@ -0,0 +1,3 @@
+destructive
+needs/root
+shippable/posix/group1

test/integration/targets/openssl_pkcs12/meta/main.yml

@@ -0,0 +1,2 @@
+dependencies:
+  - setup_openssl

test/integration/targets/openssl_pkcs12/tasks/main.yml

@@ -0,0 +1,65 @@
+- block:
+  - name: 'Generate privatekey'
+    openssl_privatekey:
+      path: "{{ output_dir }}/ansible_pkey.pem"
+
+  - name: 'Generate CSR'
+    openssl_csr:
+      path: "{{ output_dir }}/ansible.csr"
+      privatekey_path: "{{ output_dir }}/ansible_pkey.pem"
+      commonName: 'www.ansible.com'
+
+  - name: 'Generate certificate'
+    openssl_certificate:
+      path: "{{ output_dir }}/ansible.crt"
+      privatekey_path: "{{ output_dir }}/ansible_pkey.pem"
+      csr_path: "{{ output_dir }}/ansible.csr"
+      provider: selfsigned
+
+  - name: 'Generate PKCS#12 file'
+    openssl_pkcs12:
+      path: "{{ output_dir }}/ansible.p12"
+      friendly_name: 'abracadabra'
+      privatekey_path: "{{ output_dir }}/ansible_pkey.pem"
+      certificate_path: "{{ output_dir }}/ansible.crt"
+      state: present
+    register: p12_standard
+
+  - name: 'Generate PKCS#12 file (force)'
+    openssl_pkcs12:
+      path: "{{ output_dir }}/ansible.p12"
+      friendly_name: 'abracadabra'
+      privatekey_path: "{{ output_dir }}/ansible_pkey.pem"
+      certificate_path: "{{ output_dir }}/ansible.crt"
+      state: present
+      force: True
+    register: p12_force
+
+  - name: 'Generate PKCS#12 file (force + change mode)'
+    openssl_pkcs12:
+      path: "{{ output_dir }}/ansible.p12"
+      friendly_name: 'abracadabra'
+      privatekey_path: "{{ output_dir }}/ansible_pkey.pem"
+      certificate_path: "{{ output_dir }}/ansible.crt"
+      state: present
+      force: True
+      mode: 0644
+    register: p12_force_and_mode
+
+  - name: 'Dump PKCS#12'
+    openssl_pkcs12:
+      src: "{{ output_dir }}/ansible.p12"
+      path: "{{ output_dir }}/ansible_parse.pem"
+      action: 'parse'
+      state: 'present'
+
+  - import_tasks: ../tests/validate.yml
+
+  always:
+  - name: 'Delete PKCS#12 file'
+    openssl_pkcs12:
+      state: absent
+      path: '{{ output_dir }}/ansible.p12'
+
+  # this is the pyopenssl version on my laptop.
+  when: pyopenssl_version.stdout is version_compare('17.1.0', '>=')

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

@@ -0,0 +1,16 @@
+- name: 'Install pexpect'
+  pip:
+    name: 'pexpect'
+    state: 'present'
+
+- name: 'Validate PKCS#12'
+  command: "openssl pkcs12 -info -in {{ output_dir }}/ansible.p12 -nodes -passin pass:''"
+  register: p12
+
+- name: 'Validate PKCS#12 (assert)'
+  assert:
+    that:
+      - p12.stdout_lines[2].split(':')[-1].strip() == 'abracadabra'
+      - p12_standard.mode == '0400'
+      - p12_force.changed
+      - p12_force_and_mode.mode == '0644' and p12_force_and_mode.changed