diff options
Diffstat (limited to 'email-print-mime-structure')
| -rwxr-xr-x | email-print-mime-structure | 75 |
1 files changed, 57 insertions, 18 deletions
diff --git a/email-print-mime-structure b/email-print-mime-structure index 27fb532..b7646e0 100755 --- a/email-print-mime-structure +++ b/email-print-mime-structure @@ -32,6 +32,7 @@ something like "cat -n" ''' import os import sys +import enum import email import logging import subprocess @@ -51,6 +52,8 @@ try: except ImportError: argcomplete = None +EncType = enum.Enum('EncType', ['PGPMIME', 'SMIME']) + class MimePrinter(object): def __init__(self, args:Namespace): self.args = args @@ -79,32 +82,66 @@ class MimePrinter(object): print(f'{prefix}{z.get_content_type()}{cset}{disposition}{fname} {nbytes:d} bytes') cryptopayload:Optional[Message] = None - ciphertext:Union[List[Message],str,bytes,None] = None try_pgp_decrypt:bool = self.args.pgpkey or self.args.use_gpg_agent + try_cms_decrypt:bool = self.args.cmskey or self.args.use_gpg_agent if try_pgp_decrypt and \ (parent is not None) and \ (parent.get_content_type().lower() == 'multipart/encrypted') and \ (str(parent.get_param('protocol')).lower() == 'application/pgp-encrypted') and \ (num == 2): - ciphertext = z.get_payload() - if not isinstance(ciphertext, str): - logging.warning('encrypted part was not a leaf mime part somehow') - return - if self.args.pgpkey: - cryptopayload = self.pgpy_decrypt(self.args.pgpkey, ciphertext) - if cryptopayload is None and self.args.use_gpg_agent: - cryptopayload = self.gpg_decrypt(ciphertext) - if cryptopayload is None: - logging.warning(f'Unable to decrypt') - return + cryptopayload = self.decrypt_part(z, EncType.PGPMIME) + + if try_cms_decrypt and \ + cryptopayload is None and \ + z.get_content_type().lower() == 'application/pkcs7-mime' and \ + str(z.get_param('smime-type')).lower() in ['authenveloped-data', + 'enveloped-data']: + cryptopayload = self.decrypt_part(z, EncType.SMIME) if cryptopayload is not None: newprefix = prefix[:-3] + ' ' print(f'{newprefix}↧ (decrypts to)') self.print_tree(cryptopayload, newprefix + '└', z, 0) - - def pgpy_decrypt(self, keys:List[str], ciphertext:str) -> Optional[Message]: + else: + if z.get_content_type().lower() == 'application/pkcs7-mime' and \ + str(z.get_param('smime-type')).lower() == 'signed-data': + bodypart:Union[List[Message],str,bytes,None] = z.get_payload(decode=True) + if isinstance(bodypart, bytes): + unwrapped = self.pipe_transform(bodypart, ['certtool', '--p7-show-data', '--p7-info', '--inder']) + if unwrapped: + newprefix = prefix[:-3] + ' ' + print(f'{newprefix}⇩ (unwraps to)') + self.print_tree(unwrapped, newprefix + '└', z, 0) + else: + logging.warning(f'Unable to unwrap one-part PKCS#7 signed message (maybe try "apt install gnutls-bin")') + + + def decrypt_part(self, msg:Message, flavor:EncType) -> Optional[Message]: + ciphertext:Union[List[Message],str,bytes,None] = msg.get_payload(decode=True) + cryptopayload:Optional[Message] = None + if not isinstance(ciphertext, bytes): + logging.warning('encrypted part was not a leaf mime part somehow') + return None + if flavor == EncType.PGPMIME: + if self.args.pgpkey: + cryptopayload = self.pgpy_decrypt(self.args.pgpkey, ciphertext) + if cryptopayload is None and self.args.use_gpg_agent: + cryptopayload = self.pipe_transform(ciphertext, ['gpg', '--batch', '--decrypt']) + elif flavor == EncType.SMIME: + if self.args.cmskey: + for keyname in self.args.cmskey: + cmd = ['openssl', 'smime', '-decrypt', '-inform', 'DER', '-inkey', keyname] + cryptopayload = self.pipe_transform(ciphertext, cmd) + if cryptopayload: + return cryptopayload + if self.args.use_gpg_agent: + cryptopayload = self.pipe_transform(ciphertext, ['gpgsm', '--batch', '--decrypt']) + if cryptopayload is None: + logging.warning(f'Unable to decrypt') + return cryptopayload + + def pgpy_decrypt(self, keys:List[str], ciphertext:bytes) -> Optional[Message]: if pgpy is None: logging.warning(f'Python module pgpy is not available, not decrypting (try "apt install python3-pgpy")') return None @@ -121,13 +158,13 @@ class MimePrinter(object): pass return None - def gpg_decrypt(self, ciphertext:str) -> Optional[Message]: + def pipe_transform(self, ciphertext:bytes, cmd:List[str]) -> Optional[Message]: inp:int outp:int inp, outp = os.pipe() - with open(outp, 'w') as outf: + with open(outp, 'wb') as outf: outf.write(ciphertext) - out:subprocess.CompletedProcess[bytes] = subprocess.run(['gpg', '--batch', '--decrypt'], + out:subprocess.CompletedProcess[bytes] = subprocess.run(cmd, stdin=inp, capture_output=True) if out.returncode == 0: @@ -157,7 +194,9 @@ def main() -> None: parser:ArgumentParser = ArgumentParser(description='Read RFC2822 MIME message from stdin and emit a tree diagram to stdout.', epilog="Example: email-print-mime-structure <message.eml") parser.add_argument('--pgpkey', metavar='KEYFILE', action='append', - help='OpenPGP Transferable Secret Key for decrypting') + help='OpenPGP Transferable Secret Key for decrypting PGP/MIME') + parser.add_argument('--cmskey', metavar='KEYFILE', action='append', + help='X.509 Private Key for decrypting S/MIME') parser.add_argument('--use-gpg-agent', action='store_true', help='Ask local GnuPG installation for decryption') parser.add_argument('--no-use-gpg-agent', action='store_false', |