From 6fa8b2a9e7f02c59f794e9dd080fac574841e50b Mon Sep 17 00:00:00 2001 From: Ruben Pollan Date: Fri, 19 Dec 2014 22:37:40 -0600 Subject: upgrade key when signed by old key --- src/leap/keymanager/__init__.py | 2 +- src/leap/keymanager/openpgp.py | 99 +++++++++++++++++----------- src/leap/keymanager/tests/test_validation.py | 55 ++++++++++++++++ src/leap/keymanager/validation.py | 6 +- 4 files changed, 121 insertions(+), 41 deletions(-) (limited to 'src/leap/keymanager') diff --git a/src/leap/keymanager/__init__.py b/src/leap/keymanager/__init__.py index fdbc206..a1a59f5 100644 --- a/src/leap/keymanager/__init__.py +++ b/src/leap/keymanager/__init__.py @@ -25,7 +25,7 @@ try: assert(GPGUtilities) # pyflakes happy from gnupg import __version__ as _gnupg_version from pkg_resources import parse_version - assert(parse_version(_gnupg_version) >= parse_version('1.2.3')) + assert(parse_version(_gnupg_version) >= parse_version('1.4.0')) except (ImportError, AssertionError): print "*******" diff --git a/src/leap/keymanager/openpgp.py b/src/leap/keymanager/openpgp.py index 0adfc52..794a0ec 100644 --- a/src/leap/keymanager/openpgp.py +++ b/src/leap/keymanager/openpgp.py @@ -163,39 +163,6 @@ class TempGPGWrapper(object): shutil.rmtree(self._gpg.homedir) -def _build_key_from_gpg(key, key_data): - """ - Build an OpenPGPKey based on C{key} from local gpg storage. - - ASCII armored GPG key data has to be queried independently in this - wrapper, so we receive it in C{key_data}. - - :param key: Key obtained from GPG storage. - :type key: dict - :param key_data: Key data obtained from GPG storage. - :type key_data: str - :return: An instance of the key. - :rtype: OpenPGPKey - """ - expiry_date = None - if key['expires']: - expiry_date = datetime.fromtimestamp(int(key['expires'])) - address = [] - for uid in key['uids']: - address.append(_parse_address(uid)) - - return OpenPGPKey( - address, - key_id=key['keyid'], - fingerprint=key['fingerprint'], - key_data=key_data, - private=True if key['type'] == 'sec' else False, - length=int(key['length']), - expiry_date=expiry_date, - refreshed_at=datetime.now(), - ) - - def _parse_address(address): """ Remove name, '<', '>' and the identity suffix after the '+' until the '@' @@ -221,6 +188,26 @@ class OpenPGPKey(EncryptionKey): Base class for OpenPGP keys. """ + def __init__(self, address, gpgbinary=None, **kwargs): + self._gpgbinary = gpgbinary + super(OpenPGPKey, self).__init__(address, **kwargs) + + @property + def signatures(self): + """ + Get the key signatures + + :return: the key IDs that have signed the key + :rtype: list(str) + """ + with TempGPGWrapper(keys=[self], gpgbinary=self._gpgbinary) as gpg: + res = gpg.list_sigs(self.key_id) + for uid, sigs in res.sigs.iteritems(): + if _parse_address(uid) in self.address: + return sigs + + return [] + class OpenPGPScheme(EncryptionScheme): """ @@ -298,7 +285,7 @@ class OpenPGPScheme(EncryptionScheme): deferreds = [] for secret in [True, False]: key = gpg.list_keys(secret=secret).pop() - openpgp_key = _build_key_from_gpg( + openpgp_key = self._build_key_from_gpg( key, gpg.export_keys(key['fingerprint'], secret=secret)) d = self.put_key(openpgp_key, address) @@ -335,7 +322,9 @@ class OpenPGPScheme(EncryptionScheme): leap_assert( address in doc.content[KEY_ADDRESS_KEY], 'Wrong address in key data.') - return build_key_from_dict(OpenPGPKey, doc.content) + key = build_key_from_dict(OpenPGPKey, doc.content) + key._gpgbinary = self._gpgbinary + return key d = self._get_key_doc(address, private) d.addCallback(build_key) @@ -375,7 +364,7 @@ class OpenPGPScheme(EncryptionScheme): openpgp_privkey = None if privkey is not None: # build private key - openpgp_privkey = _build_key_from_gpg( + openpgp_privkey = self._build_key_from_gpg( privkey, gpg.export_keys(privkey['fingerprint'], secret=True)) leap_check(pubkey['fingerprint'] == privkey['fingerprint'], @@ -383,7 +372,7 @@ class OpenPGPScheme(EncryptionScheme): errors.KeyFingerprintMismatch) # build public key - openpgp_pubkey = _build_key_from_gpg( + openpgp_pubkey = self._build_key_from_gpg( pubkey, gpg.export_keys(pubkey['fingerprint'], secret=False)) @@ -452,7 +441,7 @@ class OpenPGPScheme(EncryptionScheme): gpg.import_keys(oldkey.key_data) gpg.import_keys(key.key_data) gpgkey = gpg.list_keys(secret=key.private).pop() - mergedkey = _build_key_from_gpg( + mergedkey = self._build_key_from_gpg( gpgkey, gpg.export_keys(gpgkey['fingerprint'], secret=key.private)) @@ -571,6 +560,40 @@ class OpenPGPScheme(EncryptionScheme): d.addCallback(get_key_from_active_doc) return d + def _build_key_from_gpg(self, key, key_data): + """ + Build an OpenPGPKey for C{address} based on C{key} from + local gpg storage. + + ASCII armored GPG key data has to be queried independently in this + wrapper, so we receive it in C{key_data}. + + :param key: Key obtained from GPG storage. + :type key: dict + :param key_data: Key data obtained from GPG storage. + :type key_data: str + :return: An instance of the key. + :rtype: OpenPGPKey + """ + expiry_date = None + if key['expires']: + expiry_date = datetime.fromtimestamp(int(key['expires'])) + address = [] + for uid in key['uids']: + address.append(_parse_address(uid)) + + return OpenPGPKey( + address, + gpgbinary=self._gpgbinary, + key_id=key['keyid'], + fingerprint=key['fingerprint'], + key_data=key_data, + private=True if key['type'] == 'sec' else False, + length=int(key['length']), + expiry_date=expiry_date, + refreshed_at=datetime.now(), + ) + def delete_key(self, key): """ Remove C{key} from storage. diff --git a/src/leap/keymanager/tests/test_validation.py b/src/leap/keymanager/tests/test_validation.py index a8b35ca..15e7d27 100644 --- a/src/leap/keymanager/tests/test_validation.py +++ b/src/leap/keymanager/tests/test_validation.py @@ -117,6 +117,14 @@ class ValidationLevelTestCase(KeyManagerWithSoledadTestCase): validation=ValidationLevel.Provider_Endorsement) yield self.assertFailure(d, KeyNotValidUpgrade) + @inlineCallbacks + def test_signed_key(self): + km = self._key_manager() + yield km.put_raw_key(PUBLIC_KEY, OpenPGPKey, ADDRESS) + yield km.put_raw_key(SIGNED_KEY, OpenPGPKey, ADDRESS) + key = yield km.get_key(ADDRESS, OpenPGPKey, fetch_remote=False) + self.assertEqual(key.fingerprint, SIGNED_FINGERPRINT) + # Key material for testing @@ -284,6 +292,53 @@ HEsoyd7WOsuse7+NkyUHgMXMVW7cz+nU7iO+ht2rkBtv+Z5LGlzgHTeFjKci -----END PGP PRIVATE KEY BLOCK----- """ +# key CA1AD31E: public key "Leap Test Key " +# signed by E36E738D69173C13D709E44F2F455E2824D18DDF +SIGNED_FINGERPRINT = "6704CF3362087DA23E3D2DF8ED81DFD1CA1AD31E" +SIGNED_KEY = """ +-----BEGIN PGP PUBLIC KEY BLOCK----- +Version: GnuPG v1.4.12 (GNU/Linux) + +mQENBFQ9DHMBCADJXyNVzTQ+NnmSDbR6q8jjDsnqk/IgKrMBkpjNxUa/0HQ4o0Yh +pklzR1hIc/jsdgq42A0++pqdfQFeRc2NVw/NnE/9uzW73YuaWg5XnWGjuAP3UeRI +3xjL/cscEFmGfGkuGvFpIVa7GBPqz1SMBXWULJbkCE1pnHfgqh0R7oc5u0omnsln +0zIrmLX1ufpDRSUedjSgIfd6VqbkPm3NJuZE4NVn6spHG3zTxqcaPCG0xLfHw7eS +qgUdz0BFaxqtQiXffBpA3KvGJW0792VjDh4M6kDvpeYpKRmB9oEYlT3n3KvQrdPE +B3N5KrzJj1QIL990q4NQdqjg+jUE5zCJsTdzABEBAAG0HExlYXAgVGVzdCBLZXkg +PGxlYXBAbGVhcC5zZT6JATgEEwECACIFAlQ9DHMCGwMGCwkIBwMCBhUIAgkKCwQW +AgMBAh4BAheAAAoJEO2B39HKGtMeI/4H/0/OG1OqtQEoYscvJ+BZ3ZrM2pEk7KDd +7AEEf6QIGSd38GFyn/pve24cpRLv7phKNy9dX9VJhTDobpKvK0ZT/yQO3FVlySAN +NVpu93/jrLnrW51J3p/GP952NtUAEP5l1uyYIKZ1W3RLWws72Lh34HTaHAWC94oF +vnS42IYdTn4y6lfizL+wYD6CnfrIpHm8v3NABEQZ8e/jllrRK0pnOxAdFv/TpWEl +8AnTZXcejSBgCG6UmDtrRKfgoQlGJEIH61QSqHpRIwkepQVYexUwgcLFAZPI9Hvw +N5pZQ5Z+XcsYTGtLNEpF7YW6ykLDRTAv6LiBQPkBX8TDKhkh95Cs3sKJAhwEEAEC +AAYFAlQ9DgIACgkQL0VeKCTRjd/pABAAsNPbiGwuZ469NxwTgf3+EMoWZNHf5ZRa +ZbzKKesLFEElMdX3Q/MkVc1T8Rsy9Fdn1tf/H6glwuKyqWeXNkxa86VT6gck9WV6 +bslFIl/vJpb3dcmiCCM1tSCYpX0yE0fqebMihcfvNqDw3GdZBUo7R0pWN6BEh4iM +YYWTAtuPCrbsv+2bSid1ZLIO6FIF5kskg60h/IbSr+A+DSBgyyjf9fbUv6MoyMw8 +08GtCAx6VGJhTTC/RkWIA+N3n83W5XQFszOOg/PAAg0JMUXUBGvjfYJ5fcB8cfuw +1XZe9uWsDmYpwfVEtDajrLbatkXAu22pjIJnB4cVqiD+4hHbBCFkeZIfdRsPEINO +UacsjVZV5/EPDN9OpkvZbkrLJ6eaQnmQZnFclquNHUCqFI0QYUml0BXXaZq+aEJ9 +N9x00kdYV1xW6zkL+MGgxdViC5n6dwJcU3MANrykV8Cc5/x+wmwY8AXbHzU7MxvY +nGlAYsAZHhf4ZlEdAO6C329VotMxBLFd5DJZZoN+ysaOpsUNRl0JO41+6bbI141l +DCmzWUG4iTI70zxsgzZGgEt0HlMDoIxElPcy/jDKi1IfEDmveK+QR9WphM40Ayvx +VTeA6g9WagmoHopQs/D/Kbi3Q8izFDfXTwA52DUxTjyUEFn0jEOiG9BFmnIkQ6LE +3WkIJFd3D0+5AQ0EVD0McwEIALRXukBsOrcA/rNJ4SV4I64cGdN8q9Gl5RpLl8cS +L5+SGHp6KoCL4daBqpbxdhE/Ylb3QmPt2SBZbTkwJ2yuczELOyhH6R13OWRWCgkd +dYLZsm/sEzu7dVoFQ4peKTGDzI8mQ/s157wRkz/8iSUYjJjeM3g0NI55FVcefibN +pOOFRaYGUh8itofRFOu7ipZ9F8zRJdBwqISe1gemNBR+O3G3Srm34PYu6sZRsdLU +Nf+81+/ynQWQseVpbz8X93sx/onIYIY0w+kxIE0oR/gBBjdsMOp7EfcvtbGTgplQ ++Zxln/pV2bVFkGkjKniFEEfi4eCPknCj0+67qKRt/Fc9n98AEQEAAYkBHwQYAQIA +CQUCVD0McwIbDAAKCRDtgd/RyhrTHmmcCACpTbjOYjvvr8/rD60bDyhnfcKvpavO +1/gZtMeEmw5gysp10mOXwhc2XuC3R1A6wVbVgGuOp5RxlxI4w8xwwxMFSp6u2rS5 +jm5slXBKB2i3KE6Sao9uZKP2K4nS8Qc+DhathfgafI39vPtBmsb1SJd5W1njNnYY +hARRpViUcVdfvW3VRpDACZ79PBs4ZQQ022NsNAPwm/AJvAw2S42Y9tjTnaLVRLfH +lzdErcGTBnfEIi0mQF/11k/THBJxx7vaFt8YXiDlWLUkg5XW3xK9mkETbaTv+/tB +X2+l7IOSt+31KQCBFN/VmhTySJOVQC1d2A56lSH2c/DWVClji+x3suzn +=xprj +-----END PGP PUBLIC KEY BLOCK----- +""" + import unittest if __name__ == "__main__": diff --git a/src/leap/keymanager/validation.py b/src/leap/keymanager/validation.py index b3aff3e..c6fe478 100644 --- a/src/leap/keymanager/validation.py +++ b/src/leap/keymanager/validation.py @@ -60,8 +60,6 @@ def can_upgrade(new_key, old_key): :type old_key: EncryptionKey :rtype: bool """ - # XXX implement key signature checking (#6120) - # First contact if old_key is None: return True @@ -90,4 +88,8 @@ def can_upgrade(new_key, old_key): new_key.validation > old_key.validation): return True + # New key signed by the old key + if old_key.key_id in new_key.signatures: + return True + return False -- cgit v1.2.3