diff --git a/apps/gooyal_oauth2/tests.py b/apps/gooyal_oauth2/tests.py index df017f3..4ff6f17 100644 --- a/apps/gooyal_oauth2/tests.py +++ b/apps/gooyal_oauth2/tests.py @@ -180,17 +180,32 @@ class GooyalOAuth2Tests(APITestCase): result = self.client.post(reverse("gooyal_oauth2:token"), data=data) access_token = result.json()['access_token'] - print(AccessToken.objects.count()) - - data = { + revoke_data = { "token": access_token, } - result = self.client.post(reverse("gooyal_oauth2:revoke-token"), data=data) - print(result.content) - print(result.status_code) + result = self.client.post(reverse("gooyal_oauth2:revoke-token"), data=revoke_data) + self.assertEqual(result.status_code, 200) + self.assertEqual(AccessToken.objects.count(), 0) + + self.user.refresh_from_db() + self.user.set_otp() + self.client.credentials( + HTTP_AUTHORIZATION='Basic ' + self.basic_auth_string(self.client_id, self.client_secret) + ) + result = self.client.post(reverse("gooyal_oauth2:token"), data=data) + access_token = result.json()['access_token'] + access_token_object = AccessToken.objects.first() + revoke_data = { + "token": access_token_object.pk, + } + result = self.client.post(reverse("gooyal_oauth2:revoke-token"), data=revoke_data) + self.assertEqual(result.status_code, 200) + self.assertEqual(AccessToken.objects.count(), 0) + + + - print(AccessToken.objects.count()) diff --git a/apps/gooyal_oauth2/urls.py b/apps/gooyal_oauth2/urls.py index f0668e1..78f2c19 100644 --- a/apps/gooyal_oauth2/urls.py +++ b/apps/gooyal_oauth2/urls.py @@ -3,7 +3,7 @@ from django.urls import re_path, path from oauth2_provider import views from rest_framework import routers -from .views.oauth_views import IntrospectTokenView, IntrospectApplicationView, TokenView, RevokeTokenView +from .views.oauth_views import IntrospectTokenView, IntrospectApplicationView, TokenView, GooyalRevokeTokenView from .views import apis as api_views from .views import pages app_name = "gooyal_oauth2" @@ -14,7 +14,7 @@ app_name = "gooyal_oauth2" base_urlpatterns = [ re_path(r"^authorize/$", views.AuthorizationView.as_view(), name="authorize"), re_path(r"^token/$", TokenView.as_view(), name="token"), - re_path(r"^revoke_token/$", RevokeTokenView.as_view(), name="revoke-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 dc75f1c..6d47be1 100755 --- a/apps/gooyal_oauth2/validators.py +++ b/apps/gooyal_oauth2/validators.py @@ -266,4 +266,24 @@ class OAuth2Validator(BaseOAuth2Validator): # pylint: disable=w0223 return result def revoke_token(self, token, token_type_hint, request, *args, **kwargs): - return super().revoke_token(token, token_type_hint, request, *args, **kwargs) \ No newline at end of file + """ + Revoke an access or refresh token. + + :param token: The token string. + :param token_type_hint: access_token or refresh_token. + :param request: The HTTP Request (oauthlib.common.Request) + """ + if token_type_hint not in ["access_token", "refresh_token"]: + token_type_hint = None + + token_types = { + "access_token": AccessToken, + "refresh_token": RefreshToken, + } + + token_type = token_types.get(token_type_hint, AccessToken) + + try: + token_type.objects.get(pk=token).revoke() + except: + token_type.objects.get(token=token).revoke() diff --git a/apps/gooyal_oauth2/views/oauth_views.py b/apps/gooyal_oauth2/views/oauth_views.py index 8c48607..561d1bf 100644 --- a/apps/gooyal_oauth2/views/oauth_views.py +++ b/apps/gooyal_oauth2/views/oauth_views.py @@ -169,7 +169,7 @@ class TokenView(OAuthLibMixin, View): @method_decorator(csrf_exempt, name="dispatch") @method_decorator(login_not_required, name="dispatch") -class RevokeTokenView(OAuthLibMixin, View): +class GooyalRevokeTokenView(OAuthLibMixin, View): """ Implements an endpoint to revoke access or refresh tokens """ diff --git a/apps/users/models.py b/apps/users/models.py index dac8845..0ab60af 100644 --- a/apps/users/models.py +++ b/apps/users/models.py @@ -195,6 +195,8 @@ class User(AbstractUser): if result: self.otp = None + self.otp_expire = None + if not self.date_joined: self.date_joined = timezone.now() else: @@ -202,7 +204,9 @@ class User(AbstractUser): self.save() else: result = False - + self.otp = None + self.otp_expire = None + self.save() return result diff --git a/apps/users/serializers.py b/apps/users/serializers.py index de3e137..e4bf1cc 100644 --- a/apps/users/serializers.py +++ b/apps/users/serializers.py @@ -138,7 +138,21 @@ class ChangePasswordSerializer(serializers.Serializer): AccessToken = get_access_token_model() class SessionSerializer(serializers.ModelSerializer): + is_current = serializers.SerializerMethodField() + + def get_is_current(self, obj): + request = self.context.get('request') + token = request and request.auth and request.auth.token + if token == obj.token: + result = True + + else: + result = False + + return result + + class Meta: model = AccessToken - fields = ('uuid', 'token', 'created', "detail") - read_only_fields = ['uuid', 'token', 'created', "detail"] + fields = ('uuid', 'created', "detail", "is_current") + read_only_fields = ['uuid', 'created', "detail", "is_current" ] diff --git a/apps/users/tests.py b/apps/users/tests.py index 7ce503c..8fa80f6 100644 --- a/apps/users/tests.py +++ b/apps/users/tests.py @@ -1,3 +1,154 @@ -from django.test import TestCase +import base64 +import json +import uuid +from datetime import timedelta +from unicodedata import category +from unittest.mock import patch + +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 rest_framework.test import APIClient, APITestCase + + +from apps.core.models import Config +from apps.gooyal_oauth2.models import Scope +from apps.users.models import User + +AccessToken = get_access_token_model() +Application = get_application_model() + + +class UserTests(APITestCase): + user_uuid = uuid.UUID('b14e8b86-8f4a-44d9-b29d-badceb47005f') + access_token_1 = 'au4naVsdKCbKNOhnElPyXcrwSnqqFbm' + access_token_2 = 'vu4naVsdKCbKNOhnElPyXcrwSnqqFbm' + application_uuid = uuid.UUID('a14e8b86-8f4a-44d9-b29d-badceb47005f') + client_id = '4INGOCMoulE0fNY1SQlTbPtsWqqxGj2DdqjADq6u' + client_secret = '4INGOCMoulE0fNY1SQlTbPtsWqqxGj2DdqjADq6u' + + visitor_uuid = uuid.UUID('b14e8b86-8f4a-44d9-b29d-badceb47005a') + + expire_datetime = timezone.now() + timedelta(seconds=3600) + expire_datetime.isoformat() + client = APIClient() + + def setUp(self): + from django.conf import settings + settings.SMS_SEND = False + self.user_phone_number = '+989100000000' + self.user = User.objects.create(pk=self.user_uuid, phone_number=self.user_phone_number) + + Scope.objects.create(name='accounts.status:get', description='accounts.status:get') + Scope.objects.create(name='accounts.account:retrieve', description='accounts.account:retrieve') + + self.application = Application.objects.create( + client_id=self.client_id, + client_secret=self.client_secret, + authorization_grant_type='password', + hash_client_secret=False, + uuid=self.application_uuid, + user_id=self.user_uuid, + max_allowed_session=1, + allowed_scope='accounts.status:get accounts.account:retrieve', + ) + + # self.sys_date_patcher = patch('simata_safte.models.get_sys_date', mock_get_sys_date) + # self.set_id_patcher = patch('simata_safte.models.set_id', mock_set_id) + # self.sign_pdf_patcher = patch('simata_safte.models.sign_pdf', mock_sign_pdf) + # self.check_pdf_signature_patcher = patch('simata_safte.models.check_pdf_signature', mock_check_pdf_signature) + + def tearDown(self): + super().tearDown() + + + def _create_authorization_header(self, token): + return "Bearer {0}".format(token) + + def test_authentication_allow(self): + access_token_1 = AccessToken.objects.create( + **{ + "token": self.access_token_1, + "user": self.user, + # "client_id": self.client_id, + # "client_owner": owner, + "application_id": self.application_uuid, + "scope": 'accounts.status:get accounts.account:retrieve', + "expires": self.expire_datetime.isoformat(), + }, + ) + + auth_1 = self._create_authorization_header(access_token_1.token) + + response = self.client.get(reverse("core:status"), HTTP_AUTHORIZATION=auth_1) + + self.assertContains(response, 'status') + + access_token_2 = AccessToken.objects.create( + **{ + "token": self.access_token_2, + "user": self.user, + + # "client_id": self.client_id, + # "client_owner": owner, + "application_id": self.application_uuid, + "scope": 'accounts.status:get', + "expires": self.expire_datetime.isoformat(), + }, + ) + + auth_2 = self._create_authorization_header(access_token_2.token) + + response = self.client.get(reverse("core:status"), HTTP_AUTHORIZATION=auth_1) + self.assertEqual(response.status_code, 200) + + response = self.client.get(reverse("core:status"), HTTP_AUTHORIZATION=auth_2) + self.assertEqual(response.status_code, 503) + + self.application.max_allowed_session = 0 + self.application.save() + self.application.refresh_from_db() + + response = self.client.get(reverse("core:status"), HTTP_AUTHORIZATION=auth_2) + self.assertEqual(response.status_code, 200) + + def basic_auth_string(self, username, password): + """ساخت Basic Auth string""" + import base64 + user_pass = f"{username}:{password}" + basic_credentials = base64.b64encode(user_pass.encode('utf-8')).decode('utf-8') + + return basic_credentials + + def login(self): + self.user.set_otp() + + data = { + "grant_type": "password", + "username": self.user_phone_number, + "password": '77501', + "scope": 'accounts.status:get accounts.account:retrieve', + "auth_fields": 'phone_number:otp' + } + self.client.credentials( + HTTP_AUTHORIZATION='Basic ' + self.basic_auth_string(self.client_id, self.client_secret) + ) + + result = self.client.post(reverse("gooyal_oauth2:token"), data=data) + access_token = result.json()['access_token'] + self.client.credentials(HTTP_AUTHORIZATION='Bearer ' + access_token) + + self.assertEqual(result.status_code, 200) + return result + + + def test_getSessions_allOK_success(self): + self.login() + response = self.client.get(reverse("users:user_sessions_api")) + self.assertEqual(response.json()['results'][0]['is_current'] , True) + + + + -# Create your tests here.