🎨 工具管理凭证校验逻辑调整

Author: nickcdon

server/projects/main/apps/codeproj/core/scmmgr.py

@@ -217,11 +217,13 @@ def create_checktool_auth(cls, checktool, user, scm_auth_type=None,
         """
         auth_key = "%s_%s" % (ScmAuth.KeyEnum.TOOL, checktool.id)
         logger.debug("create checktool scm auth: %s" % auth_key)
+        # 兼容逻辑
+        if scm_auth_type == ScmAuth.ScmAuthTypeEnum.OAUTH and not scm_oauth:
+            scm_oauth = cls.get_scm_auth(user)
         scm_auth = cls.create_or_update_auth(
             auth_key=auth_key, auth_type=scm_auth_type,
             scm_account=scm_account, scm_ssh=scm_ssh_info,
-            scm_oauth=scm_oauth
-        )
+            scm_oauth=scm_oauth)
         checktool.scm_auth = scm_auth
         checktool.save(user=user)
 
@@ -239,6 +241,9 @@ def create_toollib_auth(cls, toollib, user, scm_auth_type=None,
         """
         auth_key = "%s_%s" % (ScmAuth.KeyEnum.TOOLLIB, toollib.id)
         logger.debug("create toollib scm auth: %s" % auth_key)
+        # 兼容逻辑
+        if scm_auth_type == ScmAuth.ScmAuthTypeEnum.OAUTH and not scm_oauth:
+            scm_oauth = cls.get_scm_auth(user)
         scm_auth = cls.create_or_update_auth(
             auth_key=auth_key, auth_type=scm_auth_type,
             scm_account=scm_account, scm_ssh=scm_ssh_info,

server/projects/main/apps/scan_conf/core/basemgr.py

@@ -20,6 +20,7 @@
 # 项目内
 from apps.scan_conf import models
 from apps.scan_conf.api_filters import base as base_filters
+from apps.codeproj.core.scmmgr import ScmAuthManager
 from apps.base.basemodel import CDBaseModel
 
 logger = logging.getLogger(__name__)
@@ -196,3 +197,76 @@ def get_checkpackage_rules_filter(cls, queryset, filter_class=base_filters.Packa
             for field in filter_class.Meta.fields
         }
         return cls._get_filter(queryset, field_maps, models.PackageMap, user=user)
+
+
+class CommonManager(object):
+    """公共manager
+    """
+
+    @classmethod
+    def is_eq_instance_auth(cls, scm_auth, instance=None):
+        """用于更新凭证时判断，判断传入的scm_auth是否与实例原本的scm_auth相同，相同则返回该凭证的信息
+        :param scm_auth: ScmAuth
+        :param instance: ToolLib or CheckTool
+        :return is_eq, credential_info: bool, dict
+        """
+        if instance and instance.scm_auth:
+            auth_type = scm_auth.get("auth_type")
+            scm_oauth = scm_auth.get("scm_oauth")
+            scm_account = scm_auth.get("scm_account")
+            scm_ssh = scm_auth.get("scm_ssh")
+            if auth_type == models.ScmAuth.ScmAuthTypeEnum.OAUTH and \
+               instance.scm_auth.scm_oauth == scm_oauth:
+                return True, scm_oauth.credential_info
+            elif auth_type == models.ScmAuth.ScmAuthTypeEnum.PASSWORD and \
+                 instance.scm_auth.scm_account == scm_account:
+                return True, scm_account.credential_info
+            elif auth_type == models.ScmAuth.ScmAuthTypeEnum.SSHTOKEN and \
+                 instance.scm_auth.scm_ssh == scm_ssh:
+                return True, scm_ssh.credential_info
+        return False, None
+
+    @classmethod
+    def validate_scm_auth(cls, scm_auth, user):
+        """校验传入的scm_auth是否是当前用户创建的凭证，否则认为是无效凭证
+        :param scm_auth: ScmAuth
+        :param user: User
+        :return credential_info: dict
+        """
+        auth_type = scm_auth.get("auth_type")
+        scm_oauth = scm_auth.get("scm_oauth")
+        scm_account = scm_auth.get("scm_account")
+        scm_ssh = scm_auth.get("scm_ssh")
+        if auth_type == models.ScmAuth.ScmAuthTypeEnum.OAUTH:
+            if not scm_oauth:
+                # 如果没有传递scm_oauth
+                scm_oauth = ScmAuthManager.get_scm_auth(user)
+            if not scm_oauth or scm_oauth.user != user:
+                raise ParseError({"scm_auth": "请选择有效OAuth凭证"})
+            credential_info = scm_oauth.credential_info
+        elif auth_type == models.ScmAuth.ScmAuthTypeEnum.PASSWORD:
+            if not scm_account or scm_account.user != user:
+                raise ParseError({"scm_auth": "请选择有效HTTP凭证"})
+            credential_info = scm_account.credential_info
+        elif auth_type == models.ScmAuth.ScmAuthTypeEnum.SSHTOKEN:
+            if not scm_ssh or scm_ssh.user != user:
+                raise ParseError({"scm_auth": "请选择有效SSH凭证"})
+            credential_info = scm_ssh.credential_info
+        else:
+            raise ParseError({"auth_type": "不支持%s鉴权方式" % auth_type})
+        return credential_info
+
+    @classmethod
+    def get_and_check_scm_auth(cls, scm_auth, user, instance=None):
+        """获取并校验凭证，获取凭证信息
+        :param scm_auth: ScmAuth
+        :param user: User
+        :param instance: ToolLib or CheckTool
+        :return credential_info: dict
+        """
+        # 更新凭证时，用于判断凭证是否切换
+        is_eq, credential_info = cls.is_eq_instance_auth(scm_auth, instance)
+        if is_eq is False:
+            # 创建或凭证切换时，校验凭证并获取凭证信息
+            credential_info = cls.validate_scm_auth(scm_auth, user)
+        return credential_info

server/projects/main/apps/scan_conf/serializers/base/tool.py

@@ -15,13 +15,14 @@
 
 # 项目内
 from apps.scan_conf import models
-from apps.scan_conf.core import ToolLibMapManager, ToolLibSchemeManager, CheckToolManager, ToolLibManager
+from apps.scan_conf.core import CommonManager, ToolLibMapManager, ToolLibSchemeManager, \
+                                CheckToolManager, ToolLibManager
+from apps.scan_conf.serializers.base.rule import CheckRuleSerializer
 from apps.base.serializers import CDBaseModelSerializer
 from apps.authen.models import Organization
 from apps.authen.serializers.base import ScmAuthCreateSerializer, ScmAuthSerializer
 from apps.authen.serializers.base_org import OrganizationSimpleSerializer
 from apps.codeproj.core import ScmAuthManager
-from .rule import CheckRuleSerializer
 
 logger = logging.getLogger(__name__)
 
@@ -215,25 +216,8 @@ def validate(self, attrs):
         if not scm_auth:
             # raise serializers.ValidationError({"scm_auth": "凭证为必填项"})
             return super().validate(attrs)
-        auth_type = scm_auth.get("auth_type")
-        if auth_type == models.ScmAuth.ScmAuthTypeEnum.OAUTH:
-            scm_oauth = scm_auth.get("scm_oauth")
-            if not scm_oauth or scm_oauth.user != user:
-                raise serializers.ValidationError({"scm_auth": "请选择有效OAuth凭证"})
-            credential_info = scm_oauth.credential_info
-        elif auth_type == models.ScmAuth.ScmAuthTypeEnum.PASSWORD:
-            scm_account = scm_auth.get("scm_account")
-            if not scm_account or scm_account.user != user:
-                raise serializers.ValidationError({"scm_auth": "请选择有效HTTP凭证"})
-            credential_info = scm_account.credential_info
-        elif auth_type == models.ScmAuth.ScmAuthTypeEnum.SSHTOKEN:
-            scm_ssh = scm_auth.get("scm_ssh")
-            if not scm_ssh or scm_ssh.user != user:
-                raise serializers.ValidationError({"scm_auth": "请选择有效SSH凭证"})
-            credential_info = scm_ssh.credential_info
-        else:
-            raise serializers.ValidationError({"auth_type": ["不支持%s鉴权方式" % auth_type]})
-        # 校验
+        # 校验凭证有效性
+        credential_info = CommonManager.get_and_check_scm_auth(scm_auth, user, instance=self.instance)
         ScmAuthManager.check_scm_url_credential(scm_type, scm_url, credential_info)
         return super().validate(attrs)
 
@@ -243,13 +227,18 @@ def _create_or_update(self, validated_data, instance=None):
         scm_auth = validated_data.pop("scm_auth", None)
         instance, _ = ToolLibManager.create_or_update(name, user, instance=instance, **validated_data)
         # 保存凭证信息
-        if validated_data.get("scm_type") != models.ToolLib.ScmTypeEnum.LINK and scm_auth:
-            ScmAuthManager.create_toollib_auth(
-                instance, user, scm_auth_type=scm_auth.get("auth_type"),
-                scm_account=scm_auth.get("scm_account"),
-                scm_ssh_info=scm_auth.get("scm_ssh"),
-                scm_oauth=scm_auth.get("scm_oauth"),
-            )
+        if validated_data.get("scm_type") != models.ToolLib.ScmTypeEnum.LINK:
+            if scm_auth:
+                ScmAuthManager.create_toollib_auth(
+                    instance, user, scm_auth_type=scm_auth.get("auth_type"),
+                    scm_account=scm_auth.get("scm_account"),
+                    scm_ssh_info=scm_auth.get("scm_ssh"),
+                    scm_oauth=scm_auth.get("scm_oauth"),
+                )
+            elif instance:
+                # 更新时允许移除scm_auth
+                instance.scm_auth = None
+                instance.save(user=user)
         return instance
 
     def create(self, validated_data):

server/projects/main/apps/scan_conf/serializers/v3/tool.py

@@ -17,7 +17,7 @@
 # 项目内
 from apps.scan_conf import models
 from apps.scan_conf.serializers import base
-from apps.scan_conf.core import CheckToolManager
+from apps.scan_conf.core import CommonManager, CheckToolManager
 from apps.authen.models import Organization
 from apps.authen.serializers.base import ScmAuthCreateSerializer, ScmAuthSerializer
 from apps.codeproj.core import ScmAuthManager
@@ -82,6 +82,7 @@ def get_user(self):
     def validate(self, attrs):
         attrs["org"] = get_and_check_view_org(self)
         user = self.get_user()
+        scm_type = attrs.get("scm_type")
         scm_auth = attrs.get("scm_auth")
         scm_url = attrs.get('scm_url')
         run_cmd = attrs.get('run_cmd')
@@ -93,26 +94,8 @@ def validate(self, attrs):
                 raise serializers.ValidationError({"run_cmd": "工具执行命令必填"})
         # 存在时则进行凭证校验
         if scm_url and scm_auth:
-            scm_type = attrs.get("scm_type")
-            auth_type = scm_auth.get("auth_type")
-            if auth_type == models.ScmAuth.ScmAuthTypeEnum.OAUTH:
-                scm_oauth = scm_auth.get("scm_oauth")
-                if not scm_oauth or scm_oauth.user != user:
-                    raise serializers.ValidationError({"scm_auth": "请选择有效OAuth凭证"})
-                credential_info = scm_oauth.credential_info
-            elif auth_type == models.ScmAuth.ScmAuthTypeEnum.PASSWORD:
-                scm_account = scm_auth.get("scm_account")
-                if not scm_account or scm_account.user != user:
-                    raise serializers.ValidationError({"scm_auth": "请选择有效HTTP凭证"})
-                credential_info = scm_account.credential_info
-            elif auth_type == models.ScmAuth.ScmAuthTypeEnum.SSHTOKEN:
-                scm_ssh = scm_auth.get("scm_ssh")
-                if not scm_ssh or scm_ssh.user != user:
-                    raise serializers.ValidationError({"scm_auth": "请选择有效SSH凭证"})
-                credential_info = scm_ssh.credential_info
-            else:
-                raise serializers.ValidationError({"auth_type": ["不支持%s鉴权方式" % auth_type]})
-            # 校验
+            # 校验凭证有效性
+            credential_info = CommonManager.get_and_check_scm_auth(scm_auth, user, instance=self.instance)
             ScmAuthManager.check_scm_url_credential(scm_type, scm_url, credential_info)
         return super().validate(attrs)
 
@@ -129,6 +112,10 @@ def _create_or_update(self, validated_data, instance=None):
                 scm_ssh_info=scm_auth.get("scm_ssh"),
                 scm_oauth=scm_auth.get("scm_oauth"),
             )
+        elif instance:
+            # 更新时允许移除scm_auth
+            checktool.scm_auth = None
+            checktool.save(user=user)
         return checktool
 
     def create(self, validated_data):

server/projects/main/apps/scan_conf/utils/base.py

@@ -68,7 +68,7 @@ def loadlib(cls, toollib_json, user):
             scm_auth = slz.validated_data.pop("scm_auth", None)
             toollib_name = slz.validated_data.pop("name")
             toollib, _ = ToolLibManager.create_or_update(toollib_name, user, instance=toollib,
-                                                          lib_key=lib_key, **slz.validated_data)
+                                                        lib_key=lib_key, **slz.validated_data)
             return True, cls.postload_handler(toollib, scm_auth=scm_auth, user=user)
         except Exception as e:
             logger.error(e)
@@ -122,7 +122,7 @@ def postload_handler(cls, checktool, admins=None):
         :return: checktool_name
         """
         if admins and not checktool.owners.exists():
-            # 如果工具owners为农历了
+            # 如果工具owners不存在
             checktool.owners.add(*admins)
         return checktool.name
 
@@ -136,15 +136,17 @@ def loadchecker(cls, checktool_json, admins=None):
         checktool_name = checktool_json["name"]
         try:
             data = cls.preload_handler(checktool_json)
-            checktool = models.CheckTool.objects.filter(name=checktool_name).first()
-            slz = CheckerSerializer(instance=checktool, data=data, context={"is_local_script": True})
-            slz.is_valid(raise_exception=True)
-            logger.info("开始保存工具%s数据。。。" % checktool_name)
-            checktool_name = slz.validated_data.pop("name")
-            tool_key = data.get('tool_key')
-            checktool = CheckToolManager.load_by_script(checktool_name, None, checktool=checktool,
-                                              tool_key=tool_key, **slz.validated_data)
-            return True, cls.postload_handler(checktool, admins=admins)
+            if data:
+                checktool = models.CheckTool.objects.filter(name=checktool_name).first()
+                slz = CheckerSerializer(instance=checktool, data=data, context={"is_local_script": True})
+                slz.is_valid(raise_exception=True)
+                logger.info("开始保存工具%s数据。。。" % checktool_name)
+                checktool_name = slz.validated_data.pop("name")
+                tool_key = data.get('tool_key')
+                checktool = CheckToolManager.load_by_script(checktool_name, None, checktool=checktool,
+                                                tool_key=tool_key, **slz.validated_data)
+                return True, cls.postload_handler(checktool, admins=admins)
+            return True, checktool_name
         except Exception as e:
             logger.error(e)
             return False, checktool_name
@@ -221,14 +223,16 @@ def loadpkg(cls, checkpackage_json):
         checkpackage_name = checkpackage_json["name"]
         try:
             data = cls.preload_handler(checkpackage_json)
-            checkpackage = models.CheckPackage.objects.filter(name=checkpackage_name).first()
-            slz = CheckPackageJsonSerializer(instance=checkpackage, data=data)
-            slz.is_valid(raise_exception=True)
-            logger.info("开始保存规则包[%s]数据。。。" % checkpackage_name)
-            checkpackage_name = slz.validated_data.pop("name")
-            checkpackage = CheckPackageManager.load_by_script(checkpackage_name, None,
-                                              checkpackage=checkpackage, **slz.validated_data)
-            return True, cls.postload_handler(checkpackage)
+            if data:
+                checkpackage = models.CheckPackage.objects.filter(name=checkpackage_name).first()
+                slz = CheckPackageJsonSerializer(instance=checkpackage, data=data)
+                slz.is_valid(raise_exception=True)
+                logger.info("开始保存规则包[%s]数据。。。" % checkpackage_name)
+                checkpackage_name = slz.validated_data.pop("name")
+                checkpackage = CheckPackageManager.load_by_script(checkpackage_name, None,
+                                                checkpackage=checkpackage, **slz.validated_data)
+                return True, cls.postload_handler(checkpackage)
+            return True, checkpackage_name
         except Exception as e:
             logger.error(e)
             return False, checkpackage_name
@@ -252,7 +256,7 @@ def loadpkg_by_workers(cls, checkpackage_json_list, workers=10):
                         break
                     checkpackage_json = checkpackage_json_list[i_idx+j_idx]
                     logger.info('--> [%s/%s], checkpackage name: %s' % (
-                        index+1, checkpackage_count, checkpackage_json["name"]))
+                        checkpackage_count, index+1, checkpackage_json["name"]))
                     task = t_thread.submit(cls.loadpkg, checkpackage_json)
                     all_task.append(task)
                 for future in as_completed(all_task):