Server IP : 85.214.239.14 / Your IP : 3.145.85.123 Web Server : Apache/2.4.62 (Debian) System : Linux h2886529.stratoserver.net 4.9.0 #1 SMP Tue Jan 9 19:45:01 MSK 2024 x86_64 User : www-data ( 33) PHP Version : 7.4.18 Disable Function : pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_get_handler,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,pcntl_async_signals,pcntl_unshare, MySQL : OFF | cURL : OFF | WGET : ON | Perl : ON | Python : ON | Sudo : ON | Pkexec : OFF Directory : /proc/2/cwd/lib/python3/dist-packages/libcloud/storage/drivers/ |
Upload File : |
# 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)