generate-test-vectors

#!/usr/bin/env python3
'''Deterministically generate test vectors for draft-protected-headers.

We want each run to produce the same specific output, which means
hard-coding things (like timestamps, time zones, MIME boundaries,
message-ids) which are likely to be vary across real-world conditions.

Pains are taken to minimize the total amount of output, while still
preserving a plausible, minimal, correctly-formed message.
'''

import os
import sys
import codecs
import hashlib
import logging
import datetime
import subprocess
import email.policy
import email.message

import pgpy #type: ignore
from cryptography.hazmat.backends import default_backend
from cryptography import x509

from typing import Dict, Union, TextIO, Optional, List, IO
from tempfile import NamedTemporaryFile

cfg:Dict[str,Dict[str,Union[bool,bytes,int]]] = {
    'pgpmime-signed': {
        'encrypt': False,
        'minute': 0,
    },
    'smime-multipart-signed': {
        'encrypt': False,
        'smime': True,
        'minute': 3,
    },
    'smime-onepart-signed': {
        'encrypt': False,
        'smime': True,
        'multipart': False,
        'minute': 6,
    },
    'pgpmime-sign+enc': {
        'encrypt': True,
        'multilayer': False,
        'legacy': False,
        'multipart': False,
        'sk': b'8df4b2d27d5637138ac6de46415661be0bd01ed12ecf8c1db22a33cf3ede82f2',
        'minute': 9,
    },
    'pgpmime-layered': {
        'encrypt': True,
        'multilayer': True,
        'legacy': False,
        'multipart': False,
        'sk': b'5e67165ed1516333daeba32044f88fd75d4a9485a563d14705e41d31fb61a9e9',
        'minute': 12,
    },
    'smime-sign+enc': {
        'encrypt': True,
        'legacy': False,
        'multipart': False,
        'smime': True,
        'sk': b'12e2551896f77e24ce080153cda27dddd789d399bdd87757e65655d956f5f0b7',
        'minute': 15,
    },
    'pgpmime-sign+enc+legacy-disp': {
        'encrypt': True,
        'multilayer': False,
        'legacy': True,
        'multipart': False,
        'sk': b'95a71b0e344cce43a4dd52c5fd01deec5118290bfd0792a8a733c653a12d223e',
        'minute': 18,
    },
    'pgpmime-layered+legacy-disp': {
        'encrypt': True,
        'multilayer': True,
        'legacy': True,
        'multipart': False,
        'sk': b'b346a2a50fa0cf62895b74e8c0d2ad9e3ee1f02b5d564c77d879caaee7a0aa70',
        'minute': 21,
    },
    'smime-sign+enc+legacy-disp': {
        'encrypt': True,
        'legacy': True,
        'multipart': False,
        'smime': True,
        'sk': b'09e8f2a19d9e97deea7d51ee7d401be8763ab0377b6f30a68206e0bed4a0baec',
        'minute': 24,
    },
    'smime-enc+legacy-disp': {
        'encrypt': True,
        'legacy': True,
        'multipart': False,
        'signed': False,
        'smime': True,
        'sk': b'e94f6aaef7f14d6ceeac770c46d7f4885e81fbeaf1462d0fdadfce6c581525e2',
        'minute': 27,
    },
    'pgpmime-enc+legacy-disp': {
        'encrypt': True,
        'legacy': True,
        'multipart': False,
        'signed': False,
        'sk': b'4f3e7e3cb4a49747f88d232601fa98a29d7427e8f80882464cfbca3dcb847356',
        'minute': 30,
    },
    'unfortunately-complex': {
        'encrypt': True,
        'multilayer': True,
        'legacy': True,
        'multipart': True,
        'sk': b'1c489cfad9f3c0bf3214bf34e6da42b7f64005e59726baa1b17ffdefe6ecbb52',
        'minute': 33,
    },
}


def usage(to:TextIO=sys.stdout) -> None:
    subcmds = '\n  '.join(list(cfg.keys()))
    print(f'''Usage: {sys.argv[0]} SUBCMD
where SUBCMD is one of:
  help
  list-vectors
  {subcmds}''', file=to)

# We want maxheaderlen=72 here so that the example fits nicely
# in an Internet Draft.  But it is risky -- i think it could
# break a signed message with long headers.  We get away with
# it because our Cryptographic Payload is already narrow.
MAXHEADERLEN=72
    
# from https://tools.ietf.org/html/draft-bre-openpgp-samples-00#section-2.2:
alice_sec = '''-----BEGIN PGP PRIVATE KEY BLOCK-----
Comment: Alice's OpenPGP Transferable Secret Key

lFgEXEcE6RYJKwYBBAHaRw8BAQdArjWwk3FAqyiFbFBKT4TzXcVBqPTB3gmzlC/U
b7O1u10AAP9XBeW6lzGOLx7zHH9AsUDUTb2pggYGMzd0P3ulJ2AfvQ4RtCZBbGlj
ZSBMb3ZlbGFjZSA8YWxpY2VAb3BlbnBncC5leGFtcGxlPoiQBBMWCAA4AhsDBQsJ
CAcCBhUKCQgLAgQWAgMBAh4BAheAFiEE64W7X6M6deFelE5j8jFVDE9H444FAl2l
nzoACgkQ8jFVDE9H447pKwD6A5xwUqIDprBzrHfahrImaYEZzncqb25vkLV2arYf
a78A/R3AwtLQvjxwLDuzk4dUtUwvUYibL2sAHwj2kGaHnfICnF0EXEcE6RIKKwYB
BAGXVQEFAQEHQEL/BiGtq0k84Km1wqQw2DIikVYrQrMttN8d7BPfnr4iAwEIBwAA
/3/xFPG6U17rhTuq+07gmEvaFYKfxRB6sgAYiW6TMTpQEK6IeAQYFggAIBYhBOuF
u1+jOnXhXpROY/IxVQxPR+OOBQJcRwTpAhsMAAoJEPIxVQxPR+OOWdABAMUdSzpM
hzGs1O0RkWNQWbUzQ8nUOeD9wNbjE3zR+yfRAQDbYqvtWQKN4AQLTxVJN5X5AWyb
Pnn+We1aTBhaGa86AQ==
=n8OM
-----END PGP PRIVATE KEY BLOCK-----
'''
(alice_key, _) = pgpy.PGPKey.from_blob(alice_sec)

# from https://tools.ietf.org/html/draft-bre-openpgp-samples-00#section-3.1:
bob_pub = '''-----BEGIN PGP PUBLIC KEY BLOCK-----
Comment: Bob's OpenPGP certificate

mQGNBF2lnPIBDAC5cL9PQoQLTMuhjbYvb4Ncuuo0bfmgPRFywX53jPhoFf4Zg6mv
/seOXpgecTdOcVttfzC8ycIKrt3aQTiwOG/ctaR4Bk/t6ayNFfdUNxHWk4WCKzdz
/56fW2O0F23qIRd8UUJp5IIlN4RDdRCtdhVQIAuzvp2oVy/LaS2kxQoKvph/5pQ/
5whqsyroEWDJoSV0yOb25B/iwk/pLUFoyhDG9bj0kIzDxrEqW+7Ba8nocQlecMF3
X5KMN5kp2zraLv9dlBBpWW43XktjcCZgMy20SouraVma8Je/ECwUWYUiAZxLIlMv
9CurEOtxUw6N3RdOtLmYZS9uEnn5y1UkF88o8Nku890uk6BrewFzJyLAx5wRZ4F0
qV/yq36UWQ0JB/AUGhHVPdFf6pl6eaxBwT5GXvbBUibtf8YI2og5RsgTWtXfU7eb
SGXrl5ZMpbA6mbfhd0R8aPxWfmDWiIOhBufhMCvUHh1sApMKVZnvIff9/0Dca3wb
vLIwa3T4CyshfT0AEQEAAbQhQm9iIEJhYmJhZ2UgPGJvYkBvcGVucGdwLmV4YW1w
bGU+iQHOBBMBCgA4AhsDBQsJCAcCBhUKCQgLAgQWAgMBAh4BAheAFiEE0aZuGiOx
gsmYD3iM+/zIKgFeczAFAl2lnvoACgkQ+/zIKgFeczBvbAv/VNk90a6hG8Od9xTz
XxH5YRFUSGfIA1yjPIVOnKqhMwps2U+sWE3urL+MvjyQRlyRV8oY9IOhQ5Esm6DO
ZYrTnE7qVETm1ajIAP2OFChEc55uH88x/anpPOXOJY7S8jbn3naC9qad75BrZ+3g
9EBUWiy5p8TykP05WSnSxNRt7vFKLfEB4nGkehpwHXOVF0CRNwYle42bg8lpmdXF
DcCZCi+qEbafmTQzkAqyzS3nCh3IAqq6Y0kBuaKLm2tSNUOlZbD+OHYQNZ5Jix7c
ZUzs6Xh4+I55NRWl5smrLq66yOQoFPy9jot/Qxikx/wP3MsAzeGaZSEPc0fHp5G1
6rlGbxQ3vl8/usUV7W+TMEMljgwd5x8POR6HC8EaCDfVnUBCPi/Gv+egLjsIbPJZ
ZEroiE40e6/UoCiQtlpQB5exPJYSd1Q1txCwueih99PHepsDhmUQKiACszNU+RRo
zAYau2VdHqnRJ7QYdxHDiH49jPK4NTMyb/tJh2TiIwcmsIpGuQGNBF2lnPIBDADW
ML9cbGMrp12CtF9b2P6z9TTT74S8iyBOzaSvdGDQY/sUtZXRg21HWamXnn9sSXvI
DEINOQ6A9QxdxoqWdCHrOuW3ofneYXoG+zeKc4dC86wa1TR2q9vW+RMXSO4uImA+
Uzula/6k1DogDf28qhCxMwG/i/m9g1c/0aApuDyKdQ1PXsHHNlgd/Dn6rrd5y2AO
baifV7wIhEJnvqgFXDN2RXGjLeCOHV4Q2WTYPg/S4k1nMXVDwZXrvIsA0YwIMgIT
86Rafp1qKlgPNbiIlC1g9RY/iFaGN2b4Ir6GDohBQSfZW2+LXoPZuVE/wGlQ01rh
827KVZW4lXvqsge+wtnWlszcselGATyzqOK9LdHPdZGzROZYI2e8c+paLNDdVPL6
vdRBUnkCaEkOtl1mr2JpQi5nTU+gTX4IeInC7E+1a9UDF/Y85ybUz8XV8rUnR76U
qVC7KidNepdHbZjjXCt8/Zo+Tec9JNbYNQB/e9ExmDntmlHEsSEQzFwzj8sxH48A
EQEAAYkBtgQYAQoAIBYhBNGmbhojsYLJmA94jPv8yCoBXnMwBQJdpZzyAhsMAAoJ
EPv8yCoBXnMw6f8L/26C34dkjBffTzMj5Bdzm8MtF67OYneJ4TQMw7+41IL4rVcS
KhIhk/3Ud5knaRtP2ef1+5F66h9/RPQOJ5+tvBwhBAcUWSupKnUrdVaZQanYmtSx
cVV2PL9+QEiNN3tzluhaWO//rACxJ+K/ZXQlIzwQVTpNhfGzAaMVV9zpf3u0k14i
tcv6alKY8+rLZvO1wIIeRZLmU0tZDD5HtWDvUV7rIFI1WuoLb+KZgbYn3OWjCPHV
dTrdZ2CqnZbG3SXw6awH9bzRLV9EXkbhIMez0deCVdeo+wFFklh8/5VK2b0vk/+w
qMJxfpa1lHvJLobzOP9fvrswsr92MA2+k901WeISR7qEzcI0Fdg8AyFAExaEK6Vy
jP7SXGLwvfisw34OxuZr3qmx1Sufu4toH3XrB7QJN8XyqqbsGxUCBqWif9RSK4xj
zRTe56iPeiSJJOIciMP9i2ldI+KgLycyeDvGoBj0HCLO3gVaBe4ubVrj5KjhX2PV
NEJd3XZRzaXZE2aAMQ==
=NXei
-----END PGP PUBLIC KEY BLOCK-----
'''
(bob_key, _) = pgpy.PGPKey.from_blob(bob_pub)

class IdX509:
    def __init__(self, cert:Union[bytes,str], key:Union[bytes,str,None]=None):
        self._cert:bytes
        self._key:Optional[bytes] = None
        if isinstance(cert, str):
            self._cert = cert.encode()
        else:
            self._cert = cert
        if isinstance(key, str):
            self._key = key.encode()
        elif isinstance(key, bytes):
            self._key = key

        self._x509 = x509.load_pem_x509_certificate(self._cert, default_backend())

    @property
    def name(self) -> str:
        return str(self._x509.subject.get_attributes_for_oid(x509.oid.NameOID.COMMON_NAME)[0].value)

    @property
    def mailaddr(self) -> str:
        sans = self._x509.extensions.get_extension_for_oid(x509.oid.ExtensionOID.SUBJECT_ALTERNATIVE_NAME).value
        if not isinstance(sans, x509.SubjectAlternativeName):
            raise ValueError("failed to get subjectAltName as expected")
        # ignoring due to https://github.com/python/typeshed/issues/3519:
        return str(sans.get_values_for_type(x509.general_name.RFC822Name)[0]) #type: ignore

    def encrypt(self, data:email.message.MIMEPart, when:datetime.datetime, others:List["IdX509"], msg:email.message.MIMEPart) -> email.message.MIMEPart:
        txt:str = data.as_string(policy=email.policy.EmailPolicy(linesep='\r\n'))
        recips:List[IO[bytes]] = []
        me = NamedTemporaryFile(mode='wb')
        me.write(self._cert)
        me.flush()
        recips.append(me)
        for other in others:
            o = NamedTemporaryFile()
            o.write(other._cert)
            o.flush()
            recips.append(o)
        inp, outp = os.pipe()
        with open(outp, 'wb') as outf:
            outf.write(txt.encode())
        out:subprocess.CompletedProcess[bytes]
        utcwhen:datetime.datetime = datetime.datetime.utcfromtimestamp(when.timestamp())
        cmd = ['faketime', utcwhen.strftime('%Y-%m-%d %H:%M:%S'),
               'openssl', 'smime', '-encrypt',
               '-outform', 'DER']
        for recip in recips:
            cmd.append(recip.name)
        out = subprocess.run(cmd, stdin=inp, capture_output=True, env={'TZ': 'UTC'})
        if out.returncode != 0:
            logging.warning(cmd)
            logging.warning(out.stderr)
            raise Exception(f'certtool --p7-detached-sign failed with {out.returncode}')
        msg.set_content(out.stdout, 'application', 'pkcs7-mime')
        msg.set_param('name','smime.p7m')
        msg.set_param('smime-type','enveloped-data')
        return msg

    def detached_sig(self, data:email.message.MIMEPart, when:datetime.datetime) -> email.message.MIMEPart:
        txt:str = data.as_string(policy=email.policy.EmailPolicy(linesep='\r\n'))
        out = self._sign(txt.encode(), when, '--p7-detached-sign')
        sigpart = email.message.MIMEPart(email.policy.EmailPolicy(max_line_length=64))
        sigpart.set_content(out, 'application', 'pkcs7-signature')
        sigpart.set_param('name','smime.p7s')
        return sigpart

    def onepart_sig(self, data:email.message.MIMEPart, when:datetime.datetime, msg:email.message.MIMEPart) -> email.message.MIMEPart:
        txt:str = data.as_string(policy=email.policy.EmailPolicy(linesep='\r\n'))
        out = self._sign(txt.encode(), when, '--p7-sign')
        msg.set_content(out, 'application', 'pkcs7-mime')
        msg.set_param('name','smime.p7m')
        msg.set_param('smime-type','signed-data')
        return msg

    def _sign(self, data:bytes, when:datetime.datetime, subcmd:str) -> bytes:
        if self._key is None:
            raise AttributeError('no key, cannot sign')
        keyf = NamedTemporaryFile()
        keyf.write(self._key)
        certf = NamedTemporaryFile()
        certf.write(self._cert)
        keyf.flush()
        certf.flush()
        inp, outp = os.pipe()
        with open(outp, 'wb') as outf:
            outf.write(data)
        out:subprocess.CompletedProcess[bytes]
        utcwhen:datetime.datetime = datetime.datetime.utcfromtimestamp(when.timestamp())
        cmd = ['faketime', utcwhen.strftime('%Y-%m-%d %H:%M:%S'),
               'certtool', subcmd,
               '--p7-time', '--outraw',
               '--hash', 'SHA256',
               '--load-privkey', keyf.name,
               '--load-certificate', certf.name]
        out = subprocess.run(cmd, stdin=inp, capture_output=True, env={'TZ': 'UTC'})
        if out.returncode != 0:
            logging.warning(cmd)
            logging.warning(out.stderr)
            raise Exception(f'certtool --p7-detached-sign failed with {out.returncode}')
        return out.stdout

# from https://www.ietf.org/id/draft-dkg-lamps-samples-01.html#name-alices-private-key-material
alice_smime_key = '''-----BEGIN RSA PRIVATE KEY-----
MIIEogIBAAKCAQEAw+6t+WXRtiQM8yRjWQ2fbFewCodIZUX6BY02TeZuEXoEAGEs
moON6LlotcUTdGr39FE2K8IytOKkXVexswgAqBCqv8YjVDrI3yV82wrm5Td32TDl
w7ISigak4ZSu+UowPQs8YO3oxqImP4onZNHvdZ3it9EggmgUyZX0dmQ6z5O9yDzH
pLMaE2rXxfYcPXQwPvx4tcqbTf2htEP7PYnBa8a+sts0F7I7kD5ozGYI9dGg/XGs
1lYEWAoH5YZgNFdbkJdcKG2FPAwFcVZ/hoGm6soxkDKMrYSCtBp+fqH8MV11DP82
1PoOvtSEnaF8UURbaths2yKpAB2WUJvgW5xa4QIDAQABAoIBAA7vrwuIG4iLDwGq
EHjFdRXJSX5D+dzejMTHkxA1NMbYSl3NCp1s0fCf0b+pmmYRkX1qg3qqfzsS2/zR
ppZDUel9+8ZK0H6nTJDWRsJb/mYS6GwCMkHM3WTwRLl9oCkY4ryEksHA4THjQo8t
dPtWla6drp7crmHClXMYn143HdSdCIB9StRPkSgyHjyFLOThReOog2Nsm7eShmov
7WkMuESFku5OHFPLUw5FyLEzHJar8ZI7qYbT7X6IamXOf9aTMPDA1rqAcix+4KQa
zF3cNY1xgq/yIvtsv6oyknTStw1i3i46PWzMWf845Eayunrg8e6F3hWt7zndjXWQ
Jg/gAAECgYEA3SLlO2tGdb5gWHwzzZAnTzBMo1Z3toEN25LetuSmY7mxkjMTRDAi
5VOdpSXrVFaT5r8qwU9yFEm+OuB6k52CVbTE1Fp96JlbzYjZnKaLn5OG8+HSLdtn
1vj1XyCGRDJKJ8GaZpZp+WvBfp6449WpSgupXMdIOM8jfekgTEh6rgECgYEA4tKM
Da3tFEEyVy9ZSxZV9ep9dhE7kmVQnr2pvt2YfJTiKnSo2kkj/qKoMi2PhS8ZO0JQ
J90bDngqI5sIo/OGi+hwYRmcKCrvfnfJUEq3v+3BFQYPDfwktgiBu5TGDNimFA2t
l+23SwwCPfjPh5frk8GTq0IslRhXY3djNPhhbOECgYAojSegN9HZ8alVUKFnRtIO
kXrcURTu4MebxlkVDOT+UKUhfEBCNtmPWEAGcueutZm1rMS4Yks3MTazMUsJGs81
zEpz7ow8RTMyg6/0LA5amwEaZATY5+0o3MqSQTKd+uLiW3xm55pTZNE82PpqvVmn
/G94VgsGb+XARynnEzt8AQKBgDER356t+9Yf7KYT5jtqT5pt6kp6m+ql5HUTDv/t
rKl3BB6vMkBXBmR2B/EjDiN/9vNs+y5ElS/iKyucxJfDfV4TIQzAn5nJABraC0FF
iM8KvnSv5N3fqImA+Z/9JYNt8y/vbZiqoranmGyTwUHSSfKjNDEelcqDg5RPJbU1
7s3BAoGAdqDEx0K1sW/e0pOtb97fBNIRgUemSUctUiaV1imwIku1wuxVvD8z92xh
g0DszHZfhSIvZwrhxF0VqPEgh1mDWVfuSHG1g74gDyPy5p3OnEnrk4bloBhXit2Z
pUSPj7ME4rNqAEXlfdVUPq4T1Yq95lDMafQlCmUZU0DnuAy19dc=
-----END RSA PRIVATE KEY-----
'''


# from https://www.ietf.org/id/draft-dkg-lamps-samples-01.html#name-alices-end-entity-certifica
alice_smime_cert = '''-----BEGIN CERTIFICATE-----
MIIDbjCCAlagAwIBAgIUZ4K0WXNSS8H0cUcZavD9EYqqTAswDQYJKoZIhvcNAQEN
BQAwLTErMCkGA1UEAxMiU2FtcGxlIExBTVBTIENlcnRpZmljYXRlIEF1dGhvcml0
eTAgFw0xOTExMjAwNjU0MThaGA8yMDUyMDkyNzA2NTQxOFowGTEXMBUGA1UEAxMO
QWxpY2UgTG92ZWxhY2UwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQDD
7q35ZdG2JAzzJGNZDZ9sV7AKh0hlRfoFjTZN5m4RegQAYSyag43ouWi1xRN0avf0
UTYrwjK04qRdV7GzCACoEKq/xiNUOsjfJXzbCublN3fZMOXDshKKBqThlK75SjA9
Czxg7ejGoiY/iidk0e91neK30SCCaBTJlfR2ZDrPk73IPMeksxoTatfF9hw9dDA+
/Hi1yptN/aG0Q/s9icFrxr6y2zQXsjuQPmjMZgj10aD9cazWVgRYCgflhmA0V1uQ
l1wobYU8DAVxVn+GgabqyjGQMoythIK0Gn5+ofwxXXUM/zbU+g6+1ISdoXxRRFtq
2GzbIqkAHZZQm+BbnFrhAgMBAAGjgZcwgZQwDAYDVR0TAQH/BAIwADAeBgNVHREE
FzAVgRNhbGljZUBzbWltZS5leGFtcGxlMBMGA1UdJQQMMAoGCCsGAQUFBwMEMA8G
A1UdDwEB/wQFAwMHoAAwHQYDVR0OBBYEFKwuVFqk/VUYry7oZkQ40SXR1wB5MB8G
A1UdIwQYMBaAFLdSTXPAiD2yw3paDPOU9/eAonfbMA0GCSqGSIb3DQEBDQUAA4IB
AQB76o4Yz7yrVSFcpXqLrcGtdI4q93aKCXECCCzNQLp4yesh6brqaZHNJtwYcJ5T
qbUym9hJ70iJE4jGNN+yAZR1ltte0HFKYIBKM4EJumG++2hqbUaLz4tl06BHaQPC
v/9NiNY7q9R9c/B6s1YzHhwqkWht2a+AtgJ4BkpG+g+MmZMQV/Ao7RwLFKJ9OlMW
LBmEXFcpIJN0HpPasT0nEl/MmotSu+8RnClAi3yFfyTKb+8rD7VxuyXetqDZ6dU/
9/iqD/SZS7OQIjywtd343mACz3B1RlFxMHSA6dQAf2btGumqR0KiAp3KkYRAePoa
JqYkB7Zad06ngFl0G0FHON+7
-----END CERTIFICATE-----
'''
alice_x509 = IdX509(alice_smime_cert, alice_smime_key)

# from https://www.ietf.org/id/draft-dkg-lamps-samples-01.html#name-bobs-end-entity-certificate
bob_smime_cert = '''-----BEGIN CERTIFICATE-----
MIIDaTCCAlGgAwIBAgIUIlPuMG0CCx8CzfXJwT4633mmG8IwDQYJKoZIhvcNAQEN
BQAwLTErMCkGA1UEAxMiU2FtcGxlIExBTVBTIENlcnRpZmljYXRlIEF1dGhvcml0
eTAgFw0xOTExMjAwNjU0MThaGA8yMDUyMDkyNzA2NTQxOFowFjEUMBIGA1UEAxML
Qm9iIEJhYmJhZ2UwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQDCZjlu
Li00rpoCsq2s8SHqb91QPP5bdfzfaJg/G61lHUhfavEX9zZluyMwPPE50wqwV2RJ
X5dg0kStyH9s9Ja5D59pPnX8oJJ7XEqNKwxqSfJt7lRmM8BrDvSP55iP7Ofx+O+2
MzVA4tA6WUaUy2j9984CMmXH/CHjBK/+w21vSTmzFVGmeTqxxHONbd2zOqQ6Yqr/
LBaHjAWl+tj9Q+2nIjEQFKlWs6vZll3Xwid6+dAxrtpEO5rIpKZcbn40qT1pyDpr
ylNk8h3P90nwrOISpdlAJ2p71ZDdLfLd8c6qZGBPjmHwTUnjmH0oy33uBukT73RU
W6raD8MwM4AhQ4ETAgMBAAGjgZUwgZIwDAYDVR0TAQH/BAIwADAcBgNVHREEFTAT
gRFib2JAc21pbWUuZXhhbXBsZTATBgNVHSUEDDAKBggrBgEFBQcDBDAPBgNVHQ8B
Af8EBQMDB6AAMB0GA1UdDgQWBBQBrAKQ6Dj0kN4Z7pXzMnThZgAopzAfBgNVHSME
GDAWgBS3Uk1zwIg9ssN6WgzzlPf3gKJ32zANBgkqhkiG9w0BAQ0FAAOCAQEAa/tJ
ZPgdlmc7Zbn5bccc1TXNn8qBhECGHma4iSTWczDUmsNjezmDNniM3hs8QOqUZvx4
ey6diTlEngrKZ8bnwsX03k9Bn8UDPT5Y5sbxwEHpwKew41LRiLPOZFSh3DzCKYS7
HDSXJsJEGop1AwzKxtRss06C35g4ELK0Q2MwLw1u95f0+rC4q+vYndS9NzFyS3Bj
MIt37gN+Yy8h/r2wvtPVJ40mYNGmtQhdNuYnr56LOuFMmGiMIYXE8owo6L/kzCcy
YxxCy71lbnBOWLGcJz4HmRMdWJMRDV+mgLmTNnN8mPltgQU9gE3KNrYcST9v2kk+
N+cfxLhC0caHFL5G8g==
-----END CERTIFICATE-----
'''
bob_x509 = IdX509(bob_smime_cert)

def get_pgp_sigpart(pgpsig:pgpy.PGPSignature) -> email.message.MIMEPart:
    sigpart = email.message.MIMEPart()
    sigpart.set_content(str(pgpsig).strip())
    sigpart.set_type('application/pgp-signature')
    sigpart.set_charset(None) # the test vector data is 7-bit clean
    del sigpart['MIME-Version'] # MIME-Version on subparts is meaningless
    del sigpart['Content-Transfer-Encoding'] # ASCII-armored PGP Signature data is 7-bit clean
    return sigpart

def signed(params:Dict[str,Union[bool,bytes,int]], msgid:str) -> email.message.Message:
    minute = params['minute']
    if not isinstance(minute, int):
        raise TypeError('"minute" must be an integer!')
    # seconds since the unix epoch
    if params.get('smime', False):
        # smime signatures are later because the certs are later
        posixtime = (1574813489 // 3600)*3600 + 60*minute
    else:
        posixtime = (1571577491 // 3600)*3600 + 60*minute
    # America/New_York during DST:
    tz = datetime.timezone(datetime.timedelta(hours=-4))
    # 2019-10-20T09:18:11-0400
    when = datetime.datetime.fromtimestamp(posixtime, tz)
    whenstring = when.strftime('%a, %d %b %Y %T %z').strip()

    # message was received 17 seconds after it was generated:
    rcvd = datetime.datetime.fromtimestamp(posixtime + 17, tz)
    rcvdstring = rcvd.strftime('%a, %d %b %Y %T %z (%Z)').strip()

    # make the Cryptographic Payload:
    payload = email.message.MIMEPart()
    dotsig_separator = '-- ' # this is a totally different kind of "signature"
    payload.set_content(f'''Bob, we need to cancel this contract.

Please start the necessary processes to make that happen today.

(this is the '{msgid}' message)

Thanks, Alice
{dotsig_separator}
Alice Lovelace
President
Example Corp''')
    payload.set_type('text/plain')
    payload.set_charset('us-ascii') # the test vector data is 7-bit clean
    del payload['MIME-Version'] # MIME-Version on subparts is meaningless
    del payload['Content-Transfer-Encoding'] # the test vector data is 7-bit clean
    payload.set_param('protected-headers', 'v1')
    # place intended headers on the payload:
    if params.get('smime', False):
        payload['From'] = f'{alice_x509.name} <{alice_x509.mailaddr}>'
        payload['To'] = f'{bob_x509.name} <{bob_x509.mailaddr}>'
    else:
        payload['From'] = f'{alice_key.userids[0].name} <{alice_key.userids[0].email}>'
        payload['To'] = f'{bob_key.userids[0].name} <{bob_key.userids[0].email}>'
    payload['Date'] = whenstring
    payload['Subject'] = 'The FooCorp contract'
    payload['Message-ID'] = f'<{msgid}@protected-headers.example>'

    msg = email.message.MIMEPart(policy=email.policy.EmailPolicy(max_line_length=64))
    # Example transit header that is not part of the protected headers:
    msg['Received'] = f'from localhost (localhost [127.0.0.1]); {rcvdstring}'

    if params.get('smime', False):
        if params.get('multipart', True):
            sigpart = alice_x509.detached_sig(payload, when)
            micalg = 'sha-256'
        else:
            msg = alice_x509.onepart_sig(payload, when, msg)
            msg['MIME-Version'] = '1.0'
    else:
        # Note that https://github.com/SecurityInnovation/PGPy/issues/291 needs to be fixed
        # in order for the signature creation time to match the message Date header:
        pgpsig = alice_key.sign(pgpy.PGPMessage.new(str(payload), cleartext=True), created=when)
        sigpart = get_pgp_sigpart(pgpsig)
        micalg = f'pgp-{pgpsig.hash_algorithm.name.lower()}'

    if params.get('multipart', True):
        msg.set_type('multipart/signed')
        # making up an arbitrary MIME boundary based on the Message-ID:
        msg.set_boundary(hashlib.sha256(payload['Message-ID'].encode()).hexdigest()[:3])
        msg.set_param('protocol', sigpart.get_content_type())
        msg.set_param('micalg', micalg)
        msg.attach(payload)
        msg.attach(sigpart)

    # ensure that all non-Content headers from the payload are also on
    # the message:
    for h in payload.keys():
        if not h.lower().startswith('content-'): # don't transfer Content-* headers
            if msg.get(h, None) != payload[h]: # don't duplicate headers that already exist
                msg[h] = payload[h]
    return msg

def cleanpart(part:email.message.MIMEPart) -> None:
    del part['MIME-Version'] # MIME-Version on subparts is meaningless
    del part['Content-Transfer-Encoding'] # the test vector data is 7-bit clean

def make_payload(multipart:bool, msgid:str) -> email.message.MIMEPart:
    payload:email.message.MIMEPart = email.message.MIMEPart()
    dotsig_separator = '-- ' # this is a totally different kind of "signature"
    txt = f'''Hi Bob!

I just signed the contract with BarCorp and they've set us up with
an account on their system for testing.

The account information is:

        Site: https://barcorp.example/
    Username: examplecorptest
    Password: correct-horse-battery-staple

Please get the account set up and apply the test harness.

Let me know when you've got some results.

(this is the '{msgid}' message)

Thanks, Alice
{dotsig_separator}
Alice Lovelace
President
Example Corp'''

    if not multipart:
        payload.set_content(txt)
        payload.set_type('text/plain')
        payload.set_charset('us-ascii')
        cleanpart(payload)
    else:
        html = f'''<html><head></head><body><p>Hi Bob!
</p><p>
I just signed the contract with BarCorp and they've set us up with
 an account on their system for testing.
</p><p>
The account information is:
</p><dl>
<dt>Site</dt><dd>
<a href="https://barcorp.example/">https://barcorp.example/</a>
</dd>
<dt>Username</dt><dd><tt>examplecorptest</tt></dd>
<dt>Password</dt><dd>correct-horse-battery-staple</dd>
</dl><p>
Please get the account set up and apply the test harness.
</p><p>
Let me know when you've got some results.
</p><p>
(this is the '{msgid}' message)
</p><p>
Thanks, Alice<br/>
{dotsig_separator}<br/>
Alice Lovelace<br/>
President<br/>
Example Corp<br/>
</p></body></html>'''
        attachment = f'''diff -ruN a/testharness.cfg b/testharness.cfg
--- a/testharness.cfg
+++ b/testharness.cfg
@@ -13,3 +13,8 @@
 endpoint = https://openpgp.example/test/
 username = testuser
 password = MJVMZlHR75mILg
+
+[barcorp]
+endpoint = https://barcorp.example/
+username = examplecorptest
+password = correct-horse-battery-staple
'''
        txtpart:email.message.MIMEPart = email.message.MIMEPart()
        txtpart.set_content(txt)
        txtpart.set_type('text/plain')
        txtpart.set_charset('us-ascii')
        cleanpart(txtpart)

        htmlpart:email.message.MIMEPart = email.message.MIMEPart()
        htmlpart.set_content(html)
        htmlpart.set_type('text/html')
        htmlpart.set_charset('us-ascii')
        cleanpart(htmlpart)

        altpart:email.message.MIMEPart = email.message.MIMEPart()
        altpart.set_type('multipart/alternative')
        altpart.set_boundary(hashlib.sha256('multipart-alternative'.encode()).hexdigest()[:3])
        cleanpart(altpart)
        altpart.attach(txtpart)
        altpart.attach(htmlpart)
        
        attachmentpart:email.message.MIMEPart = email.message.MIMEPart()
        attachmentpart.set_content(attachment)
        attachmentpart.set_type('text/x-diff')
        attachmentpart.set_charset('us-ascii')
        cleanpart(attachmentpart)
        attachmentpart['Content-Disposition'] = 'inline; filename="testharness-config.diff"'
        
        payload.set_type('multipart/mixed')
        payload.set_boundary(hashlib.sha256('multipart-mixed'.encode()).hexdigest()[:3])
        cleanpart(payload)
        payload.attach(altpart)
        payload.attach(attachmentpart)
    return payload

def signed_encrypted(msgid:str, params:Dict[str,Union[bool,bytes,int]]) -> email.message.Message:
    minute = params['minute']
    if not isinstance(minute, int):
        raise TypeError('"minute" must be an integer!')
    # seconds since the unix epoch
    if params.get('smime', False):
        # smime signatures are later because the certs are later
        posixtime = (1574843489 // 3600)*3600 + 60*minute
    else:
        posixtime = (1571667491 // 3600)*3600 + 60*minute
    # America/Los_Angeles during DST:
    tz = datetime.timezone(datetime.timedelta(hours=-7))
    # 2019-10-21T07:18:11-0700 for PGP/MIME
    when = datetime.datetime.fromtimestamp(posixtime, tz)
    whenstring = when.strftime('%a, %d %b %Y %T %z').strip()

    # message was received 28 seconds after it was generated:
    rcvd = datetime.datetime.fromtimestamp(posixtime + 28, tz)
    rcvdstring = rcvd.strftime('%a, %d %b %Y %T %z (%Z)').strip()

    # make the Cryptographic Payload:
    if not isinstance(params['multipart'], bool):
        raise TypeError('multipart is not a bool')
    payload = make_payload(params['multipart'], msgid)

    subj = 'BarCorp contract signed, let\'s go!'

    if params['legacy']:
        legacydisplay = email.message.MIMEPart()
        legacydisplay.set_content(f'Subject: {subj}\n')
        legacydisplay.set_type('text/plain')
        legacydisplay.set_param('protected-headers', 'v1')
        legacydisplay.set_charset(None) # header data is always 7-bit clean
        del legacydisplay['MIME-Version'] # MIME-Version on subparts is meaningless
        del legacydisplay['Content-Transfer-Encoding'] # header data is always 7-bit clean
        legacydisplay['Content-Disposition'] = 'inline'
    
        innerpayload = payload
        payload = email.message.MIMEPart()
        payload.set_type('multipart/mixed')
        del payload['MIME-Version'] # MIME-Version on subparts is meaningless
        # arbitrary fixed boundary for replicability:
        payload.set_boundary(hashlib.sha256('legacy-wrapper'.encode()).hexdigest()[:3])
        payload.attach(legacydisplay)
        payload.attach(innerpayload)
    
    # place intended headers on the payload:
    if params.get('smime', False):
        payload['From'] = f'{alice_x509.name} <{alice_x509.mailaddr}>'
        payload['To'] = f'{bob_x509.name} <{bob_x509.mailaddr}>'
    else:
        payload['From'] = f'{alice_key.userids[0].name} <{alice_key.userids[0].email}>'
        payload['To'] = f'{bob_key.userids[0].name} <{bob_key.userids[0].email}>'
    payload['Date'] = whenstring
    payload['Subject'] = subj
    payload.set_param('protected-headers', 'v1')

    if not isinstance(params['sk'], bytes):
        raise TypeError('sk is not bytes')
    sessionkey = codecs.decode(params['sk'], 'hex')
    payload['Message-ID'] = f"<{msgid}@protected-headers.example>"

    msg = email.message.MIMEPart(policy=email.policy.EmailPolicy(max_line_length=64))
    # Example transit header that is not part of the protected headers:
    msg['Received'] = f'from localhost (localhost [127.0.0.1]); {rcvdstring}'
    msg['MIME-Version'] = '1.0'

    if params.get('smime', False):
        if params.get('signed', True):
            # FIXME: consider multipart/signed inner layer?
            # create PKCS7 SignedData part
            p7sigpart = email.message.MIMEPart(policy=email.policy.EmailPolicy(max_line_length=64))
            inner = alice_x509.onepart_sig(payload, when, p7sigpart)
        else:
            inner = payload
        msg = alice_x509.encrypt(inner, when, [bob_x509], msg)
    else:
        if params.get('signed', True) and params['multilayer']:
            # Note that https://github.com/SecurityInnovation/PGPy/issues/291 needs to be fixed
            # in order for the signature creation time to match the message Date header:
            pgpsig = alice_key.sign(pgpy.PGPMessage.new(str(payload), cleartext=True), created=when)
            sigpart = get_pgp_sigpart(pgpsig)
            micalg = f'pgp-{pgpsig.hash_algorithm.name.lower()}'

            sigwrapper = email.message.MIMEPart()
            sigwrapper.set_type('multipart/signed')
            del sigwrapper['MIME-Version'] # MIME-Version on subparts is meaningless
            del sigwrapper['Content-Transfer-Encoding'] # test data is all 7-bit clean
            sigwrapper.set_boundary(hashlib.sha256(f"inner{payload['Message-ID']}".encode()).hexdigest()[:3])
            sigwrapper.set_param('protocol', sigpart.get_content_type())
            sigwrapper.set_param('micalg', micalg)
            sigwrapper.attach(payload)
            sigwrapper.attach(sigpart)

            payloadmsg = pgpy.PGPMessage.new(str(sigwrapper), format='m')
        else:
            payloadmsg = pgpy.PGPMessage.new(str(payload), format='m')
            if params.get('signed', True):
                # Note that https://github.com/SecurityInnovation/PGPy/issues/291 needs to be fixed
                # in order for the signature creation time to match the message Date header:
                payloadmsg |= alice_key.sign(payloadmsg, created=when)

        cipher = pgpy.constants.SymmetricKeyAlgorithm.AES256

        encmsg = alice_key.pubkey.encrypt(payloadmsg, cipher=cipher, sessionkey=sessionkey)
        encmsg = bob_key.encrypt(encmsg, cipher=cipher, sessionkey=sessionkey)

        encpart = email.message.MIMEPart()
        encpart.set_content(str(encmsg))
        encpart.set_type('application/octet-stream')
        encpart.set_charset(None) # encrypted body is ASCII-armored already
        del encpart['MIME-Version'] # MIME-Version on subparts is meaningless
        del encpart['Content-Transfer-Encoding'] # ASCII-armored data is 7-bit clean

        cruft = email.message.MIMEPart()
        cruft.set_content('Version: 1')
        cruft.set_type('application/pgp-encrypted')
        cruft.set_charset(None) # the cruft data is 7-bit clean by definition
        del cruft['MIME-Version'] # MIME-Version on subparts is meaningless
        del cruft['Content-Transfer-Encoding'] # the cruft data is 7-bit clean by definition

        msg.set_type('multipart/encrypted')
        # making up an arbitrary MIME boundary based on the Message-ID:
        msg.set_boundary(hashlib.sha256(payload['Message-ID'].encode()).hexdigest()[:3])
        msg.set_param('protocol', 'application/pgp-encrypted')
        msg.attach(cruft)
        msg.attach(encpart)

    # ensure that all non-Content headers from the payload are also on
    # the message:
    for h in payload.keys():
        if not h.lower().startswith('content-'): # don't transfer Content-* headers
            if h.lower() != 'subject': # we will obscure Subject line
                if msg.get(h, None) != payload[h]: # don't duplicate headers that already exist
                    msg[h] = payload[h]
    msg['Subject'] = '...'
    return msg

if __name__ == '__main__':
    if len(sys.argv) < 2:
        usage(to=sys.stderr)
        exit(1)
    if sys.argv[1] == 'help':
        usage()
    elif sys.argv[1] in cfg:
        params = cfg[sys.argv[1]]
        if not params['encrypt']:
            msg = signed(params, sys.argv[1])
        else:
            msg = signed_encrypted(sys.argv[1], params)
        print(msg.as_string(maxheaderlen=MAXHEADERLEN))
    elif sys.argv[1] == 'list-vectors':
        for vector in cfg:
            print(vector)
    else:
        print(f'Unknown argument "{sys.argv[1]}"', file=sys.stderr)
        usage(to=sys.stderr)
        exit(1)