acl: add preliminary support for S3 ACL

Currently, Swift3 sets and retrieves Swift ACLs for S3 ACL requests.  However,
S3 ACL is too different from Swift ACL to implement the below reference.
  http://docs.aws.amazon.com/AmazonS3/latest/dev/S3_ACLs_UsingACLs.html
With this patch, Swift3 uses its own metadata for ACL
(e.g. X-Container-Sysmeta-Swift3-Acl) to achieve the best S3 compatibility.
This patch only embeds the S3 ACL into the Swift metadata.  The swift3
middleware does not use it for S3 requests yet; this will be addressed later.

Change-Id: I4522910b6b3a0066f24caa98727fdeb85837e42b

etc/proxy-server.conf-sample

@@ -24,6 +24,17 @@ user_test_tester3 = testing3
 [filter:swift3]
 use = egg:swift3#swift3
 
+# Swift has no concept of the S3's resource owner; the resources
+# (i.e. containers and objects) created via the Swift API have no owner
+# information. This option specifies how the swift3 middleware handles them
+# with the S3 API.  If this option is 'false', such kinds of resources will be
+# invisible and no users can access them with the S3 API.  If set to 'true',
+# the resource without owner is belong to everyone and everyone can access it
+# with the S3 API.  If you care about S3 compatibility, set 'false' here.  This
+# option makes sense only when the s3_acl option is set to 'true' and your
+# Swift cluster has the resources created via the Swift API.
+# allow_no_owner = false
+#
 # Set a region name of your Swift cluster.  Note that Swift3 doesn't choose a
 # region of the newly created bucket actually.  This value is used only for the
 # GET Bucket location API.
@@ -37,6 +48,23 @@ use = egg:swift3#swift3
 # operation.
 # max_multi_delete_objects = 1000
 #
+# If set to 'true', Swift3 uses its own metadata for ACL
+# (e.g. X-Container-Sysmeta-Swift3-Acl) to achieve the best S3 compatibility.
+# If set to 'false', Swift3 tries to use Swift ACL (e.g. X-Container-Read)
+# instead of S3 ACL as far as possible.  If you want to keep backward
+# compatibility with Swift3 1.7 or earlier, set false here
+# If set to 'false' after set to 'true' and put some container/object,
+# all users will be able to access container/object.
+# Note that s3_acl doesn't keep the acl consistency between S3 API and Swift
+# API. (e.g. when set s3acl to true and PUT acl, we won't get the acl
+# information via Swift API at all and the acl won't be applied against to
+# Swift API even if it is for a bucket currently supported.)
+# Note that s3_acl currently supports only keystone and tempauth.
+# DON'T USE THIS for production before enough testing for your use cases.
+# This stuff is still under development and it might cause something
+# you don't expect.
+# s3_acl = false
+#
 # Specify a host name of your Swift cluster.  This enables virtual-hosted style
 # requests.
 # storage_domain =

swift3/cfg.py

@@ -51,8 +51,10 @@ class Config(dict):
 
 # Global config dictionary.  The default values can be defined here.
 CONF = Config({
+    'allow_no_owner': False,
     'location': 'US',
     'max_bucket_listing': 1000,
     'max_multi_delete_objects': 1000,
+    's3_acl': False,
     'storage_domain': '',
 })

swift3/controllers/__init__.py

@@ -19,6 +19,7 @@ from swift3.controllers.bucket import BucketController
 from swift3.controllers.obj import ObjectController
 
 from swift3.controllers.acl import AclController
+from swift3.controllers.s3_acl import AclController as S3AclController
 from swift3.controllers.multi_delete import MultiObjectDeleteController
 from swift3.controllers.multi_upload import UploadController, \
     PartController, UploadsController
@@ -33,6 +34,7 @@ __all__ = [
     'ObjectController',
 
     'AclController',
+    'S3AclController',
     'MultiObjectDeleteController',
     'PartController',
     'UploadsController',

swift3/controllers/s3_acl.py

@@ -0,0 +1,128 @@
+# Copyright (c) 2014 OpenStack Foundation.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from urllib import quote
+
+from swift3.controllers.base import Controller
+from swift3.response import HTTPOk, MissingSecurityHeader, \
+    UnexpectedContent, MalformedACLError
+from swift3.etree import fromstring, tostring, XMLSyntaxError, DocumentInvalid
+from swift3.subresource import ACL
+from swift3.utils import LOGGER
+
+
+def get_acl(headers, body, bucket_owner, object_owner=None):
+    """
+    Get ACL instance from S3 (e.g. x-amz-grant) headers or S3 acl xml body.
+    """
+    acl = ACL.from_headers(headers, bucket_owner, object_owner)
+
+    if acl is None:
+        # Get acl from request body if possible.
+        if not body:
+            msg = 'Your request was missing a required header'
+            raise MissingSecurityHeader(msg, missing_header_name='x-amz-acl')
+        try:
+            elem = fromstring(body, ACL.root_tag)
+            acl = ACL.from_elem(elem)
+        except(XMLSyntaxError, DocumentInvalid):
+            raise MalformedACLError()
+        except Exception as e:
+            LOGGER.error(e)
+            raise
+    else:
+        if body:
+            # Specifying grant with both header and xml is not allowed.
+            raise UnexpectedContent
+
+    return acl
+
+
+class AclController(Controller):
+    """
+    Handles the following APIs:
+
+     - GET Bucket acl
+     - PUT Bucket acl
+     - GET Object acl
+     - PUT Object acl
+
+    Those APIs are logged as ACL operations in the S3 server log.
+    """
+    def GET(self, req):
+        """
+        Handles GET Bucket acl and GET Object acl.
+        """
+        resp = req.get_response(self.app, 'HEAD')
+        if req.is_object_request:
+            acl = resp.object_acl
+        else:
+            acl = resp.bucket_acl
+
+        acl.check_permission(req.user_id, 'READ_ACP')
+
+        resp = HTTPOk()
+        resp.body = tostring(acl.elem())
+
+        return resp
+
+    def PUT(self, req):
+        """
+        Handles PUT Bucket acl and PUT Object acl.
+        """
+        if req.is_object_request:
+            b_resp = req.get_response(self.app, 'HEAD', obj='')
+            o_resp = req.get_response(self.app, 'HEAD')
+
+            req_acl = get_acl(req.headers, req.xml(ACL.max_xml_length),
+                              b_resp.bucket_acl.owner,
+                              o_resp.object_acl.owner)
+
+            # Don't change the owner of the resource by PUT acl request.
+            o_resp.object_acl.check_owner(req_acl.owner.id)
+            o_resp.object_acl.check_permission(req.user_id, 'WRITE_ACP')
+
+            for g in req_acl.grants:
+                LOGGER.debug('Grant %s %s permission on the object /%s/%s' %
+                             (g.grantee, g.permission, req.container_name,
+                              req.object_name))
+            req.object_acl = req_acl
+            headers = {}
+            src_path = '/%s/%s' % (req.container_name, req.object_name)
+
+            # object-sysmeta' can be updated by 'Copy' method,
+            # but can not be by 'POST' method.
+            # So headers['X-Copy-From'] for copy request is added here.
+            headers['X-Copy-From'] = quote(src_path)
+            headers['Content-Length'] = 0
+            req.get_response(self.app, 'PUT', headers=headers)
+        else:
+            resp = req.get_response(self.app, 'HEAD')
+
+            req_acl = get_acl(req.headers, req.xml(ACL.max_xml_length),
+                              resp.bucket_acl.owner)
+
+            # Don't change the owner of the resource by PUT acl request.
+            resp.bucket_acl.check_owner(req_acl.owner.id)
+            resp.bucket_acl.check_permission(req.user_id, 'WRITE_ACP')
+
+            for g in req_acl.grants:
+                LOGGER.debug('Grant %s %s permission on the bucket /%s' %
+                             (g.grantee, g.permission, req.container_name))
+
+            req.bucket_acl = req_acl
+            req.get_response(self.app, 'POST')
+
+        return HTTPOk()

swift3/exception.py

@@ -28,3 +28,7 @@ class BadSwiftRequest(S3Exception):
 
 class ACLError(S3Exception):
     pass
+
+
+class InvalidSubresource(S3Exception):
+    pass

swift3/middleware.py

@@ -103,6 +103,9 @@ class Swift3Middleware(object):
     def __call__(self, env, start_response):
         try:
             req = Request(env, self.slo_enabled)
+            if CONF.s3_acl:
+                req.authenticate(self.app)
+
             resp = self.handle_request(req)
         except NotS3Request:
             resp = self.app

swift3/request.py

@@ -20,6 +20,7 @@ import base64
 import email.utils
 import datetime
 
+from swift.common.utils import split_path
 from swift.common import swob
 from swift.common.http import HTTP_OK, HTTP_CREATED, HTTP_ACCEPTED, \
     HTTP_NO_CONTENT, HTTP_UNAUTHORIZED, HTTP_FORBIDDEN, HTTP_NOT_FOUND, \
@@ -34,7 +35,7 @@ from swift3.controllers import ServiceController, BucketController, \
     ObjectController, AclController, MultiObjectDeleteController, \
     LocationController, LoggingStatusController, PartController, \
     UploadController, UploadsController, VersioningController, \
-    UnsupportedController
+    UnsupportedController, S3AclController
 from swift3.response import AccessDenied, InvalidArgument, InvalidDigest, \
     RequestTimeTooSkewed, Response, SignatureDoesNotMatch, \
     BucketAlreadyExists, BucketNotEmpty, EntityTooLarge, \
@@ -44,6 +45,8 @@ from swift3.response import AccessDenied, InvalidArgument, InvalidDigest, \
 from swift3.exception import NotS3Request, BadSwiftRequest
 from swift3.utils import utf8encode
 from swift3.cfg import CONF
+from swift3.subresource import decode_acl, encode_acl
+from swift3.utils import sysmeta_header
 
 # List of sub-resources that must be maintained as part of the HMAC
 # signature string.
@@ -57,10 +60,32 @@ ALLOWED_SUB_RESOURCES = sorted([
 ])
 
 
+def _header_acl_property(resource):
+    """
+    Set and retrieve the acl in self.headers
+    """
+    def getter(self):
+        return getattr(self, '_%s' % resource)
+
+    def setter(self, value):
+        self.headers.update(encode_acl(resource, value))
+        setattr(self, '_%s' % resource, value)
+
+    def deleter(self):
+        self.headers[sysmeta_header(resource, 'acl')] = ''
+
+    return property(getter, setter, deleter,
+                    doc='Get and set the %s acl property' % resource)
+
+
 class Request(swob.Request):
     """
     S3 request object.
     """
+
+    bucket_acl = _header_acl_property('container')
+    object_acl = _header_acl_property('object')
+
     def __init__(self, env, slo_enabled=True):
         swob.Request.__init__(self, env)
 
@@ -69,6 +94,8 @@ class Request(swob.Request):
         self.container_name, self.object_name = self._parse_uri()
         self._validate_headers()
         self.token = base64.urlsafe_b64encode(self._canonical_string())
+        self.account = None
+        self.keystone_token = None
         self.user_id = None
         self.slo_enabled = slo_enabled
 
@@ -209,6 +236,36 @@ class Request(swob.Request):
         if 'x-amz-website-redirect-location' in self.headers:
             raise S3NotImplemented('Website redirection is not supported.')
 
+    def authenticate(self, app):
+        """
+        authenticate method will run pre-authenticate request and retrieve
+        account information.
+        Note that it currently supports only keystone and tempauth.
+        (no support for the third party authentication middleware)
+        """
+        sw_req = self.to_swift_req('TEST', None, None, body='')
+        # don't show log message of this request
+        sw_req.environ['swift.proxy_access_log_made'] = True
+
+        sw_resp = sw_req.get_response(app)
+
+        if not sw_req.remote_user:
+            raise SignatureDoesNotMatch()
+
+        _, self.account, _ = split_path(sw_resp.environ['PATH_INFO'],
+                                        2, 3, True)
+        self.account = utf8encode(self.account)
+
+        if 'HTTP_X_USER_NAME' in sw_resp.environ:
+            # keystone
+            self.user_id = "%s:%s" % (sw_resp.environ['HTTP_X_TENANT_NAME'],
+                                      sw_resp.environ['HTTP_X_USER_NAME'])
+            self.user_id = utf8encode(self.user_id)
+            self.keystone_token = sw_req.environ['HTTP_X_AUTH_TOKEN']
+        else:
+            # tempauth
+            self.user_id = self.access_key
+
     @property
     def body(self):
         """
@@ -302,7 +359,10 @@ class Request(swob.Request):
                 raise S3NotImplemented("Multi-part feature isn't support")
 
         if 'acl' in self.params:
-            return AclController
+            if CONF.s3_acl:
+                return S3AclController
+            else:
+                return AclController
         if 'delete' in self.params:
             return MultiObjectDeleteController
         if 'location' in self.params:
@@ -340,10 +400,15 @@ class Request(swob.Request):
         return self.container_name and self.object_name
 
     def to_swift_req(self, method, container, obj, query=None,
-                     body=None):
+                     body=None, headers=None):
         """
         Create a Swift request based on this request's environment.
         """
+        if self.account is None:
+            account = self.access_key
+        else:
+            account = self.account
+
         env = self.environ.copy()
 
         for key in env:
@@ -358,14 +423,21 @@ class Request(swob.Request):
         env['swift.source'] = 'S3'
         if method is not None:
             env['REQUEST_METHOD'] = method
-        env['HTTP_X_AUTH_TOKEN'] = self.token
+
+        if self.keystone_token:
+            # Need to skip S3 authorization since authtoken middleware
+            # overwrites account in PATH_INFO
+            env['HTTP_X_AUTH_TOKEN'] = self.keystone_token
+            del env['HTTP_AUTHORIZATION']
+        else:
+            env['HTTP_X_AUTH_TOKEN'] = self.token
 
         if obj:
-            path = '/v1/%s/%s/%s' % (self.access_key, container, obj)
+            path = '/v1/%s/%s/%s' % (account, container, obj)
         elif container:
-            path = '/v1/%s/%s' % (self.access_key, container)
+            path = '/v1/%s/%s' % (account, container)
         else:
-            path = '/v1/%s' % (self.access_key)
+            path = '/v1/%s' % (account)
         env['PATH_INFO'] = path
 
         query_string = ''
@@ -379,7 +451,8 @@ class Request(swob.Request):
             query_string = '&'.join(params)
         env['QUERY_STRING'] = query_string
 
-        return swob.Request.blank(quote(path), environ=env, body=body)
+        return swob.Request.blank(quote(path), environ=env, body=body,
+                                  headers=headers)
 
     def _swift_success_codes(self, method, container, obj):
         """
@@ -428,6 +501,9 @@ class Request(swob.Request):
                 'PUT': [
                     HTTP_CREATED,
                 ],
+                'POST': [
+                    HTTP_ACCEPTED,
+                ],
                 'DELETE': [
                     HTTP_NO_CONTENT,
                 ],
@@ -484,6 +560,10 @@ class Request(swob.Request):
                     HTTP_REQUEST_ENTITY_TOO_LARGE: EntityTooLarge,
                     HTTP_LENGTH_REQUIRED: MissingContentLength,
                 },
+                'POST': {
+                    HTTP_NOT_FOUND: (NoSuchKey, obj),
+                    HTTP_PRECONDITION_FAILED: PreconditionFailed,
+                },
                 'DELETE': {
                     HTTP_NOT_FOUND: (NoSuchKey, obj),
                 },
@@ -492,7 +572,7 @@ class Request(swob.Request):
         return code_map[method]
 
     def get_response(self, app, method=None, container=None, obj=None,
-                     body=None, query=None):
+                     body=None, query=None, headers=None):
         """
         Calls the application with this request's environment.  Returns a
         Response object that wraps up the application's result.
@@ -503,20 +583,32 @@ class Request(swob.Request):
         if obj is None:
             obj = self.object_name
 
-        sw_req = self.to_swift_req(method, container, obj, query=query,
-                                   body=body)
+        sw_req = self.to_swift_req(method, container, obj, headers=headers,
+                                   body=body, query=query)
+
+        if CONF.s3_acl:
+            sw_req.environ['swift_owner'] = True  # needed to set ACL
+            sw_req.environ['swift.authorize_override'] = True
+            sw_req.environ['swift.authorize'] = lambda req: None
+
         sw_resp = sw_req.get_response(app)
         resp = Response.from_swift_resp(sw_resp)
         status = resp.status_int  # pylint: disable-msg=E1101
 
-        if 'HTTP_X_USER_NAME' in sw_resp.environ:
-            # keystone
-            self.user_id = utf8encode("%s:%s" %
-                                      (sw_resp.environ['HTTP_X_TENANT_NAME'],
-                                       sw_resp.environ['HTTP_X_USER_NAME']))
-        else:
-            # tempauth
-            self.user_id = self.access_key
+        if CONF.s3_acl:
+            resp.bucket_acl = decode_acl('container', resp.sysmeta_headers)
+            resp.object_acl = decode_acl('object', resp.sysmeta_headers)
+
+        if not self.user_id:
+            if 'HTTP_X_USER_NAME' in sw_resp.environ:
+                # keystone
+                self.user_id = \
+                    utf8encode("%s:%s" %
+                               (sw_resp.environ['HTTP_X_TENANT_NAME'],
+                                sw_resp.environ['HTTP_X_USER_NAME']))
+            else:
+                # tempauth
+                self.user_id = self.access_key
 
         success_codes = self._swift_success_codes(method, container, obj)
         error_codes = self._swift_error_codes(method, container, obj)

swift3/response.py

@@ -19,7 +19,7 @@ from functools import partial
 
 from swift.common import swob
 
-from swift3.utils import snake_to_camel
+from swift3.utils import snake_to_camel, sysmeta_prefix
 from swift3.etree import Element, SubElement, tostring
 
 
@@ -84,11 +84,24 @@ class Response(ResponseBase, swob.Response):
             # add double quotes to the etag header
             self.etag = self.etag
 
+        sw_sysmeta_headers = swob.HeaderKeyDict()
+        sw_headers = swob.HeaderKeyDict()
         headers = HeaderKeyDict()
+
         for key, val in self.headers.iteritems():
             _key = key.lower()
+            if _key.startswith(sysmeta_prefix('object')) or \
+                    _key.startswith(sysmeta_prefix('container')):
+                sw_sysmeta_headers[key] = val
+            else:
+                sw_headers[key] = val
+
+        # Handle swift headers
+        for key, val in sw_headers.iteritems():
+            _key = key.lower()
+
             if _key.startswith('x-object-meta-'):
-                headers['x-amz-meta-' + key[14:]] = val
+                headers['x-amz-meta-' + _key[14:]] = val
             elif _key in ('content-length', 'content-type',
                           'content-range', 'content-encoding',
                           'etag', 'last-modified'):
@@ -101,6 +114,7 @@ class Response(ResponseBase, swob.Response):
                 headers['x-rgw-bytes-used'] = val
 
         self.headers = headers
+        self.sysmeta_headers = sw_sysmeta_headers
 
     @classmethod
     def from_swift_resp(cls, sw_resp):

swift3/subresource.py

@@ -13,15 +13,19 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import re
 from functools import partial
-from swift3.response import InvalidArgument, \
+from simplejson import loads, dumps
+
+from swift3.response import InvalidArgument, MalformedACLError, \
     S3NotImplemented, InvalidRequest, AccessDenied
 from swift3.etree import Element, SubElement
+from swift3.utils import LOGGER, sysmeta_header
+from swift3.cfg import CONF
+from swift3.exception import InvalidSubresource
 
 XMLNS_XSI = 'http://www.w3.org/2001/XMLSchema-instance'
-
 PERMISSIONS = ['FULL_CONTROL', 'READ', 'WRITE', 'READ_ACP', 'WRITE_ACP']
+LOG_DELIVERY_USER = '.log_delivery'
 
 """
 An entry point of this approach is here.
@@ -49,6 +53,77 @@ http://docs.aws.amazon.com/AmazonS3/latest/dev/acl-overview.html
 """
 
 
+def encode_acl(resource, acl):
+    """
+    Encode an ACL instance to Swift metadata.
+
+    Given a resource type and an ACL instance, this method returns HTTP
+    headers, which can be used for Swift metadata.
+    """
+    header_value = {"Owner": acl.owner.id}
+    grants = []
+    for grant in acl.grants:
+        grant = {"Permission": grant.permission,
+                 "Grantee": str(grant.grantee)}
+        grants.append(grant)
+    header_value.update({"Grant": grants})
+    headers = {}
+    key = sysmeta_header(resource, 'acl')
+    headers[key] = dumps(header_value, separators=(',', ':'))
+
+    return headers
+
+
+def decode_acl(resource, headers):
+    """
+    Decode Swift metadata to an ACL instance.
+
+    Given a resource type and HTTP headers, this method returns an ACL
+    instance.
+    """
+    value = ''
+
+    key = sysmeta_header(resource, 'acl')
+    if key in headers:
+        value = headers[key]
+
+    if value == '':
+        # Fix me: In the case of value is empty or not dict instance,
+        # I want an instance of Owner as None.
+        # However, in the above process would occur error in reference
+        # to an instance variable of Owner.
+        return ACL(Owner(None, None), [])
+
+    try:
+        encode_value = loads(value)
+        if not isinstance(encode_value, dict):
+            return ACL(Owner(None, None), [])
+
+        id = None
+        name = None
+        grants = []
+        if 'Owner' in encode_value:
+            id = encode_value['Owner']
+            name = encode_value['Owner']
+        if 'Grant' in encode_value:
+            for grant in encode_value['Grant']:
+                grantee = None
+                # pylint: disable-msg=E1101
+                for group in Group.__subclasses__():
+                    if group.__name__ == grant['Grantee']:
+                        grantee = group()
+                if not grantee:
+                    grantee = User(grant['Grantee'])
+                permission = grant['Permission']
+                grants.append(Grant(grantee, permission))
+        return ACL(Owner(id, name), grants)
+    except Exception as e:
+        LOGGER.debug(e)
+        pass
+
+    raise InvalidSubresource((resource, 'acl', value))
+
+
 class Grantee(object):
     """
     Base class for grantee.
@@ -89,10 +164,14 @@ class Grantee(object):
         if type == 'CanonicalUser':
             value = elem.find('./ID').text
             return User(value)
-        if type == 'Group':
+        elif type == 'Group':
             value = elem.find('./URI').text
             subclass = get_group_subclass_from_uri(value)
             return subclass()
+        elif type == 'AmazonCustomerByEmail':
+            raise S3NotImplemented()
+        else:
+            raise MalformedACLError()
 
     @staticmethod
     def from_header(grantee):
@@ -176,8 +255,7 @@ class Group(Grantee):
         return elem
 
     def __str__(self):
-        name = re.sub('(.)([A-Z])', r'\1 \2', self.__class__.__name__)
-        return name + ' group'
+        return self.__class__.__name__
 
 
 def canned_acl_grantees(bucket_owner, object_owner=None):
@@ -253,8 +331,14 @@ class LogDelivery(Group):
     WRITE and READ_ACP permissions on a bucket enables this group to write
     server access logs to the bucket.
     """
-    # TODO: Add support for log delivery group.
-    pass
+    uri = 'http://acs.amazonaws.com/groups/s3/LogDelivery'
+
+    def __contains__(self, key):
+        if ':' in key:
+            tenant, user = key.split(':', 1)
+        else:
+            user = key
+        return user == LOG_DELIVERY_USER
 
 
 class Grant(object):
@@ -271,7 +355,6 @@ class Grant(object):
             raise S3NotImplemented()
         if not isinstance(grantee, Grantee):
             raise
-
         self.grantee = grantee
         self.permission = permission
 
@@ -320,7 +403,7 @@ class ACL(object):
         """
         :param owner: Owner Class for ACL instance
         """
-        self._owner = owner
+        self.owner = owner
         self.grants = grants
 
     @classmethod
@@ -341,8 +424,8 @@ class ACL(object):
         elem = Element(self.root_tag)
 
         owner = SubElement(elem, 'Owner')
-        SubElement(owner, 'ID').text = self._owner.id
-        SubElement(owner, 'DisplayName').text = self._owner.name
+        SubElement(owner, 'ID').text = self.owner.id
+        SubElement(owner, 'DisplayName').text = self.owner.name
 
         SubElement(elem, 'AccessControlList').extend(
             g.elem() for g in self.grants
@@ -350,15 +433,17 @@ class ACL(object):
 
         return elem
 
-    def owner(self):
-        # FIXME: maybe we should return Owner instance
-        return self._owner.id
-
     def check_owner(self, user_id):
         """
         Check that the user is an owner.
         """
-        if user_id != self._owner.id:
+        if not self.owner.id:
+            if CONF.allow_no_owner:
+                # No owner means public.
+                return
+            raise AccessDenied()
+
+        if user_id != self.owner.id:
             raise AccessDenied()
 
     def check_permission(self, user_id, permission):
@@ -390,6 +475,8 @@ class ACL(object):
                 if key.lower().startswith('x-amz-grant-'):
                     permission = key[len('x-amz-grant-'):]
                     permission = permission.upper().replace('-', '_')
+                    if permission not in PERMISSIONS:
+                        continue
                     for grantee in value.split(','):
                         grants.append(
                             Grant(Grantee.from_header(grantee), permission))
@@ -400,7 +487,6 @@ class ACL(object):
                     err_msg = 'Specifying both Canned ACLs and Header ' \
                         'Grants is not allowed'
                     raise InvalidRequest(err_msg)
-
                 grantees = canned_acl_grantees(bucket_owner, object_owner)[acl]
                 for permission, grantee in grantees:
                     grants.append(Grant(grantee, permission))

swift3/test/unit/__init__.py

@@ -39,6 +39,7 @@ class FakeApp(object):
         tenant, user = tenant_user.rsplit(':', 1)
 
         path = env['PATH_INFO']
+
         env['PATH_INFO'] = path.replace(tenant_user, 'AUTH_' + tenant)
 
     def __call__(self, env, start_response):

swift3/test/unit/helpers.py

@@ -19,6 +19,7 @@ from copy import deepcopy
 from hashlib import md5
 from swift.common import swob
 from swift.common.utils import split_path
+from swift3.cfg import CONF
 
 
 class FakeSwift(object):
@@ -34,7 +35,30 @@ class FakeSwift(object):
         # mapping of (method, path) --> (response class, headers, body)
         self._responses = {}
 
+    def _fake_auth_middleware(self, env):
+        if 'swift.authorize_override' in env:
+            return
+
+        if 'HTTP_AUTHORIZATION' not in env:
+            return
+
+        _, authorization = env['HTTP_AUTHORIZATION'].split(' ')
+        tenant_user, sign = authorization.rsplit(':', 1)
+        tenant, user = tenant_user.rsplit(':', 1)
+
+        path = env['PATH_INFO']
+        env['PATH_INFO'] = path.replace(tenant_user, 'AUTH_' + tenant)
+
+        env['REMOTE_USER'] = 'authorized'
+
+        # AccessDenied by default
+        env['swift.authorize'] = lambda req: swob.HTTPForbidden(request=req)
+
     def __call__(self, env, start_response):
+        if CONF.s3_acl:
+            self._fake_auth_middleware(env)
+
+        req = swob.Request(env)
         method = env['REQUEST_METHOD']
         path = env['PATH_INFO']
         _, acc, cont, obj = split_path(env['PATH_INFO'], 0, 4,
@@ -43,11 +67,11 @@ class FakeSwift(object):
             path += '?' + env['QUERY_STRING']
 
         if 'swift.authorize' in env:
-            resp = env['swift.authorize']()
+            resp = env['swift.authorize'](req)
             if resp:
                 return resp(env, start_response)
 
-        headers = swob.Request(env).headers
+        headers = req.headers
         self._calls.append((method, path, headers))
         self.swift_sources.append(env.get('swift.source'))
 
@@ -86,7 +110,6 @@ class FakeSwift(object):
                 self.uploaded[path][0]['Content-Type'] = env["CONTENT_TYPE"]
 
         # range requests ought to work, hence conditional_response=True
-        req = swob.Request(env)
         resp = resp_class(req=req, headers=headers, body=body,
                           conditional_response=True)
         return resp(env, start_response)

swift3/test/unit/test_s3_acl.py

@@ -0,0 +1,385 @@
+# Copyright (c) 2014 OpenStack Foundation
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import unittest
+
+from swift.common import swob
+from swift.common.swob import Request
+
+from swift3.etree import tostring, Element, SubElement
+from swift3.subresource import ACL, ACLPrivate, User, encode_acl, \
+    Owner, Grant
+from swift3.test.unit.test_middleware import Swift3TestCase
+from swift3.cfg import CONF
+
+XMLNS_XSI = 'http://www.w3.org/2001/XMLSchema-instance'
+
+
+def _gen_test_acl(owner, permission=None, grantee=None):
+    if permission is None:
+        return ACL(owner, [])
+
+    if grantee is None:
+        grantee = User('test:tester')
+    return ACL(owner, [Grant(grantee, permission)])
+
+
+def _make_xml(grantee):
+    owner = 'test:tester'
+    permission = 'READ'
+    elem = Element('AccessControlPolicy')
+    elem_owner = SubElement(elem, 'Owner')
+    SubElement(elem_owner, 'ID').text = owner
+    SubElement(elem_owner, 'DisplayName').text = owner
+    acl_list_elem = SubElement(elem, 'AccessControlList')
+    elem_grant = SubElement(acl_list_elem, 'Grant')
+    elem_grant.append(grantee)
+    SubElement(elem_grant, 'Permission').text = permission
+    return tostring(elem)
+
+
+class TestSwift3S3Acl(Swift3TestCase):
+
+    def setUp(self):
+        super(TestSwift3S3Acl, self).setUp()
+
+        CONF.s3_acl = True
+
+        # TEST method is used to resolve a tenant name
+        self.swift.register('TEST', '/v1/AUTH_test', swob.HTTPMethodNotAllowed,
+                            {}, None)
+        self.swift.register('TEST', '/v1/AUTH_X', swob.HTTPMethodNotAllowed,
+                            {}, None)
+
+        self.swift.register('HEAD', '/v1/AUTH_test/bucket', swob.HTTPNoContent,
+                            encode_acl('container',
+                                       ACLPrivate(Owner('test:tester',
+                                                        'test:tester'))),
+                            None)
+        self.swift.register('HEAD', '/v1/AUTH_test/bucket/object',
+                            swob.HTTPOk,
+                            encode_acl('object',
+                                       ACLPrivate(Owner('test:tester',
+                                                        'test:tester'))),
+                            None)
+
+        self.swift.register('PUT', '/v1/AUTH_test/bucket',
+                            swob.HTTPCreated, {}, None)
+        self.swift.register('PUT', '/v1/AUTH_test/bucket/object',
+                            swob.HTTPCreated, {}, None)
+        self.swift.register('POST', '/v1/AUTH_test/bucket/object',
+                            swob.HTTPAccepted, {}, None)
+
+    def tearDown(self):
+        CONF.s3_acl = False
+
+    def test_bucket_acl_PUT_with_other_owner(self):
+        req = Request.blank('/bucket?acl',
+                            environ={'REQUEST_METHOD': 'PUT'},
+                            headers={'Authorization': 'AWS test:tester:hmac'},
+                            body=tostring(
+                                ACLPrivate(
+                                    Owner(id='test:other',
+                                          name='test:other')).elem()))
+        status, headers, body = self.call_swift3(req)
+        self.assertEquals(self._get_error_code(body), 'AccessDenied')
+
+    def test_object_acl_PUT_xml_error(self):
+        req = Request.blank('/bucket/object?acl',
+                            environ={'REQUEST_METHOD': 'PUT'},
+                            headers={'Authorization': 'AWS test:tester:hmac'},
+                            body="invalid xml")
+        status, headers, body = self.call_swift3(req)
+        self.assertEquals(self._get_error_code(body), 'MalformedACLError')
+
+    def test_canned_acl_private(self):
+        req = Request.blank('/bucket/object?acl',
+                            environ={'REQUEST_METHOD': 'PUT'},
+                            headers={'Authorization': 'AWS test:tester:hmac',
+                                     'x-amz-acl': 'private'})
+        status, headers, body = self.call_swift3(req)
+        self.assertEquals(status.split()[0], '200')
+
+    def test_canned_acl_public_read(self):
+        req = Request.blank('/bucket/object?acl',
+                            environ={'REQUEST_METHOD': 'PUT'},
+                            headers={'Authorization': 'AWS test:tester:hmac',
+                                     'x-amz-acl': 'public-read'})
+        status, headers, body = self.call_swift3(req)
+        self.assertEquals(status.split()[0], '200')
+
+    def test_canned_acl_public_read_write(self):
+        req = Request.blank('/bucket/object?acl',
+                            environ={'REQUEST_METHOD': 'PUT'},
+                            headers={'Authorization': 'AWS test:tester:hmac',
+                                     'x-amz-acl': 'public-read-write'})
+        status, headers, body = self.call_swift3(req)
+        self.assertEquals(status.split()[0], '200')
+
+    def test_canned_acl_authenticated_read(self):
+        req = Request.blank('/bucket/object?acl',
+                            environ={'REQUEST_METHOD': 'PUT'},
+                            headers={'Authorization': 'AWS test:tester:hmac',
+                                     'x-amz-acl': 'authenticated-read'})
+        status, headers, body = self.call_swift3(req)
+        self.assertEquals(status.split()[0], '200')
+
+    def test_canned_acl_bucket_owner_read(self):
+        req = Request.blank('/bucket/object?acl',
+                            environ={'REQUEST_METHOD': 'PUT'},
+                            headers={'Authorization': 'AWS test:tester:hmac',
+                                     'x-amz-acl': 'bucket-owner-read'})
+        status, headers, body = self.call_swift3(req)
+        self.assertEquals(status.split()[0], '200')
+
+    def test_canned_acl_bucket_owner_full_control(self):
+        req = Request.blank('/bucket/object?acl',
+                            environ={'REQUEST_METHOD': 'PUT'},
+                            headers={'Authorization': 'AWS test:tester:hmac',
+                                     'x-amz-acl': 'bucket-owner-full-control'})
+        status, headers, body = self.call_swift3(req)
+        self.assertEquals(status.split()[0], '200')
+
+    def test_invalid_canned_acl(self):
+        req = Request.blank('/bucket/object?acl',
+                            environ={'REQUEST_METHOD': 'PUT'},
+                            headers={'Authorization': 'AWS test:tester:hmac',
+                                     'x-amz-acl': 'invalid'})
+        status, headers, body = self.call_swift3(req)
+        self.assertEquals(self._get_error_code(body), 'InvalidRequest')
+
+    def _test_grant_header(self, permission):
+        req = Request.blank('/bucket/object?acl',
+                            environ={'REQUEST_METHOD': 'PUT'},
+                            headers={'Authorization': 'AWS test:tester:hmac',
+                                     'x-amz-grant-' + permission:
+                                     'id=test:tester'})
+        return self.call_swift3(req)
+
+    def test_grant_read(self):
+        status, headers, body = self._test_grant_header('read')
+        self.assertEquals(status.split()[0], '200')
+
+    def test_grant_write(self):
+        status, headers, body = self._test_grant_header('write')
+        self.assertEquals(status.split()[0], '200')
+
+    def test_grant_read_acp(self):
+        status, headers, body = self._test_grant_header('read-acp')
+        self.assertEquals(status.split()[0], '200')
+
+    def test_grant_write_acp(self):
+        status, headers, body = self._test_grant_header('write-acp')
+        self.assertEquals(status.split()[0], '200')
+
+    def test_grant_full_control(self):
+        status, headers, body = self._test_grant_header('full-control')
+        self.assertEquals(status.split()[0], '200')
+
+    def test_grant_invalid_permission(self):
+        status, headers, body = self._test_grant_header('invalid')
+        self.assertEquals(self._get_error_code(body), 'MissingSecurityHeader')
+
+    def test_grant_with_both_header_and_xml(self):
+        req = Request.blank('/bucket/object?acl',
+                            environ={'REQUEST_METHOD': 'PUT'},
+                            headers={'Authorization': 'AWS test:tester:hmac',
+                                     'x-amz-grant-full-control':
+                                     'id=test:tester'},
+                            body=tostring(
+                                ACLPrivate(
+                                    Owner(id='test:tester',
+                                          name='test:tester')).elem()))
+        status, headers, body = self.call_swift3(req)
+        self.assertEquals(self._get_error_code(body), 'UnexpectedContent')
+
+    def test_grant_with_both_header_and_canned_acl(self):
+        req = Request.blank('/bucket/object?acl',
+                            environ={'REQUEST_METHOD': 'PUT'},
+                            headers={'Authorization': 'AWS test:tester:hmac',
+                                     'x-amz-grant-full-control':
+                                     'id=test:tester',
+                                     'x-amz-acl': 'public-read'})
+        status, headers, body = self.call_swift3(req)
+        self.assertEquals(self._get_error_code(body), 'InvalidRequest')
+
+    def test_grant_email(self):
+        req = Request.blank('/bucket/object?acl',
+                            environ={'REQUEST_METHOD': 'PUT'},
+                            headers={'Authorization': 'AWS test:tester:hmac',
+                                     'x-amz-grant-read': 'emailAddress=a@b.c'})
+        status, headers, body = self.call_swift3(req)
+        self.assertEquals(self._get_error_code(body), 'NotImplemented')
+
+    def test_grant_email_xml(self):
+        grantee = Element('Grantee', nsmap={'xsi': XMLNS_XSI})
+        grantee.set('{%s}type' % XMLNS_XSI, 'AmazonCustomerByEmail')
+        SubElement(grantee, 'EmailAddress').text = 'Grantees@email.com'
+        xml = _make_xml(grantee=grantee)
+        req = Request.blank('/bucket/object?acl',
+                            environ={'REQUEST_METHOD': 'PUT'},
+                            headers={'Authorization': 'AWS test:tester:hmac'},
+                            body=xml)
+        status, headers, body = self.call_swift3(req)
+        self.assertEquals(self._get_error_code(body), 'NotImplemented')
+
+    def test_grant_invalid_group_xml(self):
+        grantee = Element('Grantee', nsmap={'xsi': XMLNS_XSI})
+        grantee.set('{%s}type' % XMLNS_XSI, 'Invalid')
+        xml = _make_xml(grantee=grantee)
+        req = Request.blank('/bucket/object?acl',
+                            environ={'REQUEST_METHOD': 'PUT'},
+                            headers={'Authorization': 'AWS test:tester:hmac'},
+                            body=xml)
+        status, headers, body = self.call_swift3(req)
+        self.assertEquals(self._get_error_code(body), 'MalformedACLError')
+
+    def test_grant_authenticated_users(self):
+        req = Request.blank('/bucket/object?acl',
+                            environ={'REQUEST_METHOD': 'PUT'},
+                            headers={'Authorization': 'AWS test:tester:hmac',
+                                     'x-amz-grant-read':
+                                     'uri="http://acs.amazonaws.com/groups/'
+                                     'global/AuthenticatedUsers"'})
+        status, headers, body = self.call_swift3(req)
+        self.assertEquals(status.split()[0], '200')
+
+    def test_grant_all_users(self):
+        req = Request.blank('/bucket/object?acl',
+                            environ={'REQUEST_METHOD': 'PUT'},
+                            headers={'Authorization': 'AWS test:tester:hmac',
+                                     'x-amz-grant-read':
+                                     'uri="http://acs.amazonaws.com/groups/'
+                                     'global/AllUsers"'})
+        status, headers, body = self.call_swift3(req)
+        self.assertEquals(status.split()[0], '200')
+
+    def test_grant_invalid_uri(self):
+        req = Request.blank('/bucket/object?acl',
+                            environ={'REQUEST_METHOD': 'PUT'},
+                            headers={'Authorization': 'AWS test:tester:hmac',
+                                     'x-amz-grant-read':
+                                     'uri="http://localhost/"'})
+        status, headers, body = self.call_swift3(req)
+        self.assertEquals(self._get_error_code(body), 'InvalidArgument')
+
+    def test_grant_invalid_uri_xml(self):
+        grantee = Element('Grantee', nsmap={'xsi': XMLNS_XSI})
+        grantee.set('{%s}type' % XMLNS_XSI, 'Group')
+        SubElement(grantee, 'URI').text = 'invalid'
+        xml = _make_xml(grantee)
+
+        req = Request.blank('/bucket/object?acl',
+                            environ={'REQUEST_METHOD': 'PUT'},
+                            headers={'Authorization': 'AWS test:tester:hmac'},
+                            body=xml)
+        status, headers, body = self.call_swift3(req)
+        self.assertEquals(self._get_error_code(body), 'InvalidArgument')
+
+    def test_grant_invalid_target(self):
+        req = Request.blank('/bucket/object?acl',
+                            environ={'REQUEST_METHOD': 'PUT'},
+                            headers={'Authorization': 'AWS test:tester:hmac',
+                                     'x-amz-grant-read': 'key=value'})
+        status, headers, body = self.call_swift3(req)
+        self.assertEquals(self._get_error_code(body), 'InvalidArgument')
+
+    def _test_bucket_acl_GET(self, owner, permission):
+        owner = Owner(id=owner, name=owner)
+        acl = _gen_test_acl(owner, permission)
+
+        self.swift.register('HEAD', '/v1/AUTH_test/bucket', swob.HTTPNoContent,
+                            encode_acl('container', acl),
+                            None)
+        req = Request.blank('/bucket?acl',
+                            environ={'REQUEST_METHOD': 'GET'},
+                            headers={'Authorization': 'AWS test:tester:hmac'})
+
+        return self.call_swift3(req)
+
+    def test_bucket_acl_GET_without_permission(self):
+        status, headers, body = self._test_bucket_acl_GET('test:other', None)
+        self.assertEquals(self._get_error_code(body), 'AccessDenied')
+
+    def test_bucket_GET_with_owner_permission(self):
+        status, headers, body = self._test_bucket_acl_GET('test:tester', None)
+        self.assertEquals(status.split()[0], '200')
+
+    def _test_bucket_acl_PUT(self, owner, permission):
+        owner = Owner(id=owner, name=owner)
+        acl = _gen_test_acl(owner, permission)
+
+        req = Request.blank('/bucket?acl',
+                            environ={'REQUEST_METHOD': 'PUT'},
+                            headers={'Authorization': 'AWS test:tester:hmac'},
+                            body=tostring(acl.elem()))
+
+        return self.call_swift3(req)
+
+    def test_bucket_acl_PUT_without_permission(self):
+        status, headers, body = self._test_bucket_acl_PUT('test:other', None)
+        self.assertEquals(self._get_error_code(body), 'AccessDenied')
+
+    def test_bucket_acl_PUT_with_owner_permission(self):
+        status, headers, body = self._test_bucket_acl_PUT('test:tester', None)
+        self.assertEquals(status.split()[0], '200')
+
+    def _test_object_acl_GET(self, owner, permission):
+        owner = Owner(id=owner, name=owner)
+        acl = _gen_test_acl(owner, permission)
+
+        self.swift.register('HEAD', '/v1/AUTH_test/bucket/object',
+                            swob.HTTPOk,
+                            encode_acl('object', acl),
+                            None)
+        req = Request.blank('/bucket/object?acl',
+                            environ={'REQUEST_METHOD': 'GET'},
+                            headers={'Authorization': 'AWS test:tester:hmac'})
+
+        return self.call_swift3(req)
+
+    def test_object_acl_GET_without_permission(self):
+        status, headers, body = self._test_object_acl_GET('test:other', None)
+        self.assertEquals(self._get_error_code(body), 'AccessDenied')
+
+    def test_object_acl_GET_with_owner_permission(self):
+        status, headers, body = self._test_object_acl_GET('test:tester', None)
+        self.assertEquals(status.split()[0], '200')
+
+    def _test_object_acl_PUT(self, owner, permission):
+        owner = Owner(id=owner, name=owner)
+        acl = _gen_test_acl(owner, permission)
+
+        self.swift.register('HEAD', '/v1/AUTH_test/bucket/object',
+                            swob.HTTPOk,
+                            encode_acl('object', acl),
+                            None)
+        req = Request.blank('/bucket/object?acl',
+                            environ={'REQUEST_METHOD': 'PUT'},
+                            headers={'Authorization': 'AWS test:tester:hmac'},
+                            body=tostring(acl.elem()))
+
+        return self.call_swift3(req)
+
+    def test_object_acl_PUT_without_permission(self):
+        status, headers, body = self._test_object_acl_PUT('test:other', None)
+        self.assertEquals(self._get_error_code(body), 'AccessDenied')
+
+    def test_object_acl_PUT_with_owner_permission(self):
+        status, headers, body = self._test_object_acl_PUT('test:tester', None)
+        self.assertEquals(status.split()[0], '200')
+
+if __name__ == '__main__':
+    unittest.main()

swift3/test/unit/test_subresource.py

@@ -14,14 +14,23 @@
 # limitations under the License.
 
 import unittest
-
+from simplejson import dumps, loads
 from swift3.response import AccessDenied
 from swift3.subresource import User, AuthenticatedUsers, AllUsers, \
     ACLPrivate, ACLPublicRead, ACLPublicReadWrite, ACLAuthenticatedRead, \
-    ACLBucketOwnerRead, ACLBucketOwnerFullControl, Owner, ACL
+    ACLBucketOwnerRead, ACLBucketOwnerFullControl, Owner, ACL, encode_acl, \
+    decode_acl
+from swift3.utils import CONF, sysmeta_header
 
 
 class TestSwift3Subresource(unittest.TestCase):
+
+    def setUp(self):
+        CONF.s3_acl = True
+
+    def tearDown(self):
+        CONF.s3_acl = False
+
     def test_acl_canonical_user(self):
         grantee = User('test:tester')
 
@@ -173,6 +182,84 @@ class TestSwift3Subresource(unittest.TestCase):
         self.assertFalse(self.check_permission(acl, 'test:tester2',
                                                'WRITE_ACP'))
 
+    def test_decode_acl_container(self):
+        access_control_policy = \
+            {'Owner': 'test:tester',
+             'Grant': [{'Permission': 'FULL_CONTROL',
+                        'Grantee': 'test:tester'}]}
+        headers = {sysmeta_header('container', 'acl'):
+                   dumps(access_control_policy)}
+        acl = decode_acl('container', headers)
+
+        self.assertEqual(type(acl), ACL)
+        self.assertEqual(acl.owner.id, 'test:tester')
+        self.assertEqual(len(acl.grants), 1)
+        self.assertEqual(str(acl.grants[0].grantee), 'test:tester')
+        self.assertEqual(acl.grants[0].permission, 'FULL_CONTROL')
+
+    def test_decode_acl_object(self):
+        access_control_policy = \
+            {'Owner': 'test:tester',
+             'Grant': [{'Permission': 'FULL_CONTROL',
+                        'Grantee': 'test:tester'}]}
+        headers = {sysmeta_header('object', 'acl'):
+                   dumps(access_control_policy)}
+        acl = decode_acl('object', headers)
+
+        self.assertEqual(type(acl), ACL)
+        self.assertEqual(acl.owner.id, 'test:tester')
+        self.assertEqual(len(acl.grants), 1)
+        self.assertEqual(str(acl.grants[0].grantee), 'test:tester')
+        self.assertEqual(acl.grants[0].permission, 'FULL_CONTROL')
+
+    def test_decode_acl_undefined(self):
+        headers = {}
+        acl = decode_acl('container', headers)
+
+        self.assertEqual(type(acl), ACL)
+        self.assertEqual(None, acl.owner.id)
+        self.assertEqual(len(acl.grants), 0)
+
+    def test_encode_acl_container(self):
+        acl = ACLPrivate(Owner(id='test:tester',
+                               name='test:tester'))
+        acp = encode_acl('container', acl)
+        header_value = loads(acp[sysmeta_header('container', 'acl')])
+
+        self.assertTrue('Owner' in header_value)
+        self.assertTrue('Grant' in header_value)
+        self.assertEqual('test:tester', header_value['Owner'])
+        self.assertEqual(len(header_value['Grant']), 1)
+
+    def test_encode_acl_object(self):
+        acl = ACLPrivate(Owner(id='test:tester',
+                               name='test:tester'))
+        acp = encode_acl('object', acl)
+        header_value = loads(acp[sysmeta_header('object', 'acl')])
+
+        self.assertTrue('Owner' in header_value)
+        self.assertTrue('Grant' in header_value)
+        self.assertEqual('test:tester', header_value['Owner'])
+        self.assertEqual(len(header_value['Grant']), 1)
+
+    def test_encode_acl_many_grant(self):
+        headers = {}
+        users = []
+        for i in range(0, 99):
+            users.append('id=test:tester%s' % str(i))
+        users = ','.join(users)
+        headers['x-amz-grant-read'] = users
+        acl = ACL.from_headers(headers, Owner('test:tester', 'test:tester'))
+        acp = encode_acl('container', acl)
+
+        header_value = acp[sysmeta_header('container', 'acl')]
+        header_value = loads(header_value)
+
+        self.assertTrue('Owner' in header_value)
+        self.assertTrue('Grant' in header_value)
+        self.assertEqual('test:tester', header_value['Owner'])
+        self.assertEqual(len(header_value['Grant']), 99)
+
 
 if __name__ == '__main__':
     unittest.main()

swift3/utils.py

@@ -25,6 +25,23 @@ from swift3.cfg import CONF
 LOGGER = get_logger(CONF, log_route='swift3')
 
 
+def sysmeta_prefix(resource):
+    """
+    Returns the system metadata prefix for given resource type.
+    """
+    if resource == 'object':
+        return 'x-object-sysmeta-swift3-'
+    else:
+        return 'x-container-sysmeta-swift3-'
+
+
+def sysmeta_header(resource, name):
+    """
+    Returns the system metadata header for given resource type and name.
+    """
+    return sysmeta_prefix(resource) + name
+
+
 def camel_to_snake(camel):
     return re.sub('(.)([A-Z])', r'\1_\2', camel).lower()