requests.ssl_session

requests_ssl_session.py

from __future__ import absolute_import
import ssl

from requests import session
from requests.adapters import HTTPAdapter, DEFAULT_POOLBLOCK
from requests.packages.urllib3.poolmanager import PoolManager, SSL_KEYWORDS


try:
    # python >=2.7.9
    default_ssl_version = ssl.PROTOCOL_TLSv1_2
except AttributeError:
    # python <2.7.9
    default_ssl_version = ssl.PROTOCOL_TLSv1


def ssl_session(ssl_version=default_ssl_version):
    """
    A session object that has a pre-configured SSL transport adapter.

        ```
        url = 'https://example.com'
        tls_response = ssl_session().get(url)
        ssl_response = ssl_session(ssl_version=ssl.PROTOCOL_SSLv23).get(url)
        ```

    """
    s = session()
    s.mount('https://', SSLAdapter(ssl_version=ssl_version))
    return s


class SSLAdapter(HTTPAdapter):
    """
    An HTTPS Transport Adapter that uses an arbitrary SSL version.

        ```
        adapter_tls = SSLAdapter()
        adapter_ssl = SSLAdapter(ssl_version=ssl.PROTOCOL_SSLv23)
        ```

    """
    def __init__(self, ssl_version=default_ssl_version, **kwargs):
        self._set_ssl_keywords(ssl_version=ssl_version, **kwargs)
        super(SSLAdapter, self).__init__(**kwargs)

    def _set_ssl_keywords(self, **kwargs):
        """
        Preconfigure the SSL keywords.

            ```
            SSL_KEYWORDS = (
                'key_file',
                'cert_file',
                'cert_reqs',
                'ca_certs',
                'ca_cert_dir',
                'ssl_version',
            )
            ```

        """
        self.ssl_keywords = {k: kwargs.pop(k, None) for k in SSL_KEYWORDS if k in kwargs}

    def _append_ssl_keywords(self, **kwargs):
        """
        Returns keyword arguments with the preconfigured SSL keywords.

        """
        kwargs.update(self.ssl_keywords)
        return kwargs

    def init_poolmanager(self, connections, maxsize, block=DEFAULT_POOLBLOCK, **pool_kwargs):
        """Initializes a urllib3 PoolManager.

        This method should not be called from user code, and is only
        exposed for use when subclassing the
        :class:`HTTPAdapter <requests.adapters.HTTPAdapter>`.

        :param connections: The number of urllib3 connection pools to cache.
        :param maxsize: The maximum number of connections to save in the pool.
        :param block: Block when no free connections are available.
        :param pool_kwargs: Extra keyword arguments used to initialize the Pool Manager.
        """
        self.poolmanager = PoolManager(
            num_pools=connections,
            maxsize=maxsize,
            block=block,
            **self._append_ssl_keywords(**pool_kwargs)
        )

    def proxy_manager_for(self, proxy, **proxy_kwargs):
        """Return urllib3 ProxyManager for the given proxy.

        This method should not be called from user code, and is only
        exposed for use when subclassing the
        :class:`HTTPAdapter <requests.adapters.HTTPAdapter>`.

        :param proxy: The proxy to return a urllib3 ProxyManager for.
        :param proxy_kwargs: Extra keyword arguments used to configure the Proxy Manager.
        :returns: ProxyManager
        """
        return super(SSLAdapter, self).proxy_manager_for(
            proxy,
            **self._append_ssl_keywords(**proxy_kwargs)
        )