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 (
45
SIGNATURE_KEY_MISSING = 1
46
SIGNATURE_NOT_VALID = 2
47
SIGNATURE_NOT_SIGNED = 3
51
class DisabledGPGStrategy(object):
52
"""A GPG Strategy that makes everything fail."""
55
def verify_signatures_available():
58
def __init__(self, ignored):
59
"""Real strategies take a configuration."""
61
def sign(self, content):
62
raise errors.SigningFailed('Signing is disabled.')
64
def verify(self, content, testament):
65
raise errors.SignatureVerificationFailed('Signature verification is \
68
def set_acceptable_keys(self, command_line_input):
72
class LoopbackGPGStrategy(object):
73
"""A GPG Strategy that acts like 'cat' - data is just passed through.
78
def verify_signatures_available():
81
def __init__(self, ignored):
82
"""Real strategies take a configuration."""
84
def sign(self, content):
85
return ("-----BEGIN PSEUDO-SIGNED CONTENT-----\n" + content +
86
"-----END PSEUDO-SIGNED CONTENT-----\n")
88
def verify(self, content, testament):
89
return SIGNATURE_VALID, None
91
def set_acceptable_keys(self, command_line_input):
92
if command_line_input is not None:
93
patterns = command_line_input.split(",")
94
self.acceptable_keys = []
95
for pattern in patterns:
96
if pattern == "unknown":
99
self.acceptable_keys.append(pattern)
101
def do_verifications(self, revisions, repository):
102
count = {SIGNATURE_VALID: 0,
103
SIGNATURE_KEY_MISSING: 0,
104
SIGNATURE_NOT_VALID: 0,
105
SIGNATURE_NOT_SIGNED: 0,
106
SIGNATURE_EXPIRED: 0}
108
all_verifiable = True
109
for rev_id in revisions:
110
verification_result, uid =\
111
repository.verify_revision_signature(rev_id,self)
112
result.append([rev_id, verification_result, uid])
113
count[verification_result] += 1
114
if verification_result != SIGNATURE_VALID:
115
all_verifiable = False
116
return (count, result, all_verifiable)
118
def valid_commits_message(self, count):
119
return gettext(u"{0} commits with valid signatures").format(
120
count[SIGNATURE_VALID])
122
def unknown_key_message(self, count):
123
return ngettext(u"{0} commit with unknown key",
124
u"{0} commits with unknown keys",
125
count[SIGNATURE_KEY_MISSING]).format(
126
count[SIGNATURE_KEY_MISSING])
128
def commit_not_valid_message(self, count):
129
return ngettext(u"{0} commit not valid",
130
u"{0} commits not valid",
131
count[SIGNATURE_NOT_VALID]).format(
132
count[SIGNATURE_NOT_VALID])
134
def commit_not_signed_message(self, count):
135
return ngettext(u"{0} commit not signed",
136
u"{0} commits not signed",
137
count[SIGNATURE_NOT_SIGNED]).format(
138
count[SIGNATURE_NOT_SIGNED])
140
def expired_commit_message(self, count):
141
return ngettext(u"{0} commit with key now expired",
142
u"{0} commits with key now expired",
143
count[SIGNATURE_EXPIRED]).format(
144
count[SIGNATURE_EXPIRED])
148
tty = os.environ.get('TTY')
150
os.environ['GPG_TTY'] = tty
151
trace.mutter('setting GPG_TTY=%s', tty)
153
# This is not quite worthy of a warning, because some people
154
# don't need GPG_TTY to be set. But it is worthy of a big mark
155
# in ~/.bzr.log, so that people can debug it if it happens to them
156
trace.mutter('** Env var TTY empty, cannot set GPG_TTY.'
160
class GPGStrategy(object):
161
"""GPG Signing and checking facilities."""
163
acceptable_keys = None
165
def __init__(self, config_stack):
166
self._config_stack = config_stack
169
self.context = gpgme.Context()
170
except ImportError, error:
171
pass # can't use verify()
174
def verify_signatures_available():
176
check if this strategy can verify signatures
178
:return: boolean if this strategy can verify signatures
183
except ImportError, error:
186
def _command_line(self):
187
key = self._config_stack.get('gpg_signing_key')
188
if key is None or key == 'default':
189
# 'default' or not setting gpg_signing_key at all means we should
190
# use the user email address
191
key = config.extract_email_address(self._config_stack.get('email'))
192
return [self._config_stack.get('gpg_signing_command'), '--clearsign',
195
def sign(self, content):
196
if isinstance(content, unicode):
197
raise errors.BzrBadParameterUnicode('content')
198
ui.ui_factory.clear_term()
200
preexec_fn = _set_gpg_tty
201
if sys.platform == 'win32':
202
# Win32 doesn't support preexec_fn, but wouldn't support TTY anyway.
205
process = subprocess.Popen(self._command_line(),
206
stdin=subprocess.PIPE,
207
stdout=subprocess.PIPE,
208
preexec_fn=preexec_fn)
210
result = process.communicate(content)[0]
211
if process.returncode is None:
213
if process.returncode != 0:
214
raise errors.SigningFailed(self._command_line())
217
if e.errno == errno.EPIPE:
218
raise errors.SigningFailed(self._command_line())
222
# bad subprocess parameters, should never happen.
225
if e.errno == errno.ENOENT:
226
# gpg is not installed
227
raise errors.SigningFailed(self._command_line())
231
def verify(self, content, testament):
232
"""Check content has a valid signature.
234
:param content: the commit signature
235
:param testament: the valid testament string for the commit
237
:return: SIGNATURE_VALID or a failed SIGNATURE_ value, key uid if valid
241
except ImportError, error:
242
raise errors.GpgmeNotInstalled(error)
244
signature = StringIO(content)
245
plain_output = StringIO()
247
result = self.context.verify(signature, None, plain_output)
248
except gpgme.GpgmeError,error:
249
raise errors.SignatureVerificationFailed(error[2])
251
# No result if input is invalid.
252
# test_verify_invalid()
254
return SIGNATURE_NOT_VALID, None
255
# User has specified a list of acceptable keys, check our result is in
256
# it. test_verify_unacceptable_key()
257
fingerprint = result[0].fpr
258
if self.acceptable_keys is not None:
259
if not fingerprint in self.acceptable_keys:
260
return SIGNATURE_KEY_MISSING, fingerprint[-8:]
261
# Check the signature actually matches the testament.
262
# test_verify_bad_testament()
263
if testament != plain_output.getvalue():
264
return SIGNATURE_NOT_VALID, None
265
# Yay gpgme set the valid bit.
266
# Can't write a test for this one as you can't set a key to be
267
# trusted using gpgme.
268
if result[0].summary & gpgme.SIGSUM_VALID:
269
key = self.context.get_key(fingerprint)
270
name = key.uids[0].name
271
email = key.uids[0].email
272
return SIGNATURE_VALID, name + " <" + email + ">"
273
# Sigsum_red indicates a problem, unfortunatly I have not been able
274
# to write any tests which actually set this.
275
if result[0].summary & gpgme.SIGSUM_RED:
276
return SIGNATURE_NOT_VALID, None
277
# GPG does not know this key.
278
# test_verify_unknown_key()
279
if result[0].summary & gpgme.SIGSUM_KEY_MISSING:
280
return SIGNATURE_KEY_MISSING, fingerprint[-8:]
281
# Summary isn't set if sig is valid but key is untrusted but if user
282
# has explicity set the key as acceptable we can validate it.
283
if result[0].summary == 0 and self.acceptable_keys is not None:
284
if fingerprint in self.acceptable_keys:
285
# test_verify_untrusted_but_accepted()
286
return SIGNATURE_VALID, None
287
# test_verify_valid_but_untrusted()
288
if result[0].summary == 0 and self.acceptable_keys is None:
289
return SIGNATURE_NOT_VALID, None
290
if result[0].summary & gpgme.SIGSUM_KEY_EXPIRED:
291
expires = self.context.get_key(result[0].fpr).subkeys[0].expires
292
if expires > result[0].timestamp:
293
# The expired key was not expired at time of signing.
294
# test_verify_expired_but_valid()
295
return SIGNATURE_EXPIRED, fingerprint[-8:]
297
# I can't work out how to create a test where the signature
298
# was expired at the time of signing.
299
return SIGNATURE_NOT_VALID, None
300
# A signature from a revoked key gets this.
301
# test_verify_revoked_signature()
302
if result[0].summary & gpgme.SIGSUM_SYS_ERROR:
303
return SIGNATURE_NOT_VALID, None
304
# Other error types such as revoked keys should (I think) be caught by
305
# SIGSUM_RED so anything else means something is buggy.
306
raise errors.SignatureVerificationFailed("Unknown GnuPG key "\
307
"verification result")
309
def set_acceptable_keys(self, command_line_input):
310
"""Set the acceptable keys for verifying with this GPGStrategy.
312
:param command_line_input: comma separated list of patterns from
317
acceptable_keys_config = self._config_stack.get('acceptable_keys')
319
if isinstance(acceptable_keys_config, unicode):
320
acceptable_keys_config = str(acceptable_keys_config)
321
except UnicodeEncodeError:
322
# gpg Context.keylist(pattern) does not like unicode
323
raise errors.BzrCommandError(
324
gettext('Only ASCII permitted in option names'))
326
if acceptable_keys_config is not None:
327
key_patterns = acceptable_keys_config
328
if command_line_input is not None: # command line overrides config
329
key_patterns = command_line_input
330
if key_patterns is not None:
331
patterns = key_patterns.split(",")
333
self.acceptable_keys = []
334
for pattern in patterns:
335
result = self.context.keylist(pattern)
339
self.acceptable_keys.append(key.subkeys[0].fpr)
340
trace.mutter("Added acceptable key: " + key.subkeys[0].fpr)
343
"No GnuPG key results for pattern: {0}"
346
def do_verifications(self, revisions, repository,
347
process_events_callback=None):
348
"""do verifications on a set of revisions
350
:param revisions: list of revision ids to verify
351
:param repository: repository object
352
:param process_events_callback: method to call for GUI frontends that
353
want to keep their UI refreshed
355
:return: count dictionary of results of each type,
356
result list for each revision,
357
boolean True if all results are verified successfully
359
count = {SIGNATURE_VALID: 0,
360
SIGNATURE_KEY_MISSING: 0,
361
SIGNATURE_NOT_VALID: 0,
362
SIGNATURE_NOT_SIGNED: 0,
363
SIGNATURE_EXPIRED: 0}
365
all_verifiable = True
366
for rev_id in revisions:
367
verification_result, uid =\
368
repository.verify_revision_signature(rev_id, self)
369
result.append([rev_id, verification_result, uid])
370
count[verification_result] += 1
371
if verification_result != SIGNATURE_VALID:
372
all_verifiable = False
373
if process_events_callback is not None:
374
process_events_callback()
375
return (count, result, all_verifiable)
377
def verbose_valid_message(self, result):
378
"""takes a verify result and returns list of signed commits strings"""
380
for rev_id, validity, uid in result:
381
if validity == SIGNATURE_VALID:
382
signers.setdefault(uid, 0)
385
for uid, number in signers.items():
386
result.append( ngettext(u"{0} signed {1} commit",
387
u"{0} signed {1} commits",
388
number).format(uid, number) )
392
def verbose_not_valid_message(self, result, repo):
393
"""takes a verify result and returns list of not valid commit info"""
395
for rev_id, validity, empty in result:
396
if validity == SIGNATURE_NOT_VALID:
397
revision = repo.get_revision(rev_id)
398
authors = ', '.join(revision.get_apparent_authors())
399
signers.setdefault(authors, 0)
400
signers[authors] += 1
402
for authors, number in signers.items():
403
result.append( ngettext(u"{0} commit by author {1}",
404
u"{0} commits by author {1}",
405
number).format(number, authors) )
408
def verbose_not_signed_message(self, result, repo):
409
"""takes a verify result and returns list of not signed commit info"""
411
for rev_id, validity, empty in result:
412
if validity == SIGNATURE_NOT_SIGNED:
413
revision = repo.get_revision(rev_id)
414
authors = ', '.join(revision.get_apparent_authors())
415
signers.setdefault(authors, 0)
416
signers[authors] += 1
418
for authors, number in signers.items():
419
result.append( ngettext(u"{0} commit by author {1}",
420
u"{0} commits by author {1}",
421
number).format(number, authors) )
424
def verbose_missing_key_message(self, result):
425
"""takes a verify result and returns list of missing key info"""
427
for rev_id, validity, fingerprint in result:
428
if validity == SIGNATURE_KEY_MISSING:
429
signers.setdefault(fingerprint, 0)
430
signers[fingerprint] += 1
432
for fingerprint, number in signers.items():
433
result.append( ngettext(u"Unknown key {0} signed {1} commit",
434
u"Unknown key {0} signed {1} commits",
435
number).format(fingerprint, number) )
438
def verbose_expired_key_message(self, result, repo):
439
"""takes a verify result and returns list of expired key info"""
441
fingerprint_to_authors = {}
442
for rev_id, validity, fingerprint in result:
443
if validity == SIGNATURE_EXPIRED:
444
revision = repo.get_revision(rev_id)
445
authors = ', '.join(revision.get_apparent_authors())
446
signers.setdefault(fingerprint, 0)
447
signers[fingerprint] += 1
448
fingerprint_to_authors[fingerprint] = authors
450
for fingerprint, number in signers.items():
452
ngettext(u"{0} commit by author {1} with key {2} now expired",
453
u"{0} commits by author {1} with key {2} now expired",
455
number, fingerprint_to_authors[fingerprint], fingerprint) )
458
def valid_commits_message(self, count):
459
"""returns message for number of commits"""
460
return gettext(u"{0} commits with valid signatures").format(
461
count[SIGNATURE_VALID])
463
def unknown_key_message(self, count):
464
"""returns message for number of commits"""
465
return ngettext(u"{0} commit with unknown key",
466
u"{0} commits with unknown keys",
467
count[SIGNATURE_KEY_MISSING]).format(
468
count[SIGNATURE_KEY_MISSING])
470
def commit_not_valid_message(self, count):
471
"""returns message for number of commits"""
472
return ngettext(u"{0} commit not valid",
473
u"{0} commits not valid",
474
count[SIGNATURE_NOT_VALID]).format(
475
count[SIGNATURE_NOT_VALID])
477
def commit_not_signed_message(self, count):
478
"""returns message for number of commits"""
479
return ngettext(u"{0} commit not signed",
480
u"{0} commits not signed",
481
count[SIGNATURE_NOT_SIGNED]).format(
482
count[SIGNATURE_NOT_SIGNED])
484
def expired_commit_message(self, count):
485
"""returns message for number of commits"""
486
return ngettext(u"{0} commit with key now expired",
487
u"{0} commits with key now expired",
488
count[SIGNATURE_EXPIRED]).format(
489
count[SIGNATURE_EXPIRED])
added
removed
bzrlib/gpg.py
