Merge pull request #2826 from bacongobbler/api-loopholes

fix(controller): disallow unauthorized users from modifying apps

Author: Matthew Fisher

controller/api/__init__.py

@@ -2,4 +2,4 @@
 The **api** Django app presents a RESTful web API for interacting with the **deis** system.
 """
 
-__version__ = '1.1.1'
+__version__ = '1.1.2'

controller/api/tests/test_app.py

@@ -253,6 +253,39 @@ def test_run_without_release_should_error(self):
         self.assertEqual(response.data, "No build associated with this release "
                                         "to run this command")
 
+    def test_unauthorized_user_cannot_see_app(self):
+        """
+        An unauthorized user should not be able to access an app's resources.
+
+        Since an unauthorized user should not know about the application at all, these
+        requests should return a 404.
+        """
+        app_id = 'autotest'
+        base_url = '/v1/apps'
+        body = {'id': app_id}
+        response = self.client.post(base_url, json.dumps(body), content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(self.token))
+        unauthorized_user = User.objects.get(username='autotest2')
+        unauthorized_token = Token.objects.get(user=unauthorized_user).key
+        url = '{}/{}/run'.format(base_url, app_id)
+        body = {'command': 'foo'}
+        response = self.client.post(url, json.dumps(body), content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(unauthorized_token))
+        self.assertEqual(response.status_code, 404)
+        url = '{}/{}/logs'.format(base_url, app_id)
+        response = self.client.get(url, HTTP_AUTHORIZATION='token {}'.format(unauthorized_token))
+        self.assertEqual(response.status_code, 404)
+
+    def test_app_info_not_showing_wrong_app(self):
+        app_id = 'autotest'
+        base_url = '/v1/apps'
+        body = {'id': app_id}
+        response = self.client.post(base_url, json.dumps(body), content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(self.token))
+        url = '{}/foo'.format(base_url)
+        response = self.client.get(url, HTTP_AUTHORIZATION='token {}'.format(self.token))
+        self.assertEqual(response.status_code, 404)
+
 
 FAKE_LOG_DATA = """
 2013-08-15 12:41:25 [33454] [INFO] Starting gunicorn 17.5

controller/api/tests/test_build.py

@@ -208,16 +208,22 @@ def test_admin_can_create_builds_on_other_apps(self):
                          response.data['app'], response.data['uuid'][:7]))
 
     @mock.patch('requests.post', mock_import_repository_task)
-    def test_unauthorized_user_cannot_create_build(self):
-        """An unauthorized user should not be able to create builds for other apps."""
+    def test_unauthorized_user_cannot_modify_build(self):
+        """
+        An unauthorized user should not be able to modify other builds.
+
+        Since an unauthorized user should not know about the application at all, these
+        requests should return a 404.
+        """
+        app_id = 'autotest'
         url = '/v1/apps'
-        response = self.client.post(url, HTTP_AUTHORIZATION='token {}'.format(self.token))
-        app_id = response.data['id']
-        # attempt to create a build as a malicious user
-        evil_user = User.objects.get(username='autotest2')
-        evil_token = Token.objects.get(user=evil_user).key
-        url = "/v1/apps/{app_id}/builds".format(**locals())
-        body = {'image': 'eeeeeevillllll'}
+        body = {'id': app_id}
+        response = self.client.post(url, json.dumps(body), content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(self.token))
+        unauthorized_user = User.objects.get(username='autotest2')
+        unauthorized_token = Token.objects.get(user=unauthorized_user).key
+        url = '{}/{}/builds'.format(url, app_id)
+        body = {'image': 'foo'}
         response = self.client.post(url, json.dumps(body), content_type='application/json',
-                                    HTTP_AUTHORIZATION='token {}'.format(evil_token))
-        self.assertEqual(response.status_code, 403)
+                                    HTTP_AUTHORIZATION='token {}'.format(unauthorized_token))
+        self.assertEqual(response.status_code, 404)

controller/api/tests/test_config.py

@@ -432,3 +432,23 @@ def test_config_owner_is_requesting_user(self):
         """
         response = self.test_admin_can_create_config_on_other_apps()
         self.assertEqual(response.data['owner'], self.user.username)
+
+    def test_unauthorized_user_cannot_modify_config(self):
+        """
+        An unauthorized user should not be able to modify other config.
+
+        Since an unauthorized user should not know about the application at all, these
+        requests should return a 404.
+        """
+        app_id = 'autotest'
+        base_url = '/v1/apps'
+        body = {'id': app_id}
+        response = self.client.post(base_url, json.dumps(body), content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(self.token))
+        unauthorized_user = User.objects.get(username='autotest2')
+        unauthorized_token = Token.objects.get(user=unauthorized_user).key
+        url = '{}/{}/config'.format(base_url, app_id)
+        body = {'values': {'FOO': 'bar'}}
+        response = self.client.post(url, json.dumps(body), content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(unauthorized_token))
+        self.assertEqual(response.status_code, 404)

controller/api/tests/test_domain.py

@@ -59,15 +59,6 @@ def test_manage_domain_invalid_app(self):
                                    HTTP_AUTHORIZATION='token {}'.format(self.token))
         self.assertEqual(response.status_code, 404)
 
-    def test_manage_domain_perms_on_app(self):
-        user = User.objects.get(username='autotest2')
-        token = Token.objects.get(user=user).key
-        url = '/v1/apps/{app_id}/domains'.format(app_id=self.app_id)
-        body = {'domain': 'test-domain2.example.com'}
-        response = self.client.post(url, json.dumps(body), content_type='application/json',
-                                    HTTP_AUTHORIZATION='token {}'.format(token))
-        self.assertEqual(response.status_code, 201)
-
     def test_manage_domain_invalid_domain(self):
         url = '/v1/apps/{app_id}/domains'.format(app_id=self.app_id)
         body = {'domain': 'this_is_an.invalid.domain'}
@@ -107,3 +98,23 @@ def test_admin_can_add_domains_to_other_apps(self):
         response = self.client.post(url, json.dumps(body), content_type='application/json',
                                     HTTP_AUTHORIZATION='token {}'.format(self.token))
         self.assertEqual(response.status_code, 201)
+
+    def test_unauthorized_user_cannot_modify_domain(self):
+        """
+        An unauthorized user should not be able to modify other domains.
+
+        Since an unauthorized user should not know about the application at all, these
+        requests should return a 404.
+        """
+        app_id = 'autotest'
+        url = '/v1/apps'
+        body = {'id': app_id}
+        response = self.client.post(url, json.dumps(body), content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(self.token))
+        unauthorized_user = User.objects.get(username='autotest2')
+        unauthorized_token = Token.objects.get(user=unauthorized_user).key
+        url = '{}/{}/domains'.format(url, app_id)
+        body = {'domain': 'example.com'}
+        response = self.client.post(url, json.dumps(body), content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(unauthorized_token))
+        self.assertEqual(response.status_code, 404)

controller/api/tests/test_perm.py

@@ -159,7 +159,7 @@ def test_create(self):
         self.assertEqual(len(response.data['results']), 1)
         # check that user 2 can't see any of the app's builds, configs,
         # containers, limits, or releases
-        for model in ['builds', 'config', 'containers', 'limits', 'releases']:
+        for model in ['builds', 'config', 'containers', 'releases']:
             response = self.client.get("/v1/apps/{}/{}/".format(app_id, model),
                                        HTTP_AUTHORIZATION='token {}'.format(self.token2))
             self.assertEqual(response.data['detail'], 'Not found')
@@ -192,7 +192,7 @@ def test_create_errors(self):
         body = {'username': 'autotest-2'}
         response = self.client.post(url, json.dumps(body), content_type='application/json',
                                     HTTP_AUTHORIZATION='token {}'.format(self.token2))
-        self.assertEqual(response.status_code, 403)
+        self.assertEqual(response.status_code, 404)
 
     def test_delete(self):
         # give user 2 permission to user 1's app
@@ -211,7 +211,7 @@ def test_delete(self):
         url = "/v1/apps/{}/perms/{}".format(app_id, 'autotest-2')
         response = self.client.delete(url, content_type='application/json',
                                       HTTP_AUTHORIZATION='token {}'.format(self.token2))
-        self.assertEqual(response.status_code, 403)
+        self.assertEqual(response.status_code, 404)
         self.assertIsNone(response.data)
         # delete permission to user 1's app
         response = self.client.delete(url, content_type='application/json',
@@ -257,4 +257,24 @@ def test_list_errors(self):
         response = self.client.get(
             "/v1/apps/{}/perms".format(app_id), content_type='application/json',
             HTTP_AUTHORIZATION='token {}'.format(self.token2))
-        self.assertEqual(response.status_code, 403)
+        self.assertEqual(response.status_code, 404)
+
+    def test_unauthorized_user_cannot_modify_perms(self):
+        """
+        An unauthorized user should not be able to modify other apps' permissions.
+
+        Since an unauthorized user should not know about the application at all, these
+        requests should return a 404.
+        """
+        app_id = 'autotest'
+        url = '/v1/apps'
+        body = {'id': app_id}
+        response = self.client.post(url, json.dumps(body), content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(self.token))
+        unauthorized_user = self.user2
+        unauthorized_token = self.token2
+        url = '{}/{}/perms'.format(url, app_id)
+        body = {'username': unauthorized_user.username}
+        response = self.client.post(url, json.dumps(body), content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(unauthorized_token))
+        self.assertEqual(response.status_code, 404)

controller/api/tests/test_release.py

@@ -244,3 +244,29 @@ def test_admin_can_create_release(self):
         self.assertEqual(response.status_code, 200)
         # account for the config release as well
         self.assertEqual(response.data['count'], 2)
+
+    @mock.patch('requests.post', mock_import_repository_task)
+    def test_unauthorized_user_cannot_modify_release(self):
+        """
+        An unauthorized user should not be able to modify other releases.
+
+        Since an unauthorized user should not know about the application at all, these
+        requests should return a 404.
+        """
+        app_id = 'autotest'
+        url = '/v1/apps'
+        body = {'id': app_id}
+        response = self.client.post(url, json.dumps(body), content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(self.token))
+        # update config to roll a new release
+        url = '/v1/apps/{app_id}/config'.format(**locals())
+        body = {'values': json.dumps({'NEW_URL1': 'http://localhost:8080/'})}
+        response = self.client.post(
+            url, json.dumps(body), content_type='application/json',
+            HTTP_AUTHORIZATION='token {}'.format(self.token))
+        unauthorized_user = User.objects.get(username='autotest2')
+        unauthorized_token = Token.objects.get(user=unauthorized_user).key
+        # try to rollback
+        url = '{}/{}/releases/rollback'.format(url, app_id)
+        response = self.client.post(url, HTTP_AUTHORIZATION='token {}'.format(unauthorized_token))
+        self.assertEqual(response.status_code, 404)

controller/api/views.py

@@ -104,15 +104,15 @@ def list(self, request, **kwargs):
         if request.user != app.owner and \
                 not request.user.has_perm(perm_name, app) and \
                 not request.user.is_superuser:
-            return Response(status=status.HTTP_403_FORBIDDEN)
+            return Response(status=status.HTTP_404_NOT_FOUND)
         usernames = [u.username for u in get_users_with_perms(app)
                      if u.has_perm(perm_name, app)]
         return Response({'users': usernames})
 
     def create(self, request, **kwargs):
         app = get_object_or_404(self.model, id=kwargs['id'])
         if request.user != app.owner and not request.user.is_superuser:
-            return Response(status=status.HTTP_403_FORBIDDEN)
+            return Response(status=status.HTTP_404_NOT_FOUND)
         user = get_object_or_404(User, username=request.DATA['username'])
         assign_perm(self.perm, user, app)
         models.log_event(app, "User {} was granted access to {}".format(user, app))
@@ -121,7 +121,7 @@ def create(self, request, **kwargs):
     def destroy(self, request, **kwargs):
         app = get_object_or_404(self.model, id=kwargs['id'])
         if request.user != app.owner and not request.user.is_superuser:
-            return Response(status=status.HTTP_403_FORBIDDEN)
+            return Response(status=status.HTTP_404_NOT_FOUND)
         user = get_object_or_404(User, username=kwargs['username'])
         if user.has_perm(self.perm, app):
             remove_perm(self.perm, user, app)
@@ -138,7 +138,22 @@ class AdminPermsViewSet(viewsets.ModelViewSet):
     serializer_class = serializers.AdminUserSerializer
     permission_classes = (IsAdmin,)
 
+    def check_obj_permissions(self, obj):
+        """
+        Small wrapper around check_object_permissions().
+
+        If the user is denied permission to the object, then
+        it should return a 404 so the user is not aware of the
+        application resource.
+        """
+        try:
+            self.check_object_permissions(self.request, obj)
+        except PermissionDenied:
+            raise Http404("No {} matches the given query.".format(
+                self.model._meta.object_name))
+
     def get_queryset(self, **kwargs):
+        self.check_obj_permissions(self.request.user)
         return self.model.objects.filter(is_active=True, is_superuser=True)
 
     def create(self, request, **kwargs):
@@ -225,21 +240,31 @@ class BaseAppViewSet(viewsets.ModelViewSet):
 
     permission_classes = (permissions.IsAuthenticated, IsAppUser)
 
-    def pre_save(self, obj):
-        obj.owner = self.request.user
+    def check_obj_permissions(self, obj):
+        """
+        Small wrapper around check_object_permissions().
 
-    def get_queryset(self, **kwargs):
-        app = get_object_or_404(models.App, id=self.kwargs['id'])
+        If the user is denied permission to the object, then
+        it should return a 404 so the user is not aware of the
+        application resource.
+        """
         try:
-            self.check_object_permissions(self.request, app)
+            self.check_object_permissions(self.request, obj)
         except PermissionDenied:
             raise Http404("No {} matches the given query.".format(
                 self.model._meta.object_name))
+
+    def pre_save(self, obj):
+        obj.owner = self.request.user
+
+    def get_queryset(self, **kwargs):
+        app = get_object_or_404(models.App, id=self.kwargs['id'])
+        self.check_obj_permissions(app)
         return self.model.objects.filter(app=app)
 
     def get_object(self, *args, **kwargs):
         obj = self.get_queryset().latest('created')
-        self.check_object_permissions(self.request, obj)
+        self.check_obj_permissions(obj)
         return obj
 
 
@@ -260,7 +285,7 @@ def get_success_headers(self, data):
 
     def create(self, request, *args, **kwargs):
         app = get_object_or_404(models.App, id=self.kwargs['id'])
-        self.check_object_permissions(self.request, app)
+        self.check_obj_permissions(app)
         request._data = request.DATA.copy()
         request.DATA['app'] = app
         try:
@@ -278,12 +303,8 @@ class AppConfigViewSet(BaseAppViewSet):
     def get_object(self, *args, **kwargs):
         """Return the Config associated with the App's latest Release."""
         app = get_object_or_404(models.App, id=self.kwargs['id'])
-        try:
-            self.check_object_permissions(self.request, app)
-            return app.release_set.latest().config
-        except (PermissionDenied, models.Release.DoesNotExist):
-            raise Http404("No {} matches the given query.".format(
-                self.model._meta.object_name))
+        self.check_obj_permissions(app)
+        return app.release_set.latest().config
 
     def pre_save(self, config):
         """merge the old config with the new"""
@@ -396,20 +417,35 @@ class DomainViewSet(OwnerViewSet):
     model = models.Domain
     serializer_class = serializers.DomainSerializer
 
+    def check_obj_permissions(self, obj):
+        """
+        Small wrapper around check_object_permissions().
+
+        If the user is denied permission to the object, then
+        it should return a 404 so the user is not aware of the
+        application resource.
+        """
+        try:
+            self.check_object_permissions(self.request, obj)
+        except PermissionDenied:
+            raise Http404("No {} matches the given query.".format(
+                self.model._meta.object_name))
+
     def create(self, request, *args, **kwargs):
         app = get_object_or_404(models.App, id=self.kwargs['id'])
+        self.check_obj_permissions(app)
         request._data = request.DATA.copy()
         request.DATA['app'] = app
         return super(DomainViewSet, self).create(request, *args, **kwargs)
 
     def get_queryset(self, **kwargs):
         app = get_object_or_404(models.App, id=self.kwargs['id'])
-        qs = self.model.objects.filter(app=app)
-        return qs
+        self.check_obj_permissions(app)
+        return self.model.objects.filter(app=app)
 
     def get_object(self, *args, **kwargs):
-        qs = self.get_queryset(**kwargs)
-        obj = qs.get(domain=self.kwargs['domain'])
+        obj = self.get_queryset().get(domain=self.kwargs['domain'])
+        self.check_obj_permissions(obj)
         return obj
 
 

docs/reference/api-v1.1.rst

@@ -6,7 +6,7 @@
 Controller API v1.1
 ===================
 
-This is the v1.0 REST API for the :ref:`Controller`.
+This is the v1.1 REST API for the :ref:`Controller`.
 
 
 What's New
@@ -16,6 +16,8 @@ What's New
 
 **New!** All controller responses now return the ``X_DEIS_PLATFORM_VERSION`` header.
 
+**New!** Users see a 404 response when modifying applications they are unauthorized to see.
+
 
 Authentication
 --------------