fix: pubkey auth require svc sign

apps/authentication/backends/pubkey.py

@@ -1,8 +1,9 @@
 # -*- coding: utf-8 -*-
 #
-from django.contrib.auth import get_user_model
 from django.conf import settings
+from django.contrib.auth import get_user_model
 
+from common.permissions import ServiceAccountSignaturePermission
 from .base import JMSBaseAuthBackend
 
 UserModel = get_user_model()
@@ -18,6 +19,10 @@ class PublicKeyAuthBackend(JMSBaseAuthBackend):
     def authenticate(self, request, username=None, public_key=None, **kwargs):
         if not public_key:
             return None
+
+        permission = ServiceAccountSignaturePermission()
+        if not permission.has_permission(request, None):
+            return None
         if username is None:
             username = kwargs.get(UserModel.USERNAME_FIELD)
         try:
@@ -26,7 +31,7 @@ class PublicKeyAuthBackend(JMSBaseAuthBackend):
             return None
         else:
             if user.check_public_key(public_key) and \
-                  self.user_can_authenticate(user):
+                    self.user_can_authenticate(user):
                 return user
 
     def get_user(self, user_id):

apps/common/permissions.py

@@ -86,3 +86,38 @@ class UserConfirmation(permissions.BasePermission):
         min_level = ConfirmType.values.index(confirm_type) + 1
         name = 'UserConfirmationLevel{}TTL{}'.format(min_level, ttl)
         return type(name, (cls,), {'min_level': min_level, 'ttl': ttl, 'confirm_type': confirm_type})
+
+
+class ServiceAccountSignaturePermission(permissions.BasePermission):
+    def has_permission(self, request, view):
+        from authentication.models import AccessKey
+        from common.utils.crypto import get_aes_crypto
+        signature = request.META.get('HTTP_X_JMS_SVC', '')
+        if not signature or not signature.startswith('Sign'):
+            return False
+        data = signature[4:].strip()
+        if not data or ':' not in data:
+            return False
+        ak_id, time_sign = data.split(':', 1)
+        if not ak_id or not time_sign:
+            return False
+        ak = AccessKey.objects.filter(id=ak_id).first()
+        if not ak or not ak.is_active:
+            return False
+        if not ak.user or not ak.user.is_active or not ak.user.is_service_account:
+            return False
+        aes = get_aes_crypto(str(ak.secret).replace('-', ''), mode='ECB')
+        try:
+            timestamp = aes.decrypt(time_sign)
+            if not timestamp or not timestamp.isdigit():
+                return False
+            timestamp = int(timestamp)
+            interval = abs(int(time.time()) - timestamp)
+            if interval > 30:
+                return False
+            return True
+        except Exception:
+            return False
+
+    def has_object_permission(self, request, view, obj):
+        return False