From 6ed2c6e65988246adf9c4d83ac87a70f6bd2f44c Mon Sep 17 00:00:00 2001 From: Zachary Szczesniak Date: Thu, 7 May 2026 16:34:09 -0400 Subject: [PATCH] Add quick sign key function, fix verify file to take str or IO, and allow hidden recipients when encrypting a file. --- gnupg.py | 59 +++++++++++++++++++++++++++----- test_gnupg.py | 93 +++++++++++++++++++++++++++++++++++++++++++++++++-- 2 files changed, 140 insertions(+), 12 deletions(-) diff --git a/gnupg.py b/gnupg.py index 022138e..5f02a45 100644 --- a/gnupg.py +++ b/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,6 +2100,34 @@ 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 # @@ -2104,6 +2135,7 @@ def add_subkey(self, master_key, master_passphrase=None, algorithm='rsa', usage= 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 diff --git a/test_gnupg.py b/test_gnupg.py index 340b5b3..fc731a4 100644 --- a/test_gnupg.py +++ b/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']),