diff --git a/apps/core/admin.py b/apps/core/admin.py index 2b7593b..69b3eff 100644 --- a/apps/core/admin.py +++ b/apps/core/admin.py @@ -1,5 +1,6 @@ from django.contrib import admin -from .models import SMSPolicy +from .models import SMSPolicy, Config + class SMSPolicyAdmin(admin.ModelAdmin): def has_add_permission(self, request): @@ -9,4 +10,8 @@ class SMSPolicyAdmin(admin.ModelAdmin): return False +class ConfigAdmin(admin.ModelAdmin): + list_display = ['key', 'value', 'value_type', 'get_value', 'description', 'comment'] + +admin.site.register(Config, ConfigAdmin) admin.site.register(SMSPolicy, SMSPolicyAdmin) diff --git a/apps/gooyal_oauth2/tests.py b/apps/gooyal_oauth2/tests.py index 8336d53..df017f3 100644 --- a/apps/gooyal_oauth2/tests.py +++ b/apps/gooyal_oauth2/tests.py @@ -1,3 +1,4 @@ +import base64 import json import uuid from datetime import timedelta @@ -8,15 +9,18 @@ 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 +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() + def mock_notifications_push_user_success(user_uuid, title, message, priority=5, extras=None): import uuid as sys_uuid class Tmp(): @@ -26,32 +30,43 @@ def mock_notifications_push_user_success(user_uuid, title, message, priority=5, return data -class GooyalOAuth2Tests(TestCase): +class GooyalOAuth2Tests(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.notifications_push_user_success_patcher = patch('apps.users.models.User.notify', mock_notifications_push_user_success) self.notifications_push_user_success_patcher.start() + self.user_phone_number = '+989100000000' + self.user = User.objects.create(pk=self.user_uuid, phone_number=self.user_phone_number) - user, _ = User.objects.get_or_create(pk=self.user_uuid) - self.user = user + scope = Scope.objects.create(name='accounts.status:get', description='accounts.status:get') + + self.application = Application.objects.create( - self.application, _created = Application.objects.get_or_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 + max_allowed_session=1, + allowed_scope='accounts.status:get', ) # self.sys_date_patcher = patch('simata_safte.models.get_sys_date', mock_get_sys_date) @@ -90,6 +105,7 @@ class GooyalOAuth2Tests(TestCase): **{ "token": self.access_token_2, "user": self.user, + # "client_id": self.client_id, # "client_owner": owner, "application_id": self.application_uuid, @@ -100,12 +116,81 @@ class GooyalOAuth2Tests(TestCase): auth_2 = self._create_authorization_header(access_token_2.token) - response = self.client.get(reverse("core:status"), HTTP_AUTHORIZATION=auth_2) - self.assertEqual(response.status_code, 503) - response = self.client.get(reverse("core:status"), HTTP_AUTHORIZATION=auth_1) self.assertEqual(response.status_code, 200) - def test_1(self): - self.assertTrue(True) + 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', + "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_loginByOTP_allOK_success(self): + print(self.login()) + response = self.client.get(reverse("core:status")) + print(response.status_code) + + def test_revoke_token(self): + self.user.set_otp() + + data = { + "grant_type": "password", + "username": self.user_phone_number, + "password": '77501', + "scope": 'accounts.status:get', + "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'] + + print(AccessToken.objects.count()) + + data = { + "token": access_token, + } + + result = self.client.post(reverse("gooyal_oauth2:revoke-token"), data=data) + print(result.content) + print(result.status_code) + + print(AccessToken.objects.count()) + + diff --git a/apps/gooyal_oauth2/urls.py b/apps/gooyal_oauth2/urls.py index 62e9055..f0668e1 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.introspect import IntrospectTokenView, IntrospectApplicationView, TokenView +from .views.oauth_views import IntrospectTokenView, IntrospectApplicationView, TokenView, RevokeTokenView 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/$", views.RevokeTokenView.as_view(), name="revoke-token"), + re_path(r"^revoke_token/$", 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 5bdb8a5..dc75f1c 100755 --- a/apps/gooyal_oauth2/validators.py +++ b/apps/gooyal_oauth2/validators.py @@ -261,3 +261,9 @@ class OAuth2Validator(BaseOAuth2Validator): # pylint: disable=w0223 return result else: raise ServiceUnavailable(code='max_allowed_session_reached') + + else: + 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 diff --git a/apps/gooyal_oauth2/views/introspect.py b/apps/gooyal_oauth2/views/oauth_views.py similarity index 91% rename from apps/gooyal_oauth2/views/introspect.py rename to apps/gooyal_oauth2/views/oauth_views.py index fca06b1..8c48607 100644 --- a/apps/gooyal_oauth2/views/introspect.py +++ b/apps/gooyal_oauth2/views/oauth_views.py @@ -167,3 +167,17 @@ class TokenView(OAuthLibMixin, View): response[k] = v return response +@method_decorator(csrf_exempt, name="dispatch") +@method_decorator(login_not_required, name="dispatch") +class RevokeTokenView(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