diff --git a/etc/proxy-server.conf-sample b/etc/proxy-server.conf-sample
index 2c0a9430..200c8d07 100644
--- a/etc/proxy-server.conf-sample
+++ b/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 =
diff --git a/swift3/cfg.py b/swift3/cfg.py
index bcce2f31..811738c8 100644
--- a/swift3/cfg.py
+++ b/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': '',
 })
diff --git a/swift3/controllers/__init__.py b/swift3/controllers/__init__.py
index 22f83d7c..1a1b3eb6 100644
--- a/swift3/controllers/__init__.py
+++ b/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',
diff --git a/swift3/controllers/s3_acl.py b/swift3/controllers/s3_acl.py
new file mode 100644
index 00000000..2d4d3c51
--- /dev/null
+++ b/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()
diff --git a/swift3/exception.py b/swift3/exception.py
index feea6a56..8231ac7e 100644
--- a/swift3/exception.py
+++ b/swift3/exception.py
@@ -28,3 +28,7 @@ class BadSwiftRequest(S3Exception):
 
 class ACLError(S3Exception):
     pass
+
+
+class InvalidSubresource(S3Exception):
+    pass
diff --git a/swift3/middleware.py b/swift3/middleware.py
index 55195e92..3f84bd6d 100644
--- a/swift3/middleware.py
+++ b/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
diff --git a/swift3/request.py b/swift3/request.py
index 59c50e9c..35b99a62 100644
--- a/swift3/request.py
+++ b/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)
diff --git a/swift3/response.py b/swift3/response.py
index 5fe204d6..abe9a32a 100644
--- a/swift3/response.py
+++ b/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):
diff --git a/swift3/subresource.py b/swift3/subresource.py
index a21b3b40..ef60b89d 100644
--- a/swift3/subresource.py
+++ b/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))
diff --git a/swift3/test/unit/__init__.py b/swift3/test/unit/__init__.py
index 15d39e25..df7ee5a8 100644
--- a/swift3/test/unit/__init__.py
+++ b/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):
diff --git a/swift3/test/unit/helpers.py b/swift3/test/unit/helpers.py
index fa7a6762..308b7e66 100644
--- a/swift3/test/unit/helpers.py
+++ b/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)
diff --git a/swift3/test/unit/test_s3_acl.py b/swift3/test/unit/test_s3_acl.py
new file mode 100644
index 00000000..b815393d
--- /dev/null
+++ b/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()
diff --git a/swift3/test/unit/test_subresource.py b/swift3/test/unit/test_subresource.py
index 955c35af..65d558ff 100644
--- a/swift3/test/unit/test_subresource.py
+++ b/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()
diff --git a/swift3/utils.py b/swift3/utils.py
index 16c0010d..0b0e9ef0 100644
--- a/swift3/utils.py
+++ b/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()