/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to breezy/gpg.py

  • Committer: Jelmer Vernooij
  • Date: 2018-02-18 21:42:57 UTC
  • mto: This revision was merged to the branch mainline in revision 6859.
  • Revision ID: jelmer@jelmer.uk-20180218214257-jpevutp1wa30tz3v
Update TODO to reference Breezy, not Bazaar.

Show diffs side-by-side

added added

removed removed

Lines of Context:
breezy/gpg.py
17
17
 
18
18
"""GPG signing and checking logic."""
19
19
 
 
20
from __future__ import absolute_import
 
21
 
20
22
import os
 
23
import sys
21
24
 
22
25
from breezy.lazy_import import lazy_import
23
26
lazy_import(globals(), """
 
27
import errno
 
28
import subprocess
 
29
 
24
30
from breezy import (
25
31
    config,
26
32
    trace,
35
41
from . import (
36
42
    errors,
37
43
    )
 
44
from .sixish import (
 
45
    BytesIO,
 
46
    )
38
47
 
39
 
# verification results
 
48
#verification results
40
49
SIGNATURE_VALID = 0
41
50
SIGNATURE_KEY_MISSING = 1
42
51
SIGNATURE_NOT_VALID = 2
43
52
SIGNATURE_NOT_SIGNED = 3
44
53
SIGNATURE_EXPIRED = 4
45
54
 
46
 
MODE_NORMAL = 0
47
 
MODE_DETACH = 1
48
 
MODE_CLEAR = 2
49
 
 
50
55
 
51
56
class GpgNotInstalled(errors.DependencyNotPresent):
52
57
 
53
 
    _fmt = ('python-gpg is not installed, it is needed to create or '
54
 
            'verify signatures. %(error)s')
 
58
    _fmt = 'python-gpg is not installed, it is needed to verify signatures'
55
59
 
56
60
    def __init__(self, error):
57
61
        errors.DependencyNotPresent.__init__(self, 'gpg', error)
74
78
 
75
79
 
76
80
def bulk_verify_signatures(repository, revids, strategy,
77
 
                           process_events_callback=None):
 
81
        process_events_callback=None):
78
82
    """Do verifications on a set of revisions
79
83
 
80
84
    :param repository: repository object
95
99
    result = []
96
100
    all_verifiable = True
97
101
    total = len(revids)
98
 
    with ui.ui_factory.nested_progress_bar() as pb:
 
102
    pb = ui.ui_factory.nested_progress_bar()
 
103
    try:
99
104
        for i, (rev_id, verification_result, uid) in enumerate(
100
105
                repository.verify_revision_signatures(
101
106
                    revids, strategy)):
106
111
                all_verifiable = False
107
112
            if process_events_callback is not None:
108
113
                process_events_callback()
 
114
    finally:
 
115
        pb.finished()
109
116
    return (count, result, all_verifiable)
110
117
 
111
118
 
119
126
    def __init__(self, ignored):
120
127
        """Real strategies take a configuration."""
121
128
 
122
 
    def sign(self, content, mode):
 
129
    def sign(self, content):
123
130
        raise SigningFailed('Signing is disabled.')
124
131
 
125
 
    def verify(self, signed_data, signature=None):
 
132
    def verify(self, content, testament):
126
133
        raise SignatureVerificationFailed('Signature verification is \
127
134
disabled.')
128
135
 
142
149
    def __init__(self, ignored):
143
150
        """Real strategies take a configuration."""
144
151
 
145
 
    def sign(self, content, mode):
146
 
        return (b"-----BEGIN PSEUDO-SIGNED CONTENT-----\n" + content
147
 
                + b"-----END PSEUDO-SIGNED CONTENT-----\n")
 
152
    def sign(self, content):
 
153
        return ("-----BEGIN PSEUDO-SIGNED CONTENT-----\n" + content +
 
154
                "-----END PSEUDO-SIGNED CONTENT-----\n")
148
155
 
149
 
    def verify(self, signed_data, signature=None):
150
 
        plain_text = signed_data.replace(
151
 
            b"-----BEGIN PSEUDO-SIGNED CONTENT-----\n", b"")
152
 
        plain_text = plain_text.replace(
153
 
            b"-----END PSEUDO-SIGNED CONTENT-----\n", b"")
154
 
        return SIGNATURE_VALID, None, plain_text
 
156
    def verify(self, content, testament):
 
157
        return SIGNATURE_VALID, None
155
158
 
156
159
    def set_acceptable_keys(self, command_line_input):
157
160
        if command_line_input is not None:
172
175
    else:
173
176
        # This is not quite worthy of a warning, because some people
174
177
        # don't need GPG_TTY to be set. But it is worthy of a big mark
175
 
        # in brz.log, so that people can debug it if it happens to them
 
178
        # in ~/.brz.log, so that people can debug it if it happens to them
176
179
        trace.mutter('** Env var TTY empty, cannot set GPG_TTY.'
177
180
                     '  Is TTY exported?')
178
181
 
187
190
        try:
188
191
            import gpg
189
192
            self.context = gpg.Context()
190
 
            self.context.armor = True
191
 
            self.context.signers = self._get_signing_keys()
192
 
        except ImportError:
193
 
            pass  # can't use verify()
 
193
        except ImportError as error:
 
194
            pass # can't use verify()
 
195
 
 
196
        self.context.signers = self._get_signing_keys()
194
197
 
195
198
    def _get_signing_keys(self):
196
199
        import gpg
197
200
        keyname = self._config_stack.get('gpg_signing_key')
198
 
        if keyname == 'default':
199
 
            # Leave things to gpg
200
 
            return []
201
 
 
202
201
        if keyname:
203
202
            try:
204
203
                return [self.context.get_key(keyname)]
205
204
            except gpg.errors.KeyNotFound:
206
205
                pass
207
206
 
208
 
        if keyname is None:
209
 
            # not setting gpg_signing_key at all means we should
 
207
        if keyname is None or keyname == 'default':
 
208
            # 'default' or not setting gpg_signing_key at all means we should
210
209
            # use the user email address
211
 
            keyname = config.extract_email_address(
212
 
                self._config_stack.get('email'))
213
 
        if keyname == 'default':
214
 
            return []
 
210
            keyname = config.extract_email_address(self._config_stack.get('email'))
215
211
        possible_keys = self.context.keylist(keyname, secret=True)
216
212
        try:
217
 
            return [next(possible_keys)]
 
213
            return [possible_keys.next()]
218
214
        except StopIteration:
219
215
            return []
220
216
 
226
222
        :return: boolean if this strategy can verify signatures
227
223
        """
228
224
        try:
229
 
            import gpg  # noqa: F401
 
225
            import gpg
230
226
            return True
231
 
        except ImportError:
 
227
        except ImportError as error:
232
228
            return False
233
229
 
234
 
    def sign(self, content, mode):
235
 
        try:
236
 
            import gpg
237
 
        except ImportError as error:
238
 
            raise GpgNotInstalled(
239
 
                'Set create_signatures=no to disable creating signatures.')
240
 
 
241
 
        if isinstance(content, str):
 
230
    def sign(self, content):
 
231
        import gpg
 
232
        if isinstance(content, unicode):
242
233
            raise errors.BzrBadParameterUnicode('content')
243
234
 
244
235
        plain_text = gpg.Data(content)
245
236
        try:
246
237
            output, result = self.context.sign(
247
 
                plain_text, mode={
248
 
                    MODE_DETACH: gpg.constants.sig.mode.DETACH,
249
 
                    MODE_CLEAR: gpg.constants.sig.mode.CLEAR,
250
 
                    MODE_NORMAL: gpg.constants.sig.mode.NORMAL,
251
 
                    }[mode])
 
238
                plain_text, mode=gpg.constants.sig.mode.CLEAR)
252
239
        except gpg.errors.GPGMEError as error:
253
240
            raise SigningFailed(str(error))
254
241
 
255
242
        return output
256
243
 
257
 
    def verify(self, signed_data, signature=None):
 
244
    def verify(self, content, testament):
258
245
        """Check content has a valid signature.
259
246
 
260
 
        :param signed_data; Signed data
261
 
        :param signature: optional signature (if detached)
 
247
        :param content: the commit signature
 
248
        :param testament: the valid testament string for the commit
262
249
 
263
 
        :return: SIGNATURE_VALID or a failed SIGNATURE_ value, key uid if valid, plain text
 
250
        :return: SIGNATURE_VALID or a failed SIGNATURE_ value, key uid if valid
264
251
        """
265
252
        try:
266
253
            import gpg
267
254
        except ImportError as error:
268
 
            raise GpgNotInstalled(
269
 
                'Set check_signatures=ignore to disable verifying signatures.')
 
255
            raise errors.GpgNotInstalled(error)
270
256
 
271
 
        signed_data = gpg.Data(signed_data)
272
 
        if signature:
273
 
            signature = gpg.Data(signature)
 
257
        signature = gpg.Data(content)
 
258
        sink = gpg.Data()
274
259
        try:
275
 
            plain_output, result = self.context.verify(signed_data, signature)
 
260
            plain_output, result = self.context.verify(signature)
276
261
        except gpg.errors.BadSignatures as error:
277
262
            fingerprint = error.result.signatures[0].fpr
278
263
            if error.result.signatures[0].summary & gpg.constants.SIGSUM_KEY_EXPIRED:
279
 
                expires = self.context.get_key(
280
 
                    error.result.signatures[0].fpr).subkeys[0].expires
 
264
                expires = self.context.get_key(error.result.signatures[0].fpr).subkeys[0].expires
281
265
                if expires > error.result.signatures[0].timestamp:
282
266
                    # The expired key was not expired at time of signing.
283
267
                    # test_verify_expired_but_valid()
284
 
                    return SIGNATURE_EXPIRED, fingerprint[-8:], None
 
268
                    return SIGNATURE_EXPIRED, fingerprint[-8:]
285
269
                else:
286
270
                    # I can't work out how to create a test where the signature
287
271
                    # was expired at the time of signing.
288
 
                    return SIGNATURE_NOT_VALID, None, None
 
272
                    return SIGNATURE_NOT_VALID, None
289
273
 
290
274
            # GPG does not know this key.
291
275
            # test_verify_unknown_key()
292
 
            if (error.result.signatures[0].summary &
293
 
                    gpg.constants.SIGSUM_KEY_MISSING):
294
 
                return SIGNATURE_KEY_MISSING, fingerprint[-8:], None
 
276
            if error.result.signatures[0].summary & gpg.constants.SIGSUM_KEY_MISSING:
 
277
                return SIGNATURE_KEY_MISSING, fingerprint[-8:]
295
278
 
296
 
            return SIGNATURE_NOT_VALID, None, None
 
279
            return SIGNATURE_NOT_VALID, None
297
280
        except gpg.errors.GPGMEError as error:
298
 
            raise SignatureVerificationFailed(error)
 
281
            raise SignatureVerificationFailed(error[2])
299
282
 
300
283
        # No result if input is invalid.
301
284
        # test_verify_invalid()
302
285
        if len(result.signatures) == 0:
303
 
            return SIGNATURE_NOT_VALID, None, plain_output
304
 
 
 
286
            return SIGNATURE_NOT_VALID, None
305
287
        # User has specified a list of acceptable keys, check our result is in
306
288
        # it.  test_verify_unacceptable_key()
307
289
        fingerprint = result.signatures[0].fpr
308
290
        if self.acceptable_keys is not None:
309
 
            if fingerprint not in self.acceptable_keys:
310
 
                return SIGNATURE_KEY_MISSING, fingerprint[-8:], plain_output
 
291
            if not fingerprint in self.acceptable_keys:
 
292
                return SIGNATURE_KEY_MISSING, fingerprint[-8:]
 
293
        # Check the signature actually matches the testament.
 
294
        # test_verify_bad_testament()
 
295
        if testament != plain_output:
 
296
            return SIGNATURE_NOT_VALID, None
311
297
        # Yay gpg set the valid bit.
312
298
        # Can't write a test for this one as you can't set a key to be
313
299
        # trusted using gpg.
314
300
        if result.signatures[0].summary & gpg.constants.SIGSUM_VALID:
315
301
            key = self.context.get_key(fingerprint)
316
302
            name = key.uids[0].name
317
 
            if isinstance(name, bytes):
318
 
                name = name.decode('utf-8')
319
303
            email = key.uids[0].email
320
 
            if isinstance(email, bytes):
321
 
                email = email.decode('utf-8')
322
 
            return (SIGNATURE_VALID, name + u" <" + email + u">", plain_output)
 
304
            return SIGNATURE_VALID, name + " <" + email + ">"
323
305
        # Sigsum_red indicates a problem, unfortunatly I have not been able
324
306
        # to write any tests which actually set this.
325
307
        if result.signatures[0].summary & gpg.constants.SIGSUM_RED:
326
 
            return SIGNATURE_NOT_VALID, None, plain_output
 
308
            return SIGNATURE_NOT_VALID, None
327
309
        # Summary isn't set if sig is valid but key is untrusted but if user
328
310
        # has explicity set the key as acceptable we can validate it.
329
 
        if (result.signatures[0].summary == 0 and
330
 
                self.acceptable_keys is not None):
 
311
        if result.signatures[0].summary == 0 and self.acceptable_keys is not None:
331
312
            if fingerprint in self.acceptable_keys:
332
313
                # test_verify_untrusted_but_accepted()
333
 
                return SIGNATURE_VALID, None, plain_output
 
314
                return SIGNATURE_VALID, None
334
315
        # test_verify_valid_but_untrusted()
335
316
        if result.signatures[0].summary == 0 and self.acceptable_keys is None:
336
 
            return SIGNATURE_NOT_VALID, None, plain_output
 
317
            return SIGNATURE_NOT_VALID, None
337
318
        # Other error types such as revoked keys should (I think) be caught by
338
319
        # SIGSUM_RED so anything else means something is buggy.
339
320
        raise SignatureVerificationFailed(
350
331
        acceptable_keys_config = self._config_stack.get('acceptable_keys')
351
332
        if acceptable_keys_config is not None:
352
333
            patterns = acceptable_keys_config
353
 
        if command_line_input is not None:  # command line overrides config
 
334
        if command_line_input is not None: # command line overrides config
354
335
            patterns = command_line_input.split(',')
355
336
 
356
337
        if patterns:
364
345
                    trace.mutter("Added acceptable key: " + key.subkeys[0].fpr)
365
346
                if not found_key:
366
347
                    trace.note(gettext(
367
 
                        "No GnuPG key results for pattern: {0}"
368
 
                        ).format(pattern))
 
348
                            "No GnuPG key results for pattern: {0}"
 
349
                                ).format(pattern))
369
350
 
370
351
 
371
352
def valid_commits_message(count):
372
353
    """returns message for number of commits"""
373
354
    return gettext(u"{0} commits with valid signatures").format(
374
 
        count[SIGNATURE_VALID])
 
355
                                    count[SIGNATURE_VALID])
375
356
 
376
357
 
377
358
def unknown_key_message(count):
379
360
    return ngettext(u"{0} commit with unknown key",
380
361
                    u"{0} commits with unknown keys",
381
362
                    count[SIGNATURE_KEY_MISSING]).format(
382
 
        count[SIGNATURE_KEY_MISSING])
 
363
                                    count[SIGNATURE_KEY_MISSING])
383
364
 
384
365
 
385
366
def commit_not_valid_message(count):
387
368
    return ngettext(u"{0} commit not valid",
388
369
                    u"{0} commits not valid",
389
370
                    count[SIGNATURE_NOT_VALID]).format(
390
 
        count[SIGNATURE_NOT_VALID])
 
371
                                        count[SIGNATURE_NOT_VALID])
391
372
 
392
373
 
393
374
def commit_not_signed_message(count):
395
376
    return ngettext(u"{0} commit not signed",
396
377
                    u"{0} commits not signed",
397
378
                    count[SIGNATURE_NOT_SIGNED]).format(
398
 
        count[SIGNATURE_NOT_SIGNED])
 
379
                                    count[SIGNATURE_NOT_SIGNED])
399
380
 
400
381
 
401
382
def expired_commit_message(count):
403
384
    return ngettext(u"{0} commit with key now expired",
404
385
                    u"{0} commits with key now expired",
405
386
                    count[SIGNATURE_EXPIRED]).format(
406
 
        count[SIGNATURE_EXPIRED])
 
387
                                count[SIGNATURE_EXPIRED])
407
388
 
408
389
 
409
390
def verbose_expired_key_message(result, repo):
436
417
            signers[uid] += 1
437
418
    result = []
438
419
    for uid, number in signers.items():
439
 
        result.append(ngettext(u"{0} signed {1} commit",
440
 
                               u"{0} signed {1} commits",
441
 
                               number).format(uid, number))
 
420
         result.append(ngettext(u"{0} signed {1} commit",
 
421
                                u"{0} signed {1} commits",
 
422
                                number).format(uid, number))
442
423
    return result
443
424
 
444
425