Merge branch 'skip-gpg-import' into unfreeze-ui

securedrop_client/api_jobs/downloads.py

@@ -63,6 +63,7 @@ def call_api(self, api_client: API, session: Session) -> Any:
                              remote_replies,
                              self.data_dir)
 
+        fingerprints = self.gpg.fingerprints()
         for source in remote_sources:
             if source.key and source.key.get('type', None) == 'PGP':
                 pub_key = source.key.get('public', None)
@@ -72,7 +73,13 @@ def call_api(self, api_client: API, session: Session) -> Any:
                     # as it will show as uncovered due to a cpython compiler optimziation.
                     # See: https://bugs.python.org/issue2506
                     continue  # pragma: no cover
+
+                if fingerprint in fingerprints:
+                    logger.info("Skipping import of key with fingerprint %s", fingerprint)
+                    continue
+
                 try:
+                    logger.info("Importing key with fingerprint %s", fingerprint)
                     self.gpg.import_key(source.uuid, pub_key, fingerprint)
                 except CryptoError:
                     logger.warning('Failed to import key for source {}'.format(source.uuid))

securedrop_client/crypto.py

@@ -22,6 +22,7 @@
 import struct
 import subprocess
 import tempfile
+import typing
 
 from sqlalchemy.orm import scoped_session
 from uuid import UUID
@@ -141,6 +142,28 @@ def _gpg_cmd_base(self) -> list:
         cmd.extend(['--trust-model', 'always'])
         return cmd
 
+    def fingerprints(self) -> typing.Dict[str, bool]:
+        """
+        Returns a map of key fingerprints.
+
+        The result is a map wherein each key is the fingerprint of a
+        key on our keyring, mapped to True. It's intended to help us
+        avoid expensive import operations for keys we already have.
+        """
+        cmd = self._gpg_cmd_base()
+        cmd.extend(["--list-public-keys", "--fingerprint", "--with-colons",
+                    "--fixed-list-mode", "--list-options", "no-show-photos"])
+        output = subprocess.check_output(cmd, text=True)
+
+        fingerprints = {}
+        for line in output.splitlines():
+            if line.startswith("fpr:"):
+                fields = line.split(":")
+                fingerprint = fields[9]
+                fingerprints[fingerprint] = True
+
+        return fingerprints
+
     def import_key(self, source_uuid: UUID, key_data: str, fingerprint: str) -> None:
         session = self.session_maker()
         local_source = session.query(Source).filter_by(uuid=source_uuid).one()