Add quick_sign_key function, add hidden recipients to encrypt_file, and allow paths in verify_file

gnupg.py

@@ -1562,9 +1562,12 @@ def verify_file(self, fileobj_or_path, data_filename=None, close_file=True, extr
         else:
             logger.debug('Handling detached verification')
             import tempfile
+            fileobj = self._get_fileobj(fileobj_or_path)
             fd, fn = tempfile.mkstemp(prefix='pygpg-')
-            s = fileobj_or_path.read()
-            if close_file:
+            s = fileobj.read()
+            if fileobj is not fileobj_or_path:
+                fileobj.close()
+            elif close_file:
                 fileobj_or_path.close()
             logger.debug('Wrote to temp file: %r', s)
             os.write(fd, s)
@@ -2097,13 +2100,42 @@ def add_subkey(self, master_key, master_passphrase=None, algorithm='rsa', usage=
         self._handle_io(args, f, result, passphrase=master_passphrase, binary=True)
         return result
 
+    def quick_sign_key(self, certifier_fingerprint, recipient_fingerprint, certifier_passphrase=None):
+        """
+        Certify a key using quick-sign-key function.
+
+        Args:
+            certifier_fingerprint (str): The fingerprint for the certifying key.
+
+            recipient_fingerprint (str): The fingerprint of the key being signed.
+
+            certifier_passphrase (str): The passphrase for the certifing key.
+        """
+        if self.version[0] < 2:
+            raise NotImplementedError('Not available in GnuPG 1.x')
+        if not certifier_fingerprint:  # pragma: no cover
+            raise ValueError('No certifier key fingerprint specified')
+        if not recipient_fingerprint:  # pragma: no cover
+            raise ValueError('No recipient key fingerprint specified')
+        if certifier_passphrase and not self.is_valid_passphrase(certifier_passphrase):  # pragma: no cover
+            raise ValueError('Invalid passphrase')
+
+        args = ['--local-user', certifier_fingerprint, '--quick-sign-key', recipient_fingerprint]
+
+        result = self.result_map['sign'](self)
+
+        f = _make_binary_stream('', self.encoding)
+        self._handle_io(args, f, result, passphrase=certifier_passphrase, binary=True)
+        return result
+
     #
     # ENCRYPTION
     #
 
     def encrypt_file(self,
                      fileobj_or_path,
                      recipients,
+                     hidden_recipients=None,
                      sign=None,
                      always_trust=False,
                      passphrase=None,
@@ -2119,6 +2151,8 @@ def encrypt_file(self,
 
             recipients (str|list): A key id of a recipient of the encrypted data, or a list of such key ids.
 
+            hidden_recipients (str|list): A key id of a hidden recipient of the encrypted data, or a list of such key ids.
+
             sign (str): If specified, the key id of a signer to sign the encrypted data.
 
             always_trust (bool): Whether to always trust keys.
@@ -2144,13 +2178,20 @@ def encrypt_file(self,
                 args.extend(['--cipher-algo', no_quote(symmetric)])
             # else use the default, currently CAST5
         else:
-            if not recipients:
-                raise ValueError('No recipients specified with asymmetric '
-                                 'encryption')
-            if not _is_sequence(recipients):
-                recipients = (recipients, )
-            for recipient in recipients:
-                args.extend(['--recipient', no_quote(recipient)])
+            if not recipients and not hidden_recipients:
+                raise ValueError('No recipients or hidden recipients specified with '
+                                 'asymmetric encryption')
+            if recipients:
+                if not _is_sequence(recipients):
+                    recipients = (recipients, )
+                for recipient in recipients:
+                    args.extend(['--recipient', no_quote(recipient)])
+
+            if hidden_recipients:
+                if not _is_sequence(hidden_recipients):
+                    hidden_recipients = (hidden_recipients, )
+                for hidden_recipient in hidden_recipients:
+                    args.extend(['--hidden-recipient', no_quote(hidden_recipient)])
         if armor:  # create ascii-armored output - False for binary output
             args.append('--armor')
         if output:  # pragma: no cover

test_gnupg.py

@@ -181,6 +181,34 @@ def skipIf(condition, message):
 -----END PGP PRIVATE KEY BLOCK-----
 """
 
+CERTIFYING_KEY = """
+-----BEGIN PGP PRIVATE KEY BLOCK-----
+
+lHcEaeE/WBMIKoZIzj0DAQcCAwQJX+QJbszp7FFHIaGY1ZOwLJCTnwjzy1Z5vnKw
+1AZ9UnIRO+TMPEEUizEc4FO1nQBUgCS2nOccwXpnZtavc8d5AAD/TitHUDwl1CbF
+f2FGF4alBhMBuWohWcAUNOopbKgaNO4MubQfQ2VydGlmaWVyIFRlc3QgPGNlcnRp
+ZmllckB0ZXN0PoiQBBMTCAA4FiEEl+gdlmY3v7p/43AlhW+ORWgM16QFAmnhP1gC
+GwMFCwkIBwIGFQoJCAsCBBYCAwECHgECF4AACgkQhW+ORWgM16RwdQEA9CTM/Zz+
+rWNl3ToKdsPKS7s3KaPfvGPKNIqVwUJzhT8BAN58ziizYcb85HREsFHtYOJs0Uti
+7GYLD4MPZxhIz5sr
+=he2w
+-----END PGP PRIVATE KEY BLOCK-----
+"""
+
+RECIPIENT_KEY = """
+-----BEGIN PGP PRIVATE KEY BLOCK-----
+
+lHcEaeE//hMIKoZIzj0DAQcCAwRraYDaESix05+l8b69fKzIvYmIoXbaOVPoCnjA
+Qe6hYEKQrO7p5zOUp6lLXhnZ6JWD6B7RcoGSHpAHQMWzpzkWAAEA92FWKM7TZolx
+Wpvuj+6lLf6wrg/gOVofvjKDoj9IbfARALQfUmVjaXBpZW50IFRlc3QgPFJlY2lw
+aWVudEB0ZXN0PoiQBBMTCAA4FiEEqHrbKqME4ufCGsGh2ILlpHJXa3EFAmnhP/4C
+GwMFCwkIBwIGFQoJCAsCBBYCAwECHgECF4AACgkQ2ILlpHJXa3E2NAEAlbZ+sUpL
+j88bPK7sBA0pgiAItWYclDgZXAmqBCXz9ggBAM4khSii4pshJh0XavURaYnoC2SU
+qlWoXWGVJkVDlHla
+=1zeo
+-----END PGP PRIVATE KEY BLOCK-----
+"""
+
 
 def is_list_with_len(o, n):
     return isinstance(o, list) and len(o) == n
@@ -771,6 +799,28 @@ def test_scan_keys_mem(self):
                 uids.add(d['uids'][0])
             self.assertEqual(uids, expected)
 
+    def test_quick_sign_key(self):
+        "Test the quick-sign-key functionality"
+        # GPG requires real random when signing keys
+        self.gpg.options.remove('--debug-quick-random')
+
+        recipient_key = self.gpg.import_keys(RECIPIENT_KEY)
+        certifying_key = self.gpg.import_keys(CERTIFYING_KEY)
+        self.assertEqual(len(set(recipient_key.fingerprints)), 1)
+        self.assertEqual(len(set(certifying_key.fingerprints)), 1)
+        certifying_fingerprint = certifying_key.fingerprints[0]
+        recipient_fingerprint = recipient_key.fingerprints[0]
+
+        sign_result = self.gpg.quick_sign_key(certifying_fingerprint, recipient_fingerprint)
+        self.assertEqual(sign_result.returncode, 0)
+        sigs = self.gpg.list_keys(keys=recipient_fingerprint, sigs=True)[0]['sigs']
+        key_id = sigs[1][0]
+        self.assertIn(key_id, certifying_key.fingerprints[0])
+
+        # Revert our test environment changes
+        self.gpg.options.append('--debug-quick-random')
+
+
     def test_encryption_and_decryption(self):
         "Test that encryption and decryption works"
         key = self.generate_key('Andrew', 'Able', 'alpha.com', passphrase='andy')
@@ -807,6 +857,30 @@ def test_encryption_and_decryption(self):
         self.assertEqual(data, ddata.data, 'Round-trip must work')
         ddata = gpg.decrypt(edata, passphrase='bbrown')
         self.assertEqual(data, ddata.data, 'Round-trip must work')
+        # Test with hidden recipients
+        result = gpg.encrypt(data, andrew, hidden_recipients=barbara)
+        self.assertEqual(0, result.returncode, 'Non-zero return code')
+        edata = str(result)
+        self.assertNotEqual(data, edata, 'Data must have changed')
+        ddata = gpg.decrypt(edata, passphrase='andy')
+        self.assertEqual(0, ddata.returncode, 'Non-zero return code')
+        self.assertEqual(data, ddata.data, 'Round-trip must work')
+        ddata = gpg.decrypt(edata, passphrase='bbrown')
+        self.assertEqual(data, ddata.data, 'Round-trip must work')
+        # Test only hidden recipients
+        result = gpg.encrypt(data, None, hidden_recipients=[andrew, barbara])
+        self.assertEqual(0, result.returncode, 'Non-zero return code')
+        edata = str(result)
+        self.assertNotEqual(data, edata, 'Data must have changed')
+        ddata = gpg.decrypt(edata, passphrase='andy')
+        self.assertEqual(0, ddata.returncode, 'Non-zero return code')
+        self.assertEqual(data, ddata.data, 'Round-trip must work')
+        ddata = gpg.decrypt(edata, passphrase='bbrown')
+        self.assertEqual(data, ddata.data, 'Round-trip must work')
+        # Test with no recipients
+        self.assertRaises(ValueError, gpg.encrypt, data, None)
+        self.assertRaises(ValueError, gpg.encrypt, data, None, hidden_recipients=None)
+        self.assertRaises(ValueError, gpg.encrypt, data, None, hidden_recipients=None, symmetric=False)
         # Test symmetric encryption
         data = 'chippy was here'
         self.assertRaises(ValueError, gpg.encrypt, data, None, passphrase='bbr\x00own', symmetric=True)
@@ -1053,10 +1127,23 @@ def test_signature_file(self):
         data_file.close()
         try:
             verified = self.gpg.verify_data(sig_file, data)
-            self.assertTrue(verified.username.startswith('Andrew Able'))
-            self.assertTrue(key.fingerprint.endswith(verified.key_id))
+        except Exception as e:
+            os.remove(sig_file)
+            self.fail(e)
+        self.assertTrue(verified.username.startswith('Andrew Able'))
+        self.assertTrue(key.fingerprint.endswith(verified.key_id))
+        self.assertEqual(0, verified.returncode, 'Non-zero return code')
+        if key.fingerprint != verified.fingerprint:  # pragma: no cover
+            logger.debug('key: %r', key.fingerprint)
+            logger.debug('ver: %r', verified.fingerprint)
+        self.assertEqual(key.fingerprint, verified.fingerprint, 'Fingerprints must match')
+        # Test file path verification
+        try:
+            verified = self.gpg.verify_file(sig_file, self.test_fn)
         finally:
             os.remove(sig_file)
+        self.assertTrue(verified.username.startswith('Andrew Able'))
+        self.assertTrue(key.fingerprint.endswith(verified.key_id))
         self.assertEqual(0, verified.returncode, 'Non-zero return code')
         if key.fingerprint != verified.fingerprint:  # pragma: no cover
             logger.debug('key: %r', key.fingerprint)
@@ -1631,7 +1718,7 @@ def test_exception_propagation(self):
         'test_key_generation_with_invalid_key_type', 'test_key_generation_with_escapes', 'test_key_generation_input',
         'test_key_generation_with_colons', 'test_search_keys', 'test_scan_keys', 'test_scan_keys_mem',
         'test_key_trust', 'test_add_subkey', 'test_add_subkey_with_invalid_key_type', 'test_deletion_subkey',
-        'test_list_subkey_after_generation'
+        'test_list_subkey_after_generation', 'test_quick_sign_key'
     ]),
     'import':
     set(['test_import_only', 'test_doctest_import_keys']),