certificate_complete_chain: handle duplicate intermediate subjects (#403

) (#405)

* Allow multiple intermediate CAs to have same subject.

* Add tests.

* Fix test name.

* Don't use CN for SAN.

* Make a bit more compatible.

* Include jinja2 compat for CentOS 6.

(cherry picked from commit 11a1454)

Co-authored-by: Felix Fontein <felix@fontein.de>

changelogs/fragments/403-certificate_complete_chain-same-subject.yml

@@ -0,0 +1,2 @@
+bugfixes:
+  - "certificate_complete_chain - allow multiple potential intermediate certificates to have the same subject (https://github.com/ansible-collections/community.crypto/issues/399, https://github.com/ansible-collections/community.crypto/pull/403)."

plugins/modules/certificate_complete_chain.py

@@ -237,14 +237,16 @@ class CertificateSet(object):
     def __init__(self, module):
         self.module = module
         self.certificates = set()
-        self.certificate_by_issuer = dict()
+        self.certificates_by_issuer = dict()
         self.certificate_by_cert = dict()
 
     def _load_file(self, path):
         certs = load_PEM_list(self.module, path, fail_on_error=False)
         for cert in certs:
             self.certificates.add(cert)
-            self.certificate_by_issuer[cert.cert.subject] = cert
+            if cert.cert.subject not in self.certificates_by_issuer:
+                self.certificates_by_issuer[cert.cert.subject] = []
+            self.certificates_by_issuer[cert.cert.subject].append(cert)
             self.certificate_by_cert[cert.cert] = cert
 
     def load(self, path):
@@ -263,8 +265,8 @@ def find_parent(self, cert):
         '''
         Search for the parent (issuer) of a certificate. Return ``None`` if none was found.
         '''
-        potential_parent = self.certificate_by_issuer.get(cert.cert.issuer)
-        if potential_parent is not None:
+        potential_parents = self.certificates_by_issuer.get(cert.cert.issuer, [])
+        for potential_parent in potential_parents:
             if is_parent(self.module, cert, potential_parent):
                 return potential_parent
         return None

tests/integration/targets/certificate_complete_chain/meta/main.yml

@@ -1,3 +1,4 @@
 dependencies:
+  - prepare_jinja2_compat
   - setup_openssl
   - setup_remote_tmp_dir

tests/integration/targets/certificate_complete_chain/tasks/create-single-certificate.yml

@@ -0,0 +1,20 @@
+####################################################################
+# WARNING: These are designed specifically for Ansible tests       #
+# and should not be used as examples of how to write Ansible roles #
+####################################################################
+
+- name: Generate CSR for {{ certificate.name }}
+  openssl_csr:
+    path: '{{ remote_tmp_dir }}/{{ certificate.name }}.csr'
+    privatekey_path: '{{ remote_tmp_dir }}/{{ certificate.name }}.key'
+    subject: '{{ certificate.subject }}'
+    useCommonNameForSAN: false
+
+- name: Generate certificate for {{ certificate.name }}
+  x509_certificate:
+    path: '{{ remote_tmp_dir }}/{{ certificate.name }}.pem'
+    csr_path: '{{ remote_tmp_dir }}/{{ certificate.name }}.csr'
+    privatekey_path: '{{ remote_tmp_dir }}/{{ certificate.name }}.key'
+    provider: '{{ "selfsigned" if certificate.parent is not defined else "ownca" }}'
+    ownca_path: '{{ (remote_tmp_dir ~ "/" ~ certificate.parent ~ ".pem") if certificate.parent is defined else omit }}'
+    ownca_privatekey_path: '{{ (remote_tmp_dir ~ "/" ~ certificate.parent ~ ".key") if certificate.parent is defined else omit }}'

tests/integration/targets/certificate_complete_chain/tasks/create.yml

@@ -0,0 +1,49 @@
+####################################################################
+# WARNING: These are designed specifically for Ansible tests       #
+# and should not be used as examples of how to write Ansible roles #
+####################################################################
+
+- block:
+  - name: Create private keys
+    openssl_privatekey:
+      path: '{{ remote_tmp_dir }}/{{ item.name }}.key'
+      size: '{{ default_rsa_key_size_certifiates }}'
+    loop: '{{ certificates }}'
+
+  - name: Generate certificates
+    include_tasks: create-single-certificate.yml
+    loop: '{{ certificates }}'
+    loop_control:
+      loop_var: certificate
+
+  - name: Read certificates
+    slurp:
+      src: '{{ remote_tmp_dir }}/{{ item.name }}.pem'
+    loop: '{{ certificates }}'
+    register: certificates_read
+
+  - name: Store read certificates
+    set_fact:
+      read_certificates: >-
+        {{ certificates_read.results | map(attribute='content') | map('b64decode')
+           | zip(certificates | map(attribute='name'))
+           | list
+           | items2dict(key_name=1, value_name=0) }}
+
+  vars:
+    certificates:
+      - name: a-root
+        subject:
+          commonName: root common name
+      - name: b-intermediate
+        subject:
+          commonName: intermediate common name
+        parent: a-root
+      - name: c-intermediate
+        subject:
+          commonName: intermediate common name
+        parent: a-root
+      - name: d-leaf
+        subject:
+          commonName: leaf certificate
+        parent: b-intermediate

tests/integration/targets/certificate_complete_chain/tasks/created.yml

@@ -0,0 +1,44 @@
+####################################################################
+# WARNING: These are designed specifically for Ansible tests       #
+# and should not be used as examples of how to write Ansible roles #
+####################################################################
+
+- name: Case A => works
+  certificate_complete_chain:
+    input_chain: "{{ read_certificates['d-leaf'] }}"
+    intermediate_certificates:
+    - '{{ remote_tmp_dir }}/b-intermediate.pem'
+    root_certificates:
+    - '{{ remote_tmp_dir }}/a-root.pem'
+
+- name: Case B => doesn't work, but this is expected
+  failed_when: no
+  register: caseb
+  certificate_complete_chain:
+    input_chain: "{{ read_certificates['d-leaf'] }}"
+    intermediate_certificates:
+    - '{{ remote_tmp_dir }}/c-intermediate.pem'
+    root_certificates:
+    - '{{ remote_tmp_dir }}/a-root.pem'
+
+- name: Assert that case B failed
+  assert:
+    that: "'Cannot complete chain' in caseb.msg"
+
+- name: Case C => works
+  certificate_complete_chain:
+    input_chain: "{{ read_certificates['d-leaf'] }}"
+    intermediate_certificates:
+    - '{{ remote_tmp_dir }}/c-intermediate.pem'
+    - '{{ remote_tmp_dir }}/b-intermediate.pem'
+    root_certificates:
+    - '{{ remote_tmp_dir }}/a-root.pem'
+
+- name: Case D => works as well after PR 403
+  certificate_complete_chain:
+    input_chain: "{{ read_certificates['d-leaf'] }}"
+    intermediate_certificates:
+    - '{{ remote_tmp_dir }}/b-intermediate.pem'
+    - '{{ remote_tmp_dir }}/c-intermediate.pem'
+    root_certificates:
+    - '{{ remote_tmp_dir }}/a-root.pem'

tests/integration/targets/certificate_complete_chain/tasks/existing.yml

@@ -0,0 +1,144 @@
+####################################################################
+# WARNING: These are designed specifically for Ansible tests       #
+# and should not be used as examples of how to write Ansible roles #
+####################################################################
+
+- block:
+  - name: Find root for cert 1 using directory
+    certificate_complete_chain:
+      input_chain: '{{ fullchain | trim }}'
+      root_certificates:
+      - '{{ remote_tmp_dir }}/files/roots/'
+    register: cert1_root
+  - name: Verify root for cert 1
+    assert:
+      that:
+      - cert1_root.complete_chain | join('') == (fullchain ~ root)
+      - cert1_root.root == root
+  vars:
+    fullchain: "{{ lookup('file', 'cert1-fullchain.pem', rstrip=False) }}"
+    root: "{{ lookup('file', 'cert1-root.pem', rstrip=False) }}"
+
+- block:
+  - name: Find rootchain for cert 1 using intermediate and root PEM
+    certificate_complete_chain:
+      input_chain: '{{ cert }}'
+      intermediate_certificates:
+      - '{{ remote_tmp_dir }}/files/cert1-chain.pem'
+      root_certificates:
+      - '{{ remote_tmp_dir }}/files/roots.pem'
+    register: cert1_rootchain
+  - name: Verify rootchain for cert 1
+    assert:
+      that:
+      - cert1_rootchain.complete_chain | join('') == (cert ~ chain ~ root)
+      - cert1_rootchain.chain[:-1] | join('') == chain
+      - cert1_rootchain.root == root
+  vars:
+    cert: "{{ lookup('file', 'cert1.pem', rstrip=False) }}"
+    chain: "{{ lookup('file', 'cert1-chain.pem', rstrip=False) }}"
+    root: "{{ lookup('file', 'cert1-root.pem', rstrip=False) }}"
+
+- block:
+  - name: Find root for cert 2 using directory
+    certificate_complete_chain:
+      input_chain: "{{ fullchain | trim }}"
+      root_certificates:
+      - '{{ remote_tmp_dir }}/files/roots/'
+    register: cert2_root
+  - name: Verify root for cert 2
+    assert:
+      that:
+      - cert2_root.complete_chain | join('') == (fullchain ~ root)
+      - cert2_root.root == root
+  vars:
+    fullchain: "{{ lookup('file', 'cert2-fullchain.pem', rstrip=False) }}"
+    root: "{{ lookup('file', 'cert2-root.pem', rstrip=False) }}"
+
+- block:
+  - name: Find rootchain for cert 2 using intermediate and root PEM
+    certificate_complete_chain:
+      input_chain: '{{ cert }}'
+      intermediate_certificates:
+      - '{{ remote_tmp_dir }}/files/cert2-chain.pem'
+      root_certificates:
+      - '{{ remote_tmp_dir }}/files/roots.pem'
+    register: cert2_rootchain
+  - name: Verify rootchain for cert 2
+    assert:
+      that:
+      - cert2_rootchain.complete_chain | join('') == (cert ~ chain ~ root)
+      - cert2_rootchain.chain[:-1] | join('') == chain
+      - cert2_rootchain.root == root
+  vars:
+    cert: "{{ lookup('file', 'cert2.pem', rstrip=False) }}"
+    chain: "{{ lookup('file', 'cert2-chain.pem', rstrip=False) }}"
+    root: "{{ lookup('file', 'cert2-root.pem', rstrip=False) }}"
+
+- block:
+  - name: Find alternate rootchain for cert 2 using intermediate and root PEM
+    certificate_complete_chain:
+      input_chain: '{{ cert }}'
+      intermediate_certificates:
+      - '{{ remote_tmp_dir }}/files/cert2-altchain.pem'
+      root_certificates:
+      - '{{ remote_tmp_dir }}/files/roots.pem'
+    register: cert2_rootchain_alt
+  - name: Verify rootchain for cert 2
+    assert:
+      that:
+      - cert2_rootchain_alt.complete_chain | join('') == (cert ~ chain ~ root)
+      - cert2_rootchain_alt.chain[:-1] | join('') == chain
+      - cert2_rootchain_alt.root == root
+  vars:
+    cert: "{{ lookup('file', 'cert2.pem', rstrip=False) }}"
+    chain: "{{ lookup('file', 'cert2-altchain.pem', rstrip=False) }}"
+    root: "{{ lookup('file', 'cert2-altroot.pem', rstrip=False) }}"
+
+- block:
+  - name: Find alternate rootchain for cert 2 when complete chain is already presented to the module
+    certificate_complete_chain:
+      input_chain: '{{ cert ~ chain ~ root }}'
+      root_certificates:
+      - '{{ remote_tmp_dir }}/files/roots.pem'
+    register: cert2_complete_chain
+  - name: Verify rootchain for cert 2
+    assert:
+      that:
+      - cert2_complete_chain.complete_chain | join('') == (cert ~ chain ~ root)
+      - cert2_complete_chain.chain == []
+      - cert2_complete_chain.root == root
+  vars:
+    cert: "{{ lookup('file', 'cert2.pem', rstrip=False) }}"
+    chain: "{{ lookup('file', 'cert2-altchain.pem', rstrip=False) }}"
+    root: "{{ lookup('file', 'cert2-altroot.pem', rstrip=False) }}"
+
+- name: Check failure when no intermediate certificate can be found
+  certificate_complete_chain:
+    input_chain: '{{ lookup("file", "cert2.pem", rstrip=True) }}'
+    intermediate_certificates:
+    - '{{ remote_tmp_dir }}/files/cert1-chain.pem'
+    root_certificates:
+    - '{{ remote_tmp_dir }}/files/roots.pem'
+  register: cert2_no_intermediate
+  ignore_errors: true
+- name: Verify failure
+  assert:
+    that:
+    - cert2_no_intermediate is failed
+    - "cert2_no_intermediate.msg.startswith('Cannot complete chain. Stuck at certificate ')"
+
+- name: Check failure when infinite loop is found
+  certificate_complete_chain:
+    input_chain: '{{ lookup("file", "cert2-fullchain.pem", rstrip=True) }}'
+    intermediate_certificates:
+    - '{{ remote_tmp_dir }}/files/roots.pem'
+    root_certificates:
+    - '{{ remote_tmp_dir }}/files/cert1-chain.pem'
+  register: cert2_infinite_loop
+  ignore_errors: true
+- name: Verify failure
+  assert:
+    that:
+    - cert2_infinite_loop is failed
+    - "cert2_infinite_loop.msg == 'Found cycle while building certificate chain'"