chore(user): check user and app status

Author: duanhongyi

rootfs/api/authentication.py

@@ -5,6 +5,7 @@
 from rest_framework import authentication
 from rest_framework.authentication import get_authorization_header
 from rest_framework import exceptions
+from api.clients import ManagerAPI
 
 logger = logging.getLogger(__name__)
 
@@ -21,7 +22,7 @@ def authenticate(self, request):
 class DryccAuthentication(authentication.BaseAuthentication):
 
     keywords = ('token', 'bearer')
-    ignore_authentication_failed = False
+    manager_api = ManagerAPI()
 
     def parse_header(self, request):
         try:
@@ -42,22 +43,21 @@ def parse_header(self, request):
             raise exceptions.AuthenticationFailed(msg)
 
     def authenticate(self, request):
-        token_type, token = self.parse_header(request)
+        user, (token_type, token) = None, self.parse_header(request)
         if token_type is None or token is None:
             return None
-        try:
-            if token_type == 'bearer':  # drycc oauth access token
-                from api.apps_extra.social_core.backends import OauthCacheManager
-                return OauthCacheManager().get_user(token), token
-            # drycc token
+        if token_type == 'bearer':  # drycc oauth access token
+            from api.apps_extra.social_core.backends import OauthCacheManager
+            user = OauthCacheManager().get_user(token)
+        elif token_type == 'token':  # drycc token
             user = cache.get(token, None)
             if not user:
-                return self.authenticate_credentials(token)
-            return user, token
-        except exceptions.AuthenticationFailed as e:
-            if not self.ignore_authentication_failed:
-                raise e
-        return None
+                user, token = self.authenticate_credentials(token)
+        if user:
+            is_active, message = self.manager_api.get_user_status(user.id)
+            if not is_active:
+                raise exceptions.AuthenticationFailed(message)
+        return user, token if user else None
 
     def authenticate_credentials(self, key):
         from api.models.base import Token

rootfs/api/clients.py

@@ -1,4 +1,3 @@
-import base64
 import logging
 import requests
 import urllib.parse
@@ -11,98 +10,6 @@
 logger = logging.getLogger(__name__)
 
 
-class ManagerAPI(object):
-
-    def __init__(self, timeout=3):
-        self.timeout = timeout
-        token = base64.b85encode(b"%s:%s" % (
-            settings.SOCIAL_AUTH_DRYCC_KEY.encode("utf8"),
-            settings.SOCIAL_AUTH_DRYCC_SECRET.encode("utf8"),
-        )).decode("utf8")
-        self.headers = {
-            'Content-Type': 'application/json',
-            'Authorization': 'token %s' % token,
-            'User-Agent': user_agent('Drycc Controller ', drycc_version)
-        }
-
-    def request(self, method, url, **kwargs):
-        headers = kwargs.get("headers", {})
-        headers.update(self.headers)
-        kwargs["headers"] = headers
-        kwargs["timeout"] = self.timeout
-        return requests.request(method, url, **kwargs)
-
-    def get(self, url, params=None, **kwargs):
-        return self.request('get', url, params=params, **kwargs)
-
-    def options(self, url, **kwargs):
-        return self.request('options', url, **kwargs)
-
-    def head(self, url, **kwargs):
-        kwargs.setdefault('allow_redirects', False)
-        return self.request('head', url, **kwargs)
-
-    def post(self, url, data=None, json=None, **kwargs):
-        return self.request('post', url, data=data, json=json, **kwargs)
-
-    def put(self, url, data=None, **kwargs):
-        return self.request('put', url, data=data, **kwargs)
-
-    def patch(self, url, data=None, **kwargs):
-        return self.request('patch', url, data=data, **kwargs)
-
-    def delete(self, url, **kwargs):
-        return self.request('delete', url, **kwargs)
-
-
-class WorkspaceAPI(ManagerAPI):
-
-    def get_status(self, workspace_id):
-        """
-        {
-            "is_active": False,
-            "message": "The user is in arrears"
-        }
-        """
-        key = f"workspace:status:{workspace_id}"
-        status = cache.get(key)
-        if not status:
-            url = f"{settings.WORKFLOW_MANAGER_URL}/workspaces/{workspace_id}/status/"
-            try:
-                status = self.get(url=url, timeout=self.timeout).json()
-            except requests.exceptions.Timeout as ex:
-                msg = f"request workspace {workspace_id} timeout, skipping verification."
-                status = {"is_active": True, "message": msg}
-                logger.error(msg)
-                logger.exception(ex)
-            cache.set(key, status, timeout=settings.DRYCC_CACHE_USER_TIME)
-        return status
-
-
-class UserAPI(WorkspaceAPI):
-    """Backward-compatible alias for legacy call sites."""
-
-
-class UsageAPI(ManagerAPI):
-
-    def post(self, usage: List[Dict[str, str]]):
-        """
-        [
-            {
-                "app_id":  "test",
-                "workspace": "test",
-                "name": "web",
-                "type": "limits",
-                "unit": "std1.large.c1m1",
-                "usage": "2",
-                "timestamp": "1609231998.9103732"
-            }
-        ]
-        """
-        url = "%s/usage/" % settings.WORKFLOW_MANAGER_URL
-        return super().post(url=url, json=usage)
-
-
 class PassportAPI(object):
     """Service-to-service client for Drycc Passport using OAuth2 client_credentials."""
 
@@ -113,7 +20,8 @@ def __init__(self, timeout=10):
         self.timeout = timeout
         self.base_url = settings.DRYCC_PASSPORT_URL.rstrip("/")
 
-    def _get_token(self) -> str:
+    @property
+    def headers(self) -> str:
         token = cache.get(self.TOKEN_CACHE_KEY)
         if token and self.get_scopes(token) == set(settings.DRYCC_PASSPORT_SCOPES.split()):
             return token
@@ -132,9 +40,14 @@ def _get_token(self) -> str:
         token = body["access_token"]
         ttl = max(int(body.get("expires_in", 3600)) - self.TOKEN_REFRESH_LEEWAY, 60)
         cache.set(self.TOKEN_CACHE_KEY, token, timeout=ttl)
-        return token
+        return {
+            "Authorization": f"Bearer {token}",
+            "Content-Type": "application/json",
+            "User-Agent": user_agent("Drycc Controller", drycc_version),
+        }
 
-    def get_scopes(self, token):
+    @staticmethod
+    def get_scopes(token):
         def _get_scopes():
             endpoint = getattr(settings, 'SOCIAL_AUTH_DRYCC_OIDC_ENDPOINT', None)
             if not endpoint:
@@ -156,17 +69,54 @@ def _get_scopes():
             f"drycc_oauth_scopes_v2_{token}", _get_scopes, settings.DRYCC_CACHE_USER_TIME)
 
     def send_message(self, username: str, message: Dict) -> None:
-        token = self._get_token()
-        headers = {
-            "Authorization": f"Bearer {token}",
-            "Content-Type": "application/json",
-            "User-Agent": user_agent("Drycc Controller", drycc_version),
-        }
         body = {**message, "username": username}
         resp = requests.post(
             f"{self.base_url}/messages/",
             json=body,
-            headers=headers,
+            headers=self.headers,
             timeout=self.timeout,
         )
         resp.raise_for_status()
+
+
+class ManagerAPI(object):
+
+    def __init__(self, timeout=10):
+        self.timeout = timeout
+        self.headers = PassportAPI(timeout=timeout).headers if self.enabled else None
+        self.base_url = settings.WORKFLOW_MANAGER_URL.rstrip("/") if self.enabled else None
+
+    @property
+    def enabled(self):
+        return settings.WORKFLOW_MANAGER_URL is not None
+
+    def send_usage(self, usage: List[Dict[str, str]]):
+        if not self.enabled:
+            logger.info("WORKFLOW_MANAGER_URL is not set, skipping send_usage")
+            return
+        url = f"{self.base_url}/usage/"
+        return requests.post(url=url, json=usage, headers=self.headers, timeout=self.timeout)
+
+    def get_status(self, resource_type: str, resource_id: str):
+        if not self.enabled:
+            logger.info("WORKFLOW_MANAGER_URL is not set, skipping get_status")
+            return True, None
+        key = f"{resource_type}:status:{resource_id}"
+        status = cache.get(key)
+        if not status:
+            url = f"{self.base_url}/{resource_type}s/{resource_id}/status/"
+            try:
+                status = requests.get(url=url, headers=self.headers, timeout=self.timeout).json()
+            except requests.exceptions.Timeout as ex:
+                msg = f"request {resource_type} {resource_id} timeout, skipping verification."
+                status = {"is_active": True, "message": msg}
+                logger.error(msg)
+                logger.exception(ex)
+            cache.set(key, status, timeout=settings.DRYCC_CACHE_USER_TIME)
+        return status.get("is_active", True), status.get("message", None)
+
+    def get_app_status(self, app_id):
+        return self.get_status("app", app_id)
+
+    def get_user_status(self, user_id):
+        return self.get_status("user", user_id)

rootfs/api/consumers.py

@@ -21,13 +21,15 @@
 from channels.generic.http import AsyncHttpConsumer
 from channels.generic.websocket import AsyncWebsocketConsumer
 
+from .clients import ManagerAPI
 from .models.app import App
 from .models.volume import Volume
-from .permissions import IsAppUser, get_app_status
+from .permissions import IsAppUser
 
 
 class AppPermChecker(object):
     timeout = 60 * 60
+    manager_api = ManagerAPI()
     check_permission = IsAppUser().has_object_permission
 
     def __init__(self, scope):
@@ -44,7 +46,7 @@ async def has_perm(self):
                 app = await App.objects.aget(id=app_id)
                 request = namedtuple("Request", ["user", "method"])(self.scope["user"], "GET")
                 if await sync_to_async(self.check_permission)(request, None, app):
-                    permission = await sync_to_async(get_app_status)(app)
+                    permission = await sync_to_async(self.manager_api.get_app_status)(app_id)
                 else:
                     permission = (False, "permission denied")
                 if permission[0]:

rootfs/api/management/commands/upload_app_usage.py

@@ -4,9 +4,9 @@
 from datetime import timedelta
 from django.utils import timezone
 from django.core.management.base import BaseCommand
-from django.conf import settings
 from api.models.app import App
 from api.tasks import send_usage
+from api.clients import ManagerAPI
 
 logger = logging.getLogger(__name__)
 
@@ -15,7 +15,7 @@ class Command(BaseCommand):
     """Management command for push data to manager"""
 
     def handle(self, *args, **options):
-        if settings.WORKFLOW_MANAGER_URL:
+        if ManagerAPI().enabled:
             now = timezone.now()
             start_time, timestamp, task_id = now, int(now.timestamp()), uuid.uuid4().hex
             logger.info(f"pushing {task_id} resources to workflow_manager when {now}")

rootfs/api/management/commands/upload_gateway_usage.py

@@ -4,9 +4,9 @@
 from datetime import timedelta
 from django.utils import timezone
 from django.core.management.base import BaseCommand
-from django.conf import settings
 from api.models.gateway import Gateway
 from api.tasks import send_usage
+from api.clients import ManagerAPI
 
 logger = logging.getLogger(__name__)
 
@@ -15,7 +15,7 @@ class Command(BaseCommand):
     """Management command for push data to manager"""
 
     def handle(self, *args, **options):
-        if settings.WORKFLOW_MANAGER_URL:
+        if ManagerAPI().enabled:
             now = timezone.now()
             start_time, timestamp, task_id = now, int(now.timestamp()), uuid.uuid4().hex
             logger.info(f"pushing {task_id} gateways to workflow_manager when {now}")

rootfs/api/management/commands/upload_network_usage.py

@@ -6,10 +6,10 @@
 from asgiref.sync import async_to_sync
 from django.utils import timezone
 from django.core.management.base import BaseCommand
-from django.conf import settings
 from api import monitor
 from api.models.app import App
 from api.tasks import send_usage
+from api.clients import ManagerAPI
 
 logger = logging.getLogger(__name__)
 
@@ -44,7 +44,7 @@ def _upload_network_usage(self, start_time, app_map, timestamp):
         )
 
     def handle(self, *args, **options):
-        if settings.WORKFLOW_MANAGER_URL:
+        if ManagerAPI().enabled:
             now = timezone.now()
             start_time, timestamp, task_id = now, int(now.timestamp()), uuid.uuid4().hex
             logger.info(f"pushing {task_id} networks to workflow_manager when {now}")

rootfs/api/management/commands/upload_resource_usage.py

@@ -4,9 +4,9 @@
 from datetime import timedelta
 from django.utils import timezone
 from django.core.management.base import BaseCommand
-from django.conf import settings
 from api.models.resource import Resource
 from api.tasks import send_usage
+from api.clients import ManagerAPI
 
 logger = logging.getLogger(__name__)
 
@@ -15,7 +15,7 @@ class Command(BaseCommand):
     """Management command for push data to manager"""
 
     def handle(self, *args, **options):
-        if settings.WORKFLOW_MANAGER_URL:
+        if ManagerAPI().enabled:
             now = timezone.now()
             start_time, timestamp, task_id = now, int(now.timestamp()), uuid.uuid4().hex
             logger.info(f"pushing {task_id} resources to workflow_manager when {now}")

rootfs/api/management/commands/upload_volume_usage.py

@@ -6,10 +6,10 @@
 from asgiref.sync import async_to_sync
 from django.utils import timezone
 from django.core.management.base import BaseCommand
-from django.conf import settings
 from api import monitor
 from api.models.app import App
 from api.tasks import send_usage
+from api.clients import ManagerAPI
 
 logger = logging.getLogger(__name__)
 
@@ -44,7 +44,7 @@ def _upload_volume_usage(self, start_time, app_map, timestamp):
         )
 
     def handle(self, *args, **options):
-        if settings.WORKFLOW_MANAGER_URL:
+        if ManagerAPI().enabled:
             now = timezone.now()
             start_time, timestamp, task_id = now, int(now.timestamp()), uuid.uuid4().hex
             logger.info(f"pushing {task_id} volumes to workflow_manager when {now}")

rootfs/api/migrations/0030_delete_blocklist_remove_app_suspended_state.py

@@ -0,0 +1,20 @@
+# Generated by Django 5.2.9 on 2026-05-20 08:51
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('api', '0029_key_api_key_fingerp_4e77c5_idx_and_more'),
+    ]
+
+    operations = [
+        migrations.DeleteModel(
+            name='Blocklist',
+        ),
+        migrations.RemoveField(
+            model_name='app',
+            name='suspended_state',
+        ),
+    ]