approve: 修复当前迭代 1 个功能 Bug（元景项目-初始迭代）

script_engine/engine.py

@@ -3,7 +3,15 @@
 
 Restricted script execution for script_type 0=Python / 1=SQL.
 Python runs in a whitelisted-builtins namespace; imports, classes, lambdas,
-async/await, generators and object method calls are forbidden.
+async/await, generators and unsafe attribute access are forbidden.
+
+Method calls (ast.Attribute) are allowed ONLY when the method name is on the
+ALLOWED_METHODS whitelist and is not a dunder (``__x__``) method.  This keeps
+the natural script pattern ``params.get('key')`` working (Bug
+kGZMpApWK3I1ed4OHpKQ0) while still blocking every other method call and any
+``__``-prefixed attribute -- sandbox-escape attempts such as
+``().__class__.__mro__`` / ``x.__globals__`` are rejected.
+
 SQL allows read-only statements only (SELECT/SHOW/DESCRIBE/EXPLAIN).
 """
 import ast
@@ -17,11 +25,35 @@ ALLOWED_BUILTINS = {
     'tuple', 'type', 'zip',
 }
 
+# whitelisted object methods.  The receiver is always an object that already
+# lives in the script namespace (a dict/list/str/int/float injected as
+# ``params`` or created by the script itself); there is no way to reach
+# os/subprocess/import machinery because dunder access is blocked and the
+# builtins namespace is restricted to ALLOWED_BUILTINS.  Every method NOT in
+# this list is rejected (100% coverage enforced in validate_python).
+ALLOWED_METHODS = {
+    # dict / mapping
+    'get', 'keys', 'values', 'items', 'pop', 'setdefault', 'update',
+    'clear', 'copy',
+    # str (pure transformations / queries)
+    'strip', 'lstrip', 'rstrip', 'lower', 'upper', 'title', 'capitalize',
+    'replace', 'split', 'rsplit', 'splitlines', 'join', 'format',
+    'format_map', 'startswith', 'endswith', 'find', 'rfind', 'index',
+    'rindex', 'count', 'isdigit', 'isalpha', 'isalnum', 'isspace',
+    'isupper', 'islower', 'istitle', 'encode', 'decode', 'zfill',
+    'center', 'ljust', 'rjust',
+    # list
+    'append', 'extend', 'insert', 'remove', 'pop', 'sort', 'reverse',
+    'index', 'count', 'copy', 'clear',
+    # int / float
+    'bit_length', 'is_integer',
+}
+
 # ast node types that are never allowed in a script
 FORBIDDEN_NODES = (
     ast.Import, ast.ImportFrom, ast.ClassDef, ast.Lambda,
     ast.AsyncFunctionDef, ast.AsyncFor, ast.AsyncWith, ast.Await,
-    ast.Yield, ast.YieldFrom, ast.Global,
+    ast.Yield, ast.YieldFrom, ast.Global, ast.Nonlocal,
 )
 
 # sql keywords that indicate write / dangerous statements
@@ -36,7 +68,25 @@ SQL_READONLY_RE = re.compile(r'^\s*(select|show|describe|desc|explain)\b', re.IG
 def validate_python(content):
     """Validate python script syntax and forbidden constructs.
 
-    Returns {'code': 0, 'message': 'ok'} on success else {'code': 1, 'message': ...}.
+    Safety rules (in order of evaluation):
+      1. forbidden AST node kinds: import/import-from/class/lambda/async/
+         await/yield/global/nonlocal
+      2. dunder (``__x__``) attribute access is ALWAYS rejected -- this is
+         the sandbox-escape guard (``().__class__``, ``x.__globals__``, ...)
+      3. calls:
+         - bare name -> the name must be in ALLOWED_BUILTINS
+         - object method (ast.Attribute) -> the method name must be in
+           ALLOWED_METHODS (and, per rule 2, never dunder); anything else is
+           rejected (100% interception of non-whitelisted methods)
+         - any other call expression (subscript call, call result call, ...)
+           is rejected
+      4. attribute assignment / deletion (``obj.attr = ...``,
+         ``del obj.attr``) is rejected -- scripts must not mutate object
+         attributes.  Subscript assignment on plain dicts/lists stays allowed
+         because it only touches objects already in the script namespace.
+
+    Returns {'code': 0, 'message': 'ok'} on success else
+    {'code': 1, 'message': ...}.
     """
     if not content or not content.strip():
         return {'code': 1, 'message': 'content is required'}
@@ -44,15 +94,42 @@ def validate_python(content):
         tree = ast.parse(content, mode='exec')
     except SyntaxError as e:
         return {'code': 1, 'message': 'syntax error: %s' % e}
+
+    # rule 1 + 2: forbidden node kinds and dunder attribute access
     for node in ast.walk(tree):
         if isinstance(node, FORBIDDEN_NODES):
             return {'code': 1, 'message': 'forbidden construct: %s' % type(node).__name__}
-        if isinstance(node, ast.Call):
-            func = node.func
-            if isinstance(func, ast.Attribute):
-                return {'code': 1, 'message': 'object method call is forbidden'}
-            if isinstance(func, ast.Name) and func.id not in ALLOWED_BUILTINS:
+        if isinstance(node, ast.Attribute) and node.attr.startswith('__'):
+            return {'code': 1, 'message': 'forbidden attribute access: %s' % node.attr}
+
+    # rule 3: calls
+    for node in ast.walk(tree):
+        if not isinstance(node, ast.Call):
+            continue
+        func = node.func
+        if isinstance(func, ast.Name):
+            if func.id not in ALLOWED_BUILTINS:
                 return {'code': 1, 'message': 'forbidden call: %s' % func.id}
+        elif isinstance(func, ast.Attribute):
+            if func.attr not in ALLOWED_METHODS:
+                return {'code': 1, 'message': 'forbidden method call: %s' % func.attr}
+        else:
+            return {'code': 1, 'message': 'forbidden call expression'}
+
+    # rule 4: attribute assignment / deletion
+    for node in ast.walk(tree):
+        targets = []
+        if isinstance(node, ast.Assign):
+            targets = node.targets
+        elif isinstance(node, (ast.AnnAssign, ast.AugAssign)):
+            targets = [node.target]
+        elif isinstance(node, ast.Delete):
+            targets = node.targets
+        for t in targets:
+            for sub in ast.walk(t):
+                if isinstance(sub, ast.Attribute):
+                    return {'code': 1, 'message': 'attribute assignment is forbidden'}
+
     return {'code': 0, 'message': 'ok'}
 
 
@@ -60,11 +137,14 @@ def execute_python(content, input_ns=None):
     """Execute python script in a restricted namespace.
 
     The script may assign a final variable named `result` which is returned.
+    ``input_ns`` (e.g. {'params': {...}}) is injected into the namespace so
+    scripts can read business inputs with ``params.get('key')``.
     """
+    blt = __builtins__ if isinstance(__builtins__, dict) else vars(__builtins__)
     builtins_ns = {}
     for k in ALLOWED_BUILTINS:
-        if k in __builtins__:
-            builtins_ns[k] = __builtins__[k]
+        if k in blt:
+            builtins_ns[k] = blt[k]
     namespace = {'__builtins__': builtins_ns}
     if input_ns:
         for k, v in input_ns.items():

script_engine/init.py

@@ -218,7 +218,12 @@ async def validate_script_api(request, ns):
 
 
 async def execute_script(request, ns):
-    """Execute a script by id OR by inline content+script_type."""
+    """Execute a script by id OR by inline content+script_type.
+
+    Python scripts receive the business params (everything except control
+    fields) as a ``params`` dict in the namespace, so scripts can read input
+    via ``params.get('key')``.
+    """
     data = _clean_ns(ns)
     dbname = _dbname()
     sid = (data.get('id') or '').strip()
@@ -246,7 +251,9 @@ async def execute_script(request, ns):
         return _err(1, v['message'], 'content')
     if stype == '0':
         try:
-            out = engine.execute_python(content)
+            ctrl = {'id', 'content', 'script_type', 'page', 'rows', 'keyword'}
+            params = {k: v for k, v in data.items() if k not in ctrl}
+            out = engine.execute_python(content, input_ns={'params': params})
         except Exception as e:
             exception('script_engine.execute_script python error: %s' % e)
             return _err(1, 'execute failed: %s' % e, 'content')

script_engine/tests/test_validate_python.py

@@ -0,0 +1,113 @@
+# -*- coding: utf-8 -*-
+"""Regression tests for script_engine.validate_python (Bug kGZMpApWK3I1ed4OHpKQ0).
+
+Validates that:
+  * params.get() and other whitelisted method calls now PASS validation
+  * every non-whitelisted method call is rejected (100% interception)
+  * dunder attribute access (sandbox escapes) is rejected
+  * attribute assignment / deletion is rejected
+  * forbidden constructs (import/class/lambda/async/...) are still rejected
+
+Run (from modules/script_engine):
+    python script_engine/tests/test_validate_python.py
+  or:
+    python -m pytest script_engine/tests/test_validate_python.py -v
+
+The engine module is loaded by file path so this test does not require the
+ahserver/sqlor runtime environment.
+"""
+import importlib.util
+import os
+
+_HERE = os.path.dirname(os.path.abspath(__file__))
+_ENGINE_PATH = os.path.join(os.path.dirname(_HERE), 'engine.py')
+_spec = importlib.util.spec_from_file_location('_se_engine', _ENGINE_PATH)
+_se_engine = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(_se_engine)
+
+validate_python = _se_engine.validate_python
+execute_python = _se_engine.execute_python
+
+_PASS = 0
+_FAIL = 0
+
+
+def check(content, expect_ok, why):
+    """Assert validate_python(content) == expect_ok and print the outcome."""
+    global _PASS, _FAIL
+    r = validate_python(content)
+    ok = r.get('code') == 0
+    status = 'PASS' if ok == expect_ok else 'FAIL'
+    if ok == expect_ok:
+        _PASS += 1
+    else:
+        _FAIL += 1
+    print('%-4s expect_ok=%-5s %-52s -> %s' % (
+        status, expect_ok, why, r.get('message')))
+
+
+def test_positive():
+    # the reported bug: params.get() must pass validation
+    check("result = params.get('age', 0) + 1", True, 'params.get')
+    check("result = params.get('name', '').strip().upper()", True, 'str chain on params.get')
+    check("result = sum(params.get('nums', []))", True, 'sum + params.get')
+    check("r = 0\nfor i in range(5):\n    r += i\nresult = r", True, 'loop + range')
+    check("d = {'a': 1}\nd['b'] = 2\nresult = d.get('b')", True, 'dict subscript assign')
+    check("result = [x * 2 for x in range(3)]", True, 'list comprehension')
+    check("result = params.get('a', '').replace('x', 'y')", True, 'str.replace whitelisted')
+    check("result = params.get('l', []).append(1) or params.get('l')", True, 'list.append whitelisted')
+
+
+def test_negative():
+    # forbidden constructs (must STILL be rejected)
+    check("import os\nresult = 1", False, 'import')
+    check("from os import system\nresult = 1", False, 'from-import')
+    check("class X:\n    pass", False, 'class def')
+    check("f = lambda x: x", False, 'lambda')
+    check("async def f():\n    pass", False, 'async def')
+    check("def g():\n    yield 1", False, 'yield')
+    # non-whitelisted builtin calls
+    check("result = open('/etc/passwd')", False, 'open not whitelisted')
+    check("result = os.system('id')", False, 'os.system (name call)')
+    check("result = eval('1+1')", False, 'eval not whitelisted')
+    check("result = exec('x=1')", False, 'exec not whitelisted')
+    check("result = __import__('os')", False, '__import__ not whitelisted')
+    # dunder attribute access (sandbox escapes)
+    check("result = params.__class__", False, 'dunder read __class__')
+    check("result = ().__class__.__mro__", False, 'dunder escape chain')
+    check("result = params.get.__self__", False, 'dunder __self__')
+    check("result = (lambda: 1).__globals__", False, 'dunder __globals__')
+    # non-whitelisted object methods (100% interception)
+    check("result = params.popitem()", False, 'popitem not whitelisted')
+    check("result = params.foo()", False, 'arbitrary method foo')
+    check("result = params.get('a').isidentifier()", False, 'isidentifier not whitelisted')
+    check("result = [].index.__call__()", False, '__call__ dunder')
+    # attribute assignment / deletion
+    check("obj = params\nobj.x = 1", False, 'attribute assign')
+    check("obj = params\nobj.x += 1", False, 'attribute aug-assign')
+    check("obj = params\ndel obj.x", False, 'attribute delete')
+    # other call expressions
+    check("result = params['__class__']()", False, 'subscript call expression')
+
+
+def test_execute():
+    out = execute_python("result = params.get('a', 0) + 10", {'params': {'a': 5}})
+    assert out == 15, 'execute_python params.get -> %r' % out
+    print('PASS execute_python params.get({a:5})+10 ->', out)
+    global _PASS
+    _PASS += 1
+
+
+def main():
+    test_positive()
+    test_negative()
+    test_execute()
+    print('-' * 78)
+    print('TOTAL: %d passed, %d failed' % (_PASS, _FAIL))
+    if _FAIL:
+        raise SystemExit(1)
+    print('ALL TESTS PASSED')
+
+
+if __name__ == '__main__':
+    main()