feat(controller): Add token regeneration endpoint

Author: Joshua-Anderson

api/__init__.py

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

api/permissions.py

@@ -129,3 +129,18 @@ def has_permission(self, request, view):
         if not auth_header:
             return False
         return auth_header == settings.BUILDER_KEY
+
+
+class CanRegenerateToken(permissions.BasePermission):
+    """
+    Checks if a user can regenerate a token
+    """
+
+    def has_permission(self, request, view):
+        """
+        Return `True` if permission is granted, `False` otherwise.
+        """
+        if 'username' in request.data or 'all' in request.data:
+            return request.user.is_superuser
+        else:
+            return True

api/tests/test_auth.py

@@ -278,3 +278,33 @@ def test_change_user_passwd(self):
         response = self.client.post(url, json.dumps(submit), content_type='application/json',
                                     HTTP_AUTHORIZATION='token {}'.format(self.user1_token))
         self.assertEqual(response.status_code, 200)
+
+    def test_regenerate(self):
+        """ Test that token regeneration works"""
+
+        url = '/v1/auth/tokens/'
+
+        response = self.client.post(url, '{}', content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(self.admin_token))
+
+        self.assertEqual(response.status_code, 200)
+        self.assertNotEqual(response.data['token'], self.admin_token)
+
+        self.admin_token = Token.objects.get(user=self.admin)
+
+        response = self.client.post(url, '{"username" : "autotest2"}',
+                                    content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(self.admin_token))
+
+        self.assertEqual(response.status_code, 200)
+        self.assertNotEqual(response.data['token'], self.user1_token)
+
+        response = self.client.post(url, '{"all" : "true"}',
+                                    content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(self.admin_token))
+        self.assertEqual(response.status_code, 200)
+
+        response = self.client.post(url, '{}', content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(self.admin_token))
+
+        self.assertEqual(response.status_code, 401)

api/urls.py

@@ -87,6 +87,8 @@
         views.UserManagementViewSet.as_view({'post': 'passwd'})),
     url(r'^auth/login/',
         'rest_framework.authtoken.views.obtain_auth_token'),
+    url(r'^auth/tokens/',
+        views.TokenManagementViewSet.as_view({'post': 'regenerate'})),
     # admin sharing
     url(r'^admin/perms/(?P<username>[-_\w]+)/?',
         views.AdminPermsViewSet.as_view({'delete': 'destroy'})),

api/views.py

@@ -13,6 +13,7 @@
 from rest_framework.permissions import IsAuthenticated
 from rest_framework.response import Response
 from rest_framework.viewsets import GenericViewSet
+from rest_framework.authtoken.models import Token
 
 from api import authentication, models, permissions, serializers, viewsets
 
@@ -51,6 +52,39 @@ def passwd(self, request, **kwargs):
         return Response({'status': 'password set'})
 
 
+class TokenManagementViewSet(GenericViewSet,
+                             mixins.DestroyModelMixin):
+    serializer_class = serializers.UserSerializer
+    permission_classes = [permissions.CanRegenerateToken]
+
+    def get_queryset(self):
+        return User.objects.filter(pk=self.request.user.pk)
+
+    def get_object(self):
+        return self.get_queryset()[0]
+
+    def regenerate(self, request, **kwargs):
+        obj = self.get_object()
+
+        if 'all' in request.data:
+            for user in User.objects.all():
+                if not user.is_anonymous():
+                    token = Token.objects.get(user=user)
+                    token.delete()
+                    Token.objects.create(user=user)
+            return Response("")
+
+        if 'username' in request.data:
+            obj = get_object_or_404(User,
+                                    username=request.data['username'])
+            self.check_object_permissions(self.request, obj)
+
+        token = Token.objects.get(user=obj)
+        token.delete()
+        token = Token.objects.create(user=obj)
+        return Response({'token': token.key})
+
+
 class BaseDeisViewSet(viewsets.OwnerViewSet):
     """
     A generic ViewSet for objects related to Deis.