­ ­ ­ ­ ­ ­ ­ ­ ­ ­ ­ ­ ­ ­ ­ ­ ­ ­ # 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 hmac import time import base64 import hashlib from io import FileIO as file from libcloud.utils.py3 import b, next, httplib, urlparse, urlquote, urlencode, urlunquote from libcloud.common.base import XmlResponse, ConnectionUserAndKey from libcloud.utils.files import read_in_chunks from libcloud.common.types import LibcloudError from libcloud.storage.base import CHUNK_SIZE, Object, Container, StorageDriver from libcloud.storage.types import ( ObjectDoesNotExistError, ContainerIsNotEmptyError, ContainerDoesNotExistError, ContainerAlreadyExistsError, ) def collapse(s): return " ".join([x for x in s.split(" ") if x]) class AtmosError(LibcloudError): def __init__(self, code, message, driver=None): super().__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().__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() # nosec 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(), # nosec "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() # nosec 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)