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 class PKCS7_Error(Exception): pass m2.pkcs7_init(PKCS7_Error) 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 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) def write(self, bio): # type: (BIO.BIO) -> int return m2.pkcs7_write_bio(self.pkcs7, bio._ptr()) def write_der(self, bio): # type: (BIO.BIO) -> int return m2.pkcs7_write_bio_der(self.pkcs7, bio._ptr()) 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) 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) 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) def load_pkcs7_bio(p7_bio): # type: (BIO.BIO) -> PKCS7 p7_ptr = m2.pkcs7_read_bio(p7_bio._ptr()) return PKCS7(p7_ptr, 1) 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) 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) 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) 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 class SMIME_Error(Exception): pass m2.smime_init(SMIME_Error) # FIXME class has no __init__ method class SMIME(object): 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) 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) def set_x509_stack(self, stack): # type: (X509.X509_Stack) -> None assert isinstance(stack, X509.X509_Stack) self.x509_stack = stack def set_x509_store(self, store): # type: (X509.X509_Store) -> None assert isinstance(store, X509.X509_Store) self.x509_store = store def set_cipher(self, cipher): # type: (Cipher) -> None assert isinstance(cipher, Cipher) self.cipher = cipher def unset_key(self): # type: () -> None del self.pkey del self.x509 def unset_x509_stack(self): # type: () -> None del self.x509_stack def unset_x509_store(self): # type: () -> None del self.x509_store def unset_cipher(self): # type: () -> None del self.cipher 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) 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 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) 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 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) 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()) 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())