m2_sk_x509_extension_free() (M2Crypto.X509.X509_Extension_Stack method)

m2_x509_name_entry_free() (M2Crypto.X509.X509_Name_Entry method)

m2_x509_store_ctx_free() (M2Crypto.X509.X509_Store_Context method)

get_default_session_timeout() (M2Crypto.SSL.Connection.Connection method)

get_peer_cert_chain() (M2Crypto.SSL.Connection.Connection method)

get_socket_read_timeout() (M2Crypto.SSL.Connection.Connection method)

get_socket_write_timeout() (M2Crypto.SSL.Connection.Connection method)

serverPostConnectionCheck() (M2Crypto.SSL.Connection.Connection method)

set_client_CA_list_from_context() (M2Crypto.SSL.Connection.Connection method)

set_client_CA_list_from_file() (M2Crypto.SSL.Connection.Connection method)

set_post_connection_check_callback() (M2Crypto.SSL.Connection.Connection method)

set_session_id_ctx() (M2Crypto.SSL.Connection.Connection method)

set_socket_read_timeout() (M2Crypto.SSL.Connection.Connection method)

set_socket_write_timeout() (M2Crypto.SSL.Connection.Connection method)

set_ssl_close_flag() (M2Crypto.SSL.Connection.Connection method)

set_tlsext_host_name() (M2Crypto.SSL.Connection.Connection method)

set_client_CA_list_from_file() (M2Crypto.SSL.Context.Context method)

set_default_verify_paths() (M2Crypto.SSL.Context.Context method)

TLSProtocolWrapper (class in M2Crypto.SSL.TwistedProtocolWrapper)

clear() (M2Crypto.SSL.TwistedProtocolWrapper.TLSProtocolWrapper method)

connectionLost() (M2Crypto.SSL.TwistedProtocolWrapper.TLSProtocolWrapper method)

connectionMade() (M2Crypto.SSL.TwistedProtocolWrapper.TLSProtocolWrapper method)

dataReceived() (M2Crypto.SSL.TwistedProtocolWrapper.TLSProtocolWrapper method)

loseConnection() (M2Crypto.SSL.TwistedProtocolWrapper.TLSProtocolWrapper method)

startTLS() (M2Crypto.SSL.TwistedProtocolWrapper.TLSProtocolWrapper method)

write() (M2Crypto.SSL.TwistedProtocolWrapper.TLSProtocolWrapper method)

writeSequence() (M2Crypto.SSL.TwistedProtocolWrapper.TLSProtocolWrapper method)

ssl_verify_callback_allow_unknown_ca() (in module M2Crypto.SSL.cb)

create_socket() (M2Crypto.SSL.ssl_dispatcher.ssl_dispatcher method)

from __future__ import absolute_import

"""
M2Crypto wrapper for OpenSSL ASN1 API.

Copyright (c) 1999-2004 Ng Pheng Siong. All rights reserved.

Portions created by Open Source Applications Foundation (OSAF) are
Copyright (C) 2005 OSAF. All Rights Reserved.
"""

import datetime
import time

from M2Crypto import BIO, m2, six
from typing import Optional  # noqa

MBSTRING_FLAG = 0x1000
MBSTRING_ASC = MBSTRING_FLAG | 1
MBSTRING_BMP = MBSTRING_FLAG | 2

class ASN1_Integer(object):

    m2_asn1_integer_free = m2.asn1_integer_free

    def __init__(self, asn1int, _pyfree=0):
        # type: (ASN1_Integer, int) -> None
        self.asn1int = asn1int
        self._pyfree = _pyfree

    def __cmp__(self, other):
        # type: (ASN1_Integer) -> int
        if not isinstance(other, ASN1_Integer):
            raise TypeError(
                "Comparisons supported only between ANS1_Integer objects")

        return m2.asn1_integer_cmp(self.asn1int, other.asn1int)

    def __del__(self):
        # type: () -> None
        if self._pyfree:
            self.m2_asn1_integer_free(self.asn1int)

    def __int__(self):
        # type: () -> int
        return m2.asn1_integer_get(self.asn1int)

class ASN1_String(object):

    m2_asn1_string_free = m2.asn1_string_free

    def __init__(self, asn1str, _pyfree=0):
        # type: (ASN1_String, int) -> None
        self.asn1str = asn1str

    def __bytes__(self):
        # type: () -> bytes
        buf = BIO.MemoryBuffer()
        m2.asn1_string_print(buf.bio_ptr(), self.asn1str)
        return buf.read_all()

    def __str__(self):
        # type: () -> str
        return six.ensure_text(self.__bytes__())

        if getattr(self, '_pyfree', 0):
            self.m2_asn1_string_free(self.asn1str)

    def _ptr(self):
        return self.asn1str

    def as_text(self, flags=0):
        # type: (int) -> str
        """Output an ASN1_STRING structure according to the set flags.

        :param flags: determine the format of the output by using
               predetermined constants, see ASN1_STRING_print_ex(3)
               manpage for their meaning.
        :return: output an ASN1_STRING structure.
        """
        m2.asn1_string_print_ex(buf.bio_ptr(), self.asn1str, flags)
        return six.ensure_text(buf.read_all())

class ASN1_Object(object):

    m2_asn1_object_free = m2.asn1_object_free

    def __init__(self, asn1obj, _pyfree=0):
        # type: (ASN1_Object, int) -> None
        self.asn1obj = asn1obj

            self.m2_asn1_object_free(self.asn1obj)

        return self.asn1obj

class _UTC(datetime.tzinfo):
    def tzname(self, dt):
        # type: (Optional[datetime.datetime]) -> str
        return "UTC"

    def dst(self, dt):
        # type: (Optional[datetime.datetime]) -> datetime.timedelta
        return datetime.timedelta(0)

    def utcoffset(self, dt):

    def __repr__(self):
        return "<Timezone: %s>" % self.tzname(None)

UTC = _UTC()  # type: _UTC

class LocalTimezone(datetime.tzinfo):
    """Localtimezone from datetime manual."""

    def __init__(self):
        self._stdoffset = datetime.timedelta(seconds=-time.timezone)
        if time.daylight:
            self._dstoffset = datetime.timedelta(seconds=-time.altzone)
        else:
            self._dstoffset = self._stdoffset
        self._dstdiff = self._dstoffset - self._stdoffset

        # type: (datetime.datetime) -> datetime.timedelta
        if self._isdst(dt):
            return self._dstoffset
        else:
            return self._stdoffset

            return self._dstdiff
        else:
            return datetime.timedelta(0)

        # type: (datetime.datetime) -> str
        return time.tzname[self._isdst(dt).real]

    def _isdst(self, dt):
        # type: (datetime.datetime) -> bool
        tt = (dt.year, dt.month, dt.day,
              dt.hour, dt.minute, dt.second,
              dt.weekday(), 0, -1)
        stamp = time.mktime(tt)
        tt = time.localtime(stamp)
        return tt.tm_isdst > 0

class ASN1_TIME(object):
    _ssl_months = ["Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug",
                   "Sep", "Oct", "Nov", "Dec"]
    m2_asn1_time_free = m2.asn1_time_free

    def __init__(self, asn1_time=None, _pyfree=0, asn1_utctime=None):
        # type: (Optional[ASN1_TIME], Optional[int], Optional[ASN1_TIME]) -> None
        # handle old keyword parameter
        if asn1_time is None:
            asn1_time = asn1_utctime
        if asn1_time is not None:
            assert m2.asn1_time_type_check(asn1_time), \
                "'asn1_time' type error'"
            self.asn1_time = asn1_time
            self._pyfree = _pyfree
        else:
            self.asn1_time = m2.asn1_time_new()
            self._pyfree = 1

            self.m2_asn1_time_free(self.asn1_time)

        assert m2.asn1_time_type_check(self.asn1_time), \
            "'asn1_time' type error'"
        m2.asn1_time_print(buf.bio_ptr(), self.asn1_time)

        return self.asn1_time

    def set_string(self, string):
        # type: (bytes) -> int
        """Set time from UTC string."""
        return m2.asn1_time_set_string(self.asn1_time, string)

    def set_time(self, time):
        # type: (int) -> ASN1_TIME
        """Set time from seconds since epoch (int)."""
        return m2.asn1_time_set(self.asn1_time, time)

    def get_datetime(self):
        # type: () -> ASN1_TIME
        date = str(self)

        timezone = None
        if ' ' not in date:
            raise ValueError("Invalid date: %s" % date)
        month, rest = date.split(' ', 1)
        if month not in self._ssl_months:
            raise ValueError("Invalid date %s: Invalid month: %s" %
                             (date, month))
        if rest.endswith(' GMT'):
            timezone = UTC
            rest = rest[:-4]
        if '.' in rest:
            dt = datetime.datetime.strptime(rest, "%d %H:%M:%S.%f %Y")
        else:
            dt = datetime.datetime.strptime(rest, "%d %H:%M:%S %Y")
        dt = dt.replace(month=self._ssl_months.index(month) + 1)
        if timezone:
            dt = dt.replace(tzinfo=UTC)
        return dt

    def set_datetime(self, date):
        # type: (datetime.datetime) -> ASN1_TIME
        local = LocalTimezone()
        if date.tzinfo is None:
            date = date.replace(tzinfo=local)
        date = date.astimezone(local)
        return self.set_time(int(time.mktime(date.timetuple())))

ASN1_UTCTIME = ASN1_TIME

"""Secure Authenticator Cookies

Copyright (c) 1999-2002 Ng Pheng Siong. All rights reserved."""

import logging
import re

from M2Crypto import Rand, m2, six, util
from M2Crypto.six.moves.http_cookies import SimpleCookie

from typing import re as type_re, AnyStr, Optional, Union  # noqa

_MIX_FORMAT = 'exp=%f&data=%s&digest='
_MIX_RE = re.compile(r'exp=(\d+\.\d+)&data=(.+)&digest=(\S*)')

log = logging.getLogger(__name__)

def mix(expiry, data, format=_MIX_FORMAT):
    # type: (float, AnyStr, str) -> AnyStr
    return format % (expiry, data)

def unmix(dough, regex=_MIX_RE):
    # type: (AnyStr, type_re) -> object
    mo = regex.match(dough)
    if mo:
        return float(mo.group(1)), mo.group(2)
    else:
        return None

def unmix3(dough, regex=_MIX_RE):
    # type: (AnyStr, type_re) -> Optional[tuple[float, AnyStr, AnyStr]]
    if mo:
        return float(mo.group(1)), mo.group(2), mo.group(3)
    else:

_TOKEN = '_M2AUTH_'  # type: str

class AuthCookieJar(object):

    _keylen = 20  # type: int

        self._key = Rand.rand_bytes(self._keylen)

    def _hmac(self, key, data):
        # type: (bytes, str) -> str
        return util.bin_to_hex(m2.hmac(key, six.ensure_binary(data), m2.sha1()))

    def makeCookie(self, expiry, data):
        # type: (float, str) -> AuthCookie
        """
        Make a cookie

        :param expiry: expiration time (float in seconds)
        :param data: cookie content
        :return: AuthCookie object
        """
        if not isinstance(expiry, (six.integer_types, float)):
            raise ValueError('Expiration time must be number, not "%s' % expiry)
        dough = mix(expiry, data)
        return AuthCookie(expiry, data, dough, self._hmac(self._key, dough))

    def isGoodCookie(self, cookie):
        # type: (AuthCookie) -> Union[bool, int]
        assert isinstance(cookie, AuthCookie)
        if cookie.isExpired():
            return 0
        c = self.makeCookie(cookie._expiry, cookie._data)
        return (c._expiry == cookie._expiry) \
            and (c._data == cookie._data) \
            and (c._mac == cookie._mac) \
            and (c.output() == cookie.output())

    def isGoodCookieString(self, cookie_str, _debug=False):
        # type: (Union[dict, bytes], bool) -> Union[bool, int]
        c = SimpleCookie()
        c.load(cookie_str)
        if _TOKEN not in c:
            log.debug('_TOKEN not in c (keys = %s)', dir(c))
            return 0
        undough = unmix3(c[_TOKEN].value)
        if undough is None:
            log.debug('undough is None')
            return 0
        exp, data, mac = undough
        c2 = self.makeCookie(exp, data)
        if _debug and (c2._mac == mac):
            log.error('cookie_str = %s', cookie_str)
            log.error('c2.isExpired = %s', c2.isExpired())
            log.error('mac = %s', mac)
            log.error('c2._mac = %s', c2._mac)
            log.error('c2._mac == mac: %s', str(c2._mac == mac))
        return (not c2.isExpired()) and (c2._mac == mac)

class AuthCookie(object):

    def __init__(self, expiry, data, dough, mac):
        # type: (float, str, str, str) -> None
        """
        Create new authentication cookie

        :param expiry: expiration time (in seconds)
        :param data: cookie payload (as a string)
        :param dough: expiry & data concatenated to URL compliant
                      string
        :param mac: SHA1-based HMAC of dough and random key
        """
        self._expiry = expiry
        self._data = data
        self._mac = mac
        self._cookie = SimpleCookie()
        self._cookie[_TOKEN] = '%s%s' % (dough, mac)
        self._name = '%s%s' % (dough, mac)  # WebKit only.

    def expiry(self):
        # type: () -> float
        """Return the cookie's expiry time."""
        return self._expiry

    def data(self):
        """Return the data portion of the cookie."""
        return self._data

    def mac(self):
        """Return the cookie's MAC."""
        return self._mac

    def output(self, header="Set-Cookie:"):
        # type: (Optional[str]) -> str
        """Return the cookie's output in "Set-Cookie" format."""
        return self._cookie.output(header=header)

    def value(self):
        """Return the cookie's output minus the "Set-Cookie: " portion.
        """
        return self._cookie[_TOKEN].value

    def isExpired(self):
        # type: () -> bool
        """Return 1 if the cookie has expired, 0 otherwise."""
        return isinstance(self._expiry, (float, six.integer_types)) and \
            (time.time() > self._expiry)

    # Following two methods are for WebKit only.
    # I may wish to push them to WKAuthCookie, but they are part
    # of the API now. Oh well.
    def name(self):
        return self._name

    def headerValue(self):
        return self.value()

"""M2Crypto wrapper for OpenSSL BIO API.

Copyright (c) 1999-2004 Ng Pheng Siong. All rights reserved."""

import io
from typing import Any, AnyStr, Callable, Iterable, Optional, Union  # noqa

from M2Crypto import m2, six

log = logging.getLogger('BIO')

class BIOError(ValueError):
    pass

m2.bio_init(BIOError)

class BIO(object):
    """Abstract object interface to the BIO API."""

    m2_bio_free = m2.bio_free

    def __init__(self, bio=None, _pyfree=0, _close_cb=None):
        # type: (Optional[BIO], int, Optional[Callable]) -> None
        self.bio = bio
        self._close_cb = _close_cb
        self.closed = 0
        self.write_closed = 0

            self.m2_bio_free(self.bio)

        return self.bio

    # Deprecated.
    bio_ptr = _ptr

    def fileno(self):
        return m2.bio_get_fd(self.bio)

    def readable(self):
        return not self.closed

    def read(self, size=None):
        # type: (int) -> Union[bytes, bytearray]
        if not self.readable():
            raise IOError('cannot read')
        if size is None:
            buf = bytearray()
            while 1:
                data = m2.bio_read(self.bio, 4096)
                if not data:
                    break
                buf += data
            return buf
        elif size == 0:
            return b''
        elif size < 0:
            raise ValueError('read count is negative')
        else:
            return bytes(m2.bio_read(self.bio, size))

    def readline(self, size=4096):
        # type: (int) -> bytes
        buf = m2.bio_gets(self.bio, size)
        buf = '' if buf is None else buf
        return six.ensure_binary(buf)

    def readlines(self, sizehint='ignored'):
        # type: (Union[AnyStr, int]) -> Iterable[bytes]
        lines = []
        while 1:
            buf = m2.bio_gets(self.bio, 4096)
            if buf is None:
                break
            lines.append(six.ensure_binary(buf))
        return lines

    def writeable(self):
        return (not self.closed) and (not self.write_closed)

    def write(self, data):
        # type: (AnyStr) -> int
        """Write data to BIO.

        :return: either data written, or [0, -1] for nothing written,
                 -2 not implemented
        """
        if not self.writeable():
            raise IOError('cannot write')
        if isinstance(data, six.text_type):
            data = data.encode('utf8')
        return m2.bio_write(self.bio, data)

    def write_close(self):
        self.write_closed = 1

    def flush(self):
        """Flush the buffers.

        :return: 1 for success, and 0 or -1 for failure
        """
        m2.bio_flush(self.bio)

    def reset(self):
        """Set the bio to its initial state.

        """
        return m2.bio_reset(self.bio)

    def close(self):
        self.closed = 1
        if self._close_cb:
            self._close_cb()

    def should_retry(self):
        """
        Can the call be attempted again, or was there an error
        ie do_handshake

        """
        return m2.bio_should_retry(self.bio)

    def should_read(self):
        """Should we read more data?"""

        return m2.bio_should_read(self.bio)

    def should_write(self):
        """Should we write more data?"""
        return m2.bio_should_write(self.bio)

    def tell(self):
        """Return the current offset."""
        return m2.bio_tell(self.bio)

    def seek(self, off):
        """Seek to the specified absolute offset."""
        return m2.bio_seek(self.bio, off)

    def __enter__(self):
        return self

    def __exit__(self, *args):
        # type: (*Any) -> int
        self.close()

class MemoryBuffer(BIO):
    """Object interface to BIO_s_mem.

    Empirical testing suggests that this class performs less well than
    cStringIO, because cStringIO is implemented in C, whereas this class
    is implemented in Python. Thus, the recommended practice is to use
    cStringIO for regular work and convert said cStringIO object to
    a MemoryBuffer object only when necessary.
    """

    def __init__(self, data=None):
        # type: (Optional[bytes]) -> None
        super(MemoryBuffer, self).__init__(self)
        if data is not None and not isinstance(data, bytes):
                "data must be bytes or None, not %s" % (type(data).__name__, ))
        self.bio = m2.bio_new(m2.bio_s_mem())
        self._pyfree = 1
        if data is not None:
            m2.bio_write(self.bio, data)

    def __len__(self):
        return m2.bio_ctrl_pending(self.bio)

    def read(self, size=0):
        if size:
            return m2.bio_read(self.bio, size)
        else:
            return m2.bio_read(self.bio, m2.bio_ctrl_pending(self.bio))

    # Backwards-compatibility.
    getvalue = read_all = read

        super(MemoryBuffer, self).write_close()
        m2.bio_set_mem_eof_return(self.bio, 0)

    close = write_close

class File(BIO):
    """Object interface to BIO_s_pyfd.

    This class interfaces Python to OpenSSL functions that expect BIO. For
    general file manipulation in Python, use Python's builtin file object.
    """

    def __init__(self, pyfile, close_pyfile=1, mode='rb'):
        # type: (Union[io.BytesIO, AnyStr], int, AnyStr) -> None
        super(File, self).__init__(self, _pyfree=1)

        if isinstance(pyfile, six.string_types):
            pyfile = open(pyfile, mode)

        # This is for downward compatibility, but I don't think, that it is
        # good practice to have two handles for the same file. Whats about
        # concurrent write access? Last write, last wins? Especially since Py3
        # has its own buffer management. See:
        #
        #  https://docs.python.org/3.3/c-api/file.html
        #
        pyfile.flush()
        self.fname = pyfile.name
        self.pyfile = pyfile
        # Be wary of https://github.com/openssl/openssl/pull/1925
        # BIO_new_fd is NEVER to be used before OpenSSL 1.1.1
        if hasattr(m2, "bio_new_pyfd"):
            self.bio = m2.bio_new_pyfd(pyfile.fileno(), m2.bio_noclose)
        else:
            self.bio = m2.bio_new_pyfile(pyfile, m2.bio_noclose)

        self.close_pyfile = close_pyfile
        self.closed = False

        super(File, self).flush()
        self.pyfile.flush()

        self.flush()
        super(File, self).close()
        if self.close_pyfile:
            self.pyfile.close()

        :return: 0 for success, and -1 for failure
        """
        return super(File, self).reset()

        if not self.closed:
            m2.bio_free(self.bio)

def openfile(filename, mode='rb'):
    # type: (AnyStr, AnyStr) -> File
    try:
        f = open(filename, mode)
    except IOError as ex:
        raise BIOError(ex.args)

    return File(f)

class IOBuffer(BIO):
    """Object interface to BIO_f_buffer.

    Its principal function is to be BIO_push()'ed on top of a BIO_f_ssl, so
    that makefile() of said underlying SSL socket works.
    """

    m2_bio_pop = m2.bio_pop

    def __init__(self, under_bio, mode='rwb', _pyfree=1):
        # type: (BIO, str, int) -> None
        super(IOBuffer, self).__init__(self, _pyfree=_pyfree)
        self.io = m2.bio_new(m2.bio_f_buffer())
        self.bio = m2.bio_push(self.io, under_bio._ptr())
        # This reference keeps the underlying BIO alive while we're not closed.
        self._under_bio = under_bio
        if 'w' in mode:
            self.write_closed = 0
        else:
            self.write_closed = 1

            self.m2_bio_pop(self.bio)
        self.m2_bio_free(self.io)

        BIO.close(self)

class CipherStream(BIO):
    """Object interface to BIO_f_cipher."""

    SALT_LEN = m2.PKCS5_SALT_LEN

    def __init__(self, obio):
        # type: (BIO) -> None
        super(CipherStream, self).__init__(self, _pyfree=1)
        self.obio = obio
        self.bio = m2.bio_new(m2.bio_f_cipher())

        if not getattr(self, 'closed', 1):
            self.close()

        self.m2_bio_pop(self.bio)
        self.m2_bio_free(self.bio)

        self.obio.write_close()

    def set_cipher(self, algo, key, iv, op):
        # type: (str, AnyStr, AnyStr, int) -> None
        cipher = getattr(m2, algo, None)
        if cipher is None:
            raise ValueError('unknown cipher', algo)
        else:
            if not isinstance(key, bytes):
                key = key.encode('utf8')
            if not isinstance(iv, bytes):
                iv = iv.encode('utf8')
        m2.bio_set_cipher(self.bio, cipher(), key, iv, int(op))
        m2.bio_push(self.bio, self.obio._ptr())

class SSLBio(BIO):
    """Object interface to BIO_f_ssl."""

    def __init__(self, _pyfree=1):
        # type: (int) -> None
        super(SSLBio, self).__init__(self, _pyfree=_pyfree)
        self.bio = m2.bio_new(m2.bio_f_ssl())

    def set_ssl(self, conn, close_flag=m2.bio_noclose):
        ## type: (Connection, int) -> None
        """
        Sets the bio to the SSL pointer which is
        contained in the connection object.
        """
        self._pyfree = 0
        m2.bio_set_ssl(self.bio, conn.ssl, close_flag)
        if close_flag == m2.bio_noclose:
            conn.set_ssl_close_flag(m2.bio_close)

    def do_handshake(self):
        """Do the handshake.

        Return 1 if the handshake completes
        Return 0 or a negative number if there is a problem
        """
        return m2.bio_do_handshake(self.bio)

"""
M2Crypto wrapper for OpenSSL BN (BIGNUM) API.

Copyright (c) 2005 Open Source Applications Foundation. All rights reserved.
"""

from M2Crypto import m2

def rand(bits, top=-1, bottom=0):
    # type: (int, int, int) -> Optional[int]
    """
    Generate cryptographically strong random number.

    :param bits:   Length of random number in bits.
    :param top:    If -1, the most significant bit can be 0. If 0, the most
                   significant bit is 1, and if 1, the two most significant
                   bits will be 1.
    :param bottom: If bottom is true, the number will be odd.
    """
    return m2.bn_rand(bits, top, bottom)

def rand_range(range):
    # type: (int) -> int
    """
    Generate a random number in a range.

    :param range: Upper limit for range.
    :return:      A random number in the range [0, range)
    """
    return m2.bn_rand_range(range)

def randfname(length):
    # type: (int) -> str
    """
    Return a random filename, which is simply a string where all
    the characters are from the set [a-zA-Z0-9].

    :param length: Length of filename to return.
    :return:       random filename string
    """
    import warnings
    warnings.warn(
        "Don't use BN.randfname(), use tempfile methods instead.",
        DeprecationWarning, stacklevel=2)
    letters = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ01234567890'
    lettersLen = len(letters)
    fname = []  # type: list
    for x in range(length):
        fname += [letters[m2.bn_rand_range(lettersLen)]]

    return ''.join(fname)

"""M2Crypto wrapper for OpenSSL DH API.

Copyright (c) 1999-2003 Ng Pheng Siong. All rights reserved."""

from M2Crypto import BIO, m2
from M2Crypto.util import genparam_callback
from typing import AnyStr, Callable, Optional  # noqa

class DHError(Exception):
    pass

m2.dh_init(DHError)

class DH(object):
    """Object interface to the Diffie-Hellman key exchange protocol.
    """

    m2_dh_free = m2.dh_free

    def __init__(self, dh, _pyfree=0):
        # type: (bytes, int) -> None
        assert m2.dh_type_check(dh)
        self.dh = dh

            self.m2_dh_free(self.dh)

        assert m2.dh_type_check(self.dh), "'dh' type error"
        return int(m2.dh_size(self.dh))

    def __getattr__(self, name):
        # type: (str) -> bytes
        if name in ('p', 'g', 'pub', 'priv'):
            method = getattr(m2, 'dh_get_%s' % (name,))
            assert m2.dh_type_check(self.dh), "'dh' type error"
            return method(self.dh)
        else:
            raise AttributeError

    def __setattr__(self, name, value):
        # type: (str, bytes) -> bytes
        if name in ('p', 'g'):
            raise DHError('set (p, g) via set_params()')
        elif name in ('pub', 'priv'):
            raise DHError('generate (pub, priv) via gen_key()')
        else:
            self.__dict__[name] = value

        return self.dh

    def check_params(self):
        return m2.dh_check(self.dh)

    def gen_key(self):
        m2.dh_generate_key(self.dh)

    def compute_key(self, pubkey):
        # type: (bytes) -> bytes
        return m2.dh_compute_key(self.dh, pubkey)

    def print_params(self, bio):
        # type: (BIO.BIO) -> int
        return m2.dhparams_print(bio._ptr(), self.dh)

def gen_params(plen, g, callback=genparam_callback):
    # type: (int, int, Optional[Callable]) -> DH
    dh_parms = m2.dh_generate_parameters(plen, g, callback)
    dh_obj = DH(dh_parms, 1)
    return dh_obj

def load_params(file):
    # type: (AnyStr) -> DH
    with BIO.openfile(file) as bio:
        return load_params_bio(bio)

def load_params_bio(bio):
    # type: (BIO.BIO) -> DH
    return DH(m2.dh_read_parameters(bio._ptr()), 1)

def set_params(p, g):
    # type: (bytes, bytes) -> DH
    dh = m2.dh_new()
    m2.dh_set_pg(dh, p, g)
    return DH(dh, 1)

# def free_params(cptr):
#    m2.dh_free(cptr)

DH_GENERATOR_2 = m2.DH_GENERATOR_2
DH_GENERATOR_5 = m2.DH_GENERATOR_5

from __future__ import absolute_import, print_function

"""
    M2Crypto wrapper for OpenSSL DSA API.

    Copyright (c) 1999-2003 Ng Pheng Siong. All rights reserved.

    Portions created by Open Source Applications Foundation (OSAF) are
    Copyright (C) 2004 OSAF. All Rights Reserved.
"""

from M2Crypto import BIO, m2, util
from typing import Any, AnyStr, Callable, Tuple  # noqa

class DSAError(Exception):
    pass

m2.dsa_init(DSAError)

class DSA(object):

    """
    This class is a context supporting DSA key and parameter
    values, signing and verifying.

    Simple example::

        from M2Crypto import EVP, DSA, util

        message = 'Kilroy was here!'
        md = EVP.MessageDigest('sha1')
        md.update(message)
        digest = md.final()

        dsa = DSA.gen_params(1024)
        dsa.gen_key()
        r, s = dsa.sign(digest)
        good = dsa.verify(digest, r, s)
        if good:
            print('  ** success **')
        else:
            print('  ** verification failed **')
    """

    m2_dsa_free = m2.dsa_free

    def __init__(self, dsa, _pyfree=0):
        """
        Use one of the factory functions to create an instance.
        :param dsa: binary representation of OpenSSL DSA type
        """
        assert m2.dsa_type_check(dsa), "'dsa' type error"
        self.dsa = dsa

            self.m2_dsa_free(self.dsa)

        """
        Return the key length.

        :return:  the DSA key length in bits
        """
        assert m2.dsa_type_check(self.dsa), "'dsa' type error"
        return m2.dsa_keylen(self.dsa)

        """
        Return specified DSA parameters and key values.

        :param name: name of variable to be returned.  Must be
                     one of 'p', 'q', 'g', 'pub', 'priv'.
        :return:     value of specified variable (a "byte string")
        """
        if name in ['p', 'q', 'g', 'pub', 'priv']:
            method = getattr(m2, 'dsa_get_%s' % (name,))
            assert m2.dsa_type_check(self.dsa), "'dsa' type error"
            return method(self.dsa)
        else:

        # type: (str, bytes) -> None
        if name in ['p', 'q', 'g']:
            raise DSAError('set (p, q, g) via set_params()')
        elif name in ['pub', 'priv']:
            raise DSAError('generate (pub, priv) via gen_key()')
        else:

    def set_params(self, p, q, g):
        # type: (bytes, bytes, bytes) -> None
        """
        Set new parameters.

        :param p: MPI binary representation ... format that consists of
                  the number's length in bytes represented as a 4-byte
                  big-endian number, and the number itself in big-endian
                  format, where the most significant bit signals
                  a negative number (the representation of numbers with
                  the MSB set is prefixed with null byte).
        :param q: ditto
        :param g: ditto

        @warning: This does not change the private key, so it may be
                  unsafe to use this method. It is better to use
                  gen_params function to create a new DSA object.
        """
        m2.dsa_set_pqg(self.dsa, p, q, g)

        """
        Generate a key pair.
        """
        m2.dsa_gen_key(self.dsa)

    def save_params(self, filename):
        """
        Save the DSA parameters to a file.

        :param filename: Save the DSA parameters to this file.
        :return:         1 (true) if successful
        """
        with BIO.openfile(filename, 'wb') as bio:
            ret = m2.dsa_write_params_bio(self.dsa, bio._ptr())

        return ret

    def save_params_bio(self, bio):
        """
        Save DSA parameters to a BIO object.

        :param bio: Save DSA parameters to this object.
        :return:    1 (true) if successful
        """
        return m2.dsa_write_params_bio(self.dsa, bio._ptr())

    def save_key(self, filename, cipher='aes_128_cbc',
                 callback=util.passphrase_callback):
        # type: (AnyStr, str, Callable) -> int
        """
        Save the DSA key pair to a file.

        :param filename: Save the DSA key pair to this file.
        :param cipher:   name of symmetric key algorithm and mode
                         to encrypt the private key.
        """
            ret = self.save_key_bio(bio, cipher, callback)

        return ret

    def save_key_bio(self, bio, cipher='aes_128_cbc',
                     callback=util.passphrase_callback):
        # type: (BIO.BIO, str, Callable) -> int
        """
        Save DSA key pair to a BIO object.

        :param bio:    Save DSA parameters to this object.
        :param cipher: name of symmetric key algorithm and mode
                       to encrypt the private key.
        :return:       1 (true) if successful
        """
            return m2.dsa_write_key_bio_no_cipher(self.dsa,
                                                  bio._ptr(), callback)
        else:
            ciph = getattr(m2, cipher, None)
            if ciph is None:
                raise DSAError('no such cipher: %s' % cipher)
            else:
                ciph = ciph()
            return m2.dsa_write_key_bio(self.dsa, bio._ptr(), ciph, callback)

    def save_pub_key(self, filename):
        """
        Save the DSA public key (with parameters) to a file.

        :param filename: Save DSA public key (with parameters)
                         to this file.
        """
            ret = self.save_pub_key_bio(bio)

        return ret

    def save_pub_key_bio(self, bio):
        """
        Save DSA public key (with parameters) to a BIO object.

        :param bio: Save DSA public key (with parameters)
                    to this object.
        :return:  1 (true) if successful
        """
        return m2.dsa_write_pub_key_bio(self.dsa, bio._ptr())

    def sign(self, digest):
        # type: (bytes) -> Tuple[bytes, bytes]
        """
        Sign the digest.

        :param digest: SHA-1 hash of message (same as output
                       from MessageDigest, a "byte string")
        :return:       DSA signature, a tuple of two values, r and s,
                       both "byte strings".
        """
        assert self.check_key(), 'key is not initialised'
        return m2.dsa_sign(self.dsa, digest)

    def verify(self, digest, r, s):
        # type: (bytes, bytes, bytes) -> int
        """
        Verify a newly calculated digest against the signature
        values r and s.

        :param r:      r value of the signature, a "byte string"
        :param s:      s value of the signature, a "byte string"
        :return:       1 (true) if verify succeeded, 0 if failed
        """
        return m2.dsa_verify(self.dsa, digest, r, s)

    def sign_asn1(self, digest):
        return m2.dsa_sign_asn1(self.dsa, digest)

    def verify_asn1(self, digest, blob):
        return m2.dsa_verify_asn1(self.dsa, digest, blob)

    def check_key(self):
        """
        Check to be sure the DSA object has a valid private key.

        :return:  1 (true) if a valid private key
        """
        return m2.dsa_check_key(self.dsa)

class DSA_pub(DSA):

    """
    This class is a DSA context that only supports a public key
    and verification.  It does NOT support a private key or
    signing.

    """

    def sign(self, *argv):
        # type: (*Any) -> None
        raise DSAError('DSA_pub object has no private key')

    sign_asn1 = sign

        """
        :return: does DSA_pub contain a pub key?
        """
        return m2.dsa_check_pub_key(self.dsa)

    save_key = DSA.save_pub_key

    save_key_bio = DSA.save_pub_key_bio

# --------------------------------------------------------------
# factories and other functions

def gen_params(bits, callback=util.genparam_callback):
    # type: (int, Callable) -> DSA
    """
    Factory function that generates DSA parameters and
    instantiates a DSA object from the output.

    :param bits: The length of the prime to be generated. If
                 'bits' < 512, it is set to 512.
    :param callback: A Python callback object that will be
                 invoked during parameter generation; it usual
                 purpose is to provide visual feedback.
    :return:  instance of DSA.
    """
    dsa = m2.dsa_generate_parameters(bits, callback)
    return DSA(dsa, 1)

def set_params(p, q, g):
    # type: (bytes, bytes, bytes) -> DSA
    """
    Factory function that instantiates a DSA object with DSA
    parameters.

    :param p: value of p, a "byte string"
    :param q: value of q, a "byte string"
    :param g: value of g, a "byte string"
    """
    dsa = m2.dsa_new()
    m2.dsa_set_pqg(dsa, p, q, g)

def load_params(file, callback=util.passphrase_callback):
    # type: (AnyStr, Callable) -> DSA
    """
    parameters from a file.

    :param file:     Names the file (a path) that contains the PEM
                     representation of the DSA parameters.
                     invoked if the DSA parameters file is
                     passphrase-protected.
    :return:         instance of DSA.
    """
        ret = load_params_bio(bio, callback)

    return ret

def load_params_bio(bio, callback=util.passphrase_callback):
    # type: (BIO.BIO, Callable) -> DSA
    """
    parameters from a M2Crypto.BIO object.

    :param bio:      Contains the PEM representation of the DSA
                     parameters.
    """
    dsa = m2.dsa_read_params(bio._ptr(), callback)

def load_key(file, callback=util.passphrase_callback):
    """
    Factory function that instantiates a DSA object from a
    PEM encoded DSA key pair.

                     representation of the DSA key pair.
                     invoked if the DSA key pair is
    """
        ret = load_key_bio(bio, callback)

    return ret

def load_key_bio(bio, callback=util.passphrase_callback):
    """

                     key pair.
    """
    dsa = m2.dsa_read_key(bio._ptr(), callback)

def pub_key_from_params(p, q, g, pub):
    # type: (bytes, bytes, bytes, bytes) -> DSA_pub
    """
    Factory function that instantiates a DSA_pub object using
    the parameters and public key specified.

    :param p: value of p
    :param q: value of q
    :param g: value of g
    :param pub: value of the public key
    :return:  instance of DSA_pub.
    """
    m2.dsa_set_pub(dsa, pub)
    return DSA_pub(dsa, 1)

def load_pub_key(file, callback=util.passphrase_callback):
    # type: (AnyStr, Callable) -> DSA_pub
    """
    a DSA public key contained in PEM file.  The PEM file
    must contain the parameters in addition to the public key.

                     representation of the DSA public key.
                     invoked should the DSA public key be
    :return:         instance of DSA_pub.
    """
        ret = load_pub_key_bio(bio, callback)

    return ret

def load_pub_key_bio(bio, callback=util.passphrase_callback):
    # type: (BIO.BIO, Callable) -> DSA_pub
    """
    a DSA public key contained in PEM format.  The PEM

                     public key (with params).
    """
    dsapub = m2.dsa_read_pub_key(bio._ptr(), callback)
    return DSA_pub(dsapub, 1)

"""
M2Crypto wrapper for OpenSSL ECDH/ECDSA API.

@requires: OpenSSL 0.9.8 or newer

Copyright (c) 1999-2003 Ng Pheng Siong. All rights reserved.

Portions copyright (c) 2005-2006 Vrije Universiteit Amsterdam.
All rights reserved."""

from M2Crypto import BIO, Err, EVP, m2, util
from typing import AnyStr, Callable, Dict, Optional, Tuple, Union  # noqa
from M2Crypto.EVP import PKey

EC_Key = bytes

class ECError(Exception):
    pass

m2.ec_init(ECError)

# Curve identifier constants
NID_secp112r1 = m2.NID_secp112r1  # type: int
NID_secp112r2 = m2.NID_secp112r2  # type: int
NID_secp128r1 = m2.NID_secp128r1  # type: int
NID_secp128r2 = m2.NID_secp128r2  # type: int
NID_secp160k1 = m2.NID_secp160k1  # type: int
NID_secp160r1 = m2.NID_secp160r1  # type: int
NID_secp160r2 = m2.NID_secp160r2  # type: int
NID_secp192k1 = m2.NID_secp192k1  # type: int
NID_secp224k1 = m2.NID_secp224k1  # type: int
NID_secp224r1 = m2.NID_secp224r1  # type: int
NID_secp256k1 = m2.NID_secp256k1  # type: int
NID_secp384r1 = m2.NID_secp384r1  # type: int
NID_secp521r1 = m2.NID_secp521r1  # type: int
NID_sect113r1 = m2.NID_sect113r1  # type: int
NID_sect113r2 = m2.NID_sect113r2  # type: int
NID_sect131r1 = m2.NID_sect131r1  # type: int
NID_sect131r2 = m2.NID_sect131r2  # type: int
NID_sect163k1 = m2.NID_sect163k1  # type: int
NID_sect163r1 = m2.NID_sect163r1  # type: int
NID_sect163r2 = m2.NID_sect163r2  # type: int
NID_sect193r1 = m2.NID_sect193r1  # type: int
NID_sect193r2 = m2.NID_sect193r2  # type: int
# default for secg.org TLS test server
NID_sect233k1 = m2.NID_sect233k1  # type: int
NID_sect233r1 = m2.NID_sect233r1  # type: int
NID_sect239k1 = m2.NID_sect239k1  # type: int
NID_sect283k1 = m2.NID_sect283k1  # type: int
NID_sect283r1 = m2.NID_sect283r1  # type: int
NID_sect409k1 = m2.NID_sect409k1  # type: int
NID_sect409r1 = m2.NID_sect409r1  # type: int
NID_sect571k1 = m2.NID_sect571k1  # type: int
NID_sect571r1 = m2.NID_sect571r1  # type: int

NID_prime192v1 = m2.NID_X9_62_prime192v1  # type: int
NID_prime192v2 = m2.NID_X9_62_prime192v2  # type: int
NID_prime192v3 = m2.NID_X9_62_prime192v3  # type: int
NID_prime239v1 = m2.NID_X9_62_prime239v1  # type: int
NID_prime239v2 = m2.NID_X9_62_prime239v2  # type: int
NID_prime239v3 = m2.NID_X9_62_prime239v3  # type: int
NID_prime256v1 = m2.NID_X9_62_prime256v1  # type: int
NID_c2pnb163v1 = m2.NID_X9_62_c2pnb163v1  # type: int
NID_c2pnb163v2 = m2.NID_X9_62_c2pnb163v2  # type: int
NID_c2pnb163v3 = m2.NID_X9_62_c2pnb163v3  # type: int
NID_c2pnb176v1 = m2.NID_X9_62_c2pnb176v1  # type: int
NID_c2tnb191v1 = m2.NID_X9_62_c2tnb191v1  # type: int
NID_c2tnb191v2 = m2.NID_X9_62_c2tnb191v2  # type: int
NID_c2tnb191v3 = m2.NID_X9_62_c2tnb191v3  # type: int
NID_c2pnb208w1 = m2.NID_X9_62_c2pnb208w1  # type: int
NID_c2tnb239v1 = m2.NID_X9_62_c2tnb239v1  # type: int
NID_c2tnb239v2 = m2.NID_X9_62_c2tnb239v2  # type: int
NID_c2tnb239v3 = m2.NID_X9_62_c2tnb239v3  # type: int
NID_c2pnb272w1 = m2.NID_X9_62_c2pnb272w1  # type: int
NID_c2pnb304w1 = m2.NID_X9_62_c2pnb304w1  # type: int
NID_c2tnb359v1 = m2.NID_X9_62_c2tnb359v1  # type: int
NID_c2pnb368w1 = m2.NID_X9_62_c2pnb368w1  # type: int
NID_c2tnb431r1 = m2.NID_X9_62_c2tnb431r1  # type: int

# To preserve compatibility with older names
NID_X9_62_prime192v1 = NID_prime192v1  # type: int
NID_X9_62_prime192v2 = NID_prime192v2  # type: int
NID_X9_62_prime192v3 = NID_prime192v3  # type: int
NID_X9_62_prime239v1 = NID_prime239v1  # type: int
NID_X9_62_prime239v2 = NID_prime239v2  # type: int
NID_X9_62_prime239v3 = NID_prime239v3  # type: int
NID_X9_62_prime256v1 = NID_prime256v1  # type: int
NID_X9_62_c2pnb163v1 = NID_c2pnb163v1  # type: int
NID_X9_62_c2pnb163v2 = NID_c2pnb163v2  # type: int
NID_X9_62_c2pnb163v3 = NID_c2pnb163v3  # type: int
NID_X9_62_c2pnb176v1 = NID_c2pnb176v1  # type: int
NID_X9_62_c2tnb191v1 = NID_c2tnb191v1  # type: int
NID_X9_62_c2tnb191v2 = NID_c2tnb191v2  # type: int
NID_X9_62_c2tnb191v3 = NID_c2tnb191v3  # type: int
NID_X9_62_c2pnb208w1 = NID_c2pnb208w1  # type: int
NID_X9_62_c2tnb239v1 = NID_c2tnb239v1  # type: int
NID_X9_62_c2tnb239v2 = NID_c2tnb239v2  # type: int
NID_X9_62_c2tnb239v3 = NID_c2tnb239v3  # type: int
NID_X9_62_c2pnb272w1 = NID_c2pnb272w1  # type: int
NID_X9_62_c2pnb304w1 = NID_c2pnb304w1  # type: int
NID_X9_62_c2tnb359v1 = NID_c2tnb359v1  # type: int
NID_X9_62_c2pnb368w1 = NID_c2pnb368w1  # type: int
NID_X9_62_c2tnb431r1 = NID_c2tnb431r1  # type: int

NID_wap_wsg_idm_ecid_wtls1 = m2.NID_wap_wsg_idm_ecid_wtls1  # type: int
NID_wap_wsg_idm_ecid_wtls3 = m2.NID_wap_wsg_idm_ecid_wtls3  # type: int
NID_wap_wsg_idm_ecid_wtls4 = m2.NID_wap_wsg_idm_ecid_wtls4  # type: int
NID_wap_wsg_idm_ecid_wtls5 = m2.NID_wap_wsg_idm_ecid_wtls5  # type: int
NID_wap_wsg_idm_ecid_wtls6 = m2.NID_wap_wsg_idm_ecid_wtls6  # type: int
NID_wap_wsg_idm_ecid_wtls7 = m2.NID_wap_wsg_idm_ecid_wtls7  # type: int
NID_wap_wsg_idm_ecid_wtls8 = m2.NID_wap_wsg_idm_ecid_wtls8  # type: int
NID_wap_wsg_idm_ecid_wtls9 = m2.NID_wap_wsg_idm_ecid_wtls9  # type: int
NID_wap_wsg_idm_ecid_wtls10 = m2.NID_wap_wsg_idm_ecid_wtls10  # type: int
NID_wap_wsg_idm_ecid_wtls11 = m2.NID_wap_wsg_idm_ecid_wtls11  # type: int
NID_wap_wsg_idm_ecid_wtls12 = m2.NID_wap_wsg_idm_ecid_wtls12  # type: int

# The following two curves, according to OpenSSL, have a
# "Questionable extension field!" and are not supported by
# the OpenSSL inverse function.  ECError: no inverse.
# As such they cannot be used for signing.  They might,
# however, be usable for encryption but that has not
# been tested.  Until thir usefulness can be established,
# they are not supported at this time.
# NID_ipsec3 = m2.NID_ipsec3
# NID_ipsec4 = m2.NID_ipsec4

class EC(object):

    """
    Object interface to a EC key pair.
    """

    m2_ec_key_free = m2.ec_key_free

    def __init__(self, ec, _pyfree=0):
        # type: (EC, int) -> None
        assert m2.ec_key_type_check(ec), "'ec' type error"
        self.ec = ec

            self.m2_ec_key_free(self.ec)

        assert m2.ec_key_type_check(self.ec), "'ec' type error"
        return m2.ec_key_keylen(self.ec)

        """
        Generates the key pair from its parameters. Use::

            keypair = EC.gen_params(curve)
            keypair.gen_key()

        to create an EC key pair.
        """
        m2.ec_key_gen_key(self.ec)

    def pub(self):
        # type: () -> EC_pub
        # Don't let python free
        return EC_pub(self.ec, 0)

    def sign_dsa(self, digest):
        """
        Sign the given digest using ECDSA. Returns a tuple (r,s), the two
        ECDSA signature parameters.
        """
        assert self._check_key_type(), "'ec' type error"
        return m2.ecdsa_sign(self.ec, digest)

    def verify_dsa(self, digest, r, s):
        """
        Verify the given digest using ECDSA. r and s are the ECDSA
        signature parameters.
        """
        return m2.ecdsa_verify(self.ec, digest, r, s)

    def sign_dsa_asn1(self, digest):
        return m2.ecdsa_sign_asn1(self.ec, digest)

    def verify_dsa_asn1(self, digest, blob):
        return m2.ecdsa_verify_asn1(self.ec, digest, blob)

    def compute_dh_key(self, pub_key):
        # type: (EC) -> Optional[bytes]
        """
        Compute the ECDH shared key of this key pair and the given public
        key object. They must both use the same curve. Returns the
        shared key in binary as a buffer object. No Key Derivation Function is
        applied.
        """
        return m2.ecdh_compute_key(self.ec, pub_key.ec)

        # type: (BIO.BIO, Optional[str], Callable) -> int
        """
        Save the key pair to an M2Crypto.BIO.BIO object in PEM format.

        :param bio: M2Crypto.BIO.BIO object to save key to.

        :param cipher: Symmetric cipher to protect the key. The default
                       cipher is 'aes_128_cbc'. If cipher is None, then
                       the key is saved in the clear.

        :param callback: A Python callable object that is invoked
                         to acquire a passphrase with which to protect
                         the key. The default is
                         util.passphrase_callback.
        """
            return m2.ec_key_write_bio_no_cipher(self.ec, bio._ptr(), callback)
        else:
                raise ValueError('not such cipher %s' % cipher)
            return m2.ec_key_write_bio(self.ec, bio._ptr(), ciph(), callback)

    def save_key(self, file, cipher='aes_128_cbc',
        # type: (AnyStr, Optional[str], Callable) -> int
        """
        Save the key pair to a file in PEM format.

        :param file: Name of filename to save key to.

                         the key.  The default is
        """
        with BIO.openfile(file, 'wb') as bio:
            return self.save_key_bio(bio, cipher, callback)

        """
        Save the public key to an M2Crypto.BIO.BIO object in PEM format.

        """
        return m2.ec_key_write_pubkey(self.ec, bio._ptr())

    def save_pub_key(self, file):
        """
        Save the public key to a filename in PEM format.

        """
            return m2.ec_key_write_pubkey(self.ec, bio._ptr())

    def as_pem(self, cipher='aes_128_cbc', callback=util.passphrase_callback):
        """
        Returns the key(pair) as a string in PEM format.
        If no password is passed and the cipher is set
        it exits with error
        """
        with BIO.MemoryBuffer() as bio:
            self.save_key_bio(bio, cipher, callback)
            return bio.read()

    def _check_key_type(self):
        return m2.ec_key_type_check(self.ec)

        return m2.ec_key_check_key(self.ec)

class EC_pub(EC):

    """
    Object interface to an EC public key.
    ((don't like this implementation inheritance))
    """
        EC.__init__(self, ec, _pyfree)
        self.der = None  # type: Optional[bytes]

    def get_der(self):
        """
        Returns the public key in DER format as a buffer object.
        """
        if self.der is None:
            self.der = m2.ec_key_get_public_der(self.ec)
        return self.der

    def get_key(self):
        """
        Returns the public key as a byte string.
        """
        return m2.ec_key_get_public_key(self.ec)

    save_key = EC.save_pub_key

    save_key_bio = EC.save_pub_key_bio

def gen_params(curve):
    # type: (int) -> EC
    """
    Factory function that generates EC parameters and
    instantiates a EC object from the output.

    :param curve: This is the OpenSSL nid of the curve to use.
    """
    assert curve in [x['NID'] for x in m2.ec_get_builtin_curves()], \
        'Elliptic curve %s is not available on this system.' % \
        m2.obj_nid2sn(curve)
    return EC(m2.ec_key_new_by_curve_name(curve), 1)

    # type: (AnyStr, Callable) -> EC
    """
    Factory function that instantiates a EC object.

    :param file: Names the filename that contains the PEM representation
                 of the EC key pair.

    :param callback: Python callback object that will be invoked
                     if the EC key pair is passphrase-protected.
    """
        return load_key_bio(bio, callback)

def load_key_string(string, callback=util.passphrase_callback):
    # type: (str, Callable) -> EC
    """
    Load an EC key pair from a string.

    :param string: String containing EC key pair in PEM format.

    :param callback: A Python callable object that is invoked
                     to acquire a passphrase with which to unlock the
                     key. The default is util.passphrase_callback.

    :return: M2Crypto.EC.EC object.
    """
    with BIO.MemoryBuffer(string) as bio:

    # type: (BIO.BIO, Callable) -> EC
    """

    :param bio: M2Crypto.BIO object that contains the PEM
                representation of the EC key pair.

    """
    return EC(m2.ec_key_read_bio(bio._ptr(), callback), 1)

def load_pub_key(file):
    # type: (AnyStr) -> EC_pub
    """
    Load an EC public key from filename.

    :param file: Name of filename containing EC public key in PEM
                 format.

    :return: M2Crypto.EC.EC_pub object.
    """
        return load_pub_key_bio(bio)

def load_key_string_pubkey(string, callback=util.passphrase_callback):
    # type: (str, Callable) -> PKey
    """
    Load an M2Crypto.EC.PKey from a public key as a string.

    :param string: String containing the key in PEM format.

                     to acquire a passphrase with which to protect the
                     key.

    :return: M2Crypto.EC.PKey object.
    """
        return EVP.load_key_bio_pubkey(bio, callback)

def load_pub_key_bio(bio):
    # type: (BIO.BIO) -> EC_pub
    """
    Load an EC public key from an M2Crypto.BIO.BIO object.

    :param bio: M2Crypto.BIO.BIO object containing EC public key in PEM
                format.

    """
    ec = m2.ec_key_read_pubkey(bio._ptr())
    if ec is None:
        ec_error()
    return EC_pub(ec, 1)

def ec_error():
    # type: () -> ECError
    raise ECError(Err.get_error_message())

def pub_key_from_der(der):
    # type: (bytes) -> EC_pub
    """
    Create EC_pub from DER.
    """
    return EC_pub(m2.ec_key_from_pubkey_der(der), 1)

def pub_key_from_params(curve, bytes):
    # type: (bytes, bytes) -> EC_pub
    """
    Create EC_pub from curve name and octet string.
    """
    return EC_pub(m2.ec_key_from_pubkey_params(curve, bytes), 1)

def get_builtin_curves():
    # type: () -> Tuple[Dict[str, Union[int, str]]]
    return m2.ec_get_builtin_curves()

"""M2Crypto wrapper for OpenSSL EVP API.

Portions Copyright (c) 2004-2007 Open Source Applications Foundation.
Author: Heikki Toivonen
"""

from M2Crypto import BIO, Err, RSA, m2, util
from typing import AnyStr, Optional, Callable  # noqa

log = logging.getLogger('EVP')

class EVPError(ValueError):
    pass

m2.evp_init(EVPError)

def pbkdf2(password, salt, iter, keylen):
    # type: (bytes, bytes, int, int) -> bytes
    """
    Derive a key from password using PBKDF2 algorithm specified in RFC 2898.

    :param password: Derive the key from this password.
    :param salt:     Salt.
    :param iter:     Number of iterations to perform.
    :param keylen:   Length of key to produce.
    :return:         Key.
    """
    return m2.pkcs5_pbkdf2_hmac_sha1(password, salt, iter, keylen)

class MessageDigest(object):
    """
    Message Digest
    """
    m2_md_ctx_free = m2.md_ctx_free

    def __init__(self, algo):
        # type: (str) -> None
        md = getattr(m2, algo, None)  # type: Optional[Callable]
        if md is None:
            # if the digest algorithm isn't found as an attribute of the m2
            # module, try to look up the digest using get_digestbyname()
            self.md = m2.get_digestbyname(algo)
        else:
            self.md = md()
        self.ctx = m2.md_ctx_new()
        m2.digest_init(self.ctx, self.md)

        if getattr(self, 'ctx', None):
            self.m2_md_ctx_free(self.ctx)

    def update(self, data):
        """
        Add data to be digested.

        :return: -1 for Python error, 1 for success, 0 for OpenSSL failure.
        """
        return m2.digest_update(self.ctx, data)

    def final(self):
        return m2.digest_final(self.ctx)

    digest = final

class HMAC(object):

    m2_hmac_ctx_free = m2.hmac_ctx_free

    def __init__(self, key, algo='sha1'):
        # type: (bytes, str) -> None
        md = getattr(m2, algo, None)
            raise ValueError('unknown algorithm', algo)
        self.md = md()
        self.ctx = m2.hmac_ctx_new()
        m2.hmac_init(self.ctx, key, self.md)

            self.m2_hmac_ctx_free(self.ctx)

    def reset(self, key):
        # type: (bytes) -> None

        m2.hmac_update(self.ctx, data)

        return m2.hmac_final(self.ctx)

def hmac(key, data, algo='sha1'):
    # type: (bytes, bytes, str) -> bytes
    md = getattr(m2, algo, None)
    if md is None:
        raise ValueError('unknown algorithm', algo)
    return m2.hmac(key, data, md())

class Cipher(object):

    m2_cipher_ctx_free = m2.cipher_ctx_free

    def __init__(self, alg, key, iv, op, key_as_bytes=0, d='md5',
                 salt=b'12345678', i=1, padding=1):
        # type: (str, bytes, bytes, object, int, str, bytes, int, int) -> None
        cipher = getattr(m2, alg, None)
            raise ValueError('unknown cipher', alg)
        self.cipher = cipher()
        if key_as_bytes:
            kmd = getattr(m2, d, None)
            if kmd is None:
                raise ValueError('unknown message digest', d)
            key = m2.bytes_to_key(self.cipher, kmd(), key, salt, iv, i)
        self.ctx = m2.cipher_ctx_new()
        m2.cipher_init(self.ctx, self.cipher, key, iv, op)
        self.set_padding(padding)
        del key

            self.m2_cipher_ctx_free(self.ctx)

        return m2.cipher_update(self.ctx, data)

        return m2.cipher_final(self.ctx)

    def set_padding(self, padding=1):
        # type: (int) -> int
        """
        Actually always return 1
        """
        return m2.cipher_set_padding(self.ctx, padding)

class PKey(object):
    """
    Public Key
    """

    m2_pkey_free = m2.pkey_free

    def __init__(self, pkey=None, _pyfree=0, md='sha1'):
        # type: (Optional[bytes], int, str) -> None
        if pkey is not None:
            self.pkey = pkey  # type: bytes
        else:
            self.pkey = m2.pkey_new()
        self._set_context(md)

            self.m2_pkey_free(self.pkey)

        return self.pkey

    def _set_context(self, md):
        if not md:
            self.md = None
        else:
            mda = getattr(m2, md, None)  # type: Optional[Callable]
            if mda is None:
                raise ValueError('unknown message digest', md)
            self.md = mda()
        self.ctx = m2.md_ctx_new()  ## type: Context

    def reset_context(self, md='sha1'):
        """
        Reset internal message digest context.

        :param md: The message digest algorithm.
        """

    def sign_init(self):
        """
        Initialise signing operation with self.
        """
        m2.sign_init(self.ctx, self.md)

    def sign_update(self, data):
        """
        Feed data to signing operation.

        :param data: Data to be signed.
        """
        m2.sign_update(self.ctx, data)

    def sign_final(self):
        """
        Return signature.

        :return: The signature.
        """
        return m2.sign_final(self.ctx, self.pkey)

    # Deprecated
    update = sign_update
    final = sign_final

    def verify_init(self):
        """
        Initialise signature verification operation with self.
        """
        m2.verify_init(self.ctx, self.md)

    def verify_update(self, data):
        """
        Feed data to verification operation.

        :param data: Data to be verified.
        :return: -1 on Python error, 1 for success, 0 for OpenSSL error
        """
        return m2.verify_update(self.ctx, data)

    def verify_final(self, sign):
        """
        Return result of verification.

        :param sign: Signature to use for verification
        :return: Result of verification: 1 for success, 0 for failure, -1 on
                 other error.
        """
        return m2.verify_final(self.ctx, sign, self.pkey)

    def digest_sign_init(self):
        """
        Initialise digest signing operation with self.
        """
        if self.md is None:
            m2.digest_sign_init(self.ctx, self.pkey)
        else:
            m2.digest_sign_init(self.ctx, None, self.md, None, self.pkey)

    def digest_sign_update(self, data):
        """
        Feed data to digest signing operation.

        """
        m2.digest_sign_update(self.ctx, data)

    def digest_sign_final(self):
        """

        """
        return m2.digest_sign_final(self.ctx)

    def digest_sign(self, data):
        """

        """

        if m2.OPENSSL_VERSION_NUMBER < 0x10101000:
            raise NotImplemented('This method requires OpenSSL version ' +
                    '1.1.1 or greater.')

        return m2.digest_sign(self.ctx, data)

    def digest_verify_init(self):
        """
        Initialise verification operation with self.
        """
            m2.digest_verify_init(self.ctx, self.pkey)
        else:
            m2.digest_verify_init(self.ctx, None, self.md, None, self.pkey)

    def digest_verify_update(self, data):
        """

        """
        return m2.digest_verify_update(self.ctx, data)

    def digest_verify_final(self, sign):
        """
        Feed data to digest verification operation.

        """
        return m2.digest_verify_final(self.ctx, sign)

    def digest_verify(self, sign, data):
        """

        """

        return m2.digest_verify(self.ctx, sign, data)

    def assign_rsa(self, rsa, capture=1):
        # type: (RSA.RSA, int) -> int
        """
        Assign the RSA key pair to self.

        :param rsa: M2Crypto.RSA.RSA object to be assigned to self.

        :param capture: If true (default), this PKey object will own the RSA
                        object, meaning that once the PKey object gets
                        deleted it is no longer safe to use the RSA object.

        :return: Return 1 for success and 0 for failure.
        """
        if capture:
            ret = m2.pkey_assign_rsa(self.pkey, rsa.rsa)
            if ret:
                rsa._pyfree = 0
        else:
            ret = m2.pkey_set1_rsa(self.pkey, rsa.rsa)
        return ret

    def get_rsa(self):
        # type: () -> RSA.RSA_pub
        """
        Return the underlying RSA key if that is what the EVP
        instance is holding.
        """
        rsa_ptr = m2.pkey_get1_rsa(self.pkey)

        rsa = RSA.RSA_pub(rsa_ptr, 1)
        return rsa

        """

        :param file: Name of file to save key to.

        """

        """
        Save the key pair to the M2Crypto.BIO object 'bio' in PEM format.

        :param bio: M2Crypto.BIO object to save key to.

        """
            return m2.pkey_write_pem_no_cipher(self.pkey, bio._ptr(), callback)
        else:
            proto = getattr(m2, cipher, None)
            if proto is None:
                raise ValueError('no such cipher %s' % cipher)
            return m2.pkey_write_pem(self.pkey, bio._ptr(), proto(), callback)

        # type: (Optional[str], Callable) -> bytes
        """
        Return key in PEM format in a string.

                       cipher is ``'aes_128_cbc'``. If cipher is None,
                       then the key is saved in the clear.

        """
        bio = BIO.MemoryBuffer()
        self.save_key_bio(bio, cipher, callback)
        return bio.read_all()

    def as_der(self):
        """
        Return key in DER format in a string
        """
        buf = m2.pkey_as_der(self.pkey)
        bio = BIO.MemoryBuffer(buf)

    def size(self):
        """
        Return the size of the key in bytes.
        """
        return m2.pkey_size(self.pkey)

    def get_modulus(self):
        # type: () -> Optional[bytes]
        """
        Return the modulus in hex format.
        """
        return m2.pkey_get_modulus(self.pkey)

    # type: (AnyStr, Callable) -> PKey
    """
    Load an M2Crypto.EVP.PKey from file.

    :param file: Name of file containing the key in PEM format.

                     key.

    :return: M2Crypto.EVP.PKey object.
    """
    with BIO.openfile(file, 'r') as bio:
        cptr = m2.pkey_read_pem(bio.bio, callback)

    return PKey(cptr, 1)

def load_key_pubkey(file, callback=util.passphrase_callback):
    """
    Load an M2Crypto.EVP.PKey from a public key as a file.

                     key.

    """

        cptr = m2.pkey_read_pem_pubkey(bio._ptr(), callback)
        if cptr is None:
            raise EVPError(Err.get_error())

    # type: (BIO.BIO, Callable) -> PKey
    """
    Load an M2Crypto.EVP.PKey from an M2Crypto.BIO object.

    :param bio: M2Crypto.BIO object containing the key in PEM format.

                     key.

    """
    cptr = m2.pkey_read_pem(bio._ptr(), callback)

def load_key_bio_pubkey(bio, callback=util.passphrase_callback):
    """
    Load an M2Crypto.EVP.PKey from a public key as a M2Crypto.BIO object.

                     key.

    """
    cptr = m2.pkey_read_pem_pubkey(bio._ptr(), callback)
    if cptr is None:
        raise EVPError(Err.get_error())

    """
    Load an M2Crypto.EVP.PKey from a string.

                     key.

    """
    bio = BIO.MemoryBuffer(string)
    return load_key_bio(bio, callback)

    """
    Load an M2Crypto.EVP.PKey from a public key as a string.

                     key.

    """
    return load_key_bio_pubkey(bio, callback)

# vim: sts=4 sw=4 et

"""
M2Crypto wrapper for OpenSSL ENGINE API.

Pavel Shramov
IMEC MSU
"""

from M2Crypto import EVP, Err, X509, m2, six

class EngineError(Exception):
    pass

m2.engine_init_error(EngineError)

class Engine(object):
    """Wrapper for ENGINE object."""

    m2_engine_free = m2.engine_free

    def __init__(self, id=None, _ptr=None, _pyfree=1):
        # type: (Optional[bytes], Optional[bytes], int) -> None
        """Create new Engine from ENGINE pointer or obtain by id"""
        if not _ptr and not id:
            raise ValueError("No engine id specified")
        self._ptr = _ptr
        if not self._ptr:
            self._ptr = m2.engine_by_id(id)
            if not self._ptr:
                raise ValueError("Unknown engine: %s" % id)

            self.m2_engine_free(self._ptr)

    def init(self):
        """Obtain a functional reference to the engine.

        :return: 0 on error, non-zero on success."""
        return m2.engine_init(self._ptr)

    def finish(self):
        """Release a functional and structural reference to the engine."""
        return m2.engine_finish(self._ptr)

    def ctrl_cmd_string(self, cmd, arg, optional=0):
        # type: (AnyStr, Optional[AnyStr], int) -> None
        """Call ENGINE_ctrl_cmd_string"""
        cmd = six.ensure_str(cmd)
        if arg is not None:
            arg = six.ensure_str(arg)
        if not m2.engine_ctrl_cmd_string(self._ptr, cmd, arg, optional):
            raise EngineError(Err.get_error())

    def get_name(self):
        """Return engine name"""
        return m2.engine_get_name(self._ptr)

    def get_id(self):
        """Return engine id"""
        return m2.engine_get_id(self._ptr)

    def set_default(self, methods=m2.ENGINE_METHOD_ALL):
        """
        Use this engine as default for methods specified in argument

        :param methods: Possible values are bitwise OR of m2.ENGINE_METHOD_*
        """
        return m2.engine_set_default(self._ptr, methods)

    def _engine_load_key(self, func, name, pin=None):
        # type: (Callable, bytes, Optional[bytes]) -> EVP.PKey
        """Helper function for loading keys"""
        ui = m2.ui_openssl()
        cbd = m2.engine_pkcs11_data_new(pin)
        try:
            kptr = func(self._ptr, name, ui, cbd)
            if not kptr:
                raise EngineError(Err.get_error())
            key = EVP.PKey(kptr, _pyfree=1)
        finally:
            m2.engine_pkcs11_data_free(cbd)
        return key

    def load_private_key(self, name, pin=None):
        # type: (bytes, Optional[bytes]) -> X509.X509
        """Load private key with engine methods (e.g from smartcard).
            If pin is not set it will be asked
        """
        return self._engine_load_key(m2.engine_load_private_key, name, pin)

    def load_public_key(self, name, pin=None):
        # type: (bytes, Optional[bytes]) -> EVP.PKey
        """Load public key with engine methods (e.g from smartcard)."""
        return self._engine_load_key(m2.engine_load_public_key, name, pin)

    def load_certificate(self, name):
        # type: (bytes) -> X509.X509
        """Load certificate from engine (e.g from smartcard).
        NOTE: This function may be not implemented by engine!"""
        cptr = m2.engine_load_certificate(self._ptr, name)
        if not cptr:
            raise EngineError("Certificate or card not found")
        return X509.X509(cptr, _pyfree=1)

def load_dynamic_engine(id, sopath):
    # type: (bytes, AnyStr) -> Engine
    """Load and return dymanic engine from sopath and assign id to it"""
    if isinstance(sopath, six.text_type):
        sopath = sopath.encode('utf8')
    m2.engine_load_dynamic()
    e = Engine('dynamic')
    e.ctrl_cmd_string('SO_PATH', sopath)
    e.ctrl_cmd_string('ID', id)
    e.ctrl_cmd_string('LIST_ADD', '1')
    e.ctrl_cmd_string('LOAD', None)
    return e

def load_dynamic():
    # type: () -> None
    """Load dynamic engine"""

def load_openssl():
    """Load openssl engine"""
    m2.engine_load_openssl()

def cleanup():
    """If you load any engines, you need to clean up after your application
    is finished with the engines."""
    m2.engine_cleanup()

"""M2Crypto wrapper for OpenSSL Error API.

from M2Crypto import BIO, m2, util, six  # noqa

def get_error():
    # type: () -> Optional[str]
    err = BIO.MemoryBuffer()
    m2.err_print_errors(err.bio_ptr())
    err_msg = err.read()
    if err_msg:
        return six.ensure_text(err_msg)

def get_error_code():
    # type: () -> int
    return m2.err_get_error()

def peek_error_code():
    return m2.err_peek_error()

def get_error_lib(err):
    # type: (Optional[int]) -> str
    err_str = m2.err_lib_error_string(err)
    return six.ensure_text(err_str) if err_str else ''

def get_error_func(err):
    err_str = m2.err_func_error_string(err)

def get_error_reason(err):
    err_str = m2.err_reason_error_string(err)

def get_error_message():
    # type: () -> str
    return six.ensure_text(get_error_reason(get_error_code()))

def get_x509_verify_error(err):
    err_str = m2.x509_get_verify_error(err)

class SSLError(Exception):
    def __init__(self, err, client_addr):
        # type: (int, util.AddrType) -> None
        self.err = err
        self.client_addr = client_addr

        if not isinstance(self.client_addr, six.text_type):
            s = self.client_addr.decode('utf8')
        else:
            s = self.client_addr
        return "%s: %s: %s" % (get_error_func(self.err), s,
                               get_error_reason(self.err))

class M2CryptoError(Exception):
    pass

"""M2Crypto wrapper for OpenSSL RC4 API.

from M2Crypto.m2 import rc4_free, rc4_new, rc4_set_key, rc4_update

class RC4(object):
    """Object interface to the stream cipher RC4."""

    rc4_free = rc4_free

    def __init__(self, key=None):
        self.cipher = rc4_new()
        if key:
            rc4_set_key(self.cipher, key)

        if getattr(self, 'cipher', None):
            self.rc4_free(self.cipher)

    def set_key(self, key):
        rc4_set_key(self.cipher, key)

        return rc4_update(self.cipher, data)

        return ''

"""M2Crypto wrapper for OpenSSL RSA API.

import sys

from M2Crypto import BIO, Err, m2, util
from typing import Any, AnyStr, Callable, IO, Optional, Tuple  # noqa

class RSAError(Exception):
    pass

m2.rsa_init(RSAError)

no_padding = m2.no_padding
pkcs1_padding = m2.pkcs1_padding
sslv23_padding = m2.sslv23_padding
pkcs1_oaep_padding = m2.pkcs1_oaep_padding

class RSA(object):
    """
    RSA Key Pair.
    """

    m2_rsa_free = m2.rsa_free

    def __init__(self, rsa, _pyfree=0):
        """
        :param rsa: binary representation of OpenSSL RSA type
        """
        assert m2.rsa_type_check(rsa), "'rsa' type error"
        self.rsa = rsa

            self.m2_rsa_free(self.rsa)

        return int(m2.rsa_size(self.rsa) << 3)

        if name == 'e':
            return m2.rsa_get_e(self.rsa)
        elif name == 'n':
            return m2.rsa_get_n(self.rsa)
        else:

        # type: () -> Tuple[bytes, bytes]
        return m2.rsa_get_e(self.rsa), m2.rsa_get_n(self.rsa)

    def public_encrypt(self, data, padding):
        # type: (bytes, int) -> bytes
        return m2.rsa_public_encrypt(self.rsa, data, padding)

    def public_decrypt(self, data, padding):
        return m2.rsa_public_decrypt(self.rsa, data, padding)

    def private_encrypt(self, data, padding):
        return m2.rsa_private_encrypt(self.rsa, data, padding)

    def private_decrypt(self, data, padding):
        return m2.rsa_private_decrypt(self.rsa, data, padding)

        """

        """
            return m2.rsa_write_key_no_cipher(self.rsa, bio._ptr(), callback)
        else:
                raise RSAError('not such cipher %s' % cipher)
            else:
            return m2.rsa_write_key(self.rsa, bio._ptr(), ciph, callback)

        """

        """

    save_pem = save_key

        """
        """
        return bio.read()

    def save_key_der_bio(self, bio):
        """
        Save the key pair to an M2Crypto.BIO.BIO object in DER format.

        """
        return m2.rsa_write_key_der(self.rsa, bio._ptr())

    def save_key_der(self, file):
        """
        Save the key pair to a file in DER format.

        :param file: Filename to save key to
        """
            return self.save_key_der_bio(bio)

        """

        """
        return m2.rsa_write_pub_key(self.rsa, bio._ptr())

        """
        Save the public key to a file in PEM format.

        """
            return m2.rsa_write_pub_key(self.rsa, bio._ptr())

        """
        Validate RSA keys.

        It checks that p and q are in fact prime, and that n = p*q.

        :return: returns 1 if rsa is a valid RSA key, and 0 otherwise.
                 -1 is returned if an error occurs while checking the key.
                 If the key is invalid or an error occurred, the reason
                 code can be obtained using ERR_get_error(3).
        """
        return m2.rsa_check_key(self.rsa)

    def sign_rsassa_pss(self, digest, algo='sha1', salt_length=20):
        # type: (bytes, str, int) -> bytes
        """
        Signs a digest with the private key using RSASSA-PSS

        :param digest: A digest created by using the digest method

        :param salt_length: The length of the salt to use

        :param algo: The hash algorithm to use
                     Legal values like 'sha1','sha224', 'sha256',
                     'ripemd160', and 'md5'.

        :return: a string which is the signature
        """
        hash = getattr(m2, algo, None)

        if hash is None:
            raise RSAError('not such hash algorithm %s' % algo)

        signature = m2.rsa_padding_add_pkcs1_pss(self.rsa, digest, hash(), salt_length)

        return self.private_encrypt(signature, m2.no_padding)

    def verify_rsassa_pss(self, data, signature, algo='sha1', salt_length=20):
        # type: (bytes, bytes, str, int) -> int
        """
        Verifies the signature RSASSA-PSS

        :param data: Data that has been signed

        :param signature: The signature signed with RSASSA-PSS

        :param salt_length: The length of the salt that was used

                     Legal values are for example 'sha1','sha224',
                     'sha256', 'ripemd160', and 'md5'.

        :return: 1 or 0, depending on whether the signature was
                 verified or not.
        """

        plain_signature = self.public_decrypt(signature, m2.no_padding)

        return m2.rsa_verify_pkcs1_pss(self.rsa, data, plain_signature, hash(), salt_length)

    def sign(self, digest, algo='sha1'):
        # type: (bytes, str) -> bytes
        """
        Signs a digest with the private key

        :param algo: The method that created the digest.

        """
        digest_type = getattr(m2, 'NID_' + algo, None)
        if digest_type is None:

        return m2.rsa_sign(self.rsa, digest, digest_type)

    def verify(self, data, signature, algo='sha1'):
        # type: (bytes, bytes, str) -> int
        """
        Verifies the signature with the public key

        :param signature: The signature signed with the private key

        :param algo: The method use to create digest from the data
                     before it was signed.  Legal values like
                     'sha1','sha224', 'sha256', 'ripemd160', and 'md5'.

        """

        return m2.rsa_verify(self.rsa, data, signature, digest_type)

class RSA_pub(RSA):

    """
    Object interface to an RSA public key.
    """

        if name in ['e', 'n']:
            raise RSAError('use factory function new_pub_key() to set (e, n)')
        else:

    def private_encrypt(self, *argv):
        raise RSAError('RSA_pub object has no private key')

    def private_decrypt(self, *argv):

    def save_key(self, file, *args, **kw):
        # type: (AnyStr, *Any, **Any) -> int
        """
        Save public key to file.
        """
        return self.save_pub_key(file)

    def save_key_bio(self, bio, *args, **kw):
        # type: (BIO.BIO, *Any, **Any) -> int
        """
        Save public key to BIO.
        """
        return self.save_pub_key_bio(bio)

    # save_key_der

    # save_key_der_bio

        return m2.rsa_check_pub_key(self.rsa)

def rsa_error():
    raise RSAError(Err.get_error_message())

def keygen_callback(p, n, out=sys.stdout):
    # type: (int, Any, IO[str]) -> None
    """
    Default callback for gen_key().
    """
    ch = ['.', '+', '*', '\n']
    out.write(ch[p])
    out.flush()

def gen_key(bits, e, callback=keygen_callback):
    # type: (int, int, Callable) -> RSA
    """
    Generate an RSA key pair.

    :param bits: Key length, in bits.

    :param e: The RSA public exponent.

                     during key generation; its usual purpose is to
                     provide visual feedback. The default callback is
                     keygen_callback.

    :return: M2Crypto.RSA.RSA object.
    """
    return RSA(m2.rsa_generate_key(bits, e, callback), 1)

    # type: (AnyStr, Callable) -> RSA
    """
    Load an RSA key pair from file.

    :param file: Name of file containing RSA public key in PEM format.

                     key.  The default is util.passphrase_callback.

    """

    # type: (BIO.BIO, Callable) -> RSA
    """
    Load an RSA key pair from an M2Crypto.BIO.BIO object.

    :param bio: M2Crypto.BIO.BIO object containing RSA key pair in PEM
                format.

    """
    rsa = m2.rsa_read_key(bio._ptr(), callback)
    if rsa is None:
        rsa_error()
    return RSA(rsa, 1)

    """
    Load an RSA key pair from a string.

    :param string: String containing RSA key pair in PEM format.

    """

    # type: (AnyStr) -> RSA_pub
    """
    Load an RSA public key from file.

    :return: M2Crypto.RSA.RSA_pub object.
    """

    # type: (BIO.BIO) -> RSA_pub
    """
    Load an RSA public key from an M2Crypto.BIO.BIO object.

    :param bio: M2Crypto.BIO.BIO object containing RSA public key in PEM
                format.

    """
    rsa = m2.rsa_read_pub_key(bio._ptr())
    return RSA_pub(rsa, 1)

def new_pub_key(e_n):
    # type: (Tuple[bytes, bytes]) -> RSA_pub
    """
    Instantiate an RSA_pub object from an (e, n) tuple.

    :param e: The RSA public exponent; it is a string in OpenSSL's MPINT
              format - 4-byte big-endian bit-count followed by the
              appropriate number of bits.

    :param n: The RSA composite of primes; it is a string in OpenSSL's
              MPINT format - 4-byte big-endian bit-count followed by the

    """
    (e, n) = e_n
    rsa = m2.rsa_new()
    m2.rsa_set_en(rsa, e, n)

"""M2Crypto wrapper for OpenSSL PRNG. Requires OpenSSL 0.9.5 and above.

Copyright (c) 2014-2017 Matej Cepl. All rights reserved.

See LICENCE for the license information.
"""

from typing import AnyStr, Tuple  # noqa

__all__ = ['rand_seed', 'rand_add', 'load_file', 'save_file', 'rand_bytes',
           'rand_pseudo_bytes', 'rand_file_name', 'rand_status']

class RandError(ValueError):
    pass

m2.rand_init(RandError)

def rand_add(blob, entropy):
    # type: (bytes, float) -> None
    """
    Mixes blob into the PRNG state.

    :param blob: added data
    :param entropy: (the lower bound of) an estimate of how much randomness
        is contained in blob, measured in bytes.

    Thus, if the data at buf are unpredictable to an adversary, this
    increases the uncertainty about the state and makes the PRNG output less
    predictable. Suitable input comes from user interaction (random key
    presses, mouse movements) and certain hardware events.

    Details about sources of randomness and how to estimate their entropy
    can be found in the literature, e.g. RFC 1750.
    """
    m2.rand_add(blob, entropy)  # pylint: disable=no-member

def rand_seed(seed):
    # type: (bytes) -> None
    """
    Equivalent to rand_add() when len(seed) == entropy.

    :param seed: added data (see description at rand_add)
    """
    m2.rand_seed(seed)  # pylint: disable=no-member

def rand_status():
    """
    Check whether there is enough entropy in PRNG.

    :return: 1 if the PRNG has been seeded with enough
       data, 0 otherwise.
    """
    return m2.rand_status()  # pylint: disable=no-member

def rand_file_name():
    """
    Generate a default path for the random seed file.

    :return: string with the filename.
       The seed file is $RANDFILE if that environment variable
       is set, $HOME/.rnd otherwise. If $HOME is not set either,
       an error occurs.
    """
    return six.ensure_text(m2.rand_file_name())  # pylint: disable=no-member

def load_file(filename, max_bytes):
    # type: (AnyStr, int) -> int
    """
    Read a number of bytes from file filename and adds them to the PRNG.

    If max_bytes is non-negative, up to to max_bytes are read; starting with
    OpenSSL 0.9.5, if max_bytes is -1, the complete file is read.

    :param filename:
    :param max_bytes:
    :return: the number of bytes read.
    """
    return m2.rand_load_file(six.ensure_str(filename), max_bytes)  # pylint: disable=no-member

def save_file(filename):
    # type: (AnyStr) -> int
    """
    Write a number of random bytes (currently 1024) to file.

    The file then can be used to initialize the PRNG by calling load_file() in
    a later session.

    :return: returns the number of bytes written, and -1 if the bytes
         written were generated without appropriate seed.
    """
    return m2.rand_save_file(filename)  # pylint: disable=no-member

def rand_bytes(num):
    # type: (int) -> bytes
    """
    Return n cryptographically strong pseudo-random bytes.

    An error occurs if the PRNG has not been seeded with enough randomness
    to ensure an unpredictable byte sequence.

    :param num: number of bytes to be returned
    :return: random bytes
    """
    return m2.rand_bytes(num)  # pylint: disable=no-member

def rand_pseudo_bytes(num):
    # type: (int) -> Tuple[bytes, int]
    """
    Return num pseudo-random bytes into buf.

    Pseudo-random byte sequences generated by this method will be unique
    if they are of sufficient length, but are not necessarily
    unpredictable. They can be used for non-cryptographic purposes and for
    certain purposes in cryptographic protocols, but usually not for key
    generation etc.

    Output of the function is mixed into the entropy pool before
    retrieving the new pseudo-random bytes unless disabled at compile
    time (see FAQ).

    """
    if m2.OPENSSL_VERSION_NUMBER >= 0x10100000:
        warnings.warn('The underlying OpenSSL method has been ' +
                      'deprecated. Use Rand.rand_bytes instead.',
                      DeprecationWarning)

    return m2.rand_pseudo_bytes(num)  # pylint: disable=no-member

"""M2Crypto wrapper for OpenSSL S/MIME API.

from M2Crypto import BIO, EVP, Err, X509, m2, util

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
        else:
            self.pkcs7 = m2.pkcs7_new()

            self.m2_pkcs7_free(self.pkcs7)

        return self.pkcs7

    def type(self, text_name=0):
        if text_name:
            return m2.pkcs7_type_sn(self.pkcs7)
        else:
            return m2.pkcs7_type_nid(self.pkcs7)

    def write(self, bio):
        return m2.pkcs7_write_bio(self.pkcs7, bio._ptr())

    def write_der(self, bio):
        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):
    with BIO.openfile(p7file, 'rb') as bio:
        p7_ptr = m2.pkcs7_read_bio_der(bio.bio)

def load_pkcs7_bio(p7_bio):
    # type: (BIO.BIO) -> PKCS7
    p7_ptr = m2.pkcs7_read_bio(p7_bio._ptr())

def load_pkcs7_bio_der(p7_bio):
    p7_ptr = m2.pkcs7_read_bio_der(p7_bio._ptr())

def smime_load_pkcs7(p7file):
    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):
    p7_ptr, bio_ptr = m2.smime_read_pkcs7(p7_bio._ptr())
    if p7_ptr is None:
        raise SMIME_Error(Err.get_error())
    else:

    """Object interface to EVP_CIPHER without all the frills of
    M2Crypto.EVP.Cipher.
    """

        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,
        # 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,
        # 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):
        del self.pkey
        del self.x509

    def unset_x509_stack(self):
        del self.x509_stack

    def unset_x509_store(self):
        del self.x509_store

    def unset_cipher(self):
        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

            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(),

    def verify(self, pkcs7, data_bio=None, flags=0):
        # type: (PKCS7, BIO.BIO, int) -> Optional[bytes]
            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)

    def write(self, out_bio, pkcs7, data_bio=None, flags=0):
        # type: (BIO.BIO, PKCS7, Optional[BIO.BIO], int) -> int
        assert isinstance(pkcs7, PKCS7)
            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:

def text_crlf_bio(bio_in):
    # type: (BIO.BIO) -> BIO.BIO
        return bio_out
    else:

"""M2Crypto wrapper for OpenSSL X509 API.

Copyright (C) 2004-2007 OSAF. All Rights Reserved.
"""

import binascii

from M2Crypto import ASN1, BIO, EVP, m2, six  # noqa
from typing import AnyStr, List, Optional  # noqa

FORMAT_DER = 0
FORMAT_PEM = 1

verify_allow_proxy_certs = m2.VERIFY_ALLOW_PROXY_CERTS
verify_cb_issuer_check = m2.VERIFY_CB_ISSUER_CHECK
verify_check_ss_signature = m2.VERIFY_CHECK_SS_SIGNATURE
verify_crl_check = m2.VERIFY_CRL_CHECK
verify_crl_check_all = m2.VERIFY_CRL_CHECK_ALL
verify_explicit_policy = m2.VERIFY_EXPLICIT_POLICY
verify_extended_crl_support = m2.VERIFY_EXTENDED_CRL_SUPPORT
verify_ignore_critical = m2.VERIFY_IGNORE_CRITICAL
verify_inhibit_any = m2.VERIFY_INHIBIT_ANY
verify_inhibit_map = m2.VERIFY_INHIBIT_MAP
verify_no_alt_chains = m2.VERIFY_NO_ALT_CHAINS
if hasattr(m2, "VERIFY_NO_CHECK_TIME"):
    verify_no_check_time = m2.VERIFY_NO_CHECK_TIME
verify_notify_policy = m2.VERIFY_NOTIFY_POLICY
verify_partial_chain = m2.VERIFY_PARTIAL_CHAIN
verify_policy_check = m2.VERIFY_POLICY_CHECK
verify_trusted_first = m2.VERIFY_TRUSTED_FIRST
verify_use_deltas = m2.VERIFY_USE_DELTAS
verify_x509_strict = m2.VERIFY_X509_STRICT

class X509Error(ValueError):
    pass

m2.x509_init(X509Error)

V_OK = m2.X509_V_OK  # type: int

def x509_store_default_cb(ok, ctx):
    # type: (int, X509_Store_Context) -> int
    return ok

def new_extension(name, value, critical=0, _pyfree=1):
    # type: (str, bytes, int, int) -> X509_Extension
    """
    Create new X509_Extension instance.
    """
    if name == 'subjectKeyIdentifier' and \
            value.strip('0123456789abcdefABCDEF:') != '':
        raise ValueError('value must be precomputed hash')
    ctx = m2.x509v3_set_nconf()
    x509_ext_ptr = m2.x509v3_ext_conf(None, ctx, name, value)
    if x509_ext_ptr is None:
        raise X509Error(
            "Cannot create X509_Extension with name '%s' and value '%s'" %
            (name, value))
    x509_ext = X509_Extension(x509_ext_ptr, _pyfree)
    x509_ext.set_critical(critical)
    return x509_ext

class X509_Extension(object):
    """
    X509 Extension
    """

    m2_x509_extension_free = m2.x509_extension_free

    def __init__(self, x509_ext_ptr=None, _pyfree=1):
        self.x509_ext = x509_ext_ptr

        if getattr(self, '_pyfree', 0) and self.x509_ext:
            self.m2_x509_extension_free(self.x509_ext)

        return self.x509_ext

    def set_critical(self, critical=1):
        """
        Mark this extension critical or noncritical. By default an
        extension is not critical.

        :param critical: Nonzero sets this extension as critical.
                         Calling this method without arguments will
                         set this extension to critical.
        :return: 1 for success, 0 for failure
        """
        return m2.x509_extension_set_critical(self.x509_ext, critical)

    def get_critical(self):
        """
        Return whether or not this is a critical extension.

        :return:  Nonzero if this is a critical extension.
        """
        return m2.x509_extension_get_critical(self.x509_ext)

        """
        Get the extension name, for example 'subjectAltName'.
        """
        return six.ensure_text(m2.x509_extension_get_name(self.x509_ext))

    def get_value(self, flag=0, indent=0):
        # type: (int, int) -> str
        """
        Get the extension value, for example 'DNS:www.example.com'.

        :param flag:   Flag to control what and how to print.
        :param indent: How many spaces to print before actual value.
        """
        m2.x509_ext_print(buf.bio_ptr(), self.x509_ext, flag, indent)

class X509_Extension_Stack(object):
    """
    X509 Extension Stack

    :warning: Do not modify the underlying OpenSSL stack
              except through this interface, or use any OpenSSL
              functions that do so indirectly. Doing so will get the
              OpenSSL stack and the internal pystack of this class out
              of sync, leading to python memory leaks, exceptions or
              even python crashes!
    """

    m2_sk_x509_extension_free = m2.sk_x509_extension_free

    def __init__(self, stack=None, _pyfree=0):
        if stack is not None:
            self.stack = stack
            num = m2.sk_x509_extension_num(self.stack)
            for i in range(num):
                self.pystack.append(X509_Extension(
                    m2.sk_x509_extension_value(self.stack, i),
                    _pyfree=_pyfree))
        else:
            self.stack = m2.sk_x509_extension_new_null()
            self.pystack = []  # This must be kept in sync with self.stack

        # see BIO.py - unbalanced __init__ / __del__
            self.m2_sk_x509_extension_free(self.stack)

        assert m2.sk_x509_extension_num(self.stack) == len(self.pystack)
        return len(self.pystack)

    def __getitem__(self, idx):
        # type: (int) -> X509_Extension
        return self.pystack[idx]

    def __iter__(self):
        return iter(self.pystack)

        return self.stack

    def push(self, x509_ext):
        # type: (X509_Extension) -> int
        """
        Push X509_Extension object onto the stack.

        :param x509_ext: X509_Extension object to be pushed onto the stack.
        :return: The number of extensions on the stack.
        """
        self.pystack.append(x509_ext)
        ret = m2.sk_x509_extension_push(self.stack, x509_ext._ptr())
        assert ret == len(self.pystack)
        return ret

    def pop(self):
        # type: () -> X509_Extension
        """
        Pop X509_Extension object from the stack.

        :return: X509_Extension popped
        """
        x509_ext_ptr = m2.sk_x509_extension_pop(self.stack)
        if x509_ext_ptr is None:
            assert len(self.pystack) == 0
            return None
        return self.pystack.pop()

class X509_Name_Entry(object):
    """
    X509 Name Entry
    """

    m2_x509_name_entry_free = m2.x509_name_entry_free

    def __init__(self, x509_name_entry, _pyfree=0):
        """
            :param x509_name_entry: this should be OpenSSL X509_NAME_ENTRY binary
            :param _pyfree:
        """
        self.x509_name_entry = x509_name_entry

            self.m2_x509_name_entry_free(self.x509_name_entry)

        return self.x509_name_entry

    def set_object(self, asn1obj):
        # type: (ASN1.ASN1_Object) -> int
        """
        Sets the field name to asn1obj

        :param asn1obj:
        :return: 0 on failure, 1 on success
        """
        return m2.x509_name_entry_set_object(self.x509_name_entry,
                                             asn1obj._ptr())

    def set_data(self, data, type=ASN1.MBSTRING_ASC):
        # type: (bytes, int) -> int
        """

        :param data: data in a binary form to be set
        """
        return m2.x509_name_entry_set_data(self.x509_name_entry, type, data)

    def get_object(self):
        # type: () -> ASN1.ASN1_Object
        return ASN1.ASN1_Object(
            m2.x509_name_entry_get_object(self.x509_name_entry))

    def get_data(self):
        # type: () -> ASN1.ASN1_String
        return ASN1.ASN1_String(
            m2.x509_name_entry_get_data(self.x509_name_entry))

    def create_by_txt(self, field, type, entry, len):
        return m2.x509_name_entry_create_by_txt(self.x509_name_entry._ptr(),
                                                field, type, entry, len)

class X509_Name(object):
    """
    X509 Name
    """

    nid = {'C': m2.NID_countryName,
           'SP': m2.NID_stateOrProvinceName,
           'ST': m2.NID_stateOrProvinceName,
           'stateOrProvinceName': m2.NID_stateOrProvinceName,
           'L': m2.NID_localityName,
           'localityName': m2.NID_localityName,
           'O': m2.NID_organizationName,
           'organizationName': m2.NID_organizationName,
           'OU': m2.NID_organizationalUnitName,
           'organizationUnitName': m2.NID_organizationalUnitName,
           'CN': m2.NID_commonName,
           'commonName': m2.NID_commonName,
           'Email': m2.NID_pkcs9_emailAddress,
           'emailAddress': m2.NID_pkcs9_emailAddress,
           'serialNumber': m2.NID_serialNumber,
           'SN': m2.NID_surname,
           'surname': m2.NID_surname,
           'GN': m2.NID_givenName,
           'givenName': m2.NID_givenName
           }

    m2_x509_name_free = m2.x509_name_free

    def __init__(self, x509_name=None, _pyfree=0):
        """
        :param x509_name: this should be OpenSSL X509_NAME binary
        :param _pyfree:
        """
        if x509_name is not None:
            assert m2.x509_name_type_check(x509_name), "'x509_name' type error"
            self.x509_name = x509_name
        else:
            self.x509_name = m2.x509_name_new()

            self.m2_x509_name_free(self.x509_name)

        assert m2.x509_name_type_check(self.x509_name), \
            "'x509_name' type error"
        return m2.x509_name_oneline(self.x509_name)

    def __getattr__(self, attr):
        # type: (str) -> str
        if attr in self.nid:
            assert m2.x509_name_type_check(self.x509_name), \
                "'x509_name' type error"
            return six.ensure_text(m2.x509_name_by_nid(self.x509_name, self.nid[attr]))

        if attr in self.__dict__:
            return self.__dict__[attr]

        raise AttributeError(self, attr)

    def __setattr__(self, attr, value):
        # type: (str, AnyStr) -> int
        """
        :return: 1 for success of 0 if an error occurred.
        """
            return m2.x509_name_set_by_nid(self.x509_name, self.nid[attr],
                                           six.ensure_binary(value))

        self.__dict__[attr] = value

        return m2.x509_name_entry_count(self.x509_name)

        # type: (int) -> X509_Name_Entry
        if not 0 <= idx < self.entry_count():
            raise IndexError("index out of range")
        return X509_Name_Entry(m2.x509_name_get_entry(self.x509_name, idx))

        for i in range(self.entry_count()):
            yield self[i]

        return self.x509_name

    def add_entry_by_txt(self, field, type, entry, len, loc, set):
        # entry_type: (str, int, bytes, int, int, int) -> int
        """
        Add X509_Name field whose name is identified by its name.

        :param field: name of the entry
        :param type: use MBSTRING_ASC or MBSTRING_UTF8
               (or standard ASN1 type like V_ASN1_IA5STRING)
        :param entry: value
        :param len: buf_len of the entry
               (-1 and the length is computed automagically)

        The ``loc`` and ``set`` parameters determine where a new entry
        should be added.
        For almost all applications loc can be set to -1 and set to 0.
        This adds a new entry to the end of name as a single valued
        RelativeDistinguishedName (RDN).

        :param loc: determines the index where the new entry is
               inserted: if it is -1 it is appended.
        :param set: determines how the new type is added. If it is zero
               a new RDN is created.
               If set is -1 or 1 it is added to the previous or next RDN
               structure respectively. This will then be a multivalued
               RDN: since multivalues RDNs are very seldom used set is
               almost always set to zero.

        """
        return m2.x509_name_add_entry_by_txt(self.x509_name,
                                             six.ensure_str(field), type,
                                             six.ensure_str(entry), len, loc, set)

    def entry_count(self):

    def get_entries_by_nid(self, nid):
        # type: (int) -> List[X509_Name_Entry]
        """
        Retrieve the next index matching nid.

        :param nid: name of the entry (as m2.NID* constants)

        :return: list of X509_Name_Entry items
        """
        ret = []
        lastpos = -1

        while True:
            lastpos = m2.x509_name_get_index_by_nid(self.x509_name, nid,
                                                    lastpos)
            if lastpos == -1:
                break

            ret.append(self[lastpos])

        return ret

    def as_text(self, indent=0, flags=m2.XN_FLAG_COMPAT):
        """
        as_text returns the name as a string.

        :param indent: Each line in multiline format is indented
                       by this many spaces.
        :param flags:  Flags that control how the output should be formatted.
        """
        m2.x509_name_print_ex(buf.bio_ptr(), self.x509_name, indent, flags)

        return m2.x509_name_get_der(self.x509_name)

    def as_hash(self):
        return m2.x509_name_hash(self.x509_name)

class X509(object):
    """
    X.509 Certificate
    """

    m2_x509_free = m2.x509_free

    def __init__(self, x509=None, _pyfree=0):
        """
        :param x509: binary representation of
               the underlying OpenSSL X509 object.
        """
        if x509 is not None:
            assert m2.x509_type_check(x509), "'x509' type error"
            self.x509 = x509
        else:
            self.x509 = m2.x509_new()

            self.m2_x509_free(self.x509)

        assert m2.x509_type_check(self.x509), "'x509' type error"
        return self.x509

    def as_text(self):
        m2.x509_print(buf.bio_ptr(), self.x509)

        return m2.i2d_x509(self.x509)

    def as_pem(self):
        m2.x509_write_pem(buf.bio_ptr(), self.x509)

    def save_pem(self, filename):
        """
        :param filename: name of the file to be loaded
        :return: 1 for success or 0 for failure
        """
            return m2.x509_write_pem(bio.bio_ptr(), self.x509)

    def save(self, filename, format=FORMAT_PEM):
        # type: (AnyStr, int) -> int
        """
        Saves X.509 certificate to a file. Default output
        format is PEM.

        :param filename: Name of the file the cert will be saved to.

        :param format: Controls what output format is used to save the cert.
                       Either FORMAT_PEM or FORMAT_DER to save in PEM or
                       DER format.  Raises a ValueError if an unknow
                       format is used.

        """
            if format == FORMAT_PEM:
                return m2.x509_write_pem(bio.bio_ptr(), self.x509)
            elif format == FORMAT_DER:
                return m2.i2d_x509_bio(bio.bio_ptr(), self.x509)
            else:
                raise ValueError(
                    "Unknown filetype. Must be either FORMAT_PEM or FORMAT_DER")

    def set_version(self, version):
        """
        Set version of the certificate.

        :param version: Version number.
        :return:        Returns 0 on failure.
        """
        return m2.x509_set_version(self.x509, version)

    def set_not_before(self, asn1_time):
        # type: (ASN1.ASN1_TIME) -> int
        """
        :return: 1 on success, 0 on failure
        """
        return m2.x509_set_not_before(self.x509, asn1_time._ptr())

    def set_not_after(self, asn1_time):
        """
        """
        return m2.x509_set_not_after(self.x509, asn1_time._ptr())

    def set_subject_name(self, name):
        # type: (X509_Name) -> int
        """
        """
        return m2.x509_set_subject_name(self.x509, name.x509_name)

    def set_issuer_name(self, name):
        """
        """
        return m2.x509_set_issuer_name(self.x509, name.x509_name)

    def get_version(self):
        return m2.x509_get_version(self.x509)

    def get_serial_number(self):
        # type: () -> ASN1.ASN1_Integer
        asn1_integer = m2.x509_get_serial_number(self.x509)
        return m2.asn1_integer_get(asn1_integer)

    def set_serial_number(self, serial):
        # type: (ASN1.ASN1_Integer) -> int
        """
        Set serial number.

        :param serial:  Serial number.

        :return 1 for success and 0 for failure.
        """
        # This "magically" changes serial since asn1_integer
        # is C pointer to x509's internal serial number.
        return m2.asn1_integer_set(asn1_integer, serial)
        # XXX Or should I do this?
        # asn1_integer = m2.asn1_integer_new()
        # m2.asn1_integer_set(asn1_integer, serial)
        # return m2.x509_set_serial_number(self.x509, asn1_integer)

    def get_not_before(self):
        # type: () -> ASN1.ASN1_TIME
        return ASN1.ASN1_TIME(m2.x509_get_not_before(self.x509))

    def get_not_after(self):
        out = ASN1.ASN1_TIME(m2.x509_get_not_after(self.x509))
        if 'Bad time value' in str(out):
            raise X509Error(
                '''M2Crypto cannot handle dates after year 2050.
                See RFC 5280 4.1.2.5 for more information.
                ''')
        return out

    def get_pubkey(self):
        # type: () -> EVP.PKey
        return EVP.PKey(m2.x509_get_pubkey(self.x509), _pyfree=1)

    def set_pubkey(self, pkey):
        # type: (EVP.PKey) -> int
        """
        Set the public key for the certificate

        :param pkey: Public key

        :return 1 for success and 0 for failure
        """
        return m2.x509_set_pubkey(self.x509, pkey.pkey)

    def get_issuer(self):
        # type: () -> X509_Name
        return X509_Name(m2.x509_get_issuer_name(self.x509))

    def set_issuer(self, name):
        """
        Set issuer name.

        :param name:    subjectName field.

        """

    def get_subject(self):
        return X509_Name(m2.x509_get_subject_name(self.x509))

    def set_subject(self, name):
        """
        Set subject name.

        """

    def add_ext(self, ext):
        """
        Add X509 extension to this certificate.

        :param ext:    Extension

        """
        return m2.x509_add_ext(self.x509, ext.x509_ext, -1)

    def get_ext(self, name):
        # type: (str) -> X509_Extension
        """
        Get X509 extension by name.

        :param name:    Name of the extension

        :return:       X509_Extension
        """
        # Optimizations to reduce attribute accesses
        m2x509_get_ext = m2.x509_get_ext
        m2x509_extension_get_name = m2.x509_extension_get_name
        x509 = self.x509

        name = six.ensure_binary(name)
        for i in range(m2.x509_get_ext_count(x509)):
            ext_ptr = m2x509_get_ext(x509, i)
            if m2x509_extension_get_name(ext_ptr) == name:
                return X509_Extension(ext_ptr, _pyfree=0)

        raise LookupError

    def get_ext_at(self, index):
        """
        Get X509 extension by index.

        :param index:    Name of the extension

        :return:        X509_Extension
        """
        if index < 0 or index >= self.get_ext_count():
            raise IndexError

        return X509_Extension(m2.x509_get_ext(self.x509, index),
                              _pyfree=0)

    def get_ext_count(self):
        """
        Get X509 extension count.
        """
        return m2.x509_get_ext_count(self.x509)

    def sign(self, pkey, md):
        # type: (EVP.PKey, str) -> int
        """
        Sign the certificate.

        :param md:   Message digest algorithm to use for signing,
                     for example 'sha1'.

        :return int
        """
        mda = getattr(m2, md, None)
        if mda is None:
            raise ValueError('unknown message digest', md)
        return m2.x509_sign(self.x509, pkey.pkey, mda())

    def verify(self, pkey=None):
        # type: (Optional[EVP.PKey]) -> int
        if pkey:
            return m2.x509_verify(self.x509, pkey.pkey)
        else:
            return m2.x509_verify(self.x509, self.get_pubkey().pkey)

    def check_ca(self):
        """
        Check if the certificate is a Certificate Authority (CA) certificate.

        :return: 0 if the certificate is not CA, nonzero otherwise.

        :requires: OpenSSL 0.9.8 or newer
        """
        return m2.x509_check_ca(self.x509)

    def check_purpose(self, id, ca):
        # type: (int, int) -> int
        """
        Check if the certificate's purpose matches the asked purpose.

        :param id: Purpose id. See X509_PURPOSE_* constants.

        :param ca: 1 if the certificate should be CA, 0 otherwise.

        :return: 0 if the certificate purpose does not match, nonzero
                 otherwise.
        """
        return m2.x509_check_purpose(self.x509, id, ca)

    def get_fingerprint(self, md='md5'):
        """
        Get the fingerprint of the certificate.

        :param md: Message digest algorithm to use.

        :return:   String containing the fingerprint in hex format.
        """
        der = self.as_der()
        md = EVP.MessageDigest(md)
        md.update(der)
        return six.ensure_text(binascii.hexlify(digest).upper())

def load_cert(file, format=FORMAT_PEM):
    # type: (AnyStr, int) -> X509
    """
    Load certificate from file.

    :param file: Name of file containing certificate in either DER or
                 PEM format.

    :param format: Describes the format of the file to be loaded,
                   either PEM or DER.

    :return: M2Crypto.X509.X509 object.
    """
        if format == FORMAT_PEM:
            return load_cert_bio(bio)
        elif format == FORMAT_DER:
            cptr = m2.d2i_x509(bio._ptr())
            return X509(cptr, _pyfree=1)
        else:
            raise ValueError(
                "Unknown format. Must be either FORMAT_DER or FORMAT_PEM")

def load_cert_bio(bio, format=FORMAT_PEM):
    # type: (BIO.BIO, int) -> X509
    """
    Load certificate from a bio.

    :param bio: BIO pointing at a certificate in either DER or PEM format.

    :param format: Describes the format of the cert to be loaded,
                   either PEM or DER (via constants FORMAT_PEM
                   and FORMAT_FORMAT_DER)

    """
    if format == FORMAT_PEM:
        cptr = m2.x509_read_pem(bio._ptr())
    elif format == FORMAT_DER:
        cptr = m2.d2i_x509(bio._ptr())
    else:
        raise ValueError(
            "Unknown format. Must be either FORMAT_DER or FORMAT_PEM")
    return X509(cptr, _pyfree=1)

def load_cert_string(string, format=FORMAT_PEM):
    """
    Load certificate from a string.

    :param string: String containing a certificate in either DER or PEM format.

    """
    string = six.ensure_binary(string)
    return load_cert_bio(bio, format)

def load_cert_der_string(string):
    # type: (AnyStr) -> X509
    """

    :param string: String containing a certificate in DER format.

    """
    cptr = m2.d2i_x509(bio._ptr())

class X509_Store_Context(object):
    """
    X509 Store Context
    """

    m2_x509_store_ctx_free = m2.x509_store_ctx_free

    def __init__(self, x509_store_ctx, _pyfree=0):
        """

        :param x509_store_ctx: binary data for
              OpenSSL X509_STORE_CTX type
        """
        self.ctx = x509_store_ctx

        if not hasattr(self, '_pyfree'):
            pass  # print("OOPS")
        elif self._pyfree:
            self.m2_x509_store_ctx_free(self.ctx)

        return self.ctx

    def get_current_cert(self):
        # type: () -> X509
        """
        Get current X.509 certificate.

        :warning: The returned certificate is NOT refcounted, so you can not
                  rely on it being valid once the store context goes
                  away or is modified.
        """
        return X509(m2.x509_store_ctx_get_current_cert(self.ctx), _pyfree=0)

    def get_error(self):
        """
        Get error code.
        """
        return m2.x509_store_ctx_get_error(self.ctx)

    def get_error_depth(self):
        """
        Get error depth.
        """
        return m2.x509_store_ctx_get_error_depth(self.ctx)

    def get1_chain(self):
        # type: () -> X509_Stack
        """
        Get certificate chain.

        :return: Reference counted (i.e. safe to use even after the store
                 context goes away) stack of certificates in the chain.
        """
        return X509_Stack(m2.x509_store_ctx_get1_chain(self.ctx), 1, 1)

class X509_Store(object):
    """
    X509 Store
    """

    m2_x509_store_free = m2.x509_store_free

    def __init__(self, store=None, _pyfree=0):
        """
        :param store: binary data for OpenSSL X509_STORE_CTX type.
        """
        if store is not None:
            self.store = store
        else:
            self.store = m2.x509_store_new()

            self.m2_x509_store_free(self.store)

        return self.store

    def load_info(self, file):
        """
        :param file: filename

        """
        ret = m2.x509_store_load_locations(self.store, file)
        return ret

    load_locations = load_info

    def add_x509(self, x509):
        # type: (X509) -> int
        assert isinstance(x509, X509)
        return m2.x509_store_add_cert(self.store, x509._ptr())

    def set_verify_cb(self, callback=None):
        # type: (Optional[callable]) -> None
        """
        Set callback which will be called when the store is verified.
        Wrapper over OpenSSL X509_STORE_set_verify_cb().

        :param callback:    Callable to specify verification options.
                            Type of the callable must be:
                            (int, X509_Store_Context) -> int.
                            If None: set the standard options.

        :note: compile-time or run-time errors in the callback would result
               in mysterious errors during verification, which could be hard
               to trace.

        :note: Python exceptions raised in callbacks do not propagate to
               verify() call.

        :return: None
        """
        if callback is None:
            return self.set_verify_cb(x509_store_default_cb)

        if not callable(callback):
            raise X509Error("set_verify(): callback is not callable")
        return m2.x509_store_set_verify_cb(self.store, callback)

    add_cert = add_x509

    def set_flags(self, flags):
        """
        Set the verification flags for the X509Store
        Wrapper over OpenSSL X509_STORE_set_flags()

        :param flags: `VERIFICATION FLAGS` section of the
                      X509_VERIFY_PARAM_set_flags man page has
                      a complete description of values the flags
                      parameter can take.
                      Their M2Crypto equivalent is transformed following
                      the pattern: "X509_V_FLAG_XYZ" -> lowercase("VERIFY_XYZ")
        """
        return m2.x509_store_set_flags(self.store, flags)

class X509_Stack(object):
    """
    X509 Stack

    """

    m2_sk_x509_free = m2.sk_x509_free

    def __init__(self, stack=None, _pyfree=0, _pyfree_x509=0):
        # type: (bytes, int, int) -> None
            num = m2.sk_x509_num(self.stack)
                self.pystack.append(X509(m2.sk_x509_value(self.stack, i),
                                         _pyfree=_pyfree_x509))
        else:
            self.stack = m2.sk_x509_new_null()

            self.m2_sk_x509_free(self.stack)

        assert m2.sk_x509_num(self.stack) == len(self.pystack)

        # type: (int) -> X509

    def push(self, x509):
        """
        push an X509 certificate onto the stack.

        :param x509: X509 object.

        :return: The number of X509 objects currently on the stack.
        """
        self.pystack.append(x509)
        ret = m2.sk_x509_push(self.stack, x509._ptr())
        return ret

        """
        pop a certificate from the stack.

        :return: X509 object that was popped, or None if there is
                 nothing to pop.
        """
        x509_ptr = m2.sk_x509_pop(self.stack)
        if x509_ptr is None:

        """
        Return the stack as a DER encoded string
        """
        return m2.get_der_encoding_stack(self.stack)

def new_stack_from_der(der_string):
    # type: (bytes) -> X509_Stack
    """
    Create a new X509_Stack from DER string.

    :return: X509_Stack
    """
    der_string = six.ensure_binary(der_string)
    stack_ptr = m2.make_stack_from_der_sequence(der_string)
    return X509_Stack(stack_ptr, 1, 1)

class Request(object):
    """
    X509 Certificate Request.
    """

    m2_x509_req_free = m2.x509_req_free

    def __init__(self, req=None, _pyfree=0):
        # type: (Optional[int], int) -> None
        if req is not None:
            self.req = req
        else:
            self.req = m2.x509_req_new()
            m2.x509_req_set_version(self.req, 0)

            self.m2_x509_req_free(self.req)

        m2.x509_req_print(buf.bio_ptr(), self.req)

        m2.x509_req_write_pem(buf.bio_ptr(), self.req)

        m2.i2d_x509_req_bio(buf.bio_ptr(), self.req)

            return m2.x509_req_write_pem(bio.bio_ptr(), self.req)

        """
        Saves X.509 certificate request to a file. Default output

        :param filename: Name of the file the request will be saved to.

        :param format: Controls what output format is used to save the
                       request. Either FORMAT_PEM or FORMAT_DER to save
                       in PEM or DER format. Raises ValueError if an
                       unknown format is used.

        :return: 1 for success, 0 for failure.
                 The error code can be obtained by ERR_get_error.
        """
                return m2.x509_req_write_pem(bio.bio_ptr(), self.req)
                return m2.i2d_x509_req_bio(bio.bio_ptr(), self.req)
            else:
                    "Unknown filetype. Must be either FORMAT_DER or FORMAT_PEM")

        """
        Get the public key for the request.

        :return:     Public key from the request.
        """
        return EVP.PKey(m2.x509_req_get_pubkey(self.req), _pyfree=1)

        """
        Set the public key for the request.

        :return:     Return 1 for success and 0 for failure.
        """
        return m2.x509_req_set_pubkey(self.req, pkey.pkey)

        """
        Get version.

        :return:        Returns version.
        """
        return m2.x509_req_get_version(self.req)

        """
        Set version.

        """
        return m2.x509_req_set_version(self.req, version)

        return X509_Name(m2.x509_req_get_subject_name(self.req))

        """

        :return:    1 for success and 0 for failure
        """
        return m2.x509_req_set_subject_name(self.req, name.x509_name)

    set_subject = set_subject_name

    def add_extensions(self, ext_stack):
        # type: (X509_Extension_Stack) -> int
        """
        Add X509 extensions to this request.

        :param ext_stack: Stack of extensions to add.
        :return: 1 for success and 0 for failure
        """
        return m2.x509_req_add_extensions(self.req, ext_stack._ptr())

    def verify(self, pkey):
        """

        :param pkey: PKey to be verified
        """
        return m2.x509_req_verify(self.req, pkey.pkey)

        """

        :param pkey: PKey to be signed
        :param md: used algorigthm
        """
        return m2.x509_req_sign(self.req, pkey.pkey, mda())

def load_request(file, format=FORMAT_PEM):
    # type: (AnyStr, int) -> Request
    """
    Load certificate request from file.

    :param file: Name of file containing certificate request in
                 either PEM or DER format.
                   either PEM or DER. (using constants FORMAT_PEM
                   and FORMAT_DER)
    :return: Request object.
    """
    with BIO.openfile(file) as f:
            cptr = m2.x509_req_read_pem(f.bio_ptr())
            cptr = m2.d2i_x509_req(f.bio_ptr())
        else:
                "Unknown filetype. Must be either FORMAT_PEM or FORMAT_DER")

    return Request(cptr, 1)

def load_request_bio(bio, format=FORMAT_PEM):
    # type: (BIO.BIO, int) -> Request
    """
    Load certificate request from a bio.

    :param bio: BIO pointing at a certificate request in
                either DER or PEM format.
    :param format: Describes the format of the request to be loaded,
    :return: M2Crypto.X509.Request object.
    """
        cptr = m2.x509_req_read_pem(bio._ptr())
        cptr = m2.d2i_x509_req(bio._ptr())
    else:

    return Request(cptr, _pyfree=1)

def load_request_string(string, format=FORMAT_PEM):
    """
    Load certificate request from a string.

    :param string: String containing a certificate request in
                   either DER or PEM format.

    """
    return load_request_bio(bio, format)

def load_request_der_string(string):
    # type: (AnyStr) -> Request
    """

    :param string: String containing a certificate request in DER format.
    """
    return load_request_bio(bio, FORMAT_DER)

class CRL(object):
    """
    X509 Certificate Revocation List
    """

    m2_x509_crl_free = m2.x509_crl_free

    def __init__(self, crl=None, _pyfree=0):
        """

        :param crl: binary representation of
               the underlying OpenSSL X509_CRL object.
        """
        if crl is not None:
            self.crl = crl
        else:
            self.crl = m2.x509_crl_new()

            self.m2_x509_crl_free(self.crl)

        """
        Return CRL in PEM format in a string.

        :return: String containing the CRL in PEM format.
        """
        m2.x509_crl_print(buf.bio_ptr(), self.crl)

def load_crl(file):
    # type: (AnyStr) -> CRL
    """
    Load CRL from file.

    :param file: Name of file containing CRL in PEM format.

    :return: M2Crypto.X509.CRL object.
    """
        cptr = m2.x509_crl_read_pem(f.bio_ptr())

    return CRL(cptr, 1)

"""M2Crypto client-side FTP/TLS.

This implementation complies with draft-murray-auth-ftp-ssl-07.txt.

Example:

>>> from M2Crypto import ftpslib
>>> f = ftpslib.FTP_TLS()
>>> f.connect('', 9021)
'220 spinnaker.dyndns.org M2Crypto (Medusa) FTP/TLS server v0.07 ready.'
>>> f.auth_tls()
>>> f.set_pasv(0)
>>> f.login('ftp', 'ngps@')
'230 Ok.'
>>> f.retrlines('LIST')
-rw-rw-r--   1 0        198          2326 Jul  3  1996 apache_pb.gif
drwxrwxr-x   7 0        198          1536 Oct 10  2000 manual
drwxrwxr-x   2 0        198           512 Oct 31  2000 modpy
drwxrwxr-x   2 0        198           512 Oct 31  2000 bobo
drwxr-xr-x   2 0        198         14336 May 28 15:54 postgresql
drwxr-xr-x   4 100      198           512 May 16 17:19 home
drwxr-xr-x   7 100      100          3584 Sep 23  2000 openacs
drwxr-xr-x  10 0        0             512 Aug  5  2000 python1.5
-rw-r--r--   1 100      198           326 Jul 29 03:29 index.html
drwxr-xr-x  12 0        0             512 May 31 17:08 python2.1
'226 Transfer complete'
>>> f.quit()
'221 Goodbye.'
>>>

# We want to import whole stdlib ftplib objects, because our users want
# to use them.
from ftplib import *  # noqa

# M2Crypto
from M2Crypto import SSL

class FTP_TLS(FTP):

    """Python OO interface to client-side FTP/TLS."""

    def __init__(self, host=None, ssl_ctx=None):
        """Initialise the client. If 'host' is supplied, connect to it."""
        if ssl_ctx is not None:
            self.ssl_ctx = ssl_ctx
        else:
            self.ssl_ctx = SSL.Context()
        FTP.__init__(self, host)
        self.prot = 0

    def auth_tls(self):
        """Secure the control connection per AUTH TLS, aka AUTH TLS-C."""
        self.voidcmd('AUTH TLS')
        s = SSL.Connection(self.ssl_ctx, self.sock)
        s.setup_ssl()
        s.set_connect_state()
        s.connect_ssl()
        self.sock = s
        self.file = self.sock.makefile()

    def auth_ssl(self):
        """Secure the control connection per AUTH SSL, aka AUTH TLS-P."""
        raise NotImplementedError

    def prot_p(self):
        """Set up secure data connection."""
        self.voidcmd('PBSZ 0')
        self.voidcmd('PROT P')
        self.prot = 1

    def prot_c(self):
        """Set up data connection in the clear."""
        self.voidcmd('PROT C')

    def ntransfercmd(self, cmd, rest=None):
        """Initiate a data transfer."""
        conn, size = FTP.ntransfercmd(self, cmd, rest)
        if self.prot:
            conn = SSL.Connection(self.ssl_ctx, conn)
            conn.setup_ssl()
            conn.set_connect_state()
            conn.set_session(self.sock.get_session())
            conn.connect_ssl()
        return conn, size

import warnings

"""M2Crypto support for Python's httplib.

import base64
import socket

from M2Crypto import SSL, six
from M2Crypto.six.moves.urllib_parse import urlsplit, urlunsplit
from M2Crypto.six.moves.http_client import *  # noqa
# This is not imported with just '*'
from M2Crypto.six.moves.http_client import HTTPS_PORT
from typing import Any, AnyStr, Callable, Dict, Optional  # noqa

class HTTPSConnection(HTTPConnection):

    """
    This class allows communication via SSL using M2Crypto.
    """

    default_port = HTTPS_PORT

    def __init__(self, host, port=None, strict=None, **ssl):
        # type: (str, Optional[int], Optional[bool], **Any) -> None
        """
        Represents one transaction with an HTTP server over the SSL
        connection.

        :param host: host name
        :param port: port number
        :param strict: if switched on, it raises BadStatusLine to be
                       raised if the status line can't be parsed as
                       a valid HTTP/1.0 or 1.1 status line.
        :param ssl: dict with all remaining named real parameters of the
                    function. Specifically, ``ssl_context`` is expected
                    to be included with SSL.Context; if it is not
                    default ``'sslv23'`` is substituted).
        """
        self.session = None  # type: bytes
        self.host = host
        self.port = port
        keys = set(ssl.keys()) - set(('key_file', 'cert_file', 'ssl_context'))
        if keys:
            raise ValueError('unknown keyword argument: %s', keys)
        try:
            self.ssl_ctx = ssl['ssl_context']
            assert isinstance(self.ssl_ctx, SSL.Context), self.ssl_ctx
        except KeyError:
        HTTPConnection.__init__(self, host, port, strict)

    def connect(self):
        error = None
        # We ignore the returned sockaddr because SSL.Connection.connect needs
        # a host name.
        for (family, _, _, _, _) in \
                socket.getaddrinfo(self.host, self.port, 0,
                                   socket.SOCK_STREAM):
            sock = None
            try:
                sock = SSL.Connection(self.ssl_ctx, family=family)

                # set SNI server name since we know it at this point
                sock.set_tlsext_host_name(self.host)

                if self.session is not None:
                    sock.set_session(self.session)
                sock.connect((self.host, self.port))

                self.sock = sock
                sock = None
                return
            except socket.error as e:
                # Other exception are probably SSL-related, in that case we
                # abort and the exception is forwarded to the caller.
                error = e
            finally:
                if sock is not None:
                    sock.close()

        if error is None:
            raise AssertionError("Empty list returned by getaddrinfo")
        raise error

        # This kludges around line 545 of httplib.py,
        # which closes the connection in this object;
        # the connection remains open in the response
        # object.
        #
        # M2Crypto doesn't close-here-keep-open-there,
        # so, in effect, we don't close until the whole
        # business is over and gc kicks in.
        #
        # XXX Long-running callers beware leakage.
        #
        # XXX 05-Jan-2002: This module works with Python 2.2,
        # XXX but I've not investigated if the above conditions
        # XXX remain.
        pass

    def get_session(self):
        # type: () -> SSL.Session.Session
        return self.sock.get_session()

    def set_session(self, session):
        # type: (SSL.Session.Session) -> None
        self.session = session

class ProxyHTTPSConnection(HTTPSConnection):
    """
    An HTTPS Connection that uses a proxy and the CONNECT request.

    When the connection is initiated, CONNECT is first sent to the proxy (along
    with authorization headers, if supplied). If successful, an SSL connection
    will be established over the socket through the proxy and to the target
    host.

    Finally, the actual request is sent over the SSL connection tunneling
    through the proxy.
    """

    _ports = {'http': 80, 'https': 443}
    _AUTH_HEADER = "Proxy-Authorization"
    _UA_HEADER = "User-Agent"

    def __init__(self, host, port=None, strict=None, username=None,
                 password=None, **ssl):
        # type: (str, Optional[int], Optional[bool], Optional[AnyStr], Optional[AnyStr], **Any) -> None
        """
        Create the ProxyHTTPSConnection object.

        :param host: host name of the proxy server
        :param port: port number of the proxy server
        :param username: username on the proxy server, when required
                         Username can be ``str``, but preferred type
                         is ``bytes``. M2Crypto does some conversion to
                         ``bytes`` when necessary, but it's better when
                         the user of the library does it on its own.
        :param password: password on the proxy server, when required
                         The same as with ``username``, ``str`` is accepted,
                         but ``bytes`` are preferred.
        """
        HTTPSConnection.__init__(self, host, port, strict, **ssl)

        self._username = username.encode('utf8') \
            if isinstance(username, six.string_types) else username
        self._password = password.encode('utf8') \
            if isinstance(password, six.string_types) else password
        self._proxy_auth = None  # type: str
        self._proxy_UA = None  # type: str

    def putrequest(self, method, url, skip_host=0, skip_accept_encoding=0):
        # type: (AnyStr, AnyStr, int, int) -> None
        """
        putrequest is called before connect, so can interpret url and get
        real host/port to be used to make CONNECT request to proxy
        """
        proto, netloc, path, query, fragment = urlsplit(url)
        if not proto:
            raise ValueError("unknown URL type: %s" % url)

        # get host & port
        try:
            username_password, host_port = netloc.split('@')
        except ValueError:
            host_port = netloc

        try:
            host, port_s = host_port.split(':')
            port = int(port_s)
            host = host_port
            # try to get port from proto
            try:
                port = self._ports[proto]
            except KeyError:
                raise ValueError("unknown protocol for: %s" % url)

        self._real_host = host  # type: str
        self._real_port = port  # type: int
        rest = urlunsplit(('', '', path, query, fragment))
        HTTPSConnection.putrequest(self, method, rest, skip_host,
                                   skip_accept_encoding)

    def putheader(self, header, value):
        # type: (AnyStr, AnyStr) -> None
        # Store the auth header if passed in.
        if header.lower() == self._UA_HEADER.lower():
            self._proxy_UA = value
        if header.lower() == self._AUTH_HEADER.lower():
            self._proxy_auth = value
        else:
            HTTPSConnection.putheader(self, header, value)

    def endheaders(self, *args, **kwargs):
        # type: (*Any, **Any) -> None
        # We've recieved all of hte headers. Use the supplied username
        # and password for authorization, possibly overriding the authstring
        # supplied in the headers.
        if not self._proxy_auth:
            self._proxy_auth = self._encode_auth()

        HTTPSConnection.endheaders(self, *args, **kwargs)

        HTTPConnection.connect(self)

        # send proxy CONNECT request
        self.sock.sendall(self._get_connect_msg())
        response = HTTPResponse(self.sock)
        response.begin()

        code = response.status
        if code != 200:
            # proxy returned and error, abort connection, and raise exception
            raise socket.error("Proxy connection failed: %d" % code)

        self._start_ssl()

    def _get_connect_msg(self):
        """ Return an HTTP CONNECT request to send to the proxy. """
        msg = "CONNECT %s:%d HTTP/1.1\r\n" % (self._real_host, self._real_port)
        msg = msg + "Host: %s:%d\r\n" % (self._real_host, self._real_port)
        if self._proxy_UA:
            msg = msg + "%s: %s\r\n" % (self._UA_HEADER, self._proxy_UA)
        if self._proxy_auth:
            msg = msg + "%s: %s\r\n" % (self._AUTH_HEADER, self._proxy_auth)
        msg = msg + "\r\n"
        return six.ensure_binary(msg)

    def _start_ssl(self):
        """ Make this connection's socket SSL-aware. """
        self.sock = SSL.Connection(self.ssl_ctx, self.sock)
        self.sock.setup_ssl()
        self.sock.set_connect_state()
        self.sock.connect_ssl()

    def _encode_auth(self):
        """ Encode the username and password for use in the auth header. """
        if not (self._username and self._password):
        # Authenticated proxy
        userpass = "%s:%s" % (self._username, self._password)
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", DeprecationWarning)
            enc_userpass = base64.encodestring(userpass).replace("\n", "")
        return six.ensure_binary("Basic %s" % enc_userpass)

"""M2Crypto enhancement to Python's urllib for handling
'https' url's.

FIXME: it is questionable whether we need this old-style module at all. urllib
(not urllib2) is in Python 3 support just as a legacy API.

from M2Crypto import SSL, httpslib, six

from M2Crypto.six.moves.urllib_response import addinfourl
from typing import AnyStr, Optional  # noqa

# six.moves doesn't support star imports
if six.PY3:
    from urllib.request import *  # noqa for other modules to import
    from urllib.parse import *  # noqa for other modules to import
    from urllib.error import *  # noqa for other modules to import
else:
    from urllib import *  # noqa

def open_https(self, url, data=None, ssl_context=None):
    # type: (URLOpener, AnyStr, Optional[bytes], Optional[SSL.Context]) -> addinfourl
    """
    Open URL over the SSL connection.

    :param url: URL to be opened
    :param data: data for the POST request
    :param ssl_context: SSL.Context to be used
    :return:
    """
    if six.PY3:
        warnings.warn('URLOpener has been deprecated in Py3k', DeprecationWarning)

    if ssl_context is not None and isinstance(ssl_context, SSL.Context):
        self.ctx = ssl_context
    else:
        self.ctx = SSL.Context()
    user_passwd = None
    if isinstance(url, six.string_types):
        try:               # python 2
            # http://pydoc.org/2.5.1/urllib.html
            host, selector = splithost(url)
            if host:
                user_passwd, host = splituser(host)
                host = unquote(host)
            realhost = host
        except NameError:  # python 3 has no splithost
            # https://docs.python.org/3/library/urllib.parse.html
            parsed = urlparse(url)
            host = parsed.hostname
            if parsed.port:
                host += ":{0}".format(parsed.port)
            user_passwd = parsed.password
            if parsed.password:
                user_passwd += ":{0}".format(parsed.password)
            selector = parsed.path
    else:
        host, selector = url
        urltype, rest = splittype(selector)
        url = rest
        user_passwd = None
        if urltype.lower() != 'http':
            realhost = None
        else:
            try:               # python 2
                realhost, rest = splithost(rest)
                if realhost:
                    user_passwd, realhost = splituser(realhost)
                    if user_passwd:
                        selector = "%s://%s%s" % (urltype, realhost, rest)
            except NameError:  # python 3 has no splithost
                parsed = urlparse(rest)
                host = parsed.hostname
                if parsed.port:
                    host += ":{0}".format(parsed.port)
                user_passwd = parsed.username
                if parsed.password:
                    user_passwd += ":{0}".format(parsed.password)
        # print("proxy via http:", host, selector)
    if not host:
        raise IOError('http error', 'no host given')
    if user_passwd:
        if six.PY3:
            auth = base64.encodebytes(user_passwd).strip()
        else:
            auth = base64.encodestring(user_passwd).strip()
    else:
        auth = None
    # Start here!
    h = httpslib.HTTPSConnection(host=host, ssl_context=self.ctx)
    # h.set_debuglevel(1)
    # Stop here!
    if data is not None:
        h.putrequest('POST', selector)
        h.putheader('Content-type', 'application/x-www-form-urlencoded')
        h.putheader('Content-length', '%d' % len(data))
    else:
        h.putrequest('GET', selector)
    if auth:
        h.putheader('Authorization', 'Basic %s' % auth)
    for args in self.addheaders:
        h.putheader(*args)   # for python3 - used to use apply
    h.endheaders()
        h.send(data + '\r\n')
    # Here again!
    resp = h.getresponse()
    fp = resp.fp
    return addinfourl(fp, resp.msg, "https:" + url)
    # Stop again.

# Minor brain surgery.
URLopener.open_https = open_https

"""
M2Crypto enhancement to Python's urllib2 for handling

Code from urllib2 is Copyright (c) 2001, 2002, 2003, 2004, 2005, 2006, 2007
Python Software Foundation; All Rights Reserved

Summary of changes:
 - Use an HTTPSProxyConnection if the request is going through a proxy.
 - Add the SSL context to the https connection when performing https_open.
 - Add the M2Crypto HTTPSHandler when building a default opener.
"""

from M2Crypto.six.moves.urllib_parse import urldefrag, urlparse as url_parse

    from urllib.request import *  # noqa other modules want to import
    from urllib.error import *  # noqa other modules want to import
else:
    from urllib2 import *  # noqa

try:
    mother_class = socket._fileobject
except AttributeError:
    mother_class = socket.SocketIO

class _closing_fileobject(mother_class):  # noqa
    """socket._fileobject that propagates self.close() to the socket.

    Python 2.5 provides this as socket._fileobject(sock, close=True).
    """

# for python 3
try:
    AbstractHTTPHandler
except NameError:
    # somehow this won't get imported by the import * above
    import urllib.request
    AbstractHTTPHandler = urllib.request.AbstractHTTPHandler

class HTTPSHandler(AbstractHTTPHandler):
    def __init__(self, ssl_context=None):
        # type: (SSL.Context) -> None
        AbstractHTTPHandler.__init__(self)

        if ssl_context is not None:
            assert isinstance(ssl_context, SSL.Context), ssl_context
            self.ctx = ssl_context
        else:
            self.ctx = SSL.Context()

    # Copied from urllib2, so we can set the ssl context.
    def https_open(self, req):
        # type: (Request) -> addinfourl
        """Return an addinfourl object for the request, using http_class.

        http_class must implement the HTTPConnection API from httplib.
        The addinfourl return value is a file-like object.  It also
        has methods and attributes including:

            - info(): return a mimetools.Message object for the headers

            - geturl(): return the original request URL

            - code: HTTP status code
        """
        # https://docs.python.org/3.3/library/urllib.request.html#urllib.request.Request.get_host
        try:     # up to python-3.2
            host = req.get_host()
        except AttributeError:  # from python-3.3
            host = req.host
        if not host:
            raise URLError('no host given')

        # Our change: Check to see if we're using a proxy.
        # Then create an appropriate ssl-aware connection.
        full_url = req.get_full_url()
        target_host = url_parse(full_url)[1]

        if target_host != host:
            request_uri = urldefrag(full_url)[0]
            h = httpslib.ProxyHTTPSConnection(host=host, ssl_context=self.ctx)
        else:
            try:     # up to python-3.2
                request_uri = req.get_selector()
            except AttributeError:  # from python-3.3
                request_uri = req.selector
            h = httpslib.HTTPSConnection(host=host, ssl_context=self.ctx)
        # End our change
        h.set_debuglevel(self._debuglevel)

        headers = dict(req.headers)
        headers.update(req.unredirected_hdrs)
        # We want to make an HTTP/1.1 request, but the addinfourl
        # class isn't prepared to deal with a persistent connection.
        # It will try to read all remaining data from the socket,
        # which will block while the server waits for the next request.
        # So make sure the connection gets closed after the (only)
        # request.
        headers["Connection"] = "close"
        try:
            h.request(req.get_method(), request_uri, req.data, headers)
            r = h.getresponse()
        except socket.error as err:  # XXX what error?
            raise URLError(err)

        # Pick apart the HTTPResponse object to get the addinfourl
        # object initialized properly.

        # Wrap the HTTPResponse object in socket's file object adapter
        # for Windows.  That adapter calls recv(), so delegate recv()
        # to read().  This weird wrapping allows the returned object to
        # have readline() and readlines() methods.
        r.recv = r.read
        if six.PY2:
            fp = socket._fileobject(r, close=True)
        else:
            r._decref_socketios = lambda: None
            r.ssl = h.sock.ssl
            r._timeout = -1.0
            r.recv_into = r.readinto
            fp = socket.SocketIO(r, 'rb')

        resp = addinfourl(fp, r.msg, req.get_full_url())
        resp.code = r.status
        resp.msg = r.reason
        return resp

    https_request = AbstractHTTPHandler.do_request_

# Copied from urllib2 with modifications for ssl
def build_opener(ssl_context=None, *handlers):
    # type: (Optional[SSL.Context], *object) -> OpenerDirector
    """Create an opener object from a list of handlers.

    The opener will use several default handlers, including support
    for HTTP and FTP.

    If any of the handlers passed as arguments are subclasses of the
    default handlers, the default handlers will not be used.
    """

    def isclass(obj):
        return isinstance(obj, type) or hasattr(obj, "__bases__")

    opener = OpenerDirector()
    default_classes = [ProxyHandler, UnknownHandler, HTTPHandler,
                       HTTPDefaultErrorHandler, HTTPRedirectHandler,
                       FTPHandler, FileHandler, HTTPErrorProcessor]
    skip = []
    for klass in default_classes:
        for check in handlers:
            if isclass(check):
                if issubclass(check, klass):
                    skip.append(klass)
            elif isinstance(check, klass):
                skip.append(klass)
    for klass in skip:
        default_classes.remove(klass)

        opener.add_handler(klass())

    # Add the HTTPS handler with ssl_context
    if HTTPSHandler not in skip:
        opener.add_handler(HTTPSHandler(ssl_context))

    for h in handlers:
        if isclass(h):
            h = h()
        opener.add_handler(h)
    return opener

"""An extensible library for opening URLs using a variety of protocols

The simplest way to use this module is to call the urlopen function,
which accepts a string containing a URL or a Request object (described
below).  It opens the URL and returns the results as file-like
object; the returned object has some extra methods described below.

The OpenerDirector manages a collection of Handler objects that do
all the actual work.  Each Handler implements a particular protocol or
option.  The OpenerDirector is a composite object that invokes the
Handlers needed to open the requested URL.  For example, the
HTTPHandler performs HTTP GET and POST requests and deals with
non-error returns.  The HTTPRedirectHandler automatically deals with
HTTP 301, 302, 303 and 307 redirect errors, and the HTTPDigestAuthHandler
deals with digest authentication.

urlopen(url, data=None) -- Basic usage is the same as original
urllib.  pass the url and optionally data to post to an HTTP URL, and
get a file-like object back.  One difference is that you can also pass
a Request instance instead of URL.  Raises a URLError (subclass of
OSError); for HTTP errors, raises an HTTPError, which can also be
treated as a valid response.

build_opener -- Function that creates a new OpenerDirector instance.
Will install the default handlers.  Accepts one or more Handlers as
arguments, either instances or Handler classes that it will
instantiate.  If one of the argument is a subclass of the default
handler, the argument will be installed instead of the default.

install_opener -- Installs a new opener as the default opener.

objects of interest:

OpenerDirector -- Sets up the User Agent as the Python-urllib client and manages
the Handler classes, while dealing with requests and responses.

Request -- An object that encapsulates the state of a request.  The
state can be as simple as the URL.  It can also include extra HTTP
headers, e.g. a User-Agent.

BaseHandler --

internals:
BaseHandler and parent
_call_chain conventions

Example usage:

import urllib.request

# set up authentication info
authinfo = urllib.request.HTTPBasicAuthHandler()
authinfo.add_password(realm='PDQ Application',
                      uri='https://mahler:8092/site-updates.py',
                      user='klem',
                      passwd='geheim$parole')

proxy_support = urllib.request.ProxyHandler({"http" : "http://ahad-haam:3128"})

# build a new opener that adds authentication and caching FTP handlers
opener = urllib.request.build_opener(proxy_support, authinfo,
                                     urllib.request.CacheFTPHandler)

# install it
urllib.request.install_opener(opener)

f = urllib.request.urlopen('http://www.python.org/')
"""

# XXX issues:
# If an authentication error handler that tries to perform
# authentication for some reason but fails, how should the error be
# signalled?  The client needs to know the HTTP error code.  But if
# the handler knows that the problem was, e.g., that it didn't know
# that hash algo that requested in the challenge, it would be good to
# pass that information along to the client, too.
# ftp errors aren't handled cleanly
# check digest against correct (i.e. non-apache) implementation

# Possible extensions:
# complex proxies  XXX not sure what exactly was meant by this
# abstract factory for opener

import bisect
import email
import hashlib
import http.client
import io
import os
import posixpath
import re
import string
import sys
import tempfile
import contextlib

from urllib.error import URLError, HTTPError, ContentTooShortError
from urllib.parse import (
    urlparse, urlsplit, urljoin, unwrap, quote, unquote,
    _splittype, _splithost, _splitport, _splituser, _splitpasswd,
    _splitattr, _splitquery, _splitvalue, _splittag, _to_bytes,
    unquote_to_bytes, urlunparse)
from urllib.response import addinfourl, addclosehook

# check for SSL
try:
    import ssl
except ImportError:
    _have_ssl = False
else:
    _have_ssl = True

__all__ = [
    # Classes
    'Request', 'OpenerDirector', 'BaseHandler', 'HTTPDefaultErrorHandler',
    'HTTPRedirectHandler', 'HTTPCookieProcessor', 'ProxyHandler',
    'HTTPPasswordMgr', 'HTTPPasswordMgrWithDefaultRealm',
    'HTTPPasswordMgrWithPriorAuth', 'AbstractBasicAuthHandler',
    'HTTPBasicAuthHandler', 'ProxyBasicAuthHandler', 'AbstractDigestAuthHandler',
    'HTTPDigestAuthHandler', 'ProxyDigestAuthHandler', 'HTTPHandler',
    'FileHandler', 'FTPHandler', 'CacheFTPHandler', 'DataHandler',
    'UnknownHandler', 'HTTPErrorProcessor',
    # Functions
    'urlopen', 'install_opener', 'build_opener',
    'pathname2url', 'url2pathname', 'getproxies',
    # Legacy interface
    'urlretrieve', 'urlcleanup', 'URLopener', 'FancyURLopener',
]

# used in User-Agent header sent
__version__ = '%d.%d' % sys.version_info[:2]

_opener = None
def urlopen(url, data=None, timeout=socket._GLOBAL_DEFAULT_TIMEOUT,
            *, cafile=None, capath=None, cadefault=False, context=None):
    '''Open the URL url, which can be either a string or a Request object.

    *data* must be an object specifying additional data to be sent to
    the server, or None if no such data is needed.  See Request for
    details.

    urllib.request module uses HTTP/1.1 and includes a "Connection:close"
    header in its HTTP requests.

    The optional *timeout* parameter specifies a timeout in seconds for
    blocking operations like the connection attempt (if not specified, the
    global default timeout setting will be used). This only works for HTTP,
    HTTPS and FTP connections.

    If *context* is specified, it must be a ssl.SSLContext instance describing
    the various SSL options. See HTTPSConnection for more details.

    The optional *cafile* and *capath* parameters specify a set of trusted CA
    certificates for HTTPS requests. cafile should point to a single file
    containing a bundle of CA certificates, whereas capath should point to a
    directory of hashed certificate files. More information can be found in
    ssl.SSLContext.load_verify_locations().

    The *cadefault* parameter is ignored.

    This function always returns an object which can work as a context
    manager and has methods such as

    * geturl() - return the URL of the resource retrieved, commonly used to
      determine if a redirect was followed

    * info() - return the meta-information of the page, such as headers, in the
      form of an email.message_from_string() instance (see Quick Reference to
      HTTP Headers)

    * getcode() - return the HTTP status code of the response.  Raises URLError
      on errors.

    For HTTP and HTTPS URLs, this function returns a http.client.HTTPResponse
    object slightly modified. In addition to the three new methods above, the
    msg attribute contains the same information as the reason attribute ---
    the reason phrase returned by the server --- instead of the response
    headers as it is specified in the documentation for HTTPResponse.

    For FTP, file, and data URLs and requests explicitly handled by legacy
    URLopener and FancyURLopener classes, this function returns a
    urllib.response.addinfourl object.

    Note that None may be returned if no handler handles the request (though
    the default installed global OpenerDirector uses UnknownHandler to ensure
    this never happens).

    In addition, if proxy settings are detected (for example, when a *_proxy
    environment variable like http_proxy is set), ProxyHandler is default
    installed and makes sure the requests are handled through the proxy.

    '''
    global _opener
    if cafile or capath or cadefault:
        import warnings
        warnings.warn("cafile, capath and cadefault are deprecated, use a "
                      "custom context instead.", DeprecationWarning, 2)
        if context is not None:
                "You can't pass both context and any of cafile, capath, and "
                "cadefault"
            )
        if not _have_ssl:
            raise ValueError('SSL support not available')
        context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH,
                                             cafile=cafile,
                                             capath=capath)
        https_handler = HTTPSHandler(context=context)
        opener = build_opener(https_handler)
    elif context:
    elif _opener is None:
        _opener = opener = build_opener()
    else:
        opener = _opener
    return opener.open(url, data, timeout)

def install_opener(opener):
    _opener = opener

_url_tempfiles = []
def urlretrieve(url, filename=None, reporthook=None, data=None):
    """
    Retrieve a URL into a temporary location on disk.

    Requires a URL argument. If a filename is passed, it is used as
    the temporary file location. The reporthook argument should be
    a callable that accepts a block number, a read size, and the
    total file size of the URL target. The data argument should be
    valid URL encoded data.

    If a filename is passed and the URL points to a local resource,
    the result is a copy from local file to new file.

    Returns a tuple containing the path to the newly created
    data file as well as the resulting HTTPMessage object.
    """
    url_type, path = _splittype(url)

    with contextlib.closing(urlopen(url, data)) as fp:
        headers = fp.info()

        # Just return the local path and the "headers" for file://
        # URLs. No sense in performing a copy unless requested.
        if url_type == "file" and not filename:
            return os.path.normpath(path), headers

        # Handle temporary file setup.
        if filename:
            tfp = open(filename, 'wb')
        else:
            tfp = tempfile.NamedTemporaryFile(delete=False)
            filename = tfp.name
            _url_tempfiles.append(filename)

        with tfp:
            result = filename, headers
            bs = 1024*8
            size = -1
            read = 0
            blocknum = 0
            if "content-length" in headers:
                size = int(headers["Content-Length"])

            if reporthook:
                reporthook(blocknum, bs, size)

            while True:
                block = fp.read(bs)
                if not block:
                    break
                read += len(block)
                tfp.write(block)
                blocknum += 1
                if reporthook:
                    reporthook(blocknum, bs, size)

    if size >= 0 and read < size:
        raise ContentTooShortError(
            "retrieval incomplete: got only %i out of %i bytes"
            % (read, size), result)

    return result

def urlcleanup():
    """Clean up temporary files from urlretrieve calls."""
    for temp_file in _url_tempfiles:
        try:
            os.unlink(temp_file)
        except OSError:
            pass

    del _url_tempfiles[:]
    if _opener:
        _opener = None

# copied from cookielib.py
_cut_port_re = re.compile(r":\d+$", re.ASCII)
def request_host(request):
    """Return request-host, as defined by RFC 2965.

    Variation from RFC: returned value is lowercased, for convenient
    comparison.

    """
    url = request.full_url
    host = urlparse(url)[1]
    if host == "":
        host = request.get_header("Host", "")

    # remove port, if present
    host = _cut_port_re.sub("", host, 1)
    return host.lower()

class Request:

    def __init__(self, url, data=None, headers={},
                 origin_req_host=None, unverifiable=False,
                 method=None):
        self.full_url = url
        self.headers = {}
        self.unredirected_hdrs = {}
        self._data = None
        self.data = data
        self._tunnel_host = None
        for key, value in headers.items():
            self.add_header(key, value)
        if origin_req_host is None:
            origin_req_host = request_host(self)
        self.origin_req_host = origin_req_host
        self.unverifiable = unverifiable
        if method:
            self.method = method

    @property
    def full_url(self):
        if self.fragment:
            return '{}#{}'.format(self._full_url, self.fragment)
        return self._full_url

    @full_url.setter
    def full_url(self, url):
        # unwrap('<URL:type://host/path>') --> 'type://host/path'
        self._full_url = unwrap(url)
        self._full_url, self.fragment = _splittag(self._full_url)
        self._parse()

    @full_url.deleter
        self._full_url = None
        self.fragment = None
        self.selector = ''

    @property

    @data.setter
    def data(self, data):
        if data != self._data:
            self._data = data
            # issue 16464
            # if we change data we need to remove content-length header
            # (cause it's most probably calculated for previous value)
            if self.has_header("Content-length"):
                self.remove_header("Content-length")

    @data.deleter
        self.data = None

    def _parse(self):
        self.type, rest = _splittype(self._full_url)
        if self.type is None:
            raise ValueError("unknown url type: %r" % self.full_url)
        self.host, self.selector = _splithost(rest)
        if self.host:
            self.host = unquote(self.host)

    def get_method(self):
        """Return a string indicating the HTTP request method."""
        default_method = "POST" if self.data is not None else "GET"
        return getattr(self, 'method', default_method)

    def get_full_url(self):
        return self.full_url

    def set_proxy(self, host, type):
        if self.type == 'https' and not self._tunnel_host:
            self._tunnel_host = self.host
        else:
            self.type= type
            self.selector = self.full_url

    def has_proxy(self):
        return self.selector == self.full_url

    def add_header(self, key, val):
        # useful for something like authentication
        self.headers[key.capitalize()] = val

    def add_unredirected_header(self, key, val):
        # will not be added to a redirected request
        self.unredirected_hdrs[key.capitalize()] = val

    def has_header(self, header_name):
        return (header_name in self.headers or
                header_name in self.unredirected_hdrs)

    def get_header(self, header_name, default=None):
        return self.headers.get(
            header_name,
            self.unredirected_hdrs.get(header_name, default))

    def remove_header(self, header_name):
        self.headers.pop(header_name, None)
        self.unredirected_hdrs.pop(header_name, None)

    def header_items(self):
        hdrs = {**self.unredirected_hdrs, **self.headers}
        return list(hdrs.items())

class OpenerDirector:
        client_version = "Python-urllib/%s" % __version__
        self.addheaders = [('User-agent', client_version)]
        # self.handlers is retained only for backward compatibility
        self.handlers = []
        # manage the individual handlers
        self.handle_open = {}
        self.handle_error = {}
        self.process_response = {}
        self.process_request = {}

    def add_handler(self, handler):
        if not hasattr(handler, "add_parent"):
            raise TypeError("expected BaseHandler instance, got %r" %
                            type(handler))

        added = False
        for meth in dir(handler):
            if meth in ["redirect_request", "do_open", "proxy_open"]:
                # oops, coincidental match
                continue

            i = meth.find("_")
            protocol = meth[:i]
            condition = meth[i+1:]

            if condition.startswith("error"):
                j = condition.find("_") + i + 1
                kind = meth[j+1:]
                try:
                    kind = int(kind)
                except ValueError:
                    pass
                lookup = self.handle_error.get(protocol, {})
                self.handle_error[protocol] = lookup
            elif condition == "open":
                kind = protocol
                lookup = self.handle_open
            elif condition == "response":
                lookup = self.process_response
            elif condition == "request":
                lookup = self.process_request
            else:
                continue

            handlers = lookup.setdefault(kind, [])
            if handlers:
                bisect.insort(handlers, handler)
            else:
                handlers.append(handler)
            added = True

        if added:
            bisect.insort(self.handlers, handler)
            handler.add_parent(self)

        # Only exists for backwards compatibility.
        pass

    def _call_chain(self, chain, kind, meth_name, *args):
        # Handlers raise an exception if no one else should try to handle
        # the request, or return None if they can't but another handler
        # could.  Otherwise, they return the response.
        handlers = chain.get(kind, ())
        for handler in handlers:
            func = getattr(handler, meth_name)
            result = func(*args)
            if result is not None:
                return result

    def open(self, fullurl, data=None, timeout=socket._GLOBAL_DEFAULT_TIMEOUT):
        # accept a URL or a Request object
        if isinstance(fullurl, str):
            req = Request(fullurl, data)
        else:
            req = fullurl
            if data is not None:
                req.data = data

        req.timeout = timeout
        protocol = req.type

        # pre-process request
        meth_name = protocol+"_request"
        for processor in self.process_request.get(protocol, []):
            meth = getattr(processor, meth_name)
            req = meth(req)

        sys.audit('urllib.Request', req.full_url, req.data, req.headers, req.get_method())
        response = self._open(req, data)

        # post-process response
        meth_name = protocol+"_response"
        for processor in self.process_response.get(protocol, []):
            response = meth(req, response)

        return response

    def _open(self, req, data=None):
        result = self._call_chain(self.handle_open, 'default',
                                  'default_open', req)
        if result:
            return result

        result = self._call_chain(self.handle_open, protocol, protocol +
                                  '_open', req)
        if result:

        return self._call_chain(self.handle_open, 'unknown',
                                'unknown_open', req)

    def error(self, proto, *args):
        if proto in ('http', 'https'):
            # XXX http[s] protocols are special-cased
            dict = self.handle_error['http'] # https is not different than http
            proto = args[2]  # YUCK!
            meth_name = 'http_error_%s' % proto
            http_err = 1
            orig_args = args
        else:
            dict = self.handle_error
            meth_name = proto + '_error'
            http_err = 0
        args = (dict, proto, meth_name) + args
        result = self._call_chain(*args)
        if result:

        if http_err:
            args = (dict, 'default', 'http_error_default') + orig_args
            return self._call_chain(*args)

# XXX probably also want an abstract factory that knows when it makes
# sense to skip a superclass in favor of a subclass and when it might
# make sense to include both

def build_opener(*handlers):

    for HTTP, FTP and when applicable HTTPS.

    """
                       FTPHandler, FileHandler, HTTPErrorProcessor,
                       DataHandler]
    if hasattr(http.client, "HTTPSConnection"):
        default_classes.append(HTTPSHandler)
    skip = set()
            if isinstance(check, type):
                    skip.add(klass)
                skip.add(klass)

        if isinstance(h, type):
            h = h()

class BaseHandler:
    handler_order = 500

    def add_parent(self, parent):
        self.parent = parent

        # Only exists for backwards compatibility
        pass

    def __lt__(self, other):
        if not hasattr(other, "handler_order"):
            # Try to preserve the old behavior of having custom classes
            # inserted after default ones (works only for custom user
            # classes which are not aware of handler_order).
            return True
        return self.handler_order < other.handler_order

class HTTPErrorProcessor(BaseHandler):
    """Process HTTP error responses."""
    handler_order = 1000  # after all other processing

    def http_response(self, request, response):
        code, msg, hdrs = response.code, response.msg, response.info()

        # According to RFC 2616, "2xx" code indicates that the client's
        # request was successfully received, understood, and accepted.
        if not (200 <= code < 300):
            response = self.parent.error(
                'http', request, response, code, msg, hdrs)

    https_response = http_response

class HTTPDefaultErrorHandler(BaseHandler):
    def http_error_default(self, req, fp, code, msg, hdrs):
        raise HTTPError(req.full_url, code, msg, hdrs, fp)

class HTTPRedirectHandler(BaseHandler):
    # maximum number of redirections to any single URL
    # this is needed because of the state that cookies introduce
    max_repeats = 4
    # maximum total number of redirections (regardless of URL) before
    # assuming we're in a loop
    max_redirections = 10

    def redirect_request(self, req, fp, code, msg, headers, newurl):
        """Return a Request or None in response to a redirect.

        This is called by the http_error_30x methods when a
        redirection response is received.  If a redirection should
        take place, return a new Request to allow http_error_30x to
        perform the redirect.  Otherwise, raise HTTPError if no-one
        else should try to handle this url.  Return None if you can't
        but another Handler might.
        """
        m = req.get_method()
        if (not (code in (301, 302, 303, 307) and m in ("GET", "HEAD")
            or code in (301, 302, 303) and m == "POST")):
            raise HTTPError(req.full_url, code, msg, headers, fp)

        # Strictly (according to RFC 2616), 301 or 302 in response to
        # a POST MUST NOT cause a redirection without confirmation
        # from the user (of urllib.request, in this case).  In practice,
        # essentially all clients do redirect in this case, so we do
        # the same.

        # Be conciliant with URIs containing a space.  This is mainly
        # redundant with the more complete encoding done in http_error_302(),
        # but it is kept for compatibility with other callers.
        newurl = newurl.replace(' ', '%20')

        CONTENT_HEADERS = ("content-length", "content-type")
        newheaders = {k: v for k, v in req.headers.items()
                      if k.lower() not in CONTENT_HEADERS}
        return Request(newurl,
                       headers=newheaders,
                       origin_req_host=req.origin_req_host,
                       unverifiable=True)

    # Implementation note: To avoid the server sending us into an
    # infinite loop, the request object needs to track what URLs we
    # have already seen.  Do this by adding a handler-specific
    # attribute to the Request object.
    def http_error_302(self, req, fp, code, msg, headers):
        # Some servers (incorrectly) return multiple Location headers
        # (so probably same goes for URI).  Use first header.
        if "location" in headers:
            newurl = headers["location"]
        elif "uri" in headers:
            newurl = headers["uri"]
        else:
            return

        # fix a possible malformed URL
        urlparts = urlparse(newurl)

        # For security reasons we don't allow redirection to anything other
        # than http, https or ftp.

        if urlparts.scheme not in ('http', 'https', 'ftp', ''):
            raise HTTPError(
                newurl, code,
                "%s - Redirection to url '%s' is not allowed" % (msg, newurl),
                headers, fp)

        if not urlparts.path and urlparts.netloc:
            urlparts = list(urlparts)
            urlparts[2] = "/"
        newurl = urlunparse(urlparts)

        # http.client.parse_headers() decodes as ISO-8859-1.  Recover the
        # original bytes and percent-encode non-ASCII bytes, and any special
        # characters such as the space.
        newurl = quote(
            newurl, encoding="iso-8859-1", safe=string.punctuation)
        newurl = urljoin(req.full_url, newurl)

        # XXX Probably want to forget about the state of the current
        # request, although that might interact poorly with other
        # handlers that also use handler-specific request attributes
        new = self.redirect_request(req, fp, code, msg, headers, newurl)
        if new is None:
            return

        # loop detection
        # .redirect_dict has a key url if url was previously visited.
        if hasattr(req, 'redirect_dict'):
            visited = new.redirect_dict = req.redirect_dict
            if (visited.get(newurl, 0) >= self.max_repeats or
                len(visited) >= self.max_redirections):
                raise HTTPError(req.full_url, code,
                                self.inf_msg + msg, headers, fp)
        else:
            visited = new.redirect_dict = req.redirect_dict = {}
        visited[newurl] = visited.get(newurl, 0) + 1

        # Don't close the fp until we are sure that we won't use it
        # with HTTPError.
        fp.read()
        fp.close()

        return self.parent.open(new, timeout=req.timeout)

    http_error_301 = http_error_303 = http_error_307 = http_error_302

    inf_msg = "The HTTP server returned a redirect error that would " \
              "lead to an infinite loop.\n" \
              "The last 30x error message was:\n"

def _parse_proxy(proxy):
    """Return (scheme, user, password, host/port) given a URL or an authority.

    If a URL is supplied, it must have an authority (host:port) component.
    According to RFC 3986, having an authority component means the URL must
    have two slashes after the scheme.
    """
    scheme, r_scheme = _splittype(proxy)
    if not r_scheme.startswith("/"):
        # authority
        scheme = None
        authority = proxy
    else:
        # URL
        if not r_scheme.startswith("//"):
            raise ValueError("proxy URL with no authority: %r" % proxy)
        # We have an authority, so for RFC 3986-compliant URLs (by ss 3.
        # and 3.3.), path is empty or starts with '/'
        end = r_scheme.find("/", 2)
        if end == -1:
            end = None
        authority = r_scheme[2:end]
    userinfo, hostport = _splituser(authority)
    if userinfo is not None:
        user, password = _splitpasswd(userinfo)
    else:
        user = password = None
    return scheme, user, password, hostport

class ProxyHandler(BaseHandler):
    # Proxies must be in front
    handler_order = 100

    def __init__(self, proxies=None):
        if proxies is None:
            proxies = getproxies()
        assert hasattr(proxies, 'keys'), "proxies must be a mapping"
        self.proxies = proxies
        for type, url in proxies.items():
            type = type.lower()
            setattr(self, '%s_open' % type,
                    lambda r, proxy=url, type=type, meth=self.proxy_open:
                        meth(r, proxy, type))

    def proxy_open(self, req, proxy, type):
        orig_type = req.type
        proxy_type, user, password, hostport = _parse_proxy(proxy)
        if proxy_type is None:
            proxy_type = orig_type

        if req.host and proxy_bypass(req.host):

        if user and password:
            user_pass = '%s:%s' % (unquote(user),
                                   unquote(password))
            creds = base64.b64encode(user_pass.encode()).decode("ascii")
            req.add_header('Proxy-authorization', 'Basic ' + creds)
        hostport = unquote(hostport)
        req.set_proxy(hostport, proxy_type)
        if orig_type == proxy_type or orig_type == 'https':
            # let other handlers take care of it
        else:
            # need to start over, because the other handlers don't
            # grok the proxy's URL type
            # e.g. if we have a constructor arg proxies like so:
            # {'http': 'ftp://proxy.example.com'}, we may end up turning
            # a request for http://acme.example.com/a into one for
            # ftp://proxy.example.com/a
            return self.parent.open(req, timeout=req.timeout)

class HTTPPasswordMgr:

        self.passwd = {}

    def add_password(self, realm, uri, user, passwd):
        # uri could be a single URI or a sequence
        if isinstance(uri, str):
            uri = [uri]
        if realm not in self.passwd:
            self.passwd[realm] = {}
        for default_port in True, False:
            reduced_uri = tuple(
                self.reduce_uri(u, default_port) for u in uri)
            self.passwd[realm][reduced_uri] = (user, passwd)

    def find_user_password(self, realm, authuri):
        domains = self.passwd.get(realm, {})
            reduced_authuri = self.reduce_uri(authuri, default_port)
            for uris, authinfo in domains.items():
                for uri in uris:
                    if self.is_suburi(uri, reduced_authuri):
                        return authinfo
        return None, None

    def reduce_uri(self, uri, default_port=True):
        """Accept authority or URI and extract only the authority and path."""
        # note HTTP URLs do not have a userinfo component
        parts = urlsplit(uri)
        if parts[1]:
            # URI
            scheme = parts[0]
            authority = parts[1]
            path = parts[2] or '/'
        else:
            # host or host:port
            scheme = None
            authority = uri
            path = '/'
        host, port = _splitport(authority)
        if default_port and port is None and scheme is not None:
            dport = {"http": 80,
                     "https": 443,
                     }.get(scheme)
            if dport is not None:
                authority = "%s:%d" % (host, dport)
        return authority, path

    def is_suburi(self, base, test):
        """Check if test is below base in a URI tree

        Both args must be URIs in reduced form.
        """
        if base == test:
        if base[0] != test[0]:
            return False
        common = posixpath.commonprefix((base[1], test[1]))
        if len(common) == len(base[1]):
        return False

class HTTPPasswordMgrWithDefaultRealm(HTTPPasswordMgr):

        user, password = HTTPPasswordMgr.find_user_password(self, realm,
                                                            authuri)
        if user is not None:
            return user, password
        return HTTPPasswordMgr.find_user_password(self, None, authuri)

class HTTPPasswordMgrWithPriorAuth(HTTPPasswordMgrWithDefaultRealm):

    def __init__(self, *args, **kwargs):
        self.authenticated = {}
        super().__init__(*args, **kwargs)

    def add_password(self, realm, uri, user, passwd, is_authenticated=False):
        self.update_authenticated(uri, is_authenticated)
        # Add a default for prior auth requests
        if realm is not None:
            super().add_password(None, uri, user, passwd)
        super().add_password(realm, uri, user, passwd)

    def update_authenticated(self, uri, is_authenticated=False):
            uri = [uri]

            for u in uri:
                reduced_uri = self.reduce_uri(u, default_port)
                self.authenticated[reduced_uri] = is_authenticated

    def is_authenticated(self, authuri):
            for uri in self.authenticated:
                if self.is_suburi(uri, reduced_authuri):
                    return self.authenticated[uri]

class AbstractBasicAuthHandler:

    # XXX this allows for multiple auth-schemes, but will stupidly pick
    # the last one with a realm specified.

    # allow for double- and single-quoted realm values
    # (single quotes are a violation of the RFC, but appear in the wild)
    rx = re.compile('(?:^|,)'   # start of the string or ','
                    '[ \t]*'    # optional whitespaces
                    '([^ \t]+)' # scheme like "Basic"
                    '[ \t]+'    # mandatory whitespaces
                    # realm=xxx
                    # realm='xxx'
                    # realm="xxx"
                    'realm=(["\']?)([^"\']*)\\2',
                    re.I)

    # XXX could pre-emptively send auth info already accepted (RFC 2617,
    # end of section 2, and section 1.2 immediately after "credentials"
    # production).

    def __init__(self, password_mgr=None):
        if password_mgr is None:
            password_mgr = HTTPPasswordMgr()
        self.passwd = password_mgr
        self.add_password = self.passwd.add_password

    def _parse_realm(self, header):
        # parse WWW-Authenticate header: accept multiple challenges per header
        found_challenge = False
        for mo in AbstractBasicAuthHandler.rx.finditer(header):
            scheme, quote, realm = mo.groups()
            if quote not in ['"', "'"]:
                warnings.warn("Basic Auth Realm was unquoted",
                              UserWarning, 3)

            yield (scheme, realm)

            found_challenge = True

        if not found_challenge:
            if header:
                scheme = header.split()[0]
            else:
                scheme = ''
            yield (scheme, None)

    def http_error_auth_reqed(self, authreq, host, req, headers):
        # host may be an authority (without userinfo) or a URL with an
        headers = headers.get_all(authreq)
        if not headers:
            # no header found
            return

        unsupported = None
        for header in headers:
            for scheme, realm in self._parse_realm(header):
                if scheme.lower() != 'basic':
                    unsupported = scheme
                    continue

                if realm is not None:
                    # Use the first matching Basic challenge.
                    # Ignore following challenges even if they use the Basic
                    # scheme.
                    return self.retry_http_basic_auth(host, req, realm)

        if unsupported is not None:
            raise ValueError("AbstractBasicAuthHandler does not "
                             "support the following scheme: %r"
                             % (scheme,))

    def retry_http_basic_auth(self, host, req, realm):
        user, pw = self.passwd.find_user_password(realm, host)
        if pw is not None:
            raw = "%s:%s" % (user, pw)
            auth = "Basic " + base64.b64encode(raw.encode()).decode("ascii")
            if req.get_header(self.auth_header, None) == auth:
                return None
            req.add_unredirected_header(self.auth_header, auth)
        else:

    def http_request(self, req):
        if (not hasattr(self.passwd, 'is_authenticated') or
           not self.passwd.is_authenticated(req.full_url)):
            return req

        if not req.has_header('Authorization'):
            user, passwd = self.passwd.find_user_password(None, req.full_url)
            credentials = '{0}:{1}'.format(user, passwd).encode()
            auth_str = base64.standard_b64encode(credentials).decode()
            req.add_unredirected_header('Authorization',
                                        'Basic {}'.format(auth_str.strip()))
        return req

    def http_response(self, req, response):
        if hasattr(self.passwd, 'is_authenticated'):
            if 200 <= response.code < 300:
                self.passwd.update_authenticated(req.full_url, True)
            else:
                self.passwd.update_authenticated(req.full_url, False)

    https_request = http_request

class HTTPBasicAuthHandler(AbstractBasicAuthHandler, BaseHandler):

    auth_header = 'Authorization'

    def http_error_401(self, req, fp, code, msg, headers):
        url = req.full_url
        response = self.http_error_auth_reqed('www-authenticate',
                                          url, req, headers)

class ProxyBasicAuthHandler(AbstractBasicAuthHandler, BaseHandler):

    auth_header = 'Proxy-authorization'

    def http_error_407(self, req, fp, code, msg, headers):
        # http_error_auth_reqed requires that there is no userinfo component in
        # authority.  Assume there isn't one, since urllib.request does not (and
        # should not, RFC 3986 s. 3.2.1) support requests for URLs containing
        # userinfo.
        authority = req.host
        response = self.http_error_auth_reqed('proxy-authenticate',
                                          authority, req, headers)

# Return n random bytes.
_randombytes = os.urandom

class AbstractDigestAuthHandler:
    # Digest authentication is specified in RFC 2617.

    # XXX The client does not inspect the Authentication-Info header
    # in a successful response.

    # XXX It should be possible to test this implementation against
    # a mock server that just generates a static set of challenges.

    # XXX qop="auth-int" supports is shaky

    def __init__(self, passwd=None):
        if passwd is None:
            passwd = HTTPPasswordMgr()
        self.passwd = passwd
        self.retried = 0
        self.nonce_count = 0
        self.last_nonce = None

    def reset_retry_count(self):

    def http_error_auth_reqed(self, auth_header, host, req, headers):
        authreq = headers.get(auth_header, None)
        if self.retried > 5:
            # Don't fail endlessly - if we failed once, we'll probably
            # fail a second time. Hm. Unless the Password Manager is
            # prompting for the information. Crap. This isn't great
            # but it's better than the current 'repeat until recursion
            # depth exceeded' approach <wink>
            raise HTTPError(req.full_url, 401, "digest auth failed",
                            headers, None)
        else:
            self.retried += 1
        if authreq:
            scheme = authreq.split()[0]
            if scheme.lower() == 'digest':
                return self.retry_http_digest_auth(req, authreq)
            elif scheme.lower() != 'basic':
                raise ValueError("AbstractDigestAuthHandler does not support"
                                 " the following scheme: '%s'" % scheme)

    def retry_http_digest_auth(self, req, auth):
        token, challenge = auth.split(' ', 1)
        chal = parse_keqv_list(filter(None, parse_http_list(challenge)))
        auth = self.get_authorization(req, chal)
        if auth:
            auth_val = 'Digest %s' % auth
            if req.headers.get(self.auth_header, None) == auth_val:
            req.add_unredirected_header(self.auth_header, auth_val)
            resp = self.parent.open(req, timeout=req.timeout)
            return resp

    def get_cnonce(self, nonce):
        # The cnonce-value is an opaque
        # quoted string value provided by the client and used by both client
        # and server to avoid chosen plaintext attacks, to provide mutual
        # authentication, and to provide some message integrity protection.
        # This isn't a fabulous effort, but it's probably Good Enough.
        s = "%s:%s:%s:" % (self.nonce_count, nonce, time.ctime())
        b = s.encode("ascii") + _randombytes(8)
        dig = hashlib.sha1(b).hexdigest()
        return dig[:16]

    def get_authorization(self, req, chal):
        try:
            realm = chal['realm']
            nonce = chal['nonce']
            qop = chal.get('qop')
            algorithm = chal.get('algorithm', 'MD5')
            # mod_digest doesn't send an opaque, even though it isn't
            # supposed to be optional
            opaque = chal.get('opaque', None)

        H, KD = self.get_algorithm_impls(algorithm)
        if H is None:

        user, pw = self.passwd.find_user_password(realm, req.full_url)
        if user is None:

        # XXX not implemented yet
        if req.data is not None:
            entdig = self.get_entity_digest(req.data, chal)
        else:
            entdig = None

        A1 = "%s:%s:%s" % (user, realm, pw)
        A2 = "%s:%s" % (req.get_method(),
                        # XXX selector: what about proxies and full urls
                        req.selector)
        # NOTE: As per  RFC 2617, when server sends "auth,auth-int", the client could use either `auth`
        #     or `auth-int` to the response back. we use `auth` to send the response back.
        if qop is None:
            respdig = KD(H(A1), "%s:%s" % (nonce, H(A2)))
        elif 'auth' in qop.split(','):
            if nonce == self.last_nonce:
                self.nonce_count += 1
            else:
                self.nonce_count = 1
                self.last_nonce = nonce
            ncvalue = '%08x' % self.nonce_count
            cnonce = self.get_cnonce(nonce)
            noncebit = "%s:%s:%s:%s:%s" % (nonce, ncvalue, cnonce, 'auth', H(A2))
            respdig = KD(H(A1), noncebit)
        else:
            # XXX handle auth-int.
            raise URLError("qop '%s' is not supported." % qop)

        # XXX should the partial digests be encoded too?

        base = 'username="%s", realm="%s", nonce="%s", uri="%s", ' \
               'response="%s"' % (user, realm, nonce, req.selector,
                                  respdig)
        if opaque:
            base += ', opaque="%s"' % opaque
        if entdig:
            base += ', digest="%s"' % entdig
        base += ', algorithm="%s"' % algorithm
        if qop:
            base += ', qop=auth, nc=%s, cnonce="%s"' % (ncvalue, cnonce)
        return base

    def get_algorithm_impls(self, algorithm):
        # lambdas assume digest modules are imported at the top level
        if algorithm == 'MD5':
            H = lambda x: hashlib.md5(x.encode("ascii")).hexdigest()
        elif algorithm == 'SHA':
            H = lambda x: hashlib.sha1(x.encode("ascii")).hexdigest()
        # XXX MD5-sess
        else:
            raise ValueError("Unsupported digest authentication "
                             "algorithm %r" % algorithm)
        KD = lambda s, d: H("%s:%s" % (s, d))
        return H, KD

    def get_entity_digest(self, data, chal):

class HTTPDigestAuthHandler(BaseHandler, AbstractDigestAuthHandler):
    """An authentication protocol defined by RFC 2069

    Digest authentication improves on basic authentication because it
    does not transmit passwords in the clear.
    """

    handler_order = 490  # before Basic auth

        host = urlparse(req.full_url)[1]
        retry = self.http_error_auth_reqed('www-authenticate',
                                           host, req, headers)
        self.reset_retry_count()
        return retry

class ProxyDigestAuthHandler(BaseHandler, AbstractDigestAuthHandler):

    auth_header = 'Proxy-Authorization'

        host = req.host
        retry = self.http_error_auth_reqed('proxy-authenticate',

class AbstractHTTPHandler(BaseHandler):

    def __init__(self, debuglevel=0):
        self._debuglevel = debuglevel

    def set_http_debuglevel(self, level):
        self._debuglevel = level

    def _get_content_length(self, request):
        return http.client.HTTPConnection._get_content_length(
            request.data,
            request.get_method())

    def do_request_(self, request):
        host = request.host

        if request.data is not None:  # POST
            data = request.data
            if isinstance(data, str):
                msg = "POST data should be bytes, an iterable of bytes, " \
                      "or a file object. It cannot be of type str."
                raise TypeError(msg)
            if not request.has_header('Content-type'):
                request.add_unredirected_header(
                    'Content-type',
                    'application/x-www-form-urlencoded')
            if (not request.has_header('Content-length')
                    and not request.has_header('Transfer-encoding')):
                content_length = self._get_content_length(request)
                if content_length is not None:
                    request.add_unredirected_header(
                            'Content-length', str(content_length))
                else:
                            'Transfer-encoding', 'chunked')

        sel_host = host
        if request.has_proxy():
            scheme, sel = _splittype(request.selector)
            sel_host, sel_path = _splithost(sel)
        if not request.has_header('Host'):
            request.add_unredirected_header('Host', sel_host)
        for name, value in self.parent.addheaders:
            name = name.capitalize()
            if not request.has_header(name):
                request.add_unredirected_header(name, value)

        return request

    def do_open(self, http_class, req, **http_conn_args):
        """Return an HTTPResponse object for the request, using http_class.

        http_class must implement the HTTPConnection API from http.client.
        """

        # will parse host:port
        h = http_class(host, timeout=req.timeout, **http_conn_args)

        headers = dict(req.unredirected_hdrs)
        headers.update({k: v for k, v in req.headers.items()
                        if k not in headers})

        # TODO(jhylton): Should this be redesigned to handle
        # persistent connections?

        # request.
        headers = {name.title(): val for name, val in headers.items()}

        if req._tunnel_host:
            tunnel_headers = {}
            proxy_auth_hdr = "Proxy-Authorization"
            if proxy_auth_hdr in headers:
                tunnel_headers[proxy_auth_hdr] = headers[proxy_auth_hdr]
                # Proxy-Authorization should not be sent to origin
                # server.
                del headers[proxy_auth_hdr]
            h.set_tunnel(req._tunnel_host, headers=tunnel_headers)

        try:
            try:
                h.request(req.get_method(), req.selector, req.data, headers,
                          encode_chunked=req.has_header('Transfer-encoding'))
            except OSError as err: # timeout error
                raise URLError(err)
        except:
            h.close()
            raise

        # If the server does not send us a 'Connection: close' header,
        # HTTPConnection assumes the socket should be left open. Manually
        # mark the socket to be closed when this response object goes away.
        if h.sock:
            h.sock.close()
            h.sock = None

        r.url = req.get_full_url()
        # This line replaces the .msg attribute of the HTTPResponse
        # with .headers, because urllib clients expect the response to
        # have the reason in .msg.  It would be good to mark this
        # attribute is deprecated and get then to use info() or
        # .headers.
        r.msg = r.reason
        return r

class HTTPHandler(AbstractHTTPHandler):

    def http_open(self, req):
        return self.do_open(http.client.HTTPConnection, req)

    http_request = AbstractHTTPHandler.do_request_

if hasattr(http.client, 'HTTPSConnection'):

    class HTTPSHandler(AbstractHTTPHandler):

        def __init__(self, debuglevel=0, context=None, check_hostname=None):
            AbstractHTTPHandler.__init__(self, debuglevel)
            self._context = context
            self._check_hostname = check_hostname

        def https_open(self, req):
            return self.do_open(http.client.HTTPSConnection, req,
                context=self._context, check_hostname=self._check_hostname)

        https_request = AbstractHTTPHandler.do_request_

    __all__.append('HTTPSHandler')

class HTTPCookieProcessor(BaseHandler):
    def __init__(self, cookiejar=None):
        import http.cookiejar
        if cookiejar is None:
            cookiejar = http.cookiejar.CookieJar()
        self.cookiejar = cookiejar

    def http_request(self, request):
        self.cookiejar.add_cookie_header(request)

        self.cookiejar.extract_cookies(response, request)

class UnknownHandler(BaseHandler):
    def unknown_open(self, req):
        type = req.type
        raise URLError('unknown url type: %s' % type)

def parse_keqv_list(l):
    """Parse list of key=value strings where keys are not duplicated."""
    parsed = {}
    for elt in l:
        k, v = elt.split('=', 1)
        if v[0] == '"' and v[-1] == '"':
            v = v[1:-1]
        parsed[k] = v
    return parsed

def parse_http_list(s):
    """Parse lists as described by RFC 2068 Section 2.

    In particular, parse comma-separated lists where the elements of
    the list may include quoted-strings.  A quoted-string could
    contain a comma.  A non-quoted string could have quotes in the
    middle.  Neither commas nor quotes count if they are escaped.
    Only double-quotes count, not single-quotes.
    """
    res = []
    part = ''

    escape = quote = False
    for cur in s:
        if escape:
            part += cur
            escape = False
            continue
        if quote:
            if cur == '\\':
                escape = True
                continue
            elif cur == '"':
                quote = False
            part += cur
            continue

        if cur == ',':
            res.append(part)
            part = ''
            continue

        if cur == '"':
            quote = True

        part += cur

    # append last part
    if part:
        res.append(part)

    return [part.strip() for part in res]

class FileHandler(BaseHandler):
    # Use local file or FTP depending on form of URL
    def file_open(self, req):
        url = req.selector
        if url[:2] == '//' and url[2:3] != '/' and (req.host and
                req.host != 'localhost'):
            if not req.host in self.get_names():
                raise URLError("file:// scheme is supported only on localhost")
        else:
            return self.open_local_file(req)

    # names for the localhost
    names = None
    def get_names(self):
        if FileHandler.names is None:
            try:
                FileHandler.names = tuple(
                    socket.gethostbyname_ex('localhost')[2] +
                    socket.gethostbyname_ex(socket.gethostname())[2])
            except socket.gaierror:
                FileHandler.names = (socket.gethostbyname('localhost'),)
        return FileHandler.names

    # not entirely sure what the rules are here
    def open_local_file(self, req):
        import email.utils
        import mimetypes
        filename = req.selector
        localfile = url2pathname(filename)
        try:
            stats = os.stat(localfile)
            size = stats.st_size
            modified = email.utils.formatdate(stats.st_mtime, usegmt=True)
            mtype = mimetypes.guess_type(filename)[0]
            headers = email.message_from_string(
                'Content-type: %s\nContent-length: %d\nLast-modified: %s\n' %
                (mtype or 'text/plain', size, modified))
            if host:
                host, port = _splitport(host)
            if not host or \
                (not port and _safe_gethostbyname(host) in self.get_names()):
                if host:
                    origurl = 'file://' + host + filename
                else:
                    origurl = 'file://' + filename
                return addinfourl(open(localfile, 'rb'), headers, origurl)
        except OSError as exp:
            raise URLError(exp)
        raise URLError('file not on local host')

def _safe_gethostbyname(host):
    try:
        return socket.gethostbyname(host)
    except socket.gaierror:

class FTPHandler(BaseHandler):
    def ftp_open(self, req):
        import ftplib
            raise URLError('ftp error: no host given')
        host, port = _splitport(host)
        if port is None:
            port = ftplib.FTP_PORT
        else:
            port = int(port)

        # username/password handling
        user, host = _splituser(host)
        if user:
            user, passwd = _splitpasswd(user)
        else:
            passwd = None
        host = unquote(host)
        user = user or ''
        passwd = passwd or ''

        try:
            host = socket.gethostbyname(host)
        except OSError as msg:
            raise URLError(msg)
        path, attrs = _splitattr(req.selector)
        dirs = path.split('/')
        dirs = list(map(unquote, dirs))
        dirs, file = dirs[:-1], dirs[-1]
        if dirs and not dirs[0]:
            dirs = dirs[1:]
        try:
            fw = self.connect_ftp(user, passwd, host, port, dirs, req.timeout)
            type = file and 'I' or 'D'
            for attr in attrs:
                attr, value = _splitvalue(attr)
                if attr.lower() == 'type' and \
                   value in ('a', 'A', 'i', 'I', 'd', 'D'):
                    type = value.upper()
            fp, retrlen = fw.retrfile(file, type)
            headers = ""
            mtype = mimetypes.guess_type(req.full_url)[0]
            if mtype:
                headers += "Content-type: %s\n" % mtype
            if retrlen is not None and retrlen >= 0:
                headers += "Content-length: %d\n" % retrlen
            headers = email.message_from_string(headers)
            return addinfourl(fp, headers, req.full_url)
        except ftplib.all_errors as exp:
            exc = URLError('ftp error: %r' % exp)
            raise exc.with_traceback(sys.exc_info()[2])

    def connect_ftp(self, user, passwd, host, port, dirs, timeout):
        return ftpwrapper(user, passwd, host, port, dirs, timeout,
                          persistent=False)

class CacheFTPHandler(FTPHandler):
    # XXX would be nice to have pluggable cache strategies
    # XXX this stuff is definitely not thread safe
        self.cache = {}
        self.timeout = {}
        self.soonest = 0
        self.delay = 60
        self.max_conns = 16

    def setTimeout(self, t):
        self.delay = t

    def setMaxConns(self, m):
        self.max_conns = m

        key = user, host, port, '/'.join(dirs), timeout
        if key in self.cache:
            self.timeout[key] = time.time() + self.delay
        else:
            self.cache[key] = ftpwrapper(user, passwd, host, port,
                                         dirs, timeout)
        self.check_cache()
        return self.cache[key]

    def check_cache(self):
        # first check for old ones
        t = time.time()
        if self.soonest <= t:
            for k, v in list(self.timeout.items()):
                if v < t:
                    self.cache[k].close()
                    del self.cache[k]
                    del self.timeout[k]
        self.soonest = min(list(self.timeout.values()))

        # then check the size
        if len(self.cache) == self.max_conns:
                if v == self.soonest:
                    break
            self.soonest = min(list(self.timeout.values()))

    def clear_cache(self):
        for conn in self.cache.values():
            conn.close()
        self.cache.clear()
        self.timeout.clear()

class DataHandler(BaseHandler):
    def data_open(self, req):
        # data URLs as specified in RFC 2397.
        #
        # ignores POSTed data
        #
        # syntax:
        # dataurl   := "data:" [ mediatype ] [ ";base64" ] "," data
        # mediatype := [ type "/" subtype ] *( ";" parameter )
        # data      := *urlchar
        # parameter := attribute "=" value

        scheme, data = url.split(":",1)
        mediatype, data = data.split(",",1)

        # even base64 encoded data URLs might be quoted so unquote in any case:
        data = unquote_to_bytes(data)
        if mediatype.endswith(";base64"):
            data = base64.decodebytes(data)
            mediatype = mediatype[:-7]

        if not mediatype:
            mediatype = "text/plain;charset=US-ASCII"

        headers = email.message_from_string("Content-type: %s\nContent-length: %d\n" %
            (mediatype, len(data)))

        return addinfourl(io.BytesIO(data), headers, url)

# Code move from the old urllib module

MAXFTPCACHE = 10        # Trim the ftp cache beyond this size

# Helper for non-unix systems
if os.name == 'nt':
    from nturl2path import url2pathname, pathname2url
else:
    def url2pathname(pathname):
        """OS-specific conversion from a relative URL of the 'file' scheme
        to a file system path; not recommended for general use."""
        return unquote(pathname)

    def pathname2url(pathname):
        """OS-specific conversion from a file system path to a relative URL
        of the 'file' scheme; not recommended for general use."""
        return quote(pathname)

ftpcache = {}

class URLopener:
    """Class to open URLs.
    This is a class rather than just a subroutine because we may need
    more than one set of global protocol-specific options.
    Note -- this is a base class for those who don't want the
    automatic handling of errors type 302 (relocated) and 401
    (authorization needed)."""

    __tempfiles = None

    version = "Python-urllib/%s" % __version__

    # Constructor
    def __init__(self, proxies=None, **x509):
        msg = "%(class)s style of invoking requests is deprecated. " \
              "Use newer urlopen functions/methods" % {'class': self.__class__.__name__}
        warnings.warn(msg, DeprecationWarning, stacklevel=3)
        self.key_file = x509.get('key_file')
        self.cert_file = x509.get('cert_file')
        self.addheaders = [('User-Agent', self.version), ('Accept', '*/*')]
        self.__tempfiles = []
        self.__unlink = os.unlink # See cleanup()
        self.tempcache = None
        # Undocumented feature: if you assign {} to tempcache,
        # it is used to cache files retrieved with
        # self.retrieve().  This is not enabled by default
        # since it does not work for changing documents (and I
        # haven't got the logic to check expiration headers
        # yet).
        self.ftpcache = ftpcache
        # Undocumented feature: you can use a different
        # ftp cache by assigning to the .ftpcache member;
        # in case you want logically independent URL openers
        # XXX This is not threadsafe.  Bah.

        self.cleanup()

    def cleanup(self):
        # This code sometimes runs when the rest of this module
        # has already been deleted, so it can't use any globals
        # or import anything.
        if self.__tempfiles:
            for file in self.__tempfiles:
                try:
                    self.__unlink(file)
                except OSError:
                    pass
            del self.__tempfiles[:]
        if self.tempcache:
            self.tempcache.clear()

    def addheader(self, *args):
        """Add a header to be used by the HTTP interface only
        e.g. u.addheader('Accept', 'sound/basic')"""
        self.addheaders.append(args)

    # External interface
    def open(self, fullurl, data=None):
        """Use URLopener().open(file) instead of open(file, 'r')."""
        fullurl = unwrap(_to_bytes(fullurl))
        fullurl = quote(fullurl, safe="%/:=&?~#+!$,;'@()*[]|")
        if self.tempcache and fullurl in self.tempcache:
            filename, headers = self.tempcache[fullurl]
            fp = open(filename, 'rb')
            return addinfourl(fp, headers, fullurl)
        urltype, url = _splittype(fullurl)
        if not urltype:
            urltype = 'file'
        if urltype in self.proxies:
            proxy = self.proxies[urltype]
            urltype, proxyhost = _splittype(proxy)
            host, selector = _splithost(proxyhost)
            url = (host, fullurl) # Signal special case to open_*()
        else:
            proxy = None
        name = 'open_' + urltype
        self.type = urltype
        name = name.replace('-', '_')
        if not hasattr(self, name) or name == 'open_local_file':
            if proxy:
                return self.open_unknown_proxy(proxy, fullurl, data)
            else:
                return self.open_unknown(fullurl, data)
        try:
            if data is None:
                return getattr(self, name)(url)
            else:
                return getattr(self, name)(url, data)
        except (HTTPError, URLError):
            raise
            raise OSError('socket error', msg).with_traceback(sys.exc_info()[2])

    def open_unknown(self, fullurl, data=None):
        """Overridable interface to open unknown URL type."""
        type, url = _splittype(fullurl)
        raise OSError('url error', 'unknown url type', type)

    def open_unknown_proxy(self, proxy, fullurl, data=None):
        raise OSError('url error', 'invalid proxy for %s' % type, proxy)

    def retrieve(self, url, filename=None, reporthook=None, data=None):
        """retrieve(url) returns (filename, headers) for a local object
        or (tempfilename, headers) for a remote object."""
        url = unwrap(_to_bytes(url))
        if self.tempcache and url in self.tempcache:
            return self.tempcache[url]
        type, url1 = _splittype(url)
        if filename is None and (not type or type == 'file'):
            try:
                fp = self.open_local_file(url1)
                hdrs = fp.info()
                fp.close()
                return url2pathname(_splithost(url1)[1]), hdrs
            except OSError as msg:
                pass
        fp = self.open(url, data)
        try:
            headers = fp.info()
            if filename:
                tfp = open(filename, 'wb')
            else:
                garbage, path = _splittype(url)
                garbage, path = _splithost(path or "")
                path, garbage = _splitquery(path or "")
                path, garbage = _splitattr(path or "")
                suffix = os.path.splitext(path)[1]
                (fd, filename) = tempfile.mkstemp(suffix)
                self.__tempfiles.append(filename)
                tfp = os.fdopen(fd, 'wb')
            try:
                result = filename, headers
                if self.tempcache is not None:
                    self.tempcache[url] = result
                bs = 1024*8
                size = -1
                read = 0
                blocknum = 0
                if "content-length" in headers:
                    size = int(headers["Content-Length"])
                while 1:
                    block = fp.read(bs)
                    if not block:
                        break
                    read += len(block)
                    tfp.write(block)
                    blocknum += 1
                    if reporthook:
                        reporthook(blocknum, bs, size)
            finally:
                tfp.close()
        finally:
            fp.close()

        # raise exception if actual size does not match content-length header
        if size >= 0 and read < size:
            raise ContentTooShortError(
                "retrieval incomplete: got only %i out of %i bytes"
                % (read, size), result)

        return result

    # Each method named open_<type> knows how to open that type of URL

    def _open_generic_http(self, connection_factory, url, data):
        """Make an HTTP connection using connection_class.

        This is an internal method that should be called from
        open_http() or open_https().

        Arguments:
        - connection_factory should take a host name and return an
          HTTPConnection instance.
        - url is the url to retrieval or a host, relative-path pair.
        - data is payload for a POST request or None.
        """

        proxy_passwd= None
        if isinstance(url, str):
            host, selector = _splithost(url)
            if host:
                user_passwd, host = _splituser(host)
        else:
            host, selector = url
            # check whether the proxy contains authorization information
            proxy_passwd, host = _splituser(host)
            # now we proceed with the url we want to obtain
            urltype, rest = _splittype(selector)
            url = rest
            user_passwd = None
            if urltype.lower() != 'http':
                realhost = None
            else:
                realhost, rest = _splithost(rest)
                    user_passwd, realhost = _splituser(realhost)
                if user_passwd:
                    selector = "%s://%s%s" % (urltype, realhost, rest)
                if proxy_bypass(realhost):
                    host = realhost

        if not host: raise OSError('http error', 'no host given')

        if proxy_passwd:
            proxy_passwd = unquote(proxy_passwd)
            proxy_auth = base64.b64encode(proxy_passwd.encode()).decode('ascii')
        else:
            proxy_auth = None

        if user_passwd:
            user_passwd = unquote(user_passwd)
            auth = base64.b64encode(user_passwd.encode()).decode('ascii')
        else:
            auth = None
        http_conn = connection_factory(host)
        headers = {}
        if proxy_auth:
            headers["Proxy-Authorization"] = "Basic %s" % proxy_auth
        if auth:
            headers["Authorization"] =  "Basic %s" % auth
        if realhost:
            headers["Host"] = realhost

        # Add Connection:close as we don't support persistent connections yet.
        # This helps in closing the socket and avoiding ResourceWarning

        for header, value in self.addheaders:
            headers[header] = value

            headers["Content-Type"] = "application/x-www-form-urlencoded"
            http_conn.request("POST", selector, data, headers)
        else:
            http_conn.request("GET", selector, headers=headers)

        try:
            response = http_conn.getresponse()
        except http.client.BadStatusLine:
            # something went wrong with the HTTP status line
            raise URLError("http protocol error: bad status line")

        if 200 <= response.status < 300:
            return addinfourl(response, response.msg, "http:" + url,
                              response.status)
        else:
            return self.http_error(
                url, response.fp,
                response.status, response.reason, response.msg, data)

    def open_http(self, url, data=None):
        """Use HTTP protocol."""
        return self._open_generic_http(http.client.HTTPConnection, url, data)

    def http_error(self, url, fp, errcode, errmsg, headers, data=None):
        """Handle http errors.

        Derived class can override this, or provide specific handlers
        named http_error_DDD where DDD is the 3-digit error code."""
        # First check if there's a specific handler for this error
        name = 'http_error_%d' % errcode
        if hasattr(self, name):
            method = getattr(self, name)
                result = method(url, fp, errcode, errmsg, headers)
            else:
                result = method(url, fp, errcode, errmsg, headers, data)
            if result: return result
        return self.http_error_default(url, fp, errcode, errmsg, headers)

    def http_error_default(self, url, fp, errcode, errmsg, headers):
        """Default error handler: close the connection and raise OSError."""
        raise HTTPError(url, errcode, errmsg, headers, None)

    if _have_ssl:
        def _https_connection(self, host):
            return http.client.HTTPSConnection(host,
                                           key_file=self.key_file,
                                           cert_file=self.cert_file)

        def open_https(self, url, data=None):
            """Use HTTPS protocol."""
            return self._open_generic_http(self._https_connection, url, data)

    def open_file(self, url):
        """Use local file or FTP depending on form of URL."""
        if not isinstance(url, str):
            raise URLError('file error: proxy support for file protocol currently not implemented')
        if url[:2] == '//' and url[2:3] != '/' and url[2:12].lower() != 'localhost/':
            raise ValueError("file:// scheme is supported only on localhost")
        else:
            return self.open_local_file(url)

    def open_local_file(self, url):
        """Use local file."""
        host, file = _splithost(url)
        localname = url2pathname(file)
        try:
            stats = os.stat(localname)
        except OSError as e:
            raise URLError(e.strerror, e.filename)
        size = stats.st_size
        modified = email.utils.formatdate(stats.st_mtime, usegmt=True)
        mtype = mimetypes.guess_type(url)[0]
        headers = email.message_from_string(
            'Content-Type: %s\nContent-Length: %d\nLast-modified: %s\n' %
            (mtype or 'text/plain', size, modified))
            urlfile = file
            if file[:1] == '/':
                urlfile = 'file://' + file
            return addinfourl(open(localname, 'rb'), headers, urlfile)
        if (not port
           and socket.gethostbyname(host) in ((localhost(),) + thishost())):
            elif file[:2] == './':
                raise ValueError("local file url may start with / or file:. Unknown url of type: %s" % url)
        raise URLError('local file error: not on local host')

    def open_ftp(self, url):
        """Use FTP protocol."""
            raise URLError('ftp error: proxy support for ftp protocol currently not implemented')
        host, path = _splithost(url)
        if not host: raise URLError('ftp error: no host given')
        if user: user, passwd = _splitpasswd(user)
        else: passwd = None
        user = unquote(user or '')
        passwd = unquote(passwd or '')
        host = socket.gethostbyname(host)
        if not port:
            import ftplib
        else:
        path, attrs = _splitattr(path)
        path = unquote(path)
        if dirs and not dirs[0]: dirs = dirs[1:]
        if dirs and not dirs[0]: dirs[0] = '/'
        key = user, host, port, '/'.join(dirs)
        # XXX thread unsafe!
        if len(self.ftpcache) > MAXFTPCACHE:
            # Prune the cache, rather arbitrarily
            for k in list(self.ftpcache):
                if k != key:
                    v = self.ftpcache[k]
                    del self.ftpcache[k]
                    v.close()
        try:
            if key not in self.ftpcache:
                self.ftpcache[key] = \
                    ftpwrapper(user, passwd, host, port, dirs)
            if not file: type = 'D'
            else: type = 'I'
            (fp, retrlen) = self.ftpcache[key].retrfile(file, type)
            mtype = mimetypes.guess_type("ftp:" + url)[0]
            if mtype:
                headers += "Content-Type: %s\n" % mtype
                headers += "Content-Length: %d\n" % retrlen
            return addinfourl(fp, headers, "ftp:" + url)
        except ftperrors() as exp:
            raise URLError('ftp error %r' % exp).with_traceback(sys.exc_info()[2])

    def open_data(self, url, data=None):
        """Use "data" URL."""
            raise URLError('data error: proxy support for data protocol currently not implemented')
        # ignore POSTed data
        #
        # syntax of data URLs:
        try:
            [type, data] = url.split(',', 1)
            raise OSError('data error', 'bad data URL')
        if not type:
            type = 'text/plain;charset=US-ASCII'
        semi = type.rfind(';')
        if semi >= 0 and '=' not in type[semi:]:
            encoding = type[semi+1:]
            type = type[:semi]
        else:
            encoding = ''
        msg = []
        msg.append('Date: %s'%time.strftime('%a, %d %b %Y %H:%M:%S GMT',
                                            time.gmtime(time.time())))
        msg.append('Content-type: %s' % type)
        if encoding == 'base64':
            # XXX is this encoding/decoding ok?
            data = base64.decodebytes(data.encode('ascii')).decode('latin-1')
        else:
            data = unquote(data)
        msg.append('Content-Length: %d' % len(data))
        msg.append('')
        msg.append(data)
        msg = '\n'.join(msg)
        headers = email.message_from_string(msg)
        f = io.StringIO(msg)
        #f.fileno = None     # needed for addinfourl
        return addinfourl(f, headers, url)

class FancyURLopener(URLopener):
    """Derived class with handlers for errors we can handle (perhaps)."""

        URLopener.__init__(self, *args, **kwargs)
        self.auth_cache = {}
        self.tries = 0
        self.maxtries = 10

        """Default error handling -- don't raise an exception."""
        return addinfourl(fp, headers, "http:" + url, errcode)

    def http_error_302(self, url, fp, errcode, errmsg, headers, data=None):
        """Error 302 -- relocated (temporarily)."""
        self.tries += 1
        try:
            if self.maxtries and self.tries >= self.maxtries:
                if hasattr(self, "http_error_500"):
                    meth = self.http_error_500
                else:
                    meth = self.http_error_default
                return meth(url, fp, 500,
                            "Internal Server Error: Redirect Recursion",
                            headers)
            result = self.redirect_internal(url, fp, errcode, errmsg,
                                            headers, data)
        finally:
            self.tries = 0

    def redirect_internal(self, url, fp, errcode, errmsg, headers, data):
        if 'location' in headers:
            newurl = headers['location']
        elif 'uri' in headers:
            newurl = headers['uri']
        else:
            return

        # In case the server sent a relative URL, join with original:
        newurl = urljoin(self.type + ":" + url, newurl)

        # For security reasons, we don't allow redirection to anything other
        # than http, https and ftp.

        # We are using newer HTTPError with older redirect_internal method
        # This older method will get deprecated in 3.3

            raise HTTPError(newurl, errcode,
                            errmsg +
                            " Redirection to url '%s' is not allowed." % newurl,
                            headers, fp)

        return self.open(newurl)

    def http_error_301(self, url, fp, errcode, errmsg, headers, data=None):
        """Error 301 -- also relocated (permanently)."""
        return self.http_error_302(url, fp, errcode, errmsg, headers, data)

    def http_error_303(self, url, fp, errcode, errmsg, headers, data=None):
        """Error 303 -- also relocated (essentially identical to 302)."""

    def http_error_307(self, url, fp, errcode, errmsg, headers, data=None):
        """Error 307 -- relocated, but turn POST into error."""
        if data is None:
            return self.http_error_302(url, fp, errcode, errmsg, headers, data)
        else:
            return self.http_error_default(url, fp, errcode, errmsg, headers)

    def http_error_401(self, url, fp, errcode, errmsg, headers, data=None,
            retry=False):
        """Error 401 -- authentication required.
        This function supports Basic authentication only."""
        if 'www-authenticate' not in headers:
            URLopener.http_error_default(self, url, fp,
                                         errcode, errmsg, headers)
        stuff = headers['www-authenticate']
        match = re.match('[ \t]*([^ \t]+)[ \t]+realm="([^"]*)"', stuff)
        if not match:
        scheme, realm = match.groups()
        if scheme.lower() != 'basic':
        if not retry:
            URLopener.http_error_default(self, url, fp, errcode, errmsg,
                    headers)
        name = 'retry_' + self.type + '_basic_auth'
            return getattr(self,name)(url, realm)
        else:
            return getattr(self,name)(url, realm, data)

    def http_error_407(self, url, fp, errcode, errmsg, headers, data=None,
        """Error 407 -- proxy authentication required.
        if 'proxy-authenticate' not in headers:
        stuff = headers['proxy-authenticate']
                    headers)
        name = 'retry_proxy_' + self.type + '_basic_auth'
        else:

    def retry_proxy_http_basic_auth(self, url, realm, data=None):
        host, selector = _splithost(url)
        newurl = 'http://' + host + selector
        proxy = self.proxies['http']
        urltype, proxyhost = _splittype(proxy)
        proxyhost, proxyselector = _splithost(proxyhost)
        i = proxyhost.find('@') + 1
        proxyhost = proxyhost[i:]
        user, passwd = self.get_user_passwd(proxyhost, realm, i)
        if not (user or passwd): return None
        proxyhost = "%s:%s@%s" % (quote(user, safe=''),
                                  quote(passwd, safe=''), proxyhost)
        self.proxies['http'] = 'http://' + proxyhost + proxyselector
            return self.open(newurl)
        else:
            return self.open(newurl, data)

    def retry_proxy_https_basic_auth(self, url, realm, data=None):
        newurl = 'https://' + host + selector
        proxy = self.proxies['https']
        self.proxies['https'] = 'https://' + proxyhost + proxyselector
        else:

    def retry_http_basic_auth(self, url, realm, data=None):
        i = host.find('@') + 1
        host = host[i:]
        user, passwd = self.get_user_passwd(host, realm, i)
        host = "%s:%s@%s" % (quote(user, safe=''),
                             quote(passwd, safe=''), host)
        else:

    def retry_https_basic_auth(self, url, realm, data=None):
        else:

    def get_user_passwd(self, host, realm, clear_cache=0):
        key = realm + '@' + host.lower()
        if key in self.auth_cache:
            if clear_cache:
                del self.auth_cache[key]
            else:
                return self.auth_cache[key]
        user, passwd = self.prompt_user_passwd(host, realm)
        if user or passwd: self.auth_cache[key] = (user, passwd)
        return user, passwd

    def prompt_user_passwd(self, host, realm):
        """Override this in a GUI environment!"""
        import getpass
        try:
            user = input("Enter username for %s at %s: " % (realm, host))
            passwd = getpass.getpass("Enter password for %s in %s at %s: " %
                (user, realm, host))
            return user, passwd
        except KeyboardInterrupt:
            print()
            return None, None

# Utility functions

_localhost = None
def localhost():
    """Return the IP address of the magic hostname 'localhost'."""
    global _localhost
    if _localhost is None:
        _localhost = socket.gethostbyname('localhost')
    return _localhost

_thishost = None
def thishost():
    """Return the IP addresses of the current host."""
    global _thishost
    if _thishost is None:
        try:
            _thishost = tuple(socket.gethostbyname_ex(socket.gethostname())[2])
        except socket.gaierror:
            _thishost = tuple(socket.gethostbyname_ex('localhost')[2])
    return _thishost

_ftperrors = None
def ftperrors():
    """Return the set of errors raised by the FTP class."""
    global _ftperrors
    if _ftperrors is None:
        _ftperrors = ftplib.all_errors
    return _ftperrors

_noheaders = None
def noheaders():
    """Return an empty email Message object."""
    global _noheaders
    if _noheaders is None:
        _noheaders = email.message_from_string("")
    return _noheaders

# Utility classes

class ftpwrapper:
    """Class used by open_ftp() for cache of open FTP connections."""

    def __init__(self, user, passwd, host, port, dirs, timeout=None,
                 persistent=True):
        self.user = user
        self.dirs = dirs
        self.timeout = timeout
        self.refcount = 0
        self.keepalive = persistent
        try:
            self.init()
        except:
            raise

        self.busy = 0
        self.ftp = ftplib.FTP()
        self.ftp.connect(self.host, self.port, self.timeout)
        self.ftp.login(self.user, self.passwd)
        _target = '/'.join(self.dirs)
        self.ftp.cwd(_target)

    def retrfile(self, file, type):
        self.endtransfer()
        if type in ('d', 'D'): cmd = 'TYPE A'; isdir = 1
        else: cmd = 'TYPE ' + type; isdir = 0
        try:
            self.ftp.voidcmd(cmd)
        except ftplib.all_errors:
        conn = None
        if file and not isdir:
            # Try to retrieve as a file
            try:
                cmd = 'RETR ' + file
                conn, retrlen = self.ftp.ntransfercmd(cmd)
            except ftplib.error_perm as reason:
                if str(reason)[:3] != '550':
                    raise URLError('ftp error: %r' % reason).with_traceback(
                        sys.exc_info()[2])
        if not conn:
            # Set transfer mode to ASCII!
            self.ftp.voidcmd('TYPE A')
            # Try a directory listing. Verify that directory exists.
            if file:
                pwd = self.ftp.pwd()
                try:
                    try:
                        self.ftp.cwd(file)
                    except ftplib.error_perm as reason:
                        raise URLError('ftp error: %r' % reason) from reason
                finally:
                    self.ftp.cwd(pwd)
                cmd = 'LIST ' + file
            else:
                cmd = 'LIST'
            conn, retrlen = self.ftp.ntransfercmd(cmd)
        self.busy = 1

        ftpobj = addclosehook(conn.makefile('rb'), self.file_close)
        self.refcount += 1
        conn.close()
        # Pass back both a suitably decorated object and a retrieval length
        return (ftpobj, retrlen)

    def endtransfer(self):

        self.keepalive = False
        if self.refcount <= 0:
            self.real_close()

    def file_close(self):
        self.refcount -= 1
        if self.refcount <= 0 and not self.keepalive:

    def real_close(self):
        try:
            self.ftp.close()
        except ftperrors():
            pass

# Proxy handling
def getproxies_environment():
    """Return a dictionary of scheme -> proxy server URL mappings.

    Scan the environment for variables named <scheme>_proxy;
    this seems to be the standard convention.  If you need a
    different way, you can pass a proxies dictionary to the
    [Fancy]URLopener constructor.

    """
    proxies = {}
    # in order to prefer lowercase variables, process environment in
    # two passes: first matches any, second pass matches lowercase only
    for name, value in os.environ.items():
        name = name.lower()
        if value and name[-6:] == '_proxy':
            proxies[name[:-6]] = value
    # CVE-2016-1000110 - If we are running as CGI script, forget HTTP_PROXY
    # (non-all-lowercase) as it may be set from the web server by a "Proxy:"
    # header from the client
    # If "proxy" is lowercase, it will still be used thanks to the next block
    if 'REQUEST_METHOD' in os.environ:
        proxies.pop('http', None)
        if name[-6:] == '_proxy':
            name = name.lower()
            if value:
                proxies[name[:-6]] = value
            else:
                proxies.pop(name[:-6], None)
    return proxies

def proxy_bypass_environment(host, proxies=None):
    """Test if proxies should not be used for a particular host.

    Checks the proxy dict for the value of no_proxy, which should
    be a list of comma separated DNS suffixes, or '*' for all hosts.

    """
    if proxies is None:
        proxies = getproxies_environment()
    # don't bypass, if no_proxy isn't specified
    try:
        no_proxy = proxies['no']
    except KeyError:
    # '*' is special case for always bypass
    if no_proxy == '*':
        return True
    host = host.lower()
    # strip port off host
    hostonly, port = _splitport(host)
    # check if the host ends with any of the DNS suffixes
    for name in no_proxy.split(','):
        name = name.strip()
        if name:
            name = name.lstrip('.')  # ignore leading dots
            if hostonly == name or host == name:
                return True
            name = '.' + name
            if hostonly.endswith(name) or host.endswith(name):
    # otherwise, don't bypass
    return False

# This code tests an OSX specific data structure but is testable on all
# platforms
def _proxy_bypass_macosx_sysconf(host, proxy_settings):
    """
    Return True iff this host shouldn't be accessed using a proxy

    This function uses the MacOSX framework SystemConfiguration
    to fetch the proxy information.

    proxy_settings come from _scproxy._get_proxy_settings or get mocked ie:
    { 'exclude_simple': bool,
      'exceptions': ['foo.bar', '*.bar.com', '127.0.0.1', '10.1', '10.0/16']
    }
    """
    from fnmatch import fnmatch

    def ip2num(ipAddr):
        parts = ipAddr.split('.')
        parts = list(map(int, parts))
        if len(parts) != 4:
            parts = (parts + [0, 0, 0, 0])[:4]
        return (parts[0] << 24) | (parts[1] << 16) | (parts[2] << 8) | parts[3]

    # Check for simple host names:
    if '.' not in host:
        if proxy_settings['exclude_simple']:

    hostIP = None

    for value in proxy_settings.get('exceptions', ()):
        # Items in the list are strings like these: *.local, 169.254/16
        if not value: continue

        m = re.match(r"(\d+(?:\.\d+)*)(/\d+)?", value)
        if m is not None:
            if hostIP is None:
                try:
                    hostIP = socket.gethostbyname(hostonly)
                    hostIP = ip2num(hostIP)
                    continue

            base = ip2num(m.group(1))
            mask = m.group(2)
            if mask is None:
                mask = 8 * (m.group(1).count('.') + 1)
            else:
                mask = int(mask[1:])
            mask = 32 - mask

            if (hostIP >> mask) == (base >> mask):

        elif fnmatch(host, value):

if sys.platform == 'darwin':
    from _scproxy import _get_proxy_settings, _get_proxies

    def proxy_bypass_macosx_sysconf(host):
        proxy_settings = _get_proxy_settings()
        return _proxy_bypass_macosx_sysconf(host, proxy_settings)

    def getproxies_macosx_sysconf():
        """Return a dictionary of scheme -> proxy server URL mappings.

        This function uses the MacOSX framework SystemConfiguration
        to fetch the proxy information.
        """
        return _get_proxies()

    def proxy_bypass(host):
        """Return True, if host should be bypassed.

        Checks proxy settings gathered from the environment, if specified,
        or from the MacOSX framework SystemConfiguration.

        """
        if proxies:
            return proxy_bypass_environment(host, proxies)
        else:
            return proxy_bypass_macosx_sysconf(host)

    def getproxies():
        return getproxies_environment() or getproxies_macosx_sysconf()

elif os.name == 'nt':
    def getproxies_registry():

        Win32 uses the registry to store proxies.

        """
        proxies = {}
        try:
            import winreg
        except ImportError:
            # Std module, so should be around - but you never know!
            return proxies
        try:
            internetSettings = winreg.OpenKey(winreg.HKEY_CURRENT_USER,
                r'Software\Microsoft\Windows\CurrentVersion\Internet Settings')
            proxyEnable = winreg.QueryValueEx(internetSettings,
                                               'ProxyEnable')[0]
            if proxyEnable:
                # Returned as Unicode but problems if not converted to ASCII
                proxyServer = str(winreg.QueryValueEx(internetSettings,
                                                       'ProxyServer')[0])
                if '=' in proxyServer:
                    # Per-protocol settings
                    for p in proxyServer.split(';'):
                        protocol, address = p.split('=', 1)
                        # See if address has a type:// prefix
                        if not re.match('(?:[^/:]+)://', address):
                            address = '%s://%s' % (protocol, address)
                        proxies[protocol] = address
                else:
                    # Use one setting for all protocols
                    if proxyServer[:5] == 'http:':
                        proxies['http'] = proxyServer
                    else:
                        proxies['http'] = 'http://%s' % proxyServer
                        proxies['https'] = 'https://%s' % proxyServer
                        proxies['ftp'] = 'ftp://%s' % proxyServer
            internetSettings.Close()
        except (OSError, ValueError, TypeError):
            # Either registry key not found etc, or the value in an
            # unexpected format.
            # proxies already set up to be empty so nothing to do
            pass
        return proxies

        Returns settings gathered from the environment, if specified,
        or the registry.

        """
        return getproxies_environment() or getproxies_registry()

    def proxy_bypass_registry(host):
        try:
            # Std modules, so should be around - but you never know!
            return 0
        try:
            proxyOverride = str(winreg.QueryValueEx(internetSettings,
                                                     'ProxyOverride')[0])
            # ^^^^ Returned as Unicode but problems if not converted to ASCII
            return 0
        if not proxyEnable or not proxyOverride:
            return 0
        # try to make a host list from name and IP address.
        rawHost, port = _splitport(host)
        host = [rawHost]
        try:
            addr = socket.gethostbyname(rawHost)
            if addr != rawHost:
                host.append(addr)
            pass
        try:
            fqdn = socket.getfqdn(rawHost)
            if fqdn != rawHost:
                host.append(fqdn)
            pass
        # make a check value list from the registry entry: replace the
        # '<local>' string by the localhost entry and the corresponding
        # canonical entry.
        proxyOverride = proxyOverride.split(';')
        # now check if we match one of the registry values.
        for test in proxyOverride:
            if test == '<local>':
                if '.' not in rawHost:
                    return 1
            test = test.replace(".", r"\.")     # mask dots
            test = test.replace("*", r".*")     # change glob sequence
            test = test.replace("?", r".")      # change glob char
            for val in host:
                if re.match(test, val, re.I):
                    return 1
        return 0

        """
        else:
            return proxy_bypass_registry(host)

else:
    # By default use environment variables
    getproxies = getproxies_environment
    proxy_bypass = proxy_bypass_environment

"""M2Crypto enhancement to xmlrpclib.

import M2Crypto

from M2Crypto import SSL, httpslib, m2urllib, six
from typing import Any, AnyStr, Callable, Optional  # noqa

from M2Crypto.six.moves.xmlrpc_client import ProtocolError, Transport
    from xmlrpc.client import *  # noqa
else:
    from xmlrpclib import *  # noqa

__version__ = M2Crypto.__version__

class SSL_Transport(Transport):

    user_agent = "M2Crypto_XMLRPC/%s - %s" % (__version__,
                                              Transport.user_agent)

    def __init__(self, ssl_context=None, *args, **kw):
        # type: (Optional[SSL.Context], *Any, **Any) -> None
        Transport.__init__(self, *args, **kw)
        if ssl_context is None:
        else:
            self.ssl_ctx = ssl_context

    def request(self, host, handler, request_body, verbose=0):
        # type: (AnyStr, Callable, bytes, int) -> object
        # Handle username and password.
        user_passwd, host_port = m2urllib.splituser(host)
        _host, _port = m2urllib.splitport(host_port)
        h = httpslib.HTTPSConnection(_host, int(_port),
                                     ssl_context=self.ssl_ctx)
        if verbose:
            h.set_debuglevel(1)

        # What follows is as in xmlrpclib.Transport. (Except the authz bit.)
        h.putrequest("POST", handler)

        # required by HTTP/1.1
        h.putheader("Host", _host)

        # required by XML-RPC
        h.putheader("User-Agent", self.user_agent)
        h.putheader("Content-Type", "text/xml")
        h.putheader("Content-Length", str(len(request_body)))

        # Authorisation.
        if user_passwd is not None:
            h.putheader('Authorization', 'Basic %s' % auth)

        h.endheaders()

        if request_body:
            h.send(request_body)

        response = h.getresponse()

        if response.status != 200:
            raise ProtocolError(
                host + handler,
                response.status, response.reason,
                response.getheaders()
            )

        self.verbose = verbose
        return self.parse_response(response)

"""
M2Crypto threading support, required for multithreaded applications.

# M2Crypto

def init():
    """
    Initialize threading support.
    """
    m2.threading_init()

    """
    End and cleanup threading support.
    """
    m2.threading_cleanup()

"""
    M2Crypto utility routines.

    NOTHING IN THIS MODULE IS GUARANTEED TO BE STABLE, USED ONLY FOR
    INTERNAL PURPOSES OF M2CRYPTO.

    Copyright (c) 1999-2004 Ng Pheng Siong. All rights reserved.

"""

import sys

from typing import Any, Optional, TextIO, Tuple, Union  # noqa
# see https://github.com/python/typeshed/issues/222
AddrType = Union[Tuple[str, int], str]

log = logging.getLogger('util')

class UtilError(Exception):
    pass

m2.util_init(UtilError)

def pkcs5_pad(data, blklen=8):
    # type: (str, int) -> str
    pad = (8 - (len(data) % 8))
    return data + chr(pad) * pad

def pkcs7_pad(data, blklen):
    if blklen > 255:
        raise ValueError('illegal block size')
    pad = (blklen - (len(data) % blklen))

def bin_to_hex(b):
    # type: (bytes) -> str
    return six.ensure_text(binascii.b2a_base64(b)[:-1])

def octx_to_num(x):
    # type: (bytes) -> int
    return int(binascii.hexlify(x), 16)

def genparam_callback(p, n, out=sys.stdout):
    # type: (int, Any, TextIO) -> None

def quiet_genparam_callback(p, n, out):
    # type: (Any, Any, Any) -> None
    pass

def passphrase_callback(v, prompt1='Enter passphrase:',
                        prompt2='Verify passphrase:'):
    # type: (bool, str, str) -> Optional[str]
    from getpass import getpass
    while 1:
        try:
            p1 = getpass(prompt1)
            if v:
                p2 = getpass(prompt2)
                if p1 == p2:
                    break
            else:
                break
    return p1

def no_passphrase_callback(*args):
    # type: (*Any) -> str
    return ''

"""M2Crypto SSL services.

import socket, os

# M2Crypto
from M2Crypto import _m2crypto as m2

    pass

class SSLTimeoutError(SSLError, socket.timeout):
    pass

m2.ssl_init(SSLError, SSLTimeoutError)

# M2Crypto.SSL
from M2Crypto.SSL.Cipher import Cipher, Cipher_Stack
from M2Crypto.SSL.Connection import Connection
from M2Crypto.SSL.Context import Context
from M2Crypto.SSL.SSLServer import SSLServer, ThreadingSSLServer
if os.name != 'nt':
    from M2Crypto.SSL.SSLServer import ForkingSSLServer
from M2Crypto.SSL.ssl_dispatcher import ssl_dispatcher
from M2Crypto.SSL.timeout import timeout, struct_to_timeout, struct_size

verify_none = m2.SSL_VERIFY_NONE  # type: int
verify_peer = m2.SSL_VERIFY_PEER  # type: int
verify_fail_if_no_peer_cert = m2.SSL_VERIFY_FAIL_IF_NO_PEER_CERT  # type: int
verify_client_once = m2.SSL_VERIFY_CLIENT_ONCE  # type: int
verify_crl_check_chain = m2.VERIFY_CRL_CHECK_CHAIN # type: int
verify_crl_check_leaf = m2.VERIFY_CRL_CHECK_LEAF # type: int

SSL_SENT_SHUTDOWN = m2.SSL_SENT_SHUTDOWN  # type: int
SSL_RECEIVED_SHUTDOWN = m2.SSL_RECEIVED_SHUTDOWN  # type: int

op_all = m2.SSL_OP_ALL  # type: int
op_no_sslv2 = m2.SSL_OP_NO_SSLv2  # type: int

"""
SSL peer certificate checking routines

Copyright (c) 2004-2007 Open Source Applications Foundation.
All rights reserved.

Copyright 2008 Heikki Toivonen. All rights reserved.
"""

__all__ = ['SSLVerificationError', 'NoCertificate', 'WrongCertificate',
           'WrongHost', 'Checker']

import re

from M2Crypto import X509, m2, six  # noqa

class SSLVerificationError(Exception):
    pass

class NoCertificate(SSLVerificationError):
    pass

class WrongCertificate(SSLVerificationError):
    pass

class WrongHost(SSLVerificationError):
    def __init__(self, expectedHost, actualHost, fieldName='commonName'):
        # type: (str, AnyStr, str) -> None
        """
        This exception will be raised if the certificate returned by the
        peer was issued for a different host than we tried to connect to.
        This could be due to a server misconfiguration or an active attack.

        :param expectedHost: The name of the host we expected to find in the
                             certificate.
        :param actualHost:   The name of the host we actually found in the
        :param fieldName:    The field name where we noticed the error. This
                             should be either 'commonName' or 'subjectAltName'.
        """
        if fieldName not in ('commonName', 'subjectAltName'):
                'Unknown fieldName, should be either commonName ' +
                'or subjectAltName')

        SSLVerificationError.__init__(self)
        self.expectedHost = expectedHost
        self.actualHost = actualHost
        self.fieldName = fieldName

        s = 'Peer certificate %s does not match host, expected %s, got %s' \
            % (self.fieldName, self.expectedHost, self.actualHost)
        return six.ensure_text(s)

class Checker(object):

    numericIpMatch = re.compile('^[0-9]+(\.[0-9]+)*$')

    def __init__(self, host=None, peerCertHash=None, peerCertDigest='sha1'):
        # type: (Optional[str], Optional[bytes], str) -> None
        if peerCertHash is not None:
            peerCertHash = six.ensure_binary(peerCertHash)
        self.fingerprint = peerCertHash
        self.digest = peerCertDigest  # type: str

    def __call__(self, peerCert, host=None):
        # type: (X509.X509, Optional[str]) -> bool
        if peerCert is None:
            raise NoCertificate('peer did not return certificate')

        if host is not None:
            self.host = host  # type: str

        if self.fingerprint:
            if self.digest not in ('sha1', 'md5'):
                raise ValueError('unsupported digest "%s"' % self.digest)

            if self.digest == 'sha1':
                expected_len = 40
            elif self.digest == 'md5':
                expected_len = 32
            else:
                raise ValueError('Unexpected digest {0}'.format(self.digest))

            if len(self.fingerprint) != expected_len:
                raise WrongCertificate(
                    ('peer certificate fingerprint length does not match\n' +
                     'fingerprint: {0}\nexpected = {1}\n' +
                     'observed = {2}').format(self.fingerprint,
                                              expected_len,
                                              len(self.fingerprint)))

            expected_fingerprint = six.ensure_text(self.fingerprint)
            observed_fingerprint = peerCert.get_fingerprint(md=self.digest)
            if observed_fingerprint != expected_fingerprint:
                    ('peer certificate fingerprint does not match\n' +
                     'expected = {0},\n' +
                     'observed = {1}').format(expected_fingerprint,
                                              observed_fingerprint))

            hostValidationPassed = False
            self.useSubjectAltNameOnly = False

            # subjectAltName=DNS:somehost[, ...]*
            try:
                subjectAltName = peerCert.get_ext('subjectAltName').get_value()
                if self._splitSubjectAltName(self.host, subjectAltName):
                    hostValidationPassed = True
                elif self.useSubjectAltNameOnly:
                    raise WrongHost(expectedHost=self.host,
                                    actualHost=subjectAltName,
                                    fieldName='subjectAltName')
            except LookupError:
                pass

            # commonName=somehost[, ...]*
            if not hostValidationPassed:
                hasCommonName = False
                commonNames = ''
                for entry in peerCert.get_subject().get_entries_by_nid(
                        m2.NID_commonName):
                    hasCommonName = True
                    commonName = entry.get_data().as_text()
                    if not commonNames:
                        commonNames = commonName
                    else:
                        commonNames += ',' + commonName
                    if self._match(self.host, commonName):
                        hostValidationPassed = True
                        break

                if not hasCommonName:
                    raise WrongCertificate('no commonName in peer certificate')

                if not hostValidationPassed:
                                    actualHost=commonNames,
                                    fieldName='commonName')

    def _splitSubjectAltName(self, host, subjectAltName):
        # type: (AnyStr, AnyStr) -> bool
        """
        >>> check = Checker()
        >>> check._splitSubjectAltName(host='my.example.com',
        ...     subjectAltName='DNS:my.example.com')
        True
        ...     subjectAltName='DNS:*.example.com')
        True
        ...     subjectAltName='DNS:m*.example.com')
        True
        ...     subjectAltName='DNS:m*ample.com')
        False
        >>> check.useSubjectAltNameOnly
        True
        ...     subjectAltName='DNS:m*ample.com, othername:<unsupported>')
        False
        ...     subjectAltName='DNS:m*ample.com, DNS:my.example.org')
        False
        ...     subjectAltName='DNS:m*ample.com, DNS:my.example.com')
        True
        ...     subjectAltName='DNS:my.example.com, DNS:my.example.org')
        True
        True
        ...     subjectAltName='')
        False
        ...     subjectAltName='othername:<unsupported>')
        False
        False
        """
        self.useSubjectAltNameOnly = False
        for certHost in subjectAltName.split(','):
            certHost = certHost.lower().strip()
            if certHost[:4] == 'dns:':
                self.useSubjectAltNameOnly = True
                if self._match(host, certHost[4:]):
                    return True
            elif certHost[:11] == 'ip address:':
                if self._matchIPAddress(host, certHost[11:]):

    def _match(self, host, certHost):
        # type: (str, str) -> bool
        """
        >>> check._match(host='my.example.com', certHost='my.example.com')
        True
        >>> check._match(host='my.example.com', certHost='*.example.com')
        True
        >>> check._match(host='my.example.com', certHost='m*.example.com')
        True
        >>> check._match(host='my.example.com', certHost='m*.EXAMPLE.com')
        True
        >>> check._match(host='my.example.com', certHost='m*ample.com')
        False
        >>> check._match(host='my.example.com', certHost='*.*.com')
        False
        >>> check._match(host='1.2.3.4', certHost='1.2.3.4')
        True
        >>> check._match(host='1.2.3.4', certHost='*.2.3.4')
        False
        >>> check._match(host='1234', certHost='1234')
        True
        """
        # XXX See RFC 2818 and 3280 for matching rules, this is may not
        # XXX yet be complete.

        host = host.lower()
        certHost = certHost.lower()

        if host == certHost:

        if certHost.count('*') > 1:
            # Not sure about this, but being conservative

        if self.numericIpMatch.match(host) or \
                self.numericIpMatch.match(certHost.replace('*', '')):
            # Not sure if * allowed in numeric IP, but think not.

        if certHost.find('\\') > -1:
            # Not sure about this, maybe some encoding might have these.
            # But being conservative for now, because regex below relies
            # on this.

        # Massage certHost so that it can be used in regex
        certHost = certHost.replace('.', '\.')
        certHost = certHost.replace('*', '[^\.]*')
        if re.compile('^%s$' % certHost).match(host):

    def _matchIPAddress(self, host, certHost):
        """
        >>> check._matchIPAddress(host='my.example.com',
        ...     certHost='my.example.com')
        False
        >>> check._matchIPAddress(host='1.2.3.4', certHost='1.2.3.4')
        True
        >>> check._matchIPAddress(host='1.2.3.4', certHost='*.2.3.4')
        False
        >>> check._matchIPAddress(host='1.2.3.4', certHost='1.2.3.40')
        False
        >>> check._matchIPAddress(host='::1', certHost='::1')
        True
        >>> check._matchIPAddress(host='::1', certHost='0:0:0:0:0:0:0:1')
        True
        >>> check._matchIPAddress(host='::1', certHost='::2')
        False
        """
        try:
            canonical = socket.getaddrinfo(host, 0, 0, socket.SOCK_STREAM, 0,
                                           socket.AI_NUMERICHOST)
            certCanonical = socket.getaddrinfo(certHost, 0, 0,
                                               socket.SOCK_STREAM, 0,
                                               socket.AI_NUMERICHOST)
        except:
        return canonical == certCanonical

if __name__ == '__main__':
    import doctest
    doctest.testmod()

"""SSL Ciphers

__all__ = ['Cipher', 'Cipher_Stack']

from typing import Iterable  # noqa

    def __init__(self, cipher):

        return m2.ssl_cipher_get_bits(self.cipher)

        return "%s-%s" % (self.name(), len(self))

    def version(self):
        return m2.ssl_cipher_get_version(self.cipher)

        return six.ensure_text(m2.ssl_cipher_get_name(self.cipher))

class Cipher_Stack(object):
    def __init__(self, stack):
        """
        :param stack: binary of the C-type STACK_OF(SSL_CIPHER)
        """
        self.stack = stack

        return m2.sk_ssl_cipher_num(self.stack)

        # type: (int) -> Cipher
        if not 0 <= idx < m2.sk_ssl_cipher_num(self.stack):
            raise IndexError('index out of range')
        v = m2.sk_ssl_cipher_value(self.stack, idx)
        return Cipher(v)

        # type: () -> Iterable
        for i in six.moves.range(m2.sk_ssl_cipher_num(self.stack)):

"""SSL Context

from M2Crypto import BIO, Err, RSA, X509, m2, util  # noqa
from M2Crypto.SSL import cb  # noqa
from M2Crypto.SSL.Session import Session  # noqa
from weakref import WeakValueDictionary
from typing import Any, AnyStr, Callable, Optional, Union  # noqa

__all__ = ['ctxmap', 'Context', 'map']

class _ctxmap(object):
    singleton = None  # type: Optional[_ctxmap]

        """Simple WeakReffed list.
        """
        self._ctxmap = WeakValueDictionary()

    def __getitem__(self, key):
        # type: (int) -> Any
        return self._ctxmap[key]

    def __setitem__(self, key, value):
        # type: (int, Any) -> None
        self._ctxmap[key] = value

    def __delitem__(self, key):
        del self._ctxmap[key]

def ctxmap():
    # type: () -> _ctxmap
    if _ctxmap.singleton is None:
        _ctxmap.singleton = _ctxmap()
    return _ctxmap.singleton
# deprecated!!!
map = ctxmap

class Context(object):

    """'Context' for SSL connections."""

    m2_ssl_ctx_free = m2.ssl_ctx_free

    def __init__(self, protocol='tls', weak_crypto=None,
                 post_connection_check=None):
        # type: (str, Optional[int], Optional[Callable]) -> None
        proto = getattr(m2, protocol + '_method', None)
        if proto is None:
            # default is 'sslv23' for older versions of OpenSSL
            if protocol == 'tls':
                proto = getattr(m2, 'sslv23_method')
            else:
                raise ValueError("no such protocol '%s'" % protocol)
        self.ctx = m2.ssl_ctx_new(proto())
        self.allow_unknown_ca = 0  # type: Union[int, bool]
        self.post_connection_check = post_connection_check
        ctxmap()[int(self.ctx)] = self
        m2.ssl_ctx_set_cache_size(self.ctx, 128)
        if weak_crypto is None and protocol in ('sslv23', 'tls'):
            self.set_options(m2.SSL_OP_ALL | m2.SSL_OP_NO_SSLv2 |
                             m2.SSL_OP_NO_SSLv3)

            self.m2_ssl_ctx_free(self.ctx)

        del ctxmap()[int(self.ctx)]

    def load_cert(self, certfile, keyfile=None,
                  callback=util.passphrase_callback):
        """Load certificate and private key into the context.

        :param certfile: File that contains the PEM-encoded certificate.
        :param keyfile:  File that contains the PEM-encoded private key.
                         Default value of None indicates that the private key
                         is to be found in 'certfile'.
        :param callback: Callable object to be invoked if the private key is
                         passphrase-protected. Default callback provides a
                         simple terminal-style input for the passphrase.
        """
        m2.ssl_ctx_passphrase_callback(self.ctx, callback)
        m2.ssl_ctx_use_cert(self.ctx, certfile)
        if not keyfile:
            keyfile = certfile
        m2.ssl_ctx_use_privkey(self.ctx, keyfile)
        if not m2.ssl_ctx_check_privkey(self.ctx):
            raise ValueError('public/private key mismatch')

    def load_cert_chain(self, certchainfile, keyfile=None,
                        callback=util.passphrase_callback):
        """Load certificate chain and private key into the context.

        :param certchainfile: File object containing the PEM-encoded
                              certificate chain.
        :param keyfile:       File object containing the PEM-encoded private
                              key. Default value of None indicates that the
                              private key is to be found in 'certchainfile'.
        :param callback:      Callable object to be invoked if the private key
                              is passphrase-protected. Default callback
                              provides a simple terminal-style input for the
                              passphrase.
        """
        m2.ssl_ctx_use_cert_chain(self.ctx, certchainfile)
            keyfile = certchainfile

    def set_client_CA_list_from_file(self, cafile):
        # type: (AnyStr) -> None
        """Load CA certs into the context. These CA certs are sent to the
        peer during *SSLv3 certificate request*.

        :param cafile: File object containing one or more PEM-encoded CA
                       certificates concatenated together.
        """
        m2.ssl_ctx_set_client_CA_list_from_file(self.ctx, cafile)

    load_client_CA = load_client_ca = set_client_CA_list_from_file

    def load_verify_locations(self, cafile=None, capath=None):
        # type: (Optional[AnyStr], Optional[AnyStr]) -> int
        """Load CA certs into the context.

        These CA certs are used during verification of the peer's
        certificate.

        :param cafile: File containing one or more PEM-encoded CA

        :param capath: Directory containing PEM-encoded CA certificates
                       (one certificate per file).

        :return: 0 if the operation failed because CAfile and CApath are NULL
                  or the processing at one of the locations specified failed.
                  Check the error stack to find out the reason.

                1 The operation succeeded.
        """
        if cafile is None and capath is None:
            raise ValueError("cafile and capath can not both be None.")
        return m2.ssl_ctx_load_verify_locations(self.ctx, cafile, capath)

    load_verify_info = load_verify_locations

    def set_session_id_ctx(self, id):
        """Sets the session id for the SSL.Context w/in a session can be reused.

        :param id: Sessions are generated within a certain context. When
                   exporting/importing sessions with
                   i2d_SSL_SESSION/d2i_SSL_SESSION it would be possible,
                   to re-import a session generated from another context
                   (e.g. another application), which might lead to
                   malfunctions. Therefore each application must set its
                   own session id context sid_ctx which is used to
                   distinguish the contexts and is stored in exported
                   sessions. The sid_ctx can be any kind of binary data
                   with a given length, it is therefore possible to use
                   e.g. the name of the application and/or the hostname
                   and/or service name.
        """
        ret = m2.ssl_ctx_set_session_id_context(self.ctx, id)
        if not ret:
            raise Err.SSLError(Err.get_error_code(), '')

    def set_default_verify_paths(self):
        """
        Specifies that the default locations from which CA certs are
        loaded should be used.

        There is one default directory and one default file. The default
        CA certificates directory is called "certs" in the default
        OpenSSL directory. Alternatively the SSL_CERT_DIR environment
        variable can be defined to override this location. The default
        CA certificates file is called "cert.pem" in the default OpenSSL
        directory. Alternatively the SSL_CERT_FILE environment variable
        can be defined to override this location.

        @return 0 if the operation failed. A missing default location is
                  still treated as a success. No error code is set.

        """
        ret = m2.ssl_ctx_set_default_verify_paths(self.ctx)
        if not ret:
            raise ValueError('Cannot use default SSL certificate store!')

    def set_allow_unknown_ca(self, ok):
        # type: (Union[int, bool]) -> None
        """Set the context to accept/reject a peer certificate if the
        certificate's CA is unknown.

        :param ok:       True to accept, False to reject.
        """
        self.allow_unknown_ca = ok

    def get_allow_unknown_ca(self):
        # type: () -> Union[int, bool]
        """Get the context's setting that accepts/rejects a peer
        certificate if the certificate's CA is unknown.

        FIXME 2Bconverted to bool
        """
        return self.allow_unknown_ca

    def set_verify(self, mode, depth, callback=None):
        # type: (int, int, Optional[Callable]) -> None
        """
        Set verify options. Most applications will need to call this
        method with the right options to make a secure SSL connection.

        :param mode:     The verification mode to use. Typically at least
                         SSL.verify_peer is used. Clients would also typically
                         add SSL.verify_fail_if_no_peer_cert.
        :param depth:    The maximum allowed depth of the certificate chain
                         returned by the peer.
        :param callback: Callable that can be used to specify custom
                         verification checks.
        """
            m2.ssl_ctx_set_verify_default(self.ctx, mode)
        else:
            m2.ssl_ctx_set_verify(self.ctx, mode, callback)
        m2.ssl_ctx_set_verify_depth(self.ctx, depth)

    def get_verify_mode(self):
        return m2.ssl_ctx_get_verify_mode(self.ctx)

    def get_verify_depth(self):
        """Returns the verification mode currently set in the SSL Context."""
        return m2.ssl_ctx_get_verify_depth(self.ctx)

    def set_tmp_dh(self, dhpfile):
        """Load ephemeral DH parameters into the context.

        :param dhpfile: Filename of the file containing the PEM-encoded
                        DH parameters.
        """
        f = BIO.openfile(dhpfile)
        dhp = m2.dh_read_parameters(f.bio_ptr())
        return m2.ssl_ctx_set_tmp_dh(self.ctx, dhp)

    def set_tmp_dh_callback(self, callback=None):
        # type: (Optional[Callable]) -> None
        """Sets the callback function for SSL.Context.

        :param callback: Callable to be used when a DH parameters are required.
        """
        if callback is not None:
            m2.ssl_ctx_set_tmp_dh_callback(self.ctx, callback)

    def set_tmp_rsa(self, rsa):
        # type: (RSA.RSA) -> int
        """Load ephemeral RSA key into the context.

        :param rsa: RSA.RSA instance.
        """
        if isinstance(rsa, RSA.RSA):
            return m2.ssl_ctx_set_tmp_rsa(self.ctx, rsa.rsa)
        else:
            raise TypeError("Expected an instance of RSA.RSA, got %s." % rsa)

    def set_tmp_rsa_callback(self, callback=None):
        """Sets the callback function to be used when
        a temporary/ephemeral RSA key is required.
        """
            m2.ssl_ctx_set_tmp_rsa_callback(self.ctx, callback)

    def set_info_callback(self, callback=cb.ssl_info_callback):
        # type: (Callable) -> None
        """Set a callback function to get state information.

        It can be used to get state information about the SSL
        connections that are created from this context.

        :param callback: Callback function. The default prints
                         information to stderr.
        """
        m2.ssl_ctx_set_info_callback(self.ctx, callback)

    def set_cipher_list(self, cipher_list):
        # type: (str) -> int
        """Sets the list of available ciphers.

        :param cipher_list: The format of the string is described in
                            ciphers(1).
        :return: 1 if any cipher could be selected and 0 on complete
                 failure.
        """
        return m2.ssl_ctx_set_cipher_list(self.ctx, cipher_list)

    def add_session(self, session):
        # type: (Session) -> int
        """Add the session to the context.

        :param session: the session to be added.

        :return: 0 The operation failed. It was tried to add the same
                   (identical) session twice.

                 1 The operation succeeded.
        """
        return m2.ssl_ctx_add_session(self.ctx, session._ptr())

    def remove_session(self, session):
        """Remove the session from the context.

        :param session: the session to be removed.

        :return: 0 The operation failed. The session was not found in
                   the cache.

        """
        return m2.ssl_ctx_remove_session(self.ctx, session._ptr())

    def get_session_timeout(self):
        """Get current session timeout.

        Whenever a new session is created, it is assigned a maximum
        lifetime.  This lifetime is specified by storing the creation
        time of the session and the timeout value valid at this time. If
        the actual time is later than creation time plus timeout, the
        session is not reused.

        Due to this realization, all sessions behave according to the
        timeout value valid at the time of the session negotiation.
        Changes of the timeout value do not affect already established
        sessions.

        Expired sessions are removed from the internal session cache,
        whenever SSL_CTX_flush_sessions(3) is called, either directly by
        the application or automatically (see
        SSL_CTX_set_session_cache_mode(3))

        The default value for session timeout is decided on a per
        protocol basis, see SSL_get_default_timeout(3).  All currently
        supported protocols have the same default timeout value of 300
        seconds.

        SSL_CTX_set_timeout() returns the previously set timeout value.

        :return: the currently set timeout value.
        """
        return m2.ssl_ctx_get_session_timeout(self.ctx)

    def set_session_timeout(self, timeout):
        """Set new session timeout.

        See self.get_session_timeout() for explanation of the session
        timeouts.

        :param timeout: new timeout value.

        :return: the previously set timeout value.
        """
        return m2.ssl_ctx_set_session_timeout(self.ctx, timeout)

    def set_session_cache_mode(self, mode):
        """Enables/disables session caching.

        The mode is set by using m2.SSL_SESS_CACHE_* constants.

        :param mode: new mode value.

        :return: the previously set cache mode value.
        """
        return m2.ssl_ctx_set_session_cache_mode(self.ctx, mode)

    def get_session_cache_mode(self):
        """Gets the current session caching.

        The mode is set to m2.SSL_SESS_CACHE_* constants.

        """
        return m2.ssl_ctx_get_session_cache_mode(self.ctx)

    def set_options(self, op):
        """Adds the options set via bitmask in options to the Context.

        !!! Options already set before are not cleared!

        The behaviour of the SSL library can be changed by setting
        several options.  The options are coded as bitmasks and can be
        combined by a logical or operation (|).

        SSL.Context.set_options() and SSL.set_options() affect the
        (external) protocol behaviour of the SSL library. The (internal)
        behaviour of the API can be changed by using the similar
        SSL.Context.set_mode() and SSL.set_mode() functions.

        During a handshake, the option settings of the SSL object are
        used. When a new SSL object is created from a context using
        SSL(), the current option setting is copied. Changes to ctx
        do not affect already created SSL objects. SSL.clear() does not
        affect the settings.

        :param op: bitmask of additional options specified in
                   SSL_CTX_set_options(3) manpage.

        :return: the new options bitmask after adding options.
        """
        return m2.ssl_ctx_set_options(self.ctx, op)

    def get_cert_store(self):
        # type: () -> X509.X509
        """
        Get the certificate store associated with this context.

        :warning: The store is NOT refcounted, and as such can not be relied
                  to be valid once the context goes away or is changed.
        """
        return X509.X509_Store(m2.ssl_ctx_get_cert_store(self.ctx))

"""SSLServer

# M2Crypto
from M2Crypto.SSL import SSLError
from M2Crypto.SSL.Context import Context  # noqa
# from M2Crypto import six  # noqa
from M2Crypto import util  # noqa
from M2Crypto.six.moves.socketserver import (BaseRequestHandler, BaseServer,
                                             TCPServer, ThreadingMixIn)
import os
    from M2Crypto.six.moves.socketserver import ForkingMixIn
from socket import socket  # noqa
from typing import Union  # noqa

__all__ = ['SSLServer', 'ForkingSSLServer', 'ThreadingSSLServer']

class SSLServer(TCPServer):
    def __init__(self, server_address, RequestHandlerClass, ssl_context,  # noqa
                 bind_and_activate=True):
        # type: (util.AddrType, BaseRequestHandler, Context, bool) -> None
        """
        Superclass says: Constructor. May be extended, do not override.
        This class says: Ho-hum.
        """
        BaseServer.__init__(self, server_address, RequestHandlerClass)
        self.ssl_ctx = ssl_context
        self.socket = Connection(self.ssl_ctx)
        if bind_and_activate:
            self.server_bind()
            self.server_activate()

    def handle_request(self):
        request = None
        client_address = None
        try:
            request, client_address = self.get_request()
            if self.verify_request(request, client_address):
                self.process_request(request, client_address)
        except SSLError:
            self.handle_error(request, client_address)

    def handle_error(self, request, client_address):
        # type: (Union[socket, Connection], util.AddrType) -> None
        print('-' * 40)
        import traceback
        traceback.print_exc()

class ThreadingSSLServer(ThreadingMixIn, SSLServer):
    pass

    class ForkingSSLServer(ForkingMixIn, SSLServer):
        pass

"""SSL Session

__all__ = ['Session', 'load_session']

from M2Crypto import BIO, Err, m2
from typing import AnyStr  # noqa

class Session(object):

    m2_ssl_session_free = m2.ssl_session_free

    def __init__(self, session, _pyfree=0):
        assert session is not None

            self.m2_ssl_session_free(self.session)

        return self.session

        m2.ssl_session_print(buf.bio_ptr(), self.session)

        m2.i2d_ssl_session(buf.bio_ptr(), self.session)

    def write_bio(self, bio):
        return m2.ssl_session_write_bio(bio.bio_ptr(), self.session)

    def get_time(self):
        return m2.ssl_session_get_time(self.session)

    def set_time(self, t):
        return m2.ssl_session_set_time(self.session, t)

    def get_timeout(self):
        return m2.ssl_session_get_timeout(self.session)

    def set_timeout(self, t):
        return m2.ssl_session_set_timeout(self.session, t)

def load_session(pemfile):
    # type: (AnyStr) -> Session
    with BIO.openfile(pemfile) as f:
        cptr = m2.ssl_session_read_pem(f.bio_ptr())

    return Session(cptr, 1)

"""
Make Twisted use M2Crypto for SSL

FIXME THIS HAS NOT BEEN FINISHED. NEITHER PEP484 NOR PORT PYTHON3 HAS
BEEN FINISHED. THE FURTHER WORK WILL BE DONE WHEN THE STATUS OF TWISTED
IN THE PYTHON 3 (AND ASYNCIO) WORLD WILL BE CLEAR.
"""

__all__ = ['connectSSL', 'connectTCP', 'listenSSL', 'listenTCP',
           'TLSProtocolWrapper']

from functools import partial

import twisted.internet.reactor
import twisted.protocols.policies as policies

from M2Crypto import BIO, X509, m2, util
from M2Crypto.SSL.Checker import Checker, SSLVerificationError

from twisted.internet.interfaces import ITLSTransport
from twisted.protocols.policies import ProtocolWrapper
from typing import AnyStr, Callable, Iterable, Optional  # noqa
from zope.interface import implementer

def _alwaysSucceedsPostConnectionCheck(peerX509, expectedHost):
    return 1

def connectSSL(host, port, factory, contextFactory, timeout=30,
               bindAddress=None,
               reactor=twisted.internet.reactor,
               postConnectionCheck=Checker()):
    # type: (str, int, object, object, int, Optional[str], twisted.internet.reactor, Checker) -> reactor.connectTCP
    """
    A convenience function to start an SSL/TLS connection using Twisted.

    See IReactorSSL interface in Twisted.
    """
    wrappingFactory = policies.WrappingFactory(factory)
    wrappingFactory.protocol = lambda factory, wrappedProtocol: \
        TLSProtocolWrapper(factory,
                           wrappedProtocol,
                           startPassThrough=0,
                           client=1,
                           contextFactory=contextFactory,
                           postConnectionCheck=postConnectionCheck)
    return reactor.connectTCP(host, port, wrappingFactory, timeout, bindAddress)

def connectTCP(host, port, factory, timeout=30, bindAddress=None,
    # type: (str, int, object, int, Optional[util.AddrType], object, Callable) -> object
    """
    A convenience function to start a TCP connection using Twisted.

    NOTE: You must call startTLS(ctx) to go into SSL/TLS mode.

    See IReactorTCP interface in Twisted.
    """
                           startPassThrough=1,
                           client=1,
                           contextFactory=None,

def listenSSL(port, factory, contextFactory, backlog=5, interface='',
              reactor=twisted.internet.reactor,
              postConnectionCheck=_alwaysSucceedsPostConnectionCheck):
    """
    A convenience function to listen for SSL/TLS connections using Twisted.

    """
                           client=0,
    return reactor.listenTCP(port, wrappingFactory, backlog, interface)

def listenTCP(port, factory, backlog=5, interface='',
              postConnectionCheck=None):
    """
    A convenience function to listen for TCP connections using Twisted.

    """
                           client=0,

class _BioProxy(object):
    """
    The purpose of this class is to eliminate the __del__ method from
    TLSProtocolWrapper, and thus letting it be garbage collected.
    """

    m2_bio_free_all = m2.bio_free_all

    def __init__(self, bio):

        if self.bio is not None:
            self.m2_bio_free_all(self.bio)

class _SSLProxy(object):
    """
    """

    m2_ssl_free = m2.ssl_free

    def __init__(self, ssl):
        self.ssl = ssl

        return self.ssl

        if self.ssl is not None:
            self.m2_ssl_free(self.ssl)

@implementer(ITLSTransport)
class TLSProtocolWrapper(ProtocolWrapper):
    """
    A SSL/TLS protocol wrapper to be used with Twisted. Typically
    you would not use this class directly. Use connectTCP,
    connectSSL, listenTCP, listenSSL functions defined above,
    which will hook in this class.
    """

    def __init__(self, factory, wrappedProtocol, startPassThrough, client,
                 contextFactory, postConnectionCheck):
        # type: (policies.WrappingFactory, object, int, int, object, Checker) -> None
        """
        :param factory:
        :param wrappedProtocol:
        :param startPassThrough:    If true we won't encrypt at all. Need to
                                    call startTLS() later to switch to SSL/TLS.
        :param client:              True if this should be a client protocol.
        :param contextFactory:      Factory that creates SSL.Context objects.
                                    The called function is getContext().
        :param postConnectionCheck: The post connection check callback that
                                    will be called just after connection has
                                    been established but before any real data
                                    has been exchanged. The first argument to
                                    this function is an X509 object, the second
                                    is the expected host name string.
        """
        # ProtocolWrapper.__init__(self, factory, wrappedProtocol)
        # XXX: Twisted 2.0 has a new addition where the wrappingFactory is
        #      set as the factory of the wrappedProtocol. This is an issue
        #      as the wrap should be transparent. What we want is
        #      the factory of the wrappedProtocol to be the wrappedFactory and
        #      not the outer wrappingFactory. This is how it was implemented in
        #      Twisted 1.3
        self.factory = factory
        self.wrappedProtocol = wrappedProtocol

        # wrappedProtocol == client/server instance
        # factory.wrappedFactory == client/server factory

        self.data = b''  # Clear text to encrypt and send
        self.encrypted = b''  # Encrypted data we need to decrypt and pass on
        self.tlsStarted = 0  # SSL/TLS mode or pass through
        self.checked = 0  # Post connection check done or not
        self.isClient = client
        self.helloDone = 0  # True when hello has been sent
        if postConnectionCheck is None:
            self.postConnectionCheck = _alwaysSucceedsPostConnectionCheck
        else:
            self.postConnectionCheck = postConnectionCheck

        if not startPassThrough:
            self.startTLS(contextFactory.getContext())

    def clear(self):
        """
        Clear this instance, after which it is ready for reuse.
        """
        if getattr(self, 'tlsStarted', 0):
            self.sslBio = None
            self.ssl = None
            self.internalBio = None
            self.networkBio = None
        self.data = b''
        self.encrypted = b''
        self.tlsStarted = 0
        self.checked = 0
        self.isClient = 1
        self.helloDone = 0
        # We can reuse self.ctx and it will be deleted automatically
        # when this instance dies

    def startTLS(self, ctx):
        """
        Start SSL/TLS. If this is not called, this instance just passes data
        through untouched.
        """
        # NOTE: This method signature must match the startTLS() method Twisted
        #       expects transports to have. This will be called automatically
        #       by Twisted in STARTTLS situations, for example with SMTP.
        if self.tlsStarted:
            raise Exception('TLS already started')

        self.ctx = ctx

        self.internalBio = m2.bio_new(m2.bio_s_bio())
        m2.bio_set_write_buf_size(self.internalBio, 0)
        self.networkBio = _BioProxy(m2.bio_new(m2.bio_s_bio()))
        m2.bio_set_write_buf_size(self.networkBio._ptr(), 0)
        m2.bio_make_bio_pair(self.internalBio, self.networkBio._ptr())

        self.sslBio = _BioProxy(m2.bio_new(m2.bio_f_ssl()))

        self.ssl = _SSLProxy(m2.ssl_new(self.ctx.ctx))

        if self.isClient:
            m2.ssl_set_connect_state(self.ssl._ptr())
        else:
            m2.ssl_set_accept_state(self.ssl._ptr())

        m2.ssl_set_bio(self.ssl._ptr(), self.internalBio, self.internalBio)
        m2.bio_set_ssl(self.sslBio._ptr(), self.ssl._ptr(), m2.bio_noclose)

        # Need this for writes that are larger than BIO pair buffers
        mode = m2.ssl_get_mode(self.ssl._ptr())
        m2.ssl_set_mode(self.ssl._ptr(),
                        mode |
                        m2.SSL_MODE_ENABLE_PARTIAL_WRITE |
                        m2.SSL_MODE_ACCEPT_MOVING_WRITE_BUFFER)

        self.tlsStarted = 1

        if not self.tlsStarted:
            ProtocolWrapper.write(self, data)
            return

        try:
            encryptedData = self._encrypt(data)
            ProtocolWrapper.write(self, encryptedData)
            self.helloDone = 1
        except BIO.BIOError as e:
            # See http://www.openssl.org/docs/apps/verify.html#DIAGNOSTICS
            # for the error codes returned by SSL_get_verify_result.
            e.args = (m2.ssl_get_verify_result(self.ssl._ptr()), e.args[0])
            raise e

    def writeSequence(self, data):
        # type: (Iterable[bytes]) -> None
            ProtocolWrapper.writeSequence(self, b''.join(data))
            return

        self.write(b''.join(data))

    def loseConnection(self):
        # XXX Do we need to do m2.ssl_shutdown(self.ssl._ptr())?
        ProtocolWrapper.loseConnection(self)

    def connectionMade(self):
        ProtocolWrapper.connectionMade(self)
        if self.tlsStarted and self.isClient and not self.helloDone:
            self._clientHello()

    def dataReceived(self, data):
            ProtocolWrapper.dataReceived(self, data)
            return

        self.encrypted += data

        try:
            while 1:
                decryptedData = self._decrypt()

                self._check()

                encryptedData = self._encrypt()
                ProtocolWrapper.write(self, encryptedData)

                ProtocolWrapper.dataReceived(self, decryptedData)

                if decryptedData == b'' and encryptedData == b'':
                    break
            raise e

    def connectionLost(self, reason):
        self.clear()
        ProtocolWrapper.connectionLost(self, reason)

    def _check(self):
        if not self.checked and m2.ssl_is_init_finished(self.ssl._ptr()):
            x509 = m2.ssl_get_peer_cert(self.ssl._ptr())
            if x509 is not None:
                x509 = X509.X509(x509, 1)
            if self.isClient:
                host = self.transport.addr[0]
            else:
                host = self.transport.getPeer().host
            if not self.postConnectionCheck(x509, host):
                raise SSLVerificationError('post connection check')
            self.checked = 1

    def _clientHello(self):
        try:
            # We rely on OpenSSL implicitly starting with client hello
            # when we haven't yet established an SSL connection
            encryptedData = self._encrypt(clientHello=1)
            raise e

    # Optimizations to reduce attribute accesses

    @property
    def _get_wr_guar_ssl(self):
        # type: () -> Callable[[], int]
        """Return max. length of data can be written to the BIO.

        Writes larger than this value will return a value from
        BIO_write() less than the amount requested or if the buffer is
        full request a retry.
        """
        return partial(m2.bio_ctrl_get_write_guarantee,
                       self.sslBio._ptr())

    @property
    def _get_wr_guar_net(self):
                       self.networkBio._ptr())

    @property
    def _shoud_retry_ssl(self):
        # BIO_should_retry() is true if the call that produced this
        # condition should then be retried at a later time.
        return partial(m2.bio_should_retry, self.sslBio._ptr())

    @property
    def _shoud_retry_net(self):
        return partial(m2.bio_should_retry, self.networkBio._ptr())

    @property
    def _ctrl_pend_ssl(self):
        # size_t BIO_ctrl_pending(BIO *b);
        # BIO_ctrl_pending() return the number of pending characters in
        # the BIOs read and write buffers.
        return partial(m2.bio_ctrl_pending, self.sslBio._ptr())

    @property
    def _ctrl_pend_net(self):
        return partial(m2.bio_ctrl_pending, self.networkBio._ptr())

    @property
    def _write_ssl(self):
        # type: () -> Callable[[bytes], int]
        # All these functions return either the amount of data
        # successfully read or written (if the return value is
        # positive) or that no data was successfully read or written
        # if the result is 0 or -1. If the return value is -2 then
        # the operation is not implemented in the specific BIO type.
        return partial(m2.bio_write, self.sslBio._ptr())

    @property
    def _write_net(self):
        return partial(m2.bio_write, self.networkBio._ptr())

    @property
    def _read_ssl(self):
        # type: () -> Callable[[int], Optional[bytes]]
        return partial(m2.bio_read, self.sslBio._ptr())

    @property
    def _read_net(self):
        return partial(m2.bio_read, self.networkBio._ptr())

    def _encrypt(self, data=b'', clientHello=0):
        """
        :param data:
        :param clientHello:
        :return:
        """
        encryptedData = b''
        self.data += data

        while 1:
            if (self._get_wr_guar_ssl() > 0 and self.data != b'') or clientHello:
                r = self._write_ssl(self.data)
                if r <= 0:
                    if not self._shoud_retry_ssl():
                        raise IOError(
                            ('Data left to be written to {}, ' +
                             'but cannot retry SSL connection!').format(self.sslBio))
                else:
                    assert self.checked
                    self.data = self.data[r:]

            pending = self._ctrl_pend_net()
            if pending:
                d = self._read_net(pending)
                if d is not None:  # This is strange, but d can be None
                    encryptedData += d
                else:
                    assert(self._shoud_retry_net())
            else:
                break
        return encryptedData

    def _decrypt(self, data=b''):
        decryptedData = b''

        while 1:
            if self._get_wr_guar_ssl() > 0 and self.encrypted != b'':
                r = self._write_net(self.encrypted)
                if r <= 0:
                    if not self._shoud_retry_net():
                             'but cannot retry SSL connection!').format(self.networkBio))
                else:
                    self.encrypted = self.encrypted[r:]

            pending = self._ctrl_pend_ssl()
                d = self._read_ssl(pending)
                    decryptedData += d
                else:
                    assert(self._shoud_retry_ssl())
            else:
                break

        return decryptedData

"""SSL callbacks

import sys

from typing import Any  # noqa

__all__ = ['unknown_issuer', 'ssl_verify_callback_stub', 'ssl_verify_callback',
           'ssl_verify_callback_allow_unknown_ca', 'ssl_info_callback']

def ssl_verify_callback_stub(ssl_ctx_ptr, x509_ptr, errnum, errdepth, ok):
    return ok

unknown_issuer = [
    m2.X509_V_ERR_DEPTH_ZERO_SELF_SIGNED_CERT,
    m2.X509_V_ERR_UNABLE_TO_GET_ISSUER_CERT_LOCALLY,
    m2.X509_V_ERR_UNABLE_TO_VERIFY_LEAF_SIGNATURE,
    m2.X509_V_ERR_CERT_UNTRUSTED,
]

def ssl_verify_callback(ssl_ctx_ptr, x509_ptr, errnum, errdepth, ok):
    # type: (bytes, bytes, int, int, int) -> int

    from M2Crypto.SSL.Context import Context
    ssl_ctx = Context.ctxmap()[int(ssl_ctx_ptr)]
    if errnum in unknown_issuer:
        if ssl_ctx.get_allow_unknown_ca():
            sys.stderr.write("policy: %s: permitted...\n" %
                             (m2.x509_get_verify_error(errnum)))
            sys.stderr.flush()
            ok = 1
    # CRL checking goes here...
    if ok:
        if ssl_ctx.get_verify_depth() >= errdepth:
            ok = 1
        else:
            ok = 0
    return ok

def ssl_verify_callback_allow_unknown_ca(ok, store):
    # type: (int, Any) -> int
    errnum = store.get_error()
        ok = 1
    return ok

# Cribbed from OpenSSL's apps/s_cb.c.
def ssl_info_callback(where, ret, ssl_ptr):
    # type: (int, int, bytes) -> None

    w = where & ~m2.SSL_ST_MASK
    if w & m2.SSL_ST_CONNECT:
        state = "SSL connect"
    elif w & m2.SSL_ST_ACCEPT:
        state = "SSL accept"
    else:
        state = "SSL state unknown"

    if where & m2.SSL_CB_LOOP:
        sys.stderr.write("LOOP: %s: %s\n" %
                         (state, m2.ssl_get_state_v(ssl_ptr)))
        sys.stderr.flush()
        return

    if where & m2.SSL_CB_EXIT:
        if not ret:
            sys.stderr.write("FAILED: %s: %s\n" %
                             (state, m2.ssl_get_state_v(ssl_ptr)))
        else:
            sys.stderr.write("INFO: %s: %s\n" %
        return

    if where & m2.SSL_CB_ALERT:
        if where & m2.SSL_CB_READ:
            w = 'read'
        else:
            w = 'write'
        sys.stderr.write("ALERT: %s: %s: %s\n" %
                         (w, m2.ssl_get_alert_type_v(ret),
                          m2.ssl_get_alert_desc_v(ret)))
        return

"""SSL dispatcher

# Python
import asyncore

# M2Crypto

__all__ = ['ssl_dispatcher']

class ssl_dispatcher(asyncore.dispatcher):

    def create_socket(self, ssl_context):
        # type: (Context) -> None
        self.family_and_type = socket.AF_INET, socket.SOCK_STREAM
        # self.socket.setblocking(0)
        self.add_channel()

    def connect(self, addr):
        # type: (util.AddrType) -> None
        self.socket.setblocking(1)
        self.socket.connect(addr)
        self.socket.setblocking(0)

    def recv(self, buffer_size=4096):
        """Receive data over SSL."""
        return self.socket.recv(buffer_size)

    def send(self, buffer):
        """Send data over SSL."""
        return self.socket.send(buffer)

"""Support for SSL socket timeouts.

"""

__all__ = ['DEFAULT_TIMEOUT', 'timeout', 'struct_to_timeout', 'struct_size']

import sys
import struct

DEFAULT_TIMEOUT = 600  # type: int

class timeout(object):

    def __init__(self, sec=DEFAULT_TIMEOUT, microsec=0):
        # type: (int, int) -> None
        self.sec = sec
        self.microsec = microsec

    def pack(self):
        if sys.platform == 'win32':
            millisec = int(self.sec * 1000 + round(float(self.microsec) / 1000))
            binstr = struct.pack('l', millisec)
        else:
            binstr = struct.pack('ll', self.sec, self.microsec)
        return binstr

def struct_to_timeout(binstr):
    # type: (bytes) -> timeout
    if sys.platform == 'win32':
        millisec = struct.unpack('l', binstr)[0]
        # On py3, int/int performs exact division and returns float. We want
        # the whole number portion of the exact division result:
        sec = int(millisec / 1000)
        microsec = (millisec % 1000) * 1000
    else:
        (sec, microsec) = struct.unpack('ll', binstr)
    return timeout(sec, microsec)

def struct_size():
        return struct.calcsize('l')
    else:
        return struct.calcsize('ll')