FM / libcloud

/opt/imh-python/lib/python3.9/site-packages/libcloud/

/opt/imh-python/lib/python3.9/site-packages/libcloud

Name	Size	Mode	Actions
backup/	-	0755	rm
common/	-	0755	rm
compute/	-	0755	rm
container/	-	0755	rm
data/	-	0755	rm
dns/	-	0755	rm
loadbalancer/	-	0755	rm
storage/	-	0755	rm
test/	-	0755	rm
utils/	-	0755	rm
__pycache__/	-	0755	rm
base.py	2809	0644	edit dl rm
http.py	10501	0644	edit dl rm
pricing.py	6978	0644	edit dl rm
py.typed	0	0644	edit dl rm
security.py	2940	0644	edit dl rm
__init__.py	3621	0644	edit dl rm

Edit: /opt/imh-python/lib/python3.9/site-packages/libcloud/http.py (10501B)

# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""
Subclass for httplib.HTTPSConnection with optional certificate name
verification, depending on libcloud.security settings.
"""

import os
import warnings
import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.poolmanager import PoolManager

import libcloud.security
from libcloud.utils.py3 import urlparse, PY3

__all__ = [
    'LibcloudBaseConnection',
    'LibcloudConnection'
]

ALLOW_REDIRECTS = 1

HTTP_PROXY_ENV_VARIABLE_NAME = 'http_proxy'
HTTPS_PROXY_ENV_VARIABLE_NAME = 'https_proxy'

class SignedHTTPSAdapter(HTTPAdapter):
    def __init__(self, cert_file, key_file):
        self.cert_file = cert_file
        self.key_file = key_file
        super(SignedHTTPSAdapter, self).__init__()

def init_poolmanager(self, connections, maxsize, block=False):
        self.poolmanager = PoolManager(
            num_pools=connections, maxsize=maxsize,
            block=block,
            cert_file=self.cert_file,
            key_file=self.key_file)

class LibcloudBaseConnection(object):
    """
    Base connection class to inherit from.

Note: This class should not be instantiated directly.
    """

session = None

proxy_scheme = None
    proxy_host = None
    proxy_port = None

proxy_username = None
    proxy_password = None

http_proxy_used = False

ca_cert = None

def __init__(self):
        self.session = requests.Session()

def set_http_proxy(self, proxy_url):
        """
        Set a HTTP proxy which will be used with this connection.

:param proxy_url: Proxy URL (e.g. http://<hostname>:<port> without
                          authentication and
                          http://<username>:<password>@<hostname>:<port> for
                          basic auth authentication information.
        :type proxy_url: ``str``
        """
        result = self._parse_proxy_url(proxy_url=proxy_url)

scheme = result[0]
        host = result[1]
        port = result[2]
        username = result[3]
        password = result[4]

self.proxy_scheme = scheme
        self.proxy_host = host
        self.proxy_port = port
        self.proxy_username = username
        self.proxy_password = password
        self.http_proxy_used = True

self.session.proxies = {
            'http': proxy_url,
            'https': proxy_url,
        }

def _parse_proxy_url(self, proxy_url):
        """
        Parse and validate a proxy URL.

:param proxy_url: Proxy URL (e.g. http://hostname:3128)
        :type proxy_url: ``str``

:rtype: ``tuple`` (``scheme``, ``hostname``, ``port``)
        """
        parsed = urlparse.urlparse(proxy_url)

if parsed.scheme not in ('http', 'https'):
            raise ValueError('Only http and https proxies are supported')

if not parsed.hostname or not parsed.port:
            raise ValueError('proxy_url must be in the following format: '
                             '<scheme>://<proxy host>:<proxy port>')

proxy_scheme = parsed.scheme
        proxy_host, proxy_port = parsed.hostname, parsed.port

netloc = parsed.netloc

if '@' in netloc:
            username_password = netloc.split('@', 1)[0]
            split = username_password.split(':', 1)

if len(split) < 2:
                raise ValueError('URL is in an invalid format')

proxy_username, proxy_password = split[0], split[1]
        else:
            proxy_username = None
            proxy_password = None

return (proxy_scheme, proxy_host, proxy_port, proxy_username,
                proxy_password)

def _setup_verify(self):
        self.verify = libcloud.security.VERIFY_SSL_CERT

def _setup_ca_cert(self, **kwargs):
        # simulating keyword-only argument in Python 2
        ca_certs_path = kwargs.get('ca_cert', libcloud.security.CA_CERTS_PATH)

if self.verify is False:
            pass
        else:
            if isinstance(ca_certs_path, list):
                msg = (
                    'Providing a list of CA trusts is no longer supported '
                    'since libcloud 2.0. Using the first element in the list. '
                    'See http://libcloud.readthedocs.io/en/latest/other/'
                    'changes_in_2_0.html#providing-a-list-of-ca-trusts-is-no-'
                    'longer-supported')
                warnings.warn(msg, DeprecationWarning)
                self.ca_cert = ca_certs_path[0]
            else:
                self.ca_cert = ca_certs_path

def _setup_signing(self, cert_file=None, key_file=None):
        """
        Setup request signing by mounting a signing
        adapter to the session
        """
        self.session.mount('https://', SignedHTTPSAdapter(cert_file, key_file))

class LibcloudConnection(LibcloudBaseConnection):
    timeout = None
    host = None
    response = None

def __init__(self, host, port, secure=None, **kwargs):
        scheme = 'https' if secure is not None and secure else 'http'
        self.host = '{0}://{1}{2}'.format(
            'https' if port == 443 else scheme,
            host,
            ":{0}".format(port) if port not in (80, 443) else ""
        )

# Support for HTTP(s) proxy
        # NOTE: We always only use a single proxy (either HTTP or HTTPS)
        https_proxy_url_env = os.environ.get(HTTPS_PROXY_ENV_VARIABLE_NAME,
                                             None)
        http_proxy_url_env = os.environ.get(HTTP_PROXY_ENV_VARIABLE_NAME,
                                            https_proxy_url_env)

# Connection argument has precedence over environment variables
        proxy_url = kwargs.pop('proxy_url', http_proxy_url_env)

self._setup_verify()
        self._setup_ca_cert()

LibcloudBaseConnection.__init__(self)

self.session.timeout = kwargs.pop('timeout', 60)

if 'cert_file' in kwargs or 'key_file' in kwargs:
            self._setup_signing(**kwargs)

if proxy_url:
            self.set_http_proxy(proxy_url=proxy_url)

@property
    def verification(self):
        """
        The option for SSL verification given to underlying requests
        """
        return self.ca_cert if self.ca_cert is not None else self.verify

def request(self, method, url, body=None, headers=None, raw=False,
                stream=False):
        url = urlparse.urljoin(self.host, url)
        headers = self._normalize_headers(headers=headers)

self.response = self.session.request(
            method=method.lower(),
            url=url,
            data=body,
            headers=headers,
            allow_redirects=ALLOW_REDIRECTS,
            stream=stream,
            verify=self.verification
        )

def prepared_request(self, method, url, body=None,
                         headers=None, raw=False, stream=False):
        headers = self._normalize_headers(headers=headers)

req = requests.Request(method, ''.join([self.host, url]),
                               data=body, headers=headers)

prepped = self.session.prepare_request(req)

prepped.body = body

self.response = self.session.send(
            prepped,
            stream=stream,
            verify=self.ca_cert if self.ca_cert is not None else self.verify)

def getresponse(self):
        return self.response

def getheaders(self):
        # urlib decoded response body, libcloud has a bug
        # and will not check if content is gzipped, so let's
        # remove headers indicating compressed content.
        if 'content-encoding' in self.response.headers:
            del self.response.headers['content-encoding']
        return self.response.headers

@property
    def status(self):
        return self.response.status_code

@property
    def reason(self):
        return None if self.response.status_code > 400 else self.response.text

def connect(self):  # pragma: no cover
        pass

def read(self):
        return self.response.content

def close(self):  # pragma: no cover
        # return connection back to pool
        self.response.close()

def _normalize_headers(self, headers):
        headers = headers or {}

# all headers should be strings
        for key, value in headers.items():
            if isinstance(value, (int, float)):
                headers[key] = str(value)

return headers

class HttpLibResponseProxy(object):
    """
    Provides a proxy pattern around the :class:`requests.Reponse`
    object to a :class:`httplib.HTTPResponse` object
    """
    def __init__(self, response):
        self._response = response

def read(self, amt=None):
        return self._response.text

def getheader(self, name, default=None):
        """
        Get the contents of the header name, or default
        if there is no matching header.
        """
        if name in self._response.headers.keys():
            return self._response.headers[name]
        else:
            return default

def getheaders(self):
        """
        Return a list of (header, value) tuples.
        """
        if PY3:
            return list(self._response.headers.items())
        else:
            return self._response.headers.items()

@property
    def status(self):
        return self._response.status_code

@property
    def reason(self):
        return self._response.reason

@property
    def version(self):
        # requests doesn't expose this
        return '11'

@property
    def body(self):
        # NOTE: We use property to avoid saving whole response body into RAM
        # See https://github.com/apache/libcloud/pull/1132 for details
        return self._response.content

cmd: