Merge "add belongs_to check"

2 files changed, 47 insertions, 3 deletions

diff --git a/keystone/token/controllers.py b/keystone/token/controllers.py
index 07a94451..06a1fe64 100644
--- a/keystone/token/controllers.py
+++ b/keystone/token/controllers.py
@@ -462,9 +462,16 @@ class Auth(controller.V2Controller):
         """
         # TODO(termie): this stuff should probably be moved to middleware
         self.assert_admin(context)
-        token_ref = self.token_api.get_token(context=context,
-                                             token_id=token_id)
-        return token_ref
+        data = self.token_api.get_token(context=context,
+                                        token_id=token_id)
+        if belongs_to:
+            if data.get('tenant') is None:
+                raise exception.Unauthorized(
+                    _('Token does not belong to specified tenant.'))
+            if data['tenant'].get('id') != belongs_to:
+                raise exception.Unauthorized(
+                    _('Token does not belong to specified tenant.'))
+        return data
 
     # admin only
     def validate_token_head(self, context, token_id):
diff --git a/tests/test_auth.py b/tests/test_auth.py
index 1436d8ec..dd729b73 100644
--- a/tests/test_auth.py
+++ b/tests/test_auth.py
@@ -334,6 +334,43 @@ class AuthWithToken(AuthTest):
         self.assertNotIn(role_foo_domain1['id'], roles)
         self.assertNotIn(role_group_domain1['id'], roles)
 
+    def test_belongs_to_no_tenant(self):
+        r = self.controller.authenticate(
+            {},
+            auth={
+                'passwordCredentials': {
+                    'username': self.user_foo['name'],
+                    'password': self.user_foo['password']
+                }
+            })
+        unscoped_token_id = r['access']['token']['id']
+        self.assertRaises(
+            exception.Unauthorized,
+            self.controller.validate_token,
+            dict(is_admin=True, query_string={'belongsTo': 'BAR'}),
+            token_id=unscoped_token_id)
+
+    def test_belongs_to(self):
+        body_dict = _build_user_auth(
+            username='FOO',
+            password='foo2',
+            tenant_name="BAR")
+
+        scoped_token = self.controller.authenticate({}, body_dict)
+        scoped_token_id = scoped_token['access']['token']['id']
+
+        self.assertRaises(
+            exception.Unauthorized,
+            self.controller.validate_token,
+            dict(is_admin=True, query_string={'belongsTo': 'me'}),
+            token_id=scoped_token_id)
+
+        self.assertRaises(
+            exception.Unauthorized,
+            self.controller.validate_token,
+            dict(is_admin=True, query_string={'belongsTo': 'BAR'}),
+            token_id=scoped_token_id)
+
 
 class AuthWithPasswordCredentials(AuthTest):
     def setUp(self):