Add features signature subpacket parsing to PGPPacket
[gpg-migrate.git] / gpg-migrate.py
index 066463be4455c49b8246ced5c4f30c009bca0f55..6055eb105020b88528ce07b113151590aa95ec2e 100755 (executable)
@@ -1,5 +1,6 @@
 #!/usr/bin/python
 
+import hashlib as _hashlib
 import re as _re
 import subprocess as _subprocess
 import struct as _struct
@@ -51,10 +52,272 @@ class PGPPacket (dict):
         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,
+        }
+
+    _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',
+        }
+
+    _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):
-        return self._clean_type_regex.sub('_', self['type'])
+    def _clean_type(self, type=None):
+        if type is None:
+            type = self['type']
+        return self._clean_type_regex.sub('_', type)
+
+    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_issuer_signature_subpacket(self, subpacket):
+        return subpacket['issuer'][-8:].upper()
+
+    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_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)
@@ -96,6 +359,50 @@ class PGPPacket (dict):
         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)
+
+    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-coded-count'] = 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)
 
@@ -110,10 +417,201 @@ class PGPPacket (dict):
                 'public (sub)key packet version {}'.format(
                     self['key-version']))
         length = 5
-        self['creation_time'], self['public-key-algorithm'] = _struct.unpack(
+        self['creation-time'], algorithm = _struct.unpack(
             '>IB', data[offset: offset + length])
         offset += length
-        self['key'] = data[offset:]
+        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
+        if string_to_key_usage in [0, 255]:
+            key_end = -2
+        else:
+            key_end = 0
+        self['secret-key'] = data[offset:key_end]
+        if key_end:
+            self['secret-key-checksum'] = data[key_end:]
+
+    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_issuer_signature_subpacket(self, data, subpacket):
+        subpacket['issuer'] = ''.join('{:02x}'.format(byte) for byte in data)
+
+    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_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
@@ -127,6 +625,50 @@ def packets_from_bytes(data):
         yield packet
 
 
+class PGPKey (object):
+    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))
+
+    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
 
@@ -134,22 +676,14 @@ def migrate(old_key, 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))
-    if old_key_packets[0]['type'] != 'public-key packet':
-        raise ValueError(
-            '{} does not start with a public-key packet'.format(
-                old_key))
-    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__':