Privileged acct ACL header, new ACL syntax, TempAuth impl.

* Introduce a new privileged account header: X-Account-Access-Control
 * Introduce JSON-based version 2 ACL syntax -- see below for discussion
 * Implement account ACL authorization in TempAuth

X-Account-Access-Control Header
-------------------------------

Accounts now have a new privileged header to represent ACLs or any other
form of account-level access control.  The value of the header is an opaque
string to be interpreted by the auth system, but it must be a JSON-encoded
dictionary.  A reference implementation is given in TempAuth, with the
knowledge that historically other auth systems often use TempAuth as a
starting point.

The reference implementation describes three levels of account access:
"admin", "read-write", and "read-only".  Adding new access control
features in a future patch (e.g. "write-only" account access) will
automatically be forward- and backward-compatible, due to the JSON
dictionary header format.

The privileged X-Account-Access-Control header may only be read or written
by a user with "swift_owner" status, traditionally the account owner but
now also any user on the "admin" ACL.

Access Levels:

Read-only access is intended to indicate to the auth system that this
list of identities can read everything (except privileged headers) in
the account.  Specifically, a user with read-only account access can get
a list of containers in the account, list the contents of any container,
retrieve any object, and see the (non-privileged) headers of the
account, any container, or any object.

Read-write access is intended to indicate to the auth system that this
list of identities can read or write (or create) any container.  A user
with read-write account access can create new containers, set any
unprivileged container headers, overwrite objects, delete containers,
etc.  A read-write user can NOT set account headers (or perform any
PUT/POST/DELETE requests on the account).

Admin access is intended to indicate to the auth system that this list of
identities has "swift_owner" privileges.  A user with admin account access
can do anything the account owner can, including setting account headers
and any privileged headers -- and thus changing the value of
X-Account-Access-Control and thereby granting read-only, read-write, or
admin access to other users.

The auth system is responsible for making decisions based on this header,
if it chooses to support its use.  Therefore the above access level
descriptions are necessarily advisory only for other auth systems.

When setting the value of the header, callers are urged to use the new
format_acl() method, described below.

New ACL Format
--------------

The account ACLs introduce a new format for ACLs, rather than reusing the
existing format from X-Container-Read/X-Container-Write.  There are several
reasons for this:
 * Container ACL format does not support Unicode
 * Container ACLs have a different structure than account ACLs
  + account ACLs have no concept of referrers or rlistings
  + accounts have additional "admin" access level
  + account access levels are structured as admin > rw > ro, which seems more
    appropriate for how people access accounts, rather than reusing
    container ACLs' orthogonal read and write access

In addition, the container ACL syntax is a bit arbitrary and highly custom,
so instead of parsing additional custom syntax, I'd rather propose a next
version and introduce a means for migration.  The V2 ACL syntax has the
following benefits:
 * JSON is a well-known standard syntax with parsers in all languages
 * no artificial value restrictions (you can grant access to a user named
    ".rlistings" if you want)
 * forward and backward compatibility: you may have extraneous keys, but
    your attempt to parse the header won't raise an exception

I've introduced hooks in parse_acl and format_acl which currently default
to the old V1 syntax but tolerate the V2 syntax and can easily be flipped
to default to V2.  I'm not changing the default or adding code to rewrite
V1 ACLs to V2, because this patch has suffered a lot of scope creep already,
but this seems like a sensible milestone in the migration.

TempAuth Account ACL Implementation
-----------------------------------

As stated above, core Swift is responsible for privileging the
X-Account-Access-Control header (making it only accessible to swift_owners),
for translating it to -sysmeta-* headers to trigger persistence by the
account server, and for including the header in the responses to requests
by privileged users.  Core Swift puts no expectation on the *content* of
this header.  Auth systems (including TempAuth) are responsible for
defining the content of the header and taking action based on it.

In addition to the changes described above, this patch defines a format
to be used by TempAuth for these headers in the common.middleware.acl
module, in the methods format_v2_acl() and parse_v2_acl().  This patch
also teaches TempAuth to take action based on the header contents.  TempAuth
now sets swift_owner=True if the user is on the Admin ACL, authorizes
GET/HEAD/OPTIONS requests if the user is on any ACL, authorizes
PUT/POST/DELETE requests if the user is on the admin or read-write ACL, etc.

Note that the action of setting swift_owner=True triggers core Swift to
add or strip the privileged headers from the responses.  Core Swift (not
the auth system) is responsible for that.

DocImpact: Documentation for the new ACL usage and format appears in
summary form in doc/source/overview_auth.rst, and in more detail in
swift/common/middleware/tempauth.py in the TempAuth class docstring.
I leave it to the Swift doc team to determine whether more is needed.

Change-Id: I836a99eaaa6bb0e92dc03e1ca46a474522e6e826

doc/source/overview_auth.rst

@@ -29,16 +29,72 @@ validated. For a valid token, the auth system responds with an overall
 expiration in seconds from now. Swift will cache the token up to the expiration
 time.
 
-The included TempAuth also has the concept of admin and non-admin users within
+The included TempAuth also has the concept of admin and non-admin users
-an account. Admin users can do anything within the account. Non-admin users can
+within an account.  Admin users can do anything within the account.
-only perform operations per container based on the container's X-Container-Read
+Non-admin users can only perform operations per container based on the
-and X-Container-Write ACLs. For more information on ACLs, see
+container's X-Container-Read and X-Container-Write ACLs.  Container ACLs
-:mod:`swift.common.middleware.acl`.
+use the "V1" ACL syntax, which looks like this:
+``name1, name2, .r:referrer1.com, .r:-bad.referrer1.com, .rlistings``
+For more information on ACLs, see :mod:`swift.common.middleware.acl`.
 
 Additionally, if the auth system sets the request environ's swift_owner key to
 True, the proxy will return additional header information in some requests,
 such as the X-Container-Sync-Key for a container GET or HEAD.
 
+In addition to container ACLs, TempAuth allows account-level ACLs.  Any auth
+system may use the special header ``X-Account-Access-Control`` to specify
+account-level ACLs in a format specific to that auth system.  (Following the
+TempAuth format is strongly recommended.)  These headers are visible and
+settable only by account owners (those for whom ``swift_owner`` is true).
+Behavior of account ACLs is auth-system-dependent.  In the case of TempAuth,
+if an authenticated user has membership in a group which is listed in the
+ACL, then the user is allowed the access level of that ACL.
+
+Account ACLs use the "V2" ACL syntax, which is a JSON dictionary with keys
+named "admin", "read-write", and "read-only".  (Note the case sensitivity.)
+An example value for the ``X-Account-Access-Control`` header looks like this:
+``{"admin":["a","b"],"read-only":["c"]}``  Keys may be absent (as shown).
+The recommended way to generate ACL strings is as follows::
+
+  from swift.common.middleware.acl import format_acl
+  acl_data = { 'admin': ['alice'], 'read-write': ['bob', 'carol'] }
+  acl_string = format_acl(version=2, acl_dict=acl_data)
+
+Using the :func:`format_acl` method will ensure
+that JSON is encoded as ASCII (using e.g. '\u1234' for Unicode).  While
+it's permissible to manually send ``curl`` commands containing
+``X-Account-Access-Control`` headers, you should exercise caution when
+doing so, due to the potential for human error.
+
+Within the JSON dictionary stored in ``X-Account-Access-Control``, the keys
+have the following meanings:
+
+============   ==============================================================
+Access Level   Description
+============   ==============================================================
+read-only      These identities can read *everything* (except privileged
+               headers) in the account.  Specifically, a user with read-only
+	       account access can get a list of containers in the account,
+	       list the contents of any container, retrieve any object, and
+	       see the (non-privileged) headers of the account, any
+	       container, or any object.
+read-write     These identities can read or write (or create) any container.
+               A user with read-write account access can create new
+	       containers, set any unprivileged container headers, overwrite
+	       objects, delete containers, etc.  A read-write user can NOT
+	       set account headers (or perform any PUT/POST/DELETE requests
+	       on the account).
+admin          These identities have "swift_owner" privileges.  A user with
+               admin account access can do anything the account owner can,
+	       including setting account headers and any privileged headers
+	       -- and thus granting read-only, read-write, or admin access
+	       to other users.
+============   ==============================================================
+
+
+For more details, see :mod:`swift.common.middleware.tempauth`.  For details
+on the ACL format, see :mod:`swift.common.middleware.acl`.
+
 Users with the special group ``.reseller_admin`` can operate on any account.
 For an example usage please see :mod:`swift.common.middleware.tempauth`.
 If a request is coming from a reseller the auth system sets the request environ

etc/proxy-server.conf-sample

@@ -191,7 +191,7 @@ use = egg:swift#proxy
 # These are the headers whose values will only be shown to swift_owners. The
 # exact definition of a swift_owner is up to the auth system in use, but
 # usually indicates administrative responsibilities.
-# swift_owner_headers = x-container-read, x-container-write, x-container-sync-key, x-container-sync-to, x-account-meta-temp-url-key, x-account-meta-temp-url-key-2
+# swift_owner_headers = x-container-read, x-container-write, x-container-sync-key, x-container-sync-to, x-account-meta-temp-url-key, x-account-meta-temp-url-key-2, x-account-access-control
 
 
 [filter:tempauth]

swift/common/middleware/acl.py

@@ -13,7 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from swift.common.utils import urlparse
+from swift.common.utils import urlparse, json
 
 
 def clean_acl(name, value):
@@ -89,35 +89,98 @@ def clean_acl(name, value):
     values = []
     for raw_value in value.split(','):
         raw_value = raw_value.strip()
-        if raw_value:
+        if not raw_value:
-            if ':' not in raw_value:
+            continue
-                values.append(raw_value)
+        if ':' not in raw_value:
-            else:
+            values.append(raw_value)
-                first, second = (v.strip() for v in raw_value.split(':', 1))
+            continue
-                if not first or first[0] != '.':
+        first, second = (v.strip() for v in raw_value.split(':', 1))
-                    values.append(raw_value)
+        if not first or first[0] != '.':
-                elif first in ('.r', '.ref', '.referer', '.referrer'):
+            values.append(raw_value)
-                    if 'write' in name:
+        elif first in ('.r', '.ref', '.referer', '.referrer'):
-                        raise ValueError('Referrers not allowed in write ACL: '
+            if 'write' in name:
-                                         '%s' % repr(raw_value))
+                raise ValueError('Referrers not allowed in write ACL: '
-                    negate = False
+                                 '%s' % repr(raw_value))
-                    if second and second[0] == '-':
+            negate = False
-                        negate = True
+            if second and second[0] == '-':
-                        second = second[1:].strip()
+                negate = True
-                    if second and second != '*' and second[0] == '*':
+                second = second[1:].strip()
-                        second = second[1:].strip()
+            if second and second != '*' and second[0] == '*':
-                    if not second or second == '.':
+                second = second[1:].strip()
-                        raise ValueError('No host/domain value after referrer '
+            if not second or second == '.':
-                                         'designation in ACL: %s' %
+                raise ValueError('No host/domain value after referrer '
-                                         repr(raw_value))
+                                 'designation in ACL: %s' % repr(raw_value))
-                    values.append('.r:%s%s' % ('-' if negate else '', second))
+            values.append('.r:%s%s' % ('-' if negate else '', second))
-                else:
+        else:
-                    raise ValueError('Unknown designator %s in ACL: %s' %
+            raise ValueError('Unknown designator %s in ACL: %s' %
-                                     (repr(first), repr(raw_value)))
+                             (repr(first), repr(raw_value)))
     return ','.join(values)
 
 
-def parse_acl(acl_string):
+def format_acl_v1(groups=None, referrers=None, header_name=None):
+    """
+    Returns a standard Swift ACL string for the given inputs.
+
+    Caller is responsible for ensuring that :referrers: parameter is only given
+    if the ACL is being generated for X-Container-Read.  (X-Container-Write
+    and the account ACL headers don't support referrers.)
+
+    :param groups: a list of groups (and/or members in most auth systems) to
+                   grant access
+    :param referrers: a list of referrer designations (without the leading .r:)
+    :param header_name: (optional) header name of the ACL we're preparing, for
+                        clean_acl; if None, returned ACL won't be cleaned
+    :returns: a Swift ACL string for use in X-Container-{Read,Write},
+              X-Account-Access-Control, etc.
+    """
+    groups, referrers = groups or [], referrers or []
+    referrers = ['.r:%s' % r for r in referrers]
+    result = ','.join(groups + referrers)
+    return (clean_acl(header_name, result) if header_name else result)
+
+
+def format_acl_v2(acl_dict):
+    """
+    Returns a version-2 Swift ACL JSON string.
+
+    HTTP headers for Version 2 ACLs have the following form:
+      Header-Name: {"arbitrary":"json","encoded":"string"}
+
+    JSON will be forced ASCII (containing six-char \uNNNN sequences rather
+    than UTF-8; UTF-8 is valid JSON but clients vary in their support for
+    UTF-8 headers), and without extraneous whitespace.
+
+    Advantages over V1: forward compatibility (new keys don't cause parsing
+    exceptions); Unicode support; no reserved words (you can have a user
+    named .rlistings if you want).
+
+    :param acl_dict: dict of arbitrary data to put in the ACL; see specific
+                     auth systems such as tempauth for supported values
+    :returns: a JSON string which encodes the ACL
+    """
+    return json.dumps(acl_dict, ensure_ascii=True, separators=(',', ':'),
+                      sort_keys=True)
+
+
+def format_acl(version=1, **kwargs):
+    """
+    Compatibility wrapper to help migrate ACL syntax from version 1 to 2.
+    Delegates to the appropriate version-specific format_acl method, defaulting
+    to version 1 for backward compatibility.
+
+    :param kwargs: keyword args appropriate for the selected ACL syntax version
+                   (see :func:`format_acl_v1` or :func:`format_acl_v2`)
+    """
+    if version == 1:
+        return format_acl_v1(
+            groups=kwargs.get('groups'), referrers=kwargs.get('referrers'),
+            header_name=kwargs.get('header_name'))
+    elif version == 2:
+        return format_acl_v2(kwargs.get('acl_dict'))
+    raise ValueError("Invalid ACL version: %r" % version)
+
+
+def parse_acl_v1(acl_string):
     """
     Parses a standard Swift ACL string into a referrers list and groups list.
 
@@ -139,6 +202,45 @@ def parse_acl(acl_string):
     return referrers, groups
 
 
+def parse_acl_v2(data):
+    """
+    Parses a version-2 Swift ACL string and returns a dict of ACL info.
+
+    :param data: string containing the ACL data in JSON format
+    :returns: A dict containing ACL info, e.g.:
+              {"groups": [...], "referrers": [...]}
+    :returns: None if data is None
+    :returns: empty dictionary if data does not parse as valid JSON
+    """
+    if data is None:
+        return None
+    try:
+        return json.loads(data)
+    except ValueError:
+        return {}
+
+
+def parse_acl(*args, **kwargs):
+    """
+    Compatibility wrapper to help migrate ACL syntax from version 1 to 2.
+    Delegates to the appropriate version-specific parse_acl method, attempting
+    to determine the version from the types of args/kwargs.
+
+    :param args: positional args for the selected ACL syntax version
+    :param kwargs: keyword args for the selected ACL syntax version
+                   (see :func:`parse_acl_v1` or :func:`parse_acl_v2`)
+    :returns: the return value of :func:`parse_acl_v1` or :func:`parse_acl_v2`
+    """
+    version = kwargs.pop('version', None)
+    if version in (1, None):
+        return parse_acl_v1(*args)
+    elif version == 2:
+        return parse_acl_v2(*args, **kwargs)
+    else:
+        raise ValueError('Unknown ACL version: parse_acl(%r, %r)' %
+                         (args, kwargs))
+
+
 def referrer_allowed(referrer, referrer_acl):
     """
     Returns True if the referrer should be allowed based on the referrer_acl
@@ -164,3 +266,29 @@ def referrer_allowed(referrer, referrer_acl):
                     (mhost[0] == '.' and rhost.endswith(mhost)):
                 allow = True
     return allow
+
+
+def acls_from_account_info(info):
+    """
+    Extract the account ACLs from the given account_info, and return the ACLs.
+
+    :param info: a dict of the form returned by get_account_info
+    :returns: None (no ACL system metadata is set), or a dict of the form::
+       {'admin': [...], 'read-write': [...], 'read-only': [...]}
+
+    :raises ValueError: on a syntactically invalid header
+    """
+    acl = parse_acl(
+        version=2, data=info.get('sysmeta', {}).get('core-access-control'))
+    if acl is None:
+        return None
+    admin_members = acl.get('admin', [])
+    readwrite_members = acl.get('read-write', [])
+    readonly_members = acl.get('read-only', [])
+    if not any((admin_members, readwrite_members, readonly_members)):
+        return None
+    return {
+        'admin': admin_members,
+        'read-write': readwrite_members,
+        'read-only': readonly_members,
+    }

swift/common/middleware/tempauth.py

@@ -26,9 +26,12 @@ from swift.common.swob import Response, Request
 from swift.common.swob import HTTPBadRequest, HTTPForbidden, HTTPNotFound, \
     HTTPUnauthorized
 
-from swift.common.middleware.acl import clean_acl, parse_acl, referrer_allowed
+from swift.common.request_helpers import get_sys_meta_prefix
+from swift.common.middleware.acl import (
+    clean_acl, parse_acl, referrer_allowed, acls_from_account_info)
 from swift.common.utils import cache_from_env, get_logger, \
     split_path, config_true_value, register_swift_info
+from swift.proxy.controllers.base import get_account_info
 
 
 class TempAuth(object):
@@ -61,8 +64,45 @@ class TempAuth(object):
 
     See the proxy-server.conf-sample for more information.
 
+    Account ACLs:
+        If a swift_owner issues a POST or PUT to the account, with the
+        X-Account-Access-Control header set in the request, then this may
+        allow certain types of access for additional users.
+
+        * Read-Only: Users with read-only access can list containers in the
+          account, list objects in any container, retrieve objects, and view
+          unprivileged account/container/object metadata.
+        * Read-Write: Users with read-write access can (in addition to the
+          read-only privileges) create objects, overwrite existing objects,
+          create new containers, and set unprivileged container/object
+          metadata.
+        * Admin: Users with admin access are swift_owners and can perform
+          any action, including viewing/setting privileged metadata (e.g.
+          changing account ACLs).
+
+    To generate headers for setting an account ACL::
+
+        from swift.common.middleware.acl import format_acl
+        acl_data = { 'admin': ['alice'], 'read-write': ['bob', 'carol'] }
+        header_value = format_acl(version=2, acl_dict=acl_data)
+
+    To generate a curl command line from the above::
+
+        token=...
+        storage_url=...
+        python -c '
+          from swift.common.middleware.acl import format_acl
+          acl_data = { 'admin': ['alice'], 'read-write': ['bob', 'carol'] }
+          headers = {'X-Account-Access-Control':
+                     format_acl(version=2, acl_dict=acl_data)}
+          header_str = ' '.join(["-H '%s: %s'" % (k, v)
+                                 for k, v in headers.items()])
+          print ('curl -D- -X POST -H "x-auth-token: $token" %s '
+                 '$storage_url' % header_str)
+        '
+
     :param app: The next WSGI app in the pipeline
-    :param conf: The dict of configuration values
+    :param conf: The dict of configuration values from the Paste config file
     """
 
     def __init__(self, app, conf):
@@ -249,6 +289,66 @@ class TempAuth(object):
 
         return groups
 
+    def account_acls(self, req):
+        """
+        Return a dict of ACL data from the account server via get_account_info.
+
+        Auth systems may define their own format, serialization, structure,
+        and capabilities implemented in the ACL headers and persisted in the
+        sysmeta data.  However, auth systems are strongly encouraged to be
+        interoperable with Tempauth.
+
+        Account ACLs are set and retrieved via the header
+           X-Account-Access-Control
+
+        For header format and syntax, see:
+         * :func:`swift.common.middleware.acl.parse_acl()`
+         * :func:`swift.common.middleware.acl.format_acl()`
+        """
+        info = get_account_info(req.environ, self.app, swift_source='TA')
+        try:
+            acls = acls_from_account_info(info)
+        except ValueError as e1:
+            self.logger.warn("Invalid ACL stored in metadata: %r" % e1)
+            return None
+        except NotImplementedError as e2:
+            self.logger.warn("ACL version exceeds middleware version: %r" % e2)
+            return None
+        return acls
+
+    def extract_acl_and_report_errors(self, req):
+        """
+        Return a user-readable string indicating the errors in the input ACL,
+        or None if there are no errors.
+        """
+        acl_header = 'x-account-access-control'
+        acl_data = req.headers.get(acl_header)
+        result = parse_acl(version=2, data=acl_data)
+        if (not result and acl_data not in ('', '{}')):
+            return 'Syntax error in input (%r)' % acl_data
+
+        tempauth_acl_keys = 'admin read-write read-only'.split()
+        for key in result:
+            # While it is possible to construct auth systems that collaborate
+            # on ACLs, TempAuth is not such an auth system.  At this point,
+            # it thinks it is authoritative.
+            if key not in tempauth_acl_keys:
+                return 'Key %r not recognized' % key
+
+        for key in tempauth_acl_keys:
+            if key not in result:
+                continue
+            if not isinstance(result[key], list):
+                return 'Value for key %r must be a list' % key
+            for grantee in result[key]:
+                if not isinstance(grantee, str):
+                    return 'Elements of %r list must be strings' % key
+
+        # Everything looks fine, no errors found
+        internal_hdr = get_sys_meta_prefix('account') + 'core-access-control'
+        req.headers[internal_hdr] = req.headers.pop(acl_header)
+        return None
+
     def authorize(self, req):
         """
         Returns None if the request is authorized to continue or a standard
@@ -256,7 +356,7 @@ class TempAuth(object):
         """
 
         try:
-            version, account, container, obj = req.split_path(1, 4, True)
+            _junk, account, container, obj = req.split_path(1, 4, True)
         except ValueError:
             self.logger.increment('errors')
             return HTTPNotFound(request=req)
@@ -267,6 +367,18 @@ class TempAuth(object):
                               % (account, self.reseller_prefix))
             return self.denied_response(req)
 
+        # At this point, TempAuth is convinced that it is authoritative.
+        # If you are sending an ACL header, it must be syntactically valid
+        # according to TempAuth's rules for ACL syntax.
+        acl_data = req.headers.get('x-account-access-control')
+        if acl_data is not None:
+            error = self.extract_acl_and_report_errors(req)
+            if error:
+                msg = 'X-Account-Access-Control invalid: %s\n\nInput: %s\n' % (
+                    error, acl_data)
+                headers = [('Content-Type', 'text/plain; charset=UTF-8')]
+                return HTTPBadRequest(request=req, headers=headers, body=msg)
+
         user_groups = (req.remote_user or '').split(',')
         account_user = user_groups[1] if len(user_groups) > 1 else None
 
@@ -314,6 +426,30 @@ class TempAuth(object):
                                   % (account_user, user_group))
                 return None
 
+        # Check for access via X-Account-Access-Control
+        acct_acls = self.account_acls(req)
+        if acct_acls:
+            # At least one account ACL is set in this account's sysmeta data,
+            # so we should see whether this user is authorized by the ACLs.
+            user_group_set = set(user_groups)
+            if user_group_set.intersection(acct_acls['admin']):
+                req.environ['swift_owner'] = True
+                self.logger.debug('User %s allowed by X-Account-Access-Control'
+                                  ' (admin)' % account_user)
+                return None
+            if (user_group_set.intersection(acct_acls['read-write']) and
+                    (container or req.method in ('GET', 'HEAD'))):
+                # The RW ACL allows all operations to containers/objects, but
+                # only GET/HEAD to accounts (and OPTIONS, above)
+                self.logger.debug('User %s allowed by X-Account-Access-Control'
+                                  ' (read-write)' % account_user)
+                return None
+            if (user_group_set.intersection(acct_acls['read-only']) and
+                    req.method in ('GET', 'HEAD')):
+                self.logger.debug('User %s allowed by X-Account-Access-Control'
+                                  ' (read-only)' % account_user)
+                return None
+
         return self.denied_response(req)
 
     def denied_response(self, req):
@@ -510,7 +646,7 @@ def filter_factory(global_conf, **local_conf):
     """Returns a WSGI filter app for use with paste.deploy."""
     conf = global_conf.copy()
     conf.update(local_conf)
-    register_swift_info('tempauth')
+    register_swift_info('tempauth', account_acls=True)
 
     def auth_filter(app):
         return TempAuth(app, conf)

swift/common/swob.py

@@ -288,6 +288,9 @@ class HeaderKeyDict(dict):
             self[key] = value
         return self[key]
 
+    def pop(self, key, default=None):
+        return dict.pop(self, key.title(), default)
+
 
 def _resp_status_property():
     """

swift/proxy/controllers/account.py

@@ -18,11 +18,13 @@ from urllib import unquote
 
 from swift.account.utils import account_listing_response
 from swift.common.request_helpers import get_listing_content_type
+from swift.common.middleware.acl import parse_acl, format_acl
 from swift.common.utils import public
 from swift.common.constraints import check_metadata, MAX_ACCOUNT_NAME_LENGTH
 from swift.common.http import HTTP_NOT_FOUND, HTTP_GONE
 from swift.proxy.controllers.base import Controller, clear_info_cache
 from swift.common.swob import HTTPBadRequest, HTTPMethodNotAllowed
+from swift.common.request_helpers import get_sys_meta_prefix
 
 
 class AccountController(Controller):
@@ -36,6 +38,16 @@ class AccountController(Controller):
             self.allowed_methods.remove('PUT')
             self.allowed_methods.remove('DELETE')
 
+    def add_acls_from_sys_metadata(self, resp):
+        if resp.environ['REQUEST_METHOD'] in ('HEAD', 'GET', 'PUT', 'POST'):
+            prefix = get_sys_meta_prefix('account') + 'core-'
+            name = 'access-control'
+            (extname, intname) = ('x-account-' + name, prefix + name)
+            acl_dict = parse_acl(version=2, data=resp.headers.pop(intname))
+            if acl_dict:  # treat empty dict as empty header
+                resp.headers[extname] = format_acl(
+                    version=2, acl_dict=acl_dict)
+
     def GETorHEAD(self, req):
         """Handler for HTTP GET/HEAD requests."""
         if len(self.account_name) > MAX_ACCOUNT_NAME_LENGTH:
@@ -54,10 +66,11 @@ class AccountController(Controller):
             elif self.app.account_autocreate:
                 resp = account_listing_response(self.account_name, req,
                                                 get_listing_content_type(req))
-        if not req.environ.get('swift_owner', False):
+        if req.environ.get('swift_owner'):
-            for key in self.app.swift_owner_headers:
+            self.add_acls_from_sys_metadata(resp)
-                if key in resp.headers:
+        else:
-                    del resp.headers[key]
+            for header in self.app.swift_owner_headers:
+                resp.headers.pop(header, None)
         return resp
 
     @public
@@ -82,6 +95,7 @@ class AccountController(Controller):
         resp = self.make_requests(
             req, self.app.account_ring, account_partition, 'PUT',
             req.swift_entity_path, [headers] * len(accounts))
+        self.add_acls_from_sys_metadata(resp)
         return resp
 
     @public
@@ -107,6 +121,7 @@ class AccountController(Controller):
             resp = self.make_requests(
                 req, self.app.account_ring, account_partition, 'POST',
                 req.swift_entity_path, [headers] * len(accounts))
+        self.add_acls_from_sys_metadata(resp)
         return resp
 
     @public

swift/proxy/controllers/base.py

@@ -273,7 +273,7 @@ def get_container_info(env, app, swift_source=None):
     .. note::
 
         This call bypasses auth. Success does not imply that the request has
-        authorization to the account.
+        authorization to the container.
     """
     (version, account, container, unused) = \
         split_path(env['PATH_INFO'], 3, 4, True)
@@ -292,7 +292,7 @@ def get_account_info(env, app, swift_source=None):
     .. note::
 
         This call bypasses auth. Success does not imply that the request has
-        authorization to the container.
+        authorization to the account.
     """
     (version, account, _junk, _junk) = \
         split_path(env['PATH_INFO'], 2, 4, True)

swift/proxy/controllers/container.py

@@ -99,6 +99,9 @@ class ContainerController(Controller):
             self.clean_acls(req) or check_metadata(req, 'container')
         if error_response:
             return error_response
+        if not req.environ.get('swift_owner'):
+            for key in self.app.swift_owner_headers:
+                req.headers.pop(key, None)
         if len(self.container_name) > MAX_CONTAINER_NAME_LENGTH:
             resp = HTTPBadRequest(request=req)
             resp.body = 'Container name length of %d longer than %d' % \
@@ -138,6 +141,9 @@ class ContainerController(Controller):
             self.clean_acls(req) or check_metadata(req, 'container')
         if error_response:
             return error_response
+        if not req.environ.get('swift_owner'):
+            for key in self.app.swift_owner_headers:
+                req.headers.pop(key, None)
         account_partition, accounts, container_count = \
             self.account_info(self.account_name, req)
         if not accounts:

swift/proxy/server.py

@@ -171,13 +171,17 @@ class Application(object):
         else:
             raise ValueError(
                 'Invalid write_affinity_node_count value: %r' % ''.join(value))
+        # swift_owner_headers are stripped by the account and container
+        # controllers; we should extend header stripping to object controller
+        # when a privileged object header is implemented.
         swift_owner_headers = conf.get(
             'swift_owner_headers',
             'x-container-read, x-container-write, '
             'x-container-sync-key, x-container-sync-to, '
-            'x-account-meta-temp-url-key, x-account-meta-temp-url-key-2')
+            'x-account-meta-temp-url-key, x-account-meta-temp-url-key-2, '
+            'x-account-access-control')
         self.swift_owner_headers = [
-            name.strip()
+            name.strip().title()
             for name in swift_owner_headers.split(',') if name.strip()]
         # Initialization was successful, so now apply the client chunk size
         # parameter as the default read / write buffer size for the network

test/functional/swift_testing.py

@@ -18,6 +18,7 @@ import os
 import socket
 import sys
 from time import sleep
+from urlparse import urlparse
 
 from test import get_config
 
@@ -119,18 +120,23 @@ conn = [None, None, None]
 
 def retry(func, *args, **kwargs):
     """
-    You can use the kwargs to override the 'retries' (default: 5) and
+    You can use the kwargs to override:
-    'use_account' (default: 1).
+      'retries' (default: 5)
+      'use_account' (default: 1) - which user's token to pass
+      'url_account' (default: matches 'use_account') - which user's storage URL
+      'resource' (default: url[url_account] - URL to connect to; retry()
+          will interpolate the variable :storage_url: if present
     """
     global url, token, parsed, conn
     retries = kwargs.get('retries', 5)
-    use_account = 1
+    attempts, backoff = 0, 1
-    if 'use_account' in kwargs:
+
-        use_account = kwargs['use_account']
+    # use account #1 by default; turn user's 1-indexed account into 0-indexed
-        del kwargs['use_account']
+    use_account = kwargs.pop('use_account', 1) - 1
-    use_account -= 1
+
-    attempts = 0
+    # access our own account by default
-    backoff = 1
+    url_account = kwargs.pop('url_account', use_account + 1) - 1
+
     while attempts <= retries:
         attempts += 1
         try:
@@ -146,8 +152,13 @@ def retry(func, *args, **kwargs):
             if not parsed[use_account] or not conn[use_account]:
                 parsed[use_account], conn[use_account] = \
                     http_connection(url[use_account])
-            return func(url[use_account], token[use_account],
+
-                        parsed[use_account], conn[use_account],
+            # default resource is the account url[url_account]
+            resource = kwargs.pop('resource', '%(storage_url)s')
+            template_vars = {'storage_url': url[url_account]}
+            parsed_result = urlparse(resource % template_vars)
+            return func(url[url_account], token[use_account],
+                        parsed_result, conn[url_account],
                         *args, **kwargs)
         except (socket.error, HTTPException):
             if attempts > retries:

test/functional/test_account.py

@@ -16,12 +16,16 @@
 # limitations under the License.
 
 import unittest
+import json
 from nose import SkipTest
 
 from swift.common.constraints import MAX_META_COUNT, MAX_META_NAME_LENGTH, \
     MAX_META_OVERALL_SIZE, MAX_META_VALUE_LENGTH
-
+from swift.common.middleware.acl import format_acl
+from test.functional.swift_test_client import Connection
+from test import get_config
 from swift_testing import check_response, retry, skip, web_front_end
+import swift_testing
 
 
 class TestAccount(unittest.TestCase):
@@ -66,6 +70,148 @@ class TestAccount(unittest.TestCase):
         self.assert_(resp.status in (200, 204), resp.status)
         self.assertEquals(resp.getheader('x-account-meta-test'), 'Value')
 
+    def test_tempauth_account_acls(self):
+        if skip:
+            raise SkipTest
+
+        # Determine whether this cluster has account ACLs; if not, skip test
+        conn = Connection(get_config('func_test'))
+        conn.authenticate()
+        status = conn.make_request(
+            'GET', '/info', cfg={'verbatim_path': True})
+        if status // 100 != 2:
+            # Can't tell if account ACLs are enabled; skip tests proactively.
+            raise SkipTest
+        else:
+            cluster_info = json.loads(conn.response.read())
+            if not cluster_info.get('tempauth', {}).get('account_acls'):
+                raise SkipTest
+            if 'keystoneauth' in cluster_info:
+                # Unfortunate hack -- tempauth (with account ACLs) is expected
+                # to play nice with Keystone (without account ACLs), but Zuul
+                # functest framework doesn't give us an easy way to get a
+                # tempauth user.
+                raise SkipTest
+
+        def post(url, token, parsed, conn, headers):
+            new_headers = dict({'X-Auth-Token': token}, **headers)
+            conn.request('POST', parsed.path, '', new_headers)
+            return check_response(conn)
+
+        def put(url, token, parsed, conn, headers):
+            new_headers = dict({'X-Auth-Token': token}, **headers)
+            conn.request('PUT', parsed.path, '', new_headers)
+            return check_response(conn)
+
+        def delete(url, token, parsed, conn, headers):
+            new_headers = dict({'X-Auth-Token': token}, **headers)
+            conn.request('DELETE', parsed.path, '', new_headers)
+            return check_response(conn)
+
+        def head(url, token, parsed, conn):
+            conn.request('HEAD', parsed.path, '', {'X-Auth-Token': token})
+            return check_response(conn)
+
+        def get(url, token, parsed, conn):
+            conn.request('GET', parsed.path, '', {'X-Auth-Token': token})
+            return check_response(conn)
+
+        try:
+            # User1 can POST to their own account (and reset the ACLs)
+            resp = retry(post, headers={'X-Account-Access-Control': '{}'},
+                         use_account=1)
+            resp.read()
+            self.assertEqual(resp.status, 204)
+            self.assertEqual(resp.getheader('X-Account-Access-Control'), None)
+
+            # User1 can GET their own empty account
+            resp = retry(get, use_account=1)
+            resp.read()
+            self.assertEqual(resp.status // 100, 2)
+            self.assertEqual(resp.getheader('X-Account-Access-Control'), None)
+
+            # User2 can't GET User1's account
+            resp = retry(get, use_account=2, url_account=1)
+            resp.read()
+            self.assertEqual(resp.status, 403)
+
+            # User1 is swift_owner of their own account, so they can POST an
+            # ACL -- let's do this and make User2 (test_user[1]) an admin
+            acl_user = swift_testing.swift_test_user[1]
+            acl = {'admin': [acl_user]}
+            headers = {'x-account-access-control': format_acl(
+                version=2, acl_dict=acl)}
+            resp = retry(post, headers=headers, use_account=1)
+            resp.read()
+            self.assertEqual(resp.status, 204)
+
+            # User1 can see the new header
+            resp = retry(get, use_account=1)
+            resp.read()
+            self.assertEqual(resp.status // 100, 2)
+            data_from_headers = resp.getheader('x-account-access-control')
+            expected = json.dumps(acl, separators=(',', ':'))
+            self.assertEqual(data_from_headers, expected)
+
+            # Now User2 should be able to GET the account and see the ACL
+            resp = retry(head, use_account=2, url_account=1)
+            resp.read()
+            data_from_headers = resp.getheader('x-account-access-control')
+            self.assertEqual(data_from_headers, expected)
+
+            # Revoke User2's admin access, grant User2 read-write access
+            acl = {'read-write': [acl_user]}
+            headers = {'x-account-access-control': format_acl(
+                version=2, acl_dict=acl)}
+            resp = retry(post, headers=headers, use_account=1)
+            resp.read()
+            self.assertEqual(resp.status, 204)
+
+            # User2 can still GET the account, but not see the ACL
+            # (since it's privileged data)
+            resp = retry(head, use_account=2, url_account=1)
+            resp.read()
+            self.assertEqual(resp.status, 204)
+            self.assertEqual(resp.getheader('x-account-access-control'), None)
+
+            # User2 can PUT and DELETE a container
+            resp = retry(put, use_account=2, url_account=1,
+                         resource='%(storage_url)s/mycontainer', headers={})
+            resp.read()
+            self.assertEqual(resp.status, 201)
+            resp = retry(delete, use_account=2, url_account=1,
+                         resource='%(storage_url)s/mycontainer', headers={})
+            resp.read()
+            self.assertEqual(resp.status, 204)
+
+            # Revoke User2's read-write access, grant User2 read-only access
+            acl = {'read-only': [acl_user]}
+            headers = {'x-account-access-control': format_acl(
+                version=2, acl_dict=acl)}
+            resp = retry(post, headers=headers, use_account=1)
+            resp.read()
+            self.assertEqual(resp.status, 204)
+
+            # User2 can still GET the account, but not see the ACL
+            # (since it's privileged data)
+            resp = retry(head, use_account=2, url_account=1)
+            resp.read()
+            self.assertEqual(resp.status, 204)
+            self.assertEqual(resp.getheader('x-account-access-control'), None)
+
+            # User2 can't PUT a container
+            resp = retry(put, use_account=2, url_account=1,
+                         resource='%(storage_url)s/mycontainer', headers={})
+            resp.read()
+            self.assertEqual(resp.status, 403)
+
+        finally:
+            # Make sure to clean up even if tests fail -- User2 should not
+            # have access to User1's account in other functional tests!
+            resp = retry(post, headers={'X-Account-Access-Control': '{}'},
+                         use_account=1)
+            resp.read()
+
     def test_unicode_metadata(self):
         if skip:
             raise SkipTest

test/functional/tests.py

@@ -118,7 +118,7 @@ def timeout(seconds, method, *args, **kwargs):
     return False
 
 
-class Utils:
+class Utils(object):
     @classmethod
     def create_ascii_name(cls, length=None):
         return uuid.uuid4().hex
@@ -170,7 +170,7 @@ class Base2(object):
         Utils.create_name = Utils.create_ascii_name
 
 
-class TestAccountEnv:
+class TestAccountEnv(object):
     @classmethod
     def setUp(cls):
         cls.conn = Connection(config)
@@ -357,7 +357,7 @@ class TestAccountUTF8(Base2, TestAccount):
     set_up = False
 
 
-class TestAccountNoContainersEnv:
+class TestAccountNoContainersEnv(object):
     @classmethod
     def setUp(cls):
         cls.conn = Connection(config)
@@ -386,7 +386,7 @@ class TestAccountNoContainersUTF8(Base2, TestAccountNoContainers):
     set_up = False
 
 
-class TestContainerEnv:
+class TestContainerEnv(object):
     @classmethod
     def setUp(cls):
         cls.conn = Connection(config)
@@ -677,7 +677,7 @@ class TestContainerUTF8(Base2, TestContainer):
     set_up = False
 
 
-class TestContainerPathsEnv:
+class TestContainerPathsEnv(object):
     @classmethod
     def setUp(cls):
         cls.conn = Connection(config)
@@ -856,7 +856,7 @@ class TestContainerPaths(Base):
                        ['dir1/subdir with spaces/file B'])
 
 
-class TestFileEnv:
+class TestFileEnv(object):
     @classmethod
     def setUp(cls):
         cls.conn = Connection(config)
@@ -1650,7 +1650,7 @@ class TestDloUTF8(Base2, TestDlo):
     set_up = False
 
 
-class TestFileComparisonEnv:
+class TestFileComparisonEnv(object):
     @classmethod
     def setUp(cls):
         cls.conn = Connection(config)

test/unit/common/middleware/test_acl.py

@@ -81,6 +81,72 @@ class TestACL(unittest.TestCase):
             (['ref3', 'ref5', '-ref6'],
              ['acc1', 'acc2:usr2', 'acc3', 'acc4:usr4']))
 
+    def test_parse_v2_acl(self):
+        # For all these tests, the header name will be "hdr".
+        tests = [
+            # Simple case: all ACL data in one header line
+            ({'hdr': '{"a":1,"b":"foo"}'}, {'a': 1, 'b': 'foo'}),
+
+            # No header "hdr" exists -- should return None
+            ({}, None),
+            ({'junk': 'junk'}, None),
+        ]
+
+        for hdrs_in, expected in tests:
+            result = acl.parse_acl(version=2, data=hdrs_in.get('hdr'))
+            self.assertEquals(expected, result,
+                              '%r: %r != %r' % (hdrs_in, result, expected))
+
+    def test_format_v1_acl(self):
+        tests = [
+            ((['a', 'b'], ['c.com']), 'a,b,.r:c.com'),
+            ((['a', 'b'], ['c.com', '-x.c.com']), 'a,b,.r:c.com,.r:-x.c.com'),
+            ((['a', 'b'], None), 'a,b'),
+            ((None, ['c.com']), '.r:c.com'),
+            ((None, None), ''),
+        ]
+
+        for (groups, refs), expected in tests:
+            result = acl.format_acl(
+                version=1, groups=groups, referrers=refs, header_name='hdr')
+            self.assertEquals(expected, result, 'groups=%r, refs=%r: %r != %r'
+                              % (groups, refs, result, expected))
+
+    def test_format_v2_acl(self):
+        tests = [
+            ({}, '{}'),
+            ({'foo': 'bar'}, '{"foo":"bar"}'),
+            ({'groups': ['a', 'b'], 'referrers': ['c.com', '-x.c.com']},
+             '{"groups":["a","b"],"referrers":["c.com","-x.c.com"]}'),
+        ]
+
+        for data, expected in tests:
+            result = acl.format_acl(version=2, acl_dict=data)
+            self.assertEquals(expected, result,
+                              'data=%r: %r *!=* %r' % (data, result, expected))
+
+    def test_acls_from_account_info(self):
+        test_data = [
+            ({}, None),
+            ({'sysmeta': {}}, None),
+            ({'sysmeta':
+              {'core-access-control': '{"VERSION":1,"admin":["a","b"]}'}},
+             {'admin': ['a', 'b'], 'read-write': [], 'read-only': []}),
+            ({
+                'some-key': 'some-value',
+                'other-key': 'other-value',
+                'sysmeta': {
+                    'core-access-control': '{"VERSION":1,"admin":["a","b"],"r'
+                                           'ead-write":["c"],"read-only":[]}',
+                }},
+             {'admin': ['a', 'b'], 'read-write': ['c'], 'read-only': []}),
+        ]
+
+        for args, expected in test_data:
+            result = acl.acls_from_account_info(args)
+            self.assertEqual(expected, result, "%r: Got %r, expected %r" %
+                             (args, result, expected))
+
     def test_referrer_allowed(self):
         self.assert_(not acl.referrer_allowed('host', None))
         self.assert_(not acl.referrer_allowed('host', []))

test/unit/common/middleware/test_tempauth.py

@@ -19,7 +19,11 @@ from base64 import b64encode
 from time import time
 
 from swift.common.middleware import tempauth as auth
+from swift.common.middleware.acl import format_acl
 from swift.common.swob import Request, Response
+from swift.common.utils import split_path
+
+NO_CONTENT_RESP = (('204 No Content', {}, ''),)   # mock server response
 
 
 class FakeMemcache(object):
@@ -62,7 +66,7 @@ class FakeApp(object):
 
     def __call__(self, env, start_response):
         self.calls += 1
-        self.request = Request.blank('', environ=env)
+        self.request = Request(env)
         if self.acl:
             self.request.acl = self.acl
         if self.sync_key:
@@ -244,8 +248,7 @@ class TestAuth(unittest.TestCase):
 
     def test_auth_no_reseller_prefix_no_token(self):
         # Check that normally we set up a call back to our authorize.
-        local_auth = \
+        local_auth = auth.filter_factory({'reseller_prefix': ''})(FakeApp())
-            auth.filter_factory({'reseller_prefix': ''})(FakeApp(iter([])))
         req = self._make_request('/v1/account')
         resp = req.get_response(local_auth)
         self.assertEquals(resp.status_int, 401)
@@ -293,6 +296,8 @@ class TestAuth(unittest.TestCase):
         self.assertEquals(resp.status_int, 403)
 
     def test_authorize_acl_group_access(self):
+        self.test_auth = auth.filter_factory({})(
+            FakeApp(iter(NO_CONTENT_RESP * 3)))
         req = self._make_request('/v1/AUTH_cfa')
         req.remote_user = 'act:usr,act'
         resp = self.test_auth.authorize(req)
@@ -331,6 +336,8 @@ class TestAuth(unittest.TestCase):
         self.assertEquals(self.test_auth.authorize(req), None)
 
     def test_authorize_acl_referrer_access(self):
+        self.test_auth = auth.filter_factory({})(
+            FakeApp(iter(NO_CONTENT_RESP * 6)))
         req = self._make_request('/v1/AUTH_cfa/c')
         req.remote_user = 'act:usr,act'
         resp = self.test_auth.authorize(req)
@@ -389,6 +396,8 @@ class TestAuth(unittest.TestCase):
         self.assertTrue(req.environ.get('reseller_request', False))
 
     def test_account_put_permissions(self):
+        self.test_auth = auth.filter_factory({})(
+            FakeApp(iter(NO_CONTENT_RESP * 4)))
         req = self._make_request('/v1/AUTH_new',
                                  environ={'REQUEST_METHOD': 'PUT'})
         req.remote_user = 'act:usr,act'
@@ -423,6 +432,8 @@ class TestAuth(unittest.TestCase):
         self.assertEquals(resp.status_int, 403)
 
     def test_account_delete_permissions(self):
+        self.test_auth = auth.filter_factory({})(
+            FakeApp(iter(NO_CONTENT_RESP * 4)))
         req = self._make_request('/v1/AUTH_new',
                                  environ={'REQUEST_METHOD': 'DELETE'})
         req.remote_user = 'act:usr,act'
@@ -456,6 +467,30 @@ class TestAuth(unittest.TestCase):
         resp = self.test_auth.authorize(req)
         self.assertEquals(resp.status_int, 403)
 
+    def test_get_token_success(self):
+        # Example of how to simulate the auth transaction
+        test_auth = auth.filter_factory({'user_ac_user': 'testing'})(FakeApp())
+        req = self._make_request(
+            '/auth/v1.0',
+            headers={'X-Auth-User': 'ac:user', 'X-Auth-Key': 'testing'})
+        resp = req.get_response(test_auth)
+        self.assertEquals(resp.status_int, 200)
+        self.assertTrue(resp.headers['x-storage-url'].endswith('/v1/AUTH_ac'))
+        self.assertTrue(resp.headers['x-auth-token'].startswith('AUTH_'))
+        self.assertTrue(len(resp.headers['x-auth-token']) > 10)
+
+    def test_use_token_success(self):
+        # Example of how to simulate an authorized request
+        test_auth = auth.filter_factory({'user_acct_user': 'testing'})(
+            FakeApp(iter(NO_CONTENT_RESP * 1)))
+        req = self._make_request('/v1/AUTH_acct',
+                                 headers={'X-Auth-Token': 'AUTH_t'})
+        cache_key = 'AUTH_/token/AUTH_t'
+        cache_entry = (time() + 3600, 'AUTH_acct')
+        req.environ['swift.cache'].set(cache_key, cache_entry)
+        resp = req.get_response(test_auth)
+        self.assertEquals(resp.status_int, 204)
+
     def test_get_token_fail(self):
         resp = self._make_request('/auth/v1.0').get_response(self.test_auth)
         self.assertEquals(resp.status_int, 401)
@@ -503,6 +538,17 @@ class TestAuth(unittest.TestCase):
         self.assertEquals(resp.headers.get('Www-Authenticate'),
                           'Swift realm="act"')
 
+    def test_object_name_containing_slash(self):
+        test_auth = auth.filter_factory({'user_acct_user': 'testing'})(
+            FakeApp(iter(NO_CONTENT_RESP * 1)))
+        req = self._make_request('/v1/AUTH_acct/cont/obj/name/with/slash',
+                                 headers={'X-Auth-Token': 'AUTH_t'})
+        cache_key = 'AUTH_/token/AUTH_t'
+        cache_entry = (time() + 3600, 'AUTH_acct')
+        req.environ['swift.cache'].set(cache_key, cache_entry)
+        resp = req.get_response(test_auth)
+        self.assertEquals(resp.status_int, 204)
+
     def test_storage_url_default(self):
         self.test_auth = \
             auth.filter_factory({'user_test_tester': 'testing'})(FakeApp())
@@ -653,7 +699,7 @@ class TestAuth(unittest.TestCase):
         self.assertEquals(owner_values, [False])
 
     def test_sync_request_success(self):
-        self.test_auth.app = FakeApp(iter([('204 No Content', {}, '')]),
+        self.test_auth.app = FakeApp(iter(NO_CONTENT_RESP * 1),
                                      sync_key='secret')
         req = self._make_request(
             '/v1/AUTH_cfa/c/o',
@@ -665,8 +711,7 @@ class TestAuth(unittest.TestCase):
         self.assertEquals(resp.status_int, 204)
 
     def test_sync_request_fail_key(self):
-        self.test_auth.app = FakeApp(iter([('204 No Content', {}, '')]),
+        self.test_auth.app = FakeApp(sync_key='secret')
-                                     sync_key='secret')
         req = self._make_request(
             '/v1/AUTH_cfa/c/o',
             environ={'REQUEST_METHOD': 'DELETE'},
@@ -678,8 +723,7 @@ class TestAuth(unittest.TestCase):
         self.assertEquals(resp.headers.get('Www-Authenticate'),
                           'Swift realm="AUTH_cfa"')
 
-        self.test_auth.app = FakeApp(iter([('204 No Content', {}, '')]),
+        self.test_auth.app = FakeApp(sync_key='othersecret')
-                                     sync_key='othersecret')
         req = self._make_request(
             '/v1/AUTH_cfa/c/o',
             environ={'REQUEST_METHOD': 'DELETE'},
@@ -691,8 +735,7 @@ class TestAuth(unittest.TestCase):
         self.assertEquals(resp.headers.get('Www-Authenticate'),
                           'Swift realm="AUTH_cfa"')
 
-        self.test_auth.app = FakeApp(iter([('204 No Content', {}, '')]),
+        self.test_auth.app = FakeApp(sync_key=None)
-                                     sync_key=None)
         req = self._make_request(
             '/v1/AUTH_cfa/c/o',
             environ={'REQUEST_METHOD': 'DELETE'},
@@ -705,8 +748,7 @@ class TestAuth(unittest.TestCase):
                           'Swift realm="AUTH_cfa"')
 
     def test_sync_request_fail_no_timestamp(self):
-        self.test_auth.app = FakeApp(iter([('204 No Content', {}, '')]),
+        self.test_auth.app = FakeApp(sync_key='secret')
-                                     sync_key='secret')
         req = self._make_request(
             '/v1/AUTH_cfa/c/o',
             environ={'REQUEST_METHOD': 'DELETE'},
@@ -718,7 +760,7 @@ class TestAuth(unittest.TestCase):
                           'Swift realm="AUTH_cfa"')
 
     def test_sync_request_success_lb_sync_host(self):
-        self.test_auth.app = FakeApp(iter([('204 No Content', {}, '')]),
+        self.test_auth.app = FakeApp(iter(NO_CONTENT_RESP * 1),
                                      sync_key='secret')
         req = self._make_request(
             '/v1/AUTH_cfa/c/o',
@@ -730,7 +772,7 @@ class TestAuth(unittest.TestCase):
         resp = req.get_response(self.test_auth)
         self.assertEquals(resp.status_int, 204)
 
-        self.test_auth.app = FakeApp(iter([('204 No Content', {}, '')]),
+        self.test_auth.app = FakeApp(iter(NO_CONTENT_RESP * 1),
                                      sync_key='secret')
         req = self._make_request(
             '/v1/AUTH_cfa/c/o',
@@ -826,5 +868,243 @@ class TestParseUserCreation(unittest.TestCase):
         }), FakeApp())
 
 
+class TestAccountAcls(unittest.TestCase):
+    def _make_request(self, path, **kwargs):
+        # Our TestAccountAcls default request will have a valid auth token
+        version, acct, _ = split_path(path, 1, 3, True)
+        headers = kwargs.pop('headers', {'X-Auth-Token': 'AUTH_t'})
+        user_groups = kwargs.pop('user_groups', 'AUTH_firstacct')
+
+        # The account being accessed will have account ACLs
+        acl = {'admin': ['AUTH_admin'], 'read-write': ['AUTH_rw'],
+               'read-only': ['AUTH_ro']}
+        header_data = {'core-access-control':
+                       format_acl(version=2, acl_dict=acl)}
+        acls = kwargs.pop('acls', header_data)
+
+        req = Request.blank(path, headers=headers, **kwargs)
+
+        # Authorize the token by populating the request's cache
+        req.environ['swift.cache'] = FakeMemcache()
+        cache_key = 'AUTH_/token/AUTH_t'
+        cache_entry = (time() + 3600, user_groups)
+        req.environ['swift.cache'].set(cache_key, cache_entry)
+
+        # Pretend get_account_info returned ACLs in sysmeta, and we cached that
+        cache_key = 'account/%s' % acct
+        cache_entry = {'sysmeta': acls}
+        req.environ['swift.cache'].set(cache_key, cache_entry)
+
+        return req
+
+    def test_account_acl_success(self):
+        test_auth = auth.filter_factory({'user_admin_user': 'testing'})(
+            FakeApp(iter(NO_CONTENT_RESP * 1)))
+
+        # admin (not a swift admin) wants to read from otheracct
+        req = self._make_request('/v1/AUTH_otheract', user_groups="AUTH_admin")
+
+        # The request returned by _make_request should be allowed
+        resp = req.get_response(test_auth)
+        self.assertEquals(resp.status_int, 204)
+
+    def test_account_acl_failures(self):
+        test_auth = auth.filter_factory({'user_admin_user': 'testing'})(
+            FakeApp())
+
+        # If I'm not authed as anyone on the ACLs, I shouldn't get in
+        req = self._make_request('/v1/AUTH_otheract', user_groups="AUTH_bob")
+        resp = req.get_response(test_auth)
+        self.assertEquals(resp.status_int, 403)
+
+        # If the target account has no ACLs, a non-owner shouldn't get in
+        req = self._make_request('/v1/AUTH_otheract', user_groups="AUTH_admin",
+                                 acls={})
+        resp = req.get_response(test_auth)
+        self.assertEquals(resp.status_int, 403)
+
+    def test_admin_privileges(self):
+        test_auth = auth.filter_factory({'user_admin_user': 'testing'})(
+            FakeApp(iter(NO_CONTENT_RESP * 18)))
+
+        for target in ('/v1/AUTH_otheracct', '/v1/AUTH_otheracct/container',
+                       '/v1/AUTH_otheracct/container/obj'):
+            for method in ('GET', 'HEAD', 'OPTIONS', 'PUT', 'POST', 'DELETE'):
+                # Admin ACL user can do anything
+                req = self._make_request(target, user_groups="AUTH_admin",
+                                         environ={'REQUEST_METHOD': method})
+                resp = req.get_response(test_auth)
+                self.assertEquals(resp.status_int, 204)
+
+                # swift_owner should be set to True
+                if method != 'OPTIONS':
+                    self.assertTrue(req.environ.get('swift_owner'))
+
+    def test_readwrite_privileges(self):
+        test_auth = auth.filter_factory({'user_rw_user': 'testing'})(
+            FakeApp(iter(NO_CONTENT_RESP * 15)))
+
+        for target in ('/v1/AUTH_otheracct',):
+            for method in ('GET', 'HEAD', 'OPTIONS'):
+                # Read-Write user can read account data
+                req = self._make_request(target, user_groups="AUTH_rw",
+                                         environ={'REQUEST_METHOD': method})
+                resp = req.get_response(test_auth)
+                self.assertEquals(resp.status_int, 204)
+
+                # swift_owner should NOT be set to True
+                self.assertFalse(req.environ.get('swift_owner'))
+
+            # RW user should NOT be able to PUT, POST, or DELETE to the account
+            for method in ('PUT', 'POST', 'DELETE'):
+                req = self._make_request(target, user_groups="AUTH_rw",
+                                         environ={'REQUEST_METHOD': method})
+                resp = req.get_response(test_auth)
+                self.assertEquals(resp.status_int, 403)
+
+        # RW user should be able to GET, PUT, POST, or DELETE to containers
+        # and objects
+        for target in ('/v1/AUTH_otheracct/c', '/v1/AUTH_otheracct/c/o'):
+            for method in ('GET', 'HEAD', 'OPTIONS', 'PUT', 'POST', 'DELETE'):
+                req = self._make_request(target, user_groups="AUTH_rw",
+                                         environ={'REQUEST_METHOD': method})
+                resp = req.get_response(test_auth)
+                self.assertEquals(resp.status_int, 204)
+
+    def test_readonly_privileges(self):
+        test_auth = auth.filter_factory({'user_ro_user': 'testing'})(
+            FakeApp(iter(NO_CONTENT_RESP * 9)))
+
+        # ReadOnly user should NOT be able to PUT, POST, or DELETE to account,
+        # container, or object
+        for target in ('/v1/AUTH_otheracct', '/v1/AUTH_otheracct/cont',
+                       '/v1/AUTH_otheracct/cont/obj'):
+            for method in ('GET', 'HEAD', 'OPTIONS'):
+                req = self._make_request(target, user_groups="AUTH_ro",
+                                         environ={'REQUEST_METHOD': method})
+                resp = req.get_response(test_auth)
+                self.assertEquals(resp.status_int, 204)
+                # swift_owner should NOT be set to True for the ReadOnly ACL
+                self.assertFalse(req.environ.get('swift_owner'))
+            for method in ('PUT', 'POST', 'DELETE'):
+                req = self._make_request(target, user_groups="AUTH_ro",
+                                         environ={'REQUEST_METHOD': method})
+                resp = req.get_response(test_auth)
+                self.assertEquals(resp.status_int, 403)
+                # swift_owner should NOT be set to True for the ReadOnly ACL
+                self.assertFalse(req.environ.get('swift_owner'))
+
+    def test_user_gets_best_acl(self):
+        test_auth = auth.filter_factory({'user_acct_username': 'testing'})(
+            FakeApp(iter(NO_CONTENT_RESP * 18)))
+
+        mygroups = "AUTH_acct,AUTH_ro,AUTH_something,AUTH_admin"
+        for target in ('/v1/AUTH_otheracct', '/v1/AUTH_otheracct/container',
+                       '/v1/AUTH_otheracct/container/obj'):
+            for method in ('GET', 'HEAD', 'OPTIONS', 'PUT', 'POST', 'DELETE'):
+                # Admin ACL user can do anything
+                req = self._make_request(target, user_groups=mygroups,
+                                         environ={'REQUEST_METHOD': method})
+                resp = req.get_response(test_auth)
+                self.assertEquals(
+                    resp.status_int, 204, "%s (%s) - expected 204, got %d" %
+                    (target, method, resp.status_int))
+
+                # swift_owner should be set to True
+                if method != 'OPTIONS':
+                    self.assertTrue(req.environ.get('swift_owner'))
+
+    def test_acl_syntax_verification(self):
+        test_auth = auth.filter_factory({'user_admin_user': 'testing'})(
+            FakeApp(iter(NO_CONTENT_RESP * 3)))
+
+        good_headers = {'X-Auth-Token': 'AUTH_t'}
+        good_acl = '{"read-only":["a","b"]}'
+        bad_acl = 'syntactically invalid acl -- this does not parse as JSON'
+        wrong_acl = '{"other-auth-system":["valid","json","but","wrong"]}'
+        bad_value_acl = '{"read-write":["fine"],"admin":"should be a list"}'
+        target = '/v1/AUTH_firstacct'
+
+        # no acls -- no problem!
+        req = self._make_request(target, headers=good_headers)
+        resp = req.get_response(test_auth)
+        self.assertEquals(resp.status_int, 204)
+
+        # syntactically valid acls should go through
+        update = {'x-account-access-control': good_acl}
+        req = self._make_request(target, headers=dict(good_headers, **update))
+        resp = req.get_response(test_auth)
+        self.assertEquals(resp.status_int, 204)
+
+        errmsg = 'X-Account-Access-Control invalid: %s'
+        # syntactically invalid acls get a 400
+        update = {'x-account-access-control': bad_acl}
+        req = self._make_request(target, headers=dict(good_headers, **update))
+        resp = req.get_response(test_auth)
+        self.assertEquals(resp.status_int, 400)
+        self.assertEquals(errmsg % "Syntax error", resp.body[:46])
+
+        # syntactically valid acls with bad keys also get a 400
+        update = {'x-account-access-control': wrong_acl}
+        req = self._make_request(target, headers=dict(good_headers, **update))
+        resp = req.get_response(test_auth)
+        self.assertEquals(resp.status_int, 400)
+        self.assertEquals(errmsg % "Key '", resp.body[:39])
+
+        # acls with good keys but bad values also get a 400
+        update = {'x-account-access-control': bad_value_acl}
+        req = self._make_request(target, headers=dict(good_headers, **update))
+        resp = req.get_response(test_auth)
+        self.assertEquals(resp.status_int, 400)
+        self.assertEquals(errmsg % "Value", resp.body[:39])
+
+    def test_acls_propagate_to_sysmeta(self):
+        test_auth = auth.filter_factory({'user_admin_user': 'testing'})(
+            FakeApp(iter(NO_CONTENT_RESP * 3)))
+
+        sysmeta_hdr = 'x-account-sysmeta-core-access-control'
+        target = '/v1/AUTH_firstacct'
+        good_headers = {'X-Auth-Token': 'AUTH_t'}
+        good_acl = '{"read-only":["a","b"]}'
+
+        # no acls -- no problem!
+        req = self._make_request(target, headers=good_headers)
+        resp = req.get_response(test_auth)
+        self.assertEquals(resp.status_int, 204)
+        self.assertEqual(None, req.headers.get(sysmeta_hdr))
+
+        # syntactically valid acls should go through
+        update = {'x-account-access-control': good_acl}
+        req = self._make_request(target, headers=dict(good_headers, **update))
+        resp = req.get_response(test_auth)
+        self.assertEquals(resp.status_int, 204)
+        self.assertEqual(good_acl, req.headers.get(sysmeta_hdr))
+
+    def test_bad_acls_get_denied(self):
+        test_auth = auth.filter_factory({'user_admin_user': 'testing'})(
+            FakeApp(iter(NO_CONTENT_RESP * 3)))
+
+        target = '/v1/AUTH_firstacct'
+        good_headers = {'X-Auth-Token': 'AUTH_t'}
+        bad_acls = (
+            'syntax error',
+            '{"bad_key":"should_fail"}',
+            '{"admin":"not a list, should fail"}',
+            '{"admin":["valid"],"read-write":"not a list, should fail"}',
+        )
+
+        for bad_acl in bad_acls:
+            hdrs = dict(good_headers, **{'x-account-access-control': bad_acl})
+            req = self._make_request(target, headers=hdrs)
+            resp = req.get_response(test_auth)
+            self.assertEquals(resp.status_int, 400)
+
+
+class TestUtilityMethods(unittest.TestCase):
+    def test_account_acls_bad_path_raises_exception(self):
+        auth_inst = auth.filter_factory({})(FakeApp())
+        req = Request({'PATH_INFO': '/'})
+        self.assertRaises(ValueError, auth_inst.account_acls, req)
+
 if __name__ == '__main__':
     unittest.main()

test/unit/proxy/controllers/test_account.py

@@ -16,12 +16,14 @@
 import mock
 import unittest
 
-from swift.common.swob import Request
+from swift.common.swob import Request, Response
+from swift.common.middleware.acl import format_acl
 from swift.proxy import server as proxy_server
 from swift.proxy.controllers.base import headers_to_account_info
 from swift.common.constraints import MAX_ACCOUNT_NAME_LENGTH as MAX_ANAME_LEN
 from test.unit import fake_http_connect, FakeRing, FakeMemcache
 from swift.common.request_helpers import get_sys_meta_prefix
+import swift.proxy.controllers.base
 
 
 class TestAccountController(unittest.TestCase):
@@ -152,6 +154,91 @@ class TestAccountController(unittest.TestCase):
         self.assertEqual(context['headers'][user_meta_key], 'bar')
         self.assertNotEqual(context['headers']['x-timestamp'], '1.0')
 
+    def _make_user_and_sys_acl_headers_data(self):
+        acl = {
+            'admin': ['AUTH_alice', 'AUTH_bob'],
+            'read-write': ['AUTH_carol'],
+            'read-only': [],
+        }
+        user_prefix = 'x-account-'  # external, user-facing
+        user_headers = {(user_prefix + 'access-control'): format_acl(
+            version=2, acl_dict=acl)}
+        sys_prefix = get_sys_meta_prefix('account')   # internal, system-facing
+        sys_headers = {(sys_prefix + 'core-access-control'): format_acl(
+            version=2, acl_dict=acl)}
+        return user_headers, sys_headers
+
+    def test_account_acl_headers_translated_for_GET_HEAD(self):
+        # Verify that a GET/HEAD which receives X-Account-Sysmeta-Acl-* headers
+        # from the account server will remap those headers to X-Account-Acl-*
+
+        hdrs_ext, hdrs_int = self._make_user_and_sys_acl_headers_data()
+        controller = proxy_server.AccountController(self.app, 'acct')
+
+        for verb in ('GET', 'HEAD'):
+            req = Request.blank('/v1/acct', environ={'swift_owner': True})
+            controller.GETorHEAD_base = lambda *_: Response(
+                headers=hdrs_int, environ={
+                    'PATH_INFO': '/acct',
+                    'REQUEST_METHOD': verb,
+                })
+            method = getattr(controller, verb)
+            resp = method(req)
+            for header, value in hdrs_ext.items():
+                if value:
+                    self.assertEqual(resp.headers.get(header), value)
+                else:
+                    # blank ACLs should result in no header
+                    self.assert_(header not in resp.headers)
+
+    def test_add_acls_impossible_cases(self):
+        # For test coverage: verify that defensive coding does defend, in cases
+        # that shouldn't arise naturally
+
+        # add_acls should do nothing if REQUEST_METHOD isn't HEAD/GET/PUT/POST
+        resp = Response()
+        controller = proxy_server.AccountController(self.app, 'a')
+        resp.environ['PATH_INFO'] = '/a'
+        resp.environ['REQUEST_METHOD'] = 'OPTIONS'
+        controller.add_acls_from_sys_metadata(resp)
+        self.assertEqual(1, len(resp.headers))  # we always get Content-Type
+        self.assertEqual(2, len(resp.environ))
+
+    def test_memcache_key_impossible_cases(self):
+        # For test coverage: verify that defensive coding does defend, in cases
+        # that shouldn't arise naturally
+        self.assertRaises(
+            ValueError,
+            lambda: swift.proxy.controllers.base.get_container_memcache_key(
+                '/a', None))
+
+    def test_stripping_swift_admin_headers(self):
+        # Verify that a GET/HEAD which receives privileged headers from the
+        # account server will strip those headers for non-swift_owners
+
+        hdrs_ext, hdrs_int = self._make_user_and_sys_acl_headers_data()
+        headers = {
+            'x-account-meta-harmless': 'hi mom',
+            'x-account-meta-temp-url-key': 's3kr1t',
+        }
+        controller = proxy_server.AccountController(self.app, 'acct')
+
+        for verb in ('GET', 'HEAD'):
+            for env in ({'swift_owner': True}, {'swift_owner': False}):
+                req = Request.blank('/v1/acct', environ=env)
+                controller.GETorHEAD_base = lambda *_: Response(
+                    headers=headers, environ={
+                        'PATH_INFO': '/acct',
+                        'REQUEST_METHOD': verb,
+                    })
+                method = getattr(controller, verb)
+                resp = method(req)
+                self.assertEqual(resp.headers.get('x-account-meta-harmless'),
+                                 'hi mom')
+                privileged_header_present = (
+                    'x-account-meta-temp-url-key' in resp.headers)
+                self.assertEqual(privileged_header_present, env['swift_owner'])
+
 
 if __name__ == '__main__':
     unittest.main()

test/unit/proxy/test_server.py

@@ -40,6 +40,7 @@ from swift.container import server as container_server
 from swift.obj import server as object_server
 from swift.common import ring
 from swift.common.middleware import proxy_logging
+from swift.common.middleware.acl import parse_acl, format_acl
 from swift.common.exceptions import ChunkReadTimeout
 from swift.common.constraints import MAX_META_NAME_LENGTH, \
     MAX_META_VALUE_LENGTH, MAX_META_COUNT, MAX_META_OVERALL_SIZE, \
@@ -55,6 +56,7 @@ from swift.proxy.controllers.base import get_container_memcache_key, \
 import swift.proxy.controllers
 from swift.common.swob import Request, Response, HTTPNotFound, \
     HTTPUnauthorized
+from swift.common.request_helpers import get_sys_meta_prefix
 
 # mocks
 logging.getLogger().addHandler(logging.StreamHandler(sys.stdout))
@@ -4895,9 +4897,10 @@ class TestContainerController(unittest.TestCase):
                 controller = \
                     proxy_server.ContainerController(self.app, 'a', 'c')
                 set_http_connect(200, 201, 201, 201, give_connect=test_connect)
-                req = Request.blank('/v1/a/c',
+                req = Request.blank(
-                                    environ={'REQUEST_METHOD': method},
+                    '/v1/a/c',
-                                    headers={test_header: test_value})
+                    environ={'REQUEST_METHOD': method, 'swift_owner': True},
+                    headers={test_header: test_value})
                 self.app.update_request(req)
                 getattr(controller, method)(req)
                 self.assertEquals(test_errors, [])
@@ -5925,6 +5928,143 @@ class TestAccountControllerFakeGetResponse(unittest.TestCase):
             resp = req.get_response(self.app)
             self.assertEqual(400, resp.status_int)
 
+    def test_account_acl_header_access(self):
+        acl = {
+            'admin': ['AUTH_alice'],
+            'read-write': ['AUTH_bob'],
+            'read-only': ['AUTH_carol'],
+        }
+        prefix = get_sys_meta_prefix('account')
+        privileged_headers = {(prefix + 'core-access-control'): format_acl(
+            version=2, acl_dict=acl)}
+
+        app = proxy_server.Application(
+            None, FakeMemcache(), account_ring=FakeRing(),
+            container_ring=FakeRing(), object_ring=FakeRing())
+
+        with save_globals():
+            # Mock account server will provide privileged information (ACLs)
+            set_http_connect(200, 200, 200, headers=privileged_headers)
+            req = Request.blank('/v1/a', environ={'REQUEST_METHOD': 'GET'})
+            resp = app.handle_request(req)
+
+            # Not a swift_owner -- ACLs should NOT be in response
+            header = 'X-Account-Access-Control'
+            self.assert_(header not in resp.headers, '%r was in %r' % (
+                header, resp.headers))
+
+            # Same setup -- mock acct server will provide ACLs
+            set_http_connect(200, 200, 200, headers=privileged_headers)
+            req = Request.blank('/v1/a', environ={'REQUEST_METHOD': 'GET',
+                                                  'swift_owner': True})
+            resp = app.handle_request(req)
+
+            # For a swift_owner, the ACLs *should* be in response
+            self.assert_(header in resp.headers, '%r not in %r' % (
+                header, resp.headers))
+
+    def test_account_acls_through_delegation(self):
+
+        # Define a way to grab the requests sent out from the AccountController
+        # to the Account Server, and a way to inject responses we'd like the
+        # Account Server to return.
+        resps_to_send = []
+
+        @contextmanager
+        def patch_account_controller_method(verb):
+            old_method = getattr(proxy_server.AccountController, verb)
+            new_method = lambda self, req, *_, **__: resps_to_send.pop(0)
+            try:
+                setattr(proxy_server.AccountController, verb, new_method)
+                yield
+            finally:
+                setattr(proxy_server.AccountController, verb, old_method)
+
+        def make_test_request(http_method, swift_owner=True):
+            env = {
+                'REQUEST_METHOD': http_method,
+                'swift_owner': swift_owner,
+            }
+            acl = {
+                'admin': ['foo'],
+                'read-write': ['bar'],
+                'read-only': ['bas'],
+            }
+            headers = {} if http_method in ('GET', 'HEAD') else {
+                'x-account-access-control': format_acl(version=2, acl_dict=acl)
+            }
+
+            return Request.blank('/v1/a', environ=env, headers=headers)
+
+        # Our AccountController will invoke methods to communicate with the
+        # Account Server, and they will return responses like these:
+        def make_canned_response(http_method):
+            acl = {
+                'admin': ['foo'],
+                'read-write': ['bar'],
+                'read-only': ['bas'],
+            }
+            headers = {'x-account-sysmeta-core-access-control': format_acl(
+                version=2, acl_dict=acl)}
+            canned_resp = Response(headers=headers)
+            canned_resp.environ = {
+                'PATH_INFO': '/acct',
+                'REQUEST_METHOD': http_method,
+            }
+            resps_to_send.append(canned_resp)
+
+        app = proxy_server.Application(
+            None, FakeMemcache(), account_ring=FakeRing(),
+            container_ring=FakeRing(), object_ring=FakeRing())
+        app.allow_account_management = True
+
+        ext_header = 'x-account-access-control'
+        with patch_account_controller_method('GETorHEAD_base'):
+            # GET/HEAD requests should remap sysmeta headers from acct server
+            for verb in ('GET', 'HEAD'):
+                make_canned_response(verb)
+                req = make_test_request(verb)
+                resp = app.handle_request(req)
+                h = parse_acl(version=2, data=resp.headers.get(ext_header))
+                self.assertEqual(h['admin'], ['foo'])
+                self.assertEqual(h['read-write'], ['bar'])
+                self.assertEqual(h['read-only'], ['bas'])
+
+                # swift_owner = False: GET/HEAD shouldn't return sensitive info
+                make_canned_response(verb)
+                req = make_test_request(verb, swift_owner=False)
+                resp = app.handle_request(req)
+                h = resp.headers
+                self.assertEqual(None, h.get(ext_header))
+
+                # swift_owner unset: GET/HEAD shouldn't return sensitive info
+                make_canned_response(verb)
+                req = make_test_request(verb, swift_owner=False)
+                del req.environ['swift_owner']
+                resp = app.handle_request(req)
+                h = resp.headers
+                self.assertEqual(None, h.get(ext_header))
+
+        # Verify that PUT/POST requests remap sysmeta headers from acct server
+        with patch_account_controller_method('make_requests'):
+            make_canned_response('PUT')
+            req = make_test_request('PUT')
+            resp = app.handle_request(req)
+
+            h = parse_acl(version=2, data=resp.headers.get(ext_header))
+            self.assertEqual(h['admin'], ['foo'])
+            self.assertEqual(h['read-write'], ['bar'])
+            self.assertEqual(h['read-only'], ['bas'])
+
+            make_canned_response('POST')
+            req = make_test_request('POST')
+            resp = app.handle_request(req)
+
+            h = parse_acl(version=2, data=resp.headers.get(ext_header))
+            self.assertEqual(h['admin'], ['foo'])
+            self.assertEqual(h['read-write'], ['bar'])
+            self.assertEqual(h['read-only'], ['bas'])
+
 
 class FakeObjectController(object):