Issue unscoped token if user's default project is invalid

Implemented blueprint unscoped-token-for-invalid-default-project

DocImpact

Change-Id: I2b1b0a8dbb12350446cec6cdf319e3e50856be32

keystone/auth/controllers.py

@@ -320,14 +320,49 @@ class Auth(controller.V3Controller):
         # fill in default_project_id if it is available
         try:
             user_ref = self.identity_api.get_user(auth_context['user_id'])
-            default_project_id = user_ref.get('default_project_id')
-            if default_project_id:
-                auth_info.set_scope(domain_id=None,
-                                    project_id=default_project_id)
         except exception.UserNotFound as e:
             LOG.exception(e)
             raise exception.Unauthorized(e)
 
+        default_project_id = user_ref.get('default_project_id')
+        if not default_project_id:
+            # User has no default project. He shall get an unscoped token.
+            return
+
+        # make sure user's default project is legit before scoping to it
+        try:
+            default_project_ref = self.assignment_api.get_project(
+                default_project_id)
+            default_project_domain_ref = self.assignment_api.get_domain(
+                default_project_ref['domain_id'])
+            if (default_project_ref.get('enabled', True) and
+                    default_project_domain_ref.get('enabled', True)):
+                if self.assignment_api.get_roles_for_user_and_project(
+                        user_ref['id'], default_project_id):
+                    auth_info.set_scope(project_id=default_project_id)
+                else:
+                    msg = _("User %(user_id)s doesn't have access to"
+                            " default project %(project_id)s. The token will"
+                            " be unscoped rather than scoped to the project.")
+                    LOG.warning(msg,
+                                {'user_id': user_ref['id'],
+                                 'project_id': default_project_id})
+            else:
+                msg = _("User %(user_id)s's default project %(project_id)s is"
+                        " disabled. The token will be unscoped rather than"
+                        " scoped to the project.")
+                LOG.warning(msg,
+                            {'user_id': user_ref['id'],
+                             'project_id': default_project_id})
+        except (exception.ProjectNotFound, exception.DomainNotFound):
+            # default project or default project domain doesn't exist,
+            # will issue unscoped token instead
+            msg = _("User %(user_id)s's default project %(project_id)s not"
+                    " found. The token will be unscoped rather than"
+                    " scoped to the project.")
+            LOG.warning(msg, {'user_id': user_ref['id'],
+                              'project_id': default_project_id})
+
     def authenticate(self, context, auth_info, auth_context):
         """Authenticate user."""
 

keystone/tests/test_v3.py

@@ -263,6 +263,21 @@ class RestfulTestCase(rest.RestfulTestCase):
 
         return ref
 
+    def create_new_default_project_for_user(self, user_id, domain_id,
+                                            enable_project=True):
+        ref = self.new_project_ref(domain_id=domain_id)
+        ref['enabled'] = enable_project
+        r = self.post('/projects', body={'project': ref})
+        project = self.assertValidProjectResponse(r, ref)
+        # set the user's preferred project
+        body = {'user': {'default_project_id': project['id']}}
+        r = self.patch('/users/%(user_id)s' % {
+            'user_id': user_id},
+            body=body)
+        self.assertValidUserResponse(r)
+
+        return project
+
     def admin_request(self, *args, **kwargs):
         """Translates XML responses to dicts.
 

keystone/tests/test_v3_auth.py

@@ -1247,28 +1247,6 @@ class TestAuthJSON(test_v3.RestfulTestCase):
         self.assertEqual(r.result['token']['project']['id'],
                          self.project['id'])
 
-    def test_default_project_id_scoped_token_with_user_id_401(self):
-        # create a second project to work with
-        ref = self.new_project_ref(domain_id=self.domain['id'])
-        del ref['id']
-        r = self.post('/projects', body={'project': ref})
-        project = self.assertValidProjectResponse(r, ref)
-
-        # set the user's preferred project without having authz on that project
-        body = {'user': {'default_project_id': project['id']}}
-        r = self.patch('/users/%(user_id)s' % {
-            'user_id': self.user['id']},
-            body=body)
-        self.assertValidUserResponse(r)
-
-        # attempt to authenticate without requesting a project
-        # the default_project_id should be the assumed scope of the request,
-        # and fail because the user doesn't have explicit authz on that scope
-        auth_data = self.build_authentication_request(
-            user_id=self.user['id'],
-            password=self.user['password'])
-        self.post('/auth/tokens', body=auth_data, expected_status=401)
-
     def test_project_id_scoped_token_with_user_id_401(self):
         project_id = uuid.uuid4().hex
         project = self.new_project_ref(domain_id=self.domain_id)
@@ -1778,6 +1756,61 @@ class TestAuthJSON(test_v3.RestfulTestCase):
 
         self.post('/auth/tokens', body=auth_data, expected_status=401)
 
+    def test_disabled_default_project_result_in_unscoped_token(self):
+        # create a disabled project to work with
+        project = self.create_new_default_project_for_user(
+            self.user['id'], self.domain_id, enable_project=False)
+
+        # assign a role to user for the new project
+        self.assignment_api.add_role_to_user_and_project(self.user['id'],
+                                                         project['id'],
+                                                         self.role_id)
+
+        # attempt to authenticate without requesting a project
+        auth_data = self.build_authentication_request(
+            user_id=self.user['id'],
+            password=self.user['password'])
+        r = self.post('/auth/tokens', body=auth_data)
+        self.assertValidUnscopedTokenResponse(r)
+
+    def test_disabled_default_project_domain_result_in_unscoped_token(self):
+        domain_ref = self.new_domain_ref()
+        r = self.post('/domains', body={'domain': domain_ref})
+        domain = self.assertValidDomainResponse(r, domain_ref)
+
+        project = self.create_new_default_project_for_user(
+            self.user['id'], domain['id'])
+
+        # assign a role to user for the new project
+        self.assignment_api.add_role_to_user_and_project(self.user['id'],
+                                                         project['id'],
+                                                         self.role_id)
+
+        # now disable the project domain
+        body = {'domain': {'enabled': False}}
+        r = self.patch('/domains/%(domain_id)s' % {'domain_id': domain['id']},
+                       body=body)
+        self.assertValidDomainResponse(r)
+
+        # attempt to authenticate without requesting a project
+        auth_data = self.build_authentication_request(
+            user_id=self.user['id'],
+            password=self.user['password'])
+        r = self.post('/auth/tokens', body=auth_data)
+        self.assertValidUnscopedTokenResponse(r)
+
+    def test_no_access_to_default_project_result_in_unscoped_token(self):
+        # create a disabled project to work with
+        self.create_new_default_project_for_user(self.user['id'],
+                                                 self.domain_id)
+
+        # attempt to authenticate without requesting a project
+        auth_data = self.build_authentication_request(
+            user_id=self.user['id'],
+            password=self.user['password'])
+        r = self.post('/auth/tokens', body=auth_data)
+        self.assertValidUnscopedTokenResponse(r)
+
 
 class TestAuthXML(TestAuthJSON):
     content_type = 'xml'