Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified
📁 ..	-	2026-03-21 15:56
📄 ACL.py # -- coding: utf-8 -- ## -------------------------------------------------------------------- ## Amazon S3 - Access Control List representation ## ## Authors : Michal Ludvig <michal@logix.cz> (https://www.logix.cz/michal) ## Florent Viard <florent@sodria.com> (https://www.sodria.com) ## Copyright : TGRMN Software, Sodria SAS and contributors ## License : GPL Version 2 ## Website : https://s3tools.org ## -------------------------------------------------------------------- from __future__ import absolute_import, print_function import sys from .BaseUtils import getTreeFromXml, encode_to_s3, decode_from_s3 from .Utils import deunicodise try: import xml.etree.ElementTree as ET except ImportError: import elementtree.ElementTree as ET PY3 = (sys.version_info >= (3, 0)) class Grantee(object): ALL_USERS_URI = "http://acs.amazonaws.com/groups/global/AllUsers" LOG_DELIVERY_URI = "http://acs.amazonaws.com/groups/s3/LogDelivery" def __init__(self): self.xsi_type = None self.tag = None self.name = None self.display_name = '' self.permission = None def __repr__(self): return repr('Grantee("%(tag)s", "%(name)s", "%(permission)s")' % { "tag" : self.tag, "name" : self.name, "permission" : self.permission }) def isAllUsers(self): return self.tag == "URI" and self.name == Grantee.ALL_USERS_URI def isAnonRead(self): return self.isAllUsers() and (self.permission == "READ" or self.permission == "FULL_CONTROL") def isAnonWrite(self): return self.isAllUsers() and (self.permission == "WRITE" or self.permission == "FULL_CONTROL") def getElement(self): el = ET.Element("Grant") grantee = ET.SubElement(el, "Grantee", { 'xmlns:xsi' : 'http://www.w3.org/2001/XMLSchema-instance', 'xsi:type' : self.xsi_type }) name = ET.SubElement(grantee, self.tag) name.text = self.name permission = ET.SubElement(el, "Permission") permission.text = self.permission return el class GranteeAnonRead(Grantee): def __init__(self): Grantee.__init__(self) self.xsi_type = "Group" self.tag = "URI" self.name = Grantee.ALL_USERS_URI self.permission = "READ" class GranteeLogDelivery(Grantee): def __init__(self, permission): """ permission must be either READ_ACP or WRITE """ Grantee.__init__(self) self.xsi_type = "Group" self.tag = "URI" self.name = Grantee.LOG_DELIVERY_URI self.permission = permission class ACL(object): EMPTY_ACL = b"<AccessControlPolicy><Owner><ID></ID></Owner><AccessControlList></AccessControlList></AccessControlPolicy>" def __init__(self, xml = None): if not xml: xml = ACL.EMPTY_ACL self.grantees = [] self.owner_id = "" self.owner_nick = "" tree = getTreeFromXml(encode_to_s3(xml)) self.parseOwner(tree) self.parseGrants(tree) def parseOwner(self, tree): self.owner_id = tree.findtext(".//Owner//ID") self.owner_nick = tree.findtext(".//Owner//DisplayName") def parseGrants(self, tree): for grant in tree.findall(".//Grant"): grantee = Grantee() g = grant.find(".//Grantee") grantee.xsi_type = g.attrib['{http://www.w3.org/2001/XMLSchema-instance}type'] grantee.permission = grant.find('Permission').text for el in g: if el.tag == "DisplayName": grantee.display_name = el.text else: grantee.tag = el.tag grantee.name = el.text self.grantees.append(grantee) def getGrantList(self): acl = [] for grantee in self.grantees: if grantee.display_name: user = grantee.display_name elif grantee.isAllUsers(): user = "anon" else: user = grantee.name acl.append({'grantee': user, 'permission': grantee.permission}) return acl def getOwner(self): return { 'id' : self.owner_id, 'nick' : self.owner_nick } def isAnonRead(self): for grantee in self.grantees: if grantee.isAnonRead(): return True return False def isAnonWrite(self): for grantee in self.grantees: if grantee.isAnonWrite(): return True return False def grantAnonRead(self): if not self.isAnonRead(): self.appendGrantee(GranteeAnonRead()) def revokeAnonRead(self): self.grantees = [g for g in self.grantees if not g.isAnonRead()] def revokeAnonWrite(self): self.grantees = [g for g in self.grantees if not g.isAnonWrite()] def appendGrantee(self, grantee): self.grantees.append(grantee) def hasGrant(self, name, permission): name = name.lower() permission = permission.upper() for grantee in self.grantees: if grantee.name.lower() == name: if grantee.permission == "FULL_CONTROL": return True elif grantee.permission.upper() == permission: return True return False def grant(self, name, permission): if self.hasGrant(name, permission): return permission = permission.upper() if "ALL" == permission: permission = "FULL_CONTROL" if "FULL_CONTROL" == permission: self.revoke(name, "ALL") grantee = Grantee() grantee.name = name grantee.permission = permission if '@' in name: grantee.name = grantee.name.lower() grantee.xsi_type = "AmazonCustomerByEmail" grantee.tag = "EmailAddress" elif 'http://acs.amazonaws.com/groups/' in name: grantee.xsi_type = "Group" grantee.tag = "URI" else: grantee.name = grantee.name.lower() grantee.xsi_type = "CanonicalUser" grantee.tag = "ID" self.appendGrantee(grantee) def revoke(self, name, permission): name = name.lower() permission = permission.upper() if "ALL" == permission: self.grantees = [g for g in self.grantees if not (g.name.lower() == name or (g.display_name is not None and g.display_name.lower() == name))] else: self.grantees = [g for g in self.grantees if not (((g.display_name is not None and g.display_name.lower() == name) or g.name.lower() == name) and g.permission.upper() == permission)] def get_printable_tree(self): tree = getTreeFromXml(ACL.EMPTY_ACL) tree.attrib['xmlns'] = "http://s3.amazonaws.com/doc/2006-03-01/" owner = tree.find(".//Owner//ID") owner.text = self.owner_id acl = tree.find(".//AccessControlList") for grantee in self.grantees: acl.append(grantee.getElement()) return tree def __unicode__(self): return decode_from_s3(ET.tostring(self.get_printable_tree())) def __str__(self): if PY3: # Return unicode return ET.tostring(self.get_printable_tree(), encoding="unicode") else: # Return bytes return ET.tostring(self.get_printable_tree()) if __name__ == "__main__": xml = b"""<?xml version="1.0" encoding="UTF-8"?> <AccessControlPolicy xmlns="http://s3.amazonaws.com/doc/2006-03-01/"> <Owner> <ID>12345678901234567890</ID> <DisplayName>owner-nickname</DisplayName> </Owner> <AccessControlList> <Grant> <Grantee xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:type="CanonicalUser"> <ID>12345678901234567890</ID> <DisplayName>owner-nickname</DisplayName> </Grantee> <Permission>FULL_CONTROL</Permission> </Grant> <Grant> <Grantee xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:type="Group"> <URI>http://acs.amazonaws.com/groups/global/AllUsers</URI> </Grantee> <Permission>READ</Permission> </Grant> </AccessControlList> </AccessControlPolicy> """ acl = ACL(xml) print("Grants:", acl.getGrantList()) acl.revokeAnonRead() print("Grants:", acl.getGrantList()) acl.grantAnonRead() print("Grants:", acl.getGrantList()) print(acl) # vim:et:ts=4:sts=4:ai	8.4 KB	2023-12-06 09:07
📄 AccessLog.py # -- coding: utf-8 -- ## -------------------------------------------------------------------- ## Amazon S3 - Access Control List representation ## ## Authors : Michal Ludvig <michal@logix.cz> (https://www.logix.cz/michal) ## Florent Viard <florent@sodria.com> (https://www.sodria.com) ## Copyright : TGRMN Software, Sodria SAS and contributors ## License : GPL Version 2 ## Website : https://s3tools.org ## -------------------------------------------------------------------- from __future__ import absolute_import, print_function import sys from . import S3Uri from .Exceptions import ParameterError from .BaseUtils import getTreeFromXml, decode_from_s3 from .ACL import GranteeAnonRead try: import xml.etree.ElementTree as ET except ImportError: import elementtree.ElementTree as ET PY3 = (sys.version_info >= (3,0)) __all__ = [] class AccessLog(object): LOG_DISABLED = "<BucketLoggingStatus></BucketLoggingStatus>" LOG_TEMPLATE = "<LoggingEnabled><TargetBucket></TargetBucket><TargetPrefix></TargetPrefix></LoggingEnabled>" def __init__(self, xml = None): if not xml: xml = self.LOG_DISABLED self.tree = getTreeFromXml(xml) self.tree.attrib['xmlns'] = "http://doc.s3.amazonaws.com/2006-03-01" def isLoggingEnabled(self): return (self.tree.find(".//LoggingEnabled") is not None) def disableLogging(self): el = self.tree.find(".//LoggingEnabled") if el: self.tree.remove(el) def enableLogging(self, target_prefix_uri): el = self.tree.find(".//LoggingEnabled") if not el: el = getTreeFromXml(self.LOG_TEMPLATE) self.tree.append(el) el.find(".//TargetBucket").text = target_prefix_uri.bucket() el.find(".//TargetPrefix").text = target_prefix_uri.object() def targetPrefix(self): if self.isLoggingEnabled(): target_prefix = u"s3://%s/%s" % ( self.tree.find(".//LoggingEnabled//TargetBucket").text, self.tree.find(".//LoggingEnabled//TargetPrefix").text) return S3Uri.S3Uri(target_prefix) else: return "" def setAclPublic(self, acl_public): le = self.tree.find(".//LoggingEnabled") if le is None: raise ParameterError("Logging not enabled, can't set default ACL for logs") tg = le.find(".//TargetGrants") if not acl_public: if not tg: ## All good, it's not been there return else: le.remove(tg) else: # acl_public == True anon_read = GranteeAnonRead().getElement() if not tg: tg = ET.SubElement(le, "TargetGrants") ## What if TargetGrants already exists? We should check if ## AnonRead is there before appending a new one. Later... tg.append(anon_read) def isAclPublic(self): raise NotImplementedError() def __unicode__(self): return decode_from_s3(ET.tostring(self.tree)) def __str__(self): if PY3: # Return unicode return ET.tostring(self.tree, encoding="unicode") else: # Return bytes return ET.tostring(self.tree) __all__.append("AccessLog") if __name__ == "__main__": log = AccessLog() print(log) log.enableLogging(S3Uri.S3Uri(u"s3://targetbucket/prefix/log-")) print(log) log.setAclPublic(True) print(log) log.setAclPublic(False) print(log) log.disableLogging() print(log) # vim:et:ts=4:sts=4:ai	3.54 KB	2023-12-06 09:07
📄 BaseUtils.py # -- coding: utf-8 -- ## -------------------------------------------------------------------- ## Amazon S3 manager ## ## Authors : Michal Ludvig <michal@logix.cz> (https://www.logix.cz/michal) ## Florent Viard <florent@sodria.com> (https://www.sodria.com) ## Copyright : TGRMN Software, Sodria SAS and contributors ## License : GPL Version 2 ## Website : https://s3tools.org ## -------------------------------------------------------------------- from __future__ import absolute_import, division import functools import re import posixpath import sys from calendar import timegm from hashlib import md5 from logging import debug, warning, error import xml.dom.minidom import xml.etree.ElementTree as ET from .ExitCodes import EX_OSFILE try: import dateutil.parser except ImportError: sys.stderr.write(u""" !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! ImportError trying to import dateutil.parser. Please install the python dateutil module: $ sudo apt-get install python-dateutil or $ sudo yum install python-dateutil or $ pip install python-dateutil !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! """) sys.stderr.flush() sys.exit(EX_OSFILE) try: from urllib import quote except ImportError: # python 3 support from urllib.parse import quote try: unicode = unicode except NameError: # python 3 support # In python 3, unicode -> str, and str -> bytes unicode = str __all__ = [] s3path = posixpath __all__.append("s3path") try: md5() except ValueError as exc: # md5 is disabled for FIPS-compliant Python builds. # Since s3cmd does not use md5 in a security context, # it is safe to allow the use of it by setting useforsecurity to False. try: md5(usedforsecurity=False) md5 = functools.partial(md5, usedforsecurity=False) except Exception: # "usedforsecurity" is only available on python >= 3.9 or RHEL distributions raise exc __all__.append("md5") RE_S3_DATESTRING = re.compile('\\.[0-9](?:[Z\\-\\+]?)') RE_XML_NAMESPACE = re.compile(b'^(<?[^>]+?>\\s\|\\s)(<\\w+) xmlns=[\'"](https?://[^\'"]+)[\'"]', re.MULTILINE) # Date and time helpers def dateS3toPython(date): # Reset milliseconds to 000 date = RE_S3_DATESTRING.sub(".000", date) return dateutil.parser.parse(date, fuzzy=True) __all__.append("dateS3toPython") def dateS3toUnix(date): ## NOTE: This is timezone-aware and return the timestamp regarding GMT return timegm(dateS3toPython(date).utctimetuple()) __all__.append("dateS3toUnix") def dateRFC822toPython(date): """ Convert a string formatted like '2020-06-27T15:56:34Z' into a python datetime """ return dateutil.parser.parse(date, fuzzy=True) __all__.append("dateRFC822toPython") def dateRFC822toUnix(date): return timegm(dateRFC822toPython(date).utctimetuple()) __all__.append("dateRFC822toUnix") def formatDateTime(s3timestamp): date_obj = dateutil.parser.parse(s3timestamp, fuzzy=True) return date_obj.strftime("%Y-%m-%d %H:%M") __all__.append("formatDateTime") # Encoding / Decoding def base_unicodise(string, encoding='UTF-8', errors='replace', silent=False): """ Convert 'string' to Unicode or raise an exception. """ if type(string) == unicode: return string if not silent: debug("Unicodising %r using %s" % (string, encoding)) try: return unicode(string, encoding, errors) except UnicodeDecodeError: raise UnicodeDecodeError("Conversion to unicode failed: %r" % string) __all__.append("base_unicodise") def base_deunicodise(string, encoding='UTF-8', errors='replace', silent=False): """ Convert unicode 'string' to <type str>, by default replacing all invalid characters with '?' or raise an exception. """ if type(string) != unicode: return string if not silent: debug("DeUnicodising %r using %s" % (string, encoding)) try: return string.encode(encoding, errors) except UnicodeEncodeError: raise UnicodeEncodeError("Conversion from unicode failed: %r" % string) __all__.append("base_deunicodise") def decode_from_s3(string, errors = "replace"): """ Convert S3 UTF-8 'string' to Unicode or raise an exception. """ return base_unicodise(string, "UTF-8", errors, True) __all__.append("decode_from_s3") def encode_to_s3(string, errors='replace'): """ Convert Unicode to S3 UTF-8 'string', by default replacing all invalid characters with '?' or raise an exception. """ return base_deunicodise(string, "UTF-8", errors, True) __all__.append("encode_to_s3") def s3_quote(param, quote_backslashes=True, unicode_output=False): """ URI encode every byte. UriEncode() must enforce the following rules: - URI encode every byte except the unreserved characters: 'A'-'Z', 'a'-'z', '0'-'9', '-', '.', '_', and '~'. - The space character is a reserved character and must be encoded as "%20" (and not as "+"). - Each URI encoded byte is formed by a '%' and the two-digit hexadecimal value of the byte. - Letters in the hexadecimal value must be uppercase, for example "%1A". - Encode the forward slash character, '/', everywhere except in the object key name. For example, if the object key name is photos/Jan/sample.jpg, the forward slash in the key name is not encoded. """ if quote_backslashes: safe_chars = "~" else: safe_chars = "~/" param = encode_to_s3(param) param = quote(param, safe=safe_chars) if unicode_output: param = decode_from_s3(param) else: param = encode_to_s3(param) return param __all__.append("s3_quote") def base_urlencode_string(string, urlencoding_mode = None, unicode_output=False): string = encode_to_s3(string) if urlencoding_mode == "verbatim": ## Don't do any pre-processing return string encoded = quote(string, safe="~/") debug("String '%s' encoded to '%s'" % (string, encoded)) if unicode_output: return decode_from_s3(encoded) else: return encode_to_s3(encoded) __all__.append("base_urlencode_string") def base_replace_nonprintables(string, with_message=False): """ replace_nonprintables(string) Replaces all non-printable characters 'ch' in 'string' where ord(ch) <= 26 with ^@, ^A, ... ^Z """ new_string = "" modified = 0 for c in string: o = ord(c) if (o <= 31): new_string += "^" + chr(ord('@') + o) modified += 1 elif (o == 127): new_string += "^?" modified += 1 else: new_string += c if modified and with_message: warning("%d non-printable characters replaced in: %s" % (modified, new_string)) return new_string __all__.append("base_replace_nonprintables") # XML helpers def parseNodes(nodes): ## WARNING: Ignores text nodes from mixed xml/text. ## For instance <tag1>some text<tag2>other text</tag2></tag1> ## will be ignore "some text" node ## WARNING 2: Any node at first level without children will also be ignored retval = [] for node in nodes: retval_item = {} for child in node: name = decode_from_s3(child.tag) if len(child): retval_item[name] = parseNodes([child]) else: found_text = node.findtext(".//%s" % child.tag) if found_text is not None: retval_item[name] = decode_from_s3(found_text) else: retval_item[name] = None if retval_item: retval.append(retval_item) return retval __all__.append("parseNodes") def getPrettyFromXml(xmlstr): xmlparser = xml.dom.minidom.parseString(xmlstr) return xmlparser.toprettyxml() __all__.append("getPrettyFromXml") def stripNameSpace(xml): """ removeNameSpace(xml) -- remove top-level AWS namespace Operate on raw byte(utf-8) xml string. (Not unicode) """ xmlns_match = RE_XML_NAMESPACE.match(xml) if xmlns_match: xmlns = xmlns_match.group(3) xml = RE_XML_NAMESPACE.sub(b"\\1\\2", xml, 1) else: xmlns = None return xml, xmlns __all__.append("stripNameSpace") def getTreeFromXml(xml): xml, xmlns = stripNameSpace(encode_to_s3(xml)) try: tree = ET.fromstring(xml) if xmlns: tree.attrib['xmlns'] = xmlns return tree except Exception as e: error("Error parsing xml: %s", e) error(xml) raise __all__.append("getTreeFromXml") def getListFromXml(xml, node): tree = getTreeFromXml(xml) nodes = tree.findall('.//%s' % (node)) return parseNodes(nodes) __all__.append("getListFromXml") def getDictFromTree(tree): ret_dict = {} for child in tree: if len(child): ## Complex-type child. Recurse content = getDictFromTree(child) else: content = decode_from_s3(child.text) if child.text is not None else None child_tag = decode_from_s3(child.tag) if child_tag in ret_dict: if not type(ret_dict[child_tag]) == list: ret_dict[child_tag] = [ret_dict[child_tag]] ret_dict[child_tag].append(content or "") else: ret_dict[child_tag] = content or "" return ret_dict __all__.append("getDictFromTree") def getTextFromXml(xml, xpath): tree = getTreeFromXml(xml) if tree.tag.endswith(xpath): return decode_from_s3(tree.text) if tree.text is not None else None else: result = tree.findtext(xpath) return decode_from_s3(result) if result is not None else None __all__.append("getTextFromXml") def getRootTagName(xml): tree = getTreeFromXml(xml) return decode_from_s3(tree.tag) if tree.tag is not None else None __all__.append("getRootTagName") def xmlTextNode(tag_name, text): el = ET.Element(tag_name) el.text = decode_from_s3(text) return el __all__.append("xmlTextNode") def appendXmlTextNode(tag_name, text, parent): """ Creates a new <tag_name> Node and sets its content to 'text'. Then appends the created Node to 'parent' element if given. Returns the newly created Node. """ el = xmlTextNode(tag_name, text) parent.append(el) return el __all__.append("appendXmlTextNode") # vim:et:ts=4:sts=4:ai	10.21 KB	2023-12-06 09:07
📄 BidirMap.py # -- coding: utf-8 -- ## -------------------------------------------------------------------- ## Amazon S3 manager ## ## Authors : Michal Ludvig <michal@logix.cz> (https://www.logix.cz/michal) ## Florent Viard <florent@sodria.com> (https://www.sodria.com) ## Copyright : TGRMN Software, Sodria SAS and contributors ## License : GPL Version 2 ## Website : https://s3tools.org ## -------------------------------------------------------------------- class BidirMap(object): def __init__(self, **map): self.k2v = {} self.v2k = {} for key in map: self.__setitem__(key, map[key]) def __setitem__(self, key, value): if value in self.v2k: if self.v2k[value] != key: raise KeyError("Value '"+str(value)+"' already in use with key '"+str(self.v2k[value])+"'") try: del(self.v2k[self.k2v[key]]) except KeyError: pass self.k2v[key] = value self.v2k[value] = key def __getitem__(self, key): return self.k2v[key] def __str__(self): return self.v2k.__str__() def getkey(self, value): return self.v2k[value] def getvalue(self, key): return self.k2v[key] def keys(self): return [key for key in self.k2v] def values(self): return [value for value in self.v2k] # vim:et:ts=4:sts=4:ai	1.37 KB	2023-12-06 09:07
📄 CloudFront.py # -- coding: utf-8 -- ## -------------------------------------------------------------------- ## Amazon CloudFront support ## ## Authors : Michal Ludvig <michal@logix.cz> (https://www.logix.cz/michal) ## Florent Viard <florent@sodria.com> (https://www.sodria.com) ## Copyright : TGRMN Software, Sodria SAS and contributors ## License : GPL Version 2 ## Website : https://s3tools.org ## -------------------------------------------------------------------- from __future__ import absolute_import import sys import time import random from collections import defaultdict from datetime import datetime from logging import debug, info, warning, error try: import xml.etree.ElementTree as ET except ImportError: import elementtree.ElementTree as ET from .S3 import S3 from .Config import Config from .Exceptions import CloudFrontError, ParameterError from .ExitCodes import EX_OK, EX_GENERAL, EX_PARTIAL from .BaseUtils import (getTreeFromXml, appendXmlTextNode, getDictFromTree, dateS3toPython, encode_to_s3, decode_from_s3) from .Utils import (getBucketFromHostname, getHostnameFromBucket, deunicodise, convertHeaderTupleListToDict) from .Crypto import sign_string_v2 from .S3Uri import S3Uri, S3UriS3 from .ConnMan import ConnMan from .SortedDict import SortedDict PY3 = (sys.version_info >= (3, 0)) cloudfront_api_version = "2010-11-01" cloudfront_resource = "/%(api_ver)s/distribution" % { 'api_ver' : cloudfront_api_version } def output(message): sys.stdout.write(message + "\n") def pretty_output(label, message): #label = ("%s " % label).ljust(20, ".") label = ("%s:" % label).ljust(15) output("%s %s" % (label, message)) class DistributionSummary(object): ## Example: ## ## <DistributionSummary> ## <Id>1234567890ABC</Id> ## <Status>Deployed</Status> ## <LastModifiedTime>2009-01-16T11:49:02.189Z</LastModifiedTime> ## <DomainName>blahblahblah.cloudfront.net</DomainName> ## <S3Origin> ## <DNSName>example.bucket.s3.amazonaws.com</DNSName> ## </S3Origin> ## <CNAME>cdn.example.com</CNAME> ## <CNAME>img.example.com</CNAME> ## <Comment>What Ever</Comment> ## <Enabled>true</Enabled> ## </DistributionSummary> def __init__(self, tree): if tree.tag != "DistributionSummary": raise ValueError("Expected <DistributionSummary /> xml, got: <%s />" % tree.tag) self.parse(tree) def parse(self, tree): self.info = getDictFromTree(tree) self.info['Enabled'] = (self.info['Enabled'].lower() == "true") if "CNAME" in self.info and type(self.info['CNAME']) != list: self.info['CNAME'] = [self.info['CNAME']] def uri(self): return S3Uri(u"cf://%s" % self.info['Id']) class DistributionList(object): ## Example: ## ## <DistributionList xmlns="http://cloudfront.amazonaws.com/doc/2010-07-15/"> ## <Marker /> ## <MaxItems>100</MaxItems> ## <IsTruncated>false</IsTruncated> ## <DistributionSummary> ## ... handled by DistributionSummary() class ... ## </DistributionSummary> ## </DistributionList> def __init__(self, xml): tree = getTreeFromXml(xml) if tree.tag != "DistributionList": raise ValueError("Expected <DistributionList /> xml, got: <%s />" % tree.tag) self.parse(tree) def parse(self, tree): self.info = getDictFromTree(tree) ## Normalise some items self.info['IsTruncated'] = (self.info['IsTruncated'].lower() == "true") self.dist_summs = [] for dist_summ in tree.findall(".//DistributionSummary"): self.dist_summs.append(DistributionSummary(dist_summ)) class Distribution(object): ## Example: ## ## <Distribution xmlns="http://cloudfront.amazonaws.com/doc/2010-07-15/"> ## <Id>1234567890ABC</Id> ## <Status>InProgress</Status> ## <LastModifiedTime>2009-01-16T13:07:11.319Z</LastModifiedTime> ## <DomainName>blahblahblah.cloudfront.net</DomainName> ## <DistributionConfig> ## ... handled by DistributionConfig() class ... ## </DistributionConfig> ## </Distribution> def __init__(self, xml): tree = getTreeFromXml(xml) if tree.tag != "Distribution": raise ValueError("Expected <Distribution /> xml, got: <%s />" % tree.tag) self.parse(tree) def parse(self, tree): self.info = getDictFromTree(tree) ## Normalise some items self.info['LastModifiedTime'] = dateS3toPython(self.info['LastModifiedTime']) self.info['DistributionConfig'] = DistributionConfig(tree = tree.find(".//DistributionConfig")) def uri(self): return S3Uri(u"cf://%s" % self.info['Id']) class DistributionConfig(object): ## Example: ## ## <DistributionConfig> ## <Origin>somebucket.s3.amazonaws.com</Origin> ## <CallerReference>s3://somebucket/</CallerReference> ## <Comment>http://somebucket.s3.amazonaws.com/</Comment> ## <Enabled>true</Enabled> ## <Logging> ## <Bucket>bu.ck.et</Bucket> ## <Prefix>/cf-somebucket/</Prefix> ## </Logging> ## </DistributionConfig> EMPTY_CONFIG = "<DistributionConfig><S3Origin><DNSName/></S3Origin><CallerReference/><Enabled>true</Enabled></DistributionConfig>" xmlns = "http://cloudfront.amazonaws.com/doc/%(api_ver)s/" % { 'api_ver' : cloudfront_api_version } def __init__(self, xml = None, tree = None): if xml is None: xml = DistributionConfig.EMPTY_CONFIG if tree is None: tree = getTreeFromXml(xml) if tree.tag != "DistributionConfig": raise ValueError("Expected <DistributionConfig /> xml, got: <%s />" % tree.tag) self.parse(tree) def parse(self, tree): self.info = getDictFromTree(tree) self.info['Enabled'] = (self.info['Enabled'].lower() == "true") if "CNAME" not in self.info: self.info['CNAME'] = [] if type(self.info['CNAME']) != list: self.info['CNAME'] = [self.info['CNAME']] self.info['CNAME'] = [cname.lower() for cname in self.info['CNAME']] if "Comment" not in self.info: self.info['Comment'] = "" if "DefaultRootObject" not in self.info: self.info['DefaultRootObject'] = "" ## Figure out logging - complex node not parsed by getDictFromTree() logging_nodes = tree.findall(".//Logging") if logging_nodes: logging_dict = getDictFromTree(logging_nodes[0]) logging_dict['Bucket'], success = getBucketFromHostname(logging_dict['Bucket']) if not success: warning("Logging to unparsable bucket name: %s" % logging_dict['Bucket']) self.info['Logging'] = S3UriS3(u"s3://%(Bucket)s/%(Prefix)s" % logging_dict) else: self.info['Logging'] = None def get_printable_tree(self): tree = ET.Element("DistributionConfig") tree.attrib['xmlns'] = DistributionConfig.xmlns ## Retain the order of the following calls! s3org = appendXmlTextNode("S3Origin", '', tree) appendXmlTextNode("DNSName", self.info['S3Origin']['DNSName'], s3org) appendXmlTextNode("CallerReference", self.info['CallerReference'], tree) for cname in self.info['CNAME']: appendXmlTextNode("CNAME", cname.lower(), tree) if self.info['Comment']: appendXmlTextNode("Comment", self.info['Comment'], tree) appendXmlTextNode("Enabled", str(self.info['Enabled']).lower(), tree) # don't create a empty DefaultRootObject element as it would result in a MalformedXML error if str(self.info['DefaultRootObject']): appendXmlTextNode("DefaultRootObject", str(self.info['DefaultRootObject']), tree) if self.info['Logging']: logging_el = ET.Element("Logging") appendXmlTextNode("Bucket", getHostnameFromBucket(self.info['Logging'].bucket()), logging_el) appendXmlTextNode("Prefix", self.info['Logging'].object(), logging_el) tree.append(logging_el) return tree def __unicode__(self): return decode_from_s3(ET.tostring(self.get_printable_tree())) def __str__(self): if PY3: # Return unicode return ET.tostring(self.get_printable_tree(), encoding="unicode") else: # Return bytes return ET.tostring(self.get_printable_tree()) class Invalidation(object): ## Example: ## ## <Invalidation xmlns="http://cloudfront.amazonaws.com/doc/2010-11-01/"> ## <Id>id</Id> ## <Status>status</Status> ## <CreateTime>date</CreateTime> ## <InvalidationBatch> ## <Path>/image1.jpg</Path> ## <Path>/image2.jpg</Path> ## <Path>/videos/movie.flv</Path> ## <CallerReference>my-batch</CallerReference> ## </InvalidationBatch> ## </Invalidation> def __init__(self, xml): tree = getTreeFromXml(xml) if tree.tag != "Invalidation": raise ValueError("Expected <Invalidation /> xml, got: <%s />" % tree.tag) self.parse(tree) def parse(self, tree): self.info = getDictFromTree(tree) def __str__(self): return str(self.info) class InvalidationList(object): ## Example: ## ## <InvalidationList> ## <Marker/> ## <NextMarker>Invalidation ID</NextMarker> ## <MaxItems>2</MaxItems> ## <IsTruncated>true</IsTruncated> ## <InvalidationSummary> ## <Id>[Second Invalidation ID]</Id> ## <Status>Completed</Status> ## </InvalidationSummary> ## <InvalidationSummary> ## <Id>[First Invalidation ID]</Id> ## <Status>Completed</Status> ## </InvalidationSummary> ## </InvalidationList> def __init__(self, xml): tree = getTreeFromXml(xml) if tree.tag != "InvalidationList": raise ValueError("Expected <InvalidationList /> xml, got: <%s />" % tree.tag) self.parse(tree) def parse(self, tree): self.info = getDictFromTree(tree) def __str__(self): return str(self.info) class InvalidationBatch(object): ## Example: ## ## <InvalidationBatch> ## <Path>/image1.jpg</Path> ## <Path>/image2.jpg</Path> ## <Path>/videos/movie.flv</Path> ## <Path>/sound%20track.mp3</Path> ## <CallerReference>my-batch</CallerReference> ## </InvalidationBatch> def __init__(self, reference = None, distribution = None, paths = []): if reference: self.reference = reference else: if not distribution: distribution="0" self.reference = "%s.%s.%s" % (distribution, datetime.strftime(datetime.now(),"%Y%m%d%H%M%S"), random.randint(1000,9999)) self.paths = [] self.add_objects(paths) def add_objects(self, paths): self.paths.extend(paths) def get_reference(self): return self.reference def get_printable_tree(self): tree = ET.Element("InvalidationBatch") for path in self.paths: if len(path) < 1 or path[0] != "/": path = "/" + path appendXmlTextNode("Path", path, tree) appendXmlTextNode("CallerReference", self.reference, tree) return tree def __unicode__(self): return decode_from_s3(ET.tostring(self.get_printable_tree())) def __str__(self): if PY3: # Return unicode return ET.tostring(self.get_printable_tree(), encoding="unicode") else: # Return bytes return ET.tostring(self.get_printable_tree()) class CloudFront(object): operations = { "CreateDist" : { 'method' : "POST", 'resource' : "" }, "DeleteDist" : { 'method' : "DELETE", 'resource' : "/%(dist_id)s" }, "GetList" : { 'method' : "GET", 'resource' : "" }, "GetDistInfo" : { 'method' : "GET", 'resource' : "/%(dist_id)s" }, "GetDistConfig" : { 'method' : "GET", 'resource' : "/%(dist_id)s/config" }, "SetDistConfig" : { 'method' : "PUT", 'resource' : "/%(dist_id)s/config" }, "Invalidate" : { 'method' : "POST", 'resource' : "/%(dist_id)s/invalidation" }, "GetInvalList" : { 'method' : "GET", 'resource' : "/%(dist_id)s/invalidation" }, "GetInvalInfo" : { 'method' : "GET", 'resource' : "/%(dist_id)s/invalidation/%(request_id)s" }, } dist_list = None def __init__(self, config): self.config = config ## -------------------------------------------------- ## Methods implementing CloudFront API ## -------------------------------------------------- def GetList(self): response = self.send_request("GetList") response['dist_list'] = DistributionList(response['data']) if response['dist_list'].info['IsTruncated']: raise NotImplementedError("List is truncated. Ask s3cmd author to add support.") ## TODO: handle Truncated return response def CreateDistribution(self, uri, cnames_add = [], comment = None, logging = None, default_root_object = None): dist_config = DistributionConfig() dist_config.info['Enabled'] = True dist_config.info['S3Origin']['DNSName'] = uri.host_name() dist_config.info['CallerReference'] = str(uri) dist_config.info['DefaultRootObject'] = default_root_object if comment == None: dist_config.info['Comment'] = uri.public_url() else: dist_config.info['Comment'] = comment for cname in cnames_add: if dist_config.info['CNAME'].count(cname) == 0: dist_config.info['CNAME'].append(cname) if logging: dist_config.info['Logging'] = S3UriS3(logging) request_body = str(dist_config) debug("CreateDistribution(): request_body: %s" % request_body) response = self.send_request("CreateDist", body = request_body) response['distribution'] = Distribution(response['data']) return response def ModifyDistribution(self, cfuri, cnames_add = [], cnames_remove = [], comment = None, enabled = None, logging = None, default_root_object = None): if cfuri.type != "cf": raise ValueError("Expected CFUri instead of: %s" % cfuri) # Get current dist status (enabled/disabled) and Etag info("Checking current status of %s" % cfuri) response = self.GetDistConfig(cfuri) dc = response['dist_config'] if enabled != None: dc.info['Enabled'] = enabled if comment != None: dc.info['Comment'] = comment if default_root_object != None: dc.info['DefaultRootObject'] = default_root_object for cname in cnames_add: if dc.info['CNAME'].count(cname) == 0: dc.info['CNAME'].append(cname) for cname in cnames_remove: while dc.info['CNAME'].count(cname) > 0: dc.info['CNAME'].remove(cname) if logging != None: if logging == False: dc.info['Logging'] = False else: dc.info['Logging'] = S3UriS3(logging) response = self.SetDistConfig(cfuri, dc, response['headers']['etag']) return response def DeleteDistribution(self, cfuri): if cfuri.type != "cf": raise ValueError("Expected CFUri instead of: %s" % cfuri) # Get current dist status (enabled/disabled) and Etag info("Checking current status of %s" % cfuri) response = self.GetDistConfig(cfuri) if response['dist_config'].info['Enabled']: info("Distribution is ENABLED. Disabling first.") response['dist_config'].info['Enabled'] = False response = self.SetDistConfig(cfuri, response['dist_config'], response['headers']['etag']) warning("Waiting for Distribution to become disabled.") warning("This may take several minutes, please wait.") while True: response = self.GetDistInfo(cfuri) d = response['distribution'] if d.info['Status'] == "Deployed" and d.info['Enabled'] == False: info("Distribution is now disabled") break warning("Still waiting...") time.sleep(10) headers = SortedDict(ignore_case = True) headers['if-match'] = response['headers']['etag'] response = self.send_request("DeleteDist", dist_id = cfuri.dist_id(), headers = headers) return response def GetDistInfo(self, cfuri): if cfuri.type != "cf": raise ValueError("Expected CFUri instead of: %s" % cfuri) response = self.send_request("GetDistInfo", dist_id = cfuri.dist_id()) response['distribution'] = Distribution(response['data']) return response def GetDistConfig(self, cfuri): if cfuri.type != "cf": raise ValueError("Expected CFUri instead of: %s" % cfuri) response = self.send_request("GetDistConfig", dist_id = cfuri.dist_id()) response['dist_config'] = DistributionConfig(response['data']) return response def SetDistConfig(self, cfuri, dist_config, etag = None): if etag == None: debug("SetDistConfig(): Etag not set. Fetching it first.") etag = self.GetDistConfig(cfuri)['headers']['etag'] debug("SetDistConfig(): Etag = %s" % etag) request_body = str(dist_config) debug("SetDistConfig(): request_body: %s" % request_body) headers = SortedDict(ignore_case = True) headers['if-match'] = etag response = self.send_request("SetDistConfig", dist_id = cfuri.dist_id(), body = request_body, headers = headers) return response def InvalidateObjects(self, uri, paths, default_index_file, invalidate_default_index_on_cf, invalidate_default_index_root_on_cf): # joseprio: if the user doesn't want to invalidate the default index # path, or if the user wants to invalidate the root of the default # index, we need to process those paths if default_index_file is not None and (not invalidate_default_index_on_cf or invalidate_default_index_root_on_cf): new_paths = [] default_index_suffix = '/' + default_index_file for path in paths: if path.endswith(default_index_suffix) or path == default_index_file: if invalidate_default_index_on_cf: new_paths.append(path) if invalidate_default_index_root_on_cf: new_paths.append(path[:-len(default_index_file)]) else: new_paths.append(path) paths = new_paths # uri could be either cf:// or s3:// uri cfuris = self.get_dist_name_for_bucket(uri) if len(paths) > 999: try: tmp_filename = Utils.mktmpfile() with open(deunicodise(tmp_filename), "w") as fp: fp.write(deunicodise("\n".join(paths)+"\n")) warning("Request to invalidate %d paths (max 999 supported)" % len(paths)) warning("All the paths are now saved in: %s" % tmp_filename) except Exception: pass raise ParameterError("Too many paths to invalidate") responses = [] for cfuri in cfuris: invalbatch = InvalidationBatch(distribution = cfuri.dist_id(), paths = paths) debug("InvalidateObjects(): request_body: %s" % invalbatch) response = self.send_request("Invalidate", dist_id = cfuri.dist_id(), body = str(invalbatch)) response['dist_id'] = cfuri.dist_id() if response['status'] == 201: inval_info = Invalidation(response['data']).info response['request_id'] = inval_info['Id'] debug("InvalidateObjects(): response: %s" % response) responses.append(response) return responses def GetInvalList(self, cfuri): if cfuri.type != "cf": raise ValueError("Expected CFUri instead of: %s" % cfuri) response = self.send_request("GetInvalList", dist_id = cfuri.dist_id()) response['inval_list'] = InvalidationList(response['data']) return response def GetInvalInfo(self, cfuri): if cfuri.type != "cf": raise ValueError("Expected CFUri instead of: %s" % cfuri) if cfuri.request_id() is None: raise ValueError("Expected CFUri with Request ID") response = self.send_request("GetInvalInfo", dist_id = cfuri.dist_id(), request_id = cfuri.request_id()) response['inval_status'] = Invalidation(response['data']) return response ## -------------------------------------------------- ## Low-level methods for handling CloudFront requests ## -------------------------------------------------- def send_request(self, op_name, dist_id = None, request_id = None, body = None, headers = None, retries = None): if retries is None: retries = self.config.max_retries if headers is None: headers = SortedDict(ignore_case = True) operation = self.operations[op_name] if body: headers['content-type'] = 'text/plain' request = self.create_request(operation, dist_id, request_id, headers) conn = self.get_connection() debug("send_request(): %s %s" % (request['method'], request['resource'])) conn.c.request(request['method'], request['resource'], body, request['headers']) 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() ConnMan.put(conn) debug("CloudFront: response: %r" % response) if response["status"] >= 500: e = CloudFrontError(response) if retries: warning(u"Retrying failed request: %s (%s)" % (op_name, e)) warning("Waiting %d sec..." % self._fail_wait(retries)) time.sleep(self._fail_wait(retries)) return self.send_request(op_name, dist_id, body = body, retries = retries - 1) else: raise e if response["status"] < 200 or response["status"] > 299: raise CloudFrontError(response) return response def create_request(self, operation, dist_id = None, request_id = None, headers = None): resource = cloudfront_resource + ( operation['resource'] % { 'dist_id' : dist_id, 'request_id' : request_id }) if not headers: headers = SortedDict(ignore_case = True) if "date" in headers: if "x-amz-date" not in headers: headers["x-amz-date"] = headers["date"] del(headers["date"]) if "x-amz-date" not in headers: headers["x-amz-date"] = time.strftime("%a, %d %b %Y %H:%M:%S +0000", time.gmtime()) if len(self.config.access_token)>0: self.config.role_refresh() headers['x-amz-security-token']=self.config.access_token signature = self.sign_request(headers) headers["Authorization"] = "AWS "+self.config.access_key+":"+signature request = {} request['resource'] = resource request['headers'] = headers request['method'] = operation['method'] return request def sign_request(self, headers): string_to_sign = headers['x-amz-date'] signature = decode_from_s3(sign_string_v2(encode_to_s3(string_to_sign))) debug(u"CloudFront.sign_request('%s') = %s" % (string_to_sign, signature)) return signature def get_connection(self): conn = ConnMan.get(self.config.cloudfront_host, ssl = True) return conn def _fail_wait(self, retries): # Wait a few seconds. The more it fails the more we wait. return (self.config.max_retries - retries + 1) * 3 def get_dist_name_for_bucket(self, uri): if uri.type == "cf": return [uri] if uri.type != "s3": raise ParameterError("CloudFront or S3 URI required instead of: %s" % uri) debug("_get_dist_name_for_bucket(%r)" % uri) if CloudFront.dist_list is None: response = self.GetList() CloudFront.dist_list = {} for d in response['dist_list'].dist_summs: distListIndex = "" if "S3Origin" in d.info: distListIndex = getBucketFromHostname(d.info['S3Origin']['DNSName'])[0] elif "CustomOrigin" in d.info: # Aral: This used to skip over distributions with CustomOrigin, however, we mustn't # do this since S3 buckets that are set up as websites use custom origins. # Thankfully, the custom origin URLs they use start with the URL of the # S3 bucket. Here, we make use this naming convention to support this use case. distListIndex = getBucketFromHostname(d.info['CustomOrigin']['DNSName'])[0] distListIndex = distListIndex[:len(uri.bucket())] else: # Aral: I'm not sure when this condition will be reached, but keeping it in there. continue if CloudFront.dist_list.get(distListIndex, None) is None: CloudFront.dist_list[distListIndex] = set() CloudFront.dist_list[distListIndex].add(d.uri()) debug("dist_list: %s" % CloudFront.dist_list) try: return CloudFront.dist_list[uri.bucket()] except Exception as e: debug(e) raise ParameterError("Unable to translate S3 URI to CloudFront distribution name: %s" % uri) class Cmd(object): """ Class that implements CloudFront commands """ class Options(object): cf_cnames_add = [] cf_cnames_remove = [] cf_comment = None cf_enable = None cf_logging = None cf_default_root_object = None def option_list(self): return [opt for opt in dir(self) if opt.startswith("cf_")] def update_option(self, option, value): setattr(Cmd.options, option, value) options = Options() @staticmethod def _parse_args(args): cf = CloudFront(Config()) cfuris = [] for arg in args: uris = cf.get_dist_name_for_bucket(S3Uri(arg)) cfuris.extend(uris) return cfuris @staticmethod def info(args): cf = CloudFront(Config()) if not args: response = cf.GetList() for d in response['dist_list'].dist_summs: if "S3Origin" in d.info: origin = S3UriS3.httpurl_to_s3uri(d.info['S3Origin']['DNSName']) elif "CustomOrigin" in d.info: origin = "http://%s/" % d.info['CustomOrigin']['DNSName'] else: origin = "<unknown>" pretty_output("Origin", origin) pretty_output("DistId", d.uri()) pretty_output("DomainName", d.info['DomainName']) if "CNAME" in d.info: pretty_output("CNAMEs", ", ".join(d.info['CNAME'])) pretty_output("Status", d.info['Status']) pretty_output("Enabled", d.info['Enabled']) output("") else: cfuris = Cmd._parse_args(args) for cfuri in cfuris: response = cf.GetDistInfo(cfuri) d = response['distribution'] dc = d.info['DistributionConfig'] if "S3Origin" in dc.info: origin = S3UriS3.httpurl_to_s3uri(dc.info['S3Origin']['DNSName']) elif "CustomOrigin" in dc.info: origin = "http://%s/" % dc.info['CustomOrigin']['DNSName'] else: origin = "<unknown>" pretty_output("Origin", origin) pretty_output("DistId", d.uri()) pretty_output("DomainName", d.info['DomainName']) if "CNAME" in dc.info: pretty_output("CNAMEs", ", ".join(dc.info['CNAME'])) pretty_output("Status", d.info['Status']) pretty_output("Comment", dc.info['Comment']) pretty_output("Enabled", dc.info['Enabled']) pretty_output("DfltRootObject", dc.info['DefaultRootObject']) pretty_output("Logging", dc.info['Logging'] or "Disabled") pretty_output("Etag", response['headers']['etag']) @staticmethod def create(args): cf = CloudFront(Config()) buckets = [] for arg in args: uri = S3Uri(arg) if uri.type != "s3": raise ParameterError("Distribution can only be created from a s3:// URI instead of: %s" % arg) if uri.object(): raise ParameterError("Use s3:// URI with a bucket name only instead of: %s" % arg) if not uri.is_dns_compatible(): raise ParameterError("CloudFront can only handle lowercase-named buckets.") buckets.append(uri) if not buckets: raise ParameterError("No valid bucket names found") for uri in buckets: info("Creating distribution from: %s" % uri) response = cf.CreateDistribution(uri, cnames_add = Cmd.options.cf_cnames_add, comment = Cmd.options.cf_comment, logging = Cmd.options.cf_logging, default_root_object = Cmd.options.cf_default_root_object) d = response['distribution'] dc = d.info['DistributionConfig'] output("Distribution created:") pretty_output("Origin", S3UriS3.httpurl_to_s3uri(dc.info['S3Origin']['DNSName'])) pretty_output("DistId", d.uri()) pretty_output("DomainName", d.info['DomainName']) pretty_output("CNAMEs", ", ".join(dc.info['CNAME'])) pretty_output("Comment", dc.info['Comment']) pretty_output("Status", d.info['Status']) pretty_output("Enabled", dc.info['Enabled']) pretty_output("DefaultRootObject", dc.info['DefaultRootObject']) pretty_output("Etag", response['headers']['etag']) @staticmethod def delete(args): cf = CloudFront(Config()) cfuris = Cmd._parse_args(args) for cfuri in cfuris: response = cf.DeleteDistribution(cfuri) if response['status'] >= 400: error("Distribution %s could not be deleted: %s" % (cfuri, response['reason'])) output("Distribution %s deleted" % cfuri) @staticmethod def modify(args): cf = CloudFront(Config()) if len(args) > 1: raise ParameterError("Too many parameters. Modify one Distribution at a time.") try: cfuri = Cmd._parse_args(args)[0] except IndexError: raise ParameterError("No valid Distribution URI found.") response = cf.ModifyDistribution(cfuri, cnames_add = Cmd.options.cf_cnames_add, cnames_remove = Cmd.options.cf_cnames_remove, comment = Cmd.options.cf_comment, enabled = Cmd.options.cf_enable, logging = Cmd.options.cf_logging, default_root_object = Cmd.options.cf_default_root_object) if response['status'] >= 400: error("Distribution %s could not be modified: %s" % (cfuri, response['reason'])) output("Distribution modified: %s" % cfuri) response = cf.GetDistInfo(cfuri) d = response['distribution'] dc = d.info['DistributionConfig'] pretty_output("Origin", S3UriS3.httpurl_to_s3uri(dc.info['S3Origin']['DNSName'])) pretty_output("DistId", d.uri()) pretty_output("DomainName", d.info['DomainName']) pretty_output("Status", d.info['Status']) pretty_output("CNAMEs", ", ".join(dc.info['CNAME'])) pretty_output("Comment", dc.info['Comment']) pretty_output("Enabled", dc.info['Enabled']) pretty_output("DefaultRootObject", dc.info['DefaultRootObject']) pretty_output("Etag", response['headers']['etag']) @staticmethod def invalinfo(args): cf = CloudFront(Config()) cfuris = Cmd._parse_args(args) requests = [] for cfuri in cfuris: if cfuri.request_id(): requests.append(str(cfuri)) else: inval_list = cf.GetInvalList(cfuri) try: for i in inval_list['inval_list'].info['InvalidationSummary']: requests.append("/".join(["cf:/", cfuri.dist_id(), i["Id"]])) except Exception: continue for req in requests: cfuri = S3Uri(req) inval_info = cf.GetInvalInfo(cfuri) st = inval_info['inval_status'].info paths = st['InvalidationBatch']['Path'] nr_of_paths = len(paths) if isinstance(paths, list) else 1 pretty_output("URI", str(cfuri)) pretty_output("Status", st['Status']) pretty_output("Created", st['CreateTime']) pretty_output("Nr of paths", nr_of_paths) pretty_output("Reference", st['InvalidationBatch']['CallerReference']) output("") @staticmethod def invalidate(args): cfg = Config() cf = CloudFront(cfg) s3 = S3(cfg) bucket_paths = defaultdict(list) for arg in args: uri = S3Uri(arg) uobject = uri.object() if not uobject: # If object is not defined, we want to invalidate the whole bucket uobject = '' elif uobject[-1] == '/': # If object is folder (ie prefix), we want to invalidate the whole content uobject += '' bucket_paths[uri.bucket()].append(uobject) ret = EX_OK params = [] for bucket, paths in bucket_paths.items(): base_uri = S3Uri(u's3://%s' % bucket) cfuri = next(iter(cf.get_dist_name_for_bucket(base_uri))) default_index_file = None if cfg.invalidate_default_index_on_cf or cfg.invalidate_default_index_root_on_cf: info_response = s3.website_info(base_uri, cfg.bucket_location) if info_response: default_index_file = info_response['index_document'] if not default_index_file: default_index_file = None if cfg.dry_run: fulluri_paths = [S3UriS3.compose_uri(bucket, path) for path in paths] output(u"[--dry-run] Would invalidate %r" % fulluri_paths) continue params.append((bucket, paths, base_uri, cfuri, default_index_file)) if cfg.dry_run: warning(u"Exiting now because of --dry-run") return EX_OK nb_success = 0 first = True for bucket, paths, base_uri, cfuri, default_index_file in params: if not first: output("") else: first = False results = cf.InvalidateObjects( cfuri, paths, default_index_file, cfg.invalidate_default_index_on_cf, cfg.invalidate_default_index_root_on_cf ) dist_id = cfuri.dist_id() pretty_output("URI", str(base_uri)) pretty_output("DistId", dist_id) pretty_output("Nr of paths", len(paths)) for result in results: result_code = result['status'] if result_code != 201: pretty_output("Status", "Failed: %d" % result_code) ret = EX_GENERAL continue request_id = result['request_id'] nb_success += 1 pretty_output("Status", "Created") pretty_output("RequestId", request_id) pretty_output("Info", u"Check progress with: s3cmd cfinvalinfo %s/%s" % (dist_id, request_id)) if ret != EX_OK and cfg.stop_on_error: error(u"Exiting now because of --stop-on-error") break if ret != EX_OK and nb_success: ret = EX_PARTIAL return ret # vim:et:ts=4:sts=4:ai	36.71 KB	2023-12-06 09:07
📄 Config.py # -- coding: utf-8 -- ## -------------------------------------------------------------------- ## Amazon S3 manager ## ## Authors : Michal Ludvig <michal@logix.cz> (https://www.logix.cz/michal) ## Florent Viard <florent@sodria.com> (https://www.sodria.com) ## Copyright : TGRMN Software, Sodria SAS and contributors ## License : GPL Version 2 ## Website : https://s3tools.org ## -------------------------------------------------------------------- from __future__ import absolute_import import logging import datetime import locale import re import os import io import sys import json import time from logging import debug, warning from .ExitCodes import EX_OSFILE try: import dateutil.parser import dateutil.tz except ImportError: sys.stderr.write(u""" !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! ImportError trying to import dateutil.parser and dateutil.tz. Please install the python dateutil module: $ sudo apt-get install python-dateutil or $ sudo yum install python-dateutil or $ pip install python-dateutil !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! """) sys.stderr.flush() sys.exit(EX_OSFILE) try: # python 3 support import httplib except ImportError: import http.client as httplib try: from configparser import (NoOptionError, NoSectionError, MissingSectionHeaderError, ParsingError, ConfigParser as PyConfigParser) except ImportError: # Python2 fallback code from ConfigParser import (NoOptionError, NoSectionError, MissingSectionHeaderError, ParsingError, ConfigParser as PyConfigParser) from . import Progress from .SortedDict import SortedDict from .BaseUtils import (s3_quote, getTreeFromXml, getDictFromTree, base_unicodise, dateRFC822toPython) try: unicode except NameError: # python 3 support # In python 3, unicode -> str, and str -> bytes unicode = str def is_bool_true(value): """Check to see if a string is true, yes, on, or 1 value may be a str, or unicode. Return True if it is """ if type(value) == unicode: return value.lower() in ["true", "yes", "on", "1"] elif type(value) == bool and value == True: return True else: return False def is_bool_false(value): """Check to see if a string is false, no, off, or 0 value may be a str, or unicode. Return True if it is """ if type(value) == unicode: return value.lower() in ["false", "no", "off", "0"] elif type(value) == bool and value == False: return True else: return False def is_bool(value): """Check a string value to see if it is bool""" return is_bool_true(value) or is_bool_false(value) class Config(object): _instance = None _parsed_files = [] _doc = {} access_key = u"" secret_key = u"" access_token = u"" _access_token_refresh = True _access_token_expiration = None _access_token_last_update = None host_base = u"s3.amazonaws.com" host_bucket = u"%(bucket)s.s3.amazonaws.com" kms_key = u"" #can't set this and Server Side Encryption at the same time # simpledb_host looks useless, legacy? to remove? simpledb_host = u"sdb.amazonaws.com" cloudfront_host = u"cloudfront.amazonaws.com" verbosity = logging.WARNING progress_meter = sys.stdout.isatty() progress_class = Progress.ProgressCR send_chunk = 64 * 1024 recv_chunk = 64 * 1024 list_md5 = False long_listing = False human_readable_sizes = False extra_headers = SortedDict(ignore_case = True) force = False server_side_encryption = False enable = None get_continue = False put_continue = False upload_id = u"" skip_existing = False recursive = False restore_days = 1 restore_priority = u"Standard" acl_public = None acl_grants = [] acl_revokes = [] proxy_host = u"" proxy_port = 3128 encrypt = False dry_run = False add_encoding_exts = u"" preserve_attrs = True preserve_attrs_list = [ u'uname', # Verbose owner Name (e.g. 'root') u'uid', # Numeric user ID (e.g. 0) u'gname', # Group name (e.g. 'users') u'gid', # Numeric group ID (e.g. 100) u'atime', # Last access timestamp u'mtime', # Modification timestamp u'ctime', # Creation timestamp u'mode', # File mode (e.g. rwxr-xr-x = 755) u'md5', # File MD5 (if known) #u'acl', # Full ACL (not yet supported) ] keep_dirs = False delete_removed = False delete_after = False delete_after_fetch = False max_delete = -1 limit = -1 _doc['delete_removed'] = u"[sync] Remove remote S3 objects when local file has been deleted" delay_updates = False # OBSOLETE gpg_passphrase = u"" gpg_command = u"" gpg_encrypt = u"%(gpg_command)s -c --verbose --no-use-agent --batch --yes --passphrase-fd %(passphrase_fd)s -o %(output_file)s %(input_file)s" gpg_decrypt = u"%(gpg_command)s -d --verbose --no-use-agent --batch --yes --passphrase-fd %(passphrase_fd)s -o %(output_file)s %(input_file)s" use_https = True ca_certs_file = u"" ssl_client_key_file = u"" ssl_client_cert_file = u"" check_ssl_certificate = True check_ssl_hostname = True bucket_location = u"US" default_mime_type = u"binary/octet-stream" guess_mime_type = True use_mime_magic = True mime_type = u"" enable_multipart = True # Chunk size is at the same time the chunk size and the threshold multipart_chunk_size_mb = 15 # MiB # Maximum chunk size for s3-to-s3 copy is 5 GiB. # But, use a lot lower value by default (1GiB) multipart_copy_chunk_size_mb = 1 * 1024 # Maximum chunks on AWS S3, could be different on other S3-compatible APIs multipart_max_chunks = 10000 # List of checks to be performed for 'sync' sync_checks = ['size', 'md5'] # 'weak-timestamp' # List of compiled REGEXPs exclude = [] include = [] # Dict mapping compiled REGEXPs back to their textual form debug_exclude = {} debug_include = {} encoding = locale.getpreferredencoding() or "UTF-8" urlencoding_mode = u"normal" log_target_prefix = u"" reduced_redundancy = False storage_class = u"" follow_symlinks = False # If too big, this value can be overridden by the OS socket timeouts max values. # For example, on Linux, a connection attempt will automatically timeout after 120s. socket_timeout = 300 invalidate_on_cf = False # joseprio: new flags for default index invalidation invalidate_default_index_on_cf = False invalidate_default_index_root_on_cf = True website_index = u"index.html" website_error = u"" website_endpoint = u"http://%(bucket)s.s3-website-%(location)s.amazonaws.com/" additional_destinations = [] files_from = [] cache_file = u"" add_headers = u"" remove_headers = [] expiry_days = u"" expiry_date = u"" expiry_prefix = u"" skip_destination_validation = False signature_v2 = False limitrate = 0 requester_pays = False stop_on_error = False content_disposition = u"" content_type = u"" stats = False # Disabled by default because can create a latency with a CONTINUE status reply # expected for every send file requests. use_http_expect = False signurl_use_https = False # Maximum sleep duration for throttle / limitrate. # s3 will timeout if a request/transfer is stuck for more than a short time throttle_max = 100 public_url_use_https = False connection_pooling = True # How long in seconds a connection can be kept idle in the pool and still # be alive. AWS s3 is supposed to close connections that are idle for 20 # seconds or more, but in real life, undocumented, it closes https conns # after around 6s of inactivity. connection_max_age = 5 # Not an AWS standard # allow the listing results to be returned in unsorted order. # This may be faster when listing very large buckets. list_allow_unordered = False # Maximum attempts of re-issuing failed requests max_retries = 5 ## Creating a singleton def __new__(self, configfile = None, access_key=None, secret_key=None, access_token=None): if self._instance is None: self._instance = object.__new__(self) return self._instance def __init__(self, configfile = None, access_key=None, secret_key=None, access_token=None): if configfile: try: self.read_config_file(configfile) except IOError: if 'AWS_CREDENTIAL_FILE' in os.environ or 'AWS_PROFILE' in os.environ: self.aws_credential_file() # override these if passed on the command-line if access_key and secret_key: self.access_key = access_key self.secret_key = secret_key if access_token: self.access_token = access_token # Do not refresh the IAM role when an access token is provided. self._access_token_refresh = False if len(self.access_key) == 0: env_access_key = os.getenv('AWS_ACCESS_KEY') or os.getenv('AWS_ACCESS_KEY_ID') env_secret_key = os.getenv('AWS_SECRET_KEY') or os.getenv('AWS_SECRET_ACCESS_KEY') env_access_token = os.getenv('AWS_SESSION_TOKEN') or os.getenv('AWS_SECURITY_TOKEN') if env_access_key: if not env_secret_key: raise ValueError( "AWS_ACCESS_KEY environment variable is used but" " AWS_SECRET_KEY variable is missing" ) # py3 getenv returns unicode and py2 returns bytes. self.access_key = base_unicodise(env_access_key) self.secret_key = base_unicodise(env_secret_key) if env_access_token: # Do not refresh the IAM role when an access token is provided. self._access_token_refresh = False self.access_token = base_unicodise(env_access_token) else: self.role_config() #TODO check KMS key is valid if self.kms_key and self.server_side_encryption == True: warning('Cannot have server_side_encryption (S3 SSE) and KMS_key set (S3 KMS). KMS encryption will be used. Please set server_side_encryption to False') if self.kms_key and self.signature_v2 == True: raise Exception('KMS encryption requires signature v4. Please set signature_v2 to False') def role_config(self): """ Get credentials from IAM authentication and STS AssumeRole """ try: role_arn = os.environ.get('AWS_ROLE_ARN') if role_arn: role_session_name = 'role-session-%s' % (int(time.time())) params = { 'Action': 'AssumeRole', 'Version': '2011-06-15', 'RoleArn': role_arn, 'RoleSessionName': role_session_name, } web_identity_token_file = os.environ.get('AWS_WEB_IDENTITY_TOKEN_FILE') if web_identity_token_file: with open(web_identity_token_file) as f: web_identity_token = f.read() params['Action'] = 'AssumeRoleWithWebIdentity' params['WebIdentityToken'] = web_identity_token encoded_params = '&'.join([ '%s=%s' % (k, s3_quote(v, unicode_output=True)) for k, v in params.items() ]) sts_endpoint = "sts.amazonaws.com" if os.environ.get("AWS_STS_REGIONAL_ENDPOINTS") == "regional": # Check if the AWS_REGION variable is available to use as a region. region = os.environ.get("AWS_REGION") if not region: # Otherwise use the bucket location region = self.bucket_location sts_endpoint = "sts.%s.amazonaws.com" % region conn = httplib.HTTPSConnection(host=sts_endpoint, timeout=2) conn.request('POST', '/?' + encoded_params) resp = conn.getresponse() resp_content = resp.read() if resp.status == 200 and len(resp_content) > 1: tree = getTreeFromXml(resp_content) result_dict = getDictFromTree(tree) if tree.tag == "AssumeRoleResponse": creds = result_dict['AssumeRoleResult']['Credentials'] elif tree.tag == "AssumeRoleWithWebIdentityResponse": creds = result_dict['AssumeRoleWithWebIdentityResult']['Credentials'] else: raise IOError("Unexpected XML message from STS server: <%s />" % tree.tag) Config().update_option('access_key', creds['AccessKeyId']) Config().update_option('secret_key', creds['SecretAccessKey']) Config().update_option('access_token', creds['SessionToken']) expiration = dateRFC822toPython(base_unicodise(creds['Expiration'])) # Add a timedelta to prevent any expiration if the EC2 machine is not at the right date self._access_token_expiration = expiration - datetime.timedelta(minutes=15) # last update date is not provided in STS responses self._access_token_last_update = datetime.datetime.now(dateutil.tz.tzutc()) # Others variables : Code / Type else: raise IOError else: conn = httplib.HTTPConnection(host='169.254.169.254', timeout=2) # To use Instance Metadata Service (IMDSv2), we first need to obtain a token, then # supply it with every IMDS HTTP call. More info: # # https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/configuring-instance-metadata-service.html # # 60 seconds is arbitrary, but since we're just pulling small bits of data from the # local instance, it should be plenty of time. # # There's a chance that there are "mostly AWS compatible" systems that might offer # only IMDSv1 emulation, so we make this optional -- if we can't get the token, we # just proceed without. # # More discussion at https://github.com/Hyperbase/hyperbase/pull/22259 # imds_auth = {} try: imds_ttl = {"X-aws-ec2-metadata-token-ttl-seconds": "60"} conn.request('PUT', "/latest/api/token", headers=imds_ttl) resp = conn.getresponse() resp_content = resp.read() if resp.status == 200: imds_token = base_unicodise(resp_content) imds_auth = {"X-aws-ec2-metadata-token": imds_token} except Exception: # Ensure to close the connection in case of timeout or # anything. This will avoid CannotSendRequest errors for # the next request. conn.close() conn.request('GET', "/latest/meta-data/iam/security-credentials/", headers=imds_auth) resp = conn.getresponse() files = resp.read() if resp.status == 200 and len(files) > 1: conn.request('GET', "/latest/meta-data/iam/security-credentials/%s" % files.decode('utf-8'), headers=imds_auth) resp=conn.getresponse() if resp.status == 200: resp_content = base_unicodise(resp.read()) creds = json.loads(resp_content) Config().update_option('access_key', base_unicodise(creds['AccessKeyId'])) Config().update_option('secret_key', base_unicodise(creds['SecretAccessKey'])) Config().update_option('access_token', base_unicodise(creds['Token'])) expiration = dateRFC822toPython(base_unicodise(creds['Expiration'])) # Add a timedelta to prevent any expiration if the EC2 machine is not at the right date self._access_token_expiration = expiration - datetime.timedelta(minutes=15) self._access_token_last_update = dateRFC822toPython(base_unicodise(creds['LastUpdated'])) # Others variables : Code / Type else: raise IOError else: raise IOError except Exception: raise def role_refresh(self): if self._access_token_refresh: now = datetime.datetime.now(dateutil.tz.tzutc()) if self._access_token_expiration \ and now < self._access_token_expiration \ and self._access_token_last_update \ and self._access_token_last_update <= now: # current token is still valid. No need to refresh it return try: self.role_config() except Exception: warning("Could not refresh role") def aws_credential_file(self): try: aws_credential_file = os.path.expanduser('~/.aws/credentials') credential_file_from_env = os.environ.get('AWS_CREDENTIAL_FILE') if credential_file_from_env and \ os.path.isfile(credential_file_from_env): aws_credential_file = base_unicodise(credential_file_from_env) elif not os.path.isfile(aws_credential_file): return config = PyConfigParser() debug("Reading AWS credentials from %s" % (aws_credential_file)) with io.open(aws_credential_file, "r", encoding=getattr(self, 'encoding', 'UTF-8')) as fp: config_string = fp.read() try: try: # readfp is replaced by read_file in python3, # but so far readfp it is still available. config.readfp(io.StringIO(config_string)) except MissingSectionHeaderError: # if header is missing, this could be deprecated # credentials file format as described here: # https://blog.csanchez.org/2011/05/ # then do the hacky-hack and add default header # to be able to read the file with PyConfigParser() config_string = u'[default]\n' + config_string config.readfp(io.StringIO(config_string)) except ParsingError as exc: raise ValueError( "Error reading aws_credential_file " "(%s): %s" % (aws_credential_file, str(exc))) profile = base_unicodise(os.environ.get('AWS_PROFILE', "default")) debug("Using AWS profile '%s'" % (profile)) # get_key - helper function to read the aws profile credentials # including the legacy ones as described here: # https://blog.csanchez.org/2011/05/ def get_key(profile, key, legacy_key, print_warning=True): result = None try: result = config.get(profile, key) except NoOptionError as e: # we may want to skip warning message for optional keys if print_warning: warning("Couldn't find key '%s' for the AWS Profile " "'%s' in the credentials file '%s'", e.option, e.section, aws_credential_file) # if the legacy_key defined and original one wasn't found, # try read the legacy_key if legacy_key: try: key = legacy_key profile = "default" result = config.get(profile, key) warning( "Legacy configuration key '%s' used, please use" " the standardized config format as described " "here: https://aws.amazon.com/blogs/security/a-new-and-standardized-way-to-manage-credentials-in-the-aws-sdks/", key) except NoOptionError as e: pass if result: debug("Found the configuration option '%s' for the AWS " "Profile '%s' in the credentials file %s", key, profile, aws_credential_file) return result profile_access_key = get_key(profile, "aws_access_key_id", "AWSAccessKeyId") if profile_access_key: Config().update_option('access_key', base_unicodise(profile_access_key)) profile_secret_key = get_key(profile, "aws_secret_access_key", "AWSSecretKey") if profile_secret_key: Config().update_option('secret_key', base_unicodise(profile_secret_key)) profile_access_token = get_key(profile, "aws_session_token", None, False) if profile_access_token: Config().update_option('access_token', base_unicodise(profile_access_token)) except IOError as e: warning("Errno %d accessing credentials file %s", e.errno, aws_credential_file) except NoSectionError as e: warning("Couldn't find AWS Profile '%s' in the credentials file " "'%s'", profile, aws_credential_file) def option_list(self): retval = [] for option in dir(self): ## Skip attributes that start with underscore or are not string, int or bool option_type = type(getattr(Config, option)) if option.startswith("_") or \ not (option_type in ( type(u"string"), # str type(42), # int type(True))): # bool continue retval.append(option) return retval def read_config_file(self, configfile): cp = ConfigParser(configfile) for option in self.option_list(): _option = cp.get(option) if _option is not None: _option = _option.strip() self.update_option(option, _option) # allow acl_public to be set from the config file too, even though by # default it is set to None, and not present in the config file. if cp.get('acl_public'): self.update_option('acl_public', cp.get('acl_public')) if cp.get('add_headers'): for option in cp.get('add_headers').split(","): (key, value) = option.split(':', 1) self.extra_headers[key.strip()] = value.strip() self._parsed_files.append(configfile) def dump_config(self, stream): ConfigDumper(stream).dump(u"default", self) def update_option(self, option, value): if value is None: return #### Handle environment reference if unicode(value).startswith("$"): return self.update_option(option, os.getenv(value[1:])) #### Special treatment of some options ## verbosity must be known to "logging" module if option == "verbosity": # support integer verboisities try: value = int(value) except ValueError: try: # otherwise it must be a key known to the logging module try: # python 3 support value = logging._levelNames[value] except AttributeError: value = logging._nameToLevel[value] except KeyError: raise ValueError("Config: verbosity level '%s' is not valid" % value) elif option == "limitrate": #convert kb,mb to bytes if value.endswith("k") or value.endswith("K"): shift = 10 elif value.endswith("m") or value.endswith("M"): shift = 20 else: shift = 0 try: value = shift and int(value[:-1]) << shift or int(value) except Exception: raise ValueError("Config: value of option %s must have suffix m, k, or nothing, not '%s'" % (option, value)) ## allow yes/no, true/false, on/off and 1/0 for boolean options ## Some options default to None, if that's the case check the value to see if it is bool elif (type(getattr(Config, option)) is type(True) or # Config is bool (getattr(Config, option) is None and is_bool(value))): # Config is None and value is bool if is_bool_true(value): value = True elif is_bool_false(value): value = False else: raise ValueError("Config: value of option '%s' must be Yes or No, not '%s'" % (option, value)) elif type(getattr(Config, option)) is type(42): # int try: value = int(value) except ValueError: raise ValueError("Config: value of option '%s' must be an integer, not '%s'" % (option, value)) elif option in ["host_base", "host_bucket", "cloudfront_host"]: if value.startswith("http://"): value = value[7:] elif value.startswith("https://"): value = value[8:] setattr(Config, option, value) class ConfigParser(object): def __init__(self, file, sections = []): self.cfg = {} self.parse_file(file, sections) def parse_file(self, file, sections = []): debug("ConfigParser: Reading file '%s'" % file) if type(sections) != type([]): sections = [sections] in_our_section = True r_comment = re.compile(r'^\s#.') r_empty = re.compile(r'^\s$') r_section = re.compile(r'^\[([^\]]+)\]') r_data = re.compile(r'^\s(?P<key>\w+)\s=\s(?P<value>.)') r_quotes = re.compile(r'^"(.)"\s*$') with io.open(file, "r", encoding=self.get('encoding', 'UTF-8')) as fp: for line in fp: if r_comment.match(line) or r_empty.match(line): continue is_section = r_section.match(line) if is_section: section = is_section.groups()[0] in_our_section = (section in sections) or (len(sections) == 0) continue is_data = r_data.match(line) if is_data and in_our_section: data = is_data.groupdict() if r_quotes.match(data["value"]): data["value"] = data["value"][1:-1] self.__setitem__(data["key"], data["value"]) if data["key"] in ("access_key", "secret_key", "gpg_passphrase"): print_value = ("%s...%d_chars...%s") % (data["value"][:2], len(data["value"]) - 3, data["value"][-1:]) else: print_value = data["value"] debug("ConfigParser: %s->%s" % (data["key"], print_value)) continue warning("Ignoring invalid line in '%s': %s" % (file, line)) def __getitem__(self, name): return self.cfg[name] def __setitem__(self, name, value): self.cfg[name] = value def get(self, name, default = None): if name in self.cfg: return self.cfg[name] return default class ConfigDumper(object): def __init__(self, stream): self.stream = stream def dump(self, section, config): self.stream.write(u"[%s]\n" % section) for option in config.option_list(): value = getattr(config, option) if option == "verbosity": # we turn level numbers back into strings if possible if isinstance(value, int): try: try: # python 3 support value = logging._levelNames[value] except AttributeError: value = logging._levelToName[value] except KeyError: pass self.stream.write(u"%s = %s\n" % (option, value)) # vim:et:ts=4:sts=4:ai	29.24 KB	2023-12-06 09:07
📄 ConnMan.py # -- coding: utf-8 -- ## -------------------------------------------------------------------- ## Amazon S3 manager ## ## Authors : Michal Ludvig <michal@logix.cz> (https://www.logix.cz/michal) ## Florent Viard <florent@sodria.com> (https://www.sodria.com) ## Copyright : TGRMN Software, Sodria SAS and contributors ## License : GPL Version 2 ## Website : https://s3tools.org ## -------------------------------------------------------------------- from __future__ import absolute_import import sys if sys.version_info >= (3, 0): from .Custom_httplib3x import httplib else: from .Custom_httplib27 import httplib import ssl from logging import debug from threading import Semaphore from time import time try: # python 3 support from urlparse import urlparse except ImportError: from urllib.parse import urlparse from .Config import Config from .Exceptions import ParameterError, S3SSLCertificateError from .Utils import getBucketFromHostname __all__ = ["ConnMan"] class http_connection(object): context = None context_set = False @staticmethod def _ssl_verified_context(cafile): cfg = Config() context = None try: context = ssl.create_default_context(cafile=cafile) except AttributeError: # no ssl.create_default_context pass if context and not cfg.check_ssl_hostname: context.check_hostname = False debug(u'Disabling SSL certificate hostname checking') return context @staticmethod def _ssl_unverified_context(cafile): debug(u'Disabling SSL certificate checking') context = None try: context = ssl._create_unverified_context(cafile=cafile, cert_reqs=ssl.CERT_NONE) except AttributeError: # no ssl._create_unverified_context pass return context @staticmethod def _ssl_client_auth_context(certfile, keyfile, check_server_cert, cafile): context = None try: cert_reqs = ssl.CERT_REQUIRED if check_server_cert else ssl.CERT_NONE context = ssl._create_unverified_context(cafile=cafile, keyfile=keyfile, certfile=certfile, cert_reqs=cert_reqs) except AttributeError: # no ssl._create_unverified_context pass return context @staticmethod def _ssl_context(): if http_connection.context_set: return http_connection.context cfg = Config() cafile = cfg.ca_certs_file if cafile == "": cafile = None certfile = cfg.ssl_client_cert_file or None keyfile = cfg.ssl_client_key_file or None # the key may be embedded into cert file debug(u"Using ca_certs_file %s", cafile) debug(u"Using ssl_client_cert_file %s", certfile) debug(u"Using ssl_client_key_file %s", keyfile) if certfile is not None: context = http_connection._ssl_client_auth_context(certfile, keyfile, cfg.check_ssl_certificate, cafile) elif cfg.check_ssl_certificate: context = http_connection._ssl_verified_context(cafile) else: context = http_connection._ssl_unverified_context(cafile) http_connection.context = context http_connection.context_set = True return context def forgive_wildcard_cert(self, cert, hostname): """ Wildcard matching for .s3.amazonaws.com and similar per region. Per http://docs.aws.amazon.com/AmazonS3/latest/dev/BucketRestrictions.html: "We recommend that all bucket names comply with DNS naming conventions." Per http://docs.aws.amazon.com/AmazonS3/latest/dev/VirtualHosting.html: "When using virtual hosted-style buckets with SSL, the SSL wild card certificate only matches buckets that do not contain periods. To work around this, use HTTP or write your own certificate verification logic." Therefore, we need a custom validation routine that allows mybucket.example.com.s3.amazonaws.com to be considered a valid hostname for the .s3.amazonaws.com wildcard cert, and for the region-specific .s3-[region].amazonaws.com wildcard cert. We also forgive non-S3 wildcard certificates should the hostname match, to allow compatibility with other S3 API-compatible storage providers. """ debug(u'checking SSL subjectAltName as forgiving wildcard cert') san = cert.get('subjectAltName', ()) hostname = hostname.lower() cleaned_host_bucket_config = urlparse('https://' + Config.host_bucket).hostname for key, value in san: if key == 'DNS': value = value.lower() if value.startswith('.s3') and \ (value.endswith('.amazonaws.com') and hostname.endswith('.amazonaws.com')) or \ (value.endswith('.amazonaws.com.cn') and hostname.endswith('.amazonaws.com.cn')): return True elif value == cleaned_host_bucket_config % \ {'bucket': '*', 'location': Config.bucket_location.lower()} and \ hostname.endswith(cleaned_host_bucket_config % \ {'bucket': '', 'location': Config.bucket_location.lower()}): return True return False def match_hostname(self): cert = self.c.sock.getpeercert() try: ssl.match_hostname(cert, self.hostname) except AttributeError: # old ssl module doesn't have this function return except ValueError: # empty SSL cert means underlying SSL library didn't validate it, we don't either. return except S3CertificateError as e: if not self.forgive_wildcard_cert(cert, self.hostname): raise e @staticmethod def _https_connection(hostname, port=None): try: context = http_connection._ssl_context() # Wildcard certificates do not work with DNS-style named buckets. bucket_name, success = getBucketFromHostname(hostname) if success and '.' in bucket_name: # this merely delays running the hostname check until # after the connection is made and we get control # back. We then run the same check, relaxed for S3's # wildcard certificates. debug(u'Bucket name contains "." character, disabling initial SSL hostname check') check_hostname = False if context: context.check_hostname = False else: if context: check_hostname = context.check_hostname else: # Earliest version of python that don't have context, # don't check hostnames anyway check_hostname = True # Note, we are probably needed to try to set check_hostname because of that bug: # http://bugs.python.org/issue22959 conn = httplib.HTTPSConnection(hostname, port, context=context, check_hostname=check_hostname) debug(u'httplib.HTTPSConnection() has both context and check_hostname') except TypeError: try: # in case check_hostname parameter is not present try again conn = httplib.HTTPSConnection(hostname, port, context=context) debug(u'httplib.HTTPSConnection() has only context') except TypeError: # in case even context parameter is not present try one last time conn = httplib.HTTPSConnection(hostname, port) debug(u'httplib.HTTPSConnection() has neither context nor check_hostname') return conn def __init__(self, id, hostname, ssl, cfg): self.ssl = ssl self.id = id self.counter = 0 # Whatever is the input, ensure to have clean hostname and port parsed_hostname = urlparse('https://' + hostname) self.hostname = parsed_hostname.hostname self.port = parsed_hostname.port if parsed_hostname.path and parsed_hostname.path != '/': self.path = parsed_hostname.path.rstrip('/') debug(u'endpoint path set to %s', self.path) else: self.path = None """ History note: In a perfect world, or in the future: - All http proxies would support CONNECT/tunnel, and so there would be no need for using "absolute URIs" in format_uri. - All s3-like servers would work well whether using relative or ABSOLUTE URIs. But currently, what is currently common: - Proxies without support for CONNECT for http, and so "absolute URIs" have to be used. - Proxies with support for CONNECT for httpS but s3-like servers having issues with "absolute URIs", so relative one still have to be used as the requests will pass as-is, through the proxy because of the CONNECT mode. """ if not cfg.proxy_host: if ssl: self.c = http_connection._https_connection(self.hostname, self.port) debug(u'non-proxied HTTPSConnection(%s, %s)', self.hostname, self.port) else: self.c = httplib.HTTPConnection(self.hostname, self.port) debug(u'non-proxied HTTPConnection(%s, %s)', self.hostname, self.port) else: if ssl: self.c = http_connection._https_connection(cfg.proxy_host, cfg.proxy_port) debug(u'proxied HTTPSConnection(%s, %s)', cfg.proxy_host, cfg.proxy_port) port = self.port and self.port or 443 self.c.set_tunnel(self.hostname, port) debug(u'tunnel to %s, %s', self.hostname, port) else: self.c = httplib.HTTPConnection(cfg.proxy_host, cfg.proxy_port) debug(u'proxied HTTPConnection(%s, %s)', cfg.proxy_host, cfg.proxy_port) # No tunnel here for the moment self.last_used_time = time() class ConnMan(object): _CS_REQ_SENT = httplib._CS_REQ_SENT CONTINUE = httplib.CONTINUE conn_pool_sem = Semaphore() conn_pool = {} conn_max_counter = 800 ## AWS closes connection after some ~90 requests @staticmethod def get(hostname, ssl=None): cfg = Config() if ssl is None: ssl = cfg.use_https conn = None if cfg.proxy_host != "": if ssl and sys.hexversion < 0x02070000: raise ParameterError("use_https=True can't be used with proxy on Python <2.7") conn_id = "proxy://%s:%s" % (cfg.proxy_host, cfg.proxy_port) else: conn_id = "http%s://%s" % (ssl and "s" or "", hostname) ConnMan.conn_pool_sem.acquire() if conn_id not in ConnMan.conn_pool: ConnMan.conn_pool[conn_id] = [] while ConnMan.conn_pool[conn_id]: conn = ConnMan.conn_pool[conn_id].pop() cur_time = time() if cur_time < conn.last_used_time + cfg.connection_max_age \ and cur_time >= conn.last_used_time: debug("ConnMan.get(): re-using connection: %s#%d" % (conn.id, conn.counter)) break # Conn is too old or wall clock went back in the past debug("ConnMan.get(): closing expired connection") ConnMan.close(conn) conn = None ConnMan.conn_pool_sem.release() if not conn: debug("ConnMan.get(): creating new connection: %s" % conn_id) conn = http_connection(conn_id, hostname, ssl, cfg) conn.c.connect() if conn.ssl and cfg.check_ssl_certificate and cfg.check_ssl_hostname: conn.match_hostname() conn.counter += 1 return conn @staticmethod def put(conn): if conn.id.startswith("proxy://"): ConnMan.close(conn) debug("ConnMan.put(): closing proxy connection (keep-alive not yet" " supported)") return if conn.counter >= ConnMan.conn_max_counter: ConnMan.close(conn) debug("ConnMan.put(): closing over-used connection") return cfg = Config() if not cfg.connection_pooling: ConnMan.close(conn) debug("ConnMan.put(): closing connection (connection pooling disabled)") return # Update timestamp of conn to record when was its last use conn.last_used_time = time() ConnMan.conn_pool_sem.acquire() ConnMan.conn_pool[conn.id].append(conn) ConnMan.conn_pool_sem.release() debug("ConnMan.put(): connection put back to pool (%s#%d)" % (conn.id, conn.counter)) @staticmethod def close(conn): if conn: conn.c.close()	12.98 KB	2023-12-06 09:07
📄 Crypto.py # -- coding: utf-8 -- ## -------------------------------------------------------------------- ## Amazon S3 manager ## ## Authors : Michal Ludvig <michal@logix.cz> (https://www.logix.cz/michal) ## Florent Viard <florent@sodria.com> (https://www.sodria.com) ## Copyright : TGRMN Software, Sodria SAS and contributors ## License : GPL Version 2 ## Website : https://s3tools.org ## -------------------------------------------------------------------- from __future__ import absolute_import import sys import hmac try: from base64 import encodebytes as encodestring except ImportError: # Python 2 support from base64 import encodestring from . import Config from logging import debug from .BaseUtils import encode_to_s3, decode_from_s3, s3_quote, md5, unicode from .Utils import time_to_epoch, deunicodise, check_bucket_name_dns_support from .SortedDict import SortedDict import datetime from hashlib import sha1, sha256 __all__ = [] def format_param_str(params, always_have_equal=False, limited_keys=None): """ Format URL parameters from a params dict and returns ?parm1=val1&parm2=val2 or an empty string if there are no parameters. Output of this function should be appended directly to self.resource['uri'] - Set "always_have_equal" to always have the "=" char for a param even when there is no value for it. - Set "limited_keys" list to restrict the param string to keys that are defined in it. """ if not params: return "" param_str = "" equal_str = always_have_equal and u'=' or '' for key in sorted(params.keys()): if limited_keys and key not in limited_keys: continue value = params[key] if value in (None, ""): param_str += "&%s%s" % (s3_quote(key, unicode_output=True), equal_str) else: param_str += "&%s=%s" % (key, s3_quote(params[key], unicode_output=True)) return param_str and "?" + param_str[1:] __all__.append("format_param_str") ### AWS Version 2 signing def sign_string_v2(string_to_sign): """Sign a string with the secret key, returning base64 encoded results. By default the configured secret key is used, but may be overridden as an argument. Useful for REST authentication. See http://s3.amazonaws.com/doc/s3-developer-guide/RESTAuthentication.html string_to_sign should be utf-8 "bytes". and returned signature will be utf-8 encoded "bytes". """ secret_key = Config.Config().secret_key signature = encodestring(hmac.new(encode_to_s3(secret_key), string_to_sign, sha1).digest()).strip() return signature __all__.append("sign_string_v2") def sign_request_v2(method='GET', canonical_uri='/', params=None, cur_headers=None): """Sign a string with the secret key, returning base64 encoded results. By default the configured secret key is used, but may be overridden as an argument. Useful for REST authentication. See http://s3.amazonaws.com/doc/s3-developer-guide/RESTAuthentication.html string_to_sign should be utf-8 "bytes". """ # valid sub-resources to be included in sign v2: SUBRESOURCES_TO_INCLUDE = ['acl', 'lifecycle', 'location', 'logging', 'notification', 'partNumber', 'policy', 'requestPayment', 'tagging', 'torrent', 'uploadId', 'uploads', 'versionId', 'versioning', 'versions', 'website', # Missing of aws s3 doc but needed 'delete', 'cors', 'restore'] if cur_headers is None: cur_headers = SortedDict(ignore_case = True) access_key = Config.Config().access_key string_to_sign = method + "\n" string_to_sign += cur_headers.get("content-md5", "") + "\n" string_to_sign += cur_headers.get("content-type", "") + "\n" string_to_sign += cur_headers.get("date", "") + "\n" for header in sorted(cur_headers.keys()): if header.startswith("x-amz-"): string_to_sign += header + ":" + cur_headers[header] + "\n" if header.startswith("x-emc-"): string_to_sign += header + ":"+ cur_headers[header] + "\n" canonical_uri = s3_quote(canonical_uri, quote_backslashes=False, unicode_output=True) canonical_querystring = format_param_str(params, limited_keys=SUBRESOURCES_TO_INCLUDE) # canonical_querystring would be empty if no param given, otherwise it will # starts with a "?" canonical_uri += canonical_querystring string_to_sign += canonical_uri debug("SignHeaders: " + repr(string_to_sign)) signature = decode_from_s3(sign_string_v2(encode_to_s3(string_to_sign))) new_headers = SortedDict(list(cur_headers.items()), ignore_case=True) new_headers["Authorization"] = "AWS " + access_key + ":" + signature return new_headers __all__.append("sign_request_v2") def sign_url_v2(url_to_sign, expiry): """Sign a URL in s3://bucket/object form with the given expiry time. The object will be accessible via the signed URL until the AWS key and secret are revoked or the expiry time is reached, even if the object is otherwise private. See: http://s3.amazonaws.com/doc/s3-developer-guide/RESTAuthentication.html """ return sign_url_base_v2( bucket = url_to_sign.bucket(), object = url_to_sign.object(), expiry = expiry ) __all__.append("sign_url_v2") def sign_url_base_v2(*parms): """Shared implementation of sign_url methods. Takes a hash of 'bucket', 'object' and 'expiry' as args.""" content_disposition=Config.Config().content_disposition content_type=Config.Config().content_type parms['expiry']=time_to_epoch(parms['expiry']) parms['access_key']=Config.Config().access_key parms['host_base']=Config.Config().host_base parms['object'] = s3_quote(parms['object'], quote_backslashes=False, unicode_output=True) parms['proto'] = 'http' if Config.Config().signurl_use_https: parms['proto'] = 'https' debug("Expiry interpreted as epoch time %s", parms['expiry']) signtext = 'GET\n\n\n%(expiry)d\n/%(bucket)s/%(object)s' % parms param_separator = '?' if content_disposition: signtext += param_separator + 'response-content-disposition=' + content_disposition param_separator = '&' if content_type: signtext += param_separator + 'response-content-type=' + content_type param_separator = '&' debug("Signing plaintext: %r", signtext) parms['sig'] = s3_quote(sign_string_v2(encode_to_s3(signtext)), unicode_output=True) debug("Urlencoded signature: %s", parms['sig']) if check_bucket_name_dns_support(Config.Config().host_bucket, parms['bucket']): url = "%(proto)s://%(bucket)s.%(host_base)s/%(object)s" else: url = "%(proto)s://%(host_base)s/%(bucket)s/%(object)s" url += "?AWSAccessKeyId=%(access_key)s&Expires=%(expiry)d&Signature=%(sig)s" url = url % parms if content_disposition: url += "&response-content-disposition=" + s3_quote(content_disposition, unicode_output=True) if content_type: url += "&response-content-type=" + s3_quote(content_type, unicode_output=True) return url __all__.append("sign_url_base_v2") def sign(key, msg): return hmac.new(key, encode_to_s3(msg), sha256).digest() def getSignatureKey(key, dateStamp, regionName, serviceName): """ Input: unicode params Output: bytes """ kDate = sign(encode_to_s3('AWS4' + key), dateStamp) kRegion = sign(kDate, regionName) kService = sign(kRegion, serviceName) kSigning = sign(kService, 'aws4_request') return kSigning def sign_request_v4(method='GET', host='', canonical_uri='/', params=None, region='us-east-1', cur_headers=None, body=b''): service = 's3' if cur_headers is None: cur_headers = SortedDict(ignore_case = True) cfg = Config.Config() access_key = cfg.access_key secret_key = cfg.secret_key t = datetime.datetime.utcnow() amzdate = t.strftime('%Y%m%dT%H%M%SZ') datestamp = t.strftime('%Y%m%d') signing_key = getSignatureKey(secret_key, datestamp, region, service) canonical_uri = s3_quote(canonical_uri, quote_backslashes=False, unicode_output=True) canonical_querystring = format_param_str(params, always_have_equal=True).lstrip('?') if type(body) == type(sha256(b'')): payload_hash = decode_from_s3(body.hexdigest()) else: payload_hash = decode_from_s3(sha256(encode_to_s3(body)).hexdigest()) canonical_headers = {'host' : host, 'x-amz-content-sha256': payload_hash, 'x-amz-date' : amzdate } signed_headers = 'host;x-amz-content-sha256;x-amz-date' for header in cur_headers.keys(): # avoid duplicate headers and previous Authorization if header == 'Authorization' or header in signed_headers.split(';'): continue canonical_headers[header.strip()] = cur_headers[header].strip() signed_headers += ';' + header.strip() # sort headers into a string canonical_headers_str = '' for k, v in sorted(canonical_headers.items()): canonical_headers_str += k + ":" + v + "\n" canonical_headers = canonical_headers_str debug(u"canonical_headers = %s" % canonical_headers) signed_headers = ';'.join(sorted(signed_headers.split(';'))) canonical_request = method + '\n' + canonical_uri + '\n' + canonical_querystring + '\n' + canonical_headers + '\n' + signed_headers + '\n' + payload_hash debug('Canonical Request:\n%s\n----------------------' % canonical_request) algorithm = 'AWS4-HMAC-SHA256' credential_scope = datestamp + '/' + region + '/' + service + '/' + 'aws4_request' string_to_sign = algorithm + '\n' + amzdate + '\n' + credential_scope + '\n' + decode_from_s3(sha256(encode_to_s3(canonical_request)).hexdigest()) signature = decode_from_s3(hmac.new(signing_key, encode_to_s3(string_to_sign), sha256).hexdigest()) authorization_header = algorithm + ' ' + 'Credential=' + access_key + '/' + credential_scope + ',' + 'SignedHeaders=' + signed_headers + ',' + 'Signature=' + signature new_headers = SortedDict(cur_headers.items()) new_headers.update({'x-amz-date':amzdate, 'Authorization':authorization_header, 'x-amz-content-sha256': payload_hash}) debug("signature-v4 headers: %s" % new_headers) return new_headers __all__.append("sign_request_v4") def checksum_file_descriptor(file_desc, offset=0, size=None, hash_func=sha256): hash = hash_func() if size is None: for chunk in iter(lambda: file_desc.read(8192), b''): hash.update(chunk) else: file_desc.seek(offset) size_left = size while size_left > 0: chunk = file_desc.read(min(8192, size_left)) if not chunk: break size_left -= len(chunk) hash.update(chunk) return hash __all__.append("checksum_file_stream") def checksum_sha256_file(file, offset=0, size=None): if not isinstance(file, unicode): # file is directly a file descriptor return checksum_file_descriptor(file, offset, size, sha256) # Otherwise, we expect file to be a filename with open(deunicodise(file),'rb') as fp: return checksum_file_descriptor(fp, offset, size, sha256) __all__.append("checksum_sha256_file") def checksum_sha256_buffer(buffer, offset=0, size=None): hash = sha256() if size is None: hash.update(buffer) else: hash.update(buffer[offset:offset+size]) return hash __all__.append("checksum_sha256_buffer") def generate_content_md5(body): m = md5(encode_to_s3(body)) base64md5 = encodestring(m.digest()) base64md5 = decode_from_s3(base64md5) if base64md5[-1] == '\n': base64md5 = base64md5[0:-1] return decode_from_s3(base64md5) __all__.append("generate_content_md5") def hash_file_md5(filename): h = md5() with open(deunicodise(filename), "rb") as fp: while True: # Hash 32kB chunks data = fp.read(321024) if not data: break h.update(data) return h.hexdigest() __all__.append("hash_file_md5") def calculateChecksum(buffer, mfile, offset, chunk_size, send_chunk): md5_hash = md5() size_left = chunk_size if buffer == '': mfile.seek(offset) while size_left > 0: data = mfile.read(min(send_chunk, size_left)) if not data: break md5_hash.update(data) size_left -= len(data) else: md5_hash.update(buffer) return md5_hash.hexdigest() __all__.append("calculateChecksum")	12.62 KB	2023-12-12 01:00
📄 Custom_httplib27.py from __future__ import absolute_import, print_function import os import httplib from httplib import (_CS_REQ_SENT, _CS_REQ_STARTED, CONTINUE, UnknownProtocol, CannotSendHeader, NO_CONTENT, NOT_MODIFIED, EXPECTATION_FAILED, HTTPMessage, HTTPException) try: from cStringIO import StringIO except ImportError: from StringIO import StringIO from .BaseUtils import encode_to_s3 _METHODS_EXPECTING_BODY = ['PATCH', 'POST', 'PUT'] # Fixed python 2.X httplib to be able to support # Expect: 100-Continue http feature # Inspired by: # http://bugs.python.org/file26357/issue1346874-273.patch def httpresponse_patched_begin(self): """ Re-implemented httplib begin function to not loop over "100 CONTINUE" status replies but to report it to higher level so it can be processed. """ if self.msg is not None: # we've already started reading the response return # read only one status even if we get a non-100 response version, status, reason = self._read_status() self.status = status self.reason = reason.strip() if version == 'HTTP/1.0': self.version = 10 elif version.startswith('HTTP/1.'): self.version = 11 # use HTTP/1.1 code for HTTP/1.x where x>=1 elif version == 'HTTP/0.9': self.version = 9 else: raise UnknownProtocol(version) if self.version == 9: self.length = None self.chunked = 0 self.will_close = 1 self.msg = HTTPMessage(StringIO()) return self.msg = HTTPMessage(self.fp, 0) if self.debuglevel > 0: for hdr in self.msg.headers: print("header:", hdr, end=" ") # don't let the msg keep an fp self.msg.fp = None # are we using the chunked-style of transfer encoding? tr_enc = self.msg.getheader('transfer-encoding') if tr_enc and tr_enc.lower() == "chunked": self.chunked = 1 self.chunk_left = None else: self.chunked = 0 # will the connection close at the end of the response? self.will_close = self._check_close() # do we have a Content-Length? # NOTE: RFC 2616, S4.4, #3 says we ignore this if tr_enc is "chunked" length = self.msg.getheader('content-length') if length and not self.chunked: try: self.length = int(length) except ValueError: self.length = None else: if self.length < 0: # ignore nonsensical negative lengths self.length = None else: self.length = None # does the body have a fixed length? (of zero) if (status == NO_CONTENT or status == NOT_MODIFIED or 100 <= status < 200 or # 1xx codes self._method == 'HEAD'): self.length = 0 # if the connection remains open, and we aren't using chunked, and # a content-length was not provided, then assume that the connection # WILL close. if not self.will_close and \ not self.chunked and \ self.length is None: self.will_close = 1 def httpconnection_patched_set_content_length(self, body, method): ## REIMPLEMENTED because new in last httplib but needed by send_request # Set the content-length based on the body. If the body is "empty", we # set Content-Length: 0 for methods that expect a body (RFC 7230, # Section 3.3.2). If the body is set for other methods, we set the # header provided we can figure out what the length is. thelen = None if body is None and method.upper() in _METHODS_EXPECTING_BODY: thelen = '0' elif body is not None: try: thelen = str(len(body)) except (TypeError, AttributeError): # If this is a file-like object, try to # fstat its file descriptor try: thelen = str(os.fstat(body.fileno()).st_size) except (AttributeError, OSError): # Don't send a length if this failed if self.debuglevel > 0: print("Cannot stat!!") if thelen is not None: self.putheader('Content-Length', thelen) def httpconnection_patched_send_request(self, method, url, body, headers): # Honor explicitly requested Host: and Accept-Encoding: headers. header_names = dict.fromkeys([k.lower() for k in headers]) skips = {} if 'host' in header_names: skips['skip_host'] = 1 if 'accept-encoding' in header_names: skips['skip_accept_encoding'] = 1 expect_continue = False for hdr, value in headers.iteritems(): if 'expect' == hdr.lower() and '100-continue' in value.lower(): expect_continue = True url = encode_to_s3(url) self.putrequest(method, url, **skips) if 'content-length' not in header_names: self._set_content_length(body, method) for hdr, value in headers.iteritems(): self.putheader(encode_to_s3(hdr), encode_to_s3(value)) # If an Expect: 100-continue was sent, we need to check for a 417 # Expectation Failed to avoid unnecessarily sending the body # See RFC 2616 8.2.3 if not expect_continue: self.endheaders(body) else: if not body: raise HTTPException("A body is required when expecting " "100-continue") self.endheaders() resp = self.getresponse() resp.read() self._HTTPConnection__state = _CS_REQ_SENT if resp.status == EXPECTATION_FAILED: raise ExpectationFailed() elif resp.status == CONTINUE: self.send(body) def httpconnection_patched_endheaders(self, message_body=None): """Indicate that the last header line has been sent to the server. This method sends the request to the server. The optional message_body argument can be used to pass a message body associated with the request. The message body will be sent in the same packet as the message headers if it is string, otherwise it is sent as a separate packet. """ if self._HTTPConnection__state == _CS_REQ_STARTED: self._HTTPConnection__state = _CS_REQ_SENT else: raise CannotSendHeader() self._send_output(message_body) # TCP Maximum Segment Size (MSS) is determined by the TCP stack on # a per-connection basis. There is no simple and efficient # platform independent mechanism for determining the MSS, so # instead a reasonable estimate is chosen. The getsockopt() # interface using the TCP_MAXSEG parameter may be a suitable # approach on some operating systems. A value of 16KiB is chosen # as a reasonable estimate of the maximum MSS. mss = 16384 def httpconnection_patched_send_output(self, message_body=None): """Send the currently buffered request and clear the buffer. Appends an extra \\r\\n to the buffer. A message_body may be specified, to be appended to the request. """ self._buffer.extend((b"", b"")) msg = b"\r\n".join(self._buffer) del self._buffer[:] msg = encode_to_s3(msg) # If msg and message_body are sent in a single send() call, # it will avoid performance problems caused by the interaction # between delayed ack and the Nagle algorithm. if isinstance(message_body, str) and len(message_body) < mss: msg += message_body message_body = None self.send(msg) if message_body is not None: #message_body was not a string (i.e. it is a file) and #we must run the risk of Nagle self.send(message_body) class ExpectationFailed(HTTPException): pass # Wrappers # def httpconnection_patched_wrapper_send_body(self, message_body): self.send(message_body) httplib.HTTPResponse.begin = httpresponse_patched_begin httplib.HTTPConnection.endheaders = httpconnection_patched_endheaders httplib.HTTPConnection._send_output = httpconnection_patched_send_output httplib.HTTPConnection._set_content_length = httpconnection_patched_set_content_length httplib.HTTPConnection._send_request = httpconnection_patched_send_request # Interfaces added to httplib.HTTPConnection: httplib.HTTPConnection.wrapper_send_body = httpconnection_patched_wrapper_send_body	7.99 KB	2023-12-06 09:07
📄 Custom_httplib3x.py from __future__ import absolute_import, print_function import os import sys import http.client as httplib from http.client import (_CS_REQ_SENT, _CS_REQ_STARTED, CONTINUE, UnknownProtocol, CannotSendHeader, NO_CONTENT, NOT_MODIFIED, EXPECTATION_FAILED, HTTPMessage, HTTPException) from io import StringIO from .BaseUtils import encode_to_s3 _METHODS_EXPECTING_BODY = ['PATCH', 'POST', 'PUT'] # Fixed python 2.X httplib to be able to support # Expect: 100-Continue http feature # Inspired by: # http://bugs.python.org/file26357/issue1346874-273.patch def _encode(data, name='data'): """Call data.encode("latin-1") but show a better error message.""" try: return data.encode("latin-1") except UnicodeEncodeError as err: # The following is equivalent to raise Exception() from None # but is still byte-compilable compatible with python 2. exc = UnicodeEncodeError( err.encoding, err.object, err.start, err.end, "%s (%.20r) is not valid Latin-1. Use %s.encode('utf-8') " "if you want to send it encoded in UTF-8." % (name.title(), data[err.start:err.end], name)) exc.__cause__ = None raise exc def httpresponse_patched_begin(self): """ Re-implemented httplib begin function to not loop over "100 CONTINUE" status replies but to report it to higher level so it can be processed. """ if self.headers is not None: # we've already started reading the response return # read only one status even if we get a non-100 response version, status, reason = self._read_status() self.code = self.status = status self.reason = reason.strip() if version in ('HTTP/1.0', 'HTTP/0.9'): # Some servers might still return "0.9", treat it as 1.0 anyway self.version = 10 elif version.startswith('HTTP/1.'): self.version = 11 # use HTTP/1.1 code for HTTP/1.x where x>=1 else: raise UnknownProtocol(version) self.headers = self.msg = httplib.parse_headers(self.fp) if self.debuglevel > 0: for hdr in self.headers: print("header:", hdr, end=" ") # are we using the chunked-style of transfer encoding? tr_enc = self.headers.get('transfer-encoding') if tr_enc and tr_enc.lower() == "chunked": self.chunked = True self.chunk_left = None else: self.chunked = False # will the connection close at the end of the response? self.will_close = self._check_close() # do we have a Content-Length? # NOTE: RFC 2616, S4.4, #3 says we ignore this if tr_enc is "chunked" self.length = None length = self.headers.get('content-length') if length and not self.chunked: try: self.length = int(length) except ValueError: self.length = None else: if self.length < 0: # ignore nonsensical negative lengths self.length = None else: self.length = None # does the body have a fixed length? (of zero) if (status == NO_CONTENT or status == NOT_MODIFIED or 100 <= status < 200 or # 1xx codes self._method == 'HEAD'): self.length = 0 # if the connection remains open, and we aren't using chunked, and # a content-length was not provided, then assume that the connection # WILL close. if (not self.will_close and not self.chunked and self.length is None): self.will_close = True # No need to override httplib with this one, as it is only used by send_request def httpconnection_patched_get_content_length(body, method): """## REIMPLEMENTED because new in last httplib but needed by send_request""" """Get the content-length based on the body. If the body is None, we set Content-Length: 0 for methods that expect a body (RFC 7230, Section 3.3.2). We also set the Content-Length for any method if the body is a str or bytes-like object and not a file. """ if body is None: # do an explicit check for not None here to distinguish # between unset and set but empty if method.upper() in _METHODS_EXPECTING_BODY: return 0 else: return None if hasattr(body, 'read'): # file-like object. return None try: # does it implement the buffer protocol (bytes, bytearray, array)? mv = memoryview(body) return mv.nbytes except TypeError: pass if isinstance(body, str): return len(body) return None def httpconnection_patched_send_request(self, method, url, body, headers, encode_chunked=False): # Honor explicitly requested Host: and Accept-Encoding: headers. header_names = dict.fromkeys([k.lower() for k in headers]) skips = {} if 'host' in header_names: skips['skip_host'] = 1 if 'accept-encoding' in header_names: skips['skip_accept_encoding'] = 1 expect_continue = False for hdr, value in headers.items(): if 'expect' == hdr.lower() and '100-continue' in value.lower(): expect_continue = True self.putrequest(method, url, **skips) # chunked encoding will happen if HTTP/1.1 is used and either # the caller passes encode_chunked=True or the following # conditions hold: # 1. content-length has not been explicitly set # 2. the body is a file or iterable, but not a str or bytes-like # 3. Transfer-Encoding has NOT been explicitly set by the caller if 'content-length' not in header_names: # only chunk body if not explicitly set for backwards # compatibility, assuming the client code is already handling the # chunking if 'transfer-encoding' not in header_names: # if content-length cannot be automatically determined, fall # back to chunked encoding encode_chunked = False content_length = httpconnection_patched_get_content_length(body, method) if content_length is None: if body is not None: if self.debuglevel > 0: print('Unable to determine size of %r' % body) encode_chunked = True self.putheader('Transfer-Encoding', 'chunked') else: self.putheader('Content-Length', str(content_length)) else: encode_chunked = False for hdr, value in headers.items(): self.putheader(encode_to_s3(hdr), encode_to_s3(value)) if isinstance(body, str): # RFC 2616 Section 3.7.1 says that text default has a # default charset of iso-8859-1. body = _encode(body, 'body') # If an Expect: 100-continue was sent, we need to check for a 417 # Expectation Failed to avoid unnecessarily sending the body # See RFC 2616 8.2.3 if not expect_continue: self.endheaders(body, encode_chunked=encode_chunked) else: if not body: raise HTTPException("A body is required when expecting " "100-continue") self.endheaders() resp = self.getresponse() resp.read() self._HTTPConnection__state = _CS_REQ_SENT if resp.status == EXPECTATION_FAILED: raise ExpectationFailed() elif resp.status == CONTINUE: self.wrapper_send_body(body, encode_chunked) def httpconnection_patched_endheaders(self, message_body=None, encode_chunked=False): """REIMPLEMENTED because new argument encode_chunked added after py 3.4""" """Indicate that the last header line has been sent to the server. This method sends the request to the server. The optional message_body argument can be used to pass a message body associated with the request. """ if self._HTTPConnection__state == _CS_REQ_STARTED: self._HTTPConnection__state = _CS_REQ_SENT else: raise CannotSendHeader() self._send_output(message_body, encode_chunked=encode_chunked) def httpconnection_patched_read_readable(self, readable): """REIMPLEMENTED because needed by send_output and added after py 3.4 """ blocksize = 8192 if self.debuglevel > 0: print("sendIng a read()able") encode = self._is_textIO(readable) if encode and self.debuglevel > 0: print("encoding file using iso-8859-1") while True: datablock = readable.read(blocksize) if not datablock: break if encode: datablock = datablock.encode("iso-8859-1") yield datablock def httpconnection_patched_send_output(self, message_body=None, encode_chunked=False): """REIMPLEMENTED because needed by endheaders and parameter encode_chunked was added""" """Send the currently buffered request and clear the buffer. Appends an extra \\r\\n to the buffer. A message_body may be specified, to be appended to the request. """ self._buffer.extend((b"", b"")) msg = b"\r\n".join(self._buffer) del self._buffer[:] self.send(msg) if message_body is not None: self.wrapper_send_body(message_body, encode_chunked) class ExpectationFailed(HTTPException): pass # Wrappers # def httpconnection_patched_wrapper_send_body(self, message_body, encode_chunked=False): # create a consistent interface to message_body if hasattr(message_body, 'read'): # Let file-like take precedence over byte-like. This # is needed to allow the current position of mmap'ed # files to be taken into account. chunks = self._read_readable(message_body) else: try: # this is solely to check to see if message_body # implements the buffer API. it /would/ be easier # to capture if PyObject_CheckBuffer was exposed # to Python. memoryview(message_body) except TypeError: try: chunks = iter(message_body) except TypeError: raise TypeError("message_body should be a bytes-like " "object or an iterable, got %r" % type(message_body)) else: # the object implements the buffer interface and # can be passed directly into socket methods chunks = (message_body,) for chunk in chunks: if not chunk: if self.debuglevel > 0: print('Zero length chunk ignored') continue if encode_chunked and self._http_vsn == 11: # chunked encoding chunk = '{:X}\r\n'.format(len(chunk)).encode('ascii') + chunk \ + b'\r\n' self.send(chunk) if encode_chunked and self._http_vsn == 11: # end chunked transfer self.send(b'0\r\n\r\n') httplib.HTTPResponse.begin = httpresponse_patched_begin httplib.HTTPConnection.endheaders = httpconnection_patched_endheaders httplib.HTTPConnection._send_readable = httpconnection_patched_read_readable httplib.HTTPConnection._send_output = httpconnection_patched_send_output httplib.HTTPConnection._send_request = httpconnection_patched_send_request # Interfaces added to httplib.HTTPConnection: httplib.HTTPConnection.wrapper_send_body = httpconnection_patched_wrapper_send_body	11.24 KB	2023-12-06 09:07
📄 Exceptions.py # -- coding: utf-8 -- ## -------------------------------------------------------------------- ## Amazon S3 manager - Exceptions library ## ## Authors : Michal Ludvig <michal@logix.cz> (https://www.logix.cz/michal) ## Florent Viard <florent@sodria.com> (https://www.sodria.com) ## Copyright : TGRMN Software, Sodria SAS and contributors ## License : GPL Version 2 ## Website : https://s3tools.org ## -------------------------------------------------------------------- from __future__ import absolute_import from logging import debug, error import sys import S3.BaseUtils import S3.Utils from . import ExitCodes if sys.version_info >= (3, 0): PY3 = True # In python 3, unicode -> str, and str -> bytes unicode = str else: PY3 = False ## External exceptions from ssl import SSLError as S3SSLError try: from ssl import CertificateError as S3SSLCertificateError except ImportError: class S3SSLCertificateError(Exception): pass try: from xml.etree.ElementTree import ParseError as XmlParseError except ImportError: # ParseError was only added in python2.7, before ET was raising ExpatError from xml.parsers.expat import ExpatError as XmlParseError ## s3cmd exceptions class S3Exception(Exception): def __init__(self, message=""): self.message = S3.Utils.unicodise(message) def __str__(self): ## Don't return self.message directly because ## __unicode__() method could be overridden in subclasses! if PY3: return self.__unicode__() else: return S3.Utils.deunicodise(self.__unicode__()) def __unicode__(self): return self.message ## (Base)Exception.message has been deprecated in Python 2.6 def _get_message(self): return self._message def _set_message(self, message): self._message = message message = property(_get_message, _set_message) class S3Error(S3Exception): def __init__(self, response): self.status = response["status"] self.reason = response["reason"] self.info = { "Code": "", "Message": "", "Resource": "" } debug("S3Error: %s (%s)" % (self.status, self.reason)) if "headers" in response: for header in response["headers"]: debug("HttpHeader: %s: %s" % (header, response["headers"][header])) if "data" in response and response["data"]: try: tree = S3.BaseUtils.getTreeFromXml(response["data"]) except XmlParseError: debug("Not an XML response") else: try: self.info.update(self.parse_error_xml(tree)) except Exception as e: error("Error parsing xml: %s. ErrorXML: %s" % (e, response["data"])) self.code = self.info["Code"] self.message = self.info["Message"] self.resource = self.info["Resource"] def __unicode__(self): retval = u"%d " % (self.status) retval += (u"(%s)" % (self.code or self.reason)) error_msg = self.message if error_msg: retval += (u": %s" % error_msg) return retval def get_error_code(self): if self.status in [301, 307]: return ExitCodes.EX_SERVERMOVED elif self.status in [400, 405, 411, 416, 417, 501, 504]: return ExitCodes.EX_SERVERERROR elif self.status == 403: return ExitCodes.EX_ACCESSDENIED elif self.status == 404: return ExitCodes.EX_NOTFOUND elif self.status == 409: return ExitCodes.EX_CONFLICT elif self.status == 412: return ExitCodes.EX_PRECONDITION elif self.status == 500: return ExitCodes.EX_SOFTWARE elif self.status in [429, 503]: return ExitCodes.EX_SERVICE else: return ExitCodes.EX_SOFTWARE @staticmethod def parse_error_xml(tree): info = {} error_node = tree if not error_node.tag == "Error": error_node = tree.find(".//Error") if error_node is not None: for child in error_node: if child.text != "": debug("ErrorXML: " + child.tag + ": " + repr(child.text)) info[child.tag] = child.text else: raise S3ResponseError("Malformed error XML returned from remote server.") return info class CloudFrontError(S3Error): pass class S3UploadError(S3Exception): pass class S3DownloadError(S3Exception): pass class S3RequestError(S3Exception): pass class S3ResponseError(S3Exception): pass class InvalidFileError(S3Exception): pass class ParameterError(S3Exception): pass # vim:et:ts=4:sts=4:ai	4.74 KB	2023-12-06 09:07
📄 ExitCodes.py # -- coding: utf-8 -- # patterned on /usr/include/sysexits.h EX_OK = 0 EX_GENERAL = 1 EX_PARTIAL = 2 # some parts of the command succeeded, while others failed EX_SERVERMOVED = 10 # 301: Moved permanently & 307: Moved temp EX_SERVERERROR = 11 # 400, 405, 411, 416, 417, 501: Bad request, 504: Gateway Time-out EX_NOTFOUND = 12 # 404: Not found EX_CONFLICT = 13 # 409: Conflict (ex: bucket error) EX_PRECONDITION = 14 # 412: Precondition failed EX_SERVICE = 15 # 503: Service not available or slow down EX_USAGE = 64 # The command was used incorrectly (e.g. bad command line syntax) EX_DATAERR = 65 # Failed file transfer, upload or download EX_SOFTWARE = 70 # internal software error (e.g. S3 error of unknown specificity) EX_OSERR = 71 # system error (e.g. out of memory) EX_OSFILE = 72 # OS error (e.g. invalid Python version) EX_IOERR = 74 # An error occurred while doing I/O on some file. EX_TEMPFAIL = 75 # temporary failure (S3DownloadError or similar, retry later) EX_ACCESSDENIED = 77 # Insufficient permissions to perform the operation on S3 EX_CONFIG = 78 # Configuration file error EX_CONNECTIONREFUSED = 111 # TCP connection refused (e.g. connecting to a closed server port) _EX_SIGNAL = 128 _EX_SIGINT = 2 EX_BREAK = _EX_SIGNAL + _EX_SIGINT # Control-C (KeyboardInterrupt raised) class ExitScoreboard(object): """Helper to return best return code""" def __init__(self): self._success = 0 self._notfound = 0 self._failed = 0 def success(self): self._success += 1 def notfound(self): self._notfound += 1 def failed(self): self._failed += 1 def rc(self): if self._success: if not self._failed and not self._notfound: return EX_OK elif self._failed: return EX_PARTIAL else: if self._failed: return EX_GENERAL else: if self._notfound: return EX_NOTFOUND return EX_GENERAL	2.2 KB	2023-12-06 09:07
📄 FileDict.py # -- coding: utf-8 -- ## -------------------------------------------------------------------- ## Amazon S3 manager ## ## Authors : Michal Ludvig <michal@logix.cz> (https://www.logix.cz/michal) ## Florent Viard <florent@sodria.com> (https://www.sodria.com) ## Copyright : TGRMN Software, Sodria SAS and contributors ## License : GPL Version 2 ## Website : https://s3tools.org ## -------------------------------------------------------------------- from __future__ import absolute_import import logging from .SortedDict import SortedDict from .Crypto import hash_file_md5 from . import Utils from . import Config zero_length_md5 = "d41d8cd98f00b204e9800998ecf8427e" cfg = Config.Config() class FileDict(SortedDict): def __init__(self, mapping = None, ignore_case = True, kwargs): SortedDict.__init__(self, mapping = mapping or {}, ignore_case = ignore_case, kwargs) self.hardlinks_md5 = dict() # { dev: { inode : {'md5':, 'relative_files':}}} self.by_md5 = dict() # {md5: set(relative_files)} def record_md5(self, relative_file, md5): if not relative_file: return if md5 is None: return if md5 == zero_length_md5: return if md5 not in self.by_md5: self.by_md5[md5] = relative_file def find_md5_one(self, md5): if not md5: return None return self.by_md5.get(md5, None) def get_md5(self, relative_file): """returns md5 if it can, or raises IOError if file is unreadable""" md5 = None if 'md5' in self[relative_file]: return self[relative_file]['md5'] md5 = self.get_hardlink_md5(relative_file) if md5 is None and 'md5' in cfg.sync_checks: logging.debug(u"doing file I/O to read md5 of %s" % relative_file) md5 = hash_file_md5(self[relative_file]['full_name']) self.record_md5(relative_file, md5) self[relative_file]['md5'] = md5 return md5 def record_hardlink(self, relative_file, dev, inode, md5, size): if md5 is None: return if size == 0: # don't record 0-length files return if dev == 0 or inode == 0: # Windows return if dev not in self.hardlinks_md5: self.hardlinks_md5[dev] = dict() if inode not in self.hardlinks_md5[dev]: self.hardlinks_md5[dev][inode] = md5 def get_hardlink_md5(self, relative_file): try: dev = self[relative_file]['dev'] inode = self[relative_file]['inode'] md5 = self.hardlinks_md5[dev][inode] except KeyError: md5 = None return md5	2.68 KB	2023-12-06 09:07
📄 FileLists.py # -- coding: utf-8 -- ## -------------------------------------------------------------------- ## Create and compare lists of files/objects ## ## Authors : Michal Ludvig <michal@logix.cz> (https://www.logix.cz/michal) ## Florent Viard <florent@sodria.com> (https://www.sodria.com) ## Copyright : TGRMN Software, Sodria SAS and contributors ## License : GPL Version 2 ## Website : https://s3tools.org ## -------------------------------------------------------------------- from __future__ import absolute_import from .S3 import S3 from .Config import Config from .S3Uri import S3Uri from .FileDict import FileDict from .BaseUtils import dateS3toUnix, dateRFC822toUnix, s3path from .Utils import unicodise, deunicodise, deunicodise_s, replace_nonprintables from .Exceptions import ParameterError from .HashCache import HashCache from logging import debug, info, warning import os import sys import glob import re import errno import io from stat import S_ISDIR PY3 = (sys.version_info >= (3, 0)) __all__ = ["fetch_local_list", "fetch_remote_list", "compare_filelists"] def _os_walk_unicode(top): ''' Reimplementation of python's os.walk to nicely support unicode in input as in output. ''' try: names = os.listdir(deunicodise(top)) except Exception: return dirs, nondirs = [], [] for name in names: name = unicodise(name) if os.path.isdir(deunicodise(os.path.join(top, name))): if not handle_exclude_include_walk_dir(top, name): dirs.append(name) else: nondirs.append(name) yield top, dirs, nondirs for name in dirs: new_path = os.path.join(top, name) if not os.path.islink(deunicodise(new_path)): for x in _os_walk_unicode(new_path): yield x def handle_exclude_include_walk_dir(root, dirname): ''' Should this root/dirname directory be excluded? (otherwise included by default) Exclude dir matches in the current directory This prevents us from recursing down trees we know we want to ignore return True for excluding, and False for including ''' cfg = Config() # python versions end their patterns (from globs) differently, test for different styles; check python3.6+ styles first directory_patterns = (u'/)$', u'/)\\Z', u'\\/$', u'\\/\\Z(?ms)') d = os.path.join(root, dirname, '') debug(u"CHECK: '%s'" % d) excluded = False for r in cfg.exclude: if not any(r.pattern.endswith(dp) for dp in directory_patterns): # we only check for directory patterns here continue if r.search(d): excluded = True debug(u"EXCL-MATCH: '%s'" % cfg.debug_exclude[r]) break if excluded: ## No need to check for --include if not excluded for r in cfg.include: if not any(r.pattern.endswith(dp) for dp in directory_patterns): # we only check for directory patterns here continue debug(u"INCL-TEST: '%s' ~ %s" % (d, r.pattern)) if r.search(d): excluded = False debug(u"INCL-MATCH: '%s'" % (cfg.debug_include[r])) break if excluded: ## Still excluded - ok, action it debug(u"EXCLUDE: '%s'" % d) else: debug(u"PASS: '%s'" % d) return excluded def _fswalk_follow_symlinks(path): ''' Walk filesystem, following symbolic links (but without recursion), on python2.4 and later If a symlink directory loop is detected, emit a warning and skip. E.g.: dir1/dir2/sym-dir -> ../dir2 ''' assert os.path.isdir(deunicodise(path)) # only designed for directory argument walkdirs = set([path]) for dirpath, dirnames, filenames in _os_walk_unicode(path): real_dirpath = unicodise(os.path.realpath(deunicodise(dirpath))) for dirname in dirnames: current = os.path.join(dirpath, dirname) real_current = unicodise(os.path.realpath(deunicodise(current))) if os.path.islink(deunicodise(current)): if (real_dirpath == real_current or real_dirpath.startswith(real_current + os.path.sep)): warning("Skipping recursively symlinked directory %s" % dirname) else: walkdirs.add(current) for walkdir in walkdirs: for dirpath, dirnames, filenames in _os_walk_unicode(walkdir): yield (dirpath, dirnames, filenames) def _fswalk_no_symlinks(path): ''' Directory tree generator path (str) is the root of the directory tree to walk ''' for dirpath, dirnames, filenames in _os_walk_unicode(path): yield (dirpath, dirnames, filenames) def filter_exclude_include(src_list): debug(u"Applying --exclude/--include") cfg = Config() exclude_list = FileDict(ignore_case = False) for file in src_list.keys(): debug(u"CHECK: '%s'" % file) excluded = False for r in cfg.exclude: if r.search(file): excluded = True debug(u"EXCL-MATCH: '%s'" % cfg.debug_exclude[r]) break if excluded: ## No need to check for --include if not excluded for r in cfg.include: if r.search(file): excluded = False debug(u"INCL-MATCH: '%s'" % cfg.debug_include[r]) break if excluded: ## Still excluded - ok, action it debug(u"EXCLUDE: '%s'" % file) exclude_list[file] = src_list[file] del(src_list[file]) continue else: debug(u"PASS: '%s'" % file) return src_list, exclude_list def _get_filelist_from_file(cfg, local_path): def _append(d, key, value): if key not in d: d[key] = [value] else: d[key].append(value) filelist = {} for fname in cfg.files_from: try: f = None if fname == u'-': f = io.open(sys.stdin.fileno(), mode='r', closefd=False) else: try: f = io.open(deunicodise(fname), mode='r') except IOError as e: warning(u"--files-from input file %s could not be opened for reading (%s), skipping." % (fname, e.strerror)) continue for line in f: line = unicodise(line).strip() line = os.path.normpath(os.path.join(local_path, line)) dirname = unicodise(os.path.dirname(deunicodise(line))) basename = unicodise(os.path.basename(deunicodise(line))) _append(filelist, dirname, basename) finally: if f: f.close() # reformat to match os.walk() result = [] for key in sorted(filelist): values = filelist[key] values.sort() result.append((key, [], values)) return result def fetch_local_list(args, is_src = False, recursive = None, with_dirs=False): def _fetch_local_list_info(loc_list): len_loc_list = len(loc_list) total_size = 0 info(u"Running stat() and reading/calculating MD5 values on %d files, this may take some time..." % len_loc_list) counter = 0 for relative_file in loc_list: counter += 1 if counter % 1000 == 0: info(u"[%d/%d]" % (counter, len_loc_list)) if relative_file == '-': continue loc_list_item = loc_list[relative_file] full_name = loc_list_item['full_name'] is_dir = loc_list_item['is_dir'] try: sr = os.stat_result(os.stat(deunicodise(full_name))) except OSError as e: if e.errno == errno.ENOENT: # file was removed async to us getting the list continue else: raise if is_dir: size = 0 else: size = sr.st_size loc_list[relative_file].update({ 'size' : size, 'mtime' : sr.st_mtime, 'dev' : sr.st_dev, 'inode' : sr.st_ino, 'uid' : sr.st_uid, 'gid' : sr.st_gid, 'sr': sr, # save it all, may need it in preserve_attrs_list ## TODO: Possibly more to save here... }) total_size += sr.st_size if is_dir: # A md5 can't be calculated with a directory path continue if 'md5' in cfg.sync_checks: md5 = cache.md5(sr.st_dev, sr.st_ino, sr.st_mtime, sr.st_size) if md5 is None: try: # this does the file I/O md5 = loc_list.get_md5(relative_file) except IOError: continue cache.add(sr.st_dev, sr.st_ino, sr.st_mtime, sr.st_size, md5) loc_list.record_hardlink(relative_file, sr.st_dev, sr.st_ino, md5, sr.st_size) return total_size def _get_filelist_local(loc_list, local_uri, cache, with_dirs): info(u"Compiling list of local files...") if local_uri.basename() == "-": try: uid = os.geteuid() gid = os.getegid() except Exception: uid = 0 gid = 0 loc_list["-"] = { 'full_name' : '-', 'size' : -1, 'mtime' : -1, 'uid' : uid, 'gid' : gid, 'dev' : 0, 'inode': 0, 'is_dir': False, } return loc_list, True if local_uri.isdir(): local_base = local_uri.basename() local_path = local_uri.path() if is_src and len(cfg.files_from): filelist = _get_filelist_from_file(cfg, local_path) single_file = False else: if cfg.follow_symlinks: filelist = _fswalk_follow_symlinks(local_path) else: filelist = _fswalk_no_symlinks(local_path) single_file = False else: local_base = "" local_path = local_uri.dirname() filelist = [( local_path, [], [local_uri.basename()] )] single_file = True for root, dirs, files in filelist: rel_root = root.replace(local_path, local_base, 1) if not with_dirs: iter_elements = ((files, False),) else: iter_elements = ((dirs, True), (files, False)) for elements, is_dir in iter_elements: for f in elements: full_name = os.path.join(root, f) if not is_dir and not os.path.isfile(deunicodise(full_name)): if os.path.exists(deunicodise(full_name)): warning(u"Skipping over non regular file: %s" % full_name) continue if os.path.islink(deunicodise(full_name)): if not cfg.follow_symlinks: warning(u"Skipping over symbolic link: %s" % full_name) continue relative_file = os.path.join(rel_root, f) if os.path.sep != "/": # Convert non-unix dir separators to '/' relative_file = "/".join(relative_file.split(os.path.sep)) if cfg.urlencoding_mode == "normal": relative_file = replace_nonprintables(relative_file) if relative_file.startswith('./'): relative_file = relative_file[2:] if is_dir and relative_file and relative_file[-1] != '/': relative_file += '/' loc_list[relative_file] = { 'full_name' : full_name, 'is_dir': is_dir, } return loc_list, single_file def _maintain_cache(cache, local_list): # if getting the file list from files_from, it is going to be # a subset of the actual tree. We should not purge content # outside of that subset as we don't know if it's valid or # not. Leave it to a non-files_from run to purge. if cfg.cache_file and len(cfg.files_from) == 0: cache.mark_all_for_purge() if PY3: local_list_val_iter = local_list.values() else: local_list_val_iter = local_list.itervalues() for f_info in local_list_val_iter: inode = f_info.get('inode', 0) if not inode: continue cache.unmark_for_purge(f_info['dev'], inode, f_info['mtime'], f_info['size']) cache.purge() cache.save(cfg.cache_file) cfg = Config() cache = HashCache() if cfg.cache_file and os.path.isfile(deunicodise_s(cfg.cache_file)) and os.path.getsize(deunicodise_s(cfg.cache_file)) > 0: cache.load(cfg.cache_file) else: info(u"Cache file not found or empty, creating/populating it.") local_uris = [] local_list = FileDict(ignore_case = False) single_file = False if type(args) not in (list, tuple, set): args = [args] if recursive == None: recursive = cfg.recursive for arg in args: uri = S3Uri(arg) if not uri.type == 'file': raise ParameterError("Expecting filename or directory instead of: %s" % arg) if uri.isdir() and not recursive: raise ParameterError("Use --recursive to upload a directory: %s" % arg) local_uris.append(uri) for uri in local_uris: list_for_uri, single_file = _get_filelist_local(local_list, uri, cache, with_dirs) ## Single file is True if and only if the user ## specified one local URI and that URI represents ## a FILE. Ie it is False if the URI was of a DIR ## and that dir contained only one FILE. That's not ## a case of single_file==True. if len(local_list) > 1: single_file = False local_list, exclude_list = filter_exclude_include(local_list) total_size = _fetch_local_list_info(local_list) _maintain_cache(cache, local_list) return local_list, single_file, exclude_list, total_size def fetch_remote_list(args, require_attribs = False, recursive = None, uri_params = {}): def _get_remote_attribs(uri, remote_item): response = S3(cfg).object_info(uri) if not response.get('headers'): return remote_item.update({ 'size': int(response['headers']['content-length']), 'md5': response['headers']['etag'].strip('"\''), 'timestamp': dateRFC822toUnix(response['headers']['last-modified']) }) try: md5 = response['s3cmd-attrs']['md5'] remote_item.update({'md5': md5}) debug(u"retrieved md5=%s from headers" % md5) except KeyError: pass def _get_filelist_remote(remote_uri, recursive = True): ## If remote_uri ends with '/' then all remote files will have ## the remote_uri prefix removed in the relative path. ## If, on the other hand, the remote_uri ends with something else ## (probably alphanumeric symbol) we'll use the last path part ## in the relative path. ## ## Complicated, eh? See an example: ## _get_filelist_remote("s3://bckt/abc/def") may yield: ## { 'def/file1.jpg' : {}, 'def/xyz/blah.txt' : {} } ## _get_filelist_remote("s3://bckt/abc/def/") will yield: ## { 'file1.jpg' : {}, 'xyz/blah.txt' : {} } ## Furthermore a prefix-magic can restrict the return list: ## _get_filelist_remote("s3://bckt/abc/def/x") yields: ## { 'xyz/blah.txt' : {} } info(u"Retrieving list of remote files for %s ..." % remote_uri) total_size = 0 s3 = S3(Config()) response = s3.bucket_list(remote_uri.bucket(), prefix = remote_uri.object(), recursive = recursive, uri_params = uri_params) rem_base_original = rem_base = remote_uri.object() remote_uri_original = remote_uri if rem_base != '' and rem_base[-1] != '/': rem_base = rem_base[:rem_base.rfind('/')+1] remote_uri = S3Uri(u"s3://%s/%s" % (remote_uri.bucket(), rem_base)) rem_base_len = len(rem_base) rem_list = FileDict(ignore_case = False) break_now = False for object in response['list']: object_key = object['Key'] object_size = int(object['Size']) is_dir = (object_key[-1] == '/') if object_key == rem_base_original and not is_dir: ## We asked for one file and we got that file :-) key = s3path.basename(object_key) object_uri_str = remote_uri_original.uri() break_now = True # Remove whatever has already been put to rem_list rem_list = FileDict(ignore_case = False) else: # Beware - this may be '' if object_key==rem_base !! key = object_key[rem_base_len:] object_uri_str = remote_uri.uri() + key if not key: # Objects may exist on S3 with empty names (''), which don't map so well to common filesystems. warning(u"Found empty root object name on S3, ignoring.") continue rem_list[key] = { 'size' : object_size, 'timestamp' : dateS3toUnix(object['LastModified']), ## Sadly it's upload time, not our lastmod time :-( 'md5' : object['ETag'].strip('"\''), 'object_key' : object_key, 'object_uri_str' : object_uri_str, 'base_uri' : remote_uri, 'dev' : None, 'inode' : None, 'is_dir': is_dir, } if '-' in rem_list[key]['md5']: # always get it for multipart uploads _get_remote_attribs(S3Uri(object_uri_str), rem_list[key]) md5 = rem_list[key]['md5'] rem_list.record_md5(key, md5) total_size += object_size if break_now: break return rem_list, total_size cfg = Config() remote_uris = [] remote_list = FileDict(ignore_case = False) if type(args) not in (list, tuple, set): args = [args] if recursive == None: recursive = cfg.recursive for arg in args: uri = S3Uri(arg) if not uri.type == 's3': raise ParameterError("Expecting S3 URI instead of '%s'" % arg) remote_uris.append(uri) total_size = 0 if recursive: for uri in remote_uris: objectlist, tmp_total_size = _get_filelist_remote(uri, recursive = True) total_size += tmp_total_size for key in objectlist: remote_list[key] = objectlist[key] remote_list.record_md5(key, objectlist.get_md5(key)) else: for uri in remote_uris: uri_str = uri.uri() ## Wildcards used in remote URI? ## If yes we'll need a bucket listing... wildcard_split_result = re.split("\*\|\?", uri_str, maxsplit=1) if len(wildcard_split_result) == 2: ## If wildcards found prefix, rest = wildcard_split_result ## Only request recursive listing if the 'rest' of the URI, ## i.e. the part after first wildcard, contains '/' need_recursion = '/' in rest objectlist, tmp_total_size = _get_filelist_remote(S3Uri(prefix), recursive = need_recursion) total_size += tmp_total_size for key in objectlist: ## Check whether the 'key' matches the requested wildcards if glob.fnmatch.fnmatch(objectlist[key]['object_uri_str'], uri_str): remote_list[key] = objectlist[key] else: ## No wildcards - simply append the given URI to the list key = s3path.basename(uri.object()) if not key: raise ParameterError(u"Expecting S3 URI with a filename or --recursive: %s" % uri.uri()) is_dir = (key and key[-1] == '/') remote_item = { 'base_uri': uri, 'object_uri_str': uri.uri(), 'object_key': uri.object(), 'is_dir': is_dir, } if require_attribs: _get_remote_attribs(uri, remote_item) remote_list[key] = remote_item md5 = remote_item.get('md5') if md5: remote_list.record_md5(key, md5) total_size += remote_item.get('size', 0) remote_list, exclude_list = filter_exclude_include(remote_list) return remote_list, exclude_list, total_size def compare_filelists(src_list, dst_list, src_remote, dst_remote): def __direction_str(is_remote): return is_remote and "remote" or "local" def _compare(src_list, dst_lst, src_remote, dst_remote, file): """Return True if src_list[file] matches dst_list[file], else False""" attribs_match = True src_file = src_list.get(file) dst_file = dst_list.get(file) if not src_file or not dst_file: info(u"%s: does not exist in one side or the other: src_list=%s, dst_list=%s" % (file, bool(src_file), bool(dst_file))) return False ## check size first if 'size' in cfg.sync_checks: src_size = src_file.get('size') dst_size = dst_file.get('size') if dst_size is not None and src_size is not None and dst_size != src_size: debug(u"xfer: %s (size mismatch: src=%s dst=%s)" % (file, src_size, dst_size)) attribs_match = False ## check md5 compare_md5 = 'md5' in cfg.sync_checks # Multipart-uploaded files don't have a valid md5 sum - it ends with "...-nn" if compare_md5: if (src_remote == True and '-' in src_file['md5']) or (dst_remote == True and '-' in dst_file['md5']): compare_md5 = False info(u"disabled md5 check for %s" % file) if compare_md5 and src_file['is_dir'] == True: # For directories, nothing to do if they already exist compare_md5 = False if attribs_match and compare_md5: try: src_md5 = src_list.get_md5(file) dst_md5 = dst_list.get_md5(file) except (IOError, OSError): # md5 sum verification failed - ignore that file altogether debug(u"IGNR: %s (disappeared)" % (file)) warning(u"%s: file disappeared, ignoring." % (file)) raise if src_md5 != dst_md5: ## checksums are different. attribs_match = False debug(u"XFER: %s (md5 mismatch: src=%s dst=%s)" % (file, src_md5, dst_md5)) return attribs_match # we don't support local->local sync, use 'rsync' or something like that instead ;-) assert(not(src_remote == False and dst_remote == False)) info(u"Verifying attributes...") cfg = Config() ## Items left on src_list will be transferred ## Items left on update_list will be transferred after src_list ## Items left on copy_pairs will be copied from dst1 to dst2 update_list = FileDict(ignore_case = False) ## Items left on dst_list will be deleted copy_pairs = {} debug("Comparing filelists (direction: %s -> %s)" % (__direction_str(src_remote), __direction_str(dst_remote))) src_dir_cache = set() for relative_file in src_list.keys(): debug(u"CHECK: '%s'" % relative_file) if src_remote: # Most of the time, there will not be dir objects on the remote side # we still need to have a "virtual" list of them to not think that there # are unmatched dirs with the local side. dir_idx = relative_file.rfind('/') if dir_idx > 0: path = relative_file[:dir_idx+1] while path and path not in src_dir_cache: src_dir_cache.add(path) # Also add to cache, all the parent dirs try: path = path[:path.rindex('/', 0, -1)+1] except ValueError: continue if relative_file in dst_list: ## Was --skip-existing requested? if cfg.skip_existing: debug(u"IGNR: '%s' (used --skip-existing)" % relative_file) del(src_list[relative_file]) del(dst_list[relative_file]) continue try: same_file = _compare(src_list, dst_list, src_remote, dst_remote, relative_file) except (IOError,OSError): debug(u"IGNR: '%s' (disappeared)" % relative_file) warning(u"%s: file disappeared, ignoring." % relative_file) del(src_list[relative_file]) del(dst_list[relative_file]) continue if same_file: debug(u"IGNR: '%s' (transfer not needed)" % relative_file) del(src_list[relative_file]) del(dst_list[relative_file]) else: # look for matching file in src try: md5 = src_list.get_md5(relative_file) except IOError: md5 = None if md5 is not None and md5 in dst_list.by_md5: # Found one, we want to copy copy_src_file = dst_list.find_md5_one(md5) debug(u"DST COPY src: '%s' -> '%s'" % (copy_src_file, relative_file)) src_item = src_list[relative_file] src_item["md5"] = md5 src_item["copy_src"] = copy_src_file copy_pairs[relative_file] = src_item del(src_list[relative_file]) del(dst_list[relative_file]) else: # record that we will get this file transferred to us (before all the copies), so if we come across it later again, # we can copy from _this_ copy (e.g. we only upload it once, and copy thereafter). dst_list.record_md5(relative_file, md5) update_list[relative_file] = src_list[relative_file] del src_list[relative_file] del dst_list[relative_file] else: # dst doesn't have this file # look for matching file elsewhere in dst try: md5 = src_list.get_md5(relative_file) except IOError: md5 = None copy_src_file = dst_list.find_md5_one(md5) if copy_src_file is not None: # Found one, we want to copy debug(u"DST COPY dst: '%s' -> '%s'" % (copy_src_file, relative_file)) src_item = src_list[relative_file] src_item["md5"] = md5 src_item["copy_src"] = copy_src_file copy_pairs[relative_file] = src_item del(src_list[relative_file]) else: # we don't have this file, and we don't have a copy of this file elsewhere. Get it. # record that we will get this file transferred to us (before all the copies), so if we come across it later again, # we can copy from _this_ copy (e.g. we only upload it once, and copy thereafter). dst_list.record_md5(relative_file, md5) for f in dst_list.keys(): if f in src_list or f in update_list or f in src_dir_cache: # leave only those not on src_list + update_list + src_dir_cache del dst_list[f] return src_list, dst_list, update_list, copy_pairs # vim:et:ts=4:sts=4:ai	28.15 KB	2023-12-06 09:07
📄 HashCache.py # -- coding: utf-8 -- from __future__ import absolute_import try: # python 3 support import cPickle as pickle except ImportError: import pickle from .Utils import deunicodise class HashCache(object): def __init__(self): self.inodes = dict() def add(self, dev, inode, mtime, size, md5): if dev == 0 or inode == 0: return # Windows if dev not in self.inodes: self.inodes[dev] = dict() if inode not in self.inodes[dev]: self.inodes[dev][inode] = dict() self.inodes[dev][inode][mtime] = dict(md5=md5, size=size) def md5(self, dev, inode, mtime, size): try: d = self.inodes[dev][inode][mtime] if d['size'] != size: return None except Exception: return None return d['md5'] def mark_all_for_purge(self): for d in tuple(self.inodes): for i in tuple(self.inodes[d]): for c in tuple(self.inodes[d][i]): self.inodes[d][i][c]['purge'] = True def unmark_for_purge(self, dev, inode, mtime, size): try: d = self.inodes[dev][inode][mtime] except KeyError: return if d['size'] == size and 'purge' in d: del self.inodes[dev][inode][mtime]['purge'] def purge(self): for d in tuple(self.inodes): for i in tuple(self.inodes[d]): for m in tuple(self.inodes[d][i]): if 'purge' in self.inodes[d][i][m]: del self.inodes[d][i] break def save(self, f): d = dict(inodes=self.inodes, version=1) with open(deunicodise(f), 'wb') as fp: pickle.dump(d, fp) def load(self, f): with open(deunicodise(f), 'rb') as fp: d = pickle.load(fp) if d.get('version') == 1 and 'inodes' in d: self.inodes = d['inodes']	1.91 KB	2023-12-06 09:07
📄 MultiPart.py # -- coding: utf-8 -- ## Amazon S3 Multipart upload support ## Author: Jerome Leclanche <jerome.leclanche@gmail.com> ## License: GPL Version 2 from __future__ import absolute_import import sys from logging import debug, info, warning, error from .Crypto import calculateChecksum from .Exceptions import ParameterError from .S3Uri import S3UriS3 from .BaseUtils import getTextFromXml, getTreeFromXml, s3_quote, parseNodes from .Utils import formatSize SIZE_1MB = 1024 * 1024 class MultiPartUpload(object): """Supports MultiPartUpload and MultiPartUpload(Copy) operation""" MIN_CHUNK_SIZE_MB = 5 # 5MB MAX_CHUNK_SIZE_MB = 5 * 1024 # 5GB MAX_FILE_SIZE = 5 * 1024 * 1024 # 5TB def __init__(self, s3, src, dst_uri, headers_baseline=None, src_size=None): self.s3 = s3 self.file_stream = None self.src_uri = None self.src_size = src_size self.dst_uri = dst_uri self.parts = {} self.headers_baseline = headers_baseline or {} if isinstance(src, S3UriS3): # Source is the uri of an object to s3-to-s3 copy with multipart. self.src_uri = src if not src_size: raise ParameterError("Source size is missing for " "MultipartUploadCopy operation") c_size = self.s3.config.multipart_copy_chunk_size_mb * SIZE_1MB else: # Source is a file_stream to upload self.file_stream = src c_size = self.s3.config.multipart_chunk_size_mb * SIZE_1MB self.chunk_size = c_size self.upload_id = self.initiate_multipart_upload() def get_parts_information(self, uri, upload_id): part_list = self.s3.list_multipart(uri, upload_id) parts = dict() for elem in part_list: try: parts[int(elem['PartNumber'])] = { 'checksum': elem['ETag'], 'size': elem['Size'] } except KeyError: pass return parts def get_unique_upload_id(self, uri): upload_id = "" multipart_list = self.s3.get_multipart(uri) for mpupload in multipart_list: try: mp_upload_id = mpupload['UploadId'] mp_path = mpupload['Key'] info("mp_path: %s, object: %s" % (mp_path, uri.object())) if mp_path == uri.object(): if upload_id: raise ValueError( "More than one UploadId for URI %s. Disable " "multipart upload, or use\n %s multipart %s\n" "to list the Ids, then pass a unique --upload-id " "into the put command." % (uri, sys.argv[0], uri)) upload_id = mp_upload_id except KeyError: pass return upload_id def initiate_multipart_upload(self): """ Begin a multipart upload http://docs.amazonwebservices.com/AmazonS3/latest/API/index.html?mpUploadInitiate.html """ if self.s3.config.upload_id: self.upload_id = self.s3.config.upload_id elif self.s3.config.put_continue: self.upload_id = self.get_unique_upload_id(self.dst_uri) else: self.upload_id = "" if not self.upload_id: request = self.s3.create_request("OBJECT_POST", uri=self.dst_uri, headers=self.headers_baseline, uri_params={'uploads': None}) response = self.s3.send_request(request) data = response["data"] self.upload_id = getTextFromXml(data, "UploadId") return self.upload_id def upload_all_parts(self, extra_label=''): """ Execute a full multipart upload on a file Returns the seq/etag dict TODO use num_processes to thread it """ if not self.upload_id: raise ParameterError("Attempting to use a multipart upload that " "has not been initiated.") remote_statuses = {} if self.src_uri: filename = self.src_uri.uri() # Continue is not possible with multipart copy else: filename = self.file_stream.stream_name if self.s3.config.put_continue: remote_statuses = self.get_parts_information(self.dst_uri, self.upload_id) if extra_label: extra_label = u' ' + extra_label labels = { 'source': filename, 'destination': self.dst_uri.uri(), } seq = 1 if self.src_size: size_left = self.src_size nr_parts = self.src_size // self.chunk_size \ + (self.src_size % self.chunk_size and 1) debug("MultiPart: Uploading %s in %d parts" % (filename, nr_parts)) while size_left > 0: offset = self.chunk_size * (seq - 1) current_chunk_size = min(self.src_size - offset, self.chunk_size) size_left -= current_chunk_size labels['extra'] = "[part %d of %d, %s]%s" % ( seq, nr_parts, "%d%sB" % formatSize(current_chunk_size, human_readable=True), extra_label) try: if self.file_stream: self.upload_part( seq, offset, current_chunk_size, labels, remote_status=remote_statuses.get(seq)) else: self.copy_part( seq, offset, current_chunk_size, labels, remote_status=remote_statuses.get(seq)) except: error(u"\nUpload of '%s' part %d failed. Use\n " "%s abortmp %s %s\nto abort the upload, or\n " "%s --upload-id %s put ...\nto continue the upload." % (filename, seq, sys.argv[0], self.dst_uri, self.upload_id, sys.argv[0], self.upload_id)) raise seq += 1 debug("MultiPart: Upload finished: %d parts", seq - 1) return # Else -> Case of u"<stdin>" source debug("MultiPart: Uploading from %s" % filename) while True: buffer = self.file_stream.read(self.chunk_size) offset = 0 # send from start of the buffer current_chunk_size = len(buffer) labels['extra'] = "[part %d of -, %s]%s" % ( seq, "%d%sB" % formatSize(current_chunk_size, human_readable=True), extra_label) if not buffer: # EOF break try: self.upload_part(seq, offset, current_chunk_size, labels, buffer, remote_status=remote_statuses.get(seq)) except: error(u"\nUpload of '%s' part %d failed. Use\n " "%s abortmp %s %s\nto abort, or\n " "%s --upload-id %s put ...\nto continue the upload." % (filename, seq, sys.argv[0], self.dst_uri, self.upload_id, sys.argv[0], self.upload_id)) raise seq += 1 debug("MultiPart: Upload finished: %d parts", seq - 1) def upload_part(self, seq, offset, chunk_size, labels, buffer='', remote_status=None): """ Upload a file chunk http://docs.amazonwebservices.com/AmazonS3/latest/API/index.html?mpUploadUploadPart.html """ # TODO implement Content-MD5 debug("Uploading part %i of %r (%s bytes)" % (seq, self.upload_id, chunk_size)) if remote_status is not None: if int(remote_status['size']) == chunk_size: checksum = calculateChecksum(buffer, self.file_stream, offset, chunk_size, self.s3.config.send_chunk) remote_checksum = remote_status['checksum'].strip('"\'') if remote_checksum == checksum: warning("MultiPart: size and md5sum match for %s part %d, " "skipping." % (self.dst_uri, seq)) self.parts[seq] = remote_status['checksum'] return None else: warning("MultiPart: checksum (%s vs %s) does not match for" " %s part %d, reuploading." % (remote_checksum, checksum, self.dst_uri, seq)) else: warning("MultiPart: size (%d vs %d) does not match for %s part" " %d, reuploading." % (int(remote_status['size']), chunk_size, self.dst_uri, seq)) headers = {"content-length": str(chunk_size)} query_string_params = {'partNumber': '%s' % seq, 'uploadId': self.upload_id} request = self.s3.create_request("OBJECT_PUT", uri=self.dst_uri, headers=headers, uri_params=query_string_params) response = self.s3.send_file(request, self.file_stream, labels, buffer, offset=offset, chunk_size=chunk_size) self.parts[seq] = response["headers"].get('etag', '').strip('"\'') return response def copy_part(self, seq, offset, chunk_size, labels, remote_status=None): """ Copy a remote file chunk http://docs.amazonwebservices.com/AmazonS3/latest/API/index.html?mpUploadUploadPart.html http://docs.amazonwebservices.com/AmazonS3/latest/API/mpUploadUploadPartCopy.html """ debug("Copying part %i of %r (%s bytes)" % (seq, self.upload_id, chunk_size)) # set up headers with copy-params. # Examples: # x-amz-copy-source: /source_bucket/sourceObject # x-amz-copy-source-range:bytes=first-last # x-amz-copy-source-if-match: etag # x-amz-copy-source-if-none-match: etag # x-amz-copy-source-if-unmodified-since: time_stamp # x-amz-copy-source-if-modified-since: time_stamp headers = { "x-amz-copy-source": s3_quote("/%s/%s" % (self.src_uri.bucket(), self.src_uri.object()), quote_backslashes=False, unicode_output=True) } # byte range, with end byte included. A 10 byte file has bytes=0-9 headers["x-amz-copy-source-range"] = \ "bytes=%d-%d" % (offset, (offset + chunk_size - 1)) query_string_params = {'partNumber': '%s' % seq, 'uploadId': self.upload_id} request = self.s3.create_request("OBJECT_PUT", uri=self.dst_uri, headers=headers, uri_params=query_string_params) labels[u'action'] = u'remote copy' response = self.s3.send_request_with_progress(request, labels, chunk_size) # NOTE: Amazon sends whitespace while upload progresses, which # accumulates in response body and seems to confuse XML parser. # Strip newlines to find ETag in XML response data #data = response["data"].replace("\n", '') self.parts[seq] = getTextFromXml(response['data'], "ETag") or '' return response def complete_multipart_upload(self): """ Finish a multipart upload http://docs.amazonwebservices.com/AmazonS3/latest/API/index.html?mpUploadComplete.html """ debug("MultiPart: Completing upload: %s" % self.upload_id) parts_xml = [] part_xml = "<Part><PartNumber>%i</PartNumber><ETag>%s</ETag></Part>" for seq, etag in self.parts.items(): parts_xml.append(part_xml % (seq, etag)) body = "<CompleteMultipartUpload>%s</CompleteMultipartUpload>" \ % "".join(parts_xml) headers = {"content-length": str(len(body))} request = self.s3.create_request( "OBJECT_POST", uri=self.dst_uri, headers=headers, body=body, uri_params={'uploadId': self.upload_id}) response = self.s3.send_request(request) return response def abort_upload(self): """ Abort multipart upload http://docs.amazonwebservices.com/AmazonS3/latest/API/index.html?mpUploadAbort.html """ debug("MultiPart: Aborting upload: %s" % self.upload_id) #request = self.s3.create_request("OBJECT_DELETE", uri = self.uri, # uri_params = {'uploadId': self.upload_id}) #response = self.s3.send_request(request) response = None return response # vim:et:ts=4:sts=4:ai	13.33 KB	2023-12-06 09:07
📄 PkgInfo.py # -- coding: utf-8 -- ## -------------------------------------------------------------------- ## Amazon S3 manager ## ## Authors : Michal Ludvig <michal@logix.cz> (https://www.logix.cz/michal) ## Florent Viard <florent@sodria.com> (https://www.sodria.com) ## Copyright : TGRMN Software, Sodria SAS and contributors ## License : GPL Version 2 ## Website : https://s3tools.org ## -------------------------------------------------------------------- package = "s3cmd" version = "2.4.0" url = "http://s3tools.org" license = "GNU GPL v2+" short_description = "Command line tool for managing Amazon S3 and CloudFront services" long_description = """ S3cmd lets you copy files from/to Amazon S3 (Simple Storage Service) using a simple to use command line client. Supports rsync-like backup, GPG encryption, and more. Also supports management of Amazon's CloudFront content delivery network. """ # vim:et:ts=4:sts=4:ai	934 B	2023-12-12 01:08
📄 Progress.py # -- coding: utf-8 -- ## -------------------------------------------------------------------- ## Amazon S3 manager ## ## Authors : Michal Ludvig <michal@logix.cz> (https://www.logix.cz/michal) ## Florent Viard <florent@sodria.com> (https://www.sodria.com) ## Copyright : TGRMN Software, Sodria SAS and contributors ## License : GPL Version 2 ## Website : https://s3tools.org ## -------------------------------------------------------------------- from __future__ import absolute_import, division import sys import datetime import time import S3.Utils class Progress(object): _stdout = sys.stdout _last_display = 0 def __init__(self, labels, total_size): self._stdout = sys.stdout self.new_file(labels, total_size) def new_file(self, labels, total_size): self.labels = labels self.total_size = total_size # Set initial_position to something in the # case we're not counting from 0. For instance # when appending to a partially downloaded file. # Setting initial_position will let the speed # be computed right. self.initial_position = 0 self.current_position = self.initial_position self.time_start = datetime.datetime.now() self.time_last = self.time_start self.time_current = self.time_start self.display(new_file = True) def update(self, current_position = -1, delta_position = -1): self.time_last = self.time_current self.time_current = datetime.datetime.now() if current_position > -1: self.current_position = current_position elif delta_position > -1: self.current_position += delta_position #else: # no update, just call display() self.display() def done(self, message): self.display(done_message = message) def output_labels(self): self._stdout.write(u"%(action)s: '%(source)s' -> '%(destination)s' %(extra)s\n" % self.labels) self._stdout.flush() def _display_needed(self): # We only need to update the display every so often. if time.time() - self._last_display > 1: self._last_display = time.time() return True return False def display(self, new_file = False, done_message = None): """ display(new_file = False[/True], done = False[/True]) Override this method to provide a nicer output. """ if new_file: self.output_labels() self.last_milestone = 0 return if self.current_position == self.total_size: print_size = S3.Utils.formatSize(self.current_position, True) if print_size[1] != "": print_size[1] += "B" timedelta = self.time_current - self.time_start sec_elapsed = timedelta.days * 86400 + timedelta.seconds + float(timedelta.microseconds) / 1000000.0 print_speed = S3.Utils.formatSize((self.current_position - self.initial_position) / sec_elapsed, True, True) self._stdout.write("100%% %s%s in %.2fs (%.2f %sB/s)\n" % (print_size[0], print_size[1], sec_elapsed, print_speed[0], print_speed[1])) self._stdout.flush() return rel_position = (self.current_position * 100) // self.total_size if rel_position >= self.last_milestone: # Move by increments of 5. # NOTE: to check: Looks like to not do what is looks like to be designed to do self.last_milestone = (rel_position // 5) * 5 self._stdout.write("%d%% ", self.last_milestone) self._stdout.flush() return class ProgressANSI(Progress): ## http://en.wikipedia.org/wiki/ANSI_escape_code SCI = '\x1b[' ANSI_hide_cursor = SCI + "?25l" ANSI_show_cursor = SCI + "?25h" ANSI_save_cursor_pos = SCI + "s" ANSI_restore_cursor_pos = SCI + "u" ANSI_move_cursor_to_column = SCI + "%uG" ANSI_erase_to_eol = SCI + "0K" ANSI_erase_current_line = SCI + "2K" def display(self, new_file = False, done_message = None): """ display(new_file = False[/True], done_message = None) """ if new_file: self.output_labels() self._stdout.write(self.ANSI_save_cursor_pos) self._stdout.flush() return # Only display progress every so often if not (new_file or done_message) and not self._display_needed(): return timedelta = self.time_current - self.time_start sec_elapsed = timedelta.days * 86400 + timedelta.seconds + float(timedelta.microseconds)/1000000.0 if (sec_elapsed > 0): print_speed = S3.Utils.formatSize((self.current_position - self.initial_position) / sec_elapsed, True, True) else: print_speed = (0, "") self._stdout.write(self.ANSI_restore_cursor_pos) self._stdout.write(self.ANSI_erase_to_eol) self._stdout.write("%(current)s of %(total)s %(percent)3d%% in %(elapsed)ds %(speed).2f %(speed_coeff)sB/s" % { "current" : str(self.current_position).rjust(len(str(self.total_size))), "total" : self.total_size, "percent" : self.total_size and ((self.current_position * 100) // self.total_size) or 0, "elapsed" : sec_elapsed, "speed" : print_speed[0], "speed_coeff" : print_speed[1] }) if done_message: self._stdout.write(" %s\n" % done_message) self._stdout.flush() class ProgressCR(Progress): ## Uses CR char (Carriage Return) just like other progress bars do. CR_char = chr(13) def display(self, new_file = False, done_message = None): """ display(new_file = False[/True], done_message = None) """ if new_file: self.output_labels() return # Only display progress every so often if not (new_file or done_message) and not self._display_needed(): return timedelta = self.time_current - self.time_start sec_elapsed = timedelta.days * 86400 + timedelta.seconds + float(timedelta.microseconds)/1000000.0 if (sec_elapsed > 0): print_speed = S3.Utils.formatSize((self.current_position - self.initial_position) / sec_elapsed, True, True) else: print_speed = (0, "") self._stdout.write(self.CR_char) output = " %(current)s of %(total)s %(percent)3d%% in %(elapsed)4ds %(speed)7.2f %(speed_coeff)sB/s" % { "current" : str(self.current_position).rjust(len(str(self.total_size))), "total" : self.total_size, "percent" : self.total_size and ((self.current_position * 100) // self.total_size) or 0, "elapsed" : sec_elapsed, "speed" : print_speed[0], "speed_coeff" : print_speed[1] } self._stdout.write(output) if done_message: self._stdout.write(" %s\n" % done_message) self._stdout.flush() class StatsInfo(object): """Holding info for stats totals""" def __init__(self): self.files = None self.size = None self.files_transferred = None self.size_transferred = None self.files_copied = None self.size_copied = None self.files_deleted = None self.size_deleted = None def format_output(self): outstr = u"" if self.files is not None: tmp_str = u"Number of files: %d"% self.files if self.size is not None: tmp_str += " (%d bytes) "% self.size outstr += u"\nStats: " + tmp_str if self.files_transferred: tmp_str = u"Number of files transferred: %d"% self.files_transferred if self.size_transferred is not None: tmp_str += " (%d bytes) "% self.size_transferred outstr += u"\nStats: " + tmp_str if self.files_copied: tmp_str = u"Number of files copied: %d"% self.files_copied if self.size_copied is not None: tmp_str += " (%d bytes) "% self.size_copied outstr += u"\nStats: " + tmp_str if self.files_deleted: tmp_str = u"Number of files deleted: %d"% self.files_deleted if self.size_deleted is not None: tmp_str += " (%d bytes) "% self.size_deleted outstr += u"\nStats: " + tmp_str return outstr # vim:et:ts=4:sts=4:ai	8.35 KB	2023-12-06 09:07
📄 S3.py # -- coding: utf-8 -- ## -------------------------------------------------------------------- ## Amazon S3 manager ## ## Authors : Michal Ludvig <michal@logix.cz> (https://www.logix.cz/michal) ## Florent Viard <florent@sodria.com> (https://www.sodria.com) ## Copyright : TGRMN Software, Sodria SAS and contributors ## License : GPL Version 2 ## Website : https://s3tools.org ## -------------------------------------------------------------------- from __future__ import absolute_import, division import sys import os import time import errno 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, ST_MODE, S_ISDIR, S_ISREG try: # python 3 support from urlparse import urlparse except ImportError: from urllib.parse import urlparse import select from .BaseUtils import (getListFromXml, getTextFromXml, getRootTagName, decode_from_s3, encode_to_s3, md5, s3_quote) from .Utils import (convertHeaderTupleListToDict, unicodise, deunicodise, check_bucket_name, check_bucket_name_dns_support, getHostnameFromBucket) 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, generate_content_md5, hash_file_md5, calculateChecksum, 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( UNDEFINED = 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", } 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 <dir_list>, <object_list> pairs of groups of content of a specified bucket. """ def _list_truncated(data): ## <IsTruncated> 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_elts, key): return getTextFromXml(response["data"], "NextMarker") or current_elts[-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, "Key") elif current_prefixes: uri_params['marker'] = \ _get_next_marker(response["data"], current_prefixes, "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) if self.config.list_allow_unordered: uri_params['allow-unordered'] = "true" 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, extra_headers = None): headers = SortedDict(ignore_case = True) if extra_headers: headers.update(extra_headers) 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 = "<CreateBucketConfiguration><LocationConstraint>" body += bucket_location body += "</LocationConstraint></CreateBucketConfiguration>" 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" # AWS suddenly changed the default "ownership" control value mid 2023. # ACL are disabled by default, so obviously the bucket can't be public. # See: https://aws.amazon.com/fr/blogs/aws/heads-up-amazon-s3-security-changes-are-coming-in-april-of-2023/ # To be noted: "Block Public Access" flags should also be disabled after the bucket creation to be able to set a "public" acl for an object. headers["x-amz-object-ownership"] = 'ObjectWriter' 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(resp_data, "Payer") else: payer = None return payer def set_bucket_ownership(self, uri, ownership): headers = SortedDict(ignore_case=True) body = '<OwnershipControls xmlns="http://s3.amazonaws.com/doc/2006-03-01/">' \ '<Rule>' \ '<ObjectOwnership>%s</ObjectOwnership>' \ '</Rule>' \ '</OwnershipControls>' body = body % ownership debug(u"set_bucket_ownership(%s)" % body) headers['content-md5'] = generate_content_md5(body) request = self.create_request("BUCKET_CREATE", uri = uri, headers = headers, body = body, uri_params = {'ownershipControls': None}) response = self.send_request(request) return response def get_bucket_ownership(self, uri): request = self.create_request("BUCKET_LIST", bucket=uri.bucket(), uri_params={'ownershipControls': None}) response = self.send_request(request) resp_data = response.get('data', '') if resp_data: ownership = getTextFromXml(resp_data, ".//Rule//ObjectOwnership") else: ownership = None return ownership def set_bucket_public_access_block(self, uri, flags): headers = SortedDict(ignore_case=True) body = '<PublicAccessBlockConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">' for tag in ('BlockPublicAcls', 'IgnorePublicAcls', 'BlockPublicPolicy', 'RestrictPublicBuckets'): val = flags.get(tag, False) and "true" or "false" body += '<%s>%s</%s>' % (tag, val, tag) body += '</PublicAccessBlockConfiguration>' debug(u"set_bucket_public_access_block(%s)" % body) headers['content-md5'] = generate_content_md5(body) request = self.create_request("BUCKET_CREATE", uri = uri, headers = headers, body = body, uri_params = {'publicAccessBlock': None}) response = self.send_request(request) return response def get_bucket_public_access_block(self, uri): request = self.create_request("BUCKET_LIST", bucket=uri.bucket(), uri_params={'publicAccessBlock': None}) response = self.send_request(request) resp_data = response.get('data', '') if resp_data: flags = { "BlockPublicAcls": getTextFromXml(resp_data, "BlockPublicAcls") == "true", "IgnorePublicAcls": getTextFromXml(resp_data, "IgnorePublicAcls") == "true", "BlockPublicPolicy": getTextFromXml(resp_data, "BlockPublicPolicy") == "true", "RestrictPublicBuckets": getTextFromXml(resp_data, "RestrictPublicBuckets") == "true", } else: flags = {} return flags def bucket_info(self, uri): response = {} response['bucket-location'] = self.get_bucket_location(uri) for key, func in (('requester-pays', self.get_bucket_requester_pays), ('versioning', self.get_versioning), ('ownership', self.get_bucket_ownership)): try: response[key] = func(uri) except S3Error as e: response[key] = None try: response['public-access-block'] = self.get_bucket_public_access_block(uri) except S3Error as e: response['public-access-block'] = {} 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 = '<WebsiteConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">' body += ' <IndexDocument>' body += (' <Suffix>%s</Suffix>' % self.config.website_index) body += ' </IndexDocument>' if self.config.website_error: body += ' <ErrorDocument>' body += (' <Key>%s</Key>' % self.config.website_error) body += ' </ErrorDocument>' body += '</WebsiteConfiguration>' 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 = '<LifecycleConfiguration>' body += ' <Rule>' body += ' <Filter>' body += ' <Prefix>%s</Prefix>' % self.config.expiry_prefix body += ' </Filter>' body += ' <Status>Enabled</Status>' body += ' <Expiration>' if self.config.expiry_date: body += ' <Date>%s</Date>' % self.config.expiry_date elif self.config.expiry_days: body += ' <Days>%s</Days>' % self.config.expiry_days body += ' </Expiration>' body += ' </Rule>' body += '</LifecycleConfiguration>' headers = SortedDict(ignore_case = True) headers['content-md5'] = generate_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, is_dir=False): # explicit command line argument always wins content_type = self.config.mime_type content_charset = None if filename == u'-': return self.stdin_content_type() if is_dir: content_type = 'application/x-directory' elif 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) try: is_dir = False size = 0 if filename == "-": is_stream = True src_stream = io.open(sys.stdin.fileno(), mode='rb', closefd=False) src_stream.stream_name = u'<stdin>' else: is_stream = False filename_bytes = deunicodise(filename) stat = os.stat(filename_bytes) mode = stat[ST_MODE] if S_ISDIR(mode): is_dir = True # Dirs are represented as empty objects on S3 src_stream = io.BytesIO(b'') elif not S_ISREG(mode): raise InvalidFileError(u"Not a regular file") else: # Standard normal file src_stream = io.open(filename_bytes, mode='rb') size = stat[ST_SIZE] src_stream.stream_name = filename 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, is_dir=is_dir) ## 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 is_stream: 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 is_stream: 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] return 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"<?xml version=\"1.0\" encoding=\"UTF-8\"?><Delete>" 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"<Object><Key>%s</Key></Object>" % object body += u"</Delete>" 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': generate_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 = '<RestoreRequest xmlns="http://s3.amazonaws.com/doc/2006-03-01/">' body += (' <Days>%s</Days>' % self.config.restore_days) body += ' <GlacierJobParameters>' body += (' <Tier>%s</Tier>' % self.config.restore_priority) body += ' </GlacierJobParameters>' body += '</RestoreRequest>' 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', # Cloudflare's R2 header we don't want to send 'cf-ray', # 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) try: response = self.send_request(request) except S3Error as exc: # A HEAD request will not have body, even in the case of an error # so we can't get the usual XML error content. # Add fake similar content in such a case if exc.status == 404 and not exc.code: exc.code = 'NoSuchKey' exc.message = 'The specified key does not exist.' exc.resource = uri raise exc 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 set_versioning(self, uri, enabled): headers = SortedDict(ignore_case = True) status = "Enabled" if enabled is True else "Suspended" body = '<VersioningConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">' body += '<Status>%s</Status>' % status body += '</VersioningConfiguration>' debug(u"set_versioning(%s)" % body) headers['content-md5'] = generate_content_md5(body) request = self.create_request("BUCKET_CREATE", uri = uri, headers = headers, body = body, uri_params = {'versioning': None}) response = self.send_request(request) return response def get_versioning(self, uri): request = self.create_request("BUCKET_LIST", uri = uri, uri_params = {'versioning': None}) response = self.send_request(request) return getTextFromXml(response['data'], "Status") 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_object_legal_hold(self, uri, legal_hold_status): body = '<LegalHold xmlns="http://s3.amazonaws.com/doc/2006-03-01/">' body += '<Status>%s</Status>' % legal_hold_status body += '</LegalHold>' headers = SortedDict(ignore_case = True) headers['content-type'] = 'application/xml' headers['content-md5'] = generate_content_md5(body) request = self.create_request("OBJECT_PUT", uri = uri, headers = headers, body = body, uri_params = {'legal-hold': None}) response = self.send_request(request) return response def set_object_retention(self, uri, mode, retain_until_date): body = '<Retention xmlns="http://s3.amazonaws.com/doc/2006-03-01/">' body += '<Mode>%s</Mode>' % mode body += '<RetainUntilDate>%s</RetainUntilDate>' % retain_until_date body += '</Retention>' headers = SortedDict(ignore_case = True) headers['content-type'] = 'application/xml' headers['content-md5'] = generate_content_md5(body) request = self.create_request("OBJECT_PUT", uri = uri, headers = headers, body = body, uri_params = {'retention': None}) response = self.send_request(request) return response 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'] = generate_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'] = generate_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 = '<RequestPaymentConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">\n' if self.config.requester_pays: body += '<Payer>Requester</Payer>\n' else: body += '<Payer>BucketOwner</Payer>\n' body += '</RequestPaymentConfiguration>\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 set_notification_policy(self, uri, policy): headers = SortedDict(ignore_case = True) if self.config.skip_destination_validation: headers["x-amz-skip-destination-validation"] = "True" request = self.create_request("BUCKET_CREATE", uri = uri, headers = headers, body = policy, uri_params = {'notification': None}) debug(u"set_notification_policy(%s): policy-xml: %s" % (uri, policy)) response = self.send_request(request) return response def get_notification_policy(self, uri): request = self.create_request("BUCKET_LIST", bucket = uri.bucket(), uri_params = {'notification': None}) debug(u"get_notification_policy(%s)" % uri) response = self.send_request(request) debug(u"%s: Got notification Policy" % response['status']) return response def delete_notification_policy(self, uri): empty_config = '<NotificationConfiguration></NotificationConfiguration>' return self.set_notification_policy(uri, empty_config) def set_tagging(self, uri, tagsets): if uri.type != "s3": raise ValueError("Expected URI type 's3', got '%s'" % uri.type) body = '<Tagging xmlns="http://s3.amazonaws.com/doc/2006-03-01/">' body += '<TagSet>' for (key, val) in tagsets: body += '<Tag>' body += (' <Key>%s</Key>' % key) body += (' <Value>%s</Value>' % val) body += '</Tag>' body += '</TagSet>' body += '</Tagging>' headers = SortedDict(ignore_case=True) headers['content-md5'] = generate_content_md5(body) if uri.has_object(): request = self.create_request("OBJECT_PUT", uri=uri, headers=headers, body=body, uri_params={'tagging': None}) else: request = self.create_request("BUCKET_CREATE", bucket=uri.bucket(), headers=headers, body=body, uri_params={'tagging': None}) debug(u"set_tagging(%s): tagset-xml: %s" % (uri, body)) response = self.send_request(request) return response def get_tagging(self, uri): if uri.has_object(): request = self.create_request("OBJECT_GET", uri=uri, uri_params={'tagging': None}) else: request = self.create_request("BUCKET_LIST", bucket=uri.bucket(), uri_params={'tagging': None}) debug(u"get_tagging(%s)" % uri) response = self.send_request(request) xml_data = response["data"] # extract list of tag sets tagsets = getListFromXml(xml_data, "Tag") debug(u"%s: Got object tagging" % response['status']) return tagsets def delete_tagging(self, uri): if uri.has_object(): request = self.create_request("OBJECT_DELETE", uri=uri, uri_params={'tagging': None}) else: request = self.create_request("BUCKET_DELETE", bucket=uri.bucket(), uri_params={'tagging': None}) debug(u"delete_tagging(%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.config.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=None): if retries is None: retries = self.config.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 = None, offset = 0, chunk_size = -1, use_expect_continue = None): if retries is None: retries = self.config.max_retries 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(stream, 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 not data: raise InvalidFileError("File smaller than expected. Was the file truncated?") 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 InvalidFileError as e: if self.config.progress_meter: progress.done("failed") 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"] == 507: # Not an AWS error, but s3 compatible server possible error: # InsufficientStorage try_retry = False 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 err = S3Error(response) if try_retry: if retries: warning("Upload failed: %s (%s)" % (resource['uri'], err)) 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("%s" % err) ## Non-recoverable error raise err 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("MD5 sums of sent and received files don't match!") 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=None): if retries is None: retries = self.config.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=retries - 1) else: raise S3DownloadError("Download failed for: %s" % resource['uri']) if response["status"] < 200 or response["status"] > 299: # In case of error, we still need to flush the read buffer to be able to reuse # the connection response['data'] = http_response.read() if response["status"] in [301, 307]: ## RedirectPermanent or RedirectTemporary return self._http_redirection_handler(request, response, self.recv_file, request, stream, labels, start_position) if response["status"] == 400: 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: return self._http_403_handler(request, response, self.recv_file, request, stream, labels, start_position) if response["status"] < 200 or response["status"] > 299: try_retry = False if response["status"] == 429: # Not an AWS error, but s3 compatible server possible error: # TooManyRequests/Busy/slowdown try_retry = True elif response["status"] == 503: # SlowDown error try_retry = True if try_retry: resource_uri = resource['uri'] if retries: retry_delay = self._fail_wait(retries) warning("Retrying failed request: %s (%s)" % (resource_uri, S3Error(response))) warning("Waiting %d sec..." % retry_delay) time.sleep(retry_delay) return self.recv_file(request, stream, labels, start_position, retries=retries - 1) else: warning("Too many failures. Giving up on '%s'" % resource_uri) raise S3DownloadError("Download failed for: %s" % resource_uri) # Non-recoverable error 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=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 # vim:et:ts=4:sts=4:ai	102.36 KB	2023-12-12 01:00
📄 S3Uri.py # -- coding: utf-8 -- ## -------------------------------------------------------------------- ## Amazon S3 manager ## ## Authors : Michal Ludvig <michal@logix.cz> (https://www.logix.cz/michal) ## Florent Viard <florent@sodria.com> (https://www.sodria.com) ## Copyright : TGRMN Software, Sodria SAS and contributors ## License : GPL Version 2 ## Website : https://s3tools.org ## -------------------------------------------------------------------- from __future__ import absolute_import, print_function import os import re import sys from .Utils import unicodise, deunicodise, check_bucket_name_dns_support from . import Config PY3 = (sys.version_info >= (3, 0)) class S3Uri(object): type = None _subclasses = None def __new__(self, string): if not self._subclasses: ## Generate a list of all subclasses of S3Uri self._subclasses = [] dict = sys.modules[__name__].__dict__ for something in dict: if type(dict[something]) is not type(self): continue if issubclass(dict[something], self) and dict[something] != self: self._subclasses.append(dict[something]) for subclass in self._subclasses: try: instance = object.__new__(subclass) instance.__init__(string) return instance except ValueError: continue raise ValueError("%s: not a recognized URI" % string) def __str__(self): if PY3: return self.uri() else: return deunicodise(self.uri()) def __unicode__(self): return self.uri() def __repr__(self): return repr("<%s: %s>" % (self.__class__.__name__, self.__unicode__())) def public_url(self): raise ValueError("This S3 URI does not have Anonymous URL representation") def basename(self): return self.__unicode__().split("/")[-1] class S3UriS3(S3Uri): type = "s3" _re = re.compile("^s3:///([^/])/?(.)", re.IGNORECASE \| re.UNICODE) def __init__(self, string): match = self._re.match(string) if not match: raise ValueError("%s: not a S3 URI" % string) groups = match.groups() self._bucket = groups[0] self._object = groups[1] def bucket(self): return self._bucket def object(self): return self._object def has_bucket(self): return bool(self._bucket) def has_object(self): return bool(self._object) def uri(self): return u"/".join([u"s3:/", self._bucket, self._object]) def is_dns_compatible(self): return check_bucket_name_dns_support(Config.Config().host_bucket, self._bucket) def public_url(self): public_url_protocol = "http" if Config.Config().public_url_use_https: public_url_protocol = "https" if self.is_dns_compatible(): return "%s://%s.%s/%s" % (public_url_protocol, self._bucket, Config.Config().host_base, self._object) else: return "%s://%s/%s/%s" % (public_url_protocol, Config.Config().host_base, self._bucket, self._object) def host_name(self): if self.is_dns_compatible(): return "%s.s3.amazonaws.com" % (self._bucket) else: return "s3.amazonaws.com" @staticmethod def compose_uri(bucket, object = ""): return u"s3://%s/%s" % (bucket, object) @staticmethod def httpurl_to_s3uri(http_url): m = re.match("(https?://)?([^/]+)/?(.)", http_url, re.IGNORECASE \| re.UNICODE) hostname, object = m.groups()[1:] hostname = hostname.lower() # Worst case scenario, we would like to be able to match something like # my.website.com.s3-fips.dualstack.us-west-1.amazonaws.com.cn m = re.match("(.\.)?s3(?:\-[^\.])?(?:\.dualstack)?(?:\.[^\.])?\.amazonaws\.com(?:\.cn)?$", hostname, re.IGNORECASE \| re.UNICODE) if not m: raise ValueError("Unable to parse URL: %s" % http_url) bucket = m.groups()[0] if not bucket: ## old-style url: http://s3.amazonaws.com/bucket/object if "/" not in object: ## no object given bucket = object object = "" else: ## bucket/object bucket, object = object.split("/", 1) else: ## new-style url: http://bucket.s3.amazonaws.com/object bucket = bucket.rstrip('.') return S3Uri( u"s3://%(bucket)s/%(object)s" % { 'bucket' : bucket, 'object' : object } ) class S3UriS3FS(S3Uri): type = "s3fs" _re = re.compile("^s3fs:///([^/])/?(.)", re.IGNORECASE \| re.UNICODE) def __init__(self, string): match = self._re.match(string) if not match: raise ValueError("%s: not a S3fs URI" % string) groups = match.groups() self._fsname = groups[0] self._path = groups[1].split("/") def fsname(self): return self._fsname def path(self): return "/".join(self._path) def uri(self): return u"/".join([u"s3fs:/", self._fsname, self.path()]) class S3UriFile(S3Uri): type = "file" _re = re.compile("^(\w+://)?(.)", re.UNICODE) def __init__(self, string): match = self._re.match(string) groups = match.groups() if groups[0] not in (None, "file://"): raise ValueError("%s: not a file:// URI" % string) if groups[0] is None: self._path = groups[1].split(os.sep) else: self._path = groups[1].split("/") def path(self): return os.sep.join(self._path) def uri(self): return u"/".join([u"file:/"]+self._path) def isdir(self): return os.path.isdir(deunicodise(self.path())) def dirname(self): return unicodise(os.path.dirname(deunicodise(self.path()))) def basename(self): return unicodise(os.path.basename(deunicodise(self.path()))) class S3UriCloudFront(S3Uri): type = "cf" _re = re.compile("^cf://([^/])/(.)", re.IGNORECASE \| re.UNICODE) def __init__(self, string): match = self._re.match(string) if not match: raise ValueError("%s: not a CloudFront URI" % string) groups = match.groups() self._dist_id = groups[0] self._request_id = groups[1] != "/" and groups[1] or None def dist_id(self): return self._dist_id def request_id(self): return self._request_id def uri(self): uri = u"cf://" + self.dist_id() if self.request_id(): uri += u"/" + self.request_id() return uri if __name__ == "__main__": uri = S3Uri("s3://bucket/object") print("type() =", type(uri)) print("uri =", uri) print("uri.type=", uri.type) print("bucket =", uri.bucket()) print("object =", uri.object()) print() uri = S3Uri("s3://bucket") print("type() =", type(uri)) print("uri =", uri) print("uri.type=", uri.type) print("bucket =", uri.bucket()) print() uri = S3Uri("s3fs://filesystem1/path/to/remote/file.txt") print("type() =", type(uri)) print("uri =", uri) print("uri.type=", uri.type) print("path =", uri.path()) print() uri = S3Uri("/path/to/local/file.txt") print("type() =", type(uri)) print("uri =", uri) print("uri.type=", uri.type) print("path =", uri.path()) print() uri = S3Uri("cf://1234567890ABCD/") print("type() =", type(uri)) print("uri =", uri) print("uri.type=", uri.type) print("dist_id =", uri.dist_id()) print() # vim:et:ts=4:sts=4:ai	7.69 KB	2023-12-06 09:07
📄 SortedDict.py # -- coding: utf-8 -- ## -------------------------------------------------------------------- ## Amazon S3 manager ## ## Authors : Michal Ludvig <michal@logix.cz> (https://www.logix.cz/michal) ## Florent Viard <florent@sodria.com> (https://www.sodria.com) ## Copyright : TGRMN Software, Sodria SAS and contributors ## License : GPL Version 2 ## Website : https://s3tools.org ## -------------------------------------------------------------------- from __future__ import absolute_import, print_function from .BidirMap import BidirMap class SortedDictIterator(object): def __init__(self, sorted_dict, keys, reverse=False): self.sorted_dict = sorted_dict self.keys = keys if reverse: self.pop_index = -1 else: self.pop_index = 0 def __iter__(self): return self def __next__(self): try: return self.keys.pop(self.pop_index) except IndexError: raise StopIteration next = __next__ class SortedDict(dict): def __init__(self, mapping = {}, ignore_case = True, kwargs): """ WARNING: SortedDict() with ignore_case==True will drop entries differing only in capitalisation! Eg: SortedDict({'auckland':1, 'Auckland':2}).keys() => ['Auckland'] With ignore_case==False it's all right """ dict.__init__(self, mapping, kwargs) self.ignore_case = ignore_case def keys(self): # TODO fix # Probably not anymore memory efficient on python2 # as now 2 copies of keys to sort them. keys = dict.keys(self) if self.ignore_case: # Translation map xlat_map = BidirMap() for key in keys: xlat_map[key.lower()] = key # Lowercase keys lc_keys = sorted(xlat_map.keys()) return [xlat_map[k] for k in lc_keys] else: keys = sorted(keys) return keys def __iter__(self): return SortedDictIterator(self, self.keys()) def __reversed__(self): return SortedDictIterator(self, self.keys(), reverse=True) def __getitem__(self, index): """Override to support the "get_slice" for python3 """ if isinstance(index, slice): r = SortedDict(ignore_case = self.ignore_case) for k in self.keys()[index]: r[k] = self[k] else: r = super(SortedDict, self).__getitem__(index) return r if __name__ == "__main__": d = { 'AWS' : 1, 'Action' : 2, 'america' : 3, 'Auckland' : 4, 'America' : 5 } sd = SortedDict(d) print("Wanted: Action, america, Auckland, AWS, [ignore case]") print("Got: ", end=' ') for key in sd: print("%s," % key, end=' ') print(" [used: __iter__()]") d = SortedDict(d, ignore_case = False) print("Wanted: AWS, Action, America, Auckland, america, [case sensitive]") print("Got: ", end=' ') for key in d.keys(): print("%s," % key, end=' ') print(" [used: keys()]") # vim:et:ts=4:sts=4:ai	3.08 KB	2023-12-06 09:07
📄 Utils.py # -- coding: utf-8 -- ## -------------------------------------------------------------------- ## Amazon S3 manager ## ## Authors : Michal Ludvig <michal@logix.cz> (https://www.logix.cz/michal) ## Florent Viard <florent@sodria.com> (https://www.sodria.com) ## Copyright : TGRMN Software, Sodria SAS and contributors ## License : GPL Version 2 ## Website : https://s3tools.org ## -------------------------------------------------------------------- from __future__ import absolute_import, division import os import time import re import string as string_mod import random import errno from logging import debug try: unicode except NameError: # python 3 support # In python 3, unicode -> str, and str -> bytes unicode = str import S3.Config import S3.Exceptions from S3.BaseUtils import (base_urlencode_string, base_replace_nonprintables, base_unicodise, base_deunicodise, md5) __all__ = [] def formatSize(size, human_readable=False, floating_point=False): size = floating_point and float(size) or int(size) if human_readable: coeffs = ['K', 'M', 'G', 'T'] coeff = "" while size > 2048: size /= 1024 coeff = coeffs.pop(0) return (floating_point and float(size) or int(size), coeff) else: return (size, "") __all__.append("formatSize") def convertHeaderTupleListToDict(list): """ Header keys are always in lowercase in python2 but not in python3. """ retval = {} for tuple in list: retval[tuple[0].lower()] = tuple[1] return retval __all__.append("convertHeaderTupleListToDict") _rnd_chars = string_mod.ascii_letters + string_mod.digits _rnd_chars_len = len(_rnd_chars) def rndstr(len): retval = "" while len > 0: retval += _rnd_chars[random.randint(0, _rnd_chars_len-1)] len -= 1 return retval __all__.append("rndstr") def mktmpsomething(prefix, randchars, createfunc): old_umask = os.umask(0o077) tries = 5 while tries > 0: dirname = prefix + rndstr(randchars) try: createfunc(dirname) break except OSError as e: if e.errno != errno.EEXIST: os.umask(old_umask) raise tries -= 1 os.umask(old_umask) return dirname __all__.append("mktmpsomething") def mktmpdir(prefix = os.getenv('TMP','/tmp') + "/tmpdir-", randchars = 10): return mktmpsomething(prefix, randchars, os.mkdir) __all__.append("mktmpdir") def mktmpfile(prefix = os.getenv('TMP','/tmp') + "/tmpfile-", randchars = 20): createfunc = lambda filename : os.close(os.open(deunicodise(filename), os.O_CREAT \| os.O_EXCL)) return mktmpsomething(prefix, randchars, createfunc) __all__.append("mktmpfile") def mkdir_with_parents(dir_name): """ mkdir_with_parents(dst_dir) Create directory 'dir_name' with all parent directories Returns True on success, False otherwise. """ pathmembers = dir_name.split(os.sep) tmp_stack = [] while pathmembers and not os.path.isdir(deunicodise(os.sep.join(pathmembers))): tmp_stack.append(pathmembers.pop()) while tmp_stack: pathmembers.append(tmp_stack.pop()) cur_dir = os.sep.join(pathmembers) try: debug("mkdir(%s)" % cur_dir) os.mkdir(deunicodise(cur_dir)) except (OSError, IOError) as e: debug("Can not make directory '%s' (Reason: %s)" % (cur_dir, e.strerror)) return False except Exception as e: debug("Can not make directory '%s' (Reason: %s)" % (cur_dir, e)) return False return True __all__.append("mkdir_with_parents") def unicodise(string, encoding=None, errors='replace', silent=False): if not encoding: encoding = S3.Config.Config().encoding return base_unicodise(string, encoding, errors, silent) __all__.append("unicodise") def unicodise_s(string, encoding=None, errors='replace'): """ Alias to silent version of unicodise """ return unicodise(string, encoding, errors, True) __all__.append("unicodise_s") def deunicodise(string, encoding=None, errors='replace', silent=False): if not encoding: encoding = S3.Config.Config().encoding return base_deunicodise(string, encoding, errors, silent) __all__.append("deunicodise") def deunicodise_s(string, encoding=None, errors='replace'): """ Alias to silent version of deunicodise """ return deunicodise(string, encoding, errors, True) __all__.append("deunicodise_s") def unicodise_safe(string, encoding=None): """ Convert 'string' to Unicode according to current encoding and replace all invalid characters with '?' """ return unicodise(deunicodise(string, encoding), encoding).replace(u'\ufffd', '?') __all__.append("unicodise_safe") ## Low level methods def urlencode_string(string, urlencoding_mode=None, unicode_output=False): if urlencoding_mode is None: urlencoding_mode = S3.Config.Config().urlencoding_mode return base_urlencode_string(string, urlencoding_mode, unicode_output) __all__.append("urlencode_string") def replace_nonprintables(string): """ replace_nonprintables(string) Replaces all non-printable characters 'ch' in 'string' where ord(ch) <= 26 with ^@, ^A, ... ^Z """ warning_message = (S3.Config.Config().urlencoding_mode != "fixbucket") return base_replace_nonprintables(string, warning_message) __all__.append("replace_nonprintables") def time_to_epoch(t): """Convert time specified in a variety of forms into UNIX epoch time. Accepts datetime.datetime, int, anything that has a strftime() method, and standard time 9-tuples """ if isinstance(t, int): # Already an int return t elif isinstance(t, tuple) or isinstance(t, time.struct_time): # Assume it's a time 9-tuple return int(time.mktime(t)) elif hasattr(t, 'timetuple'): # Looks like a datetime object or compatible return int(time.mktime(t.timetuple())) elif hasattr(t, 'strftime'): # Looks like the object supports standard srftime() return int(t.strftime('%s')) elif isinstance(t, str) or isinstance(t, unicode) or isinstance(t, bytes): # See if it's a string representation of an epoch try: # Support relative times (eg. "+60") if t.startswith('+'): return time.time() + int(t[1:]) return int(t) except ValueError: # Try to parse it as a timestamp string try: return time.strptime(t) except ValueError as ex: # Will fall through debug("Failed to parse date with strptime: %s", ex) pass raise S3.Exceptions.ParameterError('Unable to convert %r to an epoch time. Pass an epoch time. Try `date -d \'now + 1 year\' +%%s` (shell) or time.mktime (Python).' % t) def check_bucket_name(bucket, dns_strict=True): if dns_strict: invalid = re.search(r"([^a-z0-9\.-])", bucket, re.UNICODE) if invalid: raise S3.Exceptions.ParameterError("Bucket name '%s' contains disallowed character '%s'. The only supported ones are: lowercase us-ascii letters (a-z), digits (0-9), dot (.) and hyphen (-)." % (bucket, invalid.groups()[0])) else: invalid = re.search(r"([^A-Za-z0-9\._-])", bucket, re.UNICODE) if invalid: raise S3.Exceptions.ParameterError("Bucket name '%s' contains disallowed character '%s'. The only supported ones are: us-ascii letters (a-z, A-Z), digits (0-9), dot (.), hyphen (-) and underscore (_)." % (bucket, invalid.groups()[0])) if len(bucket) < 3: raise S3.Exceptions.ParameterError("Bucket name '%s' is too short (min 3 characters)" % bucket) if len(bucket) > 255: raise S3.Exceptions.ParameterError("Bucket name '%s' is too long (max 255 characters)" % bucket) if dns_strict: if len(bucket) > 63: raise S3.Exceptions.ParameterError("Bucket name '%s' is too long (max 63 characters)" % bucket) if re.search(r"-\.", bucket, re.UNICODE): raise S3.Exceptions.ParameterError("Bucket name '%s' must not contain sequence '-.' for DNS compatibility" % bucket) if re.search(r"\.\.", bucket, re.UNICODE): raise S3.Exceptions.ParameterError("Bucket name '%s' must not contain sequence '..' for DNS compatibility" % bucket) if not re.search(r"^[0-9a-z]", bucket, re.UNICODE): raise S3.Exceptions.ParameterError("Bucket name '%s' must start with a letter or a digit" % bucket) if not re.search(r"[0-9a-z]$", bucket, re.UNICODE): raise S3.Exceptions.ParameterError("Bucket name '%s' must end with a letter or a digit" % bucket) return True __all__.append("check_bucket_name") def check_bucket_name_dns_conformity(bucket): try: return check_bucket_name(bucket, dns_strict = True) except S3.Exceptions.ParameterError: return False __all__.append("check_bucket_name_dns_conformity") def check_bucket_name_dns_support(bucket_host, bucket_name): """ Check whether either the host_bucket support buckets and either bucket name is dns compatible """ if "%(bucket)s" not in bucket_host: return False return check_bucket_name_dns_conformity(bucket_name) __all__.append("check_bucket_name_dns_support") def getBucketFromHostname(hostname): """ bucket, success = getBucketFromHostname(hostname) Only works for hostnames derived from bucket names using Config.host_bucket pattern. Returns bucket name and a boolean success flag. """ if "%(bucket)s" not in S3.Config.Config().host_bucket: return (hostname, False) # Create RE pattern from Config.host_bucket pattern = S3.Config.Config().host_bucket.lower() % { 'bucket' : '(?P<bucket>.*)' } m = re.match(pattern, hostname, re.UNICODE) if not m: return (hostname, False) return m.group(1), True __all__.append("getBucketFromHostname") def getHostnameFromBucket(bucket): return S3.Config.Config().host_bucket.lower() % { 'bucket' : bucket } __all__.append("getHostnameFromBucket") # Deal with the fact that pwd and grp modules don't exist for Windows try: import pwd def getpwuid_username(uid): """returns a username from the password database for the given uid""" return unicodise_s(pwd.getpwuid(uid).pw_name) except ImportError: import getpass def getpwuid_username(uid): return unicodise_s(getpass.getuser()) __all__.append("getpwuid_username") try: import grp def getgrgid_grpname(gid): """returns a groupname from the group database for the given gid""" return unicodise_s(grp.getgrgid(gid).gr_name) except ImportError: def getgrgid_grpname(gid): return u"nobody" __all__.append("getgrgid_grpname") # vim:et:ts=4:sts=4:ai	10.77 KB	2023-12-06 09:07
📄 __init__.py # -- coding: utf-8 --	24 B	2023-12-06 09:07
📁 __pycache__	-	2025-10-21 13:18

Edit File:

Terminal