284 lines
9 KiB
Python
284 lines
9 KiB
Python
# Copyright 2005, 2006 Benoit Boissinot <benoit.boissinot@ens-lyon.org>
|
|
#
|
|
# This software may be used and distributed according to the terms of the
|
|
# GNU General Public License version 2, incorporated herein by reference.
|
|
|
|
'''commands to sign and verify changesets'''
|
|
|
|
import os, tempfile, binascii
|
|
from mercurial import util, commands, match
|
|
from mercurial import node as hgnode
|
|
from mercurial.i18n import _
|
|
|
|
class gpg(object):
|
|
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"""
|
|
sigfile = datafile = None
|
|
try:
|
|
# create temporary files
|
|
fd, sigfile = tempfile.mkstemp(prefix="hg-gpg-", suffix=".sig")
|
|
fp = os.fdopen(fd, 'wb')
|
|
fp.write(sig)
|
|
fp.close()
|
|
fd, datafile = tempfile.mkstemp(prefix="hg-gpg-", suffix=".txt")
|
|
fp = os.fdopen(fd, 'wb')
|
|
fp.write(data)
|
|
fp.close()
|
|
gpgcmd = ("%s --logger-fd 1 --status-fd 1 --verify "
|
|
"\"%s\" \"%s\"" % (self.path, sigfile, datafile))
|
|
ret = util.filter("", gpgcmd)
|
|
finally:
|
|
for f in (sigfile, datafile):
|
|
try:
|
|
if f: os.unlink(f)
|
|
except: pass
|
|
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"):
|
|
err = _("error while verifying signature")
|
|
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):
|
|
"""create a new gpg instance"""
|
|
gpgpath = ui.config("gpg", "cmd", "gpg")
|
|
gpgkey = opts.get('key')
|
|
if not gpgkey:
|
|
gpgkey = ui.config("gpg", "key", None)
|
|
return gpg(gpgpath, gpgkey)
|
|
|
|
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))
|
|
ln +=1
|
|
|
|
# read the heads
|
|
fl = repo.file(".hgsigs")
|
|
for r in reversed(fl.heads()):
|
|
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)
|
|
for rev in sorted(revs, reverse=True):
|
|
for k in revs[rev]:
|
|
r = "%5d:%s" % (rev, hgnode.hex(repo.changelog.node(rev)))
|
|
ui.write("%-30s %s\n" % (keystr(ui, k), r))
|
|
|
|
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 = []
|
|
|
|
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)
|
|
|
|
if not keys:
|
|
ui.write(_("No valid signature for %s\n") % hgnode.short(rev))
|
|
return
|
|
|
|
# print summary
|
|
ui.write("%s is signed by:\n" % hgnode.short(rev))
|
|
for key in keys:
|
|
ui.write(" %s\n" % keystr(ui, key))
|
|
|
|
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
|
|
|
|
def sign(ui, repo, *revs, **opts):
|
|
"""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.
|
|
|
|
See 'hg help dates' for a list of formats valid for -d/--date.
|
|
"""
|
|
|
|
mygpg = newgpg(ui, **opts)
|
|
sigver = "0"
|
|
sigmessage = ""
|
|
|
|
date = opts.get('date')
|
|
if date:
|
|
opts['date'] = util.parsedate(date)
|
|
|
|
if revs:
|
|
nodes = [repo.lookup(n) for n in revs]
|
|
else:
|
|
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()]
|
|
|
|
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:
|
|
raise util.Abort(_("Error while signing"))
|
|
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
|
|
|
|
for x in repo.status(unknown=True)[:5]:
|
|
if ".hgsigs" in x and not opts["force"]:
|
|
raise util.Abort(_("working copy of .hgsigs is changed "
|
|
"(please commit .hgsigs manually "
|
|
"or use --force)"))
|
|
|
|
repo.wfile(".hgsigs", "ab").write(sigmessage)
|
|
|
|
if '.hgsigs' not in repo.dirstate:
|
|
repo.add([".hgsigs"])
|
|
|
|
if opts["no_commit"]:
|
|
return
|
|
|
|
message = opts['message']
|
|
if not message:
|
|
# we don't translate commit messages
|
|
message = "\n".join(["Added signature for changeset %s"
|
|
% hgnode.short(n)
|
|
for n in nodes])
|
|
try:
|
|
m = match.exact(repo.root, '', ['.hgsigs'])
|
|
repo.commit(message, opts['user'], opts['date'], match=m)
|
|
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:
|
|
raise util.Abort(_("unknown signature version"))
|
|
|
|
cmdtable = {
|
|
"sign":
|
|
(sign,
|
|
[('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')),
|
|
('k', 'key', '', _('the key id to sign with')),
|
|
('m', 'message', '', _('commit message')),
|
|
] + commands.commitopts2,
|
|
_('hg sign [OPTION]... [REVISION]...')),
|
|
"sigcheck": (check, [], _('hg sigcheck REVISION')),
|
|
"sigs": (sigs, [], _('hg sigs')),
|
|
}
|
|
|