import os, tempfile, binascii, errno
from mercurial import util
from mercurial import node as hgnode
class gpg:
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"""
try:
fd, sigfile = tempfile.mkstemp(prefix="hggpgsig")
fp = os.fdopen(fd, 'wb')
fp.write(sig)
fp.close()
fd, datafile = tempfile.mkstemp(prefix="hggpgdata")
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)
#gpgcmd = "%s --status-fd 1 --verify \"%s\" \"%s\"" % (self.path, sigfile, datafile)
ret = util.filter("", gpgcmd)
except:
for f in (sigfile, datafile):
try:
if f: os.unlink(f)
except: pass
raise
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):
gpgpath = ui.config("gpg", "cmd", "gpg")
gpgkey = opts.get('key')
if not gpgkey:
gpgkey = ui.config("gpg", "key", None)
return gpg(gpgpath, gpgkey)
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 = []
def addsig(fn, ln, l):
if not l: return
n, v, sig = l.split(" ", 2)
if n == hexrev:
data = node2txt(repo, rev, v)
sig = binascii.a2b_base64(sig)
err, k = mygpg.verify(data, sig)
if not err:
keys.append((k, fn, ln))
else:
ui.warn("%s:%d %s\n" % (fn, ln , err))
fl = repo.file(".hgsigs")
h = fl.heads()
h.reverse()
# read the heads
for r in h:
ln = 1
for l in fl.read(r).splitlines():
addsig(".hgsigs|%s" % hgnode.short(r), ln, l)
ln +=1
try:
# read local signatures
ln = 1
f = repo.opener("localsigs")
for l in f:
addsig("localsigs", ln, l)
ln +=1
except IOError:
pass
if not keys:
ui.write("%s not signed\n" % hgnode.short(rev))
return
valid = []
# warn for expired key and/or sigs
for k, fn, ln in keys:
prefix = "%s:%d" % (fn, ln)
for key in k:
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]))
valid.append((key[1], key[2], key[3]))
# print summary
ui.write("%s is signed by:\n" % hgnode.short(rev))
for keyid, user, fingerprint in valid:
role = getrole(ui, fingerprint)
ui.write(" %s (%s)\n" % (user, role))
def getrole(ui, fingerprint):
return ui.config("gpg", fingerprint, "no role defined")
def sign(ui, repo, *revs, **opts):
"""add a signature for the current tip or a given revision"""
mygpg = newgpg(ui, **opts)
sigver = "0"
sigmessage = ""
if revs:
nodes = [repo.lookup(n) for n in revs]
else:
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
(c, a, d, u) = repo.changes()
for x in (c, a, d, u):
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 repo.dirstate.state(".hgsigs") == '?':
repo.add([".hgsigs"])
if opts["no_commit"]:
return
message = opts['message']
if not message:
message = "\n".join(["Added signature for changeset %s" % hgnode.hex(n)
for n in nodes])
try:
repo.commit([".hgsigs"], message, opts['user'], opts['date'])
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:
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"),
('m', 'message', "", "commit message"),
('d', 'date', "", "date code"),
('u', 'user', "", "user"),
('k', 'key', "", "the key id to sign with")],
"hg sign [OPTION]... REVISIONS"),
"sigcheck": (check, [], 'hg sigcheck REVISION')
}