security: restrict dynamic script execution in backend plugin handlers

cmdb-api/api/lib/cmdb/attribute.py

@@ -18,6 +18,8 @@
 from api.lib.cmdb.const import ValueTypeEnum
 from api.lib.cmdb.history import CITypeHistoryManager
 from api.lib.cmdb.resp_format import ErrFormat
+from api.lib.cmdb.safe_script import UnsafeScriptError
+from api.lib.cmdb.safe_script import load_class_from_script
 from api.lib.cmdb.utils import ValueTypeMap
 from api.lib.decorator import kwargs_required
 from api.lib.perm.acl.acl import is_app_admin
@@ -80,11 +82,12 @@ def _get_choice_values_from_other(choice_other):
 
         elif choice_other.get('script'):
             try:
-                x = compile(choice_other['script'], '', "exec")
-                local_ns = {}
-                exec(x, {}, local_ns)
-                res = local_ns['ChoiceValue']().values() or []
+                choice_cls = load_class_from_script(choice_other['script'], 'ChoiceValue')
+                res = choice_cls().values() or []
                 return [[i, {}] for i in res]
+            except UnsafeScriptError as e:
+                current_app.logger.error("get choice values from script: {}".format(e))
+                return []
             except Exception as e:
                 current_app.logger.error("get choice values from script: {}".format(e))
                 return []

cmdb-api/api/lib/cmdb/auto_discovery/auto_discovery.py

@@ -26,6 +26,8 @@
 from api.lib.cmdb.const import ResourceTypeEnum
 from api.lib.cmdb.custom_dashboard import SystemConfigManager
 from api.lib.cmdb.resp_format import ErrFormat
+from api.lib.cmdb.safe_script import UnsafeScriptError
+from api.lib.cmdb.safe_script import load_class_from_script
 from api.lib.cmdb.search import SearchError
 from api.lib.cmdb.search.ci import search as ci_search
 from api.lib.common_setting.role_perm_base import CMDBApp
@@ -52,11 +54,12 @@
 def parse_plugin_script(script):
     attributes = []
     try:
-        x = compile(script, '', "exec")
-        local_ns = {}
-        exec(x, {}, local_ns)
-        unique_key = local_ns['AutoDiscovery']().unique_key
-        attrs = local_ns['AutoDiscovery']().attributes() or []
+        plugin_cls = load_class_from_script(script, 'AutoDiscovery')
+        plugin = plugin_cls()
+        unique_key = plugin.unique_key
+        attrs = plugin.attributes() or []
+    except UnsafeScriptError as e:
+        return abort(400, str(e))
     except Exception as e:
         return abort(400, str(e))
 

cmdb-api/api/lib/cmdb/safe_script.py

@@ -0,0 +1,118 @@
+# -*- coding:utf-8 -*-
+
+import ast
+import builtins
+
+
+class UnsafeScriptError(Exception):
+    pass
+
+
+class _RestrictedScriptChecker(ast.NodeVisitor):
+    FORBIDDEN_NODES = (
+        ast.Import,
+        ast.ImportFrom,
+        ast.Global,
+        ast.Nonlocal,
+        ast.AsyncFunctionDef,
+        ast.Await,
+        ast.Yield,
+        ast.YieldFrom,
+        ast.Lambda,
+        ast.With,
+        ast.AsyncWith,
+        ast.Delete,
+    )
+    FORBIDDEN_NAMES = {
+        "__import__",
+        "eval",
+        "exec",
+        "open",
+        "compile",
+        "input",
+        "globals",
+        "locals",
+        "vars",
+        "dir",
+        "getattr",
+        "setattr",
+        "delattr",
+        "help",
+        "breakpoint",
+    }
+
+    def visit(self, node):
+        if isinstance(node, self.FORBIDDEN_NODES):
+            raise UnsafeScriptError("forbidden syntax: {0}".format(node.__class__.__name__))
+        return super(_RestrictedScriptChecker, self).visit(node)
+
+    def visit_Name(self, node):
+        if node.id.startswith("__") or node.id in self.FORBIDDEN_NAMES:
+            raise UnsafeScriptError("forbidden name: {0}".format(node.id))
+        return self.generic_visit(node)
+
+    def visit_Attribute(self, node):
+        if node.attr.startswith("__"):
+            raise UnsafeScriptError("forbidden attribute access: {0}".format(node.attr))
+        return self.generic_visit(node)
+
+    def visit_Call(self, node):
+        if isinstance(node.func, ast.Name):
+            if node.func.id.startswith("__") or node.func.id in self.FORBIDDEN_NAMES:
+                raise UnsafeScriptError("forbidden function call: {0}".format(node.func.id))
+        elif isinstance(node.func, ast.Attribute):
+            if node.func.attr.startswith("__"):
+                raise UnsafeScriptError("forbidden function call: {0}".format(node.func.attr))
+        return self.generic_visit(node)
+
+
+_ALLOWED_BUILTINS = {
+    "__build_class__": builtins.__build_class__,
+    "object": object,
+    "Exception": Exception,
+    "str": str,
+    "int": int,
+    "float": float,
+    "bool": bool,
+    "dict": dict,
+    "list": list,
+    "set": set,
+    "tuple": tuple,
+    "len": len,
+    "range": range,
+    "enumerate": enumerate,
+    "zip": zip,
+    "min": min,
+    "max": max,
+    "sum": sum,
+    "abs": abs,
+    "sorted": sorted,
+    "all": all,
+    "any": any,
+}
+
+
+def load_class_from_script(script, class_name):
+    if not isinstance(script, str):
+        raise UnsafeScriptError("script must be a string")
+
+    try:
+        tree = ast.parse(script, mode='exec')
+    except Exception as e:
+        raise UnsafeScriptError("invalid script: {0}".format(e))
+
+    _RestrictedScriptChecker().visit(tree)
+    code = compile(tree, '<cmdb-safe-script>', "exec")
+
+    local_ns = {}
+    global_ns = {
+        "__builtins__": _ALLOWED_BUILTINS,
+        "__name__": "__cmdb_safe_script__",
+    }
+    exec(code, global_ns, local_ns)
+
+    klass = local_ns.get(class_name) or global_ns.get(class_name)
+    if klass is None:
+        raise UnsafeScriptError("class {0} is required".format(class_name))
+
+    return klass