openssl_pkcs12: allow to specify certificate bundles in other_certificates (#166)

* Rename identify.py to pem.py.

* Move split PEM list code to pem.py crypto module_utils.

* Extend and use global certificate splitting code in acme_certificate.

* openssl_pkcs12: allow to load multiple certificates from files mentioned in other_certificates.

* Add changelog and module_utils redirect.

* Remove old check.

* Fix typo.

* Apply suggestions from code review

Co-authored-by: Andrew Klychkov <aaklychkov@mail.ru>

* Add example.

Co-authored-by: Andrew Klychkov <aaklychkov@mail.ru>

changelogs/fragments/166-openssl_pkcs12-certificate-bundles.yml

@@ -0,0 +1,3 @@
+minor_changes:
+- "openssl_pkcs12 - allow to specify certificate bundles in ``other_certificates`` by using new option ``other_certificates_parse_all`` (https://github.com/ansible-collections/community.crypto/issues/149, https://github.com/ansible-collections/community.crypto/pull/166)."
+- "The ``crypto/identify.py`` module_utils has been renamed to ``crypto/pem.py`` (https://github.com/ansible-collections/community.crypto/pull/166)."

meta/runtime.yml

@@ -23,3 +23,9 @@ plugin_routing:
       deprecation:
         removal_version: 2.0.0
         warning_text: The 'community.crypto.openssl_certificate_info' module has been renamed to 'community.crypto.x509_certificate_info'
+  module_utils:
+    crypto.identify:
+      redirect: community.crypto.crypto.pem
+      deprecation:
+        removal_version: 2.0.0
+        warning_text: The 'crypto/identify.py' module_utils has been renamed 'crypto/pem.py'. Please update your imports

plugins/module_utils/crypto/__init__.py

@@ -55,7 +55,7 @@ from .cryptography_support import (
     cryptography_compare_public_keys,
 )
 
-from .identify import (
+from .pem import (
     identify_private_key_format,
 )
 

plugins/module_utils/crypto/module_backends/privatekey.py

@@ -33,7 +33,7 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.support im
     get_fingerprint_of_privatekey,
 )
 
-from ansible_collections.community.crypto.plugins.module_utils.crypto.identify import (
+from ansible_collections.community.crypto.plugins.module_utils.crypto.pem import (
     identify_private_key_format,
 )
 

plugins/module_utils/crypto/pem.py

@@ -54,3 +54,21 @@ def identify_private_key_format(content):
     except UnicodeDecodeError:
         pass
     return 'raw'
+
+
+def split_pem_list(text, keep_inbetween=False):
+    '''
+    Split concatenated PEM objects into a list of strings, where each is one PEM object.
+    '''
+    result = []
+    current = [] if keep_inbetween else None
+    for line in text.splitlines(True):
+        if line.strip():
+            if not keep_inbetween and line.startswith('-----BEGIN '):
+                current = []
+            if current is not None:
+                current.append(line)
+                if line.startswith('-----END '):
+                    result.append(''.join(current))
+                    current = [] if keep_inbetween else None
+    return result

plugins/modules/acme_certificate.py

@@ -526,6 +526,10 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptograp
     cryptography_name_to_oid,
 )
 
+from ansible_collections.community.crypto.plugins.module_utils.crypto.pem import (
+    split_pem_list,
+)
+
 from ansible_collections.community.crypto.plugins.module_utils.acme import (
     ModuleFailException,
     write_file,
@@ -829,17 +833,10 @@ class ACMEClient(object):
         chain = []
 
         # Parse data
-        lines = content.decode('utf-8').splitlines(True)
-        current = []
-        for line in lines:
-            if line.strip():
-                current.append(line)
-            if line.startswith('-----END CERTIFICATE-----'):
-                if cert is None:
-                    cert = ''.join(current)
-                else:
-                    chain.append(''.join(current))
-                current = []
+        certs = split_pem_list(content.decode('utf-8'), keep_inbetween=True)
+        if certs:
+            cert = certs[0]
+            chain = certs[1:]
 
         alternates = []
 
@@ -855,7 +852,7 @@ class ACMEClient(object):
 
         process_links(info, f)
 
-        if cert is None or current:
+        if cert is None:
             raise ModuleFailException("Failed to parse certificate chain download from {0}: {1} (headers: {2})".format(url, content, info))
         return {'cert': cert, 'chain': chain, 'alternates': alternates}
 

plugins/modules/certificate_complete_chain.py

@@ -124,6 +124,10 @@ import traceback
 from ansible.module_utils.basic import AnsibleModule, missing_required_lib
 from ansible.module_utils._text import to_bytes
 
+from ansible_collections.community.crypto.plugins.module_utils.crypto.pem import (
+    split_pem_list,
+)
+
 CRYPTOGRAPHY_IMP_ERR = None
 try:
     import cryptography
@@ -194,27 +198,17 @@ def parse_PEM_list(module, text, source, fail_on_error=True):
     Parse concatenated PEM certificates. Return list of ``Certificate`` objects.
     '''
     result = []
-    lines = text.splitlines(True)
-    current = None
-    for line in lines:
-        if line.strip():
-            if line.startswith('-----BEGIN '):
-                current = [line]
-            elif current is not None:
-                current.append(line)
-                if line.startswith('-----END '):
-                    cert_pem = ''.join(current)
-                    current = None
-                    # Try to load PEM certificate
-                    try:
-                        cert = cryptography.x509.load_pem_x509_certificate(to_bytes(cert_pem), _cryptography_backend)
-                        result.append(Certificate(cert_pem, cert))
-                    except Exception as e:
-                        msg = 'Cannot parse certificate #{0} from {1}: {2}'.format(len(result) + 1, source, e)
-                        if fail_on_error:
-                            module.fail_json(msg=msg)
-                        else:
-                            module.warn(msg)
+    for cert_pem in split_pem_list(text):
+        # Try to load PEM certificate
+        try:
+            cert = cryptography.x509.load_pem_x509_certificate(to_bytes(cert_pem), _cryptography_backend)
+            result.append(Certificate(cert_pem, cert))
+        except Exception as e:
+            msg = 'Cannot parse certificate #{0} from {1}: {2}'.format(len(result) + 1, source, e)
+            if fail_on_error:
+                module.fail_json(msg=msg)
+            else:
+                module.warn(msg)
     return result
 
 

plugins/modules/openssl_pkcs12.py

@@ -27,10 +27,19 @@ options:
         choices: [ export, parse ]
     other_certificates:
         description:
-            - List of other certificates to include. Pre 2.8 this parameter was called C(ca_certificates)
+            - List of other certificates to include. Pre Ansible 2.8 this parameter was called I(ca_certificates).
+            - Assumes there is one PEM-encoded certificate per file. If a file contains multiple PEM certificates,
+              set I(other_certificates_parse_all) to C(true).
         type: list
         elements: path
         aliases: [ ca_certificates ]
+    other_certificates_parse_all:
+        description:
+            - If set to C(true), assumes that the files mentioned in I(other_certificates) can contain more than one
+              certificate per file (or even none per file).
+        type: bool
+        default: false
+        version_added: 1.4.0
     certificate_path:
         description:
             - The path to read certificates and private keys from.
@@ -115,6 +124,27 @@ EXAMPLES = r'''
     privatekey_path: /opt/certs/keys/key.pem
     certificate_path: /opt/certs/cert.pem
     other_certificates: /opt/certs/ca.pem
+    # Note that if /opt/certs/ca.pem contains multiple certificates,
+    # only the first one will be used. See the other_certificates_parse_all
+    # option for changing this behavior.
+    state: present
+
+- name: Generate PKCS#12 file
+  community.crypto.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
+    other_certificates_parse_all: true
+    other_certificates:
+      - /opt/certs/ca_bundle.pem
+        # Since we set other_certificates_parse_all to true, all
+        # certificates in the CA bundle are included and not just
+        # the first one.
+      - /opt/certs/intermediate.pem
+        # In case this file has multiple certificates in it,
+        # all will be included as well.
     state: present
 
 - name: Change PKCS#12 file permission
@@ -201,6 +231,10 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.support im
     load_certificate,
 )
 
+from ansible_collections.community.crypto.plugins.module_utils.crypto.pem import (
+    split_pem_list,
+)
+
 PYOPENSSL_IMP_ERR = None
 try:
     from OpenSSL import crypto
@@ -211,6 +245,15 @@ else:
     pyopenssl_found = True
 
 
+def load_certificate_set(filename):
+    '''
+    Load list of concatenated PEM files, and return a list of parsed certificates.
+    '''
+    with open(filename, 'rb') as f:
+        data = f.read().decode('utf-8')
+    return [load_certificate(None, content=cert) for cert in split_pem_list(data)]
+
+
 class PkcsError(OpenSSLObjectError):
     pass
 
@@ -226,6 +269,7 @@ class Pkcs(OpenSSLObject):
         )
         self.action = module.params['action']
         self.other_certificates = module.params['other_certificates']
+        self.other_certificates_parse_all = module.params['other_certificates_parse_all']
         self.certificate_path = module.params['certificate_path']
         self.friendly_name = module.params['friendly_name']
         self.iter_size = module.params['iter_size']
@@ -244,6 +288,17 @@ class Pkcs(OpenSSLObject):
         self.backup = module.params['backup']
         self.backup_file = None
 
+        if self.other_certificates:
+            if self.other_certificates_parse_all:
+                filenames = list(self.other_certificates)
+                self.other_certificates = []
+                for other_cert_bundle in filenames:
+                    self.other_certificates.extend(load_certificate_set(other_cert_bundle))
+            else:
+                self.other_certificates = [
+                    load_certificate(other_cert) for other_cert in self.other_certificates
+                ]
+
     def check(self, module, perms_required=True):
         """Ensure the resource is in its desired state."""
 
@@ -340,9 +395,7 @@ class Pkcs(OpenSSLObject):
         self.pkcs12 = crypto.PKCS12()
 
         if self.other_certificates:
-            other_certs = [load_certificate(other_cert) for other_cert
-                           in self.other_certificates]
-            self.pkcs12.set_ca_certificates(other_certs)
+            self.pkcs12.set_ca_certificates(self.other_certificates)
 
         if self.certificate_path:
             self.pkcs12.set_certificate(load_certificate(self.certificate_path))
@@ -402,6 +455,7 @@ def main():
     argument_spec = dict(
         action=dict(type='str', default='export', choices=['export', 'parse']),
         other_certificates=dict(type='list', elements='path', aliases=['ca_certificates']),
+        other_certificates_parse_all=dict(type='bool', default=False),
         certificate_path=dict(type='path'),
         force=dict(type='bool', default=False),
         friendly_name=dict(type='str', aliases=['name']),

plugins/modules/x509_crl.py

@@ -405,7 +405,7 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptograp
     cryptography_get_signature_algorithm_oid_from_crl,
 )
 
-from ansible_collections.community.crypto.plugins.module_utils.crypto.identify import (
+from ansible_collections.community.crypto.plugins.module_utils.crypto.pem import (
     identify_pem_format,
 )
 

plugins/modules/x509_crl_info.py

@@ -160,7 +160,7 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptograp
     cryptography_get_signature_algorithm_oid_from_crl,
 )
 
-from ansible_collections.community.crypto.plugins.module_utils.crypto.identify import (
+from ansible_collections.community.crypto.plugins.module_utils.crypto.pem import (
     identify_pem_format,
 )
 

tests/integration/targets/openssl_pkcs12/tasks/impl.yml

@@ -39,6 +39,12 @@
       pkey: ansible_pkey2.pem
     - name: ansible3
       pkey: ansible_pkey3.pem
+  - name: Generate concatenated PEM file
+    copy:
+      dest: '{{ output_dir }}/ansible23.crt'
+      content: |
+        {{ lookup("file", output_dir ~ "/ansible2.crt") }}
+        {{ lookup("file", output_dir ~ "/ansible3.crt") }}
   - name: Generate PKCS#12 file
     openssl_pkcs12:
       path: '{{ output_dir }}/ansible.p12'
@@ -113,7 +119,7 @@
       friendly_name: abracadabra
       privatekey_path: '{{ output_dir }}/ansible_pkey.pem'
       certificate_path: '{{ output_dir }}/ansible.crt'
-      ca_certificates:
+      other_certificates:
       - '{{ output_dir }}/ansible2.crt'
       - '{{ output_dir }}/ansible3.crt'
       state: present
@@ -124,7 +130,7 @@
       friendly_name: abracadabra
       privatekey_path: '{{ output_dir }}/ansible_pkey.pem'
       certificate_path: '{{ output_dir }}/ansible.crt'
-      ca_certificates:
+      other_certificates:
       - '{{ output_dir }}/ansible2.crt'
       - '{{ output_dir }}/ansible3.crt'
       state: present
@@ -237,7 +243,7 @@
     openssl_pkcs12:
       path: '{{ output_dir }}/ansible_empty.p12'
       friendly_name: abracadabra
-      ca_certificates:
+      other_certificates:
       - '{{ output_dir }}/ansible2.crt'
       - '{{ output_dir }}/ansible3.crt'
       state: present
@@ -246,11 +252,20 @@
     openssl_pkcs12:
       path: '{{ output_dir }}/ansible_empty.p12'
       friendly_name: abracadabra
-      ca_certificates:
-      - '{{ output_dir }}/ansible2.crt'
+      other_certificates:
       - '{{ output_dir }}/ansible3.crt'
+      - '{{ output_dir }}/ansible2.crt'
       state: present
     register: p12_empty_idem
+  - name: Generate 'empty' PKCS#12 file (idempotent, concatenated other certificates)
+    openssl_pkcs12:
+      path: '{{ output_dir }}/ansible_empty.p12'
+      friendly_name: abracadabra
+      other_certificates:
+      - '{{ output_dir }}/ansible23.crt'
+      other_certificates_parse_all: true
+      state: present
+    register: p12_empty_concat_idem
   - name: Generate 'empty' PKCS#12 file (parse)
     openssl_pkcs12:
       src: '{{ output_dir }}/ansible_empty.p12'

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

@@ -64,4 +64,5 @@
     that:
       - p12_empty is changed
       - p12_empty_idem is not changed
+      - p12_empty_concat_idem is not changed
       - "lookup('file', output_dir ~ '/ansible_empty.pem') == lookup('file', output_dir ~ '/ansible3.crt') ~ '\n' ~ lookup('file', output_dir ~ '/ansible2.crt')"