Merge branch 'osm' of github.com:jumpserver/jumpserver into osm

apps/audits/signal_handlers/login_log.py

@@ -33,6 +33,7 @@ class AuthBackendLabelMapping(LazyObject):
         backend_label_mapping[settings.AUTH_BACKEND_PUBKEY] = _("SSH Key")
         backend_label_mapping[settings.AUTH_BACKEND_MODEL] = _("Password")
         backend_label_mapping[settings.AUTH_BACKEND_SSO] = _("SSO")
+        backend_label_mapping[settings.AUTH_BACKEND_CUSTOM_SSO] = _("Custom SSO")
         backend_label_mapping[settings.AUTH_BACKEND_AUTH_TOKEN] = _("Auth Token")
         backend_label_mapping[settings.AUTH_BACKEND_WECOM] = _("WeCom")
         backend_label_mapping[settings.AUTH_BACKEND_FEISHU] = _("FeiShu")

apps/authentication/api/__init__.py

@@ -17,3 +17,4 @@ from .temp_token import *
 from .token import *
 from .face import *
 from .access_token import *
+from .custom_sso import *

apps/authentication/api/custom_sso.py

@@ -0,0 +1,83 @@
+from django.utils.module_loading import import_string
+from django.conf import settings
+from django.contrib.auth import get_user_model
+from django.contrib.auth import login
+from django.http.response import HttpResponseRedirect
+
+from rest_framework.generics import RetrieveAPIView
+from rest_framework.response import Response
+from rest_framework import status
+from rest_framework.permissions import AllowAny
+
+from common.utils import get_logger
+from ..mixins import AuthMixin
+
+__all__  = ['CustomSSOLoginAPIView']
+
+logger = get_logger(__file__)
+
+
+custom_sso_authenticate_method = None
+
+if settings.AUTH_CUSTOM_SSO:
+    ''' 保证自定义 SSO 认证方法在服务运行时不能被更改，只在第一次调用时加载一次 '''
+    try:
+        custom_auth_method_path = 'data.auth.custom_sso.authenticate'
+        custom_sso_authenticate_method = import_string(custom_auth_method_path)
+    except Exception as e:
+        logger.warning('Import custom SSO auth method failed: {}, Maybe not enabled'.format(e))
+
+
+class CustomSSOLoginAPIView(AuthMixin, RetrieveAPIView):
+
+    permission_classes = [AllowAny]
+
+    def retrieve(self, request, *args, **kwargs):
+        if not self.is_enabled():
+            error = 'Custom SSO authentication is disabled.'
+            return Response({'detail': error}, status=status.HTTP_403_FORBIDDEN)
+
+        query_params = {}
+        for param in settings.AUTH_CUSTOM_SSO_QUERY_PARAMS:
+            value = self.request.query_params.get(param)
+            if not value:
+                error = f'Missing required query parameter: {param}'
+                return Response({'detail': error}, status=status.HTTP_400_BAD_REQUEST)
+            query_params[param] = value
+        
+        user, error = self.authenticate(**query_params)
+        if user:
+            login(request, user, backend=settings.AUTH_BACKEND_CUSTOM_SSO)
+            self.send_auth_signal(success=True, user=user)
+            next_url = request.query_params.get('next', '/')
+            return HttpResponseRedirect(next_url)
+        else:
+            self.send_auth_signal(success=False, reason=error)
+            return Response({'detail': error}, status=status.HTTP_401_UNAUTHORIZED)
+
+    def is_enabled(self):
+        return settings.AUTH_CUSTOM_SSO and callable(custom_sso_authenticate_method)
+
+    def authenticate(self, **query_params):
+        try:
+            userinfo: dict = custom_sso_authenticate_method(**query_params)
+        except Exception as e:
+            error = f'Custom SSO authenticate error: {e}'
+            return None, error
+        
+        try:
+            user, created = self.get_or_create_user_from_userinfo(userinfo)
+            return user, ''
+        except Exception as e:
+            error = f'Custom SSO get or create user error: {e}'
+            return None, error
+
+    def get_or_create_user_from_userinfo(self, userinfo: dict):
+        username = userinfo['username']
+        attrs = ['name', 'username', 'email', 'is_active']
+        defaults = {attr: userinfo[attr] for attr in attrs}
+        user, created = get_user_model().objects.get_or_create(
+            username=username, defaults=defaults
+        )
+        # TODO: get and set role attribute for user
+        return user, created

apps/authentication/backends/sso.py

@@ -12,6 +12,15 @@ class SSOAuthentication(JMSBaseAuthBackend):
         pass
 
 
+class CustomSSOAuthentication(JMSBaseAuthBackend):
+    @staticmethod
+    def is_enabled():
+        return settings.AUTH_CUSTOM_SSO
+
+    def authenticate(self):
+        pass
+
+
 class WeComAuthentication(JMSBaseAuthBackend):
     @staticmethod
     def is_enabled():

apps/authentication/urls/api_urls.py

@@ -1,6 +1,7 @@
 # coding:utf-8
 #
 from django.urls import path
+from django.conf import settings
 from rest_framework.routers import DefaultRouter
 
 from .. import api
@@ -47,4 +48,11 @@ urlpatterns = [
     path('user-session/', api.UserSessionApi.as_view(), name='user-session'),
 ]
 
+if settings.AUTH_CUSTOM_SSO:
+    urlpatterns += [
+        path('custom-sso/login/', api.CustomSSOLoginAPIView.as_view(), name='custom-sso-login'),
+    ]
+
+
+
 urlpatterns += router.urls + passkey_urlpatterns

apps/jumpserver/conf.py

@@ -265,6 +265,10 @@ class Config(dict):
 
         'SMS_CUSTOM_FILE_MD5': '',
 
+        'AUTH_CUSTOM_SSO': False,
+        'AUTH_CUSTOM_SSO_FILE_MD5': '',
+        'AUTH_CUSTOM_SSO_QUERY_PARAMS': ['token'],
+
         # 临时密码
         'AUTH_TEMP_TOKEN': False,
 

apps/jumpserver/settings/auth.py

@@ -273,6 +273,7 @@ AUTH_BACKEND_OIDC_CODE = 'authentication.backends.oidc.OIDCAuthCodeBackend'
 AUTH_BACKEND_RADIUS = 'authentication.backends.radius.RadiusBackend'
 AUTH_BACKEND_CAS = 'authentication.backends.cas.CASBackend'
 AUTH_BACKEND_SSO = 'authentication.backends.sso.SSOAuthentication'
+AUTH_BACKEND_CUSTOM_SSO = 'authentication.backends.sso.CustomSSOAuthentication'
 AUTH_BACKEND_WECOM = 'authentication.backends.sso.WeComAuthentication'
 AUTH_BACKEND_DINGTALK = 'authentication.backends.sso.DingTalkAuthentication'
 AUTH_BACKEND_FEISHU = 'authentication.backends.sso.FeiShuAuthentication'
@@ -295,7 +296,7 @@ AUTHENTICATION_BACKENDS = [
     # 扫码模式
     AUTH_BACKEND_WECOM, AUTH_BACKEND_DINGTALK, AUTH_BACKEND_FEISHU, AUTH_BACKEND_LARK, AUTH_BACKEND_SLACK,
     # Token模式
-    AUTH_BACKEND_AUTH_TOKEN, AUTH_BACKEND_SSO, AUTH_BACKEND_TEMP_TOKEN,
+    AUTH_BACKEND_AUTH_TOKEN, AUTH_BACKEND_SSO, AUTH_BACKEND_CUSTOM_SSO, AUTH_BACKEND_TEMP_TOKEN,
     AUTH_BACKEND_PASSKEY
 ]
 
@@ -360,3 +361,11 @@ ONLY_ALLOW_AUTH_FROM_SOURCE = CONFIG.ONLY_ALLOW_AUTH_FROM_SOURCE
 PRIVACY_MODE = CONFIG.PRIVACY_MODE
 
 SAML_FOLDER = os.path.join(BASE_DIR, 'authentication', 'backends', 'saml2')
+
+AUTH_CUSTOM_SSO = CONFIG.AUTH_CUSTOM_SSO
+AUTH_CUSTOM_SSO_FILE_MD5 = CONFIG.AUTH_CUSTOM_SSO_FILE_MD5
+AUTH_CUSTOM_SSO_FILE_PATH = os.path.join(PROJECT_DIR, 'data', 'auth', 'custom_sso.py')
+if AUTH_CUSTOM_SSO and AUTH_CUSTOM_SSO_FILE_MD5 != get_file_md5(AUTH_CUSTOM_SSO_FILE_PATH):
+    # 如果启用了自定义 SSO 认证，但文件 MD5 不匹配，则不启用自定义 SSO 认证
+    AUTH_CUSTOM_SSO = False
+AUTH_CUSTOM_SSO_QUERY_PARAMS = CONFIG.AUTH_CUSTOM_SSO_QUERY_PARAMS