Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse files

Add Google Cloud Storage support for CORS configuration

  • Loading branch information...
commit b30134c5d204c4ae00973f0eba3e71612e837bd9 1 parent afee998
@mfschwartz mfschwartz authored
View
7 boto/exception.py
@@ -393,6 +393,13 @@ def __init__(self, message):
Exception.__init__(self, message)
self.message = message
+class InvalidCorsError(Exception):
+ """Exception raised when CORS XML is invalid."""
+
+ def __init__(self, message):
+ Exception.__init__(self, message)
+ self.message = message
+
class NoAuthHandlerFound(Exception):
"""Is raised when no auth handlers were found ready to authenticate."""
pass
View
32 boto/gs/bucket.py
@@ -24,14 +24,16 @@
from boto.exception import InvalidAclError
from boto.gs.acl import ACL, CannedACLStrings
from boto.gs.acl import SupportedPermissions as GSPermissions
+from boto.gs.cors import Cors
from boto.gs.key import Key as GSKey
from boto.s3.acl import Policy
from boto.s3.bucket import Bucket as S3Bucket
import xml.sax
-# constants for default object ACL and standard acl in http query args
+# constants for http query args
DEF_OBJ_ACL = 'defaultObjectAcl'
STANDARD_ACL = 'acl'
+CORS_ARG = 'cors'
class Bucket(S3Bucket):
@@ -122,6 +124,34 @@ def set_def_xml_acl(self, acl_str, key_name='', headers=None):
return self.set_xml_acl(acl_str, key_name, headers,
query_args=DEF_OBJ_ACL)
+ def get_cors(self, headers=None):
+ """returns a bucket's CORS XML"""
+ response = self.connection.make_request('GET', self.name,
+ query_args=CORS_ARG,
+ headers=headers)
+ body = response.read()
+ if response.status == 200:
+ # Success - parse XML and return Cors object.
+ cors = Cors()
+ h = handler.XmlHandler(cors, self)
+ xml.sax.parseString(body, h)
+ return cors
+ else:
+ raise self.connection.provider.storage_response_error(
+ response.status, response.reason, body)
+
+ def set_cors(self, cors, headers=None):
+ """sets or changes a bucket's CORS XML."""
+ cors_xml = cors.encode('ISO-8859-1')
+ response = self.connection.make_request('PUT', self.name,
+ data=cors_xml,
+ query_args=CORS_ARG,
+ headers=headers)
+ body = response.read()
+ if response.status != 200:
+ raise self.connection.provider.storage_response_error(
+ response.status, response.reason, body)
+
# Method with same signature as boto.s3.bucket.Bucket.add_email_grant(),
# to allow polymorphic treatment at application layer.
def add_email_grant(self, permission, email_address,
View
169 boto/gs/cors.py
@@ -0,0 +1,169 @@
+# Copyright 2012 Google Inc.
+#
+# Permission is hereby granted, free of charge, to any person obtaining a
+# copy of this software and associated documentation files (the
+# "Software"), to deal in the Software without restriction, including
+# without limitation the rights to use, copy, modify, merge, publish, dis-
+# tribute, sublicense, and/or sell copies of the Software, and to permit
+# persons to whom the Software is furnished to do so, subject to the fol-
+# lowing conditions:
+#
+# The above copyright notice and this permission notice shall be included
+# in all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL-
+# ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
+# SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
+# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+# IN THE SOFTWARE.
+
+import types
+from boto.gs.user import User
+from boto.exception import InvalidCorsError
+from xml.sax import handler
+
+# Relevant tags for the CORS XML document.
+CORS_CONFIG = 'CorsConfig'
+CORS = 'Cors'
+ORIGINS = 'Origins'
+ORIGIN = 'Origin'
+METHODS = 'Methods'
+METHOD = 'Method'
+HEADERS = 'ResponseHeaders'
+HEADER = 'ResponseHeader'
+MAXAGESEC = 'MaxAgeSec'
+
+class Cors(handler.ContentHandler):
+ """Encapsulates the CORS configuration XML document"""
+ def __init__(self):
+ # List of CORS elements found within a CorsConfig element.
+ self.cors = []
+ # List of collections (e.g. Methods, ResponseHeaders, Origins)
+ # found within a CORS element. We use a list of lists here
+ # instead of a dictionary because the collections need to be
+ # preserved in the order in which they appear in the input XML
+ # document (and Python dictionary keys are inherently unordered).
+ # The elements on this list are two element tuples of the form
+ # (collection name, [list of collection contents]).
+ self.collections = []
+ # Lists of elements within a collection. Again a list is needed to
+ # preserve ordering but also because the same element may appear
+ # multiple times within a collection.
+ self.elements = []
+ # Dictionary mapping supported collection names to element types
+ # which may be contained within each.
+ self.legal_collections = {
+ ORIGINS : [ORIGIN],
+ METHODS : [METHOD],
+ HEADERS : [HEADER],
+ MAXAGESEC: []
+ }
+ # List of supported element types within any collection, used for
+ # checking validadity of a parsed element name.
+ self.legal_elements = [ORIGIN, METHOD, HEADER]
+
+ self.parse_level = 0
+ self.collection = None
+ self.element = None
+
+ def validateParseLevel(self, tag, level):
+ """Verify parse level for a given tag."""
+ if self.parse_level != level:
+ raise InvalidCorsError('Invalid tag %s at parse level %d: ' %
+ (tag, self.parse_level))
+
+ def startElement(self, name, attrs, connection):
+ """SAX XML logic for parsing new element found."""
+ if name == CORS_CONFIG:
+ self.validateParseLevel(name, 0)
+ self.parse_level += 1;
+ elif name == CORS:
+ self.validateParseLevel(name, 1)
+ self.parse_level += 1;
+ elif name in self.legal_collections:
+ self.validateParseLevel(name, 2)
+ self.parse_level += 1;
+ self.collection = name
+ elif name in self.legal_elements:
+ self.validateParseLevel(name, 3)
+ # Make sure this tag is found inside a collection tag.
+ if self.collection is None:
+ raise InvalidCorsError('Tag %s found outside collection' % name)
+ # Make sure this tag is allowed for the current collection tag.
+ if name not in self.legal_collections[self.collection]:
+ raise InvalidCorsError('Tag %s not allowed in %s collection' %
+ (name, self.collection))
+ self.element = name
+ else:
+ raise InvalidCorsError('Unsupported tag ' + name)
+
+ def endElement(self, name, value, connection):
+ """SAX XML logic for parsing new element found."""
+ if name == CORS_CONFIG:
+ self.validateParseLevel(name, 1)
+ self.parse_level -= 1;
+ elif name == CORS:
+ self.validateParseLevel(name, 2)
+ self.parse_level -= 1;
+ # Terminating a CORS element, save any collections we found
+ # and re-initialize collections list.
+ self.cors.append(self.collections)
+ self.collections = []
+ elif name in self.legal_collections:
+ self.validateParseLevel(name, 3)
+ if name != self.collection:
+ raise InvalidCorsError('Mismatched start and end tags (%s/%s)' %
+ (self.collection, name))
+ self.parse_level -= 1;
+ if not self.legal_collections[name]:
+ # If this collection doesn't contain any sub-elements, store
+ # a tuple of name and this tag's element value.
+ self.collections.append((name, value.strip()))
+ else:
+ # Otherwise, we're terminating a collection of sub-elements,
+ # so store a tuple of name and list of contained elements.
+ self.collections.append((name, self.elements))
+ self.elements = []
+ self.collection = None
+ elif name in self.legal_elements:
+ self.validateParseLevel(name, 3)
+ # Make sure this tag is found inside a collection tag.
+ if self.collection is None:
+ raise InvalidCorsError('Tag %s found outside collection' % name)
+ # Make sure this end tag is allowed for the current collection tag.
+ if name not in self.legal_collections[self.collection]:
+ raise InvalidCorsError('Tag %s not allowed in %s collection' %
+ (name, self.collection))
+ if name != self.element:
+ raise InvalidCorsError('Mismatched start and end tags (%s/%s)' %
+ (self.element, name))
+ # Terminating an element tag, add it to the list of elements
+ # for the current collection.
+ self.elements.append((name, value.strip()))
+ self.element = None
+ else:
+ raise InvalidCorsError('Unsupported end tag ' + name)
+
+ def to_xml(self):
+ """Convert CORS object into XML string representation."""
+ s = '<' + CORS_CONFIG + '>'
+ for collections in self.cors:
+ s += '<' + CORS + '>'
+ for (collection, elements_or_value) in collections:
+ assert collection is not None
+ s += '<' + collection + '>'
+ # If collection elements has type string, append atomic value,
+ # otherwise, append sequence of values in named tags.
+ if isinstance(elements_or_value, types.StringTypes):
+ s += elements_or_value
+ else:
+ for (name, value) in elements_or_value:
+ assert name is not None
+ assert value is not None
+ s += '<' + name + '>' + value + '</' + name + '>'
+ s += '</' + collection + '>'
+ s += '</' + CORS + '>'
+ s += '</' + CORS_CONFIG + '>'
+ return s
View
16 boto/storage_uri.py
@@ -268,6 +268,22 @@ def get_def_acl(self, validate=True, headers=None):
self.check_response(acl, 'acl', self.uri)
return acl
+ def get_cors(self, validate=True, headers=None):
+ """returns a bucket's CORS XML"""
+ if not self.bucket_name:
+ raise InvalidUriError('get_cors on bucket-less URI (%s)' % self.uri)
+ bucket = self.get_bucket(validate, headers)
+ cors = bucket.get_cors(headers)
+ self.check_response(cors, 'cors', self.uri)
+ return cors
+
+ def set_cors(self, cors, validate=True, headers=None):
+ """sets or updates a bucket's CORS XML"""
+ if not self.bucket_name:
+ raise InvalidUriError('set_cors on bucket-less URI (%s)' % self.uri)
+ bucket = self.get_bucket(validate, headers)
+ bucket.set_cors(cors.to_xml(), headers)
+
def get_location(self, validate=True, headers=None):
if not self.bucket_name:
raise InvalidUriError('get_location on bucket-less URI (%s)' %
View
2  boto/utils.py
@@ -71,7 +71,7 @@
_hashfn = md5.md5
# List of Query String Arguments of Interest
-qsa_of_interest = ['acl', 'defaultObjectAcl', 'location', 'logging',
+qsa_of_interest = ['acl', 'cors', 'defaultObjectAcl', 'location', 'logging',
'partNumber', 'policy', 'requestPayment', 'torrent',
'versioning', 'versionId', 'versions', 'website',
'uploads', 'uploadId', 'response-content-type',
View
48 tests/s3/test_gsconnection.py
@@ -31,7 +31,10 @@
import time
import os
import re
+import xml
from boto.gs.connection import GSConnection
+from boto.gs.cors import Cors
+from boto import handler
from boto import storage_uri
class GSConnectionTest (unittest.TestCase):
@@ -291,5 +294,50 @@ def test_3_default_object_acls(self):
assert acl.to_xml() == '<AccessControlList></AccessControlList>'
# delete bucket
uri.delete_bucket()
+
+ def test_4_cors_xml(self):
+ """test setting and getting of CORS XML documents"""
+ # regexp for matching project-private default object ACL
+ cors_empty = '<CorsConfig></CorsConfig>'
+ cors_doc = ('<CorsConfig><Cors><Origins><Origin>origin1.example.com'
+ '</Origin><Origin>origin2.example.com</Origin></Origins>'
+ '<Methods><Method>GET</Method><Method>PUT</Method>'
+ '<Method>POST</Method></Methods><ResponseHeaders>'
+ '<ResponseHeader>foo</ResponseHeader>'
+ '<ResponseHeader>bar</ResponseHeader></ResponseHeaders>'
+ '</Cors></CorsConfig>')
+ c = GSConnection()
+ # create a new bucket
+ bucket_name = 'test-%d' % int(time.time())
+ bucket = c.create_bucket(bucket_name)
+ # now call get_bucket to see if it's really there
+ bucket = c.get_bucket(bucket_name)
+ # get new bucket cors and make sure it's empty
+ cors = re.sub(r'\s', '', bucket.get_cors().to_xml())
+ assert cors == cors_empty
+ # set cors document on new bucket
+ bucket.set_cors(cors_doc)
+ cors = re.sub(r'\s', '', bucket.get_cors().to_xml())
+ assert cors == cors_doc
+ # delete bucket
+ c.delete_bucket(bucket)
+
+ # repeat cors tests using boto's storage_uri interface
+ # create a new bucket
+ bucket_name = 'test-%d' % int(time.time())
+ uri = storage_uri('gs://' + bucket_name)
+ uri.create_bucket()
+ # get new bucket cors and make sure it's empty
+ cors = re.sub(r'\s', '', uri.get_cors().to_xml())
+ assert cors == cors_empty
+ # set cors document on new bucket
+ cors_obj = Cors()
+ h = handler.XmlHandler(cors_obj, None)
+ xml.sax.parseString(cors_doc, h)
+ uri.set_cors(cors_obj)
+ cors = re.sub(r'\s', '', uri.get_cors().to_xml())
+ assert cors == cors_doc
+ # delete bucket
+ uri.delete_bucket()
print '--- tests completed ---'
Please sign in to comment.
Something went wrong with that request. Please try again.