diff --git a/changelogs/fragments/86237-rpm-key-librpm.yml b/changelogs/fragments/86237-rpm-key-librpm.yml new file mode 100644 index 00000000000..7b4c685dbe5 --- /dev/null +++ b/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). diff --git a/lib/ansible/modules/rpm_key.py b/lib/ansible/modules/rpm_key.py index ced79688127..9f78c5028ba 100644 --- a/lib/ansible/modules/rpm_key.py +++ b/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(): diff --git a/test/integration/targets/rpm_key/files/TEST-V6-NON-PQC-KEY.cert b/test/integration/targets/rpm_key/files/TEST-V6-NON-PQC-KEY.cert new file mode 100644 index 00000000000..789abb25aeb --- /dev/null +++ b/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: +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----- diff --git a/test/integration/targets/rpm_key/tasks/main.yaml b/test/integration/targets/rpm_key/tasks/main.yaml index a72e49665fd..d4d2635514a 100644 --- a/test/integration/targets/rpm_key/tasks/main.yaml +++ b/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 }} diff --git a/test/integration/targets/rpm_key/tasks/rpm_key.yaml b/test/integration/targets/rpm_key/tasks/rpm_key.yaml index f83da4fcca1..687c87dac98 100644 --- a/test/integration/targets/rpm_key/tasks/rpm_key.yaml +++ b/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', '>=')) diff --git a/test/integration/targets/rpm_key/vars/main.yml b/test/integration/targets/rpm_key/vars/main.yml index 694a6d0c7f4..f233da93294 100644 --- a/test/integration/targets/rpm_key/vars/main.yml +++ b/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"