signing.py

   1 from zeroinstall import SafeException
   2 from zeroinstall.injector import gpg
   3 import tempfile, os, base64, sys, shutil
   4
   5 def check_signature(path):
   6         data = file(path).read()
   7         xml_comment = data.rfind('\n<!-- Base64 Signature')
   8         if xml_comment >= 0:
   9                 data_stream, sigs = gpg.check_stream(file(path))
  10                 sign_fn = sign_xml
  11                 data = data[:xml_comment + 1]
  12                 data_stream.close()
  13         elif data.startswith('-----BEGIN'):
  14                 data_stream, sigs = gpg.check_stream(file(path))
  15                 sign_fn = sign_plain
  16                 data = data_stream.read()
  17         else:
  18                 return data, sign_unsigned, None
  19         for sig in sigs:
  20                 if isinstance(sig, gpg.ValidSig):
  21                         return data, sign_fn, sig.fingerprint
  22         print "ERROR: No valid signatures found!"
  23         for sig in sigs:
  24                 print "Got:", sig
  25         ok = raw_input('Ignore and load anyway? (y/N) ').lower()
  26         if ok and 'yes'.startswith(ok):
  27                 import __main__
  28                 __main__.force_save = True
  29                 return data, sign_unsigned, None
  30         sys.exit(1)
  31
  32 def write_tmp(path, data):
  33         """Create a temporary file in the same directory as 'path' and write data to it."""
  34         tmpdir = os.path.dirname(path)
  35         if tmpdir:
  36                 assert os.path.isdir(tmpdir), "Not a directory: " + tmpdir
  37         fd, tmp = tempfile.mkstemp(prefix = 'tmp-', dir = tmpdir)
  38         stream = os.fdopen(fd, 'w')
  39         stream.write(data)
  40         stream.close()
  41
  42         umask = os.umask(0)
  43         os.umask(umask)
  44         os.chmod(tmp, 0644 & ~umask)
  45
  46         return tmp
  47
  48 def run_gpg(default_key, *arguments):
  49         arguments = list(arguments)
  50         if default_key is not None:
  51                 arguments = ['--default-key', default_key] + arguments
  52         arguments.insert(0, 'gpg')
  53         if os.spawnvp(os.P_WAIT, 'gpg', arguments):
  54                 raise SafeException("Command '%s' failed" % arguments)
  55
  56 def sign_unsigned(path, data, key):
  57         os.rename(write_tmp(path, data), path)
  58
  59 def sign_plain(path, data, key):
  60         tmp = write_tmp(path, data)
  61         try:
  62                 run_gpg(key, '--clearsign', tmp)
  63         finally:
  64                 os.unlink(tmp)
  65         os.rename(tmp + '.asc', 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         os.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