gpg.py
286 lines
| 9.0 KiB
| text/x-python
|
PythonLexer
/ hgext / gpg.py
|
r1681 | # Copyright 2005, 2006 Benoit Boissinot <benoit.boissinot@ens-lyon.org> | ||
# | ||||
|
r8225 | # This software may be used and distributed according to the terms of the | ||
|
r10263 | # GNU General Public License version 2 or any later version. | ||
|
r1681 | |||
|
r8934 | '''commands to sign and verify changesets''' | ||
|
r8873 | |||
|
r1681 | import os, tempfile, binascii | ||
|
r8701 | from mercurial import util, commands, match | ||
|
r1592 | from mercurial import node as hgnode | ||
|
r3891 | from mercurial.i18n import _ | ||
|
r1592 | |||
|
r8778 | class gpg(object): | ||
|
r1592 | def __init__(self, path, key=None): | ||
self.path = path | ||||
self.key = (key and " --local-user \"%s\"" % key) or "" | ||||
def sign(self, data): | ||||
gpgcmd = "%s --sign --detach-sign%s" % (self.path, self.key) | ||||
return util.filter(data, gpgcmd) | ||||
def verify(self, data, sig): | ||||
""" returns of the good and bad signatures""" | ||||
|
r2231 | sigfile = datafile = None | ||
|
r1592 | try: | ||
|
r1681 | # create temporary files | ||
|
r2165 | fd, sigfile = tempfile.mkstemp(prefix="hg-gpg-", suffix=".sig") | ||
|
r1592 | fp = os.fdopen(fd, 'wb') | ||
fp.write(sig) | ||||
fp.close() | ||||
|
r2165 | fd, datafile = tempfile.mkstemp(prefix="hg-gpg-", suffix=".txt") | ||
|
r1592 | fp = os.fdopen(fd, 'wb') | ||
fp.write(data) | ||||
fp.close() | ||||
|
r1681 | gpgcmd = ("%s --logger-fd 1 --status-fd 1 --verify " | ||
"\"%s\" \"%s\"" % (self.path, sigfile, datafile)) | ||||
|
r1592 | ret = util.filter("", gpgcmd) | ||
|
r2231 | finally: | ||
|
r1592 | for f in (sigfile, datafile): | ||
try: | ||||
|
r10282 | if f: | ||
os.unlink(f) | ||||
except: | ||||
pass | ||||
|
r1592 | keys = [] | ||
key, fingerprint = None, None | ||||
err = "" | ||||
for l in ret.splitlines(): | ||||
# see DETAILS in the gnupg documentation | ||||
# filter the logger output | ||||
if not l.startswith("[GNUPG:]"): | ||||
continue | ||||
l = l[9:] | ||||
if l.startswith("ERRSIG"): | ||||
|
r1681 | err = _("error while verifying signature") | ||
|
r1592 | break | ||
elif l.startswith("VALIDSIG"): | ||||
# fingerprint of the primary key | ||||
fingerprint = l.split()[10] | ||||
elif (l.startswith("GOODSIG") or | ||||
l.startswith("EXPSIG") or | ||||
l.startswith("EXPKEYSIG") or | ||||
l.startswith("BADSIG")): | ||||
if key is not None: | ||||
keys.append(key + [fingerprint]) | ||||
key = l.split(" ", 2) | ||||
fingerprint = None | ||||
if err: | ||||
return err, [] | ||||
if key is not None: | ||||
keys.append(key + [fingerprint]) | ||||
return err, keys | ||||
def newgpg(ui, **opts): | ||||
|
r1681 | """create a new gpg instance""" | ||
|
r1592 | gpgpath = ui.config("gpg", "cmd", "gpg") | ||
gpgkey = opts.get('key') | ||||
if not gpgkey: | ||||
gpgkey = ui.config("gpg", "key", None) | ||||
return gpg(gpgpath, gpgkey) | ||||
|
r1681 | def sigwalk(repo): | ||
""" | ||||
walk over every sigs, yields a couple | ||||
((node, version, sig), (filename, linenumber)) | ||||
""" | ||||
def parsefile(fileiter, context): | ||||
ln = 1 | ||||
for l in fileiter: | ||||
if not l: | ||||
continue | ||||
yield (l.split(" ", 2), (context, ln)) | ||||
|
r10394 | ln += 1 | ||
|
r1681 | |||
|
r8210 | # read the heads | ||
|
r1681 | fl = repo.file(".hgsigs") | ||
|
r8210 | for r in reversed(fl.heads()): | ||
|
r1681 | fn = ".hgsigs|%s" % hgnode.short(r) | ||
for item in parsefile(fl.read(r).splitlines(), fn): | ||||
yield item | ||||
try: | ||||
# read local signatures | ||||
fn = "localsigs" | ||||
for item in parsefile(repo.opener(fn), fn): | ||||
yield item | ||||
except IOError: | ||||
pass | ||||
def getkeys(ui, repo, mygpg, sigdata, context): | ||||
"""get the keys who signed a data""" | ||||
fn, ln = context | ||||
node, version, sig = sigdata | ||||
prefix = "%s:%d" % (fn, ln) | ||||
node = hgnode.bin(node) | ||||
data = node2txt(repo, node, version) | ||||
sig = binascii.a2b_base64(sig) | ||||
err, keys = mygpg.verify(data, sig) | ||||
if err: | ||||
ui.warn("%s:%d %s\n" % (fn, ln , err)) | ||||
return None | ||||
validkeys = [] | ||||
# warn for expired key and/or sigs | ||||
for key in keys: | ||||
if key[0] == "BADSIG": | ||||
ui.write(_("%s Bad signature from \"%s\"\n") % (prefix, key[2])) | ||||
continue | ||||
if key[0] == "EXPSIG": | ||||
ui.write(_("%s Note: Signature has expired" | ||||
" (signed by: \"%s\")\n") % (prefix, key[2])) | ||||
elif key[0] == "EXPKEYSIG": | ||||
ui.write(_("%s Note: This key has expired" | ||||
" (signed by: \"%s\")\n") % (prefix, key[2])) | ||||
validkeys.append((key[1], key[2], key[3])) | ||||
return validkeys | ||||
def sigs(ui, repo): | ||||
"""list signed changesets""" | ||||
mygpg = newgpg(ui) | ||||
revs = {} | ||||
for data, context in sigwalk(repo): | ||||
node, version, sig = data | ||||
fn, ln = context | ||||
try: | ||||
n = repo.lookup(node) | ||||
except KeyError: | ||||
ui.warn(_("%s:%d node does not exist\n") % (fn, ln)) | ||||
continue | ||||
r = repo.changelog.rev(n) | ||||
keys = getkeys(ui, repo, mygpg, data, context) | ||||
if not keys: | ||||
continue | ||||
revs.setdefault(r, []) | ||||
revs[r].extend(keys) | ||||
|
r8303 | for rev in sorted(revs, reverse=True): | ||
|
r1682 | for k in revs[rev]: | ||
r = "%5d:%s" % (rev, hgnode.hex(repo.changelog.node(rev))) | ||||
|
r1681 | ui.write("%-30s %s\n" % (keystr(ui, k), r)) | ||
|
r1592 | def check(ui, repo, rev): | ||
"""verify all the signatures there may be for a particular revision""" | ||||
mygpg = newgpg(ui) | ||||
rev = repo.lookup(rev) | ||||
hexrev = hgnode.hex(rev) | ||||
keys = [] | ||||
|
r1681 | for data, context in sigwalk(repo): | ||
node, version, sig = data | ||||
if node == hexrev: | ||||
k = getkeys(ui, repo, mygpg, data, context) | ||||
if k: | ||||
keys.extend(k) | ||||
|
r1592 | |||
if not keys: | ||||
|
r1681 | ui.write(_("No valid signature for %s\n") % hgnode.short(rev)) | ||
|
r1592 | return | ||
|
r1681 | |||
|
r1592 | # print summary | ||
ui.write("%s is signed by:\n" % hgnode.short(rev)) | ||||
|
r1681 | for key in keys: | ||
ui.write(" %s\n" % keystr(ui, key)) | ||||
|
r1592 | |||
|
r1681 | def keystr(ui, key): | ||
"""associate a string to a key (username, comment)""" | ||||
keyid, user, fingerprint = key | ||||
comment = ui.config("gpg", fingerprint, None) | ||||
if comment: | ||||
return "%s (%s)" % (user, comment) | ||||
else: | ||||
return user | ||||
|
r1592 | |||
def sign(ui, repo, *revs, **opts): | ||||
|
r3916 | """add a signature for the current or given revision | ||
If no revision is given, the parent of the working directory is used, | ||||
or tip if no revision is checked out. | ||||
|
r6163 | |||
See 'hg help dates' for a list of formats valid for -d/--date. | ||||
|
r3916 | """ | ||
|
r1592 | mygpg = newgpg(ui, **opts) | ||
sigver = "0" | ||||
sigmessage = "" | ||||
|
r6139 | |||
date = opts.get('date') | ||||
if date: | ||||
opts['date'] = util.parsedate(date) | ||||
|
r1592 | if revs: | ||
nodes = [repo.lookup(n) for n in revs] | ||||
else: | ||||
|
r3916 | nodes = [node for node in repo.dirstate.parents() | ||
if node != hgnode.nullid] | ||||
if len(nodes) > 1: | ||||
raise util.Abort(_('uncommitted merge - please provide a ' | ||||
'specific revision')) | ||||
if not nodes: | ||||
nodes = [repo.changelog.tip()] | ||||
|
r1592 | |||
for n in nodes: | ||||
hexnode = hgnode.hex(n) | ||||
ui.write("Signing %d:%s\n" % (repo.changelog.rev(n), | ||||
hgnode.short(n))) | ||||
# build data | ||||
data = node2txt(repo, n, sigver) | ||||
sig = mygpg.sign(data) | ||||
if not sig: | ||||
|
r1681 | raise util.Abort(_("Error while signing")) | ||
|
r1592 | sig = binascii.b2a_base64(sig) | ||
sig = sig.replace("\n", "") | ||||
sigmessage += "%s %s %s\n" % (hexnode, sigver, sig) | ||||
# write it | ||||
if opts['local']: | ||||
repo.opener("localsigs", "ab").write(sigmessage) | ||||
return | ||||
|
r6760 | for x in repo.status(unknown=True)[:5]: | ||
|
r1592 | if ".hgsigs" in x and not opts["force"]: | ||
|
r1681 | raise util.Abort(_("working copy of .hgsigs is changed " | ||
"(please commit .hgsigs manually " | ||||
"or use --force)")) | ||||
|
r1592 | |||
repo.wfile(".hgsigs", "ab").write(sigmessage) | ||||
|
r4906 | if '.hgsigs' not in repo.dirstate: | ||
|
r1592 | repo.add([".hgsigs"]) | ||
if opts["no_commit"]: | ||||
return | ||||
message = opts['message'] | ||||
if not message: | ||||
|
r9183 | # we don't translate commit messages | ||
message = "\n".join(["Added signature for changeset %s" | ||||
|
r5475 | % hgnode.short(n) | ||
|
r1592 | for n in nodes]) | ||
try: | ||||
|
r8808 | m = match.exact(repo.root, '', ['.hgsigs']) | ||
|
r8706 | repo.commit(message, opts['user'], opts['date'], match=m) | ||
|
r1592 | except ValueError, inst: | ||
raise util.Abort(str(inst)) | ||||
def node2txt(repo, node, ver): | ||||
"""map a manifest into some text""" | ||||
if ver == "0": | ||||
return "%s\n" % hgnode.hex(node) | ||||
else: | ||||
|
r1685 | raise util.Abort(_("unknown signature version")) | ||
|
r1592 | |||
cmdtable = { | ||||
"sign": | ||||
(sign, | ||||
|
r4730 | [('l', 'local', None, _('make the signature local')), | ||
('f', 'force', None, _('sign even if the sigfile is modified')), | ||||
('', 'no-commit', None, _('do not commit the sigfile after signing')), | ||||
|
r5147 | ('k', 'key', '', _('the key id to sign with')), | ||
|
r4730 | ('m', 'message', '', _('commit message')), | ||
|
r5147 | ] + commands.commitopts2, | ||
|
r4730 | _('hg sign [OPTION]... [REVISION]...')), | ||
|
r1681 | "sigcheck": (check, [], _('hg sigcheck REVISION')), | ||
"sigs": (sigs, [], _('hg sigs')), | ||||
|
r1592 | } | ||