Improved REST API permission system (FooCard authentication)

requirements/base.txt

@@ -14,3 +14,4 @@ requests==2.8.1
 pytz==2015.7
 raven==5.8.1
 pynarlivs==0.9.0
+PyJWT==1.4.2

src/authtoken/authentication.py

@@ -1,7 +1,33 @@
+import jwt
+from jwt.exceptions import InvalidKeyError, InvalidTokenError
+from django.conf import settings
+from django.utils.translation import ugettext_lazy as _
 from rest_framework import authentication, exceptions
 from rest_framework.authentication import get_authorization_header
-from django.utils.translation import ugettext_lazy as _
+
 from .models import Token
+from foobar.api import get_account
+
+
+def validate_header(auth):
+    if len(auth) == 1:
+        msg = _('Invalid token header. No credentials provided.')
+        raise exceptions.AuthenticationFailed(msg)
+    elif len(auth) > 2:
+        msg = _('Invalid token header. Token string should not contain '
+                'spaces.')
+        raise exceptions.AuthenticationFailed(msg)
+
+
+class FallbackAuthentication(authentication.BaseAuthentication):
+    """
+    This is a fallback intended to run as a last resort for authentication.
+    It does however only raise an exception, telling the requestor that all
+    attempts on authenticating has failed.
+    """
+
+    def authenticate(self):
+        raise exceptions.AuthenticationFailed(_('Invalid token'))
 
 
 class TokenAuthentication(authentication.BaseAuthentication):
@@ -24,14 +50,7 @@ def authenticate(self, request):
         if not auth or auth[0].lower() != b'token':
             return None
 
-        if len(auth) == 1:
-            msg = _('Invalid token header. No credentials provided.')
-            raise exceptions.AuthenticationFailed(msg)
-        elif len(auth) > 2:
-            msg = _('Invalid token header. Token string should not contain '
-                    'spaces.')
-            raise exceptions.AuthenticationFailed(msg)
-
+        validate_header(auth)
         try:
             token = auth[1].decode()
         except UnicodeError:
@@ -51,3 +70,44 @@ def authenticate_credentials(self, key):
 
     def authenticate_header(self, request):
         return 'Token'
+
+
+class FooCardAuthentication(authentication.BaseAuthentication):
+    """
+    JSON Web Token authentication.
+    Acts identically to token based authentication, as seen from the outside.
+    Clients should authenticate by passing the token key in the "Authorization"
+    HTTP header, prepended with the string "CardToken ". For example:
+        Authorization: CardToken: 401f7ac837da42b97f613d789819ff93537bee6a
+    """
+    algorithm = 'HS256'
+    message = 'Invalid token'
+
+    def authenticate(self, request):
+        auth = get_authorization_header(request).split()
+        if not auth or auth[0].lower() != b'cardtoken':
+            return None
+
+        validate_header(auth)
+        try:
+            payload = jwt.decode(
+                auth[1].decode(),
+                settings.SECRET_WEBTOKEN,
+                algorithm=self.algorithm
+            )
+        except (InvalidKeyError, InvalidTokenError):
+            msg = _('{0} header'.format(self.message))
+            raise exceptions.AuthenticationFailed(msg)
+
+        return self.authenticate_credentials(payload)
+
+    def authenticate_credentials(self, payload):
+        card_id = payload.get('card_id')
+        account = get_account(card_id)
+        if account is None:
+            raise exceptions.AuthenticationFailed(_(self.message))
+
+        return (account, None)
+
+    def authenticate_header(self, request):
+        return 'CardToken'

src/authtoken/serializers.py

@@ -0,0 +1,16 @@
+from django.utils.translation import ugettext_lazy as _
+from rest_framework import serializers
+
+from foobar.api import get_account
+
+
+class CardTokenSerializer(serializers.Serializer):
+    number = serializers.CharField(required=True, allow_blank=False)
+
+    def validate_number(self, value):
+        account = get_account(value)
+        if account is None:
+            msg = {'number': _('Card is not registered')}
+            raise serializers.ValidationError(msg)
+
+        return value

src/authtoken/test.py

@@ -0,0 +1,163 @@
+import datetime
+import jwt
+from jwt.exceptions import InvalidTokenError
+from unittest.mock import patch
+
+from django.conf.urls import url
+from django.http import HttpResponse
+from django.test import override_settings, TestCase
+from django.utils import timezone
+from rest_framework import status
+from rest_framework.test import APIClient, APIRequestFactory
+from rest_framework.views import APIView
+
+from .authentication import FooCardAuthentication
+from .models import Token
+from .views import obtain_foocard_token
+from foobar.tests.factories import AccountFactory, CardFactory
+
+factory = APIRequestFactory()
+
+
+class MockView(APIView):
+    def get(self, request):
+        return HttpResponse({'some': 1, 'thing': 2, 'dandy': 3})
+
+    def post(self, request):
+        return HttpResponse({'some': 1, 'thing': 2, 'dandy': 3})
+
+
+urlpatterns = [
+    url(r'^cardtoken/$',
+        MockView.as_view(authentication_classes=[FooCardAuthentication])),
+    url(r'^auth-token/$', obtain_foocard_token),
+]
+
+
+@override_settings(ROOT_URLCONF='authtoken.test', SECRET_WEBTOKEN='secret')
+class AuthenticateFooCardTokenTests(TestCase):
+    enc = 'HS256'
+    path = '/cardtoken/'
+
+    def setUp(self):
+        self.csrf_client = APIClient(enforce_csrf_checks=True)
+
+    @patch('authtoken.authentication.jwt.decode')
+    def test_correct_auth_of_token(self, mock_jwt):
+        card = CardFactory()
+        token = jwt.encode(
+            {'card_id': card.number},
+            'secret',
+            algorithm=self.enc
+        )
+        token = token.decode()
+        stamp = timezone.now() + timezone.timedelta(minutes=5)
+        mock_jwt.return_value = {'exp': stamp, 'card_id': card.number}
+
+        auth = 'CardToken {0}'.format(token)
+        response = self.csrf_client.post(
+            self.path,
+            {'example': 'example'},
+            format='json',
+            HTTP_AUTHORIZATION=auth
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        mock_jwt.assert_called_once_with(token, 'secret', algorithm=self.enc)
+
+    @patch('authtoken.authentication.jwt.decode')
+    def test_auth_with_bad_token(self, mock_jwt):
+        token = '401f7ac837da42b97f613d789819ff93537bee6a'
+        auth = 'CardToken {0}'.format(token)
+
+        mock_jwt.side_effect = InvalidTokenError('Invalid token')
+
+        response = self.csrf_client.get(self.path, HTTP_AUTHORIZATION=auth)
+
+        self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
+        mock_jwt.assert_called_once_with(token, 'secret', algorithm=self.enc)
+
+    @patch('authtoken.authentication.jwt.decode')
+    def test_auth_with_bad_account(self, mock_jwt):
+        AccountFactory()  # Keep this here so we just don't have an empty QS
+        stamp = timezone.now() + datetime.timedelta(minutes=10)
+        token = jwt.encode(
+            {'exp': stamp, 'card_id': None},
+            'secret',
+            algorithm=self.enc
+        )
+        token = token.decode()
+        mock_jwt.return_value = {'exp': stamp, 'card_id': None}
+
+        auth = 'CardToken {0}'.format(token)
+        response = self.csrf_client.get(self.path, HTTP_AUTHORIZATION=auth)
+
+        self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
+        mock_jwt.assert_called_once_with(token, 'secret', algorithm=self.enc)
+
+    def test_auth_with_bad_header(self):
+        account = AccountFactory()
+        stamp = timezone.now() + datetime.timedelta(minutes=10)
+        token = jwt.encode(
+            {'exp': stamp, 'card_id': str(account.pk)},
+            'secret',
+            algorithm=self.enc
+        )
+        auth = 'ThisIsIncorrect {0}'.format(token.decode())
+        response = self.csrf_client.get(self.path, HTTP_AUTHORIZATION=auth)
+        # Returns 200_OK as it is not allowed to even try to authenticate
+        # And there is no other authentication that is tested
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+
+        auth = 'CardToken'
+        response = self.csrf_client.get(self.path, HTTP_AUTHORIZATION=auth)
+        self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
+
+        auth = 'CardToken {0} {1}'.format(token.decode()[:10], token.decode())
+        response = self.csrf_client.get(self.path, HTTP_AUTHORIZATION=auth)
+        self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
+
+
+@override_settings(ROOT_URLCONF='authtoken.test', SECRET_WEBTOKEN='secret')
+class CreateFooCardTokenTests(TestCase):
+    enc = 'HS256'
+    path = '/auth-token/'
+
+    def setUp(self):
+        self.csrf_client = APIClient(enforce_csrf_checks=True)
+
+    def test_create_new_account_token(self):
+        card = CardFactory()
+        api_token = Token.objects.create(key='abc123')
+
+        auth = 'Token {0}'.format(api_token)
+        response = self.csrf_client.post(
+            self.path,
+            {'number': card.number},
+            format='json',
+            HTTP_AUTHORIZATION=auth
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertIn('card_token', response.data.keys())
+
+        token = response.data.get('card_token')
+        self.assertIsNotNone(token)
+        self.assertTrue(len(token) > 0)
+
+        auth = 'CardToken {0}'.format(token)
+        response = self.csrf_client.get('/cardtoken/', HTTP_AUTHORIZATION=auth)
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+
+    def test_create_token_for_invalid_account(self):
+        api_token = Token.objects.create(key='abc123')
+        auth = 'Token {0}'.format(api_token)
+        response = self.csrf_client.post(
+            self.path,
+            {'number': '123456'},
+            format='json',
+            HTTP_AUTHORIZATION=auth
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertIn('number', response.data.keys())

src/authtoken/views.py

@@ -0,0 +1,29 @@
+import jwt
+
+from django.conf import settings
+from django.utils import timezone
+from rest_framework.response import Response
+from rest_framework.views import APIView
+
+from .serializers import CardTokenSerializer
+
+
+class ObtainFooCardToken(APIView):
+    serializer_class = CardTokenSerializer
+
+    def post(self, request, *args, **kwargs):
+        serializer = self.serializer_class(data=request.data)
+        serializer.is_valid(raise_exception=True)
+        number = serializer.validated_data['number']
+
+        expiration_date = timezone.now() + timezone.timedelta(minutes=5)
+        token = jwt.encode(
+            {'exp': expiration_date, 'card_id': number},
+            settings.SECRET_WEBTOKEN,
+            algorithm='HS256'
+        )
+
+        return Response({'card_token': token.decode()})
+
+
+obtain_foocard_token = ObtainFooCardToken.as_view()

src/foobar/settings/base.py

@@ -19,6 +19,7 @@
 # SECURITY WARNING: keep the secret key used in production secret!
 SECRET_KEY = os.getenv('SECRET_KEY',
                        '%^5x9&idy09abn3my1)p+9_g!=aglt4&qog*5ztxwc@xjjp0m%')
+SECRET_WEBTOKEN = os.getenv('SECRET_WEBTOKEN', SECRET_KEY)
 
 # SECURITY WARNING: don't run with debug turned on in production!
 DEBUG = os.getenv('DEBUG', True)

src/foobar/tests/factories.py

@@ -1,3 +1,5 @@
+import random
+
 import factory.fuzzy
 from .. import models
 
@@ -12,4 +14,8 @@ class Meta:
         model = models.Card
 
     account = factory.SubFactory(AccountFactory)
-    number = factory.fuzzy.FuzzyInteger(0, (1 << 32) - 1)
+
+    @factory.lazy_attribute
+    def number(self):
+        value = random.randint(0, (1 << 32) - 1)
+        return str(value)