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                 return data, sign_unsigned, None
  28         sys.exit(1)
  29
  30 def write_tmp(path, data):
  31         """Create a temporary file in the same directory as 'path' and write data to it."""
  32         fd, tmp = tempfile.mkstemp(prefix = 'tmp-', dir = os.path.dirname(path))
  33         stream = os.fdopen(fd, 'w')
  34         stream.write(data)
  35         stream.close()
  36         return tmp
  37
  38 def run_gpg(default_key, *arguments):
  39         arguments = list(arguments)
  40         if default_key is not None:
  41                 arguments = ['--default-key', default_key] + arguments
  42         arguments.insert(0, 'gpg')
  43         if os.spawnvp(os.P_WAIT, 'gpg', arguments):
  44                 raise SafeException("Command '%s' failed" % arguments)
  45
  46 def sign_unsigned(path, data, key):
  47         os.rename(write_tmp(path, data), path)
  48
  49 def sign_plain(path, data, key):
  50         tmp = write_tmp(path, data)
  51         try:
  52                 run_gpg(key, '--clearsign', tmp)
  53         finally:
  54                 os.unlink(tmp)
  55         os.rename(tmp + '.asc', path)
  56
  57 def sign_xml(path, data, key):
  58         tmp = write_tmp(path, data)
  59         try:
  60                 run_gpg(key, '--detach-sign', tmp)
  61         finally:
  62                 os.unlink(tmp)
  63         tmp += '.sig'
  64         encoded = base64.encodestring(file(tmp).read())
  65         os.unlink(tmp)
  66         sig = "<!-- Base64 Signature\n" + encoded + "\n-->\n"
  67         os.rename(write_tmp(path, data + sig), path)
  68
  69 def export_key(dir, fingerprint):
  70         assert fingerprint is not None
  71         # Convert fingerprint to key ID
  72         stream = os.popen('gpg --with-colons --list-keys %s' % fingerprint)
  73         try:
  74                 keyID = None
  75                 for line in stream:
  76                         parts = line.split(':')
  77                         if parts[0] == 'pub':
  78                                 if keyID:
  79                                         raise Exception('Two key IDs returned from GPG!')
  80                                 keyID = parts[4]
  81         finally:
  82                 stream.close()
  83         key_file = os.path.join(dir, keyID + '.gpg')
  84         if os.path.isfile(key_file):
  85                 return
  86         key_stream = file(key_file, 'w')
  87         stream = os.popen("gpg -a --export '%s'" % fingerprint)
  88         shutil.copyfileobj(stream, key_stream)
  89         stream.close()
  90         key_stream.close()
  91         print "Exported public key as '%s'" % key_file