M2Crypto.SSL.Context — M2Crypto documentation

# Source code for M2Crypto.SSL.Context

    from __future__ import absolute_import

    """SSL Context

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

    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]

        def __init__(self):
            # type: () -> None
            """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):
            # type: (int) -> None
            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)

        def __del__(self):
            # type: () -> None
            if getattr(self, 'ctx', None):
                self.m2_ssl_ctx_free(self.ctx)

        def close(self):
            # type: () -> None
            del ctxmap()[int(self.ctx)]

        def load_cert(self, certfile, keyfile=None,
                      callback=util.passphrase_callback):
            # type: (AnyStr, Optional[AnyStr], Callable) -> None
            """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):
            # type: (AnyStr, Optional[AnyStr], Callable) -> None
            """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_passphrase_callback(self.ctx, callback)
            m2.ssl_ctx_use_cert_chain(self.ctx, certchainfile)
            if not keyfile:
                keyfile = certchainfile
            m2.ssl_ctx_use_privkey(self.ctx, keyfile)
            if not m2.ssl_ctx_check_privkey(self.ctx):
                raise ValueError('public/private key mismatch')

        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)

        # Deprecated.
        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
                           certificates concatenated together.

            :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)

        # Deprecated.
        load_verify_info = load_verify_locations

        def set_session_id_ctx(self, id):
            # type: (bytes) -> None
            """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):
            # type: () -> int
            """
            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.

                    1 The operation succeeded.
            """
            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.
            """
            if callback is None:
                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):
            # type: () -> int
            return m2.ssl_ctx_get_verify_mode(self.ctx)

        def get_verify_depth(self):
            # type: () -> int
            """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):
            # type: (AnyStr) -> int
            """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):
            # type: (Optional[Callable]) -> None
            """Sets the callback function to be used when
            a temporary/ephemeral RSA key is required.
            """
            if callback is not None:
                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):
            # type: (Session) -> int
            """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.

                     1 The operation succeeded.
            """
            return m2.ssl_ctx_remove_session(self.ctx, session._ptr())

        def get_session_timeout(self):
            # type: () -> int
            """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):
            # type: (int) -> int
            """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):
            # type: (int) -> int
            """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):
            # type: () -> int
            """Gets the current session caching.

            The mode is set to m2.SSL_SESS_CACHE_* constants.

            :return: the previously set cache mode value.
            """
            return m2.ssl_ctx_get_session_cache_mode(self.ctx)

        def set_options(self, op):
            # type: (int) -> int
            """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))