1
# Copyright (C) 2005, 2011 Canonical Ltd
2
# Authors: Robert Collins <robert.collins@canonical.com>
4
# This program is free software; you can redistribute it and/or modify
5
# it under the terms of the GNU General Public License as published by
6
# the Free Software Foundation; either version 2 of the License, or
7
# (at your option) any later version.
9
# This program is distributed in the hope that it will be useful,
10
# but WITHOUT ANY WARRANTY; without even the implied warranty of
11
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12
# GNU General Public License for more details.
14
# You should have received a copy of the GNU General Public License
15
# along with this program; if not, write to the Free Software
16
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
18
"""GPG signing and checking logic."""
20
from __future__ import absolute_import
24
from StringIO import StringIO
26
from bzrlib.lazy_import import lazy_import
27
lazy_import(globals(), """
37
from bzrlib.i18n import (
43
from bzrlib.symbol_versioning import (
50
SIGNATURE_KEY_MISSING = 1
51
SIGNATURE_NOT_VALID = 2
52
SIGNATURE_NOT_SIGNED = 3
56
def bulk_verify_signatures(repository, revids, strategy,
57
process_events_callback=None):
58
"""Do verifications on a set of revisions
60
:param repository: repository object
61
:param revids: list of revision ids to verify
62
:param strategy: GPG strategy to use
63
:param process_events_callback: method to call for GUI frontends that
64
want to keep their UI refreshed
66
:return: count dictionary of results of each type,
67
result list for each revision,
68
boolean True if all results are verified successfully
70
count = {SIGNATURE_VALID: 0,
71
SIGNATURE_KEY_MISSING: 0,
72
SIGNATURE_NOT_VALID: 0,
73
SIGNATURE_NOT_SIGNED: 0,
78
pb = ui.ui_factory.nested_progress_bar()
80
for i, (rev_id, verification_result, uid) in enumerate(
81
repository.verify_revision_signatures(
83
pb.update("verifying signatures", i, total)
84
result.append([rev_id, verification_result, uid])
85
count[verification_result] += 1
86
if verification_result != SIGNATURE_VALID:
87
all_verifiable = False
88
if process_events_callback is not None:
89
process_events_callback()
92
return (count, result, all_verifiable)
95
class DisabledGPGStrategy(object):
96
"""A GPG Strategy that makes everything fail."""
99
def verify_signatures_available():
102
def __init__(self, ignored):
103
"""Real strategies take a configuration."""
105
def sign(self, content):
106
raise errors.SigningFailed('Signing is disabled.')
108
def verify(self, content, testament):
109
raise errors.SignatureVerificationFailed('Signature verification is \
112
def set_acceptable_keys(self, command_line_input):
116
class LoopbackGPGStrategy(object):
117
"""A GPG Strategy that acts like 'cat' - data is just passed through.
122
def verify_signatures_available():
125
def __init__(self, ignored):
126
"""Real strategies take a configuration."""
128
def sign(self, content):
129
return ("-----BEGIN PSEUDO-SIGNED CONTENT-----\n" + content +
130
"-----END PSEUDO-SIGNED CONTENT-----\n")
132
def verify(self, content, testament):
133
return SIGNATURE_VALID, None
135
def set_acceptable_keys(self, command_line_input):
136
if command_line_input is not None:
137
patterns = command_line_input.split(",")
138
self.acceptable_keys = []
139
for pattern in patterns:
140
if pattern == "unknown":
143
self.acceptable_keys.append(pattern)
145
@deprecated_method(deprecated_in((2, 6, 0)))
146
def do_verifications(self, revisions, repository):
147
return bulk_verify_signatures(repository, revisions, self)
149
@deprecated_method(deprecated_in((2, 6, 0)))
150
def valid_commits_message(self, count):
151
return valid_commits_message(count)
153
@deprecated_method(deprecated_in((2, 6, 0)))
154
def unknown_key_message(self, count):
155
return unknown_key_message(count)
157
@deprecated_method(deprecated_in((2, 6, 0)))
158
def commit_not_valid_message(self, count):
159
return commit_not_valid_message(count)
161
@deprecated_method(deprecated_in((2, 6, 0)))
162
def commit_not_signed_message(self, count):
163
return commit_not_signed_message(count)
165
@deprecated_method(deprecated_in((2, 6, 0)))
166
def expired_commit_message(self, count):
167
return expired_commit_message(count)
171
tty = os.environ.get('TTY')
173
os.environ['GPG_TTY'] = tty
174
trace.mutter('setting GPG_TTY=%s', tty)
176
# This is not quite worthy of a warning, because some people
177
# don't need GPG_TTY to be set. But it is worthy of a big mark
178
# in ~/.bzr.log, so that people can debug it if it happens to them
179
trace.mutter('** Env var TTY empty, cannot set GPG_TTY.'
183
class GPGStrategy(object):
184
"""GPG Signing and checking facilities."""
186
acceptable_keys = None
188
def __init__(self, config_stack):
189
self._config_stack = config_stack
192
self.context = gpgme.Context()
193
except ImportError, error:
194
pass # can't use verify()
197
def verify_signatures_available():
199
check if this strategy can verify signatures
201
:return: boolean if this strategy can verify signatures
206
except ImportError, error:
209
def _command_line(self):
210
key = self._config_stack.get('gpg_signing_key')
211
if key is None or key == 'default':
212
# 'default' or not setting gpg_signing_key at all means we should
213
# use the user email address
214
key = config.extract_email_address(self._config_stack.get('email'))
215
return [self._config_stack.get('gpg_signing_command'), '--clearsign',
218
def sign(self, content):
219
if isinstance(content, unicode):
220
raise errors.BzrBadParameterUnicode('content')
221
ui.ui_factory.clear_term()
223
preexec_fn = _set_gpg_tty
224
if sys.platform == 'win32':
225
# Win32 doesn't support preexec_fn, but wouldn't support TTY anyway.
228
process = subprocess.Popen(self._command_line(),
229
stdin=subprocess.PIPE,
230
stdout=subprocess.PIPE,
231
preexec_fn=preexec_fn)
233
result = process.communicate(content)[0]
234
if process.returncode is None:
236
if process.returncode != 0:
237
raise errors.SigningFailed(self._command_line())
240
if e.errno == errno.EPIPE:
241
raise errors.SigningFailed(self._command_line())
245
# bad subprocess parameters, should never happen.
248
if e.errno == errno.ENOENT:
249
# gpg is not installed
250
raise errors.SigningFailed(self._command_line())
254
def verify(self, content, testament):
255
"""Check content has a valid signature.
257
:param content: the commit signature
258
:param testament: the valid testament string for the commit
260
:return: SIGNATURE_VALID or a failed SIGNATURE_ value, key uid if valid
264
except ImportError, error:
265
raise errors.GpgmeNotInstalled(error)
267
signature = StringIO(content)
268
plain_output = StringIO()
270
result = self.context.verify(signature, None, plain_output)
271
except gpgme.GpgmeError,error:
272
raise errors.SignatureVerificationFailed(error[2])
274
# No result if input is invalid.
275
# test_verify_invalid()
277
return SIGNATURE_NOT_VALID, None
278
# User has specified a list of acceptable keys, check our result is in
279
# it. test_verify_unacceptable_key()
280
fingerprint = result[0].fpr
281
if self.acceptable_keys is not None:
282
if not fingerprint in self.acceptable_keys:
283
return SIGNATURE_KEY_MISSING, fingerprint[-8:]
284
# Check the signature actually matches the testament.
285
# test_verify_bad_testament()
286
if testament != plain_output.getvalue():
287
return SIGNATURE_NOT_VALID, None
288
# Yay gpgme set the valid bit.
289
# Can't write a test for this one as you can't set a key to be
290
# trusted using gpgme.
291
if result[0].summary & gpgme.SIGSUM_VALID:
292
key = self.context.get_key(fingerprint)
293
name = key.uids[0].name
294
email = key.uids[0].email
295
return SIGNATURE_VALID, name + " <" + email + ">"
296
# Sigsum_red indicates a problem, unfortunatly I have not been able
297
# to write any tests which actually set this.
298
if result[0].summary & gpgme.SIGSUM_RED:
299
return SIGNATURE_NOT_VALID, None
300
# GPG does not know this key.
301
# test_verify_unknown_key()
302
if result[0].summary & gpgme.SIGSUM_KEY_MISSING:
303
return SIGNATURE_KEY_MISSING, fingerprint[-8:]
304
# Summary isn't set if sig is valid but key is untrusted but if user
305
# has explicity set the key as acceptable we can validate it.
306
if result[0].summary == 0 and self.acceptable_keys is not None:
307
if fingerprint in self.acceptable_keys:
308
# test_verify_untrusted_but_accepted()
309
return SIGNATURE_VALID, None
310
# test_verify_valid_but_untrusted()
311
if result[0].summary == 0 and self.acceptable_keys is None:
312
return SIGNATURE_NOT_VALID, None
313
if result[0].summary & gpgme.SIGSUM_KEY_EXPIRED:
314
expires = self.context.get_key(result[0].fpr).subkeys[0].expires
315
if expires > result[0].timestamp:
316
# The expired key was not expired at time of signing.
317
# test_verify_expired_but_valid()
318
return SIGNATURE_EXPIRED, fingerprint[-8:]
320
# I can't work out how to create a test where the signature
321
# was expired at the time of signing.
322
return SIGNATURE_NOT_VALID, None
323
# A signature from a revoked key gets this.
324
# test_verify_revoked_signature()
325
if result[0].summary & gpgme.SIGSUM_SYS_ERROR:
326
return SIGNATURE_NOT_VALID, None
327
# Other error types such as revoked keys should (I think) be caught by
328
# SIGSUM_RED so anything else means something is buggy.
329
raise errors.SignatureVerificationFailed("Unknown GnuPG key "\
330
"verification result")
332
def set_acceptable_keys(self, command_line_input):
333
"""Set the acceptable keys for verifying with this GPGStrategy.
335
:param command_line_input: comma separated list of patterns from
340
acceptable_keys_config = self._config_stack.get('acceptable_keys')
342
if isinstance(acceptable_keys_config, unicode):
343
acceptable_keys_config = str(acceptable_keys_config)
344
except UnicodeEncodeError:
345
# gpg Context.keylist(pattern) does not like unicode
346
raise errors.BzrCommandError(
347
gettext('Only ASCII permitted in option names'))
349
if acceptable_keys_config is not None:
350
key_patterns = acceptable_keys_config
351
if command_line_input is not None: # command line overrides config
352
key_patterns = command_line_input
353
if key_patterns is not None:
354
patterns = key_patterns.split(",")
356
self.acceptable_keys = []
357
for pattern in patterns:
358
result = self.context.keylist(pattern)
362
self.acceptable_keys.append(key.subkeys[0].fpr)
363
trace.mutter("Added acceptable key: " + key.subkeys[0].fpr)
366
"No GnuPG key results for pattern: {0}"
369
@deprecated_method(deprecated_in((2, 6, 0)))
370
def do_verifications(self, revisions, repository,
371
process_events_callback=None):
372
"""do verifications on a set of revisions
374
:param revisions: list of revision ids to verify
375
:param repository: repository object
376
:param process_events_callback: method to call for GUI frontends that
377
want to keep their UI refreshed
379
:return: count dictionary of results of each type,
380
result list for each revision,
381
boolean True if all results are verified successfully
383
return bulk_verify_signatures(repository, revisions, self,
384
process_events_callback)
386
@deprecated_method(deprecated_in((2, 6, 0)))
387
def verbose_valid_message(self, result):
388
"""takes a verify result and returns list of signed commits strings"""
389
return verbose_valid_message(result)
391
@deprecated_method(deprecated_in((2, 6, 0)))
392
def verbose_not_valid_message(self, result, repo):
393
"""takes a verify result and returns list of not valid commit info"""
394
return verbose_not_valid_message(result, repo)
396
@deprecated_method(deprecated_in((2, 6, 0)))
397
def verbose_not_signed_message(self, result, repo):
398
"""takes a verify result and returns list of not signed commit info"""
399
return verbose_not_valid_message(result, repo)
401
@deprecated_method(deprecated_in((2, 6, 0)))
402
def verbose_missing_key_message(self, result):
403
"""takes a verify result and returns list of missing key info"""
404
return verbose_missing_key_message(result)
406
@deprecated_method(deprecated_in((2, 6, 0)))
407
def verbose_expired_key_message(self, result, repo):
408
"""takes a verify result and returns list of expired key info"""
409
return verbose_expired_key_message(result, repo)
411
@deprecated_method(deprecated_in((2, 6, 0)))
412
def valid_commits_message(self, count):
413
"""returns message for number of commits"""
414
return valid_commits_message(count)
416
@deprecated_method(deprecated_in((2, 6, 0)))
417
def unknown_key_message(self, count):
418
"""returns message for number of commits"""
419
return unknown_key_message(count)
421
@deprecated_method(deprecated_in((2, 6, 0)))
422
def commit_not_valid_message(self, count):
423
"""returns message for number of commits"""
424
return commit_not_valid_message(count)
426
@deprecated_method(deprecated_in((2, 6, 0)))
427
def commit_not_signed_message(self, count):
428
"""returns message for number of commits"""
429
return commit_not_signed_message(count)
431
@deprecated_method(deprecated_in((2, 6, 0)))
432
def expired_commit_message(self, count):
433
"""returns message for number of commits"""
434
return expired_commit_message(count)
437
def valid_commits_message(count):
438
"""returns message for number of commits"""
439
return gettext(u"{0} commits with valid signatures").format(
440
count[SIGNATURE_VALID])
443
def unknown_key_message(count):
444
"""returns message for number of commits"""
445
return ngettext(u"{0} commit with unknown key",
446
u"{0} commits with unknown keys",
447
count[SIGNATURE_KEY_MISSING]).format(
448
count[SIGNATURE_KEY_MISSING])
451
def commit_not_valid_message(count):
452
"""returns message for number of commits"""
453
return ngettext(u"{0} commit not valid",
454
u"{0} commits not valid",
455
count[SIGNATURE_NOT_VALID]).format(
456
count[SIGNATURE_NOT_VALID])
459
def commit_not_signed_message(count):
460
"""returns message for number of commits"""
461
return ngettext(u"{0} commit not signed",
462
u"{0} commits not signed",
463
count[SIGNATURE_NOT_SIGNED]).format(
464
count[SIGNATURE_NOT_SIGNED])
467
def expired_commit_message(count):
468
"""returns message for number of commits"""
469
return ngettext(u"{0} commit with key now expired",
470
u"{0} commits with key now expired",
471
count[SIGNATURE_EXPIRED]).format(
472
count[SIGNATURE_EXPIRED])
475
def verbose_expired_key_message(result, repo):
476
"""takes a verify result and returns list of expired key info"""
478
fingerprint_to_authors = {}
479
for rev_id, validity, fingerprint in result:
480
if validity == SIGNATURE_EXPIRED:
481
revision = repo.get_revision(rev_id)
482
authors = ', '.join(revision.get_apparent_authors())
483
signers.setdefault(fingerprint, 0)
484
signers[fingerprint] += 1
485
fingerprint_to_authors[fingerprint] = authors
487
for fingerprint, number in signers.items():
489
ngettext(u"{0} commit by author {1} with key {2} now expired",
490
u"{0} commits by author {1} with key {2} now expired",
492
number, fingerprint_to_authors[fingerprint], fingerprint))
496
def verbose_valid_message(result):
497
"""takes a verify result and returns list of signed commits strings"""
499
for rev_id, validity, uid in result:
500
if validity == SIGNATURE_VALID:
501
signers.setdefault(uid, 0)
504
for uid, number in signers.items():
505
result.append(ngettext(u"{0} signed {1} commit",
506
u"{0} signed {1} commits",
507
number).format(uid, number))
511
def verbose_not_valid_message(result, repo):
512
"""takes a verify result and returns list of not valid commit info"""
514
for rev_id, validity, empty in result:
515
if validity == SIGNATURE_NOT_VALID:
516
revision = repo.get_revision(rev_id)
517
authors = ', '.join(revision.get_apparent_authors())
518
signers.setdefault(authors, 0)
519
signers[authors] += 1
521
for authors, number in signers.items():
522
result.append(ngettext(u"{0} commit by author {1}",
523
u"{0} commits by author {1}",
524
number).format(number, authors))
528
def verbose_not_signed_message(result, repo):
529
"""takes a verify result and returns list of not signed commit info"""
531
for rev_id, validity, empty in result:
532
if validity == SIGNATURE_NOT_SIGNED:
533
revision = repo.get_revision(rev_id)
534
authors = ', '.join(revision.get_apparent_authors())
535
signers.setdefault(authors, 0)
536
signers[authors] += 1
538
for authors, number in signers.items():
539
result.append(ngettext(u"{0} commit by author {1}",
540
u"{0} commits by author {1}",
541
number).format(number, authors))
545
def verbose_missing_key_message(result):
546
"""takes a verify result and returns list of missing key info"""
548
for rev_id, validity, fingerprint in result:
549
if validity == SIGNATURE_KEY_MISSING:
550
signers.setdefault(fingerprint, 0)
551
signers[fingerprint] += 1
553
for fingerprint, number in signers.items():
554
result.append(ngettext(u"Unknown key {0} signed {1} commit",
555
u"Unknown key {0} signed {1} commits",
556
number).format(fingerprint, number))
added
removed
bzrlib/gpg.py
