from __future__ import absolute_import
"""M2Crypto wrapper for OpenSSL S/MIME API.
Copyright (c) 1999-2003 Ng Pheng Siong. All rights reserved."""
from M2Crypto import BIO, EVP, Err, X509, m2, util
from typing import AnyStr, Callable, Optional  # noqa
PKCS7_TEXT = m2.PKCS7_TEXT  # type: int
PKCS7_NOCERTS = m2.PKCS7_NOCERTS  # type: int
PKCS7_NOSIGS = m2.PKCS7_NOSIGS  # type: int
PKCS7_NOCHAIN = m2.PKCS7_NOCHAIN  # type: int
PKCS7_NOINTERN = m2.PKCS7_NOINTERN  # type: int
PKCS7_NOVERIFY = m2.PKCS7_NOVERIFY  # type: int
PKCS7_DETACHED = m2.PKCS7_DETACHED  # type: int
PKCS7_BINARY = m2.PKCS7_BINARY  # type: int
PKCS7_NOATTR = m2.PKCS7_NOATTR  # type: int
PKCS7_SIGNED = m2.PKCS7_SIGNED  # type: int
PKCS7_ENVELOPED = m2.PKCS7_ENVELOPED  # type: int
PKCS7_SIGNED_ENVELOPED = m2.PKCS7_SIGNED_ENVELOPED  # Deprecated
PKCS7_DATA = m2.PKCS7_DATA  # type: int
[docs]class PKCS7_Error(Exception):
    pass 
m2.pkcs7_init(PKCS7_Error)
[docs]class PKCS7(object):
    m2_pkcs7_free = m2.pkcs7_free
    def __init__(self, pkcs7=None, _pyfree=0):
        # type: (Optional[bytes], int) -> None
        """PKCS7 object.
        :param pkcs7: binary representation of
               the OpenSSL type PKCS7
        """
        if pkcs7 is not None:
            self.pkcs7 = pkcs7
            self._pyfree = _pyfree
        else:
            self.pkcs7 = m2.pkcs7_new()
            self._pyfree = 1
    def __del__(self):
        # type: () -> None
        if getattr(self, '_pyfree', 0):
            self.m2_pkcs7_free(self.pkcs7)
    def _ptr(self):
        return self.pkcs7
[docs]    def type(self, text_name=0):
        # type: (int) -> int
        if text_name:
            return m2.pkcs7_type_sn(self.pkcs7)
        else:
            return m2.pkcs7_type_nid(self.pkcs7) 
[docs]    def write(self, bio):
        # type: (BIO.BIO) -> int
        return m2.pkcs7_write_bio(self.pkcs7, bio._ptr()) 
[docs]    def write_der(self, bio):
        # type: (BIO.BIO) -> int
        return m2.pkcs7_write_bio_der(self.pkcs7, bio._ptr()) 
[docs]    def get0_signers(self, certs, flags=0):
        # type: (X509.X509_Stack, int) -> X509.X509_Stack
        return X509.X509_Stack(m2.pkcs7_get0_signers(self.pkcs7,
                                                     certs.stack, flags), 1)  
[docs]def load_pkcs7(p7file):
    # type: (AnyStr) -> PKCS7
    with BIO.openfile(p7file, 'r') as bio:
        p7_ptr = m2.pkcs7_read_bio(bio.bio)
    return PKCS7(p7_ptr, 1) 
[docs]def load_pkcs7_der(p7file):
    # type: (AnyStr) -> PKCS7
    with BIO.openfile(p7file, 'rb') as bio:
        p7_ptr = m2.pkcs7_read_bio_der(bio.bio)
    return PKCS7(p7_ptr, 1) 
[docs]def load_pkcs7_bio(p7_bio):
    # type: (BIO.BIO) -> PKCS7
    p7_ptr = m2.pkcs7_read_bio(p7_bio._ptr())
    return PKCS7(p7_ptr, 1) 
[docs]def load_pkcs7_bio_der(p7_bio):
    # type: (BIO.BIO) -> PKCS7
    p7_ptr = m2.pkcs7_read_bio_der(p7_bio._ptr())
    return PKCS7(p7_ptr, 1) 
[docs]def smime_load_pkcs7(p7file):
    # type: (AnyStr) -> PKCS7
    bio = m2.bio_new_file(p7file, 'r')
    try:
        p7_ptr, bio_ptr = m2.smime_read_pkcs7(bio)
    finally:
        m2.bio_free(bio)
    if bio_ptr is None:
        return PKCS7(p7_ptr, 1), None
    else:
        return PKCS7(p7_ptr, 1), BIO.BIO(bio_ptr, 1) 
[docs]def smime_load_pkcs7_bio(p7_bio):
    # type: (BIO.BIO) -> PKCS7
    p7_ptr, bio_ptr = m2.smime_read_pkcs7(p7_bio._ptr())
    if p7_ptr is None:
        raise SMIME_Error(Err.get_error())
    if bio_ptr is None:
        return PKCS7(p7_ptr, 1), None
    else:
        return PKCS7(p7_ptr, 1), BIO.BIO(bio_ptr, 1) 
[docs]class Cipher(object):
    """Object interface to EVP_CIPHER without all the frills of
    M2Crypto.EVP.Cipher.
    """
    def __init__(self, algo):
        # type: (str) -> None
        cipher = getattr(m2, algo, None)
        if cipher is None:
            raise ValueError('unknown cipher', algo)
        self.cipher = cipher()
    def _ptr(self):
        return self.cipher 
[docs]class SMIME_Error(Exception):
    pass 
m2.smime_init(SMIME_Error)
# FIXME class has no __init__ method
[docs]class SMIME(object):
[docs]    def load_key(self, keyfile, certfile=None,
                 callback=util.passphrase_callback):
        # type: (AnyStr, Optional[AnyStr], Callable) -> None
        if certfile is None:
            certfile = keyfile
        self.pkey = EVP.load_key(keyfile, callback)
        self.x509 = X509.load_cert(certfile) 
[docs]    def load_key_bio(self, keybio, certbio=None,
                     callback=util.passphrase_callback):
        # type: (BIO.BIO, Optional[BIO.BIO], Callable) -> None
        if certbio is None:
            certbio = keybio
        self.pkey = EVP.load_key_bio(keybio, callback)
        self.x509 = X509.load_cert_bio(certbio) 
[docs]    def set_x509_stack(self, stack):
        # type: (X509.X509_Stack) -> None
        assert isinstance(stack, X509.X509_Stack)
        self.x509_stack = stack 
[docs]    def set_x509_store(self, store):
        # type: (X509.X509_Store) -> None
        assert isinstance(store, X509.X509_Store)
        self.x509_store = store 
[docs]    def set_cipher(self, cipher):
        # type: (Cipher) -> None
        assert isinstance(cipher, Cipher)
        self.cipher = cipher 
[docs]    def unset_key(self):
        # type: () -> None
        del self.pkey
        del self.x509 
[docs]    def unset_x509_stack(self):
        # type: () -> None
        del self.x509_stack 
[docs]    def unset_x509_store(self):
        # type: () -> None
        del self.x509_store 
[docs]    def unset_cipher(self):
        # type: () -> None
        del self.cipher 
[docs]    def encrypt(self, data_bio, flags=0):
        # type: (BIO.BIO, int) -> PKCS7
        if not hasattr(self, 'cipher'):
            raise SMIME_Error('no cipher: use set_cipher()')
        if not hasattr(self, 'x509_stack'):
            raise SMIME_Error('no recipient certs: use set_x509_stack()')
        pkcs7 = m2.pkcs7_encrypt(self.x509_stack._ptr(), data_bio._ptr(),
                                 self.cipher._ptr(), flags)
        return PKCS7(pkcs7, 1) 
[docs]    def decrypt(self, pkcs7, flags=0):
        # type: (PKCS7, int) -> Optional[bytes]
        if not hasattr(self, 'pkey'):
            raise SMIME_Error('no private key: use load_key()')
        if not hasattr(self, 'x509'):
            raise SMIME_Error('no certificate: load_key() used incorrectly?')
        blob = m2.pkcs7_decrypt(pkcs7._ptr(), self.pkey._ptr(),
                                self.x509._ptr(), flags)
        return blob 
[docs]    def sign(self, data_bio, flags=0, algo='sha1'):
        # type: (BIO.BIO, int, Optional[str]) -> PKCS7
        if not hasattr(self, 'pkey'):
            raise SMIME_Error('no private key: use load_key()')
        hash = getattr(m2, algo, None)
        if hash is None:
            raise SMIME_Error('no such hash algorithm %s' % algo)
        if hasattr(self, 'x509_stack'):
            pkcs7 = m2.pkcs7_sign1(self.x509._ptr(), self.pkey._ptr(),
                                   self.x509_stack._ptr(),
                                   data_bio._ptr(), hash(), flags)
            return PKCS7(pkcs7, 1)
        else:
            pkcs7 = m2.pkcs7_sign0(self.x509._ptr(), self.pkey._ptr(),
                                   data_bio._ptr(), hash(), flags)
            return PKCS7(pkcs7, 1) 
[docs]    def verify(self, pkcs7, data_bio=None, flags=0):
        # type: (PKCS7, BIO.BIO, int) -> Optional[bytes]
        if not hasattr(self, 'x509_stack'):
            raise SMIME_Error('no signer certs: use set_x509_stack()')
        if not hasattr(self, 'x509_store'):
            raise SMIME_Error('no x509 cert store: use set_x509_store()')
        assert isinstance(pkcs7, PKCS7), 'pkcs7 not an instance of PKCS7'
        p7 = pkcs7._ptr()
        if data_bio is None:
            blob = m2.pkcs7_verify0(p7, self.x509_stack._ptr(),
                                    self.x509_store._ptr(), flags)
        else:
            blob = m2.pkcs7_verify1(p7, self.x509_stack._ptr(),
                                    self.x509_store._ptr(),
                                    data_bio._ptr(), flags)
        return blob 
[docs]    def write(self, out_bio, pkcs7, data_bio=None, flags=0):
        # type: (BIO.BIO, PKCS7, Optional[BIO.BIO], int) -> int
        assert isinstance(pkcs7, PKCS7)
        if data_bio is None:
            return m2.smime_write_pkcs7(out_bio._ptr(), pkcs7._ptr(), flags)
        else:
            return m2.smime_write_pkcs7_multi(out_bio._ptr(), pkcs7._ptr(),
                                              data_bio._ptr(), flags)  
[docs]def text_crlf(text):
    # type: (bytes) -> bytes
    bio_in = BIO.MemoryBuffer(text)
    bio_out = BIO.MemoryBuffer()
    if m2.smime_crlf_copy(bio_in._ptr(), bio_out._ptr()):
        return bio_out.read()
    else:
        raise SMIME_Error(Err.get_error()) 
[docs]def text_crlf_bio(bio_in):
    # type: (BIO.BIO) -> BIO.BIO
    bio_out = BIO.MemoryBuffer()
    if m2.smime_crlf_copy(bio_in._ptr(), bio_out._ptr()):
        return bio_out
    else:
        raise SMIME_Error(Err.get_error())