import calendar from django.contrib.auth.decorators import login_not_required from django.core.exceptions import ObjectDoesNotExist from django.http import JsonResponse from django.utils.decorators import method_decorator from django.views.decorators.csrf import csrf_exempt from django.utils.translation import gettext_lazy as _ from oauth2_provider.models import get_access_token_model, get_application_model from oauth2_provider.signals import app_authorized from oauth2_provider.views.generic import ClientProtectedScopedResourceView from oauth2_provider.views.mixins import OAuthLibMixin @method_decorator(csrf_exempt, name="dispatch") class IntrospectTokenView(ClientProtectedScopedResourceView): """ Implements an endpoint for token introspection based on RFC 7662 https://rfc-editor.org/rfc/rfc7662.html 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.select_related("user", "application").get(token=token_value) ) except ObjectDoesNotExist: return JsonResponse({"active": False}, status=200) 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: # NOTICE: i pass uuid instead of username data["username"] = token.user.pk return JsonResponse(data) else: return JsonResponse({"active": False}, status=200) 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)) @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, # "client_id": client_id, "uuid": application.uuid } if application.user_id: data["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)) import hashlib import json from django.http import HttpResponse from django.utils.decorators import method_decorator from django.views.decorators.csrf import csrf_exempt from django.views.decorators.debug import sensitive_post_parameters from django.views.generic import FormView, View @method_decorator(csrf_exempt, name="dispatch") @method_decorator(login_not_required, name="dispatch") class TokenView(OAuthLibMixin, View): """ Implements an endpoint to provide access tokens The endpoint is used in the following flows: * Authorization code * Password * Client credentials """ @method_decorator(sensitive_post_parameters("password", "client_secret")) def post(self, request, *args, **kwargs): url, headers, body, status = self.create_token_response(request) content = json.loads(body) for k,v in content.items(): _('Invalid credentials given.') _('invalid_grant') content[k] = _(v) if status == 200: access_token = json.loads(body).get("access_token") if access_token is not None: token_checksum = hashlib.sha256(access_token.encode("utf-8")).hexdigest() token = get_access_token_model().objects.get(token_checksum=token_checksum) app_authorized.send(sender=self, request=request, token=token) response = HttpResponse(content=body, status=status) for k, v in headers.items(): response[k] = v return response @method_decorator(csrf_exempt, name="dispatch") @method_decorator(login_not_required, name="dispatch") class GooyalRevokeTokenView(OAuthLibMixin, View): """ Implements an endpoint to revoke access or refresh tokens """ def post(self, request, *args, **kwargs): url, headers, body, status = self.create_revocation_response(request) response = HttpResponse(content=body or "", status=status) for k, v in headers.items(): response[k] = v return response