Merge "Add a project scope read-only role to keystoneauth"

etc/proxy-server.conf-sample

@@ -508,6 +508,11 @@ user_test5_tester5 = testing5 service
 # only do not modify the cluster.
 # By default the list of reader roles is empty.
 # system_reader_roles =
+#
+# This is a reader role scoped for a Keystone project.
+# An identity that has this role can read anything in a project, so it is
+# basically a swiftoperator, but read-only.
+# project_reader_roles =
 
 [filter:s3api]
 use = egg:swift#s3api

swift/common/middleware/keystoneauth.py

@@ -178,7 +178,8 @@ class KeystoneAuth(object):
             config_read_reseller_options(conf,
                                          dict(operator_roles=['admin',
                                                               'swiftoperator'],
-                                              service_roles=[]))
+                                              service_roles=[],
+                                              project_reader_roles=[]))
         self.reseller_admin_role = conf.get('reseller_admin_role',
                                             'ResellerAdmin').lower()
         self.system_reader_roles = {role.lower() for role in list_from_csv(
@@ -418,15 +419,14 @@ class KeystoneAuth(object):
         user_service_roles = [r.lower() for r in env_identity.get(
                               'service_roles', [])]
 
-        # Give unconditional access to a user with the reseller_admin
-        # role.
+        # 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
 
-        # The system_reader_role is almost as good as reseller_admin.
+        # Being in system_reader_roles is almost as good as reseller_admin.
         if self.system_reader_roles.intersection(user_roles):
             # Note that if a system reader is trying to write, we're letting
             # the request fall on other access checks below. This way,
@@ -501,6 +501,20 @@ class KeystoneAuth(object):
             req.environ['swift_owner'] = True
             return
 
+        # The project_reader_roles is almost as good as operator_roles. But
+        # it does not work with service tokens and does not get 'swift_owner'.
+        # And, it only serves GET requests, obviously.
+        project_reader_roles = self.account_rules[account_prefix][
+            'project_reader_roles']
+        have_reader_role = set(project_reader_roles).intersection(
+            set(user_roles))
+        if have_reader_role:
+            if req.method in ('GET', 'HEAD'):
+                msg = 'User %s with role(s) %s has project reader authorizing'
+                self.logger.debug(msg, tenant_id,
+                                  ','.join(project_reader_roles))
+                return
+
         if acl_authorized is not None:
             return self.denied_response(req)
 

test/unit/common/middleware/test_keystoneauth.py

@@ -1510,13 +1510,15 @@ class TestSetProjectDomain(BaseTestAuthorize):
                                         sysmeta_project_domain_id='test_id')
 
 
-class TestAuthorizeReader(BaseTestAuthorizeCheck):
+class TestAuthorizeReaderSystem(BaseTestAuthorizeCheck):
 
     system_reader_role_1 = 'compliance'
     system_reader_role_2 = 'integrity'
 
     # This cannot be in SetUp because it takes arguments from tests.
     def _setup(self, system_reader_roles):
+        # We could rifle in the KeystoneAuth internals and tweak the list,
+        # but to create the middleware fresh is a clean, future-resistant way.
         self.test_auth = keystoneauth.filter_factory(
             {}, system_reader_roles=system_reader_roles)(FakeApp())
         self.test_auth.logger = debug_logger()
@@ -1524,8 +1526,6 @@ class TestAuthorizeReader(BaseTestAuthorizeCheck):
     # Zero test: make sure that reader role has no default access
     # when not in the list of system_reader_roles[].
     def test_reader_none(self):
-        # We could rifle in the KeystoneAuth internals and tweak the list,
-        # but to create the middleware fresh is a clean, future-resistant way.
         self._setup(None)
         identity = self._get_identity(roles=[self.system_reader_role_1])
         self._check_authenticate(exception=HTTP_FORBIDDEN,
@@ -1569,10 +1569,44 @@ class TestAuthorizeReader(BaseTestAuthorizeCheck):
                                  env={'REQUEST_METHOD': 'PUT'})
 
 
+class TestAuthorizeReaderProject(BaseTestAuthorizeCheck):
+
+    project_reader_role_1 = 'rdr1'
+    project_reader_role_2 = 'rdr2'
+
+    # This cannot be in SetUp because it takes arguments from tests.
+    def _setup(self, project_reader_roles):
+        self.test_auth = keystoneauth.filter_factory(
+            {}, project_reader_roles=project_reader_roles)(FakeApp())
+        self.test_auth.logger = debug_logger()
+
+    # The project reader tests do not have a zero test because it literally
+    # is the same code as system reader tests already run. See above.
+
+    # Reading is what a reader does.
+    def test_reader_get(self):
+        self._setup("%s, %s" %
+                    (self.project_reader_role_1, self.project_reader_role_2))
+        identity = self._get_identity(roles=[self.project_reader_role_2])
+        self._check_authenticate(identity=identity)
+
+    # Writing would otherwise be allowed, but not for a reader.
+    def test_reader_put(self):
+        self._setup(self.project_reader_role_1)
+        identity = self._get_identity(roles=[self.project_reader_role_1])
+        self._check_authenticate(exception=HTTP_FORBIDDEN,
+                                 identity=identity,
+                                 env={'REQUEST_METHOD': 'PUT'})
+        self._check_authenticate(exception=HTTP_FORBIDDEN,
+                                 identity=identity,
+                                 env={'REQUEST_METHOD': 'POST'})
+
+
 class ResellerInInfo(unittest.TestCase):
 
     def setUp(self):
         self.default_rules = {'operator_roles': ['admin', 'swiftoperator'],
+                              'project_reader_roles': [],
                               'service_roles': []}
 
     def test_defaults(self):