introspect_application

apps/gooyal_oauth2/models.py

@@ -28,6 +28,8 @@ class Application(AbstractApplication):
     Application model for use with Django OAuth Toolkit that allows the scopes
     available to an application to be restricted on a per-application basis.
     """
+
+    # TODO: change it to owner
     user = models.ForeignKey(
         settings.AUTH_USER_MODEL,
         related_name="%(app_label)s_%(class)s",

apps/gooyal_oauth2/urls.py

@@ -1,7 +1,7 @@
 from django.urls import re_path
 
 from oauth2_provider import views
-from .views.introspect import IntrospectTokenView
+from .views.introspect import IntrospectTokenView, IntrospectApplicationView
 
 app_name = "oauth2_provider"
 
@@ -11,6 +11,7 @@ base_urlpatterns = [
     re_path(r"^token/$", views.TokenView.as_view(), name="token"),
     re_path(r"^revoke_token/$", views.RevokeTokenView.as_view(), name="revoke-token"),
     re_path(r"^introspect/$", IntrospectTokenView.as_view(), name="introspect"),
+    re_path(r"^introspect_application/$", IntrospectApplicationView.as_view(), name="introspect-application"),
 ]
 
 

apps/gooyal_oauth2/validators.py

@@ -20,17 +20,6 @@ UserModel = get_user_model()
 
 
 class OAuth2Validator(BaseOAuth2Validator):  # pylint: disable=w0223
-    introspection_client = None
-
-    def get_introspection_client(self, introspection_client_id, introspection_client_secret):
-        if not OAuth2Validator.introspection_client:
-            OAuth2Validator.introspection_client = service_clients.Client(client_id=introspection_client_id,
-                                                                          client_secret=introspection_client_secret,
-                                                                          grant_type=service_clients.AccountsClient.GRANT_CLIENT_CREDENTIALS,
-                                                                          scopes=['introspection'])
-
-        return OAuth2Validator.introspection_client
-
     def validate_user(self, username, password, client, request, *args, **kwargs):
         auth_fields = getattr(request, 'auth_fields', 'username:password').split(':')
 
@@ -61,228 +50,3 @@ class OAuth2Validator(BaseOAuth2Validator):  # pylint: disable=w0223
             return True
 
         return False
-
-    def _get_token_from_gooyal_authentication_server(
-            self, token, introspection_url, introspection_token, introspection_credentials, introspection_client_id,
-            introspection_client_secret
-    ):
-        """Use external introspection endpoint to "crack open" the token.
-        :param introspection_url: introspection endpoint URL
-        :param introspection_token: Bearer token
-        :param introspection_credentials: Basic Auth credentials (id,secret)
-        :return: :class:`models.AccessToken`
-
-        Some RFC 7662 implementations (including this one) use a Bearer token while others use Basic
-        Auth. Depending on the external AS's implementation, provide either the introspection_token
-        or the introspection_credentials.
-
-        If the resulting access_token identifies a username (e.g. Authorization Code grant), add
-        that user to the UserModel. Also cache the access_token up until its expiry time or a
-        configured maximum time.
-
-        """
-
-        headers = None
-        response = None
-        if introspection_token:
-            headers = {"Authorization": "Bearer {}".format(introspection_token)}
-            try:
-                response = requests.post(
-                    introspection_url,
-                    data={"token": token}, headers=headers
-                )
-            except requests.exceptions.RequestException:
-                log.exception("Introspection: Failed POST to %r in token lookup", introspection_url)
-                return None
-
-        elif introspection_credentials:
-            client_id = introspection_credentials[0].encode("utf-8")
-            client_secret = introspection_credentials[1].encode("utf-8")
-            basic_auth = base64.b64encode(client_id + b":" + client_secret)
-            headers = {"Authorization": "Basic {}".format(basic_auth.decode("utf-8"))}
-            try:
-                response = requests.post(
-                    introspection_url,
-                    data={"token": token}, headers=headers
-                )
-            except requests.exceptions.RequestException:
-                log.exception("Introspection: Failed POST to %r in token lookup", introspection_url)
-                return None
-
-        elif introspection_client_id and introspection_client_secret:
-            data = {"token": token}
-            introspection_client = self.get_introspection_client(introspection_client_id, introspection_client_secret)
-            response = introspection_client.request(url=introspection_url, method='post', data=data,
-                                                    required_scopes=['introspection'], login_required=True)
-
-        try:
-            content: dict = response.json()
-        except ValueError:
-            log.exception("Introspection: Failed to parse response as json")
-            return None
-
-        user = None
-        if "active" in content and content["active"] is True:
-            if "username" in content:
-                user, content = oauth2_settings.INTROSPECTION_USER_CREATE_METHOD(token, content)
-
-            max_caching_time = datetime.now() + timedelta(
-                seconds=oauth2_settings.RESOURCE_SERVER_TOKEN_CACHING_SECONDS
-            )
-
-            if "exp" in content:
-                expires = datetime.utcfromtimestamp(content["exp"])
-                if expires > max_caching_time:
-                    expires = max_caching_time
-            else:
-                expires = max_caching_time
-
-            scope = content.get("scope", "")
-            expires = make_aware(expires)
-
-            access_token, _created = AccessTokenModel.objects.update_or_create(
-                token=token,
-                defaults={
-                    "user": user,
-                    "application": None,
-                    "scope": scope,
-                    "expires": expires,
-                    "detail": content,
-                })
-
-            # try:
-            #     access_token = AccessTokenModel.objects.select_related("application", "user").get(token=token)
-            # except AccessTokenModel.DoesNotExist:
-            #     access_token = AccessTokenModel.objects.create(
-            #         user=user,
-            #         token=token,
-            #         application=None,
-            #         scope=scope,
-            #         expires=expires,
-            #         detail=content
-            #     )
-            # else:
-            #     access_token.expires = expires
-            #     access_token.scope = scope
-            #     access_token.detail = content
-            #     access_token.save()
-
-            return access_token
-
-    # def validate_bearer_token(self, token, scopes, request):
-    #     """
-    #     When users try to access resources, check that provided token is valid
-    #     """
-    #     if not token:
-    #         return False
-    #
-    #     introspection_url = oauth2_settings.RESOURCE_SERVER_INTROSPECTION_URL
-    #     introspection_token = oauth2_settings.RESOURCE_SERVER_AUTH_TOKEN
-    #     introspection_credentials = oauth2_settings.RESOURCE_SERVER_INTROSPECTION_CREDENTIALS
-    #     introspection_client_id = oauth2_settings.RESOURCE_SERVER_CLIENT_ID
-    #     introspection_client_secret = oauth2_settings.RESOURCE_SERVER_CLIENT_SECRET
-    #
-    #     try:
-    #         access_token = AccessTokenModel.objects.select_related("application", "user").get(token=token)
-    #     except AccessTokenModel.DoesNotExist:
-    #         access_token = None
-    #
-    #     # if there is no token or it's invalid then introspect the token if there's an external OAuth server
-    #     if not access_token or not access_token.is_valid(scopes):
-    #         if introspection_url and (introspection_token or introspection_credentials or (introspection_client_id and
-    #                                                                                        introspection_client_secret)):
-    #             access_token = self._get_token_from_gooyal_authentication_server(
-    #                 token,
-    #                 introspection_url,
-    #                 introspection_token,
-    #                 introspection_credentials,
-    #                 introspection_client_id,
-    #                 introspection_client_secret
-    #             )
-    #
-    #     if access_token and access_token.is_valid(scopes):
-    #         request.client = access_token.application
-    #         request.user = access_token.user
-    #         request.scopes = scopes
-    #
-    #         # this is needed by django rest framework
-    #         request.access_token = access_token
-    #         return True
-    #     else:
-    #         self._set_oauth2_error_on_request(request, access_token, scopes)
-    #         return False
-
-    def _authenticate_basic_auth(self, request):
-        """
-        Authenticates with HTTP Basic Auth.
-
-        Note: as stated in rfc:`2.3.1`, client_id and client_secret must be encoded with
-        "application/x-www-form-urlencoded" encoding algorithm.
-        """
-        auth_string = self._extract_basic_auth(request)
-        if not auth_string:
-            return False
-
-        try:
-            encoding = request.encoding or settings.DEFAULT_CHARSET or "utf-8"
-        except AttributeError:
-            encoding = "utf-8"
-
-        try:
-            b64_decoded = base64.b64decode(auth_string)
-        except (TypeError, binascii.Error):
-            log.debug("Failed basic auth: %r can't be decoded as base64", auth_string)
-            return False
-
-        try:
-            auth_string_decoded = b64_decoded.decode(encoding)
-        except UnicodeDecodeError:
-            log.debug("Failed basic auth: %r can't be decoded as unicode by %r", auth_string, encoding)
-            return False
-
-        try:
-            client_id, client_secret = map(unquote_plus, auth_string_decoded.split(":", 1))
-        except ValueError:
-            log.debug("Failed basic auth, Invalid base64 encoding.")
-            return False
-
-        if self._load_application(client_id, request) is None:
-            log.debug("Failed basic auth: Application %s does not exist" % client_id)
-            return False
-        elif request.client.client_id != client_id:
-            log.debug("Failed basic auth: wrong client id %s" % client_id)
-            return False
-
-        # TODO: check why not work
-        elif not client_secret == request.client.client_secret:
-            log.debug("Failed basic auth: wrong client secret %s" % client_secret)
-            return False
-        else:
-            return True
-
-    def _authenticate_request_body(self, request):
-        """
-                Try to authenticate the client using client_id and client_secret
-                parameters included in body.
-
-                Remember that this method is NOT RECOMMENDED and SHOULD be limited to
-                clients unable to directly utilize the HTTP Basic authentication scheme.
-                See rfc:`2.3.1` for more details.
-                """
-        # TODO: check if oauthlib has already unquoted client_id and client_secret
-        try:
-            client_id = request.client_id
-            client_secret = request.client_secret
-        except AttributeError:
-            return False
-
-        if self._load_application(client_id, request) is None:
-            log.debug("Failed body auth: Application %s does not exists" % client_id)
-            return False
-            # TODO: check why not work
-        elif not client_secret == request.client.client_secret:
-            log.debug("Failed body auth: wrong client secret %s" % client_secret)
-            return False
-        else:
-            return True
-

apps/gooyal_oauth2/views/introspect.py

@@ -5,7 +5,7 @@ from django.http import JsonResponse
 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.models import get_access_token_model, get_application_model
 from oauth2_provider.views.generic import ClientProtectedScopedResourceView
 
 
@@ -69,3 +69,49 @@ class IntrospectTokenView(ClientProtectedScopedResourceView):
         :return:
         """
         return self.get_token_response(request.POST.get("token", None))
+
+
+@method_decorator(csrf_exempt, name="dispatch")
+class IntrospectApplicationView(ClientProtectedScopedResourceView):
+    required_scopes = ["introspection"]
+
+    # TODO: check application state
+    @staticmethod
+    def get_application_response(client_id=None):
+        try:
+            application = (
+                get_application_model().objects.get(client_id=client_id)
+            )
+        except ObjectDoesNotExist:
+            return JsonResponse({"active": False}, status=200)
+        else:
+            data = {
+                "active": True,
+            }
+            if application.user_id:
+                data["client_owner"] = str(application.user_id)
+            return JsonResponse(data)
+
+    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_application_response(request.GET.get("client_id", 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_application_response(request.POST.get("client_id", None))