Move swift_auth middleware from keystone to swift.

- Rename it to keystoneauth for consistenties.
- Implements blueprint keystone-middleware.

Change-Id: I208fecdf3ee991694b4239f065032324d297fd35

doc/source/overview_auth.rst

@@ -42,6 +42,91 @@ such as the X-Container-Sync-Key for a container GET or HEAD.
 The user starts a session by sending a ReST request to the auth system to
 receive the auth token and a URL to the Swift system.
 
+-------------
+Keystone Auth
+-------------
+
+Swift is able to authenticate against OpenStack keystone via the
+:mod:`swift.common.middleware.keystoneauth` middleware.
+
+In order to use the ``keystoneauth`` middleware the ``authtoken``
+middleware from keystone will need to be configured.
+
+The ``authtoken`` middleware performs the authentication token
+validation and retrieves actual user authentication information. It
+can be found in the Keystone distribution.
+
+The ``keystoneauth`` middleware performs authorization and mapping the
+``keystone`` roles to Swift's ACLs.
+
+Configuring Swift to use Keystone
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Configuring Swift to use Keystone is relatively straight
+forward.  The first step is to ensure that you have the auth_token
+middleware installed, distributed with keystone it can either be
+dropped in your python path or installed via the keystone package.
+
+You need at first make sure you have a service endpoint of type
+``object-store`` in keystone pointing to your Swift proxy. For example
+having this in your ``/etc/keystone/default_catalog.templates`` ::
+
+  catalog.RegionOne.object_store.name = Swift Service
+  catalog.RegionOne.object_store.publicURL = http://swiftproxy:8080/v1/AUTH_$(tenant_id)s
+  catalog.RegionOne.object_store.adminURL = http://swiftproxy:8080/
+  catalog.RegionOne.object_store.internalURL = http://swiftproxy:8080/v1/AUTH_$(tenant_id)s
+
+On your Swift Proxy server you will want to adjust your main pipeline
+and add auth_token and keystoneauth in your
+``/etc/swift/proxy-server.conf`` like this ::
+
+  [pipeline:main]
+  pipeline = [....] authtoken keystoneauth proxy-logging proxy-server
+
+add the configuration for the authtoken middleware::
+
+  [filter:authtoken]
+  paste.filter_factory = keystone.middleware.auth_token:filter_factory
+  auth_host = keystonehost
+  auth_port = 35357
+  auth_protocol = http
+  auth_uri = http://keystonehost:5000/
+  admin_tenant_name = service
+  admin_user = swift
+  admin_password = password
+
+The actual values for these variables will need to be set depending on
+your situation.  For more information, please refer to the Keystone
+documentation on the ``auth_token`` middleware, but in short:
+
+* Those variables beginning with ``auth_`` point to the Keystone
+  Admin service.  This information is used by the middleware to actually
+  query Keystone about the validity of the
+  authentication tokens.
+* The admin auth credentials (``admin_user``, ``admin_tenant_name``,
+  ``admin_password``) will be used to retrieve an admin token. That
+  token will be used to authorize user tokens behind the scenes.
+
+.. note::
+
+    If support is required for unvalidated users (as with anonymous
+    access) or for tempurl/formpost middleware, authtoken will need
+    to be configured with delay_auth_decision set to 1.
+
+and you can finally add the keystoneauth configuration::
+
+  [filter:keystoneauth]
+  use = egg:swift#keystoneauth
+  operator_roles = admin, swiftoperator
+
+By default the only users able to give ACL or to Create other
+containers are the ones who has the Keystone role specified in the
+``operator_roles`` setting.
+
+This user who have one of those role will be able to give ACLs to
+other users on containers, see the documentation on ACL here
+:mod:`swift.common.middleware.acl`.
+
 --------------
 Extending Auth
 --------------

etc/proxy-server.conf-sample

@@ -118,6 +118,32 @@ user_test_tester = testing .admin
 user_test2_tester2 = testing2 .admin
 user_test_tester3 = testing3
 
+# To enable Keystone authentication you need to have the auth token
+# middleware first to be configured. Here is an example below, please
+# refer to the keystone's documentation for details about the
+# different settings.
+#
+# You'll need to have as well the keystoneauth middleware enabled
+# and have it in your main pipeline so instead of having tempauth in
+# there you can change it to: authtoken keystone
+#
+# [filter:authtoken]
+# paste.filter_factory = keystone.middleware.auth_token:filter_factory
+# auth_host = keystonehost
+# auth_port = 35357
+# auth_protocol = http
+# auth_uri = http://keystonehost:5000/
+# admin_tenant_name = service
+# admin_user = swift
+# admin_password = password
+# delay_auth_decision = 1
+#
+# [filter:keystoneauth]
+# use = egg:swift#keystoneauth
+# Operator roles is the role which user would be allowed to manage a
+# tenant and be able to create container or give ACL to others.
+# operator_roles = admin, swiftoperator
+
 [filter:healthcheck]
 use = egg:swift#healthcheck
 # You can override the default log routing for this filter here:

setup.py

@@ -88,6 +88,7 @@ setup(
             'domain_remap=swift.common.middleware.domain_remap:filter_factory',
             'staticweb=swift.common.middleware.staticweb:filter_factory',
             'tempauth=swift.common.middleware.tempauth:filter_factory',
+            'keystoneauth=swift.common.middleware.keystoneauth:filter_factory',
             'recon=swift.common.middleware.recon:filter_factory',
             'tempurl=swift.common.middleware.tempurl:filter_factory',
             'formpost=swift.common.middleware.formpost:filter_factory',

swift/common/middleware/keystoneauth.py

@@ -0,0 +1,289 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2012 OpenStack LLC
+#
+# 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 webob
+
+from swift.common import utils as swift_utils
+from swift.common.middleware import acl as swift_acl
+
+
+class KeystoneAuth(object):
+    """Swift middleware to Keystone authorization system.
+
+    In Swift's proxy-server.conf add this middleware to your pipeline::
+
+        [pipeline:main]
+        pipeline = catch_errors cache authtoken keystoneauth proxy-server
+
+    Make sure you have the authtoken middleware before the
+    keystoneauth middleware.
+
+    The authtoken middleware will take care of validating the user and
+    keystoneauth will authorize access.
+
+    The authtoken middleware is shipped directly with keystone it
+    does not have any other dependences than itself so you can either
+    install it by copying the file directly in your python path or by
+    installing keystone.
+
+    If support is required for unvalidated users (as with anonymous
+    access) or for tempurl/formpost middleware, authtoken will need
+    to be configured with delay_auth_decision set to 1.  See the
+    Keystone documentation for more detail on how to configure the
+    authtoken middleware.
+
+    In proxy-server.conf you will need to have the setting account
+    auto creation to true::
+
+        [app:proxy-server] account_autocreate = true
+
+    And add a swift authorization filter section, such as::
+
+        [filter:keystoneauth]
+        use = egg:swift#keystoneauth
+        operator_roles = admin, swiftoperator
+
+    This maps tenants to account in Swift.
+
+    The user whose able to give ACL / create Containers permissions
+    will be the one that are inside the operator_roles
+    setting which by default includes the admin and the swiftoperator
+    roles.
+
+    The option is_admin if set to true will allow the
+    username that has the same name as the account name to be the owner.
+
+    Example: If we have the account called hellocorp with a user
+    hellocorp that user will be admin on that account and can give ACL
+    to all other users for hellocorp.
+
+    If you need to have a different reseller_prefix to be able to
+    mix different auth servers you can configure the option
+    reseller_prefix in your keystoneauth entry like this :
+
+        reseller_prefix = NEWAUTH_
+
+    Make sure you have a underscore at the end of your new
+    reseller_prefix option.
+
+    :param app: The next WSGI app in the pipeline
+    :param conf: The dict of configuration values
+    """
+    def __init__(self, app, conf):
+        self.app = app
+        self.conf = conf
+        self.logger = swift_utils.get_logger(conf, log_route='keystoneauth')
+        self.reseller_prefix = conf.get('reseller_prefix', 'AUTH_').strip()
+        self.operator_roles = conf.get('operator_roles',
+                                       'admin, swiftoperator')
+        self.reseller_admin_role = conf.get('reseller_admin_role',
+                                            'ResellerAdmin')
+        config_is_admin = conf.get('is_admin', "false").lower()
+        self.is_admin = config_is_admin in swift_utils.TRUE_VALUES
+        cfg_synchosts = conf.get('allowed_sync_hosts', '127.0.0.1')
+        self.allowed_sync_hosts = [h.strip() for h in cfg_synchosts.split(',')
+                                   if h.strip()]
+        config_overrides = conf.get('allow_overrides', 't').lower()
+        self.allow_overrides = config_overrides in swift_utils.TRUE_VALUES
+
+    def __call__(self, environ, start_response):
+        identity = self._keystone_identity(environ)
+
+        # Check if one of the middleware like tempurl or formpost have
+        # set the swift.authorize_override environ and want to control the
+        # authentication
+        if (self.allow_overrides and
+                environ.get('swift.authorize_override', False)):
+            msg = 'Authorizing from an overriding middleware (i.e: tempurl)'
+            self.logger.debug(msg)
+            return self.app(environ, start_response)
+
+        if identity:
+            self.logger.debug('Using identity: %r' % (identity))
+            environ['keystone.identity'] = identity
+            environ['REMOTE_USER'] = identity.get('tenant')
+            environ['swift.authorize'] = self.authorize
+        else:
+            self.logger.debug('Authorizing as anonymous')
+            environ['swift.authorize'] = self.authorize_anonymous
+
+        environ['swift.clean_acl'] = swift_acl.clean_acl
+
+        return self.app(environ, start_response)
+
+    def _keystone_identity(self, environ):
+        """Extract the identity from the Keystone auth component."""
+        if environ.get('HTTP_X_IDENTITY_STATUS') != 'Confirmed':
+            return
+        roles = []
+        if 'HTTP_X_ROLES' in environ:
+            roles = environ['HTTP_X_ROLES'].split(',')
+        identity = {'user': environ.get('HTTP_X_USER_NAME'),
+                    'tenant': (environ.get('HTTP_X_TENANT_ID'),
+                               environ.get('HTTP_X_TENANT_NAME')),
+                    'roles': roles}
+        return identity
+
+    def _get_account_for_tenant(self, tenant_id):
+        return '%s%s' % (self.reseller_prefix, tenant_id)
+
+    def _reseller_check(self, account, tenant_id):
+        """Check reseller prefix."""
+        return account == self._get_account_for_tenant(tenant_id)
+
+    def authorize(self, req):
+        env = req.environ
+        env_identity = env.get('keystone.identity', {})
+        tenant_id, tenant_name = env_identity.get('tenant')
+
+        try:
+            part = swift_utils.split_path(req.path, 1, 4, True)
+            version, account, container, obj = part
+        except ValueError:
+            return webob.exc.HTTPNotFound(request=req)
+
+        user_roles = env_identity.get('roles', [])
+
+        # Give unconditional access to a user with the reseller_admin
+        # role.
+        if self.reseller_admin_role in user_roles:
+            msg = 'User %s has reseller admin authorizing'
+            self.logger.debug(msg % tenant_id)
+            req.environ['swift_owner'] = True
+            return
+
+        # Check if a user tries to access an account that does not match their
+        # token
+        if not self._reseller_check(account, tenant_id):
+            log_msg = 'tenant mismatch: %s != %s' % (account, tenant_id)
+            self.logger.debug(log_msg)
+            return self.denied_response(req)
+
+        # Check the roles the user is belonging to. If the user is
+        # part of the role defined in the config variable
+        # operator_roles (like admin) then it will be
+        # promoted as an admin of the account/tenant.
+        for role in self.operator_roles.split(','):
+            role = role.strip()
+            if role in user_roles:
+                log_msg = 'allow user with role %s as account admin' % (role)
+                self.logger.debug(log_msg)
+                req.environ['swift_owner'] = True
+                return
+
+        # If user is of the same name of the tenant then make owner of it.
+        user = env_identity.get('user', '')
+        if self.is_admin and user == tenant_name:
+            req.environ['swift_owner'] = True
+            return
+
+        referrers, roles = swift_acl.parse_acl(getattr(req, 'acl', None))
+
+        authorized = self._authorize_unconfirmed_identity(req, obj, referrers,
+                                                          roles)
+        if authorized:
+            return
+        elif authorized is not None:
+            return self.denied_response(req)
+
+        # Allow ACL at individual user level (tenant:user format)
+        # For backward compatibility, check for ACL in tenant_id:user format
+        if ('%s:%s' % (tenant_name, user) in roles
+                or '%s:%s' % (tenant_id, user) in roles):
+            log_msg = 'user %s:%s or %s:%s allowed in ACL authorizing'
+            self.logger.debug(log_msg % (tenant_name, user, tenant_id, user))
+            return
+
+        # Check if we have the role in the userroles and allow it
+        for user_role in user_roles:
+            if user_role in roles:
+                log_msg = 'user %s:%s allowed in ACL: %s authorizing'
+                self.logger.debug(log_msg % (tenant_name, user, user_role))
+                return
+
+        return self.denied_response(req)
+
+    def authorize_anonymous(self, req):
+        """
+        Authorize an anonymous request.
+
+        :returns: None if authorization is granted, an error page otherwise.
+        """
+        try:
+            part = swift_utils.split_path(req.path, 1, 4, True)
+            version, account, container, obj = part
+        except ValueError:
+            return webob.exc.HTTPNotFound(request=req)
+
+        is_authoritative_authz = (account and
+                                  account.startswith(self.reseller_prefix))
+        if not is_authoritative_authz:
+            return self.denied_response(req)
+
+        referrers, roles = swift_acl.parse_acl(getattr(req, 'acl', None))
+        authorized = self._authorize_unconfirmed_identity(req, obj, referrers,
+                                                          roles)
+        if not authorized:
+            return self.denied_response(req)
+
+    def _authorize_unconfirmed_identity(self, req, obj, referrers, roles):
+        """"
+        Perform authorization for access that does not require a
+        confirmed identity.
+
+        :returns: A boolean if authorization is granted or denied.  None if
+                  a determination could not be made.
+        """
+        # Allow container sync.
+        if (req.environ.get('swift_sync_key')
+            and req.environ['swift_sync_key'] ==
+                req.headers.get('x-container-sync-key', None)
+            and 'x-timestamp' in req.headers
+            and (req.remote_addr in self.allowed_sync_hosts
+                 or swift_utils.get_remote_client(req)
+                 in self.allowed_sync_hosts)):
+            log_msg = 'allowing proxy %s for container-sync' % req.remote_addr
+            self.logger.debug(log_msg)
+            return True
+
+        # Check if referrer is allowed.
+        if swift_acl.referrer_allowed(req.referer, referrers):
+            if obj or '.rlistings' in roles:
+                log_msg = 'authorizing %s via referer ACL' % req.referrer
+                self.logger.debug(log_msg)
+                return True
+            return False
+
+    def denied_response(self, req):
+        """Deny WSGI Response.
+
+        Returns a standard WSGI response callable with the status of 403 or 401
+        depending on whether the REMOTE_USER is set or not.
+        """
+        if req.remote_user:
+            return webob.exc.HTTPForbidden(request=req)
+        else:
+            return webob.exc.HTTPUnauthorized(request=req)
+
+
+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)
+
+    def auth_filter(app):
+        return KeystoneAuth(app, conf)
+    return auth_filter

test/unit/common/middleware/test_keystoneauth.py

@@ -0,0 +1,230 @@
+# Copyright (c) 2012 OpenStack, LLC.
+#
+# 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
+import webob
+
+from swift.common.middleware import keystoneauth
+
+
+class FakeApp(object):
+    def __init__(self, status_headers_body_iter=None):
+        self.calls = 0
+        self.status_headers_body_iter = status_headers_body_iter
+        if not self.status_headers_body_iter:
+            self.status_headers_body_iter = iter([('404 Not Found', {}, '')])
+
+    def __call__(self, env, start_response):
+        self.calls += 1
+        self.request = webob.Request.blank('', environ=env)
+        if 'swift.authorize' in env:
+            resp = env['swift.authorize'](self.request)
+            if resp:
+                return resp(env, start_response)
+        status, headers, body = self.status_headers_body_iter.next()
+        return webob.Response(status=status, headers=headers,
+                              body=body)(env, start_response)
+
+
+class SwiftAuth(unittest.TestCase):
+    def setUp(self):
+        self.test_auth = keystoneauth.filter_factory({})(FakeApp())
+
+    def _make_request(self, path=None, headers=None, **kwargs):
+        if not path:
+            path = '/v1/%s/c/o' % self.test_auth._get_account_for_tenant('foo')
+        return webob.Request.blank(path, headers=headers, **kwargs)
+
+    def _get_identity_headers(self, status='Confirmed', tenant_id='1',
+                          tenant_name='acct', user='usr', role=''):
+        return dict(X_IDENTITY_STATUS=status,
+                    X_TENANT_ID=tenant_id,
+                    X_TENANT_NAME=tenant_name,
+                    X_ROLES=role,
+                    X_USER_NAME=user)
+
+    def _get_successful_middleware(self):
+        response_iter = iter([('200 OK', {}, '')])
+        return keystoneauth.filter_factory({})(FakeApp(response_iter))
+
+    def test_confirmed_identity_is_authorized(self):
+        role = self.test_auth.reseller_admin_role
+        headers = self._get_identity_headers(role=role)
+        req = self._make_request('/v1/AUTH_acct/c', headers)
+        resp = req.get_response(self._get_successful_middleware())
+        self.assertEqual(resp.status_int, 200)
+
+    def test_confirmed_identity_is_not_authorized(self):
+        headers = self._get_identity_headers()
+        req = self._make_request('/v1/AUTH_acct/c', headers)
+        resp = req.get_response(self.test_auth)
+        self.assertEqual(resp.status_int, 403)
+
+    def test_anonymous_is_authorized_for_permitted_referrer(self):
+        req = self._make_request(headers={'X_IDENTITY_STATUS': 'Invalid'})
+        req.acl = '.r:*'
+        resp = req.get_response(self._get_successful_middleware())
+        self.assertEqual(resp.status_int, 200)
+
+    def test_anonymous_is_not_authorized_for_unknown_reseller_prefix(self):
+        req = self._make_request(path='/v1/BLAH_foo/c/o',
+                                 headers={'X_IDENTITY_STATUS': 'Invalid'})
+        resp = req.get_response(self.test_auth)
+        self.assertEqual(resp.status_int, 401)
+
+    def test_blank_reseller_prefix(self):
+        conf = {'reseller_prefix': ''}
+        test_auth = keystoneauth.filter_factory(conf)(FakeApp())
+        account = tenant_id = 'foo'
+        self.assertTrue(test_auth._reseller_check(account, tenant_id))
+
+    def test_override_asked_for_but_not_allowed(self):
+        conf = {'allow_overrides': 'false'}
+        self.test_auth = keystoneauth.filter_factory(conf)(FakeApp())
+        req = self._make_request('/v1/AUTH_account',
+                                 environ={'swift.authorize_override': True})
+        resp = req.get_response(self.test_auth)
+        self.assertEquals(resp.status_int, 401)
+
+    def test_override_asked_for_and_allowed(self):
+        conf = {'allow_overrides': 'true'}
+        self.test_auth = keystoneauth.filter_factory(conf)(FakeApp())
+        req = self._make_request('/v1/AUTH_account',
+                                 environ={'swift.authorize_override': True})
+        resp = req.get_response(self.test_auth)
+        self.assertEquals(resp.status_int, 404)
+
+    def test_override_default_allowed(self):
+        req = self._make_request('/v1/AUTH_account',
+                                 environ={'swift.authorize_override': True})
+        resp = req.get_response(self.test_auth)
+        self.assertEquals(resp.status_int, 404)
+
+
+class TestAuthorize(unittest.TestCase):
+    def setUp(self):
+        self.test_auth = keystoneauth.filter_factory({})(FakeApp())
+
+    def _make_request(self, path, **kwargs):
+        return webob.Request.blank(path, **kwargs)
+
+    def _get_account(self, identity=None):
+        if not identity:
+            identity = self._get_identity()
+        return self.test_auth._get_account_for_tenant(identity['tenant'][0])
+
+    def _get_identity(self, tenant_id='tenant_id',
+                      tenant_name='tenant_name', user='user', roles=None):
+        if not roles:
+            roles = []
+        return dict(tenant=(tenant_id, tenant_name), user=user, roles=roles)
+
+    def _check_authenticate(self, account=None, identity=None, headers=None,
+                            exception=None, acl=None, env=None, path=None):
+        if not identity:
+            identity = self._get_identity()
+        if not account:
+            account = self._get_account(identity)
+        if not path:
+            path = '/v1/%s/c' % account
+        default_env = {'keystone.identity': identity,
+                       'REMOTE_USER': identity['tenant']}
+        if env:
+            default_env.update(env)
+        req = self._make_request(path, headers=headers, environ=default_env)
+        req.acl = acl
+        result = self.test_auth.authorize(req)
+        if exception:
+            self.assertTrue(isinstance(result, exception))
+        else:
+            self.assertTrue(result is None)
+        return req
+
+    def test_authorize_fails_for_unauthorized_user(self):
+        self._check_authenticate(exception=webob.exc.HTTPForbidden)
+
+    def test_authorize_fails_for_invalid_reseller_prefix(self):
+        self._check_authenticate(account='BLAN_a',
+                                 exception=webob.exc.HTTPForbidden)
+
+    def test_authorize_succeeds_for_reseller_admin(self):
+        roles = [self.test_auth.reseller_admin_role]
+        identity = self._get_identity(roles=roles)
+        req = self._check_authenticate(identity=identity)
+        self.assertTrue(req.environ.get('swift_owner'))
+
+    def test_authorize_succeeds_as_owner_for_operator_role(self):
+        roles = self.test_auth.operator_roles.split(',')[0]
+        identity = self._get_identity(roles=roles)
+        req = self._check_authenticate(identity=identity)
+        self.assertTrue(req.environ.get('swift_owner'))
+
+    def _check_authorize_for_tenant_owner_match(self, exception=None):
+        identity = self._get_identity()
+        identity['user'] = identity['tenant'][1]
+        req = self._check_authenticate(identity=identity, exception=exception)
+        expected = bool(exception is None)
+        self.assertEqual(bool(req.environ.get('swift_owner')), expected)
+
+    def test_authorize_succeeds_as_owner_for_tenant_owner_match(self):
+        self.test_auth.is_admin = True
+        self._check_authorize_for_tenant_owner_match()
+
+    def test_authorize_fails_as_owner_for_tenant_owner_match(self):
+        self.test_auth.is_admin = False
+        self._check_authorize_for_tenant_owner_match(
+            exception=webob.exc.HTTPForbidden)
+
+    def test_authorize_succeeds_for_container_sync(self):
+        env = {'swift_sync_key': 'foo', 'REMOTE_ADDR': '127.0.0.1'}
+        headers = {'x-container-sync-key': 'foo', 'x-timestamp': None}
+        self._check_authenticate(env=env, headers=headers)
+
+    def test_authorize_fails_for_invalid_referrer(self):
+        env = {'HTTP_REFERER': 'http://invalid.com/index.html'}
+        self._check_authenticate(acl='.r:example.com', env=env,
+                                 exception=webob.exc.HTTPForbidden)
+
+    def test_authorize_fails_for_referrer_without_rlistings(self):
+        env = {'HTTP_REFERER': 'http://example.com/index.html'}
+        self._check_authenticate(acl='.r:example.com', env=env,
+                                 exception=webob.exc.HTTPForbidden)
+
+    def test_authorize_succeeds_for_referrer_with_rlistings(self):
+        env = {'HTTP_REFERER': 'http://example.com/index.html'}
+        self._check_authenticate(acl='.r:example.com,.rlistings', env=env)
+
+    def test_authorize_succeeds_for_referrer_with_obj(self):
+        path = '/v1/%s/c/o' % self._get_account()
+        env = {'HTTP_REFERER': 'http://example.com/index.html'}
+        self._check_authenticate(acl='.r:example.com', env=env, path=path)
+
+    def test_authorize_succeeds_for_user_role_in_roles(self):
+        acl = 'allowme'
+        identity = self._get_identity(roles=[acl])
+        self._check_authenticate(identity=identity, acl=acl)
+
+    def test_authorize_succeeds_for_tenant_name_user_in_roles(self):
+        identity = self._get_identity()
+        acl = '%s:%s' % (identity['tenant'][1], identity['user'])
+        self._check_authenticate(identity=identity, acl=acl)
+
+    def test_authorize_succeeds_for_tenant_id_user_in_roles(self):
+        identity = self._get_identity()
+        acl = '%s:%s' % (identity['tenant'][0], identity['user'])
+        self._check_authenticate(identity=identity, acl=acl)
+
+if __name__ == '__main__':
+    unittest.main()