From 022011d2856f9055d49bb9bee150c22df90d0c82 Mon Sep 17 00:00:00 2001 From: mahdavi Date: Thu, 4 Jul 2024 15:38:20 +0330 Subject: [PATCH] introspect_application --- apps/gooyal_oauth2/models.py | 2 + apps/gooyal_oauth2/urls.py | 3 +- apps/gooyal_oauth2/validators.py | 236 ------------------------- apps/gooyal_oauth2/views/introspect.py | 48 ++++- 4 files changed, 51 insertions(+), 238 deletions(-) diff --git a/apps/gooyal_oauth2/models.py b/apps/gooyal_oauth2/models.py index 471587e..507aead 100644 --- a/apps/gooyal_oauth2/models.py +++ b/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", diff --git a/apps/gooyal_oauth2/urls.py b/apps/gooyal_oauth2/urls.py index a041ab7..8fc04cd 100644 --- a/apps/gooyal_oauth2/urls.py +++ b/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"), ] diff --git a/apps/gooyal_oauth2/validators.py b/apps/gooyal_oauth2/validators.py index 7153aaf..f0186dc 100755 --- a/apps/gooyal_oauth2/validators.py +++ b/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 - diff --git a/apps/gooyal_oauth2/views/introspect.py b/apps/gooyal_oauth2/views/introspect.py index d911fd6..516c1c6 100644 --- a/apps/gooyal_oauth2/views/introspect.py +++ b/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))