Merge "Move auth_token middleware from admin user to an RBAC policy"

etc/policy.json

@@ -1,5 +1,7 @@
 {
     "admin_required": [["role:admin"], ["is_admin:1"]],
+    "service_role": [["role:service"]],
+    "service_or_admin": [["rule:admin_required"], ["rule:service_role"]],
     "owner" : [["user_id:%(user_id)s"]],
     "admin_or_owner": [["rule:admin_required"], ["rule:owner"]],
 
@@ -71,8 +73,9 @@
     "identity:delete_policy": [["rule:admin_required"]],
 
     "identity:check_token": [["rule:admin_required"]],
-    "identity:validate_token": [["rule:admin_required"]],
-    "identity:revocation_list": [["rule:admin_required"]],
+    "identity:validate_token": [["rule:service_or_admin"]],
+    "identity:validate_token_head": [["rule:service_or_admin"]],
+    "identity:revocation_list": [["rule:service_or_admin"]],
     "identity:revoke_token": [["rule:admin_required"],
                               ["user_id:%(user_id)s"]],
 

keystone/common/controller.py

@@ -87,7 +87,7 @@ def flatten(d, parent_key=''):
 def protected(f):
     """Wraps API calls with role based access controls (RBAC)."""
     @functools.wraps(f)
-    def wrapper(self, context, **kwargs):
+    def wrapper(self, context, *args, **kwargs):
         if 'is_admin' in context and context['is_admin']:
             LOG.warning(_('RBAC: Bypassing authorization'))
         else:
@@ -100,7 +100,7 @@ def wrapper(self, context, **kwargs):
             self.policy_api.enforce(context, creds, action, flatten(kwargs))
             LOG.debug(_('RBAC: Authorization granted'))
 
-        return f(self, context, **kwargs)
+        return f(self, context, *args, **kwargs)
     return wrapper
 
 

keystone/token/controllers.py

@@ -457,8 +457,6 @@ def _get_token_ref(self, context, token_id, belongs_to=None):
         Optionally, limited to a token owned by a specific tenant.
 
         """
-        # TODO(termie): this stuff should probably be moved to middleware
-        self.assert_admin(context)
         data = self.token_api.get_token(context=context,
                                         token_id=token_id)
         if belongs_to:
@@ -510,7 +508,7 @@ def _assert_default_domain(self, context, token_ref):
                 if project_ref['domain_id'] != DEFAULT_DOMAIN_ID:
                     raise exception.Unauthorized(msg)
 
-    # admin only
+    @controller.protected
     def validate_token_head(self, context, token_id):
         """Check that a token is valid.
 
@@ -524,7 +522,7 @@ def validate_token_head(self, context, token_id):
         assert token_ref
         self._assert_default_domain(context, token_ref)
 
-    # admin only
+    @controller.protected
     def validate_token(self, context, token_id):
         """Check that a token is valid.
 
@@ -562,8 +560,8 @@ def delete_token(self, context, token_id):
         self.assert_admin(context)
         self.token_api.delete_token(context=context, token_id=token_id)
 
+    @controller.protected
     def revocation_list(self, context, auth=None):
-        self.assert_admin(context)
         tokens = self.token_api.list_revoked_tokens(context)
 
         for t in tokens:

tests/default_fixtures.py

@@ -45,6 +45,12 @@
         'description': 'description',
         'enabled': True,
         'domain_id': DEFAULT_DOMAIN_ID
+    }, {
+        'id': 'service',
+        'name': 'service',
+        'description': 'description',
+        'enabled': True,
+        'domain_id': DEFAULT_DOMAIN_ID
     }
 ]
 
@@ -115,8 +121,12 @@
     }, {
         'id': 'writer',
         'name': 'Writer',
+    }, {
+        'id': 'service',
+        'name': 'Service',
     }
 
+
 ]
 
 DOMAINS = [

tests/test_backend.py

@@ -1619,7 +1619,7 @@ def test_list_domains(self):
 
     def test_list_projects(self):
         projects = self.identity_api.list_projects()
-        self.assertEquals(len(projects), 3)
+        self.assertEquals(len(projects), 4)
         project_ids = []
         for project in projects:
             project_ids.append(project.get('id'))

tests/test_content_types.py

@@ -214,8 +214,10 @@ def public_request(self, **kwargs):
     def admin_request(self, **kwargs):
         return self._request(app=self.admin_app, **kwargs)
 
-    def get_scoped_token(self):
+    def get_scoped_token(self, tenant_id=None):
         """Convenience method so that we can test authenticated requests."""
+        if not tenant_id:
+            tenant_id = self.tenant_bar['id']
         r = self.public_request(
             method='POST',
             path='/v2.0/tokens',
@@ -225,7 +227,7 @@ def get_scoped_token(self):
                         'username': self.user_foo['name'],
                         'password': self.user_foo['password'],
                     },
-                    'tenantId': self.tenant_bar['id'],
+                    'tenantId': tenant_id,
                 },
             })
         return self._get_token_id(r)
@@ -387,6 +389,18 @@ def test_validate_token(self):
             token=token)
         self.assertValidAuthenticationResponse(r)
 
+    def test_validate_token_service_role(self):
+        self.metadata_foobar = self.identity_api.update_metadata(
+            self.user_foo['id'],
+            self.tenant_service['id'],
+            dict(roles=[self.role_service['id']]))
+
+        token = self.get_scoped_token(tenant_id='service')
+        r = self.admin_request(
+            path='/v2.0/tokens/%s' % token,
+            token=token)
+        self.assertValidAuthenticationResponse(r)
+
     def test_validate_token_belongs_to(self):
         token = self.get_scoped_token()
         path = ('/v2.0/tokens/%s?belongsTo=%s' % (token,