#!/usr/bin/python
+import getpass as _getpass
+import hashlib as _hashlib
+import math as _math
+import re as _re
import subprocess as _subprocess
import struct as _struct
+import Crypto.Cipher.AES as _crypto_cipher_aes
+import Crypto.Cipher.Blowfish as _crypto_cipher_blowfish
+import Crypto.Cipher.CAST as _crypto_cipher_cast
+import Crypto.Cipher.DES3 as _crypto_cipher_des3
+
def _get_stdout(args, stdin=None):
stdin_pipe = None
3: (None, None),
}
+ _packet_types = {
+ 0: 'reserved',
+ 1: 'public-key encrypted session key packet',
+ 2: 'signature packet',
+ 3: 'symmetric-key encrypted session key packet',
+ 4: 'one-pass signature packet',
+ 5: 'secret-key packet',
+ 6: 'public-key packet',
+ 7: 'secret-subkey packet',
+ 8: 'compressed data packet',
+ 9: 'symmetrically encrypted data packet',
+ 10: 'marker packet',
+ 11: 'literal data packet',
+ 12: 'trust packet',
+ 13: 'user id packet',
+ 14: 'public-subkey packet',
+ 17: 'user attribute packet',
+ 18: 'sym. encrypted and integrity protected data packet',
+ 19: 'modification detection code packet',
+ 60: 'private',
+ 61: 'private',
+ 62: 'private',
+ 63: 'private',
+ }
+
+ _public_key_algorithms = {
+ 1: 'rsa (encrypt or sign)',
+ 2: 'rsa encrypt-only',
+ 3: 'rsa sign-only',
+ 16: 'elgamal (encrypt-only)',
+ 17: 'dsa (digital signature algorithm)',
+ 18: 'reserved for elliptic curve',
+ 19: 'reserved for ecdsa',
+ 20: 'reserved (formerly elgamal encrypt or sign)',
+ 21: 'reserved for diffie-hellman',
+ 100: 'private',
+ 101: 'private',
+ 102: 'private',
+ 103: 'private',
+ 104: 'private',
+ 105: 'private',
+ 106: 'private',
+ 107: 'private',
+ 108: 'private',
+ 109: 'private',
+ 110: 'private',
+ }
+
+ _symmetric_key_algorithms = {
+ 0: 'plaintext or unencrypted data',
+ 1: 'idea',
+ 2: 'tripledes',
+ 3: 'cast5',
+ 4: 'blowfish',
+ 5: 'reserved',
+ 6: 'reserved',
+ 7: 'aes with 128-bit key',
+ 8: 'aes with 192-bit key',
+ 9: 'aes with 256-bit key',
+ 10: 'twofish',
+ 100: 'private',
+ 101: 'private',
+ 102: 'private',
+ 103: 'private',
+ 104: 'private',
+ 105: 'private',
+ 106: 'private',
+ 107: 'private',
+ 108: 'private',
+ 109: 'private',
+ 110: 'private',
+ }
+
+ _cipher_block_size = { # in bits
+ 'aes with 128-bit key': 128,
+ 'aes with 192-bit key': 128,
+ 'aes with 256-bit key': 128,
+ 'cast5': 64,
+ }
+
+ _crypto_module = {
+ 'aes with 128-bit key': _crypto_cipher_aes,
+ 'aes with 192-bit key': _crypto_cipher_aes,
+ 'aes with 256-bit key': _crypto_cipher_aes,
+ 'blowfish': _crypto_cipher_blowfish,
+ 'cast5': _crypto_cipher_cast,
+ 'tripledes': _crypto_cipher_des3,
+ }
+
+ _compression_algorithms = {
+ 0: 'uncompressed',
+ 1: 'zip',
+ 2: 'zlib',
+ 3: 'bzip2',
+ 100: 'private',
+ 101: 'private',
+ 102: 'private',
+ 103: 'private',
+ 104: 'private',
+ 105: 'private',
+ 106: 'private',
+ 107: 'private',
+ 108: 'private',
+ 109: 'private',
+ 110: 'private',
+ }
+
+ _hash_algorithms = {
+ 1: 'md5',
+ 2: 'sha-1',
+ 3: 'ripe-md/160',
+ 4: 'reserved',
+ 5: 'reserved',
+ 6: 'reserved',
+ 7: 'reserved',
+ 8: 'sha256',
+ 9: 'sha384',
+ 10: 'sha512',
+ 11: 'sha224',
+ 100: 'private',
+ 101: 'private',
+ 102: 'private',
+ 103: 'private',
+ 104: 'private',
+ 105: 'private',
+ 106: 'private',
+ 107: 'private',
+ 108: 'private',
+ 109: 'private',
+ 110: 'private',
+ }
+
+ _string_to_key_types = {
+ 0: 'simple',
+ 1: 'salted',
+ 2: 'reserved',
+ 3: 'iterated and salted',
+ 100: 'private',
+ 101: 'private',
+ 102: 'private',
+ 103: 'private',
+ 104: 'private',
+ 105: 'private',
+ 106: 'private',
+ 107: 'private',
+ 108: 'private',
+ 109: 'private',
+ 110: 'private',
+ }
+
+ _string_to_key_expbias = 6
+
+ _signature_types = {
+ 0x00: 'binary document',
+ 0x01: 'canonical text document',
+ 0x02: 'standalone',
+ 0x10: 'generic user id and public-key packet',
+ 0x11: 'persona user id and public-key packet',
+ 0x12: 'casual user id and public-key packet',
+ 0x13: 'postitive user id and public-key packet',
+ 0x18: 'subkey binding',
+ 0x19: 'primary key binding',
+ 0x1F: 'direct key',
+ 0x20: 'key revocation',
+ 0x28: 'subkey revocation',
+ 0x30: 'certification revocation',
+ 0x40: 'timestamp',
+ 0x50: 'third-party confirmation',
+ }
+
+ _signature_subpacket_types = {
+ 0: 'reserved',
+ 1: 'reserved',
+ 2: 'signature creation time',
+ 3: 'signature expiration time',
+ 4: 'exportable certification',
+ 5: 'trust signature',
+ 6: 'regular expression',
+ 7: 'revocable',
+ 8: 'reserved',
+ 9: 'key expiration time',
+ 10: 'placeholder for backward compatibility',
+ 11: 'preferred symmetric algorithms',
+ 12: 'revocation key',
+ 13: 'reserved',
+ 14: 'reserved',
+ 15: 'reserved',
+ 16: 'issuer',
+ 17: 'reserved',
+ 18: 'reserved',
+ 19: 'reserved',
+ 20: 'notation data',
+ 21: 'preferred hash algorithms',
+ 22: 'preferred compression algorithms',
+ 23: 'key server preferences',
+ 24: 'preferred key server',
+ 25: 'primary user id',
+ 26: 'policy uri',
+ 27: 'key flags',
+ 28: 'signer user id',
+ 29: 'reason for revocation',
+ 30: 'features',
+ 31: 'signature target',
+ 32: 'embedded signature',
+ 100: 'private',
+ 101: 'private',
+ 102: 'private',
+ 103: 'private',
+ 104: 'private',
+ 105: 'private',
+ 106: 'private',
+ 107: 'private',
+ 108: 'private',
+ 109: 'private',
+ 110: 'private',
+ }
+
+ _clean_type_regex = _re.compile('\W+')
+
+ def _clean_type(self, type=None):
+ if type is None:
+ type = self['type']
+ return self._clean_type_regex.sub('_', type)
+
+ @staticmethod
+ def _reverse(dict, value):
+ """Reverse lookups in dictionaries
+
+ >>> PGPPacket._reverse(PGPPacket._packet_types, 'public-key packet')
+ 6
+ """
+ return [k for k,v in dict.items() if v == value][0]
+
+ def __str__(self):
+ method_name = '_str_{}'.format(self._clean_type())
+ method = getattr(self, method_name, None)
+ if not method:
+ return self['type']
+ details = method()
+ return '{}: {}'.format(self['type'], details)
+
+ def _str_public_key_packet(self):
+ return self._str_generic_key_packet()
+
+ def _str_public_subkey_packet(self):
+ return self._str_generic_key_packet()
+
+ def _str_secret_key_packet(self):
+ return self._str_generic_key_packet()
+
+ def _str_secret_subkey_packet(self):
+ return self._str_generic_key_packet()
+
+ def _str_generic_key_packet(self):
+ return self['fingerprint'][-8:].upper()
+
+ def _str_signature_packet(self):
+ lines = [self['signature-type']]
+ if self['hashed-subpackets']:
+ lines.append(' hashed subpackets:')
+ lines.extend(self._str_signature_subpackets(
+ self['hashed-subpackets'], prefix=' '))
+ if self['unhashed-subpackets']:
+ lines.append(' unhashed subpackets:')
+ lines.extend(self._str_signature_subpackets(
+ self['unhashed-subpackets'], prefix=' '))
+ return '\n'.join(lines)
+
+ def _str_signature_subpackets(self, subpackets, prefix):
+ lines = []
+ for subpacket in subpackets:
+ method_name = '_str_{}_signature_subpacket'.format(
+ self._clean_type(type=subpacket['type']))
+ method = getattr(self, method_name, None)
+ if method:
+ lines.append(' {}: {}'.format(
+ subpacket['type'],
+ method(subpacket=subpacket)))
+ else:
+ lines.append(' {}'.format(subpacket['type']))
+ return lines
+
+ def _str_signature_creation_time_signature_subpacket(self, subpacket):
+ return str(subpacket['signature-creation-time'])
+
+ def _str_issuer_signature_subpacket(self, subpacket):
+ return subpacket['issuer'][-8:].upper()
+
+ def _str_key_expiration_time_signature_subpacket(self, subpacket):
+ return str(subpacket['key-expiration-time'])
+
+ def _str_preferred_symmetric_algorithms_signature_subpacket(
+ self, subpacket):
+ return ', '.join(
+ algo for algo in subpacket['preferred-symmetric-algorithms'])
+
+ def _str_preferred_hash_algorithms_signature_subpacket(
+ self, subpacket):
+ return ', '.join(
+ algo for algo in subpacket['preferred-hash-algorithms'])
+
+ def _str_preferred_compression_algorithms_signature_subpacket(
+ self, subpacket):
+ return ', '.join(
+ algo for algo in subpacket['preferred-compression-algorithms'])
+
+ def _str_key_server_preferences_signature_subpacket(self, subpacket):
+ return ', '.join(
+ x for x in sorted(subpacket['key-server-preferences']))
+
+ def _str_primary_user_id_signature_subpacket(self, subpacket):
+ return str(subpacket['primary-user-id'])
+
+ def _str_key_flags_signature_subpacket(self, subpacket):
+ return ', '.join(x for x in sorted(subpacket['key-flags']))
+
+ def _str_features_signature_subpacket(self, subpacket):
+ return ', '.join(x for x in sorted(subpacket['features']))
+
+ def _str_embedded_signature_signature_subpacket(self, subpacket):
+ return subpacket['embedded']['signature-type']
+
+ def _str_user_id_packet(self):
+ return self['user']
+
def from_bytes(self, data):
+ offset = self._parse_header(data=data)
+ packet = data[offset:offset + self['length']]
+ if len(packet) < self['length']:
+ raise ValueError('packet too short ({} < {})'.format(
+ len(packet), self['length']))
+ offset += self['length']
+ method_name = '_parse_{}'.format(self._clean_type())
+ method = getattr(self, method_name, None)
+ if not method:
+ raise NotImplementedError(
+ 'cannot parse packet type {!r}'.format(self['type']))
+ method(data=packet)
+ return offset
+
+ def _parse_header(self, data):
packet_tag = data[0]
offset = 1
always_one = packet_tag & 1 << 7
raise ValueError('most significant packet tag bit not set')
self['new-format'] = packet_tag & 1 << 6
if self['new-format']:
- self['packet-tag'] = packet_tag & 0b111111
+ type_code = packet_tag & 0b111111
raise NotImplementedError('new-format packet length')
else:
- self['packet-tag'] = packet_tag >> 2 & 0b1111
+ type_code = packet_tag >> 2 & 0b1111
self['length-type'] = packet_tag & 0b11
length_bytes, length_type = self._old_format_packet_length_type[
self['length-type']]
length_data = data[offset: offset + length_bytes]
offset += length_bytes
self['length'] = _struct.unpack(length_format, length_data)[0]
- packet = data[offset:offset + self['length']]
- if len(packet) < self['length']:
- raise ValueError('packet too short ({} < {})'.format(
- len(packet), self['length']))
- offset += self['length']
+ self['type'] = self._packet_types[type_code]
+ return offset
+
+ @staticmethod
+ def _parse_multiprecision_integer(data):
+ r"""Parse RFC 4880's multiprecision integers
+
+ >>> PGPPacket._parse_multiprecision_integer(b'\x00\x01\x01')
+ (3, 1)
+ >>> PGPPacket._parse_multiprecision_integer(b'\x00\x09\x01\xff')
+ (4, 511)
+ """
+ bits = _struct.unpack('>H', data[:2])[0]
+ offset = 2
+ length = (bits + 7) // 8
+ value = 0
+ for i in range(length):
+ value += data[offset + i] * 1 << (8 * (length - i - 1))
+ offset += length
+ return (offset, value)
+
+ @classmethod
+ def _decode_string_to_key_count(cls, data):
+ r"""Decode RFC 4880's string-to-key count
+
+ >>> PGPPacket._decode_string_to_key_count(b'\x97'[0])
+ 753664
+ """
+ return (16 + (data & 15)) << ((data >> 4) + cls._string_to_key_expbias)
+
+ def _parse_string_to_key_specifier(self, data):
+ self['string-to-key-type'] = self._string_to_key_types[data[0]]
+ offset = 1
+ if self['string-to-key-type'] == 'simple':
+ self['string-to-key-hash-algorithm'] = self._hash_algorithms[
+ data[offset]]
+ offset += 1
+ elif self['string-to-key-type'] == 'salted':
+ self['string-to-key-hash-algorithm'] = self._hash_algorithms[
+ data[offset]]
+ offset += 1
+ self['string-to-key-salt'] = data[offset: offset + 8]
+ offset += 8
+ elif self['string-to-key-type'] == 'iterated and salted':
+ self['string-to-key-hash-algorithm'] = self._hash_algorithms[
+ data[offset]]
+ offset += 1
+ self['string-to-key-salt'] = data[offset: offset + 8]
+ offset += 8
+ self['string-to-key-count'] = self._decode_string_to_key_count(
+ data=data[offset])
+ offset += 1
+ else:
+ raise NotImplementedError(
+ 'string-to-key type {}'.format(self['string-to-key-type']))
+ return offset
+
+ def _parse_public_key_packet(self, data):
+ self._parse_generic_public_key_packet(data=data)
+
+ def _parse_public_subkey_packet(self, data):
+ self._parse_generic_public_key_packet(data=data)
+
+ def _parse_generic_public_key_packet(self, data):
+ self['key-version'] = data[0]
+ offset = 1
+ if self['key-version'] != 4:
+ raise NotImplementedError(
+ 'public (sub)key packet version {}'.format(
+ self['key-version']))
+ length = 5
+ self['creation-time'], algorithm = _struct.unpack(
+ '>IB', data[offset: offset + length])
+ offset += length
+ self['public-key-algorithm'] = self._public_key_algorithms[algorithm]
+ if self['public-key-algorithm'].startswith('rsa '):
+ o, self['public-modulus'] = self._parse_multiprecision_integer(
+ data[offset:])
+ offset += o
+ o, self['public-exponent'] = self._parse_multiprecision_integer(
+ data[offset:])
+ offset += o
+ elif self['public-key-algorithm'].startswith('dsa '):
+ o, self['prime'] = self._parse_multiprecision_integer(
+ data[offset:])
+ offset += o
+ o, self['group-order'] = self._parse_multiprecision_integer(
+ data[offset:])
+ offset += o
+ o, self['group-generator'] = self._parse_multiprecision_integer(
+ data[offset:])
+ offset += o
+ o, self['public-key'] = self._parse_multiprecision_integer(
+ data[offset:])
+ offset += o
+ elif self['public-key-algorithm'].startswith('elgamal '):
+ o, self['prime'] = self._parse_multiprecision_integer(
+ data[offset:])
+ offset += o
+ o, self['group-generator'] = self._parse_multiprecision_integer(
+ data[offset:])
+ offset += o
+ o, self['public-key'] = self._parse_multiprecision_integer(
+ data[offset:])
+ offset += o
+ else:
+ raise NotImplementedError(
+ 'algorithm-specific key fields for {}'.format(
+ self['public-key-algorithm']))
+ fingerprint = _hashlib.sha1()
+ fingerprint.update(b'\x99')
+ fingerprint.update(_struct.pack('>H', len(data)))
+ fingerprint.update(data)
+ self['fingerprint'] = fingerprint.hexdigest()
return offset
+ def _parse_secret_key_packet(self, data):
+ self._parse_generic_secret_key_packet(data=data)
+
+ def _parse_secret_subkey_packet(self, data):
+ self._parse_generic_secret_key_packet(data=data)
+
+ def _parse_generic_secret_key_packet(self, data):
+ offset = self._parse_generic_public_key_packet(data=data)
+ string_to_key_usage = data[offset]
+ offset += 1
+ if string_to_key_usage in [255, 254]:
+ self['symmetric-encryption-algorithm'] = (
+ self._symmetric_key_algorithms[data[offset]])
+ offset += 1
+ offset += self._parse_string_to_key_specifier(data=data[offset:])
+ else:
+ self['symmetric-encryption-algorithm'] = (
+ self._symmetric_key_algorithms[string_to_key_usage])
+ if string_to_key_usage:
+ block_size_bits = self._cipher_block_size.get(
+ self['symmetric-encryption-algorithm'], None)
+ if block_size_bits % 8:
+ raise NotImplementedError(
+ ('{}-bit block size for {} is not an integer number of bytes'
+ ).format(
+ block_size_bits, self['symmetric-encryption-algorithm']))
+ block_size = block_size_bits // 8
+ if not block_size:
+ raise NotImplementedError(
+ 'unknown block size for {}'.format(
+ self['symmetric-encryption-algorithm']))
+ self['initial-vector'] = data[offset: offset + block_size]
+ offset += block_size
+ ciphertext = data[offset:]
+ offset += len(ciphertext)
+ decrypted_data = self.decrypt_symmetric_encryption(data=ciphertext)
+ else:
+ decrypted_data = data[offset:key_end]
+ if string_to_key_usage in [0, 255]:
+ key_end = -2
+ elif string_to_key_usage == 254:
+ key_end = -20
+ else:
+ key_end = 0
+ secret_key = decrypted_data[:key_end]
+ if key_end:
+ secret_key_checksum = decrypted_data[key_end:]
+ if key_end == -2:
+ calculated_checksum = sum(secret_key) % 65536
+ else:
+ checksum_hash = _hashlib.sha1()
+ checksum_hash.update(secret_key)
+ calculated_checksum = checksum_hash.digest()
+ if secret_key_checksum != calculated_checksum:
+ raise ValueError(
+ 'corrupt secret key (checksum {} != expected {})'.format(
+ secret_key_checksum, calculated_checksum))
+ self['secret-key'] = secret_key
+
+ def _parse_signature_subpackets(self, data):
+ offset = 0
+ while offset < len(data):
+ o, subpacket = self._parse_signature_subpacket(data=data[offset:])
+ offset += o
+ yield subpacket
+
+ def _parse_signature_subpacket(self, data):
+ subpacket = {}
+ first = data[0]
+ offset = 1
+ if first < 192:
+ length = first
+ elif first >= 192 and first < 255:
+ second = data[offset]
+ offset += 1
+ length = ((first - 192) << 8) + second + 192
+ else:
+ length = _struct.unpack(
+ '>I', data[offset: offset + 4])[0]
+ offset += 4
+ subpacket['type'] = self._signature_subpacket_types[data[offset]]
+ offset += 1
+ subpacket_data = data[offset: offset + length - 1]
+ offset += len(subpacket_data)
+ method_name = '_parse_{}_signature_subpacket'.format(
+ self._clean_type(type=subpacket['type']))
+ method = getattr(self, method_name, None)
+ if not method:
+ raise NotImplementedError(
+ 'cannot parse signature subpacket type {!r}'.format(
+ subpacket['type']))
+ method(data=subpacket_data, subpacket=subpacket)
+ return (offset, subpacket)
+
+ def _parse_signature_packet(self, data):
+ self['signature-version'] = data[0]
+ offset = 1
+ if self['signature-version'] != 4:
+ raise NotImplementedError(
+ 'signature packet version {}'.format(
+ self['signature-version']))
+ self['signature-type'] = self._signature_types[data[offset]]
+ offset += 1
+ self['public-key-algorithm'] = self._public_key_algorithms[
+ data[offset]]
+ offset += 1
+ self['hash-algorithm'] = self._hash_algorithms[data[offset]]
+ offset += 1
+ hashed_count = _struct.unpack('>H', data[offset: offset + 2])[0]
+ offset += 2
+ self['hashed-subpackets'] = list(self._parse_signature_subpackets(
+ data[offset: offset + hashed_count]))
+ offset += hashed_count
+ unhashed_count = _struct.unpack('>H', data[offset: offset + 2])[0]
+ offset += 2
+ self['unhashed-subpackets'] = list(self._parse_signature_subpackets(
+ data=data[offset: offset + unhashed_count]))
+ offset += unhashed_count
+ self['signed-hash-word'] = data[offset: offset + 2]
+ offset += 2
+ self['signature'] = data[offset:]
+
+ def _parse_signature_creation_time_signature_subpacket(
+ self, data, subpacket):
+ subpacket['signature-creation-time'] = _struct.unpack('>I', data)[0]
+
+ def _parse_issuer_signature_subpacket(self, data, subpacket):
+ subpacket['issuer'] = ''.join('{:02x}'.format(byte) for byte in data)
+
+ def _parse_key_expiration_time_signature_subpacket(
+ self, data, subpacket):
+ subpacket['key-expiration-time'] = _struct.unpack('>I', data)[0]
+
+ def _parse_preferred_symmetric_algorithms_signature_subpacket(
+ self, data, subpacket):
+ subpacket['preferred-symmetric-algorithms'] = [
+ self._symmetric_key_algorithms[d] for d in data]
+
+ def _parse_preferred_hash_algorithms_signature_subpacket(
+ self, data, subpacket):
+ subpacket['preferred-hash-algorithms'] = [
+ self._hash_algorithms[d] for d in data]
+
+ def _parse_preferred_compression_algorithms_signature_subpacket(
+ self, data, subpacket):
+ subpacket['preferred-compression-algorithms'] = [
+ self._compression_algorithms[d] for d in data]
+
+ def _parse_key_server_preferences_signature_subpacket(
+ self, data, subpacket):
+ subpacket['key-server-preferences'] = set()
+ if data[0] & 0x80:
+ subpacket['key-server-preferences'].add('no-modify')
+
+ def _parse_primary_user_id_signature_subpacket(self, data, subpacket):
+ subpacket['primary-user-id'] = bool(data[0])
+
+ def _parse_key_flags_signature_subpacket(self, data, subpacket):
+ subpacket['key-flags'] = set()
+ if data[0] & 0x1:
+ subpacket['key-flags'].add('can certify')
+ if data[0] & 0x2:
+ subpacket['key-flags'].add('can sign')
+ if data[0] & 0x4:
+ subpacket['key-flags'].add('can encrypt communications')
+ if data[0] & 0x8:
+ subpacket['key-flags'].add('can encrypt storage')
+ if data[0] & 0x10:
+ subpacket['key-flags'].add('private split')
+ if data[0] & 0x20:
+ subpacket['key-flags'].add('can authenticate')
+ if data[0] & 0x80:
+ subpacket['key-flags'].add('private shared')
+
+ def _parse_features_signature_subpacket(self, data, subpacket):
+ subpacket['features'] = set()
+ if data[0] & 0x1:
+ subpacket['features'].add('modification detection')
+
+ def _parse_embedded_signature_signature_subpacket(self, data, subpacket):
+ subpacket['embedded'] = PGPPacket()
+ subpacket['embedded']._parse_signature_packet(data=data)
+
+ def _parse_user_id_packet(self, data):
+ self['user'] = str(data, 'utf-8')
+
def to_bytes(self):
- pass
+ method_name = '_serialize_{}'.format(self._clean_type())
+ method = getattr(self, method_name, None)
+ if not method:
+ raise NotImplementedError(
+ 'cannot serialize packet type {!r}'.format(self['type']))
+ body = method()
+ self['length'] = len(body)
+ return b''.join([
+ self._serialize_header(),
+ body,
+ ])
+
+ def _serialize_header(self):
+ always_one = 1
+ new_format = 0
+ type_code = self._reverse(self._packet_types, self['type'])
+ packet_tag = (
+ always_one * (1 << 7) |
+ new_format * (1 << 6) |
+ type_code * (1 << 2) |
+ self['length-type']
+ )
+ length_bytes, length_type = self._old_format_packet_length_type[
+ self['length-type']]
+ length_format = '>{}'.format(length_type)
+ length_data = _struct.pack(length_format, self['length'])
+ return b''.join([
+ bytes([packet_tag]),
+ length_data,
+ ])
+
+ @staticmethod
+ def _serialize_multiprecision_integer(integer):
+ r"""Serialize RFC 4880's multipricision integers
+
+ >>> PGPPacket._serialize_multiprecision_integer(1)
+ b'\x00\x01\x01'
+ >>> PGPPacket._serialize_multiprecision_integer(511)
+ b'\x00\t\x01\xff'
+ """
+ bit_length = int(_math.log(integer, 2)) + 1
+ chunks = [
+ _struct.pack('>H', bit_length),
+ ]
+ while integer > 0:
+ chunks.insert(1, bytes([integer & 0xff]))
+ integer = integer >> 8
+ return b''.join(chunks)
+
+ @classmethod
+ def _encode_string_to_key_count(cls, count):
+ r"""Encode RFC 4880's string-to-key count
+
+ >>> PGPPacket._encode_string_to_key_count(753664)
+ b'\x97'
+ """
+ coded_count = 0
+ count = count >> cls._string_to_key_expbias
+ while not count & 1:
+ count = count >> 1
+ coded_count += 1 << 4
+ coded_count += count & 15
+ return bytes([coded_count])
+
+ def _serialize_string_to_key_specifier(self):
+ string_to_key_type = bytes([
+ self._reverse(
+ self._string_to_key_types, self['string-to-key-type']),
+ ])
+ chunks = [string_to_key_type]
+ if self['string-to-key-type'] == 'simple':
+ chunks.append(bytes([self._reverse(
+ self._hash_algorithms, self['string-to-key-hash-algorithm'])]))
+ elif self['string-to-key-type'] == 'salted':
+ chunks.append(bytes([self._reverse(
+ self._hash_algorithms, self['string-to-key-hash-algorithm'])]))
+ chunks.append(self['string-to-key-salt'])
+ elif self['string-to-key-type'] == 'iterated and salted':
+ chunks.append(bytes([self._reverse(
+ self._hash_algorithms, self['string-to-key-hash-algorithm'])]))
+ chunks.append(self['string-to-key-salt'])
+ chunks.append(self._encode_string_to_key_count(
+ count=self['string-to-key-count']))
+ else:
+ raise NotImplementedError(
+ 'string-to-key type {}'.format(self['string-to-key-type']))
+ return offset
+ return b''.join(chunks)
+
+ def _serialize_public_key_packet(self):
+ return self._serialize_generic_public_key_packet()
+
+ def _serialize_public_subkey_packet(self):
+ return self._serialize_generic_public_key_packet()
+
+ def _serialize_generic_public_key_packet(self):
+ key_version = bytes([self['key-version']])
+ chunks = [key_version]
+ if self['key-version'] != 4:
+ raise NotImplementedError(
+ 'public (sub)key packet version {}'.format(
+ self['key-version']))
+ chunks.append(_struct.pack('>I', self['creation-time']))
+ chunks.append(bytes([self._reverse(
+ self._public_key_algorithms, self['public-key-algorithm'])]))
+ if self['public-key-algorithm'].startswith('rsa '):
+ chunks.append(self._serialize_multiprecision_integer(
+ self['public-modulus']))
+ chunks.append(self._serialize_multiprecision_integer(
+ self['public-exponent']))
+ elif self['public-key-algorithm'].startswith('dsa '):
+ chunks.append(self._serialize_multiprecision_integer(
+ self['prime']))
+ chunks.append(self._serialize_multiprecision_integer(
+ self['group-order']))
+ chunks.append(self._serialize_multiprecision_integer(
+ self['group-generator']))
+ chunks.append(self._serialize_multiprecision_integer(
+ self['public-key']))
+ elif self['public-key-algorithm'].startswith('elgamal '):
+ chunks.append(self._serialize_multiprecision_integer(
+ self['prime']))
+ chunks.append(self._serialize_multiprecision_integer(
+ self['group-generator']))
+ chunks.append(self._serialize_multiprecision_integer(
+ self['public-key']))
+ else:
+ raise NotImplementedError(
+ 'algorithm-specific key fields for {}'.format(
+ self['public-key-algorithm']))
+ return b''.join(chunks)
+
+ def decrypt_symmetric_encryption(self, data):
+ """Decrypt OpenPGP's Cipher Feedback mode"""
+ algorithm = self['symmetric-encryption-algorithm']
+ module = self._crypto_module[algorithm]
+ key_size = self._key_size[algorithm]
+ segment_size_bits = self._cipher_block_size[algorithm]
+ if segment_size_bits % 8:
+ raise NotImplementedError(
+ ('{}-bit segment size for {} is not an integer number of bytes'
+ ).format(segment_size_bits, algorithm))
+ segment_size_bytes = segment_size_bits // 8
+ padding = segment_size_bytes - len(data) % segment_size_bytes
+ if padding:
+ data += b'\x00' * padding
+ passphrase = _getpass.getpass(
+ 'passphrase for {}: '.format(self['fingerprint'][-8:]))
+ passphrase = passphrase.encode('ascii')
+ cipher = module.new(
+ key=passphrase,
+ mode=module.MODE_CFB,
+ IV=self['initial-vector'],
+ segment_size=segment_size_bits)
+ plaintext = cipher.decrypt(data)
+ if padding:
+ plaintext = plaintext[:-padding]
+ return plaintext
def packets_from_bytes(data):
yield packet
+class PGPKey (object):
+ """An OpenPGP key with public and private parts.
+
+ From RFC 4880 [1]:
+
+ OpenPGP users may transfer public keys. The essential elements
+ of a transferable public key are as follows:
+
+ - One Public-Key packet
+ - Zero or more revocation signatures
+ - One or more User ID packets
+ - After each User ID packet, zero or more Signature packets
+ (certifications)
+ - Zero or more User Attribute packets
+ - After each User Attribute packet, zero or more Signature
+ packets (certifications)
+ - Zero or more Subkey packets
+ - After each Subkey packet, one Signature packet, plus
+ optionally a revocation
+
+ Secret keys have a similar packet stream [2]:
+
+ OpenPGP users may transfer secret keys. The format of a
+ transferable secret key is the same as a transferable public key
+ except that secret-key and secret-subkey packets are used
+ instead of the public key and public-subkey packets.
+ Implementations SHOULD include self-signatures on any user IDs
+ and subkeys, as this allows for a complete public key to be
+ automatically extracted from the transferable secret key.
+ Implementations MAY choose to omit the self-signatures,
+ especially if a transferable public key accompanies the
+ transferable secret key.
+
+ [1]: http://tools.ietf.org/search/rfc4880#section-11.1
+ [2]: http://tools.ietf.org/search/rfc4880#section-11.2
+ """
+ def __init__(self, fingerprint):
+ self.fingerprint = fingerprint
+ self.public_packets = None
+ self.secret_packets = None
+
+ def __str__(self):
+ lines = ['key: {}'.format(self.fingerprint)]
+ if self.public_packets:
+ lines.append(' public:')
+ for packet in self.public_packets:
+ lines.extend(self._str_packet(packet=packet, prefix=' '))
+ if self.secret_packets:
+ lines.append(' secret:')
+ for packet in self.secret_packets:
+ lines.extend(self._str_packet(packet=packet, prefix=' '))
+ return '\n'.join(lines)
+
+ def _str_packet(self, packet, prefix):
+ lines = str(packet).split('\n')
+ return [prefix + line for line in lines]
+
+ def import_from_gpg(self):
+ key_export = _get_stdout(
+ ['gpg', '--export', self.fingerprint])
+ self.public_packets = list(
+ packets_from_bytes(data=key_export))
+ if self.public_packets[0]['type'] != 'public-key packet':
+ raise ValueError(
+ '{} does not start with a public-key packet'.format(
+ self.fingerprint))
+ key_secret_export = _get_stdout(
+ ['gpg', '--export-secret-keys', self.fingerprint])
+ self.secret_packets = list(
+ packets_from_bytes(data=key_secret_export))
+ if self.secret_packets[0]['type'] != 'secret-key packet':
+ raise ValueError(
+ '{} does not start with a secret-key packet'.format(
+ self.fingerprint))
+
+ def export_to_gpg(self):
+ raise NotImplemetedError('export to gpg')
+
+ def import_from_key(self, key):
+ """Migrate the (sub)keys into this key"""
+ pass
+
+
def migrate(old_key, new_key):
"""Add the old key and sub-keys to the new key
signatures you'd made. You will lose signature *on* your old key
though, since sub-keys can't be signed (I don't think).
"""
- old_key_export = _get_stdout(
- ['gpg', '--export', old_key])
- old_key_packets = list(
- packets_from_bytes(data=old_key_export))
- old_key_secret_export = _get_stdout(
- ['gpg', '--export-secret-keys', old_key])
- old_key_secret_packets = list(
- packets_from_bytes(data=old_key_secret_export))
-
- import pprint
- pprint.pprint(old_key_packets)
- pprint.pprint(old_key_secret_packets)
+ old_key = PGPKey(fingerprint=old_key)
+ old_key.import_from_gpg()
+ new_key = PGPKey(fingerprint=new_key)
+ new_key.import_from_gpg()
+ new_key.import_from_key(key=old_key)
+
+ print(old_key)
+ print(new_key)
if __name__ == '__main__':
projects / gpg-migrate.git / blobdiff