Add ACL support for S3 buckets.

moto/s3/models.py

@@ -245,6 +245,7 @@ class FakeBucket(object):
         self.rules = []
         self.policy = None
         self.website_configuration = None
+        self.acl = get_canned_acl('private')
 
     @property
     def location(self):
@@ -284,6 +285,9 @@ class FakeBucket(object):
             raise NotImplementedError('"Fn::GetAtt" : [ "{0}" , "WebsiteURL" ]"')
         raise UnformattedGetAttTemplateException()
 
+    def set_acl(self, acl):
+        self.acl = acl
+
 
 class S3Backend(BaseBackend):
 
@@ -484,4 +488,13 @@ class S3Backend(BaseBackend):
         if acl is not None:
             key.set_acl(acl)
 
+    def set_bucket_acl(self, bucket_name, acl):
+        bucket = self.get_bucket(bucket_name)
+        bucket.set_acl(acl)
+
+    def get_bucket_acl(self, bucket_name):
+        bucket = self.get_bucket(bucket_name)
+        return bucket.acl
+
+
 s3_backend = S3Backend()

moto/s3/responses.py

@@ -80,7 +80,7 @@ class ResponseObject(_TemplateEnvironmentMixin):
         elif method == 'GET':
             return self._bucket_response_get(bucket_name, querystring, headers)
         elif method == 'PUT':
-            return self._bucket_response_put(body, region_name, bucket_name, querystring, headers)
+            return self._bucket_response_put(request, body, region_name, bucket_name, querystring, headers)
         elif method == 'DELETE':
             return self._bucket_response_delete(body, bucket_name, querystring, headers)
         elif method == 'POST':
@@ -128,6 +128,10 @@ class ResponseObject(_TemplateEnvironmentMixin):
         elif 'website' in querystring:
             website_configuration = self.backend.get_bucket_website_configuration(bucket_name)
             return website_configuration
+        elif 'acl' in querystring:
+            bucket = self.backend.get_bucket(bucket_name)
+            template = self.response_template(S3_OBJECT_ACL_RESPONSE)
+            return template.render(obj=bucket)
         elif 'versions' in querystring:
             delimiter = querystring.get('delimiter', [None])[0]
             encoding_type = querystring.get('encoding-type', [None])[0]
@@ -168,7 +172,7 @@ class ResponseObject(_TemplateEnvironmentMixin):
             result_folders=result_folders
         )
 
-    def _bucket_response_put(self, body, region_name, bucket_name, querystring, headers):
+    def _bucket_response_put(self, request, body, region_name, bucket_name, querystring, headers):
         if 'versioning' in querystring:
             ver = re.search('<Status>([A-Za-z]+)</Status>', body)
             if ver:
@@ -187,6 +191,11 @@ class ResponseObject(_TemplateEnvironmentMixin):
         elif 'policy' in querystring:
             self.backend.set_bucket_policy(bucket_name, body)
             return 'True'
+        elif 'acl' in querystring:
+            acl = self._acl_from_headers(request.headers)
+            # TODO: Support the XML-based ACL format
+            self.backend.set_bucket_acl(bucket_name, acl)
+            return ""
         elif 'website' in querystring:
             self.backend.set_bucket_website_configuration(bucket_name, body)
             return ""
@@ -351,7 +360,7 @@ class ResponseObject(_TemplateEnvironmentMixin):
             bucket_name, key_name, version_id=version_id)
         if 'acl' in query:
             template = self.response_template(S3_OBJECT_ACL_RESPONSE)
-            return 200, headers, template.render(key=key)
+            return 200, headers, template.render(obj=key)
 
         if key:
             headers.update(key.metadata)
@@ -696,7 +705,7 @@ S3_OBJECT_ACL_RESPONSE = """<?xml version="1.0" encoding="UTF-8"?>
         <DisplayName>webfile</DisplayName>
       </Owner>
       <AccessControlList>
-        {% for grant in key.acl.grants %}
+        {% for grant in obj.acl.grants %}
         <Grant>
           {% for grantee in grant.grantees %}
           <Grantee xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

tests/test_s3/test_s3.py

@@ -785,6 +785,31 @@ def test_acl_switching():
                    g.permission == 'READ' for g in grants), grants
 
 
+@mock_s3
+def test_bucket_acl_setting():
+    conn = boto.connect_s3()
+    bucket = conn.create_bucket('foobar')
+
+    bucket.make_public()
+
+    grants = bucket.get_acl().acl.grants
+    assert any(g.uri == 'http://acs.amazonaws.com/groups/global/AllUsers' and
+               g.permission == 'READ' for g in grants), grants
+
+
+@mock_s3
+def test_bucket_acl_switching():
+    conn = boto.connect_s3()
+    bucket = conn.create_bucket('foobar')
+    bucket.make_public()
+
+    bucket.set_acl('private')
+
+    grants = bucket.get_acl().acl.grants
+    assert not any(g.uri == 'http://acs.amazonaws.com/groups/global/AllUsers' and
+                   g.permission == 'READ' for g in grants), grants
+
+
 @mock_s3
 def test_unicode_key():
     conn = boto.connect_s3()