limit tokens

apps/core/decorators.py

@@ -0,0 +1,31 @@
+from functools import wraps
+
+from django.utils import timezone
+
+from apps.core.models import Config, ConfigValueChoices
+from utils.exceptions import ServiceUnavailable
+
+
+def service_availability(key:str):
+    def with_params(view_func):
+        @wraps(view_func)
+        def wrapper(self, request, *args, **kwargs):
+            print(key)
+            print(f"Executing {view_func.__name__} action")
+
+            service_is_available = Config.get_value_of(f"SERVICE_IS_AVAILABLE_{key.upper()}", "True", ConfigValueChoices.BOOLEAN)
+            config = Config.objects.get(key=f"SERVICE_IS_AVAILABLE_{key.upper()}")
+            if not service_is_available:
+                raise ServiceUnavailable(
+                    detail={
+                        # "code": 'service_unavailable_at_now',
+                        # "timestamp": timezone.now().isoformat(),
+                        "message": config.description or "این سرویس در حال حاضر در دسترس نیست"
+                    },
+                )
+
+            return view_func(self, request, *args, **kwargs)
+
+        return wrapper
+
+    return with_params

apps/core/migrations/0002_config.py

@@ -0,0 +1,30 @@
+# Generated by Django 6.0.2 on 2026-04-06 05:59
+
+import uuid
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('core', '0001_initial'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='Config',
+            fields=[
+                ('uuid', models.UUIDField(db_index=True, default=uuid.uuid4, primary_key=True, serialize=False, unique=True)),
+                ('created_at', models.DateTimeField(auto_now_add=True, db_index=True)),
+                ('updated_at', models.DateTimeField(auto_now=True, db_index=True)),
+                ('key', models.CharField(db_index=True, max_length=255, unique=True, verbose_name='key')),
+                ('value', models.TextField(blank=True, null=True, verbose_name='value')),
+                ('value_type', models.CharField(choices=[('INT', 'Integer'), ('FLOAT', 'Float'), ('BOOLEAN', 'Boolean'), ('STRING', 'String'), ('DATE', 'Date'), ('DATETIME', 'DateTime'), ('JSON', 'JSON')], max_length=64)),
+                ('description', models.TextField(blank=True, null=True, verbose_name='description')),
+                ('comment', models.TextField(blank=True, null=True, verbose_name='comment')),
+            ],
+            options={
+                'abstract': False,
+            },
+        ),
+    ]

apps/core/models.py

@@ -1,17 +1,21 @@
-from django.db import models
+import logging
+
 from django.db.models import TextChoices
-from django.utils.translation import gettext_lazy as _
 
 from apps.gooyal_oauth2.settings import oauth2_settings
+from utils.clients.sms.fake.client import FakeClient
+from utils.clients.sms.mobin_sms.client import MobinSMSClient
+from utils.clients.sms.payam_sms.client import PayamSMSClient
 from utils.clients.sms.sms import BaseSMSClient
 from utils.models import BaseModel
-from utils.clients.sms.payam_sms.client import PayamSMSClient
-from utils.clients.sms.mobin_sms.client import MobinSMSClient
-from utils.clients.sms.fake.client import FakeClient
 
-import logging
 logger = logging.getLogger(__name__)
 
+import json
+from datetime import datetime
+
+from django.db import models
+from django.utils.translation import gettext_lazy as _
 
 
 class SMSClienChoises(TextChoices):
@@ -19,6 +23,7 @@ class SMSClienChoises(TextChoices):
     PAYAM_SMS = 'payam_sms', _('payam_sms')
     MOBIN_SMS = 'mobin_sms', _('mobin_sms')
 
+
 class SMSPolicy(BaseModel):
     application = models.ForeignKey(oauth2_settings.APPLICATION_MODEL, on_delete=models.PROTECT,
                                     related_name='+', null=True, blank=True, unique=True)
@@ -36,3 +41,66 @@ class SMSPolicy(BaseModel):
             return PayamSMSClient()
         else:
             return FakeClient()
+
+
+class ConfigValueChoices(models.TextChoices):
+    INT = 'INT', _('Integer')
+    FLOAT = 'FLOAT', _('Float')
+    BOOLEAN = 'BOOLEAN', _('Boolean')
+    STRING = 'STRING', _('String')
+    DATE = 'DATE', _('Date')
+    DATETIME = 'DATETIME', _('DateTime')
+    JSON = 'JSON', _('JSON')
+
+
+class Config(BaseModel):
+    key = models.CharField(_('key'), max_length=255, unique=True, db_index=True)
+    value = models.TextField(_('value'), null=True, blank=True)
+    value_type = models.CharField(choices=ConfigValueChoices.choices, max_length=64)
+    description = models.TextField(_('description'), null=True, blank=True)
+    comment = models.TextField(_('comment'), null=True, blank=True)
+
+    def get_value(self):
+        return Config.get_value_of(self.key)
+
+    @staticmethod
+    def type_cast(value, value_type):
+
+        try:
+            if value_type == ConfigValueChoices.STRING.value:
+                value = str(value)
+
+            elif value_type == ConfigValueChoices.BOOLEAN.value:
+                value = value.lower() == 'true'
+
+            elif value_type == ConfigValueChoices.DATE.value:
+                value = datetime.fromisoformat(value)
+
+            elif value_type == ConfigValueChoices.DATETIME.value:
+                value = datetime.fromisoformat(value)
+
+            elif value_type == ConfigValueChoices.JSON.value:
+                value = json.loads(value)
+
+            elif value_type == ConfigValueChoices.INT.value:
+                value = int(value)
+
+            elif value_type == ConfigValueChoices.FLOAT.value:
+                value = float(value)
+
+        except Exception as e:
+            value = None
+
+        return value
+
+    @staticmethod
+    def get_value_of(key, default=None, casting_type=None):
+        # TODO: use cache
+        config, created = Config.objects.get_or_create(key=key, defaults={'value': default})
+
+        if config.value is None:
+            return default
+
+        value = Config.type_cast(config.value, casting_type or config.value_type)
+
+        return value

apps/core/serializers.py

@@ -0,0 +1,9 @@
+from rest_framework import serializers
+
+from .models import Config
+
+
+class ConfigSerializer(serializers.ModelSerializer):
+    class Meta:
+        model = Config
+        fields = ['key', 'value', 'value_type']

apps/core/urls.py

@@ -1,9 +1,17 @@
 from django.urls import path, include
-from .views import HomeView, TestIpView
+from rest_framework.routers import DefaultRouter
+
+from .views import HomeView, TestIpView, HealthcheckView, StatusView
 
 app_name = "core"
 
+router = DefaultRouter()
+
+# router.register('api/config', ConfigViewSet, basename='configs')
+
 urlpatterns = [
     path('', HomeView.as_view(), name='home'),
     path('test-ip', TestIpView.as_view(), name='test-ip'),
+    path('status', StatusView.as_view(), name='status'),
+    path('healthcheck', HealthcheckView.as_view(), name='healthcheck'),
 ]

apps/core/views.py

@@ -2,13 +2,30 @@ from django.contrib.auth.decorators import login_required
 from django.shortcuts import render
 from django.utils.decorators import method_decorator
 from django.views.generic import TemplateView
+from django.contrib.sessions.backends.db import SessionStore
+from django.contrib.sessions.models import Session
+from rest_framework import mixins
+from rest_framework.permissions import AllowAny
+from rest_framework.response import Response
+from rest_framework.views import APIView
+from rest_framework.viewsets import GenericViewSet
+
+from apps.core.decorators import service_availability
+from apps.core.models import Config
+from apps.core.serializers import ConfigSerializer
+from apps.gooyal_oauth2.rest_framework import IsAuthenticatedOrTokenMatchesOASRequirements
 
 
-# Create your views here.
 # @method_decorator(login_required, name='dispatch')
 class HomeView(TemplateView):
     template_name = 'core/home.html'
 
+    # def get(self, request, *args, **kwargs):
+    #     session = request.session
+    #     print(type(session))
+    #     print(session.session_key)
+    #     return super(HomeView, self).get(request, *args, **kwargs)
+
 
 class TestIpView(TemplateView):
     template_name = 'core/home.html'
@@ -33,3 +50,33 @@ class TestIpView(TemplateView):
                         'Priority': 'u=0, i'}
 
         return super(TestIpView, self).get(request, *args, **kwargs)
+
+
+class ConfigViewSet(mixins.ListModelMixin, GenericViewSet):
+    queryset = Config.objects.all()
+    serializer_class = ConfigSerializer
+    permission_classes = [AllowAny]
+
+
+class StatusView(APIView):
+    permission_classes = [IsAuthenticatedOrTokenMatchesOASRequirements]
+    required_alternate_scopes = {
+        "GET": [["accounts.status:get"]],
+    }
+
+    @service_availability('status')
+    def get(self, request, format=None):
+
+        data = {
+            "status": "ok",
+        }
+        return Response(data)
+
+
+class HealthcheckView(APIView):
+    permission_classes = [AllowAny]
+    def get(self, request, format=None):
+        data = {
+            "is_healthy": True
+        }
+        return Response(data)

apps/gooyal_oauth2/admin.py

@@ -7,7 +7,7 @@ from django.contrib.admin.sites import NotRegistered
 
 from oauth2_provider.admin import ApplicationAdmin
 
-from .models import Application, Resource, Scope
+from .models import Application, Scope
 from .forms import ApplicationForm
 
 
@@ -23,14 +23,9 @@ class ApplicationAdmin(ApplicationAdmin):
     form = ApplicationForm
 
 
-@admin.register(Resource)
-class ResourceAdmin(admin.ModelAdmin):
-    list_display = ("name", "user", "expires")
-
-
 @admin.register(Scope)
 class ScopeAdmin(admin.ModelAdmin):
-    list_display = ('name', "resource", 'description', 'is_default')
+    list_display = ('name', 'description', 'is_default')
 
 
 # admin.site.register(RestrictedApplication, RestrictedApplicationAdmin)

apps/gooyal_oauth2/decorators.py

@@ -0,0 +1,2 @@
+def session_limit_count(count=0):
+    pass

apps/gooyal_oauth2/migrations/0006_remove_scope_resource_remove_application_resource_and_more.py

@@ -0,0 +1,34 @@
+# Generated by Django 6.0.2 on 2026-04-05 13:48
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('gooyal_oauth2', '0005_alter_refreshtoken_unique_together_and_more'),
+    ]
+
+    operations = [
+        migrations.RemoveField(
+            model_name='scope',
+            name='resource',
+        ),
+        migrations.RemoveField(
+            model_name='application',
+            name='resource',
+        ),
+        migrations.AddField(
+            model_name='application',
+            name='max_allowed_session',
+            field=models.PositiveIntegerField(default=1),
+        ),
+        migrations.AlterField(
+            model_name='application',
+            name='authorization_grant_type',
+            field=models.CharField(choices=[('authorization-code', 'Authorization code'), ('urn:ietf:params:oauth:grant-type:device_code', 'Device Code'), ('implicit', 'Implicit'), ('password', 'Resource owner password-based'), ('client-credentials', 'Client credentials'), ('openid-hybrid', 'OpenID connect hybrid')], max_length=44),
+        ),
+        migrations.DeleteModel(
+            name='Resource',
+        ),
+    ]

apps/gooyal_oauth2/models.py

@@ -23,19 +23,6 @@ from .settings import oauth2_settings
 logger = logging.getLogger(__name__)
 
 
-class Resource(BaseModel):
-    name = models.CharField(max_length=255)
-
-    user = models.ForeignKey(
-        settings.AUTH_USER_MODEL, on_delete=models.CASCADE, blank=True, null=True,
-        related_name="resources"
-    )
-    expires = models.DateTimeField()
-
-    def __str__(self):
-        return self.name
-
-
 class Application(AbstractApplication, BaseModel):
     """
     Application model for use with Django OAuth Toolkit that allows the scopes
@@ -49,15 +36,10 @@ class Application(AbstractApplication, BaseModel):
         on_delete=models.PROTECT
     )
     allowed_scope = models.TextField(blank=True)
-    resource = models.OneToOneField(
-        Resource,
-        models.PROTECT,
-        blank=True, null=True,
-        help_text='The resource of application.',
-        related_name='application'
-    )
 
     avatar = models.ImageField(upload_to='avatars', null=True, blank=True)
+    max_allowed_session = models.PositiveIntegerField(default=1)
+
 
     @property
     def allowed_scopes(self):
@@ -96,13 +78,6 @@ class Scope(BaseModel):
         help_text='The application to which the scope belongs.',
         related_name='scopes'
     )
-    resource = models.ForeignKey(
-        Resource,
-        models.PROTECT,
-        blank=True, null=True,
-        help_text='The resource of scope.',
-        related_name='scopes'
-    )
     #: The name of the scope
     name = models.CharField(
         max_length=255,
@@ -122,19 +97,10 @@ class Scope(BaseModel):
 
     @property
     def final_name(self):
-        args = []
+        return self.name
-        if self.resource:
-            args.append(self.resource.name)
-
-        args.append(self.name)
-        return '.'.join(args)
 
     @property
     def final_description(self):
-        resource_name = self.resource and self.resource.name
-
-        if resource_name:
-            return f"{resource_name} -> {self.description}"
         return self.description
 
     @classmethod

apps/gooyal_oauth2/tests.py

@@ -0,0 +1,111 @@
+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
+
+from apps.core.models import Config
+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():
+        uuid = sys_uuid.uuid4()
+
+    data = Tmp()
+    return data
+
+
+class GooyalOAuth2Tests(TestCase):
+    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'
+
+    visitor_uuid = uuid.UUID('b14e8b86-8f4a-44d9-b29d-badceb47005a')
+
+    expire_datetime = timezone.now() + timedelta(seconds=3600)
+    expire_datetime.isoformat()
+
+    def setUp(self):
+        self.notifications_push_user_success_patcher = patch('apps.users.models.User.notify',
+                                                             mock_notifications_push_user_success)
+
+        self.notifications_push_user_success_patcher.start()
+
+        user, _ = User.objects.get_or_create(pk=self.user_uuid)
+        self.user = user
+
+        self.application, _created = Application.objects.get_or_create(
+            client_id=self.client_id,
+            uuid=self.application_uuid,
+            user_id=self.user_uuid,
+            max_allowed_session=1
+        )
+
+        # 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',
+                "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_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)
+

apps/gooyal_oauth2/validators.py

@@ -16,6 +16,7 @@ from oauth2_provider.models import get_access_token_model, get_application_model
 from oauth2_provider.oauth2_validators import OAuth2Validator as BaseOAuth2Validator
 from requests import Session
 
+from utils.exceptions import ServiceUnavailable
 from .settings import oauth2_settings
 from django.conf import settings
 from django.db import router, transaction
@@ -238,5 +239,25 @@ class OAuth2Validator(BaseOAuth2Validator):  # pylint: disable=w0223
             detail={'headers': headers},
         )
 
+    # def _get_token_from_authentication_server
+    def validate_bearer_token(self,token, scopes, request):
+        result = super().validate_bearer_token(token, scopes, request)
+        if result:
+            application = request.client
+            if hasattr(request, "max_allowed_session"):
+                max_allowed_session = request.max_allowed_session
+            else:
+                max_allowed_session = application.max_allowed_session
 
-Session
+            if max_allowed_session :
+                session_count = AccessToken.objects.filter(application=application, user=request.user).count()
+                if max_allowed_session >= session_count:
+                    return result
+                else:
+                    active_tokens = AccessToken.objects.filter(
+                        application=application, user=request.user
+                    ).order_by("created")[:max_allowed_session].values_list("token", flat=True)
+                    if token in active_tokens:
+                        return result
+                    else:
+                        raise ServiceUnavailable(code='max_allowed_session_reached')

apps/users/migrations/0008_alter_user_options.py

@@ -0,0 +1,17 @@
+# Generated by Django 6.0.2 on 2026-03-16 12:23
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('users', '0007_alter_user_otp'),
+    ]
+
+    operations = [
+        migrations.AlterModelOptions(
+            name='user',
+            options={'ordering': ['-date_joined'], 'verbose_name': 'user', 'verbose_name_plural': 'users'},
+        ),
+    ]

utils/exceptions.py

@@ -1,25 +1,52 @@
 import logging
 
+from django.conf import settings
+from django.utils import timezone
 from rest_framework.exceptions import APIException
 from rest_framework.views import exception_handler as drf_exception_handler
 
 logger = logging.getLogger(__name__)
 def exception_handler(exc, context):
+    logger.exception(exc)
     # پاسخ پیش‌فرض DRF را دریافت می‌کنیم
     response = drf_exception_handler(exc, context)
 
     if response is not None:
-        # ساختار دلخواه خود را تعریف می‌کنیم
         response_data = {
             'success': False,
             'status_code': response.status_code,
-            'message': 'An error occurred',
+            'status_message': str(exc.default_detail),
-            'details': {}
+            'details': exc.detail,
         }
 
         if isinstance(exc, APIException):
-            response_data['message'] = exc.detail if isinstance(exc.detail, str) else "Validation error"
+            try:
+                if isinstance(exc.detail, dict):
+                    if 'code' in exc.detail:
+                        error = exc.detail['code']
+                    else:
+                        attr = next(iter(exc.detail))
+                        error = exc.detail[attr].code
+                else:
+                    error = exc.detail.code or exc.code
+            except:
+                error = ''
+
+            # if isinstance(exc.detail, str):
+            #     message = exc.detail
+            # elif isinstance(exc.detail, dict):
+            #     if 'message' in exc.detail or 'string' in exc.detail:
+            #         message = exc.detail.get('message') or exc.detail.get('string')
+            #     else:
+            #         message = error
+            #
+            # else:
+            #     message = error
+            #
+            # response_data['message'] = message
             response_data['details'] = exc.detail if isinstance(exc.detail, dict) else {}
+            response_data['details']['error'] = error
+            response_data['details']['timestamp'] = timezone.now().isoformat()
 
         response.data = response_data
 
@@ -43,14 +70,24 @@ class ErrorMiddleware:
             {
                 "success": False,
                 "status_code": 500,
-                "message": "Internal server error",
+                "status_message": "Internal server error",
-                "details": str(exception)  # فقط در حالت توسعه! در تولید بهتر است لاگ شود.
+                "details": str(exception) if settings.DEBUG else None,
             },
             status=500
         )
 
+from rest_framework.exceptions import APIException
+from rest_framework import status
+from django.utils.translation import gettext_lazy as _
+
 
 class UnprocessableEntity(APIException):
     status_code = 422
     default_detail = 'The request was well-formed but cannot be processed due to semantic errors.'
-    default_code = 'unprocessable_entity'
+    default_code = 'unprocessable_entity'
+
+class ServiceUnavailable(APIException):
+    status_code = status.HTTP_503_SERVICE_UNAVAILABLE
+    default_ = _('SERVICE_UNAVAILABLE')
+    default_code = 'service_unavailable'
+    default_detail = 'Service Unavailable'