signing.py

   1 from zeroinstall import SafeException, support
   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         import subprocess
  61         if subprocess.call(arguments):
  62                 raise SafeException("Command '%s' failed" % arguments)
  63
  64 def sign_unsigned(path, data, key):
  65         support.portable_rename(write_tmp(path, data), path)
  66
  67 def sign_xml(path, data, key):
  68         tmp = write_tmp(path, data)
  69         sigtmp = tmp + '.sig'
  70         try:
  71                 run_gpg(key, '--detach-sign', '--output', sigtmp, tmp)
  72         finally:
  73                 os.unlink(tmp)
  74         encoded = base64.encodestring(file(sigtmp).read())
  75         os.unlink(sigtmp)
  76         sig = "<!-- Base64 Signature\n" + encoded + "\n-->\n"
  77         support.portable_rename(write_tmp(path, data + sig), path)
  78
  79 def export_key(dir, fingerprint):
  80         assert fingerprint is not None
  81         # Convert fingerprint to key ID
  82         stream = os.popen('gpg --with-colons --list-keys %s' % fingerprint)
  83         try:
  84                 keyID = None
  85                 for line in stream:
  86                         parts = line.split(':')
  87                         if parts[0] == 'pub':
  88                                 if keyID:
  89                                         raise Exception('Two key IDs returned from GPG!')
  90                                 keyID = parts[4]
  91         finally:
  92                 stream.close()
  93         key_file = os.path.join(dir, keyID + '.gpg')
  94         if os.path.isfile(key_file):
  95                 return
  96         key_stream = file(key_file, 'w')
  97         stream = os.popen("gpg -a --export '%s'" % fingerprint)
  98         shutil.copyfileobj(stream, key_stream)
  99         stream.close()
 100         key_stream.close()
 101         print "Exported public key as '%s'" % key_file