apt_repository add falalback to deprecated apt-key usage (#77340)

New path uses gpg directly and the stated directories and keystores to use by debian docs

Co-authored-by: Matt Clay <matt@mystile.com>

changelogs/fragments/apt_repository_sans_apt_key.yml

@@ -0,0 +1,2 @@
+minor_changes:
+  - apt_repository remove dependency on apt-key and use gpg + /usr/share/keyrings directly instead

lib/ansible/modules/apt_repository.py

@@ -135,13 +135,13 @@ EXAMPLES = '''
 
 RETURN = '''#'''
 
+import copy
 import glob
 import json
 import os
 import re
 import sys
 import tempfile
-import copy
 import random
 import time
 
@@ -314,7 +314,11 @@ class SourcesList(object):
                 except OSError as ex:
                     if not os.path.isdir(d):
                         self.module.fail_json("Failed to create directory %s: %s" % (d, to_native(ex)))
-                fd, tmp_path = tempfile.mkstemp(prefix=".%s-" % fn, dir=d)
+
+                try:
+                    fd, tmp_path = tempfile.mkstemp(prefix=".%s-" % fn, dir=d)
+                except (OSError, IOError) as e:
+                    self.module.fail_json(msg='Unable to create temp file at "%s" for apt source: %s' % (d, to_native(e)))
 
                 f = os.fdopen(fd, 'w')
                 for n, valid, enabled, source, comment in sources:
@@ -377,6 +381,7 @@ class SourcesList(object):
     def _add_valid_source(self, source_new, comment_new, file):
         # We'll try to reuse disabled source if we have it.
         # If we have more than one entry, we will enable them all - no advanced logic, remember.
+        self.module.log('ading source file: %s | %s | %s' % (source_new, comment_new, file))
         found = False
         for filename, n, enabled, source, comment in self:
             if source == source_new:
@@ -417,17 +422,18 @@ class UbuntuSourcesList(SourcesList):
 
     LP_API = 'https://launchpad.net/api/1.0/~%s/+archive/%s'
 
-    def __init__(self, module, add_ppa_signing_keys_callback=None):
+    def __init__(self, module):
         self.module = module
-        self.add_ppa_signing_keys_callback = add_ppa_signing_keys_callback
         self.codename = module.params['codename'] or distro.codename
         super(UbuntuSourcesList, self).__init__(module)
 
+        self.apt_key_bin = self.module.get_bin_path('apt-key', required=False)
+        self.gpg_bin = self.module.get_bin_path('gpg', required=False)
+        if not self.apt_key_bin and not self.gpg_bin:
+            self.module.fail_json(msg='Either apt-key or gpg binary is required, but neither could be found')
+
     def __deepcopy__(self, memo=None):
-        return UbuntuSourcesList(
+        return UbuntuSourcesList(self.module)
-            self.module,
-            add_ppa_signing_keys_callback=self.add_ppa_signing_keys_callback
-        )
 
     def _get_ppa_info(self, owner_name, ppa_name):
         lp_api = self.LP_API % (owner_name, ppa_name)
@@ -450,9 +456,39 @@ class UbuntuSourcesList(SourcesList):
         return line, ppa_owner, ppa_name
 
     def _key_already_exists(self, key_fingerprint):
-        rc, out, err = self.module.run_command('apt-key export %s' % key_fingerprint, check_rc=True)
-        return len(err) == 0
 
+        if self.apt_key_bin:
+            rc, out, err = self.module.run_command([self.apt_key_bin, 'export', key_fingerprint], check_rc=True)
+            found = len(err) == 0
+        else:
+            found = self._gpg_key_exists(key_fingerprint)
+
+        return found
+
+    def _gpg_key_exists(self, key_fingerprint):
+
+        found = False
+        keyfiles = ['/etc/apt/trusted.gpg']  # main gpg repo for apt
+        for other_dir in ('/etc/apt/trusted.gpg.d', '/usr/share/keyrings'):
+            # add other known sources of gpg sigs for apt, skip hidden files
+            keyfiles.extend([os.path.join(other_dir, x) for x in os.listdir(other_dir) if not x.startswith('.')])
+
+        for key_file in keyfiles:
+
+            if os.path.exists(key_file):
+                try:
+                    rc, out, err = self.module.run_command([self.gpg_bin, '--list-packets', key_file])
+                except (IOError, OSError) as e:
+                    self.debug("Could check key against file %s: %s" % (key_file, to_native(e)))
+                    continue
+
+                if key_fingerprint in out:
+                    found = True
+                    break
+
+        return found
+
+    # https://www.linuxuprising.com/2021/01/apt-key-is-deprecated-how-to-add.html
     def add_source(self, line, comment='', file=None):
         if line.startswith('ppa:'):
             source, ppa_owner, ppa_name = self._expand_ppa(line)
@@ -461,16 +497,39 @@ class UbuntuSourcesList(SourcesList):
                 # repository already exists
                 return
 
-            if self.add_ppa_signing_keys_callback is not None:
+            info = self._get_ppa_info(ppa_owner, ppa_name)
-                info = self._get_ppa_info(ppa_owner, ppa_name)
+
-                if not self._key_already_exists(info['signing_key_fingerprint']):
+            # add gpg sig if needed
-                    command = ['apt-key', 'adv', '--recv-keys', '--no-tty', '--keyserver', 'hkp://keyserver.ubuntu.com:80', info['signing_key_fingerprint']]
+            if not self._key_already_exists(info['signing_key_fingerprint']):
-                    self.add_ppa_signing_keys_callback(command)
+
-
+                # TODO: report file that would have been added if not check_mode
+                keyfile = ''
+                if not self.module.check_mode:
+                    if self.apt_key_bin:
+                        command = [self.apt_key_bin, 'adv', '--recv-keys', '--no-tty', '--keyserver', 'hkp://keyserver.ubuntu.com:80',
+                                   info['signing_key_fingerprint']]
+                    else:
+                        keyfile = '/usr/share/keyrings/%s-%s-%s.gpg' % (os.path.basename(source).replace(' ', '-'), ppa_owner, ppa_name)
+                        command = [self.gpg_bin, '--no-tty', '--keyserver', 'hkp://keyserver.ubuntu.com:80', '--export', info['signing_key_fingerprint']]
+
+                    rc, stdout, stderr = self.module.run_command(command, check_rc=True, encoding=None)
+                    if keyfile:
+                        # using gpg we must write keyfile ourselves
+                        if len(stdout) == 0:
+                            self.module.fail_json(msg='Unable to get required signing key', rc=rc, stderr=stderr, command=command)
+                        try:
+                            with open(keyfile, 'wb') as f:
+                                f.write(stdout)
+                            self.module.log('Added repo key "%s" for apt to file "%s"' % (info['signing_key_fingerprint'], keyfile))
+                        except (OSError, IOError) as e:
+                            self.module.fail_json(msg='Unable to add required signing key for%s ', rc=rc, stderr=stderr, error=to_native(e))
+
+            # apt source file
             file = file or self._suggest_filename('%s_%s' % (line, self.codename))
         else:
             source = self._parse(line, raise_if_invalid_or_disabled=True)[2]
             file = file or self._suggest_filename(source)
+
         self._add_valid_source(source, comment, file)
 
     def remove_source(self, line):
@@ -501,16 +560,6 @@ class UbuntuSourcesList(SourcesList):
         return _repositories
 
 
-def get_add_ppa_signing_key_callback(module):
-    def _run_command(command):
-        module.run_command(command, check_rc=True)
-
-    if module.check_mode:
-        return None
-    else:
-        return _run_command
-
-
 def revert_sources_list(sources_before, sources_after, sourceslist_before):
     '''Revert the sourcelist files to their previous state.'''
 
@@ -601,7 +650,7 @@ def main():
         module.fail_json(msg='Please set argument \'repo\' to a non-empty value')
 
     if isinstance(distro, aptsources_distro.Distribution):
-        sourceslist = UbuntuSourcesList(module, add_ppa_signing_keys_callback=get_add_ppa_signing_key_callback(module))
+        sourceslist = UbuntuSourcesList(module)
     else:
         module.fail_json(msg='Module apt_repository is not supported on target.')