Implement IntrospectTokenView (RFC 7662)

Author: symptog

oauth2_provider/migrations/0005_auto_20170514_1141.py

@@ -0,0 +1,22 @@
+# -*- coding: utf-8 -*-
+# Generated by Django 1.11.1 on 2017-05-14 11:41
+from __future__ import unicode_literals
+
+from oauth2_provider.settings import oauth2_settings
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('oauth2_provider', '0004_auto_20160525_1623'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='accesstoken',
+            name='application',
+            field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, to=oauth2_settings.APPLICATION_MODEL),
+        ),
+    ]

oauth2_provider/models.py

@@ -211,7 +211,7 @@ class AbstractAccessToken(models.Model):
                              on_delete=models.CASCADE,
                              related_name="%(app_label)s_%(class)s")
     token = models.CharField(max_length=255, unique=True, )
-    application = models.ForeignKey(oauth2_settings.APPLICATION_MODEL,
+    application = models.ForeignKey(oauth2_settings.APPLICATION_MODEL, blank=True, null=True,
                                     on_delete=models.CASCADE)
     expires = models.DateTimeField()
     scope = models.TextField(blank=True)

oauth2_provider/urls.py

@@ -12,6 +12,7 @@
     url(r'^authorize/$', views.AuthorizationView.as_view(), name="authorize"),
     url(r'^token/$', views.TokenView.as_view(), name="token"),
     url(r'^revoke_token/$', views.RevokeTokenView.as_view(), name="revoke-token"),
+    url(r'^introspect/$', views.IntrospectTokenView.as_view(), name="introspect"),
 ]
 
 

oauth2_provider/views/__init__.py

@@ -4,3 +4,4 @@
     ApplicationDelete, ApplicationUpdate
 from .generic import ProtectedResourceView, ScopedProtectedResourceView, ReadWriteScopedResourceView
 from .token import AuthorizedTokensListView, AuthorizedTokenDeleteView
+from .introspect import IntrospectTokenView

oauth2_provider/views/introspect.py

@@ -0,0 +1,75 @@
+from __future__ import unicode_literals
+
+import calendar
+import json
+
+from django.core.exceptions import ObjectDoesNotExist
+from django.http import HttpResponse
+from django.utils.decorators import method_decorator
+from django.views.decorators.csrf import csrf_exempt
+
+from oauth2_provider.models import get_access_token_model
+from oauth2_provider.views import ReadWriteScopedResourceView
+
+
+@method_decorator(csrf_exempt, name="dispatch")
+class IntrospectTokenView(ReadWriteScopedResourceView):
+    """
+    Implements an endpoint for token introspection based
+    on RFC 7662 https://tools.ietf.org/html/rfc7662
+
+    To access this view the request must pass a OAuth2 Bearer Token
+    which is allowed to access the scope `introspection`.
+    """
+    required_scopes = ["introspection"]
+
+    @staticmethod
+    def get_token_response(token_value=None):
+        try:
+            token = get_access_token_model().objects.get(token=token_value)
+        except ObjectDoesNotExist:
+            return HttpResponse(
+                content=json.dumps({"active": False}),
+                status=401,
+                content_type="application/json"
+            )
+        else:
+            if token.is_valid():
+                data = {
+                    "active": True,
+                    "scope": token.scope,
+                    "exp": int(calendar.timegm(token.expires.timetuple())),
+                }
+                if token.application:
+                    data["client_id"] = token.application.client_id
+                if token.user:
+                    data["username"] = token.user.get_username()
+                return HttpResponse(content=json.dumps(data), status=200, content_type="application/json")
+            else:
+                return HttpResponse(content=json.dumps({
+                    "active": False,
+                }), status=200, content_type="application/json")
+
+    def get(self, request, *args, **kwargs):
+        """
+        Get the token from the URL parameters.
+        URL: https://example.com/introspect?token=mF_9.B5f-4.1JqM
+
+        :param request:
+        :param args:
+        :param kwargs:
+        :return:
+        """
+        return self.get_token_response(request.GET.get("token", None))
+
+    def post(self, request, *args, **kwargs):
+        """
+        Get the token from the body form parameters.
+        Body: token=mF_9.B5f-4.1JqM
+
+        :param request:
+        :param args:
+        :param kwargs:
+        :return:
+        """
+        return self.get_token_response(request.POST.get("token", None))

setup.cfg

@@ -30,6 +30,7 @@ zip_safe = False
 install_requires =
 	django >= 1.8
 	oauthlib >= 2.0.1
+	requests >= 2.13.0
 
 [options.packages.find]
 exclude = tests

tests/test_introspection_view.py

@@ -0,0 +1,260 @@
+from __future__ import unicode_literals
+
+import calendar
+import datetime
+
+from django.contrib.auth import get_user_model
+from django.test import TestCase
+from django.urls import reverse
+from django.utils import timezone
+
+from oauth2_provider.models import get_access_token_model, get_application_model
+from oauth2_provider.settings import oauth2_settings
+
+Application = get_application_model()
+AccessToken = get_access_token_model()
+UserModel = get_user_model()
+
+
+class TestTokenIntrospectionViews(TestCase):
+    """
+    Tests for Authorized Token Introspection Views
+    """
+    def setUp(self):
+        self.resource_server_user = UserModel.objects.create_user("resource_server", "test@example.com")
+        self.test_user = UserModel.objects.create_user("bar_user", "dev@example.com")
+
+        self.application = Application(
+            name="Test Application",
+            redirect_uris="http://localhost http://example.com http://example.it",
+            user=self.test_user,
+            client_type=Application.CLIENT_CONFIDENTIAL,
+            authorization_grant_type=Application.GRANT_AUTHORIZATION_CODE,
+        )
+        self.application.save()
+
+        self.resource_server_token = AccessToken.objects.create(
+            user=self.resource_server_user, token="12345678900",
+            application=self.application,
+            expires=timezone.now() + datetime.timedelta(days=1),
+            scope="read write introspection"
+        )
+
+        self.valid_token = AccessToken.objects.create(
+            user=self.test_user, token="12345678901",
+            application=self.application,
+            expires=timezone.now() + datetime.timedelta(days=1),
+            scope="read write dolphin"
+        )
+
+        self.invalid_token = AccessToken.objects.create(
+            user=self.test_user, token="12345678902",
+            application=self.application,
+            expires=timezone.now() + datetime.timedelta(days=-1),
+            scope="read write dolphin"
+        )
+
+        self.token_without_user = AccessToken.objects.create(
+            user=None, token="12345678903",
+            application=self.application,
+            expires=timezone.now() + datetime.timedelta(days=1),
+            scope="read write dolphin"
+        )
+
+        self.token_without_app = AccessToken.objects.create(
+            user=self.test_user, token="12345678904",
+            application=None,
+            expires=timezone.now() + datetime.timedelta(days=1),
+            scope="read write dolphin"
+        )
+
+        oauth2_settings._SCOPES = ["read", "write", "introspection", "dolphin"]
+        oauth2_settings.READ_SCOPE = "read"
+        oauth2_settings.WRITE_SCOPE = "write"
+
+    def tearDown(self):
+        oauth2_settings._SCOPES = ["read", "write"]
+        AccessToken.objects.all().delete()
+        Application.objects.all().delete()
+        UserModel.objects.all().delete()
+
+    def test_view_forbidden(self):
+        """
+        Test that the view is restricted for logged-in users.
+        """
+        response = self.client.get(reverse("oauth2_provider:introspect"))
+        self.assertEqual(response.status_code, 403)
+
+    def test_view_get_valid_token(self):
+        """
+        Test that when you pass a valid token as URL parameter,
+        a json with an active token state is provided
+        """
+        auth_headers = {
+            "HTTP_AUTHORIZATION": "Bearer " + self.resource_server_token.token,
+        }
+        response = self.client.get(
+            reverse("oauth2_provider:introspect"),
+            {"token": self.valid_token.token},
+            **auth_headers)
+
+        self.assertEqual(response.status_code, 200)
+        content = response.json()
+        self.assertIsInstance(content, dict)
+        self.assertDictEqual(content, {
+            "active": True,
+            "scope": self.valid_token.scope,
+            "client_id": self.valid_token.application.client_id,
+            "username": self.valid_token.user.get_username(),
+            "exp": int(calendar.timegm(self.valid_token.expires.timetuple())),
+        })
+
+    def test_view_get_valid_token_without_user(self):
+        """
+        Test that when you pass a valid token as URL parameter,
+        a json with an active token state is provided
+        """
+        auth_headers = {
+            "HTTP_AUTHORIZATION": "Bearer " + self.resource_server_token.token,
+        }
+        response = self.client.get(
+            reverse("oauth2_provider:introspect"),
+            {"token": self.token_without_user.token},
+            **auth_headers)
+
+        self.assertEqual(response.status_code, 200)
+        content = response.json()
+        self.assertIsInstance(content, dict)
+        self.assertDictEqual(content, {
+            "active": True,
+            "scope": self.token_without_user.scope,
+            "client_id": self.token_without_user.application.client_id,
+            "exp": int(calendar.timegm(self.token_without_user.expires.timetuple())),
+        })
+
+    def test_view_get_valid_token_without_app(self):
+        """
+        Test that when you pass a valid token as URL parameter,
+        a json with an active token state is provided
+        """
+        auth_headers = {
+            "HTTP_AUTHORIZATION": "Bearer " + self.resource_server_token.token,
+        }
+        response = self.client.get(
+            reverse("oauth2_provider:introspect"),
+            {"token": self.token_without_app.token},
+            **auth_headers)
+
+        self.assertEqual(response.status_code, 200)
+        content = response.json()
+        self.assertIsInstance(content, dict)
+        self.assertDictEqual(content, {
+            "active": True,
+            "scope": self.token_without_app.scope,
+            "username": self.token_without_app.user.get_username(),
+            "exp": int(calendar.timegm(self.token_without_app.expires.timetuple())),
+        })
+
+    def test_view_get_invalid_token(self):
+        """
+        Test that when you pass an invalid token as URL parameter,
+        a json with an inactive token state is provided
+        """
+        auth_headers = {
+            "HTTP_AUTHORIZATION": "Bearer " + self.resource_server_token.token,
+        }
+        response = self.client.get(
+            reverse("oauth2_provider:introspect"),
+            {"token": self.invalid_token.token},
+            **auth_headers)
+
+        self.assertEqual(response.status_code, 200)
+        content = response.json()
+        self.assertIsInstance(content, dict)
+        self.assertDictEqual(content, {
+            "active": False,
+        })
+
+    def test_view_get_notexisting_token(self):
+        """
+        Test that when you pass an non existing token as URL parameter,
+        a json with an inactive token state is provided
+        """
+        auth_headers = {
+            "HTTP_AUTHORIZATION": "Bearer " + self.resource_server_token.token,
+        }
+        response = self.client.get(
+            reverse("oauth2_provider:introspect"),
+            {"token": "kaudawelsch"},
+            **auth_headers)
+
+        self.assertEqual(response.status_code, 401)
+        content = response.json()
+        self.assertIsInstance(content, dict)
+        self.assertDictEqual(content, {
+            "active": False,
+        })
+
+    def test_view_post_valid_token(self):
+        """
+        Test that when you pass a valid token as form parameter,
+        a json with an active token state is provided
+        """
+        auth_headers = {
+            "HTTP_AUTHORIZATION": "Bearer " + self.resource_server_token.token,
+        }
+        response = self.client.post(
+            reverse("oauth2_provider:introspect"),
+            {"token": self.valid_token.token},
+            **auth_headers)
+
+        self.assertEqual(response.status_code, 200)
+        content = response.json()
+        self.assertIsInstance(content, dict)
+        self.assertDictEqual(content, {
+            "active": True,
+            "scope": self.valid_token.scope,
+            "client_id": self.valid_token.application.client_id,
+            "username": self.valid_token.user.get_username(),
+            "exp": int(calendar.timegm(self.valid_token.expires.timetuple())),
+        })
+
+    def test_view_post_invalid_token(self):
+        """
+        Test that when you pass an invalid token as form parameter,
+        a json with an inactive token state is provided
+        """
+        auth_headers = {
+            "HTTP_AUTHORIZATION": "Bearer " + self.resource_server_token.token,
+        }
+        response = self.client.post(
+            reverse("oauth2_provider:introspect"),
+            {"token": self.invalid_token.token},
+            **auth_headers)
+
+        self.assertEqual(response.status_code, 200)
+        content = response.json()
+        self.assertIsInstance(content, dict)
+        self.assertDictEqual(content, {
+            "active": False,
+        })
+
+    def test_view_post_notexisting_token(self):
+        """
+        Test that when you pass an non existing token as form parameter,
+        a json with an inactive token state is provided
+        """
+        auth_headers = {
+            "HTTP_AUTHORIZATION": "Bearer " + self.resource_server_token.token,
+        }
+        response = self.client.post(
+            reverse("oauth2_provider:introspect"),
+            {"token": "kaudawelsch"},
+            **auth_headers)
+
+        self.assertEqual(response.status_code, 401)
+        content = response.json()
+        self.assertIsInstance(content, dict)
+        self.assertDictEqual(content, {
+            "active": False,
+        })