gpg-migrate.py

   1 #!/usr/bin/python
   2
   3 import hashlib as _hashlib
   4 import re as _re
   5 import subprocess as _subprocess
   6 import struct as _struct
   7
   8
   9 def _get_stdout(args, stdin=None):
  10     stdin_pipe = None
  11     if stdin is not None:
  12         stdin_pipe = _subprocess.PIPE
  13     p = _subprocess.Popen(args, stdin=stdin_pipe, stdout=_subprocess.PIPE)
  14     stdout, stderr = p.communicate(stdin)
  15     status = p.wait()
  16     if status != 0:
  17         raise RuntimeError(status)
  18     return stdout
  19
  20
  21 class PGPPacket (dict):
  22     # http://tools.ietf.org/search/rfc4880
  23     _old_format_packet_length_type = {  # type: (bytes, struct type)
  24         0: (1, 'B'),  # 1-byte unsigned integer
  25         1: (2, 'H'),  # 2-byte unsigned integer
  26         2: (4, 'I'),  # 4-byte unsigned integer
  27         3: (None, None),
  28         }
  29
  30     _packet_types = {
  31         0: 'reserved',
  32         1: 'public-key encrypted session key packet',
  33         2: 'signature packet',
  34         3: 'symmetric-key encrypted session key packet',
  35         4: 'one-pass signature packet',
  36         5: 'secret-key packet',
  37         6: 'public-key packet',
  38         7: 'secret-subkey packet',
  39         8: 'compressed data packet',
  40         9: 'symmetrically encrypted data packet',
  41         10: 'marker packet',
  42         11: 'literal data packet',
  43         12: 'trust packet',
  44         13: 'user id packet',
  45         14: 'public-subkey packet',
  46         17: 'user attribute packet',
  47         18: 'sym. encrypted and integrity protected data packet',
  48         19: 'modification detection code packet',
  49         60: 'private',
  50         61: 'private',
  51         62: 'private',
  52         63: 'private',
  53         }
  54
  55     _public_key_algorithms = {
  56         1: 'rsa (encrypt or sign)',
  57         2: 'rsa encrypt-only',
  58         3: 'rsa sign-only',
  59         16: 'elgamal (encrypt-only)',
  60         17: 'dsa (digital signature algorithm)',
  61         18: 'reserved for elliptic curve',
  62         19: 'reserved for ecdsa',
  63         20: 'reserved (formerly elgamal encrypt or sign)',
  64         21: 'reserved for diffie-hellman',
  65         100: 'private',
  66         101: 'private',
  67         102: 'private',
  68         103: 'private',
  69         104: 'private',
  70         105: 'private',
  71         106: 'private',
  72         107: 'private',
  73         108: 'private',
  74         109: 'private',
  75         110: 'private',
  76         }
  77
  78     _symmetric_key_algorithms = {
  79         0: 'plaintext or unencrypted data',
  80         1: 'idea',
  81         2: 'tripledes',
  82         3: 'cast5',
  83         4: 'blowfish',
  84         5: 'reserved',
  85         6: 'reserved',
  86         7: 'aes with 128-bit key',
  87         8: 'aes with 192-bit key',
  88         9: 'aes with 256-bit key',
  89         10: 'twofish',
  90         100: 'private',
  91         101: 'private',
  92         102: 'private',
  93         103: 'private',
  94         104: 'private',
  95         105: 'private',
  96         106: 'private',
  97         107: 'private',
  98         108: 'private',
  99         109: 'private',
 100         110: 'private',
 101         }
 102
 103     _cipher_block_size = {  # in bits
 104         'aes with 128-bit key': 128,
 105         'aes with 192-bit key': 128,
 106         'aes with 256-bit key': 128,
 107         'cast5': 64,
 108         }
 109
 110     _compression_algorithms = {
 111         0: 'uncompressed',
 112         1: 'zip',
 113         2: 'zlib',
 114         3: 'bzip2',
 115         100: 'private',
 116         101: 'private',
 117         102: 'private',
 118         103: 'private',
 119         104: 'private',
 120         105: 'private',
 121         106: 'private',
 122         107: 'private',
 123         108: 'private',
 124         109: 'private',
 125         110: 'private',
 126         }
 127
 128     _hash_algorithms = {
 129         1: 'md5',
 130         2: 'sha-1',
 131         3: 'ripe-md/160',
 132         4: 'reserved',
 133         5: 'reserved',
 134         6: 'reserved',
 135         7: 'reserved',
 136         8: 'sha256',
 137         9: 'sha384',
 138         10: 'sha512',
 139         11: 'sha224',
 140         100: 'private',
 141         101: 'private',
 142         102: 'private',
 143         103: 'private',
 144         104: 'private',
 145         105: 'private',
 146         106: 'private',
 147         107: 'private',
 148         108: 'private',
 149         109: 'private',
 150         110: 'private',
 151         }
 152
 153     _string_to_key_types = {
 154         0: 'simple',
 155         1: 'salted',
 156         2: 'reserved',
 157         3: 'iterated and salted',
 158         100: 'private',
 159         101: 'private',
 160         102: 'private',
 161         103: 'private',
 162         104: 'private',
 163         105: 'private',
 164         106: 'private',
 165         107: 'private',
 166         108: 'private',
 167         109: 'private',
 168         110: 'private',
 169         }
 170
 171     _signature_types = {
 172         0x00: 'binary document',
 173         0x01: 'canonical text document',
 174         0x02: 'standalone',
 175         0x10: 'generic user id and public-key packet',
 176         0x11: 'persona user id and public-key packet',
 177         0x12: 'casual user id and public-key packet',
 178         0x13: 'postitive user id and public-key packet',
 179         0x18: 'subkey binding',
 180         0x19: 'primary key binding',
 181         0x1F: 'direct key',
 182         0x20: 'key revocation',
 183         0x28: 'subkey revocation',
 184         0x30: 'certification revocation',
 185         0x40: 'timestamp',
 186         0x50: 'third-party confirmation',
 187         }
 188
 189     _signature_subpacket_types = {
 190         0: 'reserved',
 191         1: 'reserved',
 192         2: 'signature creation time',
 193         3: 'signature expiration time',
 194         4: 'exportable certification',
 195         5: 'trust signature',
 196         6: 'regular expression',
 197         7: 'revocable',
 198         8: 'reserved',
 199         9: 'key expiration time',
 200         10: 'placeholder for backward compatibility',
 201         11: 'preferred symmetric algorithms',
 202         12: 'revocation key',
 203         13: 'reserved',
 204         14: 'reserved',
 205         15: 'reserved',
 206         16: 'issuer',
 207         17: 'reserved',
 208         18: 'reserved',
 209         19: 'reserved',
 210         20: 'notation data',
 211         21: 'preferred hash algorithms',
 212         22: 'preferred compression algorithms',
 213         23: 'key server preferences',
 214         24: 'preferred key server',
 215         25: 'primary user id',
 216         26: 'policy uri',
 217         27: 'key flags',
 218         28: 'signer user id',
 219         29: 'reason for revocation',
 220         30: 'features',
 221         31: 'signature target',
 222         32: 'embedded signature',
 223         100: 'private',
 224         101: 'private',
 225         102: 'private',
 226         103: 'private',
 227         104: 'private',
 228         105: 'private',
 229         106: 'private',
 230         107: 'private',
 231         108: 'private',
 232         109: 'private',
 233         110: 'private',
 234         }
 235
 236     _clean_type_regex = _re.compile('\W+')
 237
 238     def _clean_type(self, type=None):
 239         if type is None:
 240             type = self['type']
 241         return self._clean_type_regex.sub('_', type)
 242
 243     def __str__(self):
 244         method_name = '_str_{}'.format(self._clean_type())
 245         method = getattr(self, method_name, None)
 246         if not method:
 247             return self['type']
 248         details = method()
 249         return '{}: {}'.format(self['type'], details)
 250
 251     def _str_public_key_packet(self):
 252         return self._str_generic_key_packet()
 253
 254     def _str_public_subkey_packet(self):
 255         return self._str_generic_key_packet()
 256
 257     def _str_secret_key_packet(self):
 258         return self._str_generic_key_packet()
 259
 260     def _str_secret_subkey_packet(self):
 261         return self._str_generic_key_packet()
 262
 263     def _str_generic_key_packet(self):
 264         return self['fingerprint'][-8:].upper()
 265
 266     def _str_signature_packet(self):
 267         lines = [self['signature-type']]
 268         if self['hashed-subpackets']:
 269             lines.append('  hashed subpackets:')
 270             lines.extend(self._str_signature_subpackets(
 271                 self['hashed-subpackets'], prefix='    '))
 272         if self['unhashed-subpackets']:
 273             lines.append('  unhashed subpackets:')
 274             lines.extend(self._str_signature_subpackets(
 275                 self['unhashed-subpackets'], prefix='    '))
 276         return '\n'.join(lines)
 277
 278     def _str_signature_subpackets(self, subpackets, prefix):
 279         lines = []
 280         for subpacket in subpackets:
 281             method_name = '_str_{}_signature_subpacket'.format(
 282                 self._clean_type(type=subpacket['type']))
 283             method = getattr(self, method_name, None)
 284             if method:
 285                 lines.append('    {}: {}'.format(
 286                     subpacket['type'],
 287                     method(subpacket=subpacket)))
 288             else:
 289                 lines.append('    {}'.format(subpacket['type']))
 290         return lines
 291
 292     def _str_issuer_signature_subpacket(self, subpacket):
 293         return subpacket['issuer'][-8:].upper()
 294
 295     def _str_key_flags_signature_subpacket(self, subpacket):
 296         return ', '.join(x for x in sorted(subpacket['key-flags']))
 297
 298     def _str_embedded_signature_signature_subpacket(self, subpacket):
 299         return subpacket['embedded']['signature-type']
 300
 301     def _str_user_id_packet(self):
 302         return self['user']
 303
 304     def from_bytes(self, data):
 305         offset = self._parse_header(data=data)
 306         packet = data[offset:offset + self['length']]
 307         if len(packet) < self['length']:
 308             raise ValueError('packet too short ({} < {})'.format(
 309                 len(packet), self['length']))
 310         offset += self['length']
 311         method_name = '_parse_{}'.format(self._clean_type())
 312         method = getattr(self, method_name, None)
 313         if not method:
 314             raise NotImplementedError(
 315                 'cannot parse packet type {!r}'.format(self['type']))
 316         method(data=packet)
 317         return offset
 318
 319     def _parse_header(self, data):
 320         packet_tag = data[0]
 321         offset = 1
 322         always_one = packet_tag & 1 << 7
 323         if not always_one:
 324             raise ValueError('most significant packet tag bit not set')
 325         self['new-format'] = packet_tag & 1 << 6
 326         if self['new-format']:
 327             type_code = packet_tag & 0b111111
 328             raise NotImplementedError('new-format packet length')
 329         else:
 330             type_code = packet_tag >> 2 & 0b1111
 331             self['length-type'] = packet_tag & 0b11
 332             length_bytes, length_type = self._old_format_packet_length_type[
 333                 self['length-type']]
 334             if not length_bytes:
 335                 raise NotImplementedError(
 336                     'old-format packet of indeterminate length')
 337             length_format = '>{}'.format(length_type)
 338             length_data = data[offset: offset + length_bytes]
 339             offset += length_bytes
 340             self['length'] = _struct.unpack(length_format, length_data)[0]
 341         self['type'] = self._packet_types[type_code]
 342         return offset
 343
 344     @staticmethod
 345     def _parse_multiprecision_integer(data):
 346         r"""Parse RFC 4880's multiprecision integers
 347
 348         >>> PGPPacket._parse_multiprecision_integer(b'\x00\x01\x01')
 349         (3, 1)
 350         >>> PGPPacket._parse_multiprecision_integer(b'\x00\x09\x01\xff')
 351         (4, 511)
 352         """
 353         bits = _struct.unpack('>H', data[:2])[0]
 354         offset = 2
 355         length = (bits + 7) // 8
 356         value = 0
 357         for i in range(length):
 358             value += data[offset + i] * 1 << (8 * (length - i - 1))
 359         offset += length
 360         return (offset, value)
 361
 362     def _parse_string_to_key_specifier(self, data):
 363         self['string-to-key-type'] = self._string_to_key_types[data[0]]
 364         offset = 1
 365         if self['string-to-key-type'] == 'simple':
 366             self['string-to-key-hash-algorithm'] = self._hash_algorithms[
 367                 data[offset]]
 368             offset += 1
 369         elif self['string-to-key-type'] == 'salted':
 370             self['string-to-key-hash-algorithm'] = self._hash_algorithms[
 371                 data[offset]]
 372             offset += 1
 373             self['string-to-key-salt'] = data[offset: offset + 8]
 374             offset += 8
 375         elif self['string-to-key-type'] == 'iterated and salted':
 376             self['string-to-key-hash-algorithm'] = self._hash_algorithms[
 377                 data[offset]]
 378             offset += 1
 379             self['string-to-key-salt'] = data[offset: offset + 8]
 380             offset += 8
 381             self['string-to-key-coded-count'] = data[offset]
 382             offset += 1
 383         else:
 384             raise NotImplementedError(
 385                 'string-to-key type {}'.format(self['string-to-key-type']))
 386         return offset
 387
 388     def _parse_public_key_packet(self, data):
 389         self._parse_generic_public_key_packet(data=data)
 390
 391     def _parse_public_subkey_packet(self, data):
 392         self._parse_generic_public_key_packet(data=data)
 393
 394     def _parse_generic_public_key_packet(self, data):
 395         self['key-version'] = data[0]
 396         offset = 1
 397         if self['key-version'] != 4:
 398             raise NotImplementedError(
 399                 'public (sub)key packet version {}'.format(
 400                     self['key-version']))
 401         length = 5
 402         self['creation-time'], algorithm = _struct.unpack(
 403             '>IB', data[offset: offset + length])
 404         offset += length
 405         self['public-key-algorithm'] = self._public_key_algorithms[algorithm]
 406         if self['public-key-algorithm'].startswith('rsa '):
 407             o, self['public-modulus'] = self._parse_multiprecision_integer(
 408                 data[offset:])
 409             offset += o
 410             o, self['public-exponent'] = self._parse_multiprecision_integer(
 411                 data[offset:])
 412             offset += o
 413         elif self['public-key-algorithm'].startswith('dsa '):
 414             o, self['prime'] = self._parse_multiprecision_integer(
 415                 data[offset:])
 416             offset += o
 417             o, self['group-order'] = self._parse_multiprecision_integer(
 418                 data[offset:])
 419             offset += o
 420             o, self['group-generator'] = self._parse_multiprecision_integer(
 421                 data[offset:])
 422             offset += o
 423             o, self['public-key'] = self._parse_multiprecision_integer(
 424                 data[offset:])
 425             offset += o
 426         elif self['public-key-algorithm'].startswith('elgamal '):
 427             o, self['prime'] = self._parse_multiprecision_integer(
 428                 data[offset:])
 429             offset += o
 430             o, self['group-generator'] = self._parse_multiprecision_integer(
 431                 data[offset:])
 432             offset += o
 433             o, self['public-key'] = self._parse_multiprecision_integer(
 434                 data[offset:])
 435             offset += o
 436         else:
 437             raise NotImplementedError(
 438                 'algorithm-specific key fields for {}'.format(
 439                     self['public-key-algorithm']))
 440         fingerprint = _hashlib.sha1()
 441         fingerprint.update(b'\x99')
 442         fingerprint.update(_struct.pack('>H', len(data)))
 443         fingerprint.update(data)
 444         self['fingerprint'] = fingerprint.hexdigest()
 445         return offset
 446
 447     def _parse_secret_key_packet(self, data):
 448         self._parse_generic_secret_key_packet(data=data)
 449
 450     def _parse_secret_subkey_packet(self, data):
 451         self._parse_generic_secret_key_packet(data=data)
 452
 453     def _parse_generic_secret_key_packet(self, data):
 454         offset = self._parse_generic_public_key_packet(data=data)
 455         string_to_key_usage = data[offset]
 456         offset += 1
 457         if string_to_key_usage in [255, 254]:
 458             self['symmetric-encryption-algorithm'] = (
 459                 self._symmetric_key_algorithms[data[offset]])
 460             offset += 1
 461             offset += self._parse_string_to_key_specifier(data=data[offset:])
 462         else:
 463             self['symmetric-encryption-algorithm'] = (
 464                 self._symmetric_key_algorithms[string_to_key_usage])
 465         if string_to_key_usage:
 466             block_size_bits = self._cipher_block_size.get(
 467                 self['symmetric-encryption-algorithm'], None)
 468             if block_size_bits % 8:
 469                 raise NotImplementedError(
 470                     ('{}-bit block size for {} is not an integer number of bytes'
 471                      ).format(
 472                          block_size_bits, self['symmetric-encryption-algorithm']))
 473             block_size = block_size_bits // 8
 474             if not block_size:
 475                 raise NotImplementedError(
 476                     'unknown block size for {}'.format(
 477                         self['symmetric-encryption-algorithm']))
 478             self['initial-vector'] = data[offset: offset + block_size]
 479             offset += block_size
 480         if string_to_key_usage in [0, 255]:
 481             key_end = -2
 482         else:
 483             key_end = 0
 484         self['secret-key'] = data[offset:key_end]
 485         if key_end:
 486             self['secret-key-checksum'] = data[key_end:]
 487
 488     def _parse_signature_subpackets(self, data):
 489         offset = 0
 490         while offset < len(data):
 491             o, subpacket = self._parse_signature_subpacket(data=data[offset:])
 492             offset += o
 493             yield subpacket
 494
 495     def _parse_signature_subpacket(self, data):
 496         subpacket = {}
 497         first = data[0]
 498         offset = 1
 499         if first < 192:
 500             length = first
 501         elif first >= 192 and first < 255:
 502             second = data[offset]
 503             offset += 1
 504             length = ((first - 192) << 8) + second + 192
 505         else:
 506             length = _struct.unpack(
 507                 '>I', data[offset: offset + 4])[0]
 508             offset += 4
 509         subpacket['type'] = self._signature_subpacket_types[data[offset]]
 510         offset += 1
 511         subpacket_data = data[offset: offset + length - 1]
 512         offset += len(subpacket_data)
 513         method_name = '_parse_{}_signature_subpacket'.format(
 514             self._clean_type(type=subpacket['type']))
 515         method = getattr(self, method_name, None)
 516         if not method:
 517             raise NotImplementedError(
 518                 'cannot parse signature subpacket type {!r}'.format(
 519                     subpacket['type']))
 520         method(data=subpacket_data, subpacket=subpacket)
 521         return (offset, subpacket)
 522
 523     def _parse_signature_packet(self, data):
 524         self['signature-version'] = data[0]
 525         offset = 1
 526         if self['signature-version'] != 4:
 527             raise NotImplementedError(
 528                 'signature packet version {}'.format(
 529                     self['signature-version']))
 530         self['signature-type'] = self._signature_types[data[offset]]
 531         offset += 1
 532         self['public-key-algorithm'] = self._public_key_algorithms[
 533             data[offset]]
 534         offset += 1
 535         self['hash-algorithm'] = self._hash_algorithms[data[offset]]
 536         offset += 1
 537         hashed_count = _struct.unpack('>H', data[offset: offset + 2])[0]
 538         offset += 2
 539         self['hashed-subpackets'] = list(self._parse_signature_subpackets(
 540             data[offset: offset + hashed_count]))
 541         offset += hashed_count
 542         unhashed_count = _struct.unpack('>H', data[offset: offset + 2])[0]
 543         offset += 2
 544         self['unhashed-subpackets'] = list(self._parse_signature_subpackets(
 545             data=data[offset: offset + unhashed_count]))
 546         offset += unhashed_count
 547         self['signed-hash-word'] = data[offset: offset + 2]
 548         offset += 2
 549         self['signature'] = data[offset:]
 550
 551     def _parse_issuer_signature_subpacket(self, data, subpacket):
 552         subpacket['issuer'] = ''.join('{:02x}'.format(byte) for byte in data)
 553
 554     def _parse_key_flags_signature_subpacket(self, data, subpacket):
 555         subpacket['key-flags'] = set()
 556         if data[0] & 0x1:
 557             subpacket['key-flags'].add('can certify')
 558         if data[0] & 0x2:
 559             subpacket['key-flags'].add('can sign')
 560         if data[0] & 0x4:
 561             subpacket['key-flags'].add('can encrypt communications')
 562         if data[0] & 0x8:
 563             subpacket['key-flags'].add('can encrypt storage')
 564         if data[0] & 0x10:
 565             subpacket['key-flags'].add('private split')
 566         if data[0] & 0x20:
 567             subpacket['key-flags'].add('can authenticate')
 568         if data[0] & 0x80:
 569             subpacket['key-flags'].add('private shared')
 570
 571     def _parse_embedded_signature_signature_subpacket(self, data, subpacket):
 572         subpacket['embedded'] = PGPPacket()
 573         subpacket['embedded']._parse_signature_packet(data=data)
 574
 575     def _parse_user_id_packet(self, data):
 576         self['user'] = str(data, 'utf-8')
 577
 578     def to_bytes(self):
 579         pass
 580
 581
 582 def packets_from_bytes(data):
 583     offset = 0
 584     while offset < len(data):
 585         packet = PGPPacket()
 586         offset += packet.from_bytes(data=data[offset:])
 587         yield packet
 588
 589
 590 class PGPKey (object):
 591     def __init__(self, fingerprint):
 592         self.fingerprint = fingerprint
 593         self.public_packets = None
 594         self.secret_packets = None
 595
 596     def __str__(self):
 597         lines = ['key: {}'.format(self.fingerprint)]
 598         if self.public_packets:
 599             lines.append('  public:')
 600             for packet in self.public_packets:
 601                 lines.extend(self._str_packet(packet=packet, prefix='    '))
 602         if self.secret_packets:
 603             lines.append('  secret:')
 604             for packet in self.secret_packets:
 605                 lines.extend(self._str_packet(packet=packet, prefix='    '))
 606         return '\n'.join(lines)
 607
 608     def _str_packet(self, packet, prefix):
 609         lines = str(packet).split('\n')
 610         return [prefix + line for line in lines]
 611
 612     def import_from_gpg(self):
 613         key_export = _get_stdout(
 614             ['gpg', '--export', self.fingerprint])
 615         self.public_packets = list(
 616             packets_from_bytes(data=key_export))
 617         if self.public_packets[0]['type'] != 'public-key packet':
 618             raise ValueError(
 619                 '{} does not start with a public-key packet'.format(
 620                     self.fingerprint))
 621         key_secret_export = _get_stdout(
 622             ['gpg', '--export-secret-keys', self.fingerprint])
 623         self.secret_packets = list(
 624             packets_from_bytes(data=key_secret_export))
 625
 626     def export_to_gpg(self):
 627         raise NotImplemetedError('export to gpg')
 628
 629     def import_from_key(self, key):
 630         """Migrate the (sub)keys into this key"""
 631         pass
 632
 633
 634 def migrate(old_key, new_key):
 635     """Add the old key and sub-keys to the new key
 636
 637     For example, to upgrade your master key, while preserving old
 638     signatures you'd made.  You will lose signature *on* your old key
 639     though, since sub-keys can't be signed (I don't think).
 640     """
 641     old_key = PGPKey(fingerprint=old_key)
 642     old_key.import_from_gpg()
 643     new_key = PGPKey(fingerprint=new_key)
 644     new_key.import_from_gpg()
 645     new_key.import_from_key(key=old_key)
 646
 647     print(old_key)
 648     print(new_key)
 649
 650
 651 if __name__ == '__main__':
 652     import sys as _sys
 653
 654     old_key, new_key = _sys.argv[1:3]
 655     migrate(old_key=old_key, new_key=new_key)