Add version-4-signature parsing to PGPPacket
[gpg-migrate.git] / gpg-migrate.py
1 #!/usr/bin/python
2
3 import re as _re
4 import subprocess as _subprocess
5 import struct as _struct
6
7
8 def _get_stdout(args, stdin=None):
9     stdin_pipe = None
10     if stdin is not None:
11         stdin_pipe = _subprocess.PIPE
12     p = _subprocess.Popen(args, stdin=stdin_pipe, stdout=_subprocess.PIPE)
13     stdout, stderr = p.communicate(stdin)
14     status = p.wait()
15     if status != 0:
16         raise RuntimeError(status)
17     return stdout
18
19
20 class PGPPacket (dict):
21     # http://tools.ietf.org/search/rfc4880
22     _old_format_packet_length_type = {  # type: (bytes, struct type)
23         0: (1, 'B'),  # 1-byte unsigned integer
24         1: (2, 'H'),  # 2-byte unsigned integer
25         2: (4, 'I'),  # 4-byte unsigned integer
26         3: (None, None),
27         }
28
29     _packet_types = {
30         0: 'reserved',
31         1: 'public-key encrypted session key packet',
32         2: 'signature packet',
33         3: 'symmetric-key encrypted session key packet',
34         4: 'one-pass signature packet',
35         5: 'secret-key packet',
36         6: 'public-key packet',
37         7: 'secret-subkey packet',
38         8: 'compressed data packet',
39         9: 'symmetrically encrypted data packet',
40         10: 'marker packet',
41         11: 'literal data packet',
42         12: 'trust packet',
43         13: 'user id packet',
44         14: 'public-subkey packet',
45         17: 'user attribute packet',
46         18: 'sym. encrypted and integrity protected data packet',
47         19: 'modification detection code packet',
48         60: 'private',
49         61: 'private',
50         62: 'private',
51         63: 'private',
52         }
53
54     _public_key_algorithms = {
55         1: 'rsa (encrypt or sign)',
56         2: 'rsa encrypt-only',
57         3: 'rsa sign-only',
58         16: 'elgamal (encrypt-only)',
59         17: 'dsa (digital signature algorithm)',
60         18: 'reserved for elliptic curve',
61         19: 'reserved for ecdsa',
62         20: 'reserved (formerly elgamal encrypt or sign)',
63         21: 'reserved for diffie-hellman',
64         100: 'private',
65         101: 'private',
66         102: 'private',
67         103: 'private',
68         104: 'private',
69         105: 'private',
70         106: 'private',
71         107: 'private',
72         108: 'private',
73         109: 'private',
74         110: 'private',
75         }
76
77     _symmetric_key_algorithms = {
78         0: 'plaintext or unencrypted data',
79         1: 'idea',
80         2: 'tripledes',
81         3: 'cast5',
82         4: 'blowfish',
83         5: 'reserved',
84         6: 'reserved',
85         7: 'aes with 128-bit key',
86         8: 'aes with 192-bit key',
87         9: 'aes with 256-bit key',
88         10: 'twofish',
89         100: 'private',
90         101: 'private',
91         102: 'private',
92         103: 'private',
93         104: 'private',
94         105: 'private',
95         106: 'private',
96         107: 'private',
97         108: 'private',
98         109: 'private',
99         110: 'private',
100         }
101
102     _cipher_block_size = {  # in bits
103         'aes with 128-bit key': 128,
104         'aes with 192-bit key': 128,
105         'aes with 256-bit key': 128,
106         'cast5': 64,
107         }
108
109     _compression_algorithms = {
110         0: 'uncompressed',
111         1: 'zip',
112         2: 'zlib',
113         3: 'bzip2',
114         100: 'private',
115         101: 'private',
116         102: 'private',
117         103: 'private',
118         104: 'private',
119         105: 'private',
120         106: 'private',
121         107: 'private',
122         108: 'private',
123         109: 'private',
124         110: 'private',
125         }
126
127     _hash_algorithms = {
128         1: 'md5',
129         2: 'sha-1',
130         3: 'ripe-md/160',
131         4: 'reserved',
132         5: 'reserved',
133         6: 'reserved',
134         7: 'reserved',
135         8: 'sha256',
136         9: 'sha384',
137         10: 'sha512',
138         11: 'sha224',
139         100: 'private',
140         101: 'private',
141         102: 'private',
142         103: 'private',
143         104: 'private',
144         105: 'private',
145         106: 'private',
146         107: 'private',
147         108: 'private',
148         109: 'private',
149         110: 'private',
150         }
151
152     _string_to_key_types = {
153         0: 'simple',
154         1: 'salted',
155         2: 'reserved',
156         3: 'iterated and salted',
157         100: 'private',
158         101: 'private',
159         102: 'private',
160         103: 'private',
161         104: 'private',
162         105: 'private',
163         106: 'private',
164         107: 'private',
165         108: 'private',
166         109: 'private',
167         110: 'private',
168         }
169
170     _signature_types = {
171         0x00: 'binary document',
172         0x01: 'canonical text document',
173         0x02: 'standalone',
174         0x10: 'generic user id and public-key packet',
175         0x11: 'persona user id and public-key packet',
176         0x12: 'casual user id and public-key packet',
177         0x13: 'postitive user id and public-key packet',
178         0x18: 'subkey binding',
179         0x19: 'primary key binding',
180         0x1F: 'direct key',
181         0x20: 'key revocation',
182         0x28: 'subkey revocation',
183         0x30: 'certification revocation',
184         0x40: 'timestamp',
185         0x50: 'third-party confirmation',
186         }
187
188     _clean_type_regex = _re.compile('\W+')
189
190     def _clean_type(self):
191         return self._clean_type_regex.sub('_', self['type'])
192
193     def from_bytes(self, data):
194         offset = self._parse_header(data=data)
195         packet = data[offset:offset + self['length']]
196         if len(packet) < self['length']:
197             raise ValueError('packet too short ({} < {})'.format(
198                 len(packet), self['length']))
199         offset += self['length']
200         method_name = '_parse_{}'.format(self._clean_type())
201         method = getattr(self, method_name, None)
202         if not method:
203             raise NotImplementedError(
204                 'cannot parse packet type {!r}'.format(self['type']))
205         method(data=packet)
206         return offset
207
208     def _parse_header(self, data):
209         packet_tag = data[0]
210         offset = 1
211         always_one = packet_tag & 1 << 7
212         if not always_one:
213             raise ValueError('most significant packet tag bit not set')
214         self['new-format'] = packet_tag & 1 << 6
215         if self['new-format']:
216             type_code = packet_tag & 0b111111
217             raise NotImplementedError('new-format packet length')
218         else:
219             type_code = packet_tag >> 2 & 0b1111
220             self['length-type'] = packet_tag & 0b11
221             length_bytes, length_type = self._old_format_packet_length_type[
222                 self['length-type']]
223             if not length_bytes:
224                 raise NotImplementedError(
225                     'old-format packet of indeterminate length')
226             length_format = '>{}'.format(length_type)
227             length_data = data[offset: offset + length_bytes]
228             offset += length_bytes
229             self['length'] = _struct.unpack(length_format, length_data)[0]
230         self['type'] = self._packet_types[type_code]
231         return offset
232
233     @staticmethod
234     def _parse_multiprecision_integer(data):
235         r"""Parse RFC 4880's multiprecision integers
236
237         >>> PGPPacket._parse_multiprecision_integer(b'\x00\x01\x01')
238         (3, 1)
239         >>> PGPPacket._parse_multiprecision_integer(b'\x00\x09\x01\xff')
240         (4, 511)
241         """
242         bits = _struct.unpack('>H', data[:2])[0]
243         offset = 2
244         length = (bits + 7) // 8
245         value = 0
246         for i in range(length):
247             value += data[offset + i] * 1 << (8 * (length - i - 1))
248         offset += length
249         return (offset, value)
250
251     def _parse_string_to_key_specifier(self, data):
252         self['string-to-key-type'] = self._string_to_key_types[data[0]]
253         offset = 1
254         if self['string-to-key-type'] == 'simple':
255             self['string-to-key-hash-algorithm'] = self._hash_algorithms[
256                 data[offset]]
257             offset += 1
258         elif self['string-to-key-type'] == 'salted':
259             self['string-to-key-hash-algorithm'] = self._hash_algorithms[
260                 data[offset]]
261             offset += 1
262             self['string-to-key-salt'] = data[offset: offset + 8]
263             offset += 8
264         elif self['string-to-key-type'] == 'iterated and salted':
265             self['string-to-key-hash-algorithm'] = self._hash_algorithms[
266                 data[offset]]
267             offset += 1
268             self['string-to-key-salt'] = data[offset: offset + 8]
269             offset += 8
270             self['string-to-key-coded-count'] = data[offset]
271             offset += 1
272         else:
273             raise NotImplementedError(
274                 'string-to-key type {}'.format(self['string-to-key-type']))
275         return offset
276
277     def _parse_public_key_packet(self, data):
278         self._parse_generic_public_key_packet(data=data)
279
280     def _parse_public_subkey_packet(self, data):
281         self._parse_generic_public_key_packet(data=data)
282
283     def _parse_generic_public_key_packet(self, data):
284         self['key-version'] = data[0]
285         offset = 1
286         if self['key-version'] != 4:
287             raise NotImplementedError(
288                 'public (sub)key packet version {}'.format(
289                     self['key-version']))
290         length = 5
291         self['creation-time'], algorithm = _struct.unpack(
292             '>IB', data[offset: offset + length])
293         offset += length
294         self['public-key-algorithm'] = self._public_key_algorithms[algorithm]
295         if self['public-key-algorithm'].startswith('rsa '):
296             o, self['public-modulus'] = self._parse_multiprecision_integer(
297                 data[offset:])
298             offset += o
299             o, self['public-exponent'] = self._parse_multiprecision_integer(
300                 data[offset:])
301             offset += o
302         elif self['public-key-algorithm'].startswith('dsa '):
303             o, self['prime'] = self._parse_multiprecision_integer(
304                 data[offset:])
305             offset += o
306             o, self['group-order'] = self._parse_multiprecision_integer(
307                 data[offset:])
308             offset += o
309             o, self['group-generator'] = self._parse_multiprecision_integer(
310                 data[offset:])
311             offset += o
312             o, self['public-key'] = self._parse_multiprecision_integer(
313                 data[offset:])
314             offset += o
315         elif self['public-key-algorithm'].startswith('elgamal '):
316             o, self['prime'] = self._parse_multiprecision_integer(
317                 data[offset:])
318             offset += o
319             o, self['group-generator'] = self._parse_multiprecision_integer(
320                 data[offset:])
321             offset += o
322             o, self['public-key'] = self._parse_multiprecision_integer(
323                 data[offset:])
324             offset += o
325         else:
326             raise NotImplementedError(
327                 'algorithm-specific key fields for {}'.format(
328                     self['public-key-algorithm']))
329         return offset
330
331     def _parse_secret_key_packet(self, data):
332         self._parse_generic_secret_key_packet(data=data)
333
334     def _parse_secret_subkey_packet(self, data):
335         self._parse_generic_secret_key_packet(data=data)
336
337     def _parse_generic_secret_key_packet(self, data):
338         offset = self._parse_generic_public_key_packet(data=data)
339         string_to_key_usage = data[offset]
340         offset += 1
341         if string_to_key_usage in [255, 254]:
342             self['symmetric-encryption-algorithm'] = (
343                 self._symmetric_key_algorithms[data[offset]])
344             offset += 1
345             offset += self._parse_string_to_key_specifier(data=data[offset:])
346         else:
347             self['symmetric-encryption-algorithm'] = (
348                 self._symmetric_key_algorithms[string_to_key_usage])
349         if string_to_key_usage:
350             block_size_bits = self._cipher_block_size.get(
351                 self['symmetric-encryption-algorithm'], None)
352             if block_size_bits % 8:
353                 raise NotImplementedError(
354                     ('{}-bit block size for {} is not an integer number of bytes'
355                      ).format(
356                          block_size_bits, self['symmetric-encryption-algorithm']))
357             block_size = block_size_bits // 8
358             if not block_size:
359                 raise NotImplementedError(
360                     'unknown block size for {}'.format(
361                         self['symmetric-encryption-algorithm']))
362             self['initial-vector'] = data[offset: offset + block_size]
363             offset += block_size
364         if string_to_key_usage in [0, 255]:
365             key_end = -2
366         else:
367             key_end = 0
368         self['secret-key'] = data[offset:key_end]
369         if key_end:
370             self['secret-key-checksum'] = data[key_end:]
371
372     def _parse_signature_packet(self, data):
373         self['signature-version'] = data[0]
374         offset = 1
375         if self['signature-version'] != 4:
376             raise NotImplementedError(
377                 'signature packet version {}'.format(
378                     self['signature-version']))
379         self['signature-type'] = self._signature_types[data[offset]]
380         offset += 1
381         self['public-key-algorithm'] = self._public_key_algorithms[
382             data[offset]]
383         offset += 1
384         self['hash-algorithm'] = self._hash_algorithms[data[offset]]
385         offset += 1
386         hashed_count = _struct.unpack('>H', data[offset: offset + 2])[0]
387         offset += 2
388         self['hashed-subpackets'] = data[offset: offset + hashed_count]
389         offset += hashed_count
390         unhashed_count = _struct.unpack('>H', data[offset: offset + 2])[0]
391         offset += 2
392         self['unhashed-subpackets'] = data[offset: offset + unhashed_count]
393         offset += unhashed_count
394         self['signed-hash-word'] = data[offset: offset + 2]
395         offset += 2
396         self['signature'] = data[offset:]
397
398     def to_bytes(self):
399         pass
400
401
402 def packets_from_bytes(data):
403     offset = 0
404     while offset < len(data):
405         packet = PGPPacket()
406         offset += packet.from_bytes(data=data[offset:])
407         yield packet
408
409
410 def migrate(old_key, new_key):
411     """Add the old key and sub-keys to the new key
412
413     For example, to upgrade your master key, while preserving old
414     signatures you'd made.  You will lose signature *on* your old key
415     though, since sub-keys can't be signed (I don't think).
416     """
417     old_key_export = _get_stdout(
418         ['gpg', '--export', old_key])
419     old_key_packets = list(
420         packets_from_bytes(data=old_key_export))
421     if old_key_packets[0]['type'] != 'public-key packet':
422         raise ValueError(
423             '{} does not start with a public-key packet'.format(
424                 old_key))
425     old_key_secret_export = _get_stdout(
426         ['gpg', '--export-secret-keys', old_key])
427     old_key_secret_packets = list(
428         packets_from_bytes(data=old_key_secret_export))
429
430     import pprint
431     pprint.pprint(old_key_packets)
432     pprint.pprint(old_key_secret_packets)
433
434
435 if __name__ == '__main__':
436     import sys as _sys
437
438     old_key, new_key = _sys.argv[1:3]
439     migrate(old_key=old_key, new_key=new_key)