FM / drivers

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

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

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
atmos.py	17689	0644	edit dl rm
auroraobjects.py	1933	0644	edit dl rm
azure_blobs.py	41324	0644	edit dl rm
backblaze_b2.py	21167	0644	edit dl rm
cloudfiles.py	39929	0644	edit dl rm
digitalocean_spaces.py	4499	0644	edit dl rm
dummy.py	19054	0644	edit dl rm
google_storage.py	15211	0644	edit dl rm
ktucloud.py	1946	0644	edit dl rm
local.py	21586	0644	edit dl rm
nimbus.py	4334	0644	edit dl rm
ninefold.py	1075	0644	edit dl rm
oss.py	36500	0644	edit dl rm
rgw.py	5357	0644	edit dl rm
s3.py	51021	0644	edit dl rm
__init__.py	880	0644	edit dl rm

Edit: /opt/imh-python/lib/python3.9/site-packages/libcloud/storage/drivers/atmos.py (17689B)

# 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.

import base64
import hashlib
import hmac
import time

from libcloud.utils.py3 import PY3
from libcloud.utils.py3 import b
from libcloud.utils.py3 import httplib
from libcloud.utils.py3 import next
from libcloud.utils.py3 import urlparse
from libcloud.utils.py3 import urlencode
from libcloud.utils.py3 import urlquote
from libcloud.utils.py3 import urlunquote

if PY3:
    from io import FileIO as file

from libcloud.utils.files import read_in_chunks
from libcloud.common.base import ConnectionUserAndKey, XmlResponse
from libcloud.common.types import LibcloudError

from libcloud.storage.base import Object, Container, StorageDriver, CHUNK_SIZE
from libcloud.storage.types import ContainerAlreadyExistsError, \
    ContainerDoesNotExistError, ContainerIsNotEmptyError, \
    ObjectDoesNotExistError

def collapse(s):
    return ' '.join([x for x in s.split(' ') if x])

class AtmosError(LibcloudError):
    def __init__(self, code, message, driver=None):
        super(AtmosError, self).__init__(value=message, driver=driver)
        self.code = code

class AtmosResponse(XmlResponse):
    def success(self):
        return self.status in (httplib.OK, httplib.CREATED, httplib.NO_CONTENT,
                               httplib.PARTIAL_CONTENT)

def parse_error(self):
        tree = self.parse_body()

if tree is None:
            return None

code = int(tree.find('Code').text)
        message = tree.find('Message').text
        raise AtmosError(code=code, message=message,
                         driver=self.connection.driver)

class AtmosConnection(ConnectionUserAndKey):
    responseCls = AtmosResponse

def add_default_headers(self, headers):
        headers['x-emc-uid'] = self.user_id
        headers['Date'] = time.strftime('%a, %d %b %Y %H:%M:%S GMT',
                                        time.gmtime())
        headers['x-emc-date'] = headers['Date']

if 'Content-Type' not in headers:
            headers['Content-Type'] = 'application/octet-stream'
        if 'Accept' not in headers:
            headers['Accept'] = '*/*'

return headers

def pre_connect_hook(self, params, headers):
        headers['x-emc-signature'] = self._calculate_signature(params, headers)

return params, headers

def _calculate_signature(self, params, headers):
        pathstring = urlunquote(self.action)
        driver_path = self.driver.path  # pylint: disable=no-member
        if pathstring.startswith(driver_path):
            pathstring = pathstring[len(driver_path):]
        if params:
            if type(params) is dict:
                params = list(params.items())
            pathstring += '?' + urlencode(params)
        pathstring = pathstring.lower()

xhdrs = [(k, v) for k, v in list(headers.items()) if
                 k.startswith('x-emc-')]
        xhdrs.sort(key=lambda x: x[0])

signature = [
            self.method,
            headers.get('Content-Type', ''),
            headers.get('Range', ''),
            headers.get('Date', ''),
            pathstring,
        ]
        signature.extend([k + ':' + collapse(v) for k, v in xhdrs])
        signature = '\n'.join(signature)
        key = base64.b64decode(self.key)
        signature = hmac.new(b(key), b(signature), hashlib.sha1).digest()
        return base64.b64encode(b(signature)).decode('utf-8')

class AtmosDriver(StorageDriver):
    connectionCls = AtmosConnection

host = None  # type: str
    path = None  # type: str
    api_name = 'atmos'
    supports_chunked_encoding = True
    website = 'http://atmosonline.com/'
    name = 'atmos'

DEFAULT_CDN_TTL = 60 * 60 * 24 * 7  # 1 week

def __init__(self, key, secret=None, secure=True, host=None, port=None):
        host = host or self.host
        super(AtmosDriver, self).__init__(key, secret, secure, host, port)

def iterate_containers(self):
        result = self.connection.request(self._namespace_path(''))
        entries = self._list_objects(result.object, object_type='directory')
        for entry in entries:
            extra = {
                'object_id': entry['id']
            }
            yield Container(entry['name'], extra, self)

def get_container(self, container_name):
        path = self._namespace_path(container_name) + '/?metadata/system'
        try:
            result = self.connection.request(path)
        except AtmosError as e:
            if e.code != 1003:
                raise
            raise ContainerDoesNotExistError(e, self, container_name)
        meta = self._emc_meta(result)
        extra = {
            'object_id': meta['objectid']
        }
        return Container(container_name, extra, self)

def create_container(self, container_name):
        path = self._namespace_path(container_name) + '/'
        try:
            self.connection.request(path, method='POST')
        except AtmosError as e:
            if e.code != 1016:
                raise
            raise ContainerAlreadyExistsError(e, self, container_name)
        return self.get_container(container_name)

def delete_container(self, container):
        try:
            self.connection.request(self._namespace_path(container.name) + '/',
                                    method='DELETE')
        except AtmosError as e:
            if e.code == 1003:
                raise ContainerDoesNotExistError(e, self, container.name)
            elif e.code == 1023:
                raise ContainerIsNotEmptyError(e, self, container.name)
        return True

def get_object(self, container_name, object_name):
        container = self.get_container(container_name)
        object_name_cleaned = self._clean_object_name(object_name)
        path = self._namespace_path(container_name) + '/' + object_name_cleaned

try:
            result = self.connection.request(path + '?metadata/system')
            system_meta = self._emc_meta(result)

result = self.connection.request(path + '?metadata/user')
            user_meta = self._emc_meta(result)
        except AtmosError as e:
            if e.code != 1003:
                raise
            raise ObjectDoesNotExistError(e, self, object_name)

last_modified = time.strptime(system_meta['mtime'],
                                      '%Y-%m-%dT%H:%M:%SZ')
        last_modified = time.strftime('%a, %d %b %Y %H:%M:%S GMT',
                                      last_modified)
        extra = {
            'object_id': system_meta['objectid'],
            'last_modified': last_modified
        }
        data_hash = user_meta.pop('md5', '')
        return Object(object_name, int(system_meta['size']), data_hash, extra,
                      user_meta, container, self)

def upload_object(self, file_path, container, object_name, extra=None,
                      verify_hash=True, headers=None):
        method = 'PUT'

extra = extra or {}
        object_name_cleaned = self._clean_object_name(object_name)
        request_path = self._namespace_path(container.name) + '/' +\
            object_name_cleaned
        content_type = extra.get('content_type', None)

try:
            self.connection.request(request_path + '?metadata/system')
        except AtmosError as e:
            if e.code != 1003:
                raise
            method = 'POST'

result_dict = self._upload_object(
            object_name=object_name,
            content_type=content_type,
            request_path=request_path,
            request_method=method,
            headers={}, file_path=file_path)

bytes_transferred = result_dict['bytes_transferred']

if extra is None:
            meta_data = {}
        else:
            meta_data = extra.get('meta_data', {})
        meta_data['md5'] = result_dict['data_hash']
        user_meta = ', '.join([k + '=' + str(v) for k, v in
                               list(meta_data.items())])
        self.connection.request(request_path + '?metadata/user', method='POST',
                                headers={'x-emc-meta': user_meta})
        result = self.connection.request(request_path + '?metadata/system')
        meta = self._emc_meta(result)
        del meta_data['md5']
        extra = {
            'object_id': meta['objectid'],
            'meta_data': meta_data,
        }

return Object(object_name, bytes_transferred, result_dict['data_hash'],
                      extra, meta_data, container, self)

def upload_object_via_stream(self, iterator, container, object_name,
                                 extra=None, headers=None):
        if isinstance(iterator, file):
            iterator = iter(iterator)

extra_headers = headers or {}
        data_hash = hashlib.md5()
        generator = read_in_chunks(iterator, CHUNK_SIZE, True)
        bytes_transferred = 0
        try:
            chunk = next(generator)
        except StopIteration:
            chunk = ''

path = self._namespace_path(container.name + '/' + object_name)
        method = 'PUT'

if extra is not None:
            content_type = extra.get('content_type', None)
        else:
            content_type = None

content_type = self._determine_content_type(content_type, object_name)

try:
            self.connection.request(path + '?metadata/system')
        except AtmosError as e:
            if e.code != 1003:
                raise
            method = 'POST'

while True:
            end = bytes_transferred + len(chunk) - 1
            data_hash.update(b(chunk))
            headers = dict(extra_headers)

headers.update({
                'x-emc-meta': 'md5=' + data_hash.hexdigest(),
                'Content-Type': content_type,
            })

if len(chunk) > 0 and bytes_transferred > 0:
                headers['Range'] = 'Bytes=%d-%d' % (bytes_transferred, end)
                method = 'PUT'

result = self.connection.request(path, method=method, data=chunk,
                                             headers=headers)
            bytes_transferred += len(chunk)

try:
                chunk = next(generator)
            except StopIteration:
                break
            if len(chunk) == 0:
                break

data_hash = data_hash.hexdigest()

if extra is None:
            meta_data = {}
        else:
            meta_data = extra.get('meta_data', {})
        meta_data['md5'] = data_hash
        user_meta = ', '.join([k + '=' + str(v) for k, v in
                               list(meta_data.items())])
        self.connection.request(path + '?metadata/user', method='POST',
                                headers={'x-emc-meta': user_meta})

result = self.connection.request(path + '?metadata/system')

meta = self._emc_meta(result)
        extra = {
            'object_id': meta['objectid'],
            'meta_data': meta_data,
        }

return Object(object_name, bytes_transferred, data_hash, extra,
                      meta_data, container, self)

def download_object(self, obj, destination_path, overwrite_existing=False,
                        delete_on_failure=True):
        path = self._namespace_path(obj.container.name + '/' + obj.name)
        response = self.connection.request(path, method='GET', raw=True)

return self._get_object(obj=obj, callback=self._save_object,
                                response=response,
                                callback_kwargs={
                                    'obj': obj,
                                    'response': response.response,
                                    'destination_path': destination_path,
                                    'overwrite_existing': overwrite_existing,
                                    'delete_on_failure': delete_on_failure
                                },
                                success_status_code=httplib.OK)

def download_object_as_stream(self, obj, chunk_size=None):
        path = self._namespace_path(obj.container.name + '/' + obj.name)
        response = self.connection.request(path, method='GET', raw=True)

return self._get_object(obj=obj, callback=read_in_chunks,
                                response=response,
                                callback_kwargs={
                                    'iterator': response.response,
                                    'chunk_size': chunk_size
                                },
                                success_status_code=httplib.OK)

def delete_object(self, obj):
        path = self._namespace_path(obj.container.name) + '/' +\
            self._clean_object_name(obj.name)
        try:
            self.connection.request(path, method='DELETE')
        except AtmosError as e:
            if e.code != 1003:
                raise
            raise ObjectDoesNotExistError(e, self, obj.name)
        return True

def enable_object_cdn(self, obj):
        return True

def get_object_cdn_url(self, obj, expiry=None, use_object=False):
        """
        Return an object CDN URL.

:param obj: Object instance
        :type  obj: :class:`Object`

:param expiry: Expiry
        :type expiry: ``str``

:param use_object: Use object
        :type use_object: ``bool``

:rtype: ``str``
        """
        if use_object:
            path = '/rest/objects' + obj.meta_data['object_id']
        else:
            path = '/rest/namespace/' + obj.container.name + '/' + obj.name

if self.secure:
            protocol = 'https'
        else:
            protocol = 'http'

expiry = str(expiry or int(time.time()) + self.DEFAULT_CDN_TTL)
        params = [
            ('uid', self.key),
            ('expires', expiry),
        ]
        params.append(('signature', self._cdn_signature(path, params, expiry)))

params = urlencode(params)
        path = self.path + path
        return urlparse.urlunparse((protocol, self.host, path, '', params, ''))

def _cdn_signature(self, path, params, expiry):
        key = base64.b64decode(self.secret)
        signature = '\n'.join(['GET', path.lower(), self.key, expiry])
        signature = hmac.new(key, signature, hashlib.sha1).digest()

return base64.b64encode(signature)

def _list_objects(self, tree, object_type=None):
        listing = tree.find(self._emc_tag('DirectoryList'))
        entries = []
        for entry in listing.findall(self._emc_tag('DirectoryEntry')):
            file_type = entry.find(self._emc_tag('FileType')).text
            if object_type is not None and object_type != file_type:
                continue
            entries.append({
                'id': entry.find(self._emc_tag('ObjectID')).text,
                'type': file_type,
                'name': entry.find(self._emc_tag('Filename')).text
            })
        return entries

def _clean_object_name(self, name):
        return urlquote(name.encode('ascii'))

def _namespace_path(self, path):
        return self.path + '/rest/namespace/' + urlquote(path.encode('ascii'))

def _object_path(self, object_id):
        return self.path + '/rest/objects/' + object_id.encode('ascii')

@staticmethod
    def _emc_tag(tag):
        return '{http://www.emc.com/cos/}' + tag

def _emc_meta(self, response):
        meta = response.headers.get('x-emc-meta', '')
        if len(meta) == 0:
            return {}
        meta = meta.split(', ')
        return dict([x.split('=', 1) for x in meta])

def _entries_to_objects(self, container, entries):
        for entry in entries:
            metadata = {'object_id': entry['id']}
            yield Object(entry['name'], 0, '', {}, metadata, container, self)

def iterate_container_objects(self, container, prefix=None,
                                  ex_prefix=None):
        """
        Return a generator of objects for the given container.

:param container: Container instance
        :type container: :class:`Container`

:param prefix: Filter objects starting with a prefix.
                       Filtering is performed client-side.
        :type  prefix: ``str``

:param ex_prefix: (Deprecated.) Filter objects starting with a prefix.
                          Filtering is performed client-side.
        :type  ex_prefix: ``str``

:return: A generator of Object instances.
        :rtype: ``generator`` of :class:`Object`
        """
        prefix = self._normalize_prefix_argument(prefix, ex_prefix)

headers = {'x-emc-include-meta': '1'}
        path = self._namespace_path(container.name) + '/'
        result = self.connection.request(path, headers=headers)
        entries = self._list_objects(result.object, object_type='regular')
        objects = self._entries_to_objects(container, entries)
        return self._filter_listed_container_objects(objects, prefix)

cmd: