Replace gpg in rpm_key with librpm (#86237) (#86615)

* Replace gpg in rpm_key with librpm

(cherry picked from commit 2341354ffa)

changelogs/fragments/86237-rpm-key-librpm.yml

@@ -0,0 +1,2 @@
+bugfixes:
+  - rpm_key - Use librpm library API instead of gpg utility to support version 6 PGP keys (https://github.com/ansible/ansible/issues/86157).

lib/ansible/modules/rpm_key.py

@@ -22,6 +22,7 @@ options:
       description:
         - Key that will be modified. Can be a url, a file on the managed node, or a keyid if the key
           already exists in the database.
+        - This can also be the fingerprint when attempting to delete an already installed key.
       type: str
       required: true
     state:
@@ -85,17 +86,325 @@ EXAMPLES = """
 
 RETURN = r"""#"""
 
+import ctypes
+import ctypes.util
+import hashlib
 import re
 import os.path
 import tempfile
+import typing as _t
 
 # import module snippets
 from ansible.module_utils.basic import AnsibleModule
 from ansible.module_utils.urls import fetch_url
+from ansible.module_utils.compat.version import LooseVersion
 from ansible.module_utils.common.text.converters import to_native
 
+# Type alias for ctypes pointer to uint8 array (packet data)
+# Using Any here because ctypes._Pointer is private, but documenting the actual type
+PktPointer = _t.Any  # Actually: ctypes.POINTER(ctypes.c_uint8)
 
-def is_pubkey(string):
+
+class LibRPM:
+    """
+    Wrapper for librpm PGP key functions.
+
+    The APIs in librpm vary across different versions. Since this module must work on a variety of
+    systems, we are extremely limited in the API calls that we can guarantee will be available.
+    """
+
+    # Constants
+    PGPTAG_PUBLIC_KEY = 6
+    PGPTAG_PUBLIC_SUBKEY = 14
+
+    def __init__(self) -> None:
+        # Load the librpm library
+        if not (lib_path := ctypes.util.find_library('rpm')):
+            raise ImportError("Error: Could not find librpm library")
+
+        self._lib = ctypes.CDLL(lib_path)
+        self._libc = ctypes.CDLL(None)
+
+        # void free(void *ptr)
+        self._libc.free.argtypes = [ctypes.c_void_p]
+        self._libc.free.restype = None
+
+        # pgpArmor pgpParsePkts(const char *armor, uint8_t **pkt, size_t *pktlen)
+        self._lib.pgpParsePkts.argtypes = [
+            ctypes.c_char_p,
+            ctypes.POINTER(ctypes.POINTER(ctypes.c_uint8)),
+            ctypes.POINTER(ctypes.c_size_t)
+        ]
+        self._lib.pgpParsePkts.restype = ctypes.c_int
+
+        # Identify the version of the RPM library
+        _lib_rpmversion = ctypes.c_char_p.in_dll(self._lib, "RPMVERSION")
+        self._rpmversion = _lib_rpmversion.value.decode()
+
+    @property
+    def using_librpm6(self) -> bool:
+        """
+        Check if the librpm version in use is at least version 6.0.0.
+
+        RPM version 6.0.0 and higher uses fingerprints instead of short key ID everywhere. This changes
+        how we must approach certain operations, such as key deletion from the rpmdb.
+        """
+        if LooseVersion(self._rpmversion) >= LooseVersion('6.0.0'):
+            return True
+        return False
+
+    def _parse_armor(self, armor: str) -> tuple[PktPointer | None, int]:
+        """
+        Parse ASCII armored PGP data using pgpParsePkts().
+        Returns (pkt, pktlen) tuple or (None, 0) on error.
+        """
+        pkt = ctypes.POINTER(ctypes.c_uint8)()
+        pktlen = ctypes.c_size_t()
+
+        armor_bytes = armor.encode()
+        result = self._lib.pgpParsePkts(armor_bytes, ctypes.byref(pkt), ctypes.byref(pktlen))
+
+        if result < 0 or not pkt:
+            return None, 0
+
+        return pkt, pktlen.value
+
+    def _parse_packet_header(self, pkt: PktPointer, offset: int, pktlen: int) -> tuple[int | None, int, int]:
+        """
+        Parse a PGP packet header to get tag and packet length.
+        Returns (tag, body_length, header_length) or (None, 0, 0) on error.
+
+        Per RFC 9580 - Section 4.2: Packet Headers
+        https://www.rfc-editor.org/rfc/rfc9580.html#name-packet-headers
+        """
+        if offset >= pktlen:
+            return None, 0, 0
+
+        tag_byte = pkt[offset]
+
+        # Check if it's a new format packet (bit 6 set)
+        if tag_byte & 0x40:
+            # New format
+            tag = tag_byte & 0x3f  # bits 0-5 are packet type ID
+            offset += 1
+
+            if offset >= pktlen:
+                return None, 0, 0
+
+            first_len_byte = pkt[offset]
+
+            if first_len_byte < 192:
+                # One-octet length
+                return tag, first_len_byte, 2
+            elif first_len_byte < 224:
+                # Two-octet length
+                if offset + 1 >= pktlen:
+                    return None, 0, 0
+                length = ((first_len_byte - 192) << 8) + pkt[offset + 1] + 192
+                return tag, length, 3
+            elif first_len_byte == 255:
+                # Five-octet length
+                if offset + 4 >= pktlen:
+                    return None, 0, 0
+                length = (pkt[offset + 1] << 24) | (pkt[offset + 2] << 16) | \
+                         (pkt[offset + 3] << 8) | pkt[offset + 4]
+                return tag, length, 6
+            else:
+                # Partial body length (not supported here)
+                return None, 0, 0
+        else:
+            # Old format
+            tag = (tag_byte >> 2) & 0x0f
+            length_type = tag_byte & 0x03
+
+            if length_type == 0:
+                # One-octet length
+                if offset + 1 >= pktlen:
+                    return None, 0, 0
+                return tag, pkt[offset + 1], 2
+            elif length_type == 1:
+                # Two-octet length
+                if offset + 2 >= pktlen:
+                    return None, 0, 0
+                length = (pkt[offset + 1] << 8) | pkt[offset + 2]
+                return tag, length, 3
+            elif length_type == 2:
+                # Four-octet length
+                if offset + 4 >= pktlen:
+                    return None, 0, 0
+                length = (pkt[offset + 1] << 24) | (pkt[offset + 2] << 16) | \
+                         (pkt[offset + 3] << 8) | pkt[offset + 4]
+                return tag, length, 5
+            else:
+                # Indeterminate length (not supported)
+                return None, 0, 0
+
+    def _find_key_packets(self, pkt: PktPointer, pktlen: int) -> list[tuple[int, int]]:
+        """
+        Walk the packet stream and find all PGPTAG_PUBLIC_KEY and PGPTAG_PUBLIC_SUBKEY packets.
+        Returns list of (offset, total_packet_length) tuples.
+        """
+        key_packets: list[tuple[int, int]] = []
+        offset = 0
+
+        while offset < pktlen:
+            tag, body_len, header_len = self._parse_packet_header(pkt, offset, pktlen)
+
+            if tag is None:
+                break
+
+            if tag in (self.PGPTAG_PUBLIC_KEY, self.PGPTAG_PUBLIC_SUBKEY):
+                # Found a key packet
+                total_len = header_len + body_len
+                key_packets.append((offset, total_len))
+
+            # Move to next packet
+            offset += header_len + body_len
+
+        return key_packets
+
+    def _get_key_version(self, pkt: PktPointer, offset: int, pktlen: int) -> int | None:
+        """
+        Get the version byte from a key packet.
+        Returns version number (4 or 6) or None on error.
+        """
+        tag, dummy, header_len = self._parse_packet_header(pkt, offset, pktlen)
+        if tag is None:
+            return None
+
+        # Extract packet body (skip the packet header)
+        body_offset = offset + header_len
+        if body_offset >= pktlen:
+            return None
+
+        # First byte of body is the version
+        return pkt[body_offset]
+
+    def _compute_v4_fingerprint(self, pkt: PktPointer, offset: int, pktlen: int) -> str | None:
+        """
+        Compute V4 fingerprint from packet data.
+        For V4 keys, fingerprint = SHA-1(0x99 || 2-byte-length || packet_body)
+        Per RFC 4880 Section 12.2
+        """
+        tag, body_len, header_len = self._parse_packet_header(pkt, offset, pktlen)
+
+        if tag is None:
+            return None
+
+        # Extract packet body (skip the packet header)
+        body_offset = offset + header_len
+        if body_offset + body_len > pktlen:
+            return None
+
+        # Check if it's a V4 key (first byte of body should be 0x04)
+        if pkt[body_offset] != 0x04:
+            return None
+
+        # Build the data for fingerprint: 0x99 || 2-byte length || body
+        fp_data = bytearray()
+        fp_data.append(0x99)  # V4 public key packet tag
+        fp_data.append((body_len >> 8) & 0xFF)  # Length high byte
+        fp_data.append(body_len & 0xFF)  # Length low byte
+
+        # Append the packet body
+        for i in range(body_len):
+            fp_data.append(pkt[body_offset + i])
+
+        # Compute SHA-1 hash
+        return hashlib.sha1(fp_data).hexdigest().upper()
+
+    def _compute_v6_fingerprint(self, pkt: PktPointer, offset: int, pktlen: int) -> str | None:
+        """
+        Compute V6 fingerprint from packet data.
+        For V6 keys, fingerprint = SHA-256(0x9B || 4-byte-length || packet_body)
+        Per RFC 9580 Section 5.5.4
+        """
+        tag, body_len, header_len = self._parse_packet_header(pkt, offset, pktlen)
+
+        if tag is None:
+            return None
+
+        # Extract packet body (skip the packet header)
+        body_offset = offset + header_len
+        if body_offset + body_len > pktlen:
+            return None
+
+        # Check if it's a V6 key (first byte of body should be 0x06)
+        if pkt[body_offset] != 0x06:
+            return None
+
+        # Build the data for fingerprint: 0x9B || 4-byte length || body
+        fp_data = bytearray()
+        fp_data.append(0x9B)  # V6 public key packet tag
+        fp_data.append((body_len >> 24) & 0xFF)  # Length byte 1 (MSB)
+        fp_data.append((body_len >> 16) & 0xFF)  # Length byte 2
+        fp_data.append((body_len >> 8) & 0xFF)   # Length byte 3
+        fp_data.append(body_len & 0xFF)          # Length byte 4 (LSB)
+
+        # Append the packet body
+        for i in range(body_len):
+            fp_data.append(pkt[body_offset + i])
+
+        # Compute SHA-256 hash
+        return hashlib.sha256(fp_data).hexdigest().upper()
+
+    def identify_keys(self, armor: str) -> list[dict[str, str]]:
+        """Return a list of dicts with key ID (8-byte) and fingerprint for the primary key and each subkey"""
+        key_info: list[dict[str, str]] = []
+
+        pkt, pktlen = self._parse_armor(armor)
+        if not pkt:
+            raise Exception("Unable to parse PGP key armor")
+
+        # Find all key packets in the stream and compute their fingerprints.
+        key_packets = self._find_key_packets(pkt, pktlen)
+
+        for offset, dummy in key_packets:
+            # Detect key version
+            version = self._get_key_version(pkt, offset, pktlen)
+
+            if version == 0x04:
+                # V4 key
+                computed_fp = self._compute_v4_fingerprint(pkt, offset, pktlen)
+                if computed_fp:
+                    # V4: Key ID is the last 8 bytes (16 hex chars) of the fingerprint
+                    keyid_from_fp = computed_fp[-16:]
+                    key_info.append({'keyid': keyid_from_fp, 'fingerprint': computed_fp})
+            elif version == 0x06:
+                # V6 key
+                computed_fp = self._compute_v6_fingerprint(pkt, offset, pktlen)
+                if computed_fp:
+                    # V6: Key ID is the first 8 bytes (16 hex chars) of the fingerprint
+                    keyid_from_fp = computed_fp[:16]
+                    key_info.append({'keyid': keyid_from_fp, 'fingerprint': computed_fp})
+            else:
+                raise Exception(f"Unhandled key version {version:#04x}")
+
+        self._libc.free(pkt)
+
+        return key_info
+
+    def get_key_ids_from_armor(self, armor: str) -> list[str]:
+        """
+        Get the key IDs from the primary PGP key, and all subkeys of that key, from the ASCII armored key.
+
+        'armor' is expected to be a single ASCII armored PGP key (v4 or v6). The primary key should be the
+        first item in the results, followed by its subkeys. Returned key IDs are 8-byte (16 hex characters)
+        in length. This must be accounted for if comparing against the short key ID (4-bytes).
+        """
+        return [key['keyid'] for key in self.identify_keys(armor)]
+
+    def get_fingerprints_from_armor(self, armor: str) -> list[str]:
+        """
+        Get the fingerprints from the primary PGP key, and all subkeys of that key, from the ASCII armored key.
+
+        'armor' is expected to be a single ASCII armored PGP key (v4 or v6). The primary key should be the
+        first item in the results, followed by its subkeys.
+        """
+        return [key['fingerprint'] for key in self.identify_keys(armor)]
+
+
+def is_pubkey(string: str) -> bool:
     """Verifies if string is a pubkey"""
     pgp_regex = ".*?(-----BEGIN PGP PUBLIC KEY BLOCK-----.*?-----END PGP PUBLIC KEY BLOCK-----).*"
     return bool(re.match(pgp_regex, to_native(string, errors='surrogate_or_strict'), re.DOTALL))
@@ -103,13 +412,14 @@ def is_pubkey(string):
 
 class RpmKey(object):
 
-    def __init__(self, module):
+    def __init__(self, module: AnsibleModule) -> None:
         # If the key is a url, we need to check if it's present to be idempotent,
         # to do that, we need to check the keyid, which we can get from the armor.
         keyfile = None
         should_cleanup_keyfile = False
         self.module = module
         self.rpm = self.module.get_bin_path('rpm', True)
+        self.rpmkeys = self.module.get_bin_path('rpmkeys', True)
         state = module.params['state']
         key = module.params['key']
         fingerprint = module.params['fingerprint']
@@ -120,9 +430,7 @@ class RpmKey(object):
                 fingerprint = [fingerprint]
             fingerprints = set(f.replace(' ', '').upper() for f in fingerprint)
 
-        self.gpg = self.module.get_bin_path('gpg')
-        if not self.gpg:
-            self.gpg = self.module.get_bin_path('gpg2', required=True)
+        self.librpm = LibRPM()
 
         if '://' in key:
             keyfile = self.fetch_key(key)
@@ -137,6 +445,8 @@ class RpmKey(object):
             self.module.fail_json(msg="Not a valid key %s" % key)
         keyid = self.normalize_keyid(keyid)
 
+        self.installed_keys = self.get_installed_keys()
+
         if state == 'present':
             if self.is_key_imported(keyid):
                 module.exit_json(changed=False)
@@ -161,7 +471,7 @@ class RpmKey(object):
             else:
                 module.exit_json(changed=False)
 
-    def fetch_key(self, url):
+    def fetch_key(self, url: str) -> str:
         """Downloads a key from url, returns a valid path to a gpg key"""
         rsp, info = fetch_url(self.module, url)
         if info['status'] != 200:
@@ -176,7 +486,7 @@ class RpmKey(object):
             tmpfile.write(key)
         return tmpname
 
-    def normalize_keyid(self, keyid):
+    def normalize_keyid(self, keyid: str) -> str:
         """Ensure a keyid doesn't have a leading 0x, has leading or trailing whitespace, and make sure is uppercase"""
         ret = keyid.strip().upper()
         if ret.startswith('0x'):
@@ -186,71 +496,134 @@ class RpmKey(object):
         else:
             return ret
 
-    def getkeyid(self, keyfile):
-        stdout, stderr = self.execute_command([self.gpg, '--no-tty', '--batch', '--with-colons', '--fixed-list-mode', keyfile])
-        for line in stdout.splitlines():
-            line = line.strip()
-            if line.startswith('pub:'):
-                return line.split(':')[4]
+    def getkeyid(self, keyfile: str) -> str:
+        with open(keyfile, "r") as key_fd:
+            key_ids = self.librpm.get_key_ids_from_armor(key_fd.read())
+        if not key_ids:
+            self.module.fail_json(msg="Failed to get keyid")
+        return key_ids[0]
 
-        self.module.fail_json(msg="Unexpected gpg output")
+    def getfingerprints(self, keyfile: str) -> frozenset[str]:
+        with open(keyfile, "r") as key_fd:
+            fingerprints = self.librpm.get_fingerprints_from_armor(key_fd.read())
+        if not fingerprints:
+            self.module.fail_json(msg="Failed to get fingerprint")
+        return frozenset(fingerprints)
 
-    def getfingerprints(self, keyfile):
-        stdout, stderr = self.execute_command([
-            self.gpg, '--no-tty', '--batch', '--with-colons',
-            '--fixed-list-mode', '--import', '--import-options', 'show-only',
-            '--dry-run', keyfile
-        ])
+    def is_keyid(self, keystr: str) -> re.Match[str] | None:
+        """
+        Verifies if a key, as provided by the user, is a key ID.
 
-        fingerprints = set()
-
-        for line in stdout.splitlines():
-            line = line.strip()
-            if line.startswith('fpr:'):
-                # As mentioned here,
-                #
-                # https://git.gnupg.org/cgi-bin/gitweb.cgi?p=gnupg.git;a=blob_plain;f=doc/DETAILS
-                #
-                # The description of the `fpr` field says
-                #
-                # "fpr :: Fingerprint (fingerprint is in field 10)"
-                #
-                fingerprints.add(line.split(':')[9])
-
-        if fingerprints:
-            return fingerprints
-
-        self.module.fail_json(msg="Unexpected gpg output")
-
-    def is_keyid(self, keystr):
-        """Verifies if a key, as provided by the user is a keyid"""
+        Note that this allows the short form of the key ID (4-bytes, or 8 hex characters), used in older
+        versions of RPM, while a full key ID is 8-bytes, or 16 hex characters.
+        """
+        keystr = keystr.replace(' ', '')
         return re.match('(0x)?[0-9a-f]{8}', keystr, flags=re.IGNORECASE)
 
-    def execute_command(self, cmd):
+    def execute_command(self, cmd: str | list[str]) -> tuple[str, str]:
         rc, stdout, stderr = self.module.run_command(cmd, use_unsafe_shell=True)
         if rc != 0:
             self.module.fail_json(msg=stderr)
         return stdout, stderr
 
-    def is_key_imported(self, keyid):
-        cmd = self.rpm + ' -q  gpg-pubkey'
+    def get_installed_keys(self) -> list[dict[str, str]]:
+        """
+        Get the key ID and fingerprint for every key installed on the system.
+
+        This will grab the armor string for every key reported from `rpm -q gpg-pubkey` and parse
+        it to obtain the key ID and fingerprint, including subkeys.
+        """
+        installed_keys = []
+
+        cmd = self.rpm + ' -q gpg-pubkey'
         rc, stdout, stderr = self.module.run_command(cmd)
         if rc != 0:  # No key is installed on system
-            return False
-        cmd += ' --qf "%{description}" | ' + self.gpg + ' --no-tty --batch --with-colons --fixed-list-mode -'
+            return []
+        cmd += ' --qf "%{description}"'
         stdout, stderr = self.execute_command(cmd)
+
+        # Split the content into individual key blocks
+        key_blocks = []
+        current_block = []
+        in_key_block = False
+
         for line in stdout.splitlines():
-            if keyid in line.split(':')[4]:
-                return True
+            if line.strip() == '-----BEGIN PGP PUBLIC KEY BLOCK-----':
+                in_key_block = True
+                current_block = [line]
+            elif line.strip() == '-----END PGP PUBLIC KEY BLOCK-----':
+                current_block.append(line)
+                key_blocks.append('\n'.join(current_block))
+                current_block = []
+                in_key_block = False
+            elif in_key_block:
+                current_block.append(line)
+
+        for armor_string in key_blocks:
+            installed_keys.extend(self.librpm.identify_keys(armor_string))
+
+        return installed_keys
+
+    def is_key_imported(self, keyid: str) -> bool:
+        """Check the supplied key ID value against the currently installed keys."""
+        keyid_len = len(keyid)
+        if keyid in [k['keyid'][-keyid_len:] for k in self.installed_keys]:
+            return True
+
+        # Allow the user supplied key to also be a fingerprint
+        if keyid in [f['fingerprint'] for f in self.installed_keys]:
+            return True
+
         return False
 
-    def import_key(self, keyfile):
+    def import_key(self, keyfile: str) -> None:
         if not self.module.check_mode:
             self.execute_command([self.rpm, '--import', keyfile])
 
-    def drop_key(self, keyid):
+    def _drop_key_rpm6(self, keyid: str) -> None:
+        """
+        Remove the key with the given key ID from the keyring using RPM 6+ method.
+
+        RPM version 6+ uses fingerprints and the 'rpmkeys --delete' command.
+        """
+        fingerprints = []
+        keyid_len = len(keyid)
+
+        for installed in self.installed_keys:
+            if keyid == installed['keyid'][-keyid_len:]:
+                fingerprints.append(installed['fingerprint'])
+            # We allow the user supplied 'key' to also be the full fingerprint.
+            elif keyid == installed['fingerprint']:
+                fingerprints.append(installed['fingerprint'])
+
+        if not fingerprints:
+            self.module.fail_json(msg=f"Supplied key ID {keyid} is not installed.")
+        elif len(fingerprints) == 1:
+            self.execute_command([self.rpmkeys, '--delete', fingerprints[0]])
+        else:
+            self.module.fail_json(msg=f"Supplied key ID {keyid} matches more than one fingerprint. Try using the fingerprint instead.")
+
+    def _drop_key_rpm4(self, keyid: str) -> None:
+        """
+        Remove the key with the given key ID from the keyring using RPM 4 method.
+
+        Older RPM versions use short form key ID (4-bytes) and 'rpm --erase' command.
+        """
+        # If keyid is actually a fingerprint, we need to get the associated key ID and use it.
+        for installed in self.installed_keys:
+            if keyid == installed['fingerprint']:
+                keyid = installed['keyid']
+                break
+
+        self.execute_command([self.rpm, '--erase', '--allmatches', "gpg-pubkey-%s" % keyid[-8:].lower()])
+
+    def drop_key(self, keyid: str) -> None:
+        """Remove the key with the given key ID from the keyring."""
         if not self.module.check_mode:
-            self.execute_command([self.rpm, '--erase', '--allmatches', "gpg-pubkey-%s" % keyid[-8:].lower()])
+            if self.librpm.using_librpm6:
+                self._drop_key_rpm6(keyid)
+            else:
+                self._drop_key_rpm4(keyid)
 
 
 def main():

test/integration/targets/rpm_key/files/TEST-V6-NON-PQC-KEY.cert

@@ -0,0 +1,27 @@
+-----BEGIN PGP PUBLIC KEY BLOCK-----
+Comment: 451A 6DBE 3B0F 51EA FBEF 3585 D53F 0277  62A6 C0DE A0A8 4EE5 7DB3 87B0 1651 4BBC
+Comment: <info@redhat.com>
+Comment: Ansible Example
+
+xioGaWachxsAAAAgPh2gtnpD585bPVCmyST0RA3XAn6rc0oP4lTMSZ5GjQXCrAYf
+GwoAAAA9BYJpZpyHBYkFpI+9AwsJBwMVCggCmwECHgkiIQZFGm2+Ow9R6vvvNYXV
+PwJ3YqbA3qCoTuV9s4ewFlFLvAAAAAADziCYy5PwGbFr0+ZCRuWbw54j5otd0SrF
+h5zJih5TVryL1PNLMtQZt1XtLICOSBrtgbCMIhntaxW4O6ZHmv2p4BQIjt7kNWZA
+/AJcfn57iGdoFrZD8pMTWKYR66R+FForhwXNETxpbmZvQHJlZGhhdC5jb20+wqwG
+ExsKAAAAPQWCaWachwWJBaSPvQMLCQcDFQoIApsBAh4JIiEGRRptvjsPUer77zWF
+1T8Cd2KmwN6gqE7lfbOHsBZRS7wAAAAA+NogVNCvF3Y5SSpUF4cyH9NGRE9Gjdml
+KVefqvmh4T0g0JyM/FdRmxgkzRTaUdqjRmVMLg53ScY0AwaGjuhiEjVBe2GGQrtu
+aiupfyGvcAnUxNwmhoHvh0BeOu3vQwlCk14IzQ9BbnNpYmxlIEV4YW1wbGXCrwYT
+GwoAAABABYJpZpyHBYkFpI+9AwsJBwMVCggCmQECmwECHgkiIQZFGm2+Ow9R6vvv
+NYXVPwJ3YqbA3qCoTuV9s4ewFlFLvAAAAAAowSC8pmjMJom99WiMkM19I0qzNbFm
+YwCTpfirAk4v89ggIUw6eTF1oKlk6WTUeVeK2nBGHd2cZlV6NtKAlCllPqLq0rlp
+4ZxVo6cfUKxNjdnbTrE4Seamc0hYtLG49TEkaATOKgZpZpyHGwAAACAGbi2A12S/
+VoxdCm9I5VPSmsUHRO0DznHeB/VEG8NKCsLAewYYGwoAAADMBYJpZpyHBYkFpI+9
+ApsCmaAGGRsKAAAAKQWCaWachyIhBgiuhwe1s82YIvPJXs0dT6zFgNVk7juLtL8s
+Zx6H2pNDAAAAAILuILtOYzI43a9SImbCmkWStIxesTFKhFBMvaMPJOrErcWGjiga
+Ikq7NVpmSfyVs5s0DJrr7c3TJ0+Fneku/PCnppCh8140qfhlKVx+ePwpv41jQcK9
+uj8jTdK3I8Pn6yRGAyIhBkUabb47D1Hq++81hdU/AndipsDeoKhO5X2zh7AWUUu8
+AAAAAIufIPo2sGq+KZSbp1QV4/A65DtPpSK0oSwML/B4OrPQLQ8+9iuZ2CB5iHJ6
+4xQ4OvyKGslJU6fY0A23WgFguZdIvfIAvRLo2WqFkOhyBzJc55u5IU5ukGWn8AY0
+bgG7sR6hBA==
+-----END PGP PUBLIC KEY BLOCK-----

test/integration/targets/rpm_key/tasks/main.yaml

@@ -1,7 +1,3 @@
- - name: Skip RHEL 10.1 until rpm_key has been updated
-   meta: end_play
-   when: ansible_distribution == "RedHat" and ansible_distribution_version == "10.1"
-
  - when: ansible_os_family == "RedHat"
    block:
 
@@ -23,8 +19,10 @@
 
    always:
 
+   # This will fail if the tests leave no keys in the key store
    - name: Remove all GPG keys from key ring
      shell: rpm -q gpg-pubkey | xargs rpm -e
+     ignore_errors: yes
 
    - name: Restore the previously installed GPG keys
      command: rpm --import {{ (remote_tmp_dir + '/pubkeys') | quote }}

test/integration/targets/rpm_key/tasks/rpm_key.yaml

@@ -123,6 +123,16 @@
   assert:
     that: key_id_idempotence is not changed
 
+- name: Add key to test deleting by fingerprint
+  rpm_key:
+    state: present
+    key: "{{ test_key_url }}"
+
+- name: Remove key by using fingerprint
+  rpm_key:
+    state: absent
+    key: "{{ primary_fingerprint }}"
+
 - name: Add very first key on system again
   rpm_key:
     state: present
@@ -227,3 +237,68 @@
     that:
       - result is success
       - result is not changed
+
+# Test PGPv6 keys
+- name: Test PGPv6 Keys
+  block:
+
+    # Reset to test PGPv6 keys validation
+    - name: Remove all keys from key ring
+      shell: rpm -q gpg-pubkey | xargs rpm -e
+
+    - name: Copy v6 key to remote
+      copy:
+        src: "files/{{ test_v6_key_file }}"
+        dest: "{{ remote_tmp_dir }}/{{ test_v6_key_file }}"
+        mode: "0644"
+
+    - name: Import v6 key
+      rpm_key:
+        state: present
+        key: "{{ remote_tmp_dir }}/{{ test_v6_key_file }}"
+        fingerprint: "{{ invalid_fingerprint }}"
+      register: result
+      failed_when: result is success
+
+    - name: Verify invalid fingerprint failure
+      assert:
+        that:
+           - result is success
+           - result is not changed
+           - result.msg is contains 'does not match any key fingerprints'
+
+    - name: Import v6 key
+      rpm_key:
+        state: present
+        key: "{{ remote_tmp_dir }}/{{ test_v6_key_file }}"
+        fingerprint: "{{ test_v6_key_fingerprint }}"
+
+    - name: Import v6 key (idempotent)
+      rpm_key:
+        state: present
+        key: "{{ remote_tmp_dir }}/{{ test_v6_key_file }}"
+        fingerprint: "{{ test_v6_key_fingerprint }}"
+      register: key_idempotence
+
+    - name: Verify idempotence
+      assert:
+        that: key_idempotence is not changed
+
+    - name: Delete v6 key
+      rpm_key:
+        state: absent
+        key: "{{ test_v6_key_keyid }}"
+
+    - name: Import v6 key again
+      rpm_key:
+        state: present
+        key: "{{ remote_tmp_dir }}/{{ test_v6_key_file }}"
+        fingerprint: "{{ test_v6_key_fingerprint }}"
+
+    - name: Delete v6 key by fingerprint
+      rpm_key:
+        state: absent
+        key: "{{ test_v6_key_fingerprint }}"
+
+  when: (ansible_facts['distribution'] == "RedHat" and ansible_facts['distribution_version'] is version('10.1', '>='))
+        or (ansible_facts['distribution'] == "Fedora" and ansible_facts['distribution_version'] is version('43', '>='))

test/integration/targets/rpm_key/vars/main.yml

@@ -6,3 +6,6 @@ sub_key_url: https://ci-files.testing.ansible.com/test/integration/targets/rpm_k
 invalid_fingerprint: 1111 1111 1111 1111 1111 1111 1111 1111 1111 1111
 primary_fingerprint: 66D1 5FDD 8728 7219 C8E1 5478 D200 CD70 2853 E6D0
 sub_key_fingerprint: E617 DCD4 065C 2AFC 0B2C F7A7 BA8B C08C 0F69 1F94
+test_v6_key_file: "TEST-V6-NON-PQC-KEY.cert"
+test_v6_key_fingerprint: "451A6DBE3B0F51EAFBEF3585D53F027762A6C0DEA0A84EE57DB387B016514BBC"
+test_v6_key_keyid: "451A6DBE3B0F51EA"