interfaces/email/interactive/send_pgp_mime.py

   1 #!/usr/bin/python
   2 #
   3 # Copyright (C) 2009 W. Trevor King <wking@drexel.edu>
   4 #
   5 # This program is free software; you can redistribute it and/or modify
   6 # it under the terms of the GNU General Public License as published by
   7 # the Free Software Foundation; either version 2 of the License, or
   8 # (at your option) any later version.
   9 #
  10 # This program is distributed in the hope that it will be useful,
  11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 # GNU General Public License for more details.
  14 #
  15 # You should have received a copy of the GNU General Public License along
  16 # with this program; if not, write to the Free Software Foundation, Inc.,
  17 # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  18 """
  19 Python module and command line tool for sending pgp/mime email.
  20
  21 Mostly uses subprocess to call gpg and a sendmail-compatible mailer.
  22 If you lack gpg, either don't use the encryption functions or adjust
  23 the pgp_* commands.  You may need to adjust the sendmail command to
  24 point to whichever sendmail-compatible mailer you have on your system.
  25 """
  26
  27 from cStringIO import StringIO
  28 import os
  29 import re
  30 #import GnuPGInterface # Maybe should use this instead of subprocess
  31 import smtplib
  32 import subprocess
  33 import sys
  34 import tempfile
  35 import types
  36
  37 try:
  38     from email import Message
  39     from email.mime.text import MIMEText
  40     from email.mime.multipart import MIMEMultipart
  41     from email.mime.application import MIMEApplication
  42     from email.encoders import encode_7or8bit
  43     from email.generator import Generator
  44     from email.parser import Parser
  45     from email.utils import getaddress
  46 except ImportError:
  47     # adjust to old python 2.4
  48     from email import Message
  49     from email.MIMEText import MIMEText
  50     from email.MIMEMultipart import MIMEMultipart
  51     from email.MIMENonMultipart import MIMENonMultipart
  52     from email.Encoders import encode_7or8bit
  53     from email.Generator import Generator
  54     from email.parser import Parser
  55     from email.Utils import getaddresses
  56
  57     getaddress = getaddresses
  58     class MIMEApplication (MIMENonMultipart):
  59         def __init__(self, _data, _subtype, _encoder, **params):
  60             MIMENonMultipart.__init__(self, 'application', _subtype, **params)
  61             self.set_payload(_data)
  62             _encoder(self)
  63
  64 usage="""usage: %prog [options]
  65
  66 Scriptable PGP MIME email using gpg.
  67
  68 You can use gpg-agent for passphrase caching if your key requires a
  69 passphrase (it better!).  Example usage would be to install gpg-agent,
  70 and then run
  71   export GPG_TTY=`tty`
  72   eval $(gpg-agent --daemon)
  73 in your shell before invoking this script.  See gpg-agent(1) for more
  74 details.  Alternatively, you can send your passphrase in on stdin
  75   echo 'passphrase' | %prog [options]
  76 or use the --passphrase-file option
  77   %prog [options] --passphrase-file FILE [more options]
  78 Both of these alternatives are much less secure than gpg-agent.  You
  79 have been warned.
  80 """
  81
  82 verboseInvoke = False
  83 PGP_SIGN_AS = None
  84 PASSPHRASE = None
  85
  86 # The following commands are adapted from my .mutt/pgp configuration
  87 #
  88 # Printf-like sequences:
  89 #   %a The value of PGP_SIGN_AS.
  90 #   %f Expands to the name of a file with text to be signed/encrypted.
  91 #   %p Expands to the passphrase argument.
  92 #   %R A string with some number (0 on up) of pgp_reciepient_arg
  93 #      strings.
  94 #   %r One key ID (e.g. recipient email address) to build a
  95 #      pgp_reciepient_arg string.
  96 #
  97 # The above sequences can be used to optionally print a string if
  98 # their length is nonzero. For example, you may only want to pass the
  99 # -u/--local-user argument to gpg if PGP_SIGN_AS is defined.  To
 100 # optionally print a string based upon one of the above sequences, the
 101 # following construct is used
 102 #   %?<sequence_char>?<optional_string>?
 103 # where sequence_char is a character from the table above, and
 104 # optional_string is the string you would like printed if status_char
 105 # is nonzero. optional_string may contain other sequence as well as
 106 # normal text, but it may not contain any question marks.
 107 #
 108 # see http://codesorcery.net/old/mutt/mutt-gnupg-howto
 109 #     http://www.mutt.org/doc/manual/manual-6.html#pgp_autosign
 110 #     http://tldp.org/HOWTO/Mutt-GnuPG-PGP-HOWTO-8.html
 111 # for more details
 112
 113 pgp_recipient_arg='-r "%r"'
 114 pgp_stdin_passphrase_arg='--passphrase-fd 0'
 115 pgp_sign_command='/usr/bin/gpg --no-verbose --quiet --batch %p --output - --detach-sign --armor --textmode %?a?-u "%a"? %f'
 116 pgp_encrypt_only_command='/usr/bin/gpg --no-verbose --quiet --batch --output - --encrypt --armor --textmode --always-trust --encrypt-to "%a" %R -- %f'
 117 pgp_encrypt_sign_command='/usr/bin/gpg --no-verbose --quiet --batch %p --output - --encrypt --sign %?a?-u "%a"? --armor --textmode --always-trust --encrypt-to "%a" %R -- %f'
 118 sendmail='/usr/sbin/sendmail -t'
 119
 120 def mail(msg, sendmail=None):
 121     """
 122     Send an email Message instance on its merry way.
 123
 124     We can shell out to the user specified sendmail in case
 125     the local host doesn't have an SMTP server set up
 126     for easy smtplib usage.
 127     """
 128     if sendmail != None:
 129         execute(sendmail, stdin=flatten(msg))
 130         return None
 131     s = smtplib.SMTP()
 132     s.connect()
 133     s.sendmail(from_addr=source_email(msg),
 134                to_addrs=target_emails(msg),
 135                msg=flatten(msg))
 136     s.close()
 137
 138 def header_from_text(text, encoding="us-ascii"):
 139     """
 140     Simple wrapper for instantiating an email.Message from text.
 141     >>> header = header_from_text('\\n'.join(['From: me@big.edu','To: you@big.edu','Subject: testing']))
 142     >>> print flatten(header)
 143     From: me@big.edu
 144     To: you@big.edu
 145     Subject: testing
 146     <BLANKLINE>
 147     <BLANKLINE>
 148     """
 149     text = text.strip()
 150     if type(text) == types.UnicodeType:
 151         text = text.encode(encoding)
 152     # assume StringType arguments are already encoded
 153     p = Parser()
 154     return p.parsestr(text, headersonly=True)
 155
 156 def attach_root(header, root_part):
 157     """
 158     Attach the email.Message root_part to the email.Message header
 159     without generating a multi-part message.
 160     """
 161     for k,v in header.items():
 162         root_part[k] = v
 163     return root_part
 164
 165 def execute(args, stdin=None, expect=(0,)):
 166     """
 167     Execute a command (allows us to drive gpg).
 168     """
 169     if verboseInvoke == True:
 170         print >> sys.stderr, '$ '+args
 171     try:
 172         p = subprocess.Popen(args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True, close_fds=True)
 173     except OSError, e:
 174         strerror = '%s\nwhile executing %s' % (e.args[1], args)
 175         raise Exception, strerror
 176     output, error = p.communicate(input=stdin)
 177     status = p.wait()
 178     if verboseInvoke == True:
 179         print >> sys.stderr, '(status: %d)\n%s%s' % (status, output, error)
 180     if status not in expect:
 181         strerror = '%s\nwhile executing %s\n%s\n%d' % (args[1], args, error, status)
 182         raise Exception, strerror
 183     return status, output, error
 184
 185 def replace(template, format_char, replacement_text):
 186     """
 187     >>> replace('--textmode %?a?-u %a? %f', 'f', 'file.in')
 188     '--textmode %?a?-u %a? file.in'
 189     >>> replace('--textmode %?a?-u %a? %f', 'a', '0xHEXKEY')
 190     '--textmode -u 0xHEXKEY %f'
 191     >>> replace('--textmode %?a?-u %a? %f', 'a', '')
 192     '--textmode  %f'
 193     """
 194     if replacement_text == None:
 195         replacement_text = ""
 196     regexp = re.compile('%[?]'+format_char+'[?]([^?]*)[?]')
 197     if len(replacement_text) > 0:
 198         str = regexp.sub('\g<1>', template)
 199     else:
 200         str = regexp.sub('', template)
 201     regexp = re.compile('%'+format_char)
 202     str = regexp.sub(replacement_text, str)
 203     return str
 204
 205 def flatten(msg, to_unicode=False):
 206     """
 207     Produce flat text output from an email Message instance.
 208     """
 209     assert msg != None
 210     fp = StringIO()
 211     g = Generator(fp, mangle_from_=False)
 212     g.flatten(msg)
 213     text = fp.getvalue()
 214     if to_unicode == True:
 215         encoding = msg.get_content_charset() or "utf-8"
 216         text = unicode(text, encoding=encoding)
 217     return text
 218
 219 def source_email(msg, return_realname=False):
 220     """
 221     Search the header of an email Message instance to find the
 222     sender's email address.
 223     """
 224     froms = msg.get_all('from', [])
 225     from_tuples = getaddresses(froms) # [(realname, email_address), ...]
 226     assert len(from_tuples) == 1
 227     if return_realname == True:
 228         return from_tuples[0] # (realname, email_address)
 229     return from_tuples[0][1]  # email_address
 230
 231 def target_emails(msg):
 232     """
 233     Search the header of an email Message instance to find a
 234     list of recipient's email addresses.
 235     """
 236     tos = msg.get_all('to', [])
 237     ccs = msg.get_all('cc', [])
 238     bccs = msg.get_all('bcc', [])
 239     resent_tos = msg.get_all('resent-to', [])
 240     resent_ccs = msg.get_all('resent-cc', [])
 241     resent_bccs = msg.get_all('resent-bcc', [])
 242     all_recipients = getaddresses(tos + ccs + bccs + resent_tos
 243                                   + resent_ccs + resent_bccs)
 244     return [addr[1] for addr in all_recipients]
 245
 246 class PGPMimeMessageFactory (object):
 247     """
 248     See http://www.ietf.org/rfc/rfc3156.txt for specification details.
 249     >>> from_addr = "me@big.edu"
 250     >>> to_addr = "you@you.edu"
 251     >>> header = header_from_text('\\n'.join(['From: %s'%from_addr,'To: %s'%to_addr,'Subject: testing']))
 252     >>> source_email(header) == from_addr
 253     True
 254     >>> target_emails(header) == [to_addr]
 255     True
 256     >>> m = EncryptedMessageFactory('check 1 2\\ncheck 1 2\\n')
 257     >>> print flatten(m.clearBodyPart())
 258     Content-Type: text/plain; charset="us-ascii"
 259     MIME-Version: 1.0
 260     Content-Transfer-Encoding: 7bit
 261     Content-Disposition: inline
 262     <BLANKLINE>
 263     check 1 2
 264     check 1 2
 265     <BLANKLINE>
 266     >>> print flatten(m.plain())
 267     Content-Type: text/plain; charset="us-ascii"
 268     MIME-Version: 1.0
 269     Content-Transfer-Encoding: 7bit
 270     <BLANKLINE>
 271     check 1 2
 272     check 1 2
 273     <BLANKLINE>
 274     >>> signed = m.sign(header)
 275     >>> signed.set_boundary('boundsep')
 276     >>> print flatten(signed).replace('\\t', ' '*4) # doctest: +ELLIPSIS, +NORMALIZE_WHITESPACE
 277     Content-Type: multipart/signed; protocol="application/pgp-signature";
 278         micalg="pgp-sha1"; boundary="boundsep"
 279     MIME-Version: 1.0
 280     Content-Disposition: inline
 281     <BLANKLINE>
 282     --boundsep
 283     Content-Type: text/plain; charset="us-ascii"
 284     MIME-Version: 1.0
 285     Content-Transfer-Encoding: 7bit
 286     Content-Disposition: inline
 287     <BLANKLINE>
 288     check 1 2
 289     check 1 2
 290     <BLANKLINE>
 291     --boundsep
 292     MIME-Version: 1.0
 293     Content-Transfer-Encoding: 7bit
 294     Content-Description: signature
 295     Content-Type: application/pgp-signature; name="signature.asc";
 296         charset="us-ascii"
 297     <BLANKLINE>
 298     -----BEGIN PGP SIGNATURE-----
 299     ...
 300     -----END PGP SIGNATURE-----
 301     <BLANKLINE>
 302     --boundsep--
 303     >>> encrypted = m.encrypt(header)
 304     >>> encrypted.set_boundary('boundsep')
 305     >>> print flatten(encrypted).replace('\\t', ' '*4) # doctest: +ELLIPSIS, +NORMALIZE_WHITESPACE
 306     Content-Type: multipart/encrypted;
 307         protocol="application/pgp-encrypted";
 308         micalg="pgp-sha1"; boundary="boundsep"
 309     MIME-Version: 1.0
 310     Content-Disposition: inline
 311     <BLANKLINE>
 312     --boundsep
 313     Content-Type: application/pgp-encrypted
 314     MIME-Version: 1.0
 315     Content-Transfer-Encoding: 7bit
 316     <BLANKLINE>
 317     Version: 1
 318     <BLANKLINE>
 319     --boundsep
 320     MIME-Version: 1.0
 321     Content-Transfer-Encoding: 7bit
 322     Content-Type: application/octet-stream; charset="us-ascii"
 323     <BLANKLINE>
 324     -----BEGIN PGP MESSAGE-----
 325     ...
 326     -----END PGP MESSAGE-----
 327     <BLANKLINE>
 328     --boundsep--
 329     >>> signedAndEncrypted = m.signAndEncrypt(header)
 330     >>> signedAndEncrypted.set_boundary('boundsep')
 331     >>> print flatten(signedAndEncrypted).replace('\\t', ' '*4) # doctest: +ELLIPSIS, +NORMALIZE_WHITESPACE
 332     Content-Type: multipart/encrypted;
 333         protocol="application/pgp-encrypted";
 334         micalg="pgp-sha1"; boundary="boundsep"
 335     MIME-Version: 1.0
 336     Content-Disposition: inline
 337     <BLANKLINE>
 338     --boundsep
 339     Content-Type: application/pgp-encrypted
 340     MIME-Version: 1.0
 341     Content-Transfer-Encoding: 7bit
 342     <BLANKLINE>
 343     Version: 1
 344     <BLANKLINE>
 345     --boundsep
 346     MIME-Version: 1.0
 347     Content-Transfer-Encoding: 7bit
 348     Content-Type: application/octet-stream; charset="us-ascii"
 349     <BLANKLINE>
 350     -----BEGIN PGP MESSAGE-----
 351     ...
 352     -----END PGP MESSAGE-----
 353     <BLANKLINE>
 354     --boundsep--
 355     """
 356     def __init__(self, body):
 357         self.body = body
 358     def encodedMIMEText(self, body, encoding=None):
 359         if encoding == None:
 360             if type(body) == types.StringType:
 361                 encoding = "us-ascii"
 362             elif type(body) == types.UnicodeType:
 363                 for encoding in ["us-ascii", "iso-8859-1", "utf-8"]:
 364                     try:
 365                         body.encode(encoding)
 366                     except UnicodeError:
 367                         pass
 368                     else:
 369                         break
 370                 assert encoding != None
 371         # Create the message ('plain' stands for Content-Type: text/plain)
 372         if encoding == "us-ascii":
 373             return MIMEText(body)
 374         else:
 375             return MIMEText(body.encode(encoding), 'plain', encoding)
 376     def clearBodyPart(self):
 377         body = self.encodedMIMEText(self.body)
 378         body.add_header('Content-Disposition', 'inline')
 379         return body
 380     def passphrase_arg(self, passphrase=None):
 381         if passphrase == None and PASSPHRASE != None:
 382             passphrase = PASSPHRASE
 383         if passphrase == None:
 384             return (None,'')
 385         return (passphrase, pgp_stdin_passphrase_arg)
 386     def plain(self):
 387         """
 388         text/plain
 389         """
 390         return self.encodedMIMEText(self.body)
 391     def sign(self, header, passphrase=None):
 392         """
 393         multipart/signed
 394           +-> text/plain                 (body)
 395           +-> application/pgp-signature  (signature)
 396         """
 397         passphrase,pass_arg = self.passphrase_arg(passphrase)
 398         body = self.clearBodyPart()
 399         bfile = tempfile.NamedTemporaryFile()
 400         bfile.write(flatten(body))
 401         bfile.flush()
 402
 403         args = replace(pgp_sign_command, 'f', bfile.name)
 404         if PGP_SIGN_AS == None:
 405             pgp_sign_as = '<%s>' % source_email(header)
 406         else:
 407             pgp_sign_as = PGP_SIGN_AS
 408         args = replace(args, 'a', pgp_sign_as)
 409         args = replace(args, 'p', pass_arg)
 410         status,output,error = execute(args, stdin=passphrase)
 411         signature = output
 412
 413         sig = MIMEApplication(_data=signature,
 414                               _subtype='pgp-signature; name="signature.asc"',
 415                               _encoder=encode_7or8bit)
 416         sig['Content-Description'] = 'signature'
 417         sig.set_charset('us-ascii')
 418
 419         msg = MIMEMultipart('signed', micalg='pgp-sha1',
 420                             protocol='application/pgp-signature')
 421         msg.attach(body)
 422         msg.attach(sig)
 423
 424         msg['Content-Disposition'] = 'inline'
 425         return msg
 426     def encrypt(self, header, passphrase=None):
 427         """
 428         multipart/encrypted
 429          +-> application/pgp-encrypted  (control information)
 430          +-> application/octet-stream   (body)
 431         """
 432         body = self.clearBodyPart()
 433         bfile = tempfile.NamedTemporaryFile()
 434         bfile.write(flatten(body))
 435         bfile.flush()
 436
 437         recipients = [replace(pgp_recipient_arg, 'r', recipient)
 438                       for recipient in target_emails(header)]
 439         recipient_string = ' '.join(recipients)
 440         args = replace(pgp_encrypt_only_command, 'R', recipient_string)
 441         args = replace(args, 'f', bfile.name)
 442         if PGP_SIGN_AS == None:
 443             pgp_sign_as = '<%s>' % source_email(header)
 444         else:
 445             pgp_sign_as = PGP_SIGN_AS
 446         args = replace(args, 'a', pgp_sign_as)
 447         status,output,error = execute(args)
 448         encrypted = output
 449
 450         enc = MIMEApplication(_data=encrypted, _subtype='octet-stream',
 451                               _encoder=encode_7or8bit)
 452         enc.set_charset('us-ascii')
 453
 454         control = MIMEApplication(_data='Version: 1\n', _subtype='pgp-encrypted',
 455                                   _encoder=encode_7or8bit)
 456
 457         msg = MIMEMultipart('encrypted', micalg='pgp-sha1',
 458                             protocol='application/pgp-encrypted')
 459         msg.attach(control)
 460         msg.attach(enc)
 461
 462         msg['Content-Disposition'] = 'inline'
 463         return msg
 464     def signAndEncrypt(self, header, passphrase=None):
 465         """
 466         multipart/encrypted
 467          +-> application/pgp-encrypted  (control information)
 468          +-> application/octet-stream   (body)
 469         """
 470         passphrase,pass_arg = self.passphrase_arg(passphrase)
 471         body = self.sign(header, passphrase)
 472         body.__delitem__('Bcc')
 473         bfile = tempfile.NamedTemporaryFile()
 474         bfile.write(flatten(body))
 475         bfile.flush()
 476
 477         recipients = [replace(pgp_recipient_arg, 'r', recipient)
 478                       for recipient in target_emails(header)]
 479         recipient_string = ' '.join(recipients)
 480         args = replace(pgp_encrypt_only_command, 'R', recipient_string)
 481         args = replace(args, 'f', bfile.name)
 482         if PGP_SIGN_AS == None:
 483             pgp_sign_as = '<%s>' % source_email(header)
 484         else:
 485             pgp_sign_as = PGP_SIGN_AS
 486         args = replace(args, 'a', pgp_sign_as)
 487         args = replace(args, 'p', pass_arg)
 488         status,output,error = execute(args, stdin=passphrase)
 489         encrypted = output
 490
 491         enc = MIMEApplication(_data=encrypted, _subtype='octet-stream',
 492                               _encoder=encode_7or8bit)
 493         enc.set_charset('us-ascii')
 494
 495         control = MIMEApplication(_data='Version: 1\n',
 496                                   _subtype='pgp-encrypted',
 497                                   _encoder=encode_7or8bit)
 498
 499         msg = MIMEMultipart('encrypted', micalg='pgp-sha1',
 500                             protocol='application/pgp-encrypted')
 501         msg.attach(control)
 502         msg.attach(enc)
 503
 504         msg['Content-Disposition'] = 'inline'
 505         return msg
 506
 507 def test():
 508     import doctest
 509     doctest.testmod()
 510
 511
 512 if __name__ == '__main__':
 513     from optparse import OptionParser
 514
 515     parser = OptionParser(usage=usage)
 516     parser.add_option('-t', '--test', dest='test', action='store_true',
 517                       help='Run doctests and exit')
 518
 519     parser.add_option('-H', '--header-file', dest='header_filename',
 520                       help='file containing email header', metavar='FILE')
 521     parser.add_option('-B', '--body-file', dest='body_filename',
 522                       help='file containing email body', metavar='FILE')
 523
 524     parser.add_option('-P', '--passphrase-file', dest='passphrase_file',
 525                       help='file containing gpg passphrase', metavar='FILE')
 526     parser.add_option('-p', '--passphrase-fd', dest='passphrase_fd',
 527                       help='file descriptor from which to read gpg passphrase (0 for stdin)',
 528                       type="int", metavar='DESCRIPTOR')
 529
 530     parser.add_option('--mode', dest='mode', default='sign',
 531                       help="One of 'sign', 'encrypt', 'sign-encrypt', or 'plain'.  Defaults to %default.",
 532                       metavar='MODE')
 533
 534     parser.add_option('-a', '--sign-as', dest='sign_as',
 535                       help="The gpg key to sign with (gpg's -u/--local-user)",
 536                       metavar='KEY')
 537
 538     parser.add_option('--output', dest='output', action='store_true',
 539                       help="Don't mail the generated message, print it to stdout instead.")
 540
 541     (options, args) = parser.parse_args()
 542
 543     stdin_used = False
 544
 545     if options.passphrase_file != None:
 546         PASSPHRASE = file(options.passphrase_file, 'r').read()
 547     elif options.passphrase_fd != None:
 548         if options.passphrase_fd == 0:
 549             stdin_used = True
 550             PASSPHRASE = sys.stdin.read()
 551         else:
 552             PASSPHRASE = os.read(options.passphrase_fd)
 553
 554     if options.sign_as:
 555         PGP_SIGN_AS = options.sign_as
 556
 557     if options.test == True:
 558         test()
 559         sys.exit(0)
 560
 561     header = None
 562     if options.header_filename != None:
 563         if options.header_filename == '-':
 564             assert stdin_used == False
 565             stdin_used = True
 566             header = sys.stdin.read()
 567         else:
 568             header = file(options.header_filename, 'r').read()
 569     if header == None:
 570         raise Exception, "missing header"
 571     headermsg = header_from_text(header)
 572     body = None
 573     if options.body_filename != None:
 574         if options.body_filename == '-':
 575             assert stdin_used == False
 576             stdin_used = True
 577             body = sys.stdin.read()
 578         else:
 579             body = file(options.body_filename, 'r').read()
 580     if body == None:
 581         raise Exception, "missing body"
 582
 583     m = EncryptedMessageFactory(body)
 584     if options.mode == "sign":
 585         bodymsg = m.sign(header)
 586     elif options.mode == "encrypt":
 587         bodymsg = m.encrypt(header)
 588     elif options.mode == "sign-encrypt":
 589         bodymsg = m.signAndEncrypt(header)
 590     elif options.mode == "plain":
 591         bodymsg = m.plain()
 592     else:
 593         print "Unrecognized mode '%s'" % options.mode
 594
 595     message = attach_root(headermsg, bodymsg)
 596     if options.output == True:
 597         message = flatten(message)
 598         print message
 599     else:
 600         mail(message, sendmail)