# -*- coding: utf-8 -*- ## Amazon S3 manager ## Author: Michal Ludvig ## http://www.logix.cz/michal ## License: GPL Version 2 ## Copyright: TGRMN Software and contributors from __future__ import absolute_import, division import sys import os import time import errno import base64 import mimetypes import io import pprint from xml.sax import saxutils from socket import timeout as SocketTimeoutException from logging import debug, info, warning, error from stat import ST_SIZE try: # python 3 support from urlparse import urlparse except ImportError: from urllib.parse import urlparse import select try: from hashlib import md5 except ImportError: from md5 import md5 from .Utils import (getListFromXml, getTextFromXml, getRootTagName, convertHeaderTupleListToDict, hash_file_md5, unicodise, deunicodise, decode_from_s3, encode_to_s3, s3_quote, check_bucket_name, check_bucket_name_dns_support, getHostnameFromBucket, calculateChecksum) from .SortedDict import SortedDict from .AccessLog import AccessLog from .ACL import ACL, GranteeLogDelivery from .BidirMap import BidirMap from .Config import Config from .Exceptions import * from .MultiPart import MultiPartUpload from .S3Uri import S3Uri from .ConnMan import ConnMan from .Crypto import (sign_request_v2, sign_request_v4, checksum_sha256_file, checksum_sha256_buffer, format_param_str) try: from ctypes import ArgumentError import magic try: ## https://github.com/ahupp/python-magic ## Always expect unicode for python 2 ## (has Magic class but no "open()" function) magic_ = magic.Magic(mime=True) def mime_magic_file(file): return magic_.from_file(file) except TypeError: try: ## file-5.11 built-in python bindings ## Sources: http://www.darwinsys.com/file/ ## Expects unicode since version 5.19, encoded strings before ## we can't tell if a given copy of the magic library will take a ## filesystem-encoded string or a unicode value, so try first ## with the unicode, then with the encoded string. ## (has Magic class and "open()" function) magic_ = magic.open(magic.MAGIC_MIME) magic_.load() def mime_magic_file(file): try: return magic_.file(file) except (UnicodeDecodeError, UnicodeEncodeError, ArgumentError): return magic_.file(deunicodise(file)) except AttributeError: ## http://pypi.python.org/pypi/filemagic ## Accept gracefully both unicode and encoded ## (has Magic class but not "mime" argument and no "open()" function ) magic_ = magic.Magic(flags=magic.MAGIC_MIME) def mime_magic_file(file): return magic_.id_filename(file) except AttributeError: ## Older python-magic versions doesn't have a "Magic" method ## Only except encoded strings ## (has no Magic class but "open()" function) magic_ = magic.open(magic.MAGIC_MIME) magic_.load() def mime_magic_file(file): return magic_.file(deunicodise(file)) except (ImportError, OSError) as e: error_str = str(e) if 'magic' in error_str: magic_message = "Module python-magic is not available." else: magic_message = "Module python-magic can't be used (%s)." % error_str magic_message += " Guessing MIME types based on file extensions." magic_warned = False def mime_magic_file(file): global magic_warned if (not magic_warned): warning(magic_message) magic_warned = True return mimetypes.guess_type(file)[0] def mime_magic(file): ## NOTE: So far in the code, "file" var is already unicode def _mime_magic(file): magictype = mime_magic_file(file) return magictype result = _mime_magic(file) if result is not None: if isinstance(result, str): if ';' in result: mimetype, charset = result.split(';') charset = charset[len('charset'):] result = (mimetype, charset) else: result = (result, None) if result is None: result = (None, None) return result EXPECT_CONTINUE_TIMEOUT = 2 SIZE_1MB = 1024 * 1024 __all__ = [] class S3Request(object): region_map = {} ## S3 sometimes sends HTTP-301, HTTP-307 response redir_map = {} def __init__(self, s3, method_string, resource, headers, body, params = None): self.s3 = s3 self.headers = SortedDict(headers or {}, ignore_case = True) if len(self.s3.config.access_token)>0: self.s3.config.role_refresh() self.headers['x-amz-security-token']=self.s3.config.access_token self.resource = resource self.method_string = method_string self.params = params or {} self.body = body self.requester_pays() def requester_pays(self): if self.s3.config.requester_pays and self.method_string in ("GET", "POST", "PUT", "HEAD"): self.headers['x-amz-request-payer'] = 'requester' def update_timestamp(self): if "date" in self.headers: del(self.headers["date"]) self.headers["x-amz-date"] = time.strftime("%a, %d %b %Y %H:%M:%S +0000", time.gmtime()) def use_signature_v2(self): if self.s3.endpoint_requires_signature_v4: return False if self.s3.config.signature_v2 or self.s3.fallback_to_signature_v2: return True return False def sign(self): bucket_name = self.resource.get('bucket') if self.use_signature_v2(): debug("Using signature v2") if bucket_name: resource_uri = "/%s%s" % (bucket_name, self.resource['uri']) else: resource_uri = self.resource['uri'] self.headers = sign_request_v2(self.method_string, resource_uri, self.params, self.headers) else: debug("Using signature v4") hostname = self.s3.get_hostname(self.resource['bucket']) ## Default to bucket part of DNS. ## If bucket is not part of DNS assume path style to complete the request. ## Like for format_uri, take care that redirection could be to base path if bucket_name and ( (bucket_name in S3Request.redir_map and not S3Request.redir_map.get(bucket_name, '').startswith("%s."% bucket_name)) or (bucket_name not in S3Request.redir_map and not check_bucket_name_dns_support(Config().host_bucket, bucket_name)) ): resource_uri = "/%s%s" % (bucket_name, self.resource['uri']) else: resource_uri = self.resource['uri'] bucket_region = S3Request.region_map.get(self.resource['bucket'], Config().bucket_location) ## Sign the data. self.headers = sign_request_v4(self.method_string, hostname, resource_uri, self.params, bucket_region, self.headers, self.body) def get_triplet(self): self.update_timestamp() self.sign() resource = dict(self.resource) ## take a copy # URL Encode the uri for the http request resource['uri'] = s3_quote(resource['uri'], quote_backslashes=False, unicode_output=True) # Get the final uri by adding the uri parameters resource['uri'] += format_param_str(self.params) return (self.method_string, resource, self.headers) class S3(object): http_methods = BidirMap( GET = 0x01, PUT = 0x02, HEAD = 0x04, DELETE = 0x08, POST = 0x10, MASK = 0x1F, ) targets = BidirMap( SERVICE = 0x0100, BUCKET = 0x0200, OBJECT = 0x0400, BATCH = 0x0800, MASK = 0x0700, ) operations = BidirMap( UNDFINED = 0x0000, LIST_ALL_BUCKETS = targets["SERVICE"] | http_methods["GET"], BUCKET_CREATE = targets["BUCKET"] | http_methods["PUT"], BUCKET_LIST = targets["BUCKET"] | http_methods["GET"], BUCKET_DELETE = targets["BUCKET"] | http_methods["DELETE"], OBJECT_PUT = targets["OBJECT"] | http_methods["PUT"], OBJECT_GET = targets["OBJECT"] | http_methods["GET"], OBJECT_HEAD = targets["OBJECT"] | http_methods["HEAD"], OBJECT_DELETE = targets["OBJECT"] | http_methods["DELETE"], OBJECT_POST = targets["OBJECT"] | http_methods["POST"], BATCH_DELETE = targets["BATCH"] | http_methods["POST"], ) codes = { "NoSuchBucket" : "Bucket '%s' does not exist", "AccessDenied" : "Access to bucket '%s' was denied", "BucketAlreadyExists" : "Bucket '%s' already exists", } ## Maximum attempts of re-issuing failed requests _max_retries = 5 def __init__(self, config): self.config = config self.fallback_to_signature_v2 = False self.endpoint_requires_signature_v4 = False self.expect_continue_not_supported = False def storage_class(self): # Note - you cannot specify GLACIER here # https://docs.aws.amazon.com/AmazonS3/latest/dev/storage-class-intro.html cls = 'STANDARD' if self.config.storage_class != "": return self.config.storage_class if self.config.reduced_redundancy: cls = 'REDUCED_REDUNDANCY' return cls def get_hostname(self, bucket): if bucket and bucket in S3Request.redir_map: host = S3Request.redir_map[bucket] elif bucket and check_bucket_name_dns_support(self.config.host_bucket, bucket): host = getHostnameFromBucket(bucket) else: host = self.config.host_base.lower() # The following hack is needed because it looks like that some servers # are not respecting the HTTP spec and so will fail the signature check # if the port is specified in the "Host" header for default ports. # STUPIDIEST THING EVER FOR A SERVER... # See: https://github.com/minio/minio/issues/9169 if self.config.use_https: if host.endswith(':443'): host = host[:-4] elif host.endswith(':80'): host = host[:-3] debug('get_hostname(%s): %s' % (bucket, host)) return host def set_hostname(self, bucket, redir_hostname): S3Request.redir_map[bucket] = redir_hostname.lower() def format_uri(self, resource, base_path=None): bucket_name = resource.get('bucket') if bucket_name and ( (bucket_name in S3Request.redir_map and not S3Request.redir_map.get(bucket_name, '').startswith("%s."% bucket_name)) or (bucket_name not in S3Request.redir_map and not check_bucket_name_dns_support(self.config.host_bucket, bucket_name)) ): uri = "/%s%s" % (s3_quote(bucket_name, quote_backslashes=False, unicode_output=True), resource['uri']) else: uri = resource['uri'] if base_path: uri = "%s%s" % (base_path, uri) if self.config.proxy_host != "" and not self.config.use_https: uri = "http://%s%s" % (self.get_hostname(bucket_name), uri) debug('format_uri(): ' + uri) return uri ## Commands / Actions def list_all_buckets(self): request = self.create_request("LIST_ALL_BUCKETS") response = self.send_request(request) response["list"] = getListFromXml(response["data"], "Bucket") return response def bucket_list(self, bucket, prefix = None, recursive = None, uri_params = None, limit = -1): item_list = [] prefixes = [] for truncated, dirs, objects in self.bucket_list_streaming(bucket, prefix, recursive, uri_params, limit): item_list.extend(objects) prefixes.extend(dirs) response = {} response['list'] = item_list response['common_prefixes'] = prefixes response['truncated'] = truncated return response def bucket_list_streaming(self, bucket, prefix = None, recursive = None, uri_params = None, limit = -1): """ Generator that produces , pairs of groups of content of a specified bucket. """ def _list_truncated(data): ## can either be "true" or "false" or be missing completely is_truncated = getTextFromXml(data, ".//IsTruncated") or "false" return is_truncated.lower() != "false" def _get_contents(data): return getListFromXml(data, "Contents") def _get_common_prefixes(data): return getListFromXml(data, "CommonPrefixes") def _get_next_marker(data, current_list): return getTextFromXml(response["data"], "NextMarker") or current_list[-1]["Key"] uri_params = uri_params and uri_params.copy() or {} truncated = True prefixes = [] num_objects = 0 num_prefixes = 0 max_keys = limit while truncated: response = self.bucket_list_noparse(bucket, prefix, recursive, uri_params, max_keys) current_list = _get_contents(response["data"]) current_prefixes = _get_common_prefixes(response["data"]) num_objects += len(current_list) num_prefixes += len(current_prefixes) if limit > num_objects + num_prefixes: max_keys = limit - (num_objects + num_prefixes) truncated = _list_truncated(response["data"]) if truncated: if limit == -1 or num_objects + num_prefixes < limit: if current_list: uri_params['marker'] = \ _get_next_marker(response["data"], current_list) elif current_prefixes: uri_params['marker'] = current_prefixes[-1]["Prefix"] else: # Unexpectedly, the server lied, and so the previous # response was not truncated. So, no new key to get. yield False, current_prefixes, current_list break debug("Listing continues after '%s'" % uri_params['marker']) else: yield truncated, current_prefixes, current_list break yield truncated, current_prefixes, current_list def bucket_list_noparse(self, bucket, prefix = None, recursive = None, uri_params = None, max_keys = -1): if uri_params is None: uri_params = {} if prefix: uri_params['prefix'] = prefix if not self.config.recursive and not recursive: uri_params['delimiter'] = "/" if max_keys != -1: uri_params['max-keys'] = str(max_keys) request = self.create_request("BUCKET_LIST", bucket = bucket, uri_params = uri_params) response = self.send_request(request) #debug(response) return response def bucket_create(self, bucket, bucket_location = None): headers = SortedDict(ignore_case = True) body = "" if bucket_location and bucket_location.strip().upper() != "US" and bucket_location.strip().lower() != "us-east-1": bucket_location = bucket_location.strip() if bucket_location.upper() == "EU": bucket_location = bucket_location.upper() body = "" body += bucket_location body += "" debug("bucket_location: " + body) check_bucket_name(bucket, dns_strict = True) else: check_bucket_name(bucket, dns_strict = False) if self.config.acl_public: headers["x-amz-acl"] = "public-read" request = self.create_request("BUCKET_CREATE", bucket = bucket, headers = headers, body = body) response = self.send_request(request) return response def bucket_delete(self, bucket): request = self.create_request("BUCKET_DELETE", bucket = bucket) response = self.send_request(request) return response def get_bucket_location(self, uri, force_us_default=False): bucket = uri.bucket() request = self.create_request("BUCKET_LIST", bucket = uri.bucket(), uri_params = {'location': None}) saved_redir_map = S3Request.redir_map.get(bucket, '') saved_region_map = S3Request.region_map.get(bucket, '') try: if force_us_default and not (saved_redir_map and saved_region_map): S3Request.redir_map[bucket] = self.config.host_base S3Request.region_map[bucket] = 'us-east-1' response = self.send_request(request) finally: if bucket in saved_redir_map: S3Request.redir_map[bucket] = saved_redir_map elif bucket in S3Request.redir_map: del S3Request.redir_map[bucket] if bucket in saved_region_map: S3Request.region_map[bucket] = saved_region_map elif bucket in S3Request.region_map: del S3Request.region_map[bucket] location = getTextFromXml(response['data'], "LocationConstraint") if not location or location in [ "", "US" ]: location = "us-east-1" elif location == "EU": location = "eu-west-1" return location def get_bucket_requester_pays(self, uri): request = self.create_request("BUCKET_LIST", bucket=uri.bucket(), uri_params={'requestPayment': None}) response = self.send_request(request) resp_data = response.get('data', '') if resp_data: payer = getTextFromXml(response['data'], "Payer") else: payer = None return payer def bucket_info(self, uri): response = {} response['bucket-location'] = self.get_bucket_location(uri) try: response['requester-pays'] = self.get_bucket_requester_pays(uri) except S3Error as e: response['requester-pays'] = None return response def website_info(self, uri, bucket_location = None): bucket = uri.bucket() request = self.create_request("BUCKET_LIST", bucket = bucket, uri_params = {'website': None}) try: response = self.send_request(request) response['index_document'] = getTextFromXml(response['data'], ".//IndexDocument//Suffix") response['error_document'] = getTextFromXml(response['data'], ".//ErrorDocument//Key") response['website_endpoint'] = self.config.website_endpoint % { "bucket" : uri.bucket(), "location" : self.get_bucket_location(uri)} return response except S3Error as e: if e.status == 404: debug("Could not get /?website - website probably not configured for this bucket") return None raise def website_create(self, uri, bucket_location = None): bucket = uri.bucket() body = '' body += ' ' body += (' %s' % self.config.website_index) body += ' ' if self.config.website_error: body += ' ' body += (' %s' % self.config.website_error) body += ' ' body += '' request = self.create_request("BUCKET_CREATE", bucket = bucket, body = body, uri_params = {'website': None}) response = self.send_request(request) debug("Received response '%s'" % (response)) return response def website_delete(self, uri, bucket_location = None): bucket = uri.bucket() request = self.create_request("BUCKET_DELETE", bucket = bucket, uri_params = {'website': None}) response = self.send_request(request) debug("Received response '%s'" % (response)) if response['status'] != 204: raise S3ResponseError("Expected status 204: %s" % response) return response def expiration_info(self, uri, bucket_location = None): bucket = uri.bucket() request = self.create_request("BUCKET_LIST", bucket=bucket, uri_params={'lifecycle': None}) try: response = self.send_request(request) except S3Error as e: if e.status == 404: debug("Could not get /?lifecycle - lifecycle probably not " "configured for this bucket") return None elif e.status == 501: debug("Could not get /?lifecycle - lifecycle support not " "implemented by the server") return None raise root_tag_name = getRootTagName(response['data']) if root_tag_name != "LifecycleConfiguration": debug("Could not get /?lifecycle - unexpected xml response: " "%s", root_tag_name) return None response['prefix'] = getTextFromXml(response['data'], ".//Rule//Prefix") response['date'] = getTextFromXml(response['data'], ".//Rule//Expiration//Date") response['days'] = getTextFromXml(response['data'], ".//Rule//Expiration//Days") return response def expiration_set(self, uri, bucket_location = None): if self.config.expiry_date and self.config.expiry_days: raise ParameterError("Expect either --expiry-day or --expiry-date") if not (self.config.expiry_date or self.config.expiry_days): if self.config.expiry_prefix: raise ParameterError("Expect either --expiry-day or --expiry-date") debug("del bucket lifecycle") bucket = uri.bucket() request = self.create_request("BUCKET_DELETE", bucket = bucket, uri_params = {'lifecycle': None}) else: request = self._expiration_set(uri) response = self.send_request(request) debug("Received response '%s'" % (response)) return response def _expiration_set(self, uri): debug("put bucket lifecycle") body = '' body += ' ' body += (' %s' % self.config.expiry_prefix) body += (' Enabled') body += (' ') if self.config.expiry_date: body += (' %s' % self.config.expiry_date) elif self.config.expiry_days: body += (' %s' % self.config.expiry_days) body += (' ') body += ' ' body += '' headers = SortedDict(ignore_case = True) headers['content-md5'] = compute_content_md5(body) bucket = uri.bucket() request = self.create_request("BUCKET_CREATE", bucket = bucket, headers = headers, body = body, uri_params = {'lifecycle': None}) return (request) def _guess_content_type(self, filename): content_type = self.config.default_mime_type content_charset = None if filename == "-" and not self.config.default_mime_type: raise ParameterError("You must specify --mime-type or --default-mime-type for files uploaded from stdin.") if self.config.guess_mime_type: if self.config.follow_symlinks: filename = unicodise(os.path.realpath(deunicodise(filename))) if self.config.use_mime_magic: (content_type, content_charset) = mime_magic(filename) else: (content_type, content_charset) = mimetypes.guess_type(filename) if not content_type: content_type = self.config.default_mime_type return (content_type, content_charset) def stdin_content_type(self): content_type = self.config.mime_type if not content_type: content_type = self.config.default_mime_type content_type += "; charset=" + self.config.encoding.upper() return content_type def content_type(self, filename=None): # explicit command line argument always wins content_type = self.config.mime_type content_charset = None if filename == u'-': return self.stdin_content_type() if not content_type: (content_type, content_charset) = self._guess_content_type(filename) ## add charset to content type if not content_charset: content_charset = self.config.encoding.upper() if self.add_encoding(filename, content_type) and content_charset is not None: content_type = content_type + "; charset=" + content_charset return content_type def add_encoding(self, filename, content_type): if 'charset=' in content_type: return False exts = self.config.add_encoding_exts.split(',') if exts[0]=='': return False parts = filename.rsplit('.',2) if len(parts) < 2: return False ext = parts[1] if ext in exts: return True else: return False def object_put(self, filename, uri, extra_headers = None, extra_label = ""): # TODO TODO # Make it consistent with stream-oriented object_get() if uri.type != "s3": raise ValueError("Expected URI type 's3', got '%s'" % uri.type) if filename != "-" and not os.path.isfile(deunicodise(filename)): raise InvalidFileError(u"Not a regular file") try: if filename == "-": src_stream = io.open(sys.stdin.fileno(), mode='rb', closefd=False) src_stream.stream_name = u'' size = 0 else: src_stream = io.open(deunicodise(filename), mode='rb') src_stream.stream_name = filename size = os.stat(deunicodise(filename))[ST_SIZE] except (IOError, OSError) as e: raise InvalidFileError(u"%s" % e.strerror) headers = SortedDict(ignore_case = True) if extra_headers: headers.update(extra_headers) ## Set server side encryption if self.config.server_side_encryption: headers["x-amz-server-side-encryption"] = "AES256" ## Set kms headers if self.config.kms_key: headers['x-amz-server-side-encryption'] = 'aws:kms' headers['x-amz-server-side-encryption-aws-kms-key-id'] = self.config.kms_key ## MIME-type handling headers["content-type"] = self.content_type(filename=filename) ## Other Amazon S3 attributes if self.config.acl_public: headers["x-amz-acl"] = "public-read" headers["x-amz-storage-class"] = self.storage_class() ## Multipart decision multipart = False if not self.config.enable_multipart and filename == "-": raise ParameterError("Multi-part upload is required to upload from stdin") if self.config.enable_multipart: if size > self.config.multipart_chunk_size_mb * SIZE_1MB or filename == "-": multipart = True if size > self.config.multipart_max_chunks * self.config.multipart_chunk_size_mb * SIZE_1MB: raise ParameterError("Chunk size %d MB results in more than %d chunks. Please increase --multipart-chunk-size-mb" % \ (self.config.multipart_chunk_size_mb, self.config.multipart_max_chunks)) if multipart: # Multipart requests are quite different... drop here return self.send_file_multipart(src_stream, headers, uri, size, extra_label) ## Not multipart... if self.config.put_continue: # Note, if input was stdin, we would be performing multipart upload. # So this will always work as long as the file already uploaded was # not uploaded via MultiUpload, in which case its ETag will not be # an md5. try: info = self.object_info(uri) except Exception: info = None if info is not None: remote_size = int(info['headers']['content-length']) remote_checksum = info['headers']['etag'].strip('"\'') if size == remote_size: checksum = calculateChecksum('', src_stream, 0, size, self.config.send_chunk) if remote_checksum == checksum: warning("Put: size and md5sum match for %s, skipping." % uri) return else: warning("MultiPart: checksum (%s vs %s) does not match for %s, reuploading." % (remote_checksum, checksum, uri)) else: warning("MultiPart: size (%d vs %d) does not match for %s, reuploading." % (remote_size, size, uri)) headers["content-length"] = str(size) request = self.create_request("OBJECT_PUT", uri = uri, headers = headers) labels = { 'source' : filename, 'destination' : uri.uri(), 'extra' : extra_label } response = self.send_file(request, src_stream, labels) return response def object_get(self, uri, stream, dest_name, start_position = 0, extra_label = ""): if uri.type != "s3": raise ValueError("Expected URI type 's3', got '%s'" % uri.type) request = self.create_request("OBJECT_GET", uri = uri) labels = { 'source' : uri.uri(), 'destination' : dest_name, 'extra' : extra_label } response = self.recv_file(request, stream, labels, start_position) return response def object_batch_delete(self, remote_list): """ Batch delete given a remote_list """ uris = [remote_list[item]['object_uri_str'] for item in remote_list] self.object_batch_delete_uri_strs(uris) def object_batch_delete_uri_strs(self, uris): """ Batch delete given a list of object uris """ def compose_batch_del_xml(bucket, key_list): body = u"" for key in key_list: uri = S3Uri(key) if uri.type != "s3": raise ValueError("Expected URI type 's3', got '%s'" % uri.type) if not uri.has_object(): raise ValueError("URI '%s' has no object" % key) if uri.bucket() != bucket: raise ValueError("The batch should contain keys from the same bucket") object = saxutils.escape(uri.object()) body += u"%s" % object body += u"" body = encode_to_s3(body) return body batch = uris if len(batch) == 0: raise ValueError("Key list is empty") bucket = S3Uri(batch[0]).bucket() request_body = compose_batch_del_xml(bucket, batch) headers = SortedDict({'content-md5': compute_content_md5(request_body), 'content-type': 'application/xml'}, ignore_case=True) request = self.create_request("BATCH_DELETE", bucket = bucket, headers = headers, body = request_body, uri_params = {'delete': None}) response = self.send_request(request) return response def object_delete(self, uri): if uri.type != "s3": raise ValueError("Expected URI type 's3', got '%s'" % uri.type) request = self.create_request("OBJECT_DELETE", uri = uri) response = self.send_request(request) return response def object_restore(self, uri): if uri.type != "s3": raise ValueError("Expected URI type 's3', got '%s'" % uri.type) if self.config.restore_days < 1: raise ParameterError("You must restore a file for 1 or more days") if self.config.restore_priority not in ['Standard', 'Expedited', 'Bulk']: raise ParameterError("Valid restoration priorities: bulk, standard, expedited") body = '' body += (' %s' % self.config.restore_days) body += ' ' body += (' %s' % self.config.restore_priority) body += ' ' body += '' request = self.create_request("OBJECT_POST", uri = uri, body = body, uri_params = {'restore': None}) response = self.send_request(request) debug("Received response '%s'" % (response)) return response def _sanitize_headers(self, headers): to_remove = [ # from http://docs.aws.amazon.com/AmazonS3/latest/dev/UsingMetadata.html 'date', 'content-length', 'last-modified', 'content-md5', 'x-amz-version-id', 'x-amz-delete-marker', # other headers returned from object_info() we don't want to send 'accept-ranges', 'connection', 'etag', 'server', 'x-amz-id-2', 'x-amz-request-id', # Other headers that are not copying by a direct copy 'x-amz-storage-class', ## We should probably also add server-side encryption headers ] for h in to_remove + self.config.remove_headers: if h.lower() in headers: del headers[h.lower()] return headers def object_copy(self, src_uri, dst_uri, extra_headers=None, src_size=None, extra_label="", replace_meta=False): """Remote copy an object and eventually set metadata Note: A little memo description of the nightmare for performance here: ** FOR AWS, 2 cases: - COPY will copy the metadata of the source to dest, but you can't modify them. Any additional header will be ignored anyway. - REPLACE will set the additional metadata headers that are provided but will not copy any of the source headers. So, to add to existing meta during copy, you have to do an object_info to get original source headers, then modify, then use REPLACE for the copy operation. ** For Minio and maybe other implementations: - if additional headers are sent, they will be set to the destination on top of source original meta in all cases COPY and REPLACE. It is a nice behavior except that it is different of the aws one. As it was still too easy, there is another catch: In all cases, for multipart copies, metadata data are never copied from the source. """ if src_uri.type != "s3": raise ValueError("Expected URI type 's3', got '%s'" % src_uri.type) if dst_uri.type != "s3": raise ValueError("Expected URI type 's3', got '%s'" % dst_uri.type) if self.config.acl_public is None: try: acl = self.get_acl(src_uri) except S3Error as exc: # Ignore the exception and don't fail the copy # if the server doesn't support setting ACLs if exc.status != 501: raise exc acl = None multipart = False headers = None if extra_headers or self.config.mime_type: # Force replace, that will force getting meta with object_info() replace_meta = True if replace_meta: src_info = self.object_info(src_uri) headers = src_info['headers'] src_size = int(headers["content-length"]) if self.config.enable_multipart: # Get size of remote source only if multipart is enabled and that no # size info was provided src_headers = headers if src_size is None: src_info = self.object_info(src_uri) src_headers = src_info['headers'] src_size = int(src_headers["content-length"]) # If we are over the grand maximum size for a normal copy/modify # (> 5GB) go nuclear and use multipart copy as the only option to # modify an object. # Reason is an aws s3 design bug. See: # https://github.com/aws/aws-sdk-java/issues/367 if src_uri is dst_uri: # optimisation in the case of modify threshold = MultiPartUpload.MAX_CHUNK_SIZE_MB * SIZE_1MB else: threshold = self.config.multipart_copy_chunk_size_mb * SIZE_1MB if src_size > threshold: # Sadly, s3 has a bad logic as metadata will not be copied for # multipart copy unlike what is done for direct copies. # TODO: Optimize by re-using the object_info request done # earlier earlier at fetch remote stage, and preserve headers. if src_headers is None: src_info = self.object_info(src_uri) src_headers = src_info['headers'] src_size = int(src_headers["content-length"]) headers = src_headers multipart = True if headers: self._sanitize_headers(headers) headers = SortedDict(headers, ignore_case=True) else: headers = SortedDict(ignore_case=True) # Following meta data are updated even in COPY by aws if self.config.acl_public: headers["x-amz-acl"] = "public-read" headers["x-amz-storage-class"] = self.storage_class() ## Set server side encryption if self.config.server_side_encryption: headers["x-amz-server-side-encryption"] = "AES256" ## Set kms headers if self.config.kms_key: headers['x-amz-server-side-encryption'] = 'aws:kms' headers['x-amz-server-side-encryption-aws-kms-key-id'] = \ self.config.kms_key # Following meta data are not updated in simple COPY by aws. if extra_headers: headers.update(extra_headers) if self.config.mime_type: headers["content-type"] = self.config.mime_type # "COPY" or "REPLACE" if not replace_meta: headers['x-amz-metadata-directive'] = "COPY" else: headers['x-amz-metadata-directive'] = "REPLACE" if multipart: # Multipart decision. Only do multipart copy for remote s3 files # bigger than the multipart copy threshold. # Multipart requests are quite different... delegate response = self.copy_file_multipart(src_uri, dst_uri, src_size, headers, extra_label) else: # Not multipart... direct request headers['x-amz-copy-source'] = s3_quote( "/%s/%s" % (src_uri.bucket(), src_uri.object()), quote_backslashes=False, unicode_output=True) request = self.create_request("OBJECT_PUT", uri=dst_uri, headers=headers) response = self.send_request(request) if response["data"] and getRootTagName(response["data"]) == "Error": # http://doc.s3.amazonaws.com/proposals/copy.html # Error during copy, status will be 200, so force error code 500 response["status"] = 500 error("Server error during the COPY operation. Overwrite response " "status to 500") raise S3Error(response) if self.config.acl_public is None and acl: try: self.set_acl(dst_uri, acl) except S3Error as exc: # Ignore the exception and don't fail the copy # if the server doesn't support setting ACLs if exc.status != 501: raise exc return response def object_modify(self, src_uri, dst_uri, extra_headers=None, src_size=None, extra_label=""): # dst_uri = src_uri Will optimize by using multipart just in worst case return self.object_copy(src_uri, src_uri, extra_headers, src_size, extra_label, replace_meta=True) def object_move(self, src_uri, dst_uri, extra_headers=None, src_size=None, extra_label=""): response_copy = self.object_copy(src_uri, dst_uri, extra_headers, src_size, extra_label) debug("Object %s copied to %s" % (src_uri, dst_uri)) if not response_copy["data"] \ or getRootTagName(response_copy["data"]) \ in ["CopyObjectResult", "CompleteMultipartUploadResult"]: self.object_delete(src_uri) debug("Object '%s' deleted", src_uri) else: warning("Object '%s' NOT deleted because of an unexpected " "response data content.", src_uri) return response_copy def object_info(self, uri): request = self.create_request("OBJECT_HEAD", uri=uri) response = self.send_request(request) return response def get_acl(self, uri): if uri.has_object(): request = self.create_request("OBJECT_GET", uri=uri, uri_params={'acl': None}) else: request = self.create_request("BUCKET_LIST", bucket=uri.bucket(), uri_params={'acl': None}) response = self.send_request(request) acl = ACL(response['data']) return acl def set_acl(self, uri, acl): body = u"%s"% acl debug(u"set_acl(%s): acl-xml: %s" % (uri, body)) headers = SortedDict({'content-type': 'application/xml'}, ignore_case = True) if uri.has_object(): request = self.create_request("OBJECT_PUT", uri = uri, headers = headers, body = body, uri_params = {'acl': None}) else: request = self.create_request("BUCKET_CREATE", bucket = uri.bucket(), headers = headers, body = body, uri_params = {'acl': None}) response = self.send_request(request) return response def get_policy(self, uri): request = self.create_request("BUCKET_LIST", bucket = uri.bucket(), uri_params = {'policy': None}) response = self.send_request(request) return decode_from_s3(response['data']) def set_policy(self, uri, policy): headers = SortedDict(ignore_case = True) # TODO check policy is proper json string headers['content-type'] = 'application/json' request = self.create_request("BUCKET_CREATE", uri = uri, headers=headers, body = policy, uri_params = {'policy': None}) response = self.send_request(request) return response def delete_policy(self, uri): request = self.create_request("BUCKET_DELETE", uri = uri, uri_params = {'policy': None}) debug(u"delete_policy(%s)" % uri) response = self.send_request(request) return response def get_cors(self, uri): request = self.create_request("BUCKET_LIST", bucket = uri.bucket(), uri_params = {'cors': None}) response = self.send_request(request) return decode_from_s3(response['data']) def set_cors(self, uri, cors): headers = SortedDict(ignore_case = True) # TODO check cors is proper json string headers['content-type'] = 'application/xml' headers['content-md5'] = compute_content_md5(cors) request = self.create_request("BUCKET_CREATE", uri = uri, headers=headers, body = cors, uri_params = {'cors': None}) response = self.send_request(request) return response def delete_cors(self, uri): request = self.create_request("BUCKET_DELETE", uri = uri, uri_params = {'cors': None}) debug(u"delete_cors(%s)" % uri) response = self.send_request(request) return response def set_lifecycle_policy(self, uri, policy): headers = SortedDict(ignore_case = True) headers['content-md5'] = compute_content_md5(policy) request = self.create_request("BUCKET_CREATE", uri = uri, headers=headers, body = policy, uri_params = {'lifecycle': None}) debug(u"set_lifecycle_policy(%s): policy-xml: %s" % (uri, policy)) response = self.send_request(request) return response def set_payer(self, uri): headers = SortedDict(ignore_case = True) headers['content-type'] = 'application/xml' body = '\n' if self.config.requester_pays: body += 'Requester\n' else: body += 'BucketOwner\n' body += '\n' request = self.create_request("BUCKET_CREATE", uri = uri, body = body, uri_params = {'requestPayment': None}) response = self.send_request(request) return response def get_lifecycle_policy(self, uri): request = self.create_request("BUCKET_LIST", bucket = uri.bucket(), uri_params = {'lifecycle': None}) debug(u"get_lifecycle_policy(%s)" % uri) response = self.send_request(request) debug(u"%s: Got Lifecycle Policy" % response['status']) return response def delete_lifecycle_policy(self, uri): request = self.create_request("BUCKET_DELETE", uri = uri, uri_params = {'lifecycle': None}) debug(u"delete_lifecycle_policy(%s)" % uri) response = self.send_request(request) return response def get_multipart(self, uri, uri_params=None, limit=-1): upload_list = [] for truncated, uploads in self.get_multipart_streaming(uri, uri_params, limit): upload_list.extend(uploads) return upload_list def get_multipart_streaming(self, uri, uri_params=None, limit=-1): uri_params = uri_params and uri_params.copy() or {} bucket = uri.bucket() truncated = True num_objects = 0 max_keys = limit # It is the "uploads: None" in uri_params that will change the # behavior of bucket_list to return multiparts instead of keys uri_params['uploads'] = None while truncated: response = self.bucket_list_noparse(bucket, recursive=True, uri_params=uri_params, max_keys=max_keys) xml_data = response["data"] # extract list of info of uploads upload_list = getListFromXml(xml_data, "Upload") num_objects += len(upload_list) if limit > num_objects: max_keys = limit - num_objects xml_truncated = getTextFromXml(xml_data, ".//IsTruncated") if not xml_truncated or xml_truncated.lower() == "false": truncated = False if truncated: if limit == -1 or num_objects < limit: if upload_list: next_key = getTextFromXml(xml_data, "NextKeyMarker") if not next_key: next_key = upload_list[-1]["Key"] uri_params['KeyMarker'] = next_key upload_id_marker = getTextFromXml( xml_data, "NextUploadIdMarker") if upload_id_marker: uri_params['UploadIdMarker'] = upload_id_marker elif 'UploadIdMarker' in uri_params: # Clear any pre-existing value del uri_params['UploadIdMarker'] else: # Unexpectedly, the server lied, and so the previous # response was not truncated. So, no new key to get. yield False, upload_list break debug("Listing continues after '%s'" % uri_params['KeyMarker']) else: yield truncated, upload_list break yield truncated, upload_list def list_multipart(self, uri, upload_id, uri_params=None, limit=-1): part_list = [] for truncated, parts in self.list_multipart_streaming(uri, upload_id, uri_params, limit): part_list.extend(parts) return part_list def list_multipart_streaming(self, uri, upload_id, uri_params=None, limit=-1): uri_params = uri_params and uri_params.copy() or {} truncated = True num_objects = 0 max_parts = limit while truncated: response = self.list_multipart_noparse(uri, upload_id, uri_params, max_parts) xml_data = response["data"] # extract list of multipart upload parts part_list = getListFromXml(xml_data, "Part") num_objects += len(part_list) if limit > num_objects: max_parts = limit - num_objects xml_truncated = getTextFromXml(xml_data, ".//IsTruncated") if not xml_truncated or xml_truncated.lower() == "false": truncated = False if truncated: if limit == -1 or num_objects < limit: if part_list: next_part_number = getTextFromXml( xml_data, "NextPartNumberMarker") if not next_part_number: next_part_number = part_list[-1]["PartNumber"] uri_params['part-number-marker'] = next_part_number else: # Unexpectedly, the server lied, and so the previous # response was not truncated. So, no new part to get. yield False, part_list break debug("Listing continues after Part '%s'" % uri_params['part-number-marker']) else: yield truncated, part_list break yield truncated, part_list def list_multipart_noparse(self, uri, upload_id, uri_params=None, max_parts=-1): if uri_params is None: uri_params = {} if max_parts != -1: uri_params['max-parts'] = str(max_parts) uri_params['uploadId'] = upload_id request = self.create_request("OBJECT_GET", uri=uri, uri_params=uri_params) response = self.send_request(request) return response def abort_multipart(self, uri, id): request = self.create_request("OBJECT_DELETE", uri = uri, uri_params = {'uploadId': id}) response = self.send_request(request) return response def get_accesslog(self, uri): request = self.create_request("BUCKET_LIST", bucket = uri.bucket(), uri_params = {'logging': None}) response = self.send_request(request) accesslog = AccessLog(response['data']) return accesslog def set_accesslog_acl(self, uri): acl = self.get_acl(uri) debug("Current ACL(%s): %s" % (uri.uri(), acl)) acl.appendGrantee(GranteeLogDelivery("READ_ACP")) acl.appendGrantee(GranteeLogDelivery("WRITE")) debug("Updated ACL(%s): %s" % (uri.uri(), acl)) self.set_acl(uri, acl) def set_accesslog(self, uri, enable, log_target_prefix_uri = None, acl_public = False): accesslog = AccessLog() if enable: accesslog.enableLogging(log_target_prefix_uri) accesslog.setAclPublic(acl_public) else: accesslog.disableLogging() body = "%s" % accesslog debug(u"set_accesslog(%s): accesslog-xml: %s" % (uri, body)) request = self.create_request("BUCKET_CREATE", bucket = uri.bucket(), body = body, uri_params = {'logging': None}) try: response = self.send_request(request) except S3Error as e: if e.info['Code'] == "InvalidTargetBucketForLogging": info("Setting up log-delivery ACL for target bucket.") self.set_accesslog_acl(S3Uri(u"s3://%s" % log_target_prefix_uri.bucket())) response = self.send_request(request) else: raise return accesslog, response def create_request(self, operation, uri = None, bucket = None, object = None, headers = None, body = "", uri_params = None): resource = { 'bucket' : None, 'uri' : "/" } if uri and (bucket or object): raise ValueError("Both 'uri' and either 'bucket' or 'object' parameters supplied") ## If URI is given use that instead of bucket/object parameters if uri: bucket = uri.bucket() object = uri.has_object() and uri.object() or None if bucket: resource['bucket'] = bucket if object: resource['uri'] = "/" + object method_string = S3.http_methods.getkey(S3.operations[operation] & S3.http_methods["MASK"]) request = S3Request(self, method_string, resource, headers, body, uri_params) debug("CreateRequest: resource[uri]=%s", resource['uri']) return request def _fail_wait(self, retries): # Wait a few seconds. The more it fails the more we wait. return (self._max_retries - retries + 1) * 3 def _http_redirection_handler(self, request, response, fn, *args, **kwargs): # Region info might already be available through the x-amz-bucket-region header redir_region = response['headers'].get('x-amz-bucket-region') if 'data' in response and len(response['data']) > 0: redir_bucket = getTextFromXml(response['data'], ".//Bucket") redir_hostname = getTextFromXml(response['data'], ".//Endpoint") self.set_hostname(redir_bucket, redir_hostname) info(u'Redirected to: %s', redir_hostname) if redir_region: S3Request.region_map[redir_bucket] = redir_region info(u'Redirected to region: %s', redir_region) return fn(*args, **kwargs) elif request.method_string == 'HEAD': # Head is a special case, redirection info usually are in the body # but there is no body for an HEAD request. location_url = response['headers'].get('location') if location_url: # Sometimes a "location" http header could be available and # can help us deduce the redirection path. # It is the case of "dns-style" syntax, but not for "path-style" syntax. if location_url.startswith("http://"): location_url = location_url[7:] elif location_url.startswith("https://"): location_url = location_url[8:] location_url = urlparse('https://' + location_url).hostname redir_bucket = request.resource['bucket'] self.set_hostname(redir_bucket, location_url) info(u'Redirected to: %s', location_url) if redir_region: S3Request.region_map[redir_bucket] = redir_region info(u'Redirected to region: %s', redir_region) return fn(*args, **kwargs) warning(u'Redirection error: No info provided by the server to where should be forwarded the request (HEAD request). (Hint target region: %s)', redir_region) raise S3Error(response) def _http_400_handler(self, request, response, fn, *args, **kwargs): """ Returns None if no handler available for the specific error code """ # AWS response AuthorizationHeaderMalformed means we sent the request to the wrong region # get the right region out of the response and send it there. if 'data' in response and len(response['data']) > 0: failureCode = getTextFromXml(response['data'], 'Code') if failureCode == 'AuthorizationHeaderMalformed': # we sent the request to the wrong region region = getTextFromXml(response['data'], 'Region') if region is not None: S3Request.region_map[request.resource['bucket']] = region info('Forwarding request to %s', region) return fn(*args, **kwargs) else: warning(u'Could not determine bucket the location. Please consider using the --region parameter.') elif failureCode == 'InvalidRequest': message = getTextFromXml(response['data'], 'Message') if message == 'The authorization mechanism you have provided is not supported. Please use AWS4-HMAC-SHA256.': debug(u'Endpoint requires signature v4') self.endpoint_requires_signature_v4 = True return fn(*args, **kwargs) elif failureCode == 'InvalidArgument': # returned by DreamObjects on send_request and send_file, # which doesn't support signature v4. Retry with signature v2 if not request.use_signature_v2() and not self.fallback_to_signature_v2: # have not tried with v2 yet debug(u'Falling back to signature v2') self.fallback_to_signature_v2 = True return fn(*args, **kwargs) else: # returned by DreamObjects on recv_file, which doesn't support signature v4. Retry with signature v2 if not request.use_signature_v2() and not self.fallback_to_signature_v2: # have not tried with v2 yet debug(u'Falling back to signature v2') self.fallback_to_signature_v2 = True return fn(*args, **kwargs) return None def _http_403_handler(self, request, response, fn, *args, **kwargs): if 'data' in response and len(response['data']) > 0: failureCode = getTextFromXml(response['data'], 'Code') if failureCode == 'AccessDenied': # traditional HTTP 403 message = getTextFromXml(response['data'], 'Message') if message == 'AWS authentication requires a valid Date or x-amz-date header': # message from an Eucalyptus walrus server if not request.use_signature_v2() and not self.fallback_to_signature_v2: # have not tried with v2 yet debug(u'Falling back to signature v2') self.fallback_to_signature_v2 = True return fn(*args, **kwargs) raise S3Error(response) def update_region_inner_request(self, request): """Get and update region for the request if needed. Signature v4 needs the region of the bucket or the request will fail with the indication of the correct region. We are trying to avoid this failure by pre-emptively getting the correct region to use, if not provided by the user. """ if request.resource.get('bucket') and not request.use_signature_v2() \ and S3Request.region_map.get( request.resource['bucket'], Config().bucket_location ) == "US": debug("===== SEND Inner request to determine the bucket region " "=====") try: s3_uri = S3Uri(u's3://' + request.resource['bucket']) # "force_us_default" should prevent infinite recursivity because # it will set the region_map dict. region = self.get_bucket_location(s3_uri, force_us_default=True) if region is not None: S3Request.region_map[request.resource['bucket']] = region debug("===== SUCCESS Inner request to determine the bucket " "region (%r) =====", region) except Exception as exc: # Ignore errors, it is just an optimisation, so nothing critical debug("getlocation inner request failure reason: %s", exc) debug("===== FAILED Inner request to determine the bucket " "region =====") def send_request(self, request, retries = _max_retries): self.update_region_inner_request(request) request.body = encode_to_s3(request.body) headers = request.headers method_string, resource, headers = request.get_triplet() response = {} debug("Processing request, please wait...") conn = None try: conn = ConnMan.get(self.get_hostname(resource['bucket'])) # TODO: Check what was supposed to be the usage of conn.path here # Currently this is always "None" all the time as not defined in ConnMan uri = self.format_uri(resource, conn.path) debug("Sending request method_string=%r, uri=%r, headers=%r, body=(%i bytes)" % (method_string, uri, headers, len(request.body or ""))) conn.c.request(method_string, uri, request.body, headers) http_response = conn.c.getresponse() response["status"] = http_response.status response["reason"] = http_response.reason response["headers"] = convertHeaderTupleListToDict(http_response.getheaders()) response["data"] = http_response.read() if "x-amz-meta-s3cmd-attrs" in response["headers"]: attrs = parse_attrs_header(response["headers"]["x-amz-meta-s3cmd-attrs"]) response["s3cmd-attrs"] = attrs ConnMan.put(conn) except (S3SSLError, S3SSLCertificateError): # In case of failure to validate the certificate for a ssl # connection,no need to retry, abort immediately raise except (IOError, Exception) as e: debug("Response:\n" + pprint.pformat(response)) if ((hasattr(e, 'errno') and e.errno and e.errno not in (errno.EPIPE, errno.ECONNRESET, errno.ETIMEDOUT)) or "[Errno 104]" in str(e) or "[Errno 32]" in str(e) ) and not isinstance(e, SocketTimeoutException): raise # When the connection is broken, BadStatusLine is raised with py2 # and RemoteDisconnected is raised by py3 with a trap: # RemoteDisconnected has an errno field with a None value. # close the connection and re-establish ConnMan.close(conn) if retries: warning("Retrying failed request: %s (%s)" % (resource['uri'], e)) warning("Waiting %d sec..." % self._fail_wait(retries)) time.sleep(self._fail_wait(retries)) return self.send_request(request, retries - 1) else: raise S3RequestError("Request failed for: %s" % resource['uri']) except: # Only KeyboardInterrupt and SystemExit will not be covered by Exception debug("Response:\n" + pprint.pformat(response)) raise debug("Response:\n" + pprint.pformat(response)) if response["status"] in [301, 307]: ## RedirectTemporary or RedirectPermanent return self._http_redirection_handler(request, response, self.send_request, request) if response["status"] == 400: handler_fn = self._http_400_handler(request, response, self.send_request, request) if handler_fn: return handler_fn err = S3Error(response) if retries and err.code in ['BadDigest', 'OperationAborted', 'TokenRefreshRequired', 'RequestTimeout']: warning(u"Retrying failed request: %s (%s)" % (resource['uri'], err)) warning("Waiting %d sec..." % self._fail_wait(retries)) time.sleep(self._fail_wait(retries)) return self.send_request(request, retries - 1) raise err if response["status"] == 403: return self._http_403_handler(request, response, self.send_request, request) if response["status"] == 405: # Method Not Allowed. Don't retry. raise S3Error(response) if response["status"] >= 500 or response["status"] == 429: e = S3Error(response) if response["status"] == 501: ## NotImplemented server error - no need to retry retries = 0 if retries: warning(u"Retrying failed request: %s (%s)" % (resource['uri'], e)) warning("Waiting %d sec..." % self._fail_wait(retries)) time.sleep(self._fail_wait(retries)) return self.send_request(request, retries - 1) else: raise e if response["status"] < 200 or response["status"] > 299: raise S3Error(response) return response def send_request_with_progress(self, request, labels, operation_size=0): """Wrapper around send_request for slow requests. To be able to show progression for small requests """ if not self.config.progress_meter: info("Sending slow request, please wait...") return self.send_request(request) if 'action' not in labels: labels[u'action'] = u'request' progress = self.config.progress_class(labels, operation_size) try: response = self.send_request(request) except Exception as exc: progress.done("failed") raise progress.update(current_position=operation_size) progress.done("done") return response def send_file(self, request, stream, labels, buffer = '', throttle = 0, retries = _max_retries, offset = 0, chunk_size = -1, use_expect_continue = None): self.update_region_inner_request(request) if use_expect_continue is None: use_expect_continue = self.config.use_http_expect if self.expect_continue_not_supported and use_expect_continue: use_expect_continue = False headers = request.headers size_left = size_total = int(headers["content-length"]) filename = stream.stream_name if self.config.progress_meter: labels[u'action'] = u'upload' progress = self.config.progress_class(labels, size_total) else: info("Sending file '%s', please wait..." % filename) timestamp_start = time.time() if buffer: sha256_hash = checksum_sha256_buffer(buffer, offset, size_total) else: sha256_hash = checksum_sha256_file(filename, offset, size_total) request.body = sha256_hash if use_expect_continue: if not size_total: use_expect_continue = False else: headers['expect'] = '100-continue' method_string, resource, headers = request.get_triplet() try: conn = ConnMan.get(self.get_hostname(resource['bucket'])) conn.c.putrequest(method_string, self.format_uri(resource, conn.path)) for header in headers.keys(): conn.c.putheader(encode_to_s3(header), encode_to_s3(headers[header])) conn.c.endheaders() except ParameterError as e: raise except Exception as e: if self.config.progress_meter: progress.done("failed") if retries: warning("Retrying failed request: %s (%s)" % (resource['uri'], e)) warning("Waiting %d sec..." % self._fail_wait(retries)) time.sleep(self._fail_wait(retries)) # Connection error -> same throttle value return self.send_file(request, stream, labels, buffer, throttle, retries - 1, offset, chunk_size) else: raise S3UploadError("Upload failed for: %s" % resource['uri']) if buffer == '': stream.seek(offset) md5_hash = md5() try: http_response = None if use_expect_continue: # Wait for the 100-Continue before sending the content readable, writable, exceptional = select.select([conn.c.sock],[], [], EXPECT_CONTINUE_TIMEOUT) if readable: # 100-CONTINUE STATUS RECEIVED, get it before continuing. http_response = conn.c.getresponse() elif not writable and not exceptional: warning("HTTP Expect Continue feature disabled because of no reply of the server in %.2fs.", EXPECT_CONTINUE_TIMEOUT) self.expect_continue_not_supported = True use_expect_continue = False if not use_expect_continue or (http_response and http_response.status == ConnMan.CONTINUE): if http_response: # CONTINUE case. Reset the response http_response.read() conn.c._HTTPConnection__state = ConnMan._CS_REQ_SENT while (size_left > 0): #debug("SendFile: Reading up to %d bytes from '%s' - remaining bytes: %s" % (self.config.send_chunk, filename, size_left)) l = min(self.config.send_chunk, size_left) if buffer == '': data = stream.read(l) else: data = buffer if self.config.limitrate > 0: start_time = time.time() md5_hash.update(data) conn.c.wrapper_send_body(data) if self.config.progress_meter: progress.update(delta_position = len(data)) size_left -= len(data) #throttle limitrate_throttle = throttle if self.config.limitrate > 0: real_duration = time.time() - start_time expected_duration = float(l) / self.config.limitrate limitrate_throttle = max(expected_duration - real_duration, limitrate_throttle) if limitrate_throttle: time.sleep(min(limitrate_throttle, self.config.throttle_max)) md5_computed = md5_hash.hexdigest() http_response = conn.c.getresponse() response = {} response["status"] = http_response.status response["reason"] = http_response.reason response["headers"] = convertHeaderTupleListToDict(http_response.getheaders()) response["data"] = http_response.read() response["size"] = size_total ConnMan.put(conn) debug(u"Response:\n" + pprint.pformat(response)) except ParameterError as e: raise except Exception as e: if self.config.progress_meter: progress.done("failed") if retries: known_error = False if ((hasattr(e, 'errno') and e.errno and e.errno not in (errno.EPIPE, errno.ECONNRESET, errno.ETIMEDOUT)) or "[Errno 104]" in str(e) or "[Errno 32]" in str(e) ) and not isinstance(e, SocketTimeoutException): # We have to detect these errors by looking at the error string # Connection reset by peer and Broken pipe # The server broke the connection early with an error like # in a HTTP Expect Continue case even if asked nothing. try: http_response = conn.c.getresponse() response = {} response["status"] = http_response.status response["reason"] = http_response.reason response["headers"] = convertHeaderTupleListToDict(http_response.getheaders()) response["data"] = http_response.read() response["size"] = size_total known_error = True except Exception: error("Cannot retrieve any response status before encountering an EPIPE or ECONNRESET exception") if not known_error: warning("Upload failed: %s (%s)" % (resource['uri'], e)) warning("Waiting %d sec..." % self._fail_wait(retries)) time.sleep(self._fail_wait(retries)) # Connection error -> same throttle value return self.send_file(request, stream, labels, buffer, throttle, retries - 1, offset, chunk_size, use_expect_continue) else: debug("Giving up on '%s' %s" % (filename, e)) raise S3UploadError("Upload failed for: %s" % resource['uri']) timestamp_end = time.time() response["elapsed"] = timestamp_end - timestamp_start response["speed"] = response["elapsed"] and float(response["size"]) / response["elapsed"] or float(-1) if self.config.progress_meter: ## Finalising the upload takes some time -> update() progress meter ## to correct the average speed. Otherwise people will complain that ## 'progress' and response["speed"] are inconsistent ;-) progress.update() progress.done("done") if response["status"] in [301, 307]: ## RedirectTemporary or RedirectPermanent return self._http_redirection_handler(request, response, self.send_file, request, stream, labels, buffer, offset = offset, chunk_size = chunk_size, use_expect_continue = use_expect_continue) if response["status"] == 400: handler_fn = self._http_400_handler(request, response, self.send_file, request, stream, labels, buffer, offset = offset, chunk_size = chunk_size, use_expect_continue = use_expect_continue) if handler_fn: return handler_fn err = S3Error(response) if err.code not in ['BadDigest', 'OperationAborted', 'TokenRefreshRequired', 'RequestTimeout']: raise err # else the error will be handled later with a retry if response["status"] == 403: return self._http_403_handler(request, response, self.send_file, request, stream, labels, buffer, offset = offset, chunk_size = chunk_size, use_expect_continue = use_expect_continue) if response["status"] == 417 and retries: # Expect 100-continue not supported by proxy/server self.expect_continue_not_supported = True return self.send_file(request, stream, labels, buffer, throttle, retries - 1, offset, chunk_size, use_expect_continue = False) # S3 from time to time doesn't send ETag back in a response :-( # Force re-upload here. if 'etag' not in response['headers']: response['headers']['etag'] = '' if response["status"] < 200 or response["status"] > 299: try_retry = False if response["status"] >= 500: # AWS internal error - retry try_retry = True if response["status"] == 503: ## SlowDown error throttle = throttle and throttle * 5 or 0.01 elif response["status"] == 429: # Not an AWS error, but s3 compatible server possible error: # TooManyRequests/Busy/slowdown try_retry = True throttle = throttle and throttle * 5 or 0.01 elif response["status"] >= 400: err = S3Error(response) ## Retriable client error? if err.code in ['BadDigest', 'OperationAborted', 'TokenRefreshRequired', 'RequestTimeout']: try_retry = True if try_retry: if retries: warning("Upload failed: %s (%s)" % (resource['uri'], S3Error(response))) if throttle: warning("Retrying on lower speed (throttle=%0.2f)" % throttle) warning("Waiting %d sec..." % self._fail_wait(retries)) time.sleep(self._fail_wait(retries)) return self.send_file(request, stream, labels, buffer, throttle, retries - 1, offset, chunk_size, use_expect_continue) else: warning("Too many failures. Giving up on '%s'" % filename) raise S3UploadError("Too many failures. Giving up on '%s'" % filename) ## Non-recoverable error raise S3Error(response) debug("MD5 sums: computed=%s, received=%s" % (md5_computed, response["headers"].get('etag', '').strip('"\''))) ## when using KMS encryption, MD5 etag value will not match md5_from_s3 = response["headers"].get("etag", "").strip('"\'') if ('-' not in md5_from_s3) and (md5_from_s3 != md5_hash.hexdigest()) and response["headers"].get("x-amz-server-side-encryption") != 'aws:kms': warning("MD5 Sums don't match!") if retries: warning("Retrying upload of %s" % (filename)) return self.send_file(request, stream, labels, buffer, throttle, retries - 1, offset, chunk_size, use_expect_continue) else: warning("Too many failures. Giving up on '%s'" % (filename)) raise S3UploadError("Too many failures. Giving up on '%s'" % filename) return response def send_file_multipart(self, stream, headers, uri, size, extra_label=""): timestamp_start = time.time() upload = MultiPartUpload(self, stream, uri, headers, size) upload.upload_all_parts(extra_label) response = upload.complete_multipart_upload() timestamp_end = time.time() response["elapsed"] = timestamp_end - timestamp_start response["size"] = size response["speed"] = response["elapsed"] and float(response["size"]) / response["elapsed"] or float(-1) if response["data"] and getRootTagName(response["data"]) == "Error": #http://docs.aws.amazon.com/AmazonS3/latest/API/mpUploadComplete.html # Error Complete Multipart UPLOAD, status may be 200 # raise S3UploadError raise S3UploadError(getTextFromXml(response["data"], 'Message')) return response def copy_file_multipart(self, src_uri, dst_uri, size, headers, extra_label=""): return self.send_file_multipart(src_uri, headers, dst_uri, size, extra_label) def recv_file(self, request, stream, labels, start_position = 0, retries = _max_retries): self.update_region_inner_request(request) method_string, resource, headers = request.get_triplet() filename = stream.stream_name if self.config.progress_meter: labels[u'action'] = u'download' progress = self.config.progress_class(labels, 0) else: info("Receiving file '%s', please wait..." % filename) timestamp_start = time.time() conn = None try: conn = ConnMan.get(self.get_hostname(resource['bucket'])) conn.c.putrequest(method_string, self.format_uri(resource, conn.path)) for header in headers.keys(): conn.c.putheader(encode_to_s3(header), encode_to_s3(headers[header])) if start_position > 0: debug("Requesting Range: %d .. end" % start_position) conn.c.putheader("Range", "bytes=%d-" % start_position) conn.c.endheaders() response = {} http_response = conn.c.getresponse() response["status"] = http_response.status response["reason"] = http_response.reason response["headers"] = convertHeaderTupleListToDict(http_response.getheaders()) if "x-amz-meta-s3cmd-attrs" in response["headers"]: attrs = parse_attrs_header(response["headers"]["x-amz-meta-s3cmd-attrs"]) response["s3cmd-attrs"] = attrs debug("Response:\n" + pprint.pformat(response)) except ParameterError as e: raise except (IOError, Exception) as e: if self.config.progress_meter: progress.done("failed") if ((hasattr(e, 'errno') and e.errno and e.errno not in (errno.EPIPE, errno.ECONNRESET, errno.ETIMEDOUT)) or "[Errno 104]" in str(e) or "[Errno 32]" in str(e) ) and not isinstance(e, SocketTimeoutException): raise # close the connection and re-establish ConnMan.close(conn) if retries: warning("Retrying failed request: %s (%s)" % (resource['uri'], e)) warning("Waiting %d sec..." % self._fail_wait(retries)) time.sleep(self._fail_wait(retries)) # Connection error -> same throttle value return self.recv_file(request, stream, labels, start_position, retries - 1) else: raise S3DownloadError("Download failed for: %s" % resource['uri']) if response["status"] in [301, 307]: ## RedirectPermanent or RedirectTemporary response['data'] = http_response.read() return self._http_redirection_handler(request, response, self.recv_file, request, stream, labels, start_position) if response["status"] == 400: response['data'] = http_response.read() handler_fn = self._http_400_handler(request, response, self.recv_file, request, stream, labels, start_position) if handler_fn: return handler_fn raise S3Error(response) if response["status"] == 403: response['data'] = http_response.read() return self._http_403_handler(request, response, self.recv_file, request, stream, labels, start_position) if response["status"] == 405: # Method Not Allowed. Don't retry. response['data'] = http_response.read() raise S3Error(response) if response["status"] < 200 or response["status"] > 299: response['data'] = http_response.read() raise S3Error(response) if start_position == 0: # Only compute MD5 on the fly if we're downloading from beginning # Otherwise we'd get a nonsense. md5_hash = md5() size_left = int(response["headers"]["content-length"]) size_total = start_position + size_left current_position = start_position if self.config.progress_meter: progress.total_size = size_total progress.initial_position = current_position progress.current_position = current_position try: # Fix for issue #432. Even when content size is 0, httplib expect the response to be read. if size_left == 0: data = http_response.read(1) # It is not supposed to be some data returned in that case assert(len(data) == 0) while (current_position < size_total): this_chunk = size_left > self.config.recv_chunk and self.config.recv_chunk or size_left if self.config.limitrate > 0: start_time = time.time() data = http_response.read(this_chunk) if len(data) == 0: raise S3ResponseError("EOF from S3!") #throttle if self.config.limitrate > 0: real_duration = time.time() - start_time expected_duration = float(this_chunk) / self.config.limitrate if expected_duration > real_duration: time.sleep(expected_duration - real_duration) stream.write(data) if start_position == 0: md5_hash.update(data) current_position += len(data) ## Call progress meter from here... if self.config.progress_meter: progress.update(delta_position = len(data)) ConnMan.put(conn) except OSError: raise except (IOError, Exception) as e: if self.config.progress_meter: progress.done("failed") if ((hasattr(e, 'errno') and e.errno and e.errno not in (errno.EPIPE, errno.ECONNRESET, errno.ETIMEDOUT)) or "[Errno 104]" in str(e) or "[Errno 32]" in str(e) ) and not isinstance(e, SocketTimeoutException): raise # close the connection and re-establish ConnMan.close(conn) if retries: warning("Retrying failed request: %s (%s)" % (resource['uri'], e)) warning("Waiting %d sec..." % self._fail_wait(retries)) time.sleep(self._fail_wait(retries)) # Connection error -> same throttle value return self.recv_file(request, stream, labels, current_position, retries - 1) else: raise S3DownloadError("Download failed for: %s" % resource['uri']) stream.flush() timestamp_end = time.time() if self.config.progress_meter: ## The above stream.flush() may take some time -> update() progress meter ## to correct the average speed. Otherwise people will complain that ## 'progress' and response["speed"] are inconsistent ;-) progress.update() progress.done("done") md5_from_s3 = response["headers"].get("etag", "").strip('"\'') if not 'x-amz-meta-s3tools-gpgenc' in response["headers"]: # we can't trust our stored md5 because we # encrypted the file after calculating it but before # uploading it. try: md5_from_s3 = response["s3cmd-attrs"]["md5"] except KeyError: pass # we must have something to compare against to bother with the calculation if '-' not in md5_from_s3: if start_position == 0: # Only compute MD5 on the fly if we were downloading from the beginning response["md5"] = md5_hash.hexdigest() else: # Otherwise try to compute MD5 of the output file try: response["md5"] = hash_file_md5(filename) except IOError as e: if e.errno != errno.ENOENT: warning("Unable to open file: %s: %s" % (filename, e)) warning("Unable to verify MD5. Assume it matches.") response["md5match"] = response.get("md5") == md5_from_s3 response["elapsed"] = timestamp_end - timestamp_start response["size"] = current_position response["speed"] = response["elapsed"] and float(response["size"]) / response["elapsed"] or float(-1) if response["size"] != start_position + int(response["headers"]["content-length"]): warning("Reported size (%s) does not match received size (%s)" % ( start_position + int(response["headers"]["content-length"]), response["size"])) debug("ReceiveFile: Computed MD5 = %s" % response.get("md5")) # avoid ETags from multipart uploads that aren't the real md5 if ('-' not in md5_from_s3 and not response["md5match"]) and (response["headers"].get("x-amz-server-side-encryption") != 'aws:kms'): warning("MD5 signatures do not match: computed=%s, received=%s" % ( response.get("md5"), md5_from_s3)) return response __all__.append("S3") def parse_attrs_header(attrs_header): attrs = {} for attr in attrs_header.split("/"): key, val = attr.split(":") attrs[key] = val return attrs def compute_content_md5(body): m = md5(encode_to_s3(body)) base64md5 = base64.encodestring(m.digest()) base64md5 = decode_from_s3(base64md5) if base64md5[-1] == '\n': base64md5 = base64md5[0:-1] return decode_from_s3(base64md5) # vim:et:ts=4:sts=4:ai