← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/gpg.py

(vila) Open 2.4.3 for bug fixes (Vincent Ladeuil)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/gpg.py
1
 
# Copyright (C) 2005 Canonical Ltd
 
1
# Copyright (C) 2005, 2011 Canonical Ltd
2
2
#   Authors: Robert Collins <robert.collins@canonical.com>
3
3
#
4
4
# This program is free software; you can redistribute it and/or modify
13
13
#
14
14
# You should have received a copy of the GNU General Public License
15
15
# along with this program; if not, write to the Free Software
16
 
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
16
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
17
 
18
18
"""GPG signing and checking logic."""
19
19
 
20
20
import os
21
21
import sys
 
22
from StringIO import StringIO
22
23
 
23
24
from bzrlib.lazy_import import lazy_import
24
25
lazy_import(globals(), """
32
33
    )
33
34
""")
34
35
 
 
36
class i18n:
 
37
    """this class is ready to use bzrlib.i18n but bzrlib.i18n is not ready to
 
38
    use so here is a stub until it is"""
 
39
    @staticmethod
 
40
    def gettext(string):
 
41
        return string
 
42
        
 
43
    @staticmethod
 
44
    def ngettext(single, plural, number):
 
45
        if number == 1:
 
46
            return single
 
47
        else:
 
48
            return plural
 
49
 
 
50
#verification results
 
51
SIGNATURE_VALID = 0
 
52
SIGNATURE_KEY_MISSING = 1
 
53
SIGNATURE_NOT_VALID = 2
 
54
SIGNATURE_NOT_SIGNED = 3
 
55
 
35
56
 
36
57
class DisabledGPGStrategy(object):
37
58
    """A GPG Strategy that makes everything fail."""
38
59
 
 
60
    @staticmethod
 
61
    def verify_signatures_available():
 
62
        return True
 
63
 
39
64
    def __init__(self, ignored):
40
65
        """Real strategies take a configuration."""
41
66
 
42
67
    def sign(self, content):
43
68
        raise errors.SigningFailed('Signing is disabled.')
44
69
 
 
70
    def verify(self, content, testament):
 
71
        raise errors.SignatureVerificationFailed('Signature verification is \
 
72
disabled.')
 
73
 
 
74
    def set_acceptable_keys(self, command_line_input):
 
75
        pass
 
76
 
45
77
 
46
78
class LoopbackGPGStrategy(object):
47
 
    """A GPG Strategy that acts like 'cat' - data is just passed through."""
 
79
    """A GPG Strategy that acts like 'cat' - data is just passed through.
 
80
    Used in tests.
 
81
    """
 
82
 
 
83
    @staticmethod
 
84
    def verify_signatures_available():
 
85
        return True
48
86
 
49
87
    def __init__(self, ignored):
50
88
        """Real strategies take a configuration."""
53
91
        return ("-----BEGIN PSEUDO-SIGNED CONTENT-----\n" + content +
54
92
                "-----END PSEUDO-SIGNED CONTENT-----\n")
55
93
 
 
94
    def verify(self, content, testament):
 
95
        return SIGNATURE_VALID, None
 
96
 
 
97
    def set_acceptable_keys(self, command_line_input):
 
98
        if command_line_input is not None:
 
99
            patterns = command_line_input.split(",")
 
100
            self.acceptable_keys = []
 
101
            for pattern in patterns:
 
102
                if pattern == "unknown":
 
103
                    pass
 
104
                else:
 
105
                    self.acceptable_keys.append(pattern)
 
106
 
 
107
    def do_verifications(self, revisions, repository):
 
108
        count = {SIGNATURE_VALID: 0,
 
109
                 SIGNATURE_KEY_MISSING: 0,
 
110
                 SIGNATURE_NOT_VALID: 0,
 
111
                 SIGNATURE_NOT_SIGNED: 0}
 
112
        result = []
 
113
        all_verifiable = True
 
114
        for rev_id in revisions:
 
115
            verification_result, uid =\
 
116
                                repository.verify_revision(rev_id,self)
 
117
            result.append([rev_id, verification_result, uid])
 
118
            count[verification_result] += 1
 
119
            if verification_result != SIGNATURE_VALID:
 
120
                all_verifiable = False
 
121
        return (count, result, all_verifiable)
 
122
 
 
123
    def valid_commits_message(self, count):
 
124
        return i18n.gettext(u"{0} commits with valid signatures").format(
 
125
                                        count[SIGNATURE_VALID])            
 
126
 
 
127
    def unknown_key_message(self, count):
 
128
        return i18n.ngettext(u"{0} commit with unknown key",
 
129
                             u"{0} commits with unknown keys",
 
130
                             count[SIGNATURE_KEY_MISSING]).format(
 
131
                                        count[SIGNATURE_KEY_MISSING])
 
132
 
 
133
    def commit_not_valid_message(self, count):
 
134
        return i18n.ngettext(u"{0} commit not valid",
 
135
                             u"{0} commits not valid",
 
136
                             count[SIGNATURE_NOT_VALID]).format(
 
137
                                            count[SIGNATURE_NOT_VALID])
 
138
 
 
139
    def commit_not_signed_message(self, count):
 
140
        return i18n.ngettext(u"{0} commit not signed",
 
141
                             u"{0} commits not signed",
 
142
                             count[SIGNATURE_NOT_SIGNED]).format(
 
143
                                        count[SIGNATURE_NOT_SIGNED])
 
144
 
56
145
 
57
146
def _set_gpg_tty():
58
147
    tty = os.environ.get('TTY')
70
159
class GPGStrategy(object):
71
160
    """GPG Signing and checking facilities."""
72
161
 
 
162
    acceptable_keys = None
 
163
 
 
164
    @staticmethod
 
165
    def verify_signatures_available():
 
166
        """
 
167
        check if this strategy can verify signatures
 
168
 
 
169
        :return: boolean if this strategy can verify signatures
 
170
        """
 
171
        try:
 
172
            import gpgme
 
173
            return True
 
174
        except ImportError, error:
 
175
            return False
 
176
 
73
177
    def _command_line(self):
74
178
        return [self._config.gpg_signing_command(), '--clearsign']
75
179
 
76
180
    def __init__(self, config):
77
181
        self._config = config
 
182
        try:
 
183
            import gpgme
 
184
            self.context = gpgme.Context()
 
185
        except ImportError, error:
 
186
            pass # can't use verify()
78
187
 
79
188
    def sign(self, content):
80
189
        if isinstance(content, unicode):
111
220
                raise errors.SigningFailed(self._command_line())
112
221
            else:
113
222
                raise
 
223
 
 
224
    def verify(self, content, testament):
 
225
        """Check content has a valid signature.
 
226
        
 
227
        :param content: the commit signature
 
228
        :param testament: the valid testament string for the commit
 
229
        
 
230
        :return: SIGNATURE_VALID or a failed SIGNATURE_ value, key uid if valid
 
231
        """
 
232
        try:
 
233
            import gpgme
 
234
        except ImportError, error:
 
235
            raise errors.GpgmeNotInstalled(error)
 
236
 
 
237
        signature = StringIO(content)
 
238
        plain_output = StringIO()
 
239
        
 
240
        try:
 
241
            result = self.context.verify(signature, None, plain_output)
 
242
        except gpgme.GpgmeError,error:
 
243
            raise errors.SignatureVerificationFailed(error[2])
 
244
 
 
245
        if len(result) == 0:
 
246
            return SIGNATURE_NOT_VALID, None
 
247
        fingerprint = result[0].fpr
 
248
        if self.acceptable_keys is not None:
 
249
            if not fingerprint in self.acceptable_keys:
 
250
                return SIGNATURE_KEY_MISSING, fingerprint[-8:]
 
251
        if testament != plain_output.getvalue():
 
252
            return SIGNATURE_NOT_VALID, None
 
253
        if result[0].summary & gpgme.SIGSUM_VALID:
 
254
            key = self.context.get_key(fingerprint)
 
255
            name = key.uids[0].name
 
256
            email = key.uids[0].email
 
257
            return SIGNATURE_VALID, name + " <" + email + ">"
 
258
        if result[0].summary & gpgme.SIGSUM_RED:
 
259
            return SIGNATURE_NOT_VALID, None
 
260
        if result[0].summary & gpgme.SIGSUM_KEY_MISSING:
 
261
            return SIGNATURE_KEY_MISSING, fingerprint[-8:]
 
262
        #summary isn't set if sig is valid but key is untrusted
 
263
        if result[0].summary == 0 and self.acceptable_keys is not None:
 
264
            if fingerprint in self.acceptable_keys:
 
265
                return SIGNATURE_VALID, None
 
266
        else:
 
267
            return SIGNATURE_KEY_MISSING, None
 
268
        raise errors.SignatureVerificationFailed("Unknown GnuPG key "\
 
269
                                                 "verification result")
 
270
 
 
271
    def set_acceptable_keys(self, command_line_input):
 
272
        """sets the acceptable keys for verifying with this GPGStrategy
 
273
        
 
274
        :param command_line_input: comma separated list of patterns from
 
275
                                command line
 
276
        :return: nothing
 
277
        """
 
278
        key_patterns = None
 
279
        acceptable_keys_config = self._config.acceptable_keys()
 
280
        try:
 
281
            if isinstance(acceptable_keys_config, unicode):
 
282
                acceptable_keys_config = str(acceptable_keys_config)
 
283
        except UnicodeEncodeError:
 
284
            #gpg Context.keylist(pattern) does not like unicode
 
285
            raise errors.BzrCommandError('Only ASCII permitted in option names')
 
286
 
 
287
        if acceptable_keys_config is not None:
 
288
            key_patterns = acceptable_keys_config
 
289
        if command_line_input is not None: #command line overrides config
 
290
            key_patterns = command_line_input
 
291
        if key_patterns is not None:
 
292
            patterns = key_patterns.split(",")
 
293
 
 
294
            self.acceptable_keys = []
 
295
            for pattern in patterns:
 
296
                result = self.context.keylist(pattern)
 
297
                found_key = False
 
298
                for key in result:
 
299
                    found_key = True
 
300
                    self.acceptable_keys.append(key.subkeys[0].fpr)
 
301
                    trace.mutter("Added acceptable key: " + key.subkeys[0].fpr)
 
302
                if not found_key:
 
303
                    trace.note(i18n.gettext(
 
304
                            "No GnuPG key results for pattern: {}"
 
305
                                ).format(pattern))
 
306
 
 
307
    def do_verifications(self, revisions, repository,
 
308
                            process_events_callback=None):
 
309
        """do verifications on a set of revisions
 
310
        
 
311
        :param revisions: list of revision ids to verify
 
312
        :param repository: repository object
 
313
        :param process_events_callback: method to call for GUI frontends that
 
314
                                                want to keep their UI refreshed
 
315
        
 
316
        :return: count dictionary of results of each type,
 
317
                 result list for each revision,
 
318
                 boolean True if all results are verified successfully
 
319
        """
 
320
        count = {SIGNATURE_VALID: 0,
 
321
                 SIGNATURE_KEY_MISSING: 0,
 
322
                 SIGNATURE_NOT_VALID: 0,
 
323
                 SIGNATURE_NOT_SIGNED: 0}
 
324
        result = []
 
325
        all_verifiable = True
 
326
        for rev_id in revisions:
 
327
            verification_result, uid =\
 
328
                                repository.verify_revision(rev_id,self)
 
329
            result.append([rev_id, verification_result, uid])
 
330
            count[verification_result] += 1
 
331
            if verification_result != SIGNATURE_VALID:
 
332
                all_verifiable = False
 
333
            if process_events_callback is not None:
 
334
                process_events_callback()
 
335
        return (count, result, all_verifiable)
 
336
 
 
337
    def verbose_valid_message(self, result):
 
338
        """takes a verify result and returns list of signed commits strings"""
 
339
        signers = {}
 
340
        for rev_id, validity, uid in result:
 
341
            if validity == SIGNATURE_VALID:
 
342
                signers.setdefault(uid, 0)
 
343
                signers[uid] += 1
 
344
        result = []
 
345
        for uid, number in signers.items():
 
346
             result.append( i18n.ngettext(u"{0} signed {1} commit", 
 
347
                             u"{0} signed {1} commits",
 
348
                             number).format(uid, number) )
 
349
        return result
 
350
 
 
351
 
 
352
    def verbose_not_valid_message(self, result, repo):
 
353
        """takes a verify result and returns list of not valid commit info"""
 
354
        signers = {}
 
355
        for rev_id, validity, empty in result:
 
356
            if validity == SIGNATURE_NOT_VALID:
 
357
                revision = repo.get_revision(rev_id)
 
358
                authors = ', '.join(revision.get_apparent_authors())
 
359
                signers.setdefault(authors, 0)
 
360
                signers[authors] += 1
 
361
        result = []
 
362
        for authors, number in signers.items():
 
363
            result.append( i18n.ngettext(u"{0} commit by author {1}", 
 
364
                                 u"{0} commits by author {1}",
 
365
                                 number).format(number, authors) )
 
366
        return result
 
367
 
 
368
    def verbose_not_signed_message(self, result, repo):
 
369
        """takes a verify result and returns list of not signed commit info"""
 
370
        signers = {}
 
371
        for rev_id, validity, empty in result:
 
372
            if validity == SIGNATURE_NOT_SIGNED:
 
373
                revision = repo.get_revision(rev_id)
 
374
                authors = ', '.join(revision.get_apparent_authors())
 
375
                signers.setdefault(authors, 0)
 
376
                signers[authors] += 1
 
377
        result = []
 
378
        for authors, number in signers.items():
 
379
            result.append( i18n.ngettext(u"{0} commit by author {1}", 
 
380
                                 u"{0} commits by author {1}",
 
381
                                 number).format(number, authors) )
 
382
        return result
 
383
 
 
384
    def verbose_missing_key_message(self, result):
 
385
        """takes a verify result and returns list of missing key info"""
 
386
        signers = {}
 
387
        for rev_id, validity, fingerprint in result:
 
388
            if validity == SIGNATURE_KEY_MISSING:
 
389
                signers.setdefault(fingerprint, 0)
 
390
                signers[fingerprint] += 1
 
391
        result = []
 
392
        for fingerprint, number in signers.items():
 
393
            result.append( i18n.ngettext(u"Unknown key {0} signed {1} commit", 
 
394
                                 u"Unknown key {0} signed {1} commits",
 
395
                                 number).format(fingerprint, number) )
 
396
        return result
 
397
 
 
398
    def valid_commits_message(self, count):
 
399
        """returns message for number of commits"""
 
400
        return i18n.gettext(u"{0} commits with valid signatures").format(
 
401
                                        count[SIGNATURE_VALID])
 
402
 
 
403
    def unknown_key_message(self, count):
 
404
        """returns message for number of commits"""
 
405
        return i18n.ngettext(u"{0} commit with unknown key",
 
406
                             u"{0} commits with unknown keys",
 
407
                             count[SIGNATURE_KEY_MISSING]).format(
 
408
                                        count[SIGNATURE_KEY_MISSING])
 
409
 
 
410
    def commit_not_valid_message(self, count):
 
411
        """returns message for number of commits"""
 
412
        return i18n.ngettext(u"{0} commit not valid",
 
413
                             u"{0} commits not valid",
 
414
                             count[SIGNATURE_NOT_VALID]).format(
 
415
                                            count[SIGNATURE_NOT_VALID])
 
416
 
 
417
    def commit_not_signed_message(self, count):
 
418
        """returns message for number of commits"""
 
419
        return i18n.ngettext(u"{0} commit not signed",
 
420
                             u"{0} commits not signed",
 
421
                             count[SIGNATURE_NOT_SIGNED]).format(
 
422
                                        count[SIGNATURE_NOT_SIGNED])