signing.py

   1 from zeroinstall import SafeException
   2 from zeroinstall.injector import gpg
   3 import tempfile, os, base64, sys, shutil
   4 import subprocess
   5 from logging import warn
   6
   7 def check_signature(path):
   8         data = file(path).read()
   9         xml_comment = data.rfind('\n<!-- Base64 Signature')
  10         if xml_comment >= 0:
  11                 data_stream, sigs = gpg.check_stream(file(path))
  12                 sign_fn = sign_xml
  13                 data = data[:xml_comment + 1]
  14                 data_stream.close()
  15         elif data.startswith('-----BEGIN'):
  16                 warn("Plain GPG signatures are no longer supported - not checking signature!")
  17                 warn("Will save in XML format.")
  18                 child = subprocess.Popen(['gpg', '--decrypt', path], stdout = subprocess.PIPE)
  19                 data, unused = child.communicate()
  20                 import __main__
  21                 __main__.force_save = True
  22                 return data, sign_xml, None
  23         else:
  24                 return data, sign_unsigned, None
  25         for sig in sigs:
  26                 if isinstance(sig, gpg.ValidSig):
  27                         return data, sign_fn, sig.fingerprint
  28         print "ERROR: No valid signatures found!"
  29         for sig in sigs:
  30                 print "Got:", sig
  31         ok = raw_input('Ignore and load anyway? (y/N) ').lower()
  32         if ok and 'yes'.startswith(ok):
  33                 import __main__
  34                 __main__.force_save = True
  35                 return data, sign_unsigned, None
  36         sys.exit(1)
  37
  38 def write_tmp(path, data):
  39         """Create a temporary file in the same directory as 'path' and write data to it."""
  40         tmpdir = os.path.dirname(path)
  41         if tmpdir:
  42                 assert os.path.isdir(tmpdir), "Not a directory: " + tmpdir
  43         fd, tmp = tempfile.mkstemp(prefix = 'tmp-', dir = tmpdir)
  44         stream = os.fdopen(fd, 'w')
  45         stream.write(data)
  46         stream.close()
  47
  48         umask = os.umask(0)
  49         os.umask(umask)
  50         os.chmod(tmp, 0644 & ~umask)
  51
  52         return tmp
  53
  54 def run_gpg(default_key, *arguments):
  55         arguments = list(arguments)
  56         if default_key is not None:
  57                 arguments = ['--default-key', default_key] + arguments
  58         arguments.insert(0, '--use-agent')
  59         arguments.insert(0, 'gpg')
  60         if os.spawnvp(os.P_WAIT, 'gpg', arguments):
  61                 raise SafeException("Command '%s' failed" % arguments)
  62
  63 def sign_unsigned(path, data, key):
  64         os.rename(write_tmp(path, data), path)
  65
  66 def sign_xml(path, data, key):
  67         tmp = write_tmp(path, data)
  68         sigtmp = tmp + '.sig'
  69         try:
  70                 run_gpg(key, '--detach-sign', '--output', sigtmp, tmp)
  71         finally:
  72                 os.unlink(tmp)
  73         encoded = base64.encodestring(file(sigtmp).read())
  74         os.unlink(sigtmp)
  75         sig = "<!-- Base64 Signature\n" + encoded + "\n-->\n"
  76         os.rename(write_tmp(path, data + sig), path)
  77
  78 def export_key(dir, fingerprint):
  79         assert fingerprint is not None
  80         # Convert fingerprint to key ID
  81         stream = os.popen('gpg --with-colons --list-keys %s' % fingerprint)
  82         try:
  83                 keyID = None
  84                 for line in stream:
  85                         parts = line.split(':')
  86                         if parts[0] == 'pub':
  87                                 if keyID:
  88                                         raise Exception('Two key IDs returned from GPG!')
  89                                 keyID = parts[4]
  90         finally:
  91                 stream.close()
  92         key_file = os.path.join(dir, keyID + '.gpg')
  93         if os.path.isfile(key_file):
  94                 return
  95         key_stream = file(key_file, 'w')
  96         stream = os.popen("gpg -a --export '%s'" % fingerprint)
  97         shutil.copyfileobj(stream, key_stream)
  98         stream.close()
  99         key_stream.close()
 100         print "Exported public key as '%s'" % key_file