Improve PEM identification. (#628)

changelogs/fragments/628-pem-detection.yml

@@ -0,0 +1,2 @@
+bugfixes:
+  - "Fix PEM detection/identification to also accept random other lines before the line starting with ``-----BEGIN`` (https://github.com/ansible-collections/community.crypto/issues/627, https://github.com/ansible-collections/community.crypto/pull/628)."

plugins/module_utils/crypto/pem.py

@@ -14,25 +14,31 @@
 PKCS1_PRIVATEKEY_SUFFIX = ' PRIVATE KEY'
 
 
-def identify_pem_format(content):
+def identify_pem_format(content, encoding='utf-8'):
     '''Given the contents of a binary file, tests whether this could be a PEM file.'''
     try:
-        lines = content.decode('utf-8').splitlines(False)
+        first_pem = extract_first_pem(content.decode(encoding))
+        if first_pem is None:
+            return False
+        lines = first_pem.splitlines(False)
         if lines[0].startswith(PEM_START) and lines[0].endswith(PEM_END) and len(lines[0]) > len(PEM_START) + len(PEM_END):
             return True
     except UnicodeDecodeError:
         pass
     return False
 
 
-def identify_private_key_format(content):
+def identify_private_key_format(content, encoding='utf-8'):
     '''Given the contents of a private key file, identifies its format.'''
     # See https://github.com/openssl/openssl/blob/master/crypto/pem/pem_pkey.c#L40-L85
     # (PEM_read_bio_PrivateKey)
     # and https://github.com/openssl/openssl/blob/master/include/openssl/pem.h#L46-L47
     # (PEM_STRING_PKCS8, PEM_STRING_PKCS8INF)
     try:
-        lines = content.decode('utf-8').splitlines(False)
+        first_pem = extract_first_pem(content.decode(encoding))
+        if first_pem is None:
+            return 'raw'
+        lines = first_pem.splitlines(False)
         if lines[0].startswith(PEM_START) and lines[0].endswith(PEM_END) and len(lines[0]) > len(PEM_START) + len(PEM_END):
             name = lines[0][len(PEM_START):-len(PEM_END)]
             if name in PKCS8_PRIVATEKEY_NAMES:

tests/unit/plugins/module_utils/crypto/test_pem.py

@@ -0,0 +1,67 @@
+# -*- coding: utf-8 -*-
+
+# Copyright (c) 2023, Felix Fontein <felix@fontein.de>
+# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+import pytest
+
+from ansible_collections.community.crypto.plugins.module_utils.crypto.pem import (
+    identify_pem_format,
+    identify_private_key_format,
+    split_pem_list,
+    extract_first_pem,
+)
+
+
+PEM_TEST_CASES = [
+    (b'', [], False, 'raw'),
+    (b'random stuff\nblabla', [], False, 'raw'),
+    (b'-----BEGIN PRIVATE KEY-----', [], False, 'raw'),
+    (
+        b'-----BEGIN PRIVATE KEY-----\n-----END PRIVATE KEY-----',
+        ['-----BEGIN PRIVATE KEY-----\n-----END PRIVATE KEY-----'],
+        True,
+        'pkcs8',
+    ),
+    (
+        b'foo=bar\n# random stuff\n-----BEGIN RSA PRIVATE KEY-----\nblabla\n-----END RSA PRIVATE KEY-----\nmore stuff\n',
+        ['-----BEGIN RSA PRIVATE KEY-----\nblabla\n-----END RSA PRIVATE KEY-----\n'],
+        True,
+        'pkcs1',
+    ),
+    (
+        b'foo=bar\n# random stuff\n-----BEGIN CERTIFICATE-----\nblabla\n-----END CERTIFICATE-----\nmore stuff\n'
+        b'\n-----BEGIN CERTIFICATE-----\nfoobar\n-----END CERTIFICATE-----',
+        [
+            '-----BEGIN CERTIFICATE-----\nblabla\n-----END CERTIFICATE-----\n',
+            '-----BEGIN CERTIFICATE-----\nfoobar\n-----END CERTIFICATE-----',
+        ],
+        True,
+        'unknown-pem',
+    ),
+    (
+        b'-----BEGINCERTIFICATE-----\n-----BEGIN CERTIFICATE-----\n-----BEGINCERTIFICATE-----\n-----END CERTIFICATE-----\n-----BEGINCERTIFICATE-----\n',
+        [
+            '-----BEGIN CERTIFICATE-----\n-----BEGINCERTIFICATE-----\n-----END CERTIFICATE-----\n',
+        ],
+        True,
+        'unknown-pem',
+    ),
+]
+
+
+@pytest.mark.parametrize('data, pems, is_pem, private_key_type', PEM_TEST_CASES)
+def test_pem_handling(data, pems, is_pem, private_key_type):
+    assert identify_pem_format(data) == is_pem
+    assert identify_private_key_format(data) == private_key_type
+    try:
+        text = data.decode('utf-8')
+        assert split_pem_list(text) == pems
+        first_pem = pems[0] if pems else None
+        assert extract_first_pem(text) == first_pem
+    except UnicodeDecodeError:
+        pass