Token-Based Authentication via DLA (#42)

* Finished TokenView

* Finished test cases

* Added 403 error test

accounts/urls.py

@@ -1,6 +1,6 @@
 from django.urls import path
 
-from accounts.views import CallbackView, LoginView, LogoutView
+from accounts.views import CallbackView, LoginView, LogoutView, TokenView
 
 
 app_name = "accounts"
@@ -10,4 +10,5 @@
     path("callback/", CallbackView.as_view(), name="callback"),
     path("login/", LoginView.as_view(), name="login"),
     path("logout/", LogoutView.as_view(), name="logout"),
+    path("token/", TokenView.as_view(), name="token"),
 ]

accounts/views.py

@@ -1,13 +1,22 @@
+import datetime
+
+import requests
 from django.contrib import auth
-from django.http import HttpResponseServerError
-from django.shortcuts import redirect
+from django.contrib.auth import get_user_model
+from django.http import HttpResponseServerError, JsonResponse
+from django.shortcuts import get_object_or_404, redirect
 from django.urls import reverse
+from django.utils import timezone
 from django.views import View
 from requests_oauthlib import OAuth2Session
 
+from accounts.models import AccessToken, RefreshToken
 from accounts.settings import accounts_settings
 
 
+User = get_user_model()
+
+
 def invalid_next(return_to):
     try:
         from sentry_sdk import capture_message
@@ -105,3 +114,46 @@ def get(self, request):
             invalid_next(return_to)
             return_to = "/"
         return redirect(return_to)
+
+
+class TokenView(View):
+    """
+    View for token-based authentication, specifically for mobile products that
+    do not rely on session authentication.
+    Assumes OAuth2 Authorization code has been retrieved prior to accessing this route.
+    """
+
+    def post(self, request):
+        # Hit Platform OAuth2 token provider
+        token_url = accounts_settings.PLATFORM_URL + "/accounts/token/"
+        response = requests.post(token_url, data=request.POST.dict())
+        if response.status_code == 200:
+            token = response.json()
+            # Use the access token to retrieve user information from platform
+            platform = OAuth2Session(accounts_settings.CLIENT_ID, token=token)
+            introspect_url = accounts_settings.PLATFORM_URL + "/accounts/introspect/"
+            platform_request = platform.post(
+                introspect_url, data={"token": token["access_token"]}
+            )
+            if (
+                platform_request.status_code == 200
+            ):  # Connected to platform successfully
+                user_props = platform_request.json()["user"]
+                user = get_object_or_404(
+                    User, id=user_props["pennid"], username=user_props["username"]
+                )
+                # Update user Access and Refresh tokens
+                AccessToken.objects.update_or_create(
+                    user=user,
+                    defaults={
+                        "expires_at": timezone.now()
+                        + datetime.timedelta(seconds=token["expires_in"]),
+                        "token": token["access_token"],
+                    },
+                )
+                RefreshToken.objects.update_or_create(
+                    user=user, defaults={"token": token["refresh_token"]}
+                )
+                return JsonResponse(response.json())
+            return JsonResponse({"detail": "Invalid tokens"}, status=403)
+        return JsonResponse({"detail": "Invalid parameters"}, status=400)

tests/accounts/test_views.py

@@ -5,6 +5,7 @@
 from django.test import Client, TestCase
 from django.urls import reverse
 
+from accounts.models import AccessToken, RefreshToken
 from accounts.settings import accounts_settings
 
 
@@ -173,3 +174,106 @@ def test_invalid_next(self):
         )
         self.assertEqual(response.status_code, 302)
         self.assertEqual(response.url, "/")
+
+
+class TokenViewTestCase(TestCase):
+    def setUp(self):
+        self.User = get_user_model()
+        self.client = Client()
+        self.mock_requests_json = {
+            "access_token": "abc",
+            "refresh_token": "123",
+            "expires_in": 100,
+            "token_type": "Bearer",
+            "scope": "read introspection",
+        }
+        # Response from introspect
+        self.mock_oauth_json = {
+            "user": {
+                "pennid": 1,
+                "first_name": "First",
+                "last_name": "Last",
+                "username": "user",
+                "email": "[email protected]",
+                "affiliation": [],
+                "user_permissions": [],
+                "groups": ["student", "member"],
+            }
+        }
+
+    @patch("accounts.views.OAuth2Session.post")
+    @patch("accounts.views.requests.post")
+    def test_token_valid(self, mock_requests_post, mock_oauth_post):
+        mock_requests_post.return_value.json.return_value = self.mock_requests_json
+        mock_requests_post.return_value.status_code = 200
+        mock_oauth_post.return_value.json.return_value = self.mock_oauth_json
+        mock_oauth_post.return_value.status_code = 200
+        user = self.User.objects.create(id=1, username="user", password="secret")
+        payload = {
+            "grant_type": "correct_grant_type",
+            "client_id": "correct_client_id",
+            "refresh_token": "correct_refresh_token",
+        }
+        response = self.client.post(reverse("accounts:token"), payload)
+        self.assertEqual(200, response.status_code)
+        res_json = response.json()
+        # Assert response is same as Platform response
+        self.assertEqual(
+            self.mock_requests_json["access_token"], res_json["access_token"]
+        )
+        self.assertEqual(
+            self.mock_requests_json["refresh_token"], res_json["refresh_token"]
+        )
+        self.assertEqual(self.mock_requests_json["expires_in"], res_json["expires_in"])
+        # Assert Access and Refresh tokens are correctly created in the backend
+        self.assertEqual(len(AccessToken.objects.all()), 1)
+        self.assertEqual(len(RefreshToken.objects.all()), 1)
+        self.assertEqual(
+            self.mock_requests_json["access_token"], user.accesstoken.token
+        )
+        self.assertEqual(
+            self.mock_requests_json["refresh_token"], user.refreshtoken.token
+        )
+
+    @patch("accounts.views.OAuth2Session.post")
+    @patch("accounts.views.requests.post")
+    def test_token_unknown_user(self, mock_requests_post, mock_oauth_post):
+        mock_requests_post.return_value.json.return_value = self.mock_requests_json
+        mock_requests_post.return_value.status_code = 200
+        mock_oauth_post.return_value.json.return_value = self.mock_oauth_json
+        mock_oauth_post.return_value.status_code = 200
+        payload = {
+            "grant_type": "correct_grant_type",
+            "client_id": "correct_client_id",
+            "code": "correct_code",
+            "redirect_uri": "https://example.com",
+            "verifier": "correct_verifier",
+        }
+        response = self.client.post(reverse("accounts:token"), payload)
+        # Should fail because User object is never created in this test
+        self.assertEqual(404, response.status_code)
+
+    @patch("accounts.views.requests.post")
+    def test_token_invalid_introspect(self, mock_requests_post):
+        mock_requests_post.return_value.json.return_value = self.mock_requests_json
+        mock_requests_post.return_value.status_code = 200
+        payload = {
+            "grant_type": "correct_grant_type",
+            "client_id": "correct_client_id",
+            "code": "correct_code",
+            "redirect_uri": "https://example.com",
+            "verifier": "correct_verifier",
+        }
+        response = self.client.post(reverse("accounts:token"), payload)
+        # Should fail because introspect invalidated the provided access token
+        self.assertEqual(403, response.status_code)
+
+    def test_token_invalid_parameters(self):
+        payload = {
+            "grant_type": "invalid_grant_type",
+            "client_id": "invalid_client_id",
+            "refresh_token": "invalid_refresh",
+        }
+        response = self.client.post(reverse("accounts:token"), payload)
+        # Should fail because Platform request should invalidate the payload
+        self.assertEqual(400, response.status_code)