Commit c0bdbc92 authored by Data Governance Dev's avatar Data Governance Dev

feat(web3): 沙箱 import 白名单扩到常用纯计算类库(json/unicodedata/collections/dataclass 等)

用户反馈:AI 生成 JSON 校验函数,运行时报 ImportError: 'json' is not allowed。
原白名单只有 6 个模块(datetime/time/math/decimal/re/_strptime),太窄。

设计准则:「只能改内存里的 Python 对象,不能碰进程外的东西」——
纯计算 / 数据结构类库一律放;I/O、网络、进程、动态执行类一律不放。

白名单(31 个):
  数据 / 时间  : datetime, time
  数学 / 数值  : math, decimal, fractions, statistics, numbers, operator
  字符串 / 编码: re, string, unicodedata, html
  数据结构     : collections, collections.abc, itertools, functools
  序列化 / 二进制: json, csv, struct, array
  编码 / 哈希  : hashlib, base64, binascii
  实用工具     : uuid, enum, dataclasses, typing, difflib
  其他         : copy, pprint, reprlib, ipaddress

故意不放:os/sys/pathlib/subprocess/pickle/ctypes/importlib/socket/...
连 urllib.parse 也不放(import urllib 后 urllib.request 属性访问能拿到,副作用太大)。

沙箱三件套(踩坑):
1. builtin 白名单加 __build_class__ —— class X: / enum.Enum / @dataclasses.dataclass 都靠它;
   动态 type('X', (), {}) 走 type.__call__ 不经过它,所以风险 ≈ 0
2. namespace 预填 __name__ / __qualname__ —— enum/dataclass metaclass 会查
3. __name__ 必须用 '__main__' 不能用假名 —— dataclasses._is_type 内部
   sys.modules.get(cls.__module__).__dict__,假名返 None 然后 .__dict__ 报错

另:
- ai_rule.py 的 number / date / string prompt 都加上「沙箱已预注入的常用模块」清单
- RuleModal.vue 的 string rule 提示区也列已预注入模块清单
- test_rule_runner.py 新增 TestSandboxImports 类,46 个用例
  (26 个模块 smoke + 3 个 import 形式 + 16 个危险模块被拦 + 1 个 urllib 被拦)

验证:
  python -m pytest web3/tests/  →  123 passed, 3 warnings
  npx vite build                →  ✓ built in 6.84s
parent 75f471e3
......@@ -49,7 +49,12 @@ _NUMBER_USER_TEMPLATE_INITIAL = """请把以下规则描述转换成一个 Pytho
- value 是从数据库读出来的原始值(str / int / float 都可能,按需转换)
- 返回 True 表示合规,False 表示不合规
2. 函数要短小(一般不超过 10 行),只做这一项校验
3. 不要 import,直接用 value 即可;如果需要数学常量(如 math.pi)才用 math 模块
3. 不要 import,直接用 value 即可;沙箱已预注入以下常用模块,需要时直接用:
- math(math.pi / math.isnan / math.isinf / ...)
- decimal(精确十进制)
- statistics(mean / median / stdev 等统计量)
- fractions(有理数)
- re(正则)
4. 如果描述里包含具体范围/阈值(如"0 到 150"、"大于 0"),一定要用上
5. 如果 value 不是合法数字,按不合规返回 False(用 try/except 包住转换)
......@@ -87,7 +92,8 @@ _DATE_USER_TEMPLATE_INITIAL = """请把以下规则描述转换成一个 Python
- 转回字符串:value.strftime('%Y-%m-%d')
3. 跟「今天」「某个日期」比较时:用 datetime.date.today() / datetime.datetime(2020, 1, 1) 之类
4. 函数要短小(一般不超过 10 行),只做这一项校验
5. 不要写 import,datetime 模块已预注入;不要调 strptime —— value 已经是 datetime 对象
5. 不要写 import,datetime / time / re / collections / itertools 等常用模块已预注入;
不要再调 strptime —— value 已经是 datetime 对象
6. 如果描述里包含具体范围(如"晚于 2020-01-01"、"小于今天"),一定要用上
只输出 def check(value) -> bool: 函数体的 Python 代码,不要任何解释、不要 Markdown 代码块。"""
......@@ -128,9 +134,16 @@ _STRING_USER_TEMPLATE_INITIAL = """请把以下规则描述转换成一个 Pytho
- 前缀/后缀/包含:value.startswith(...) / value.endswith(...) / ... in value
- 大小写:value.upper() / value.lower()
- 正则(如果 regex 不好写):用 re 模块(已预注入,直接 re.search / re.match 即可)
- 中文/字符集判断:遍历字符 ord(c) > 127
4. 如果 value 不是字符串或为空,先 str(value) 转一下;非空才校验
5. 不要 import,需要正则就用 re 模块
- 中文/字符集判断:遍历字符 ord(c) > 127;或用 unicodedata.category(c)
- 字符频次:collections.Counter(value)
- JSON 格式:json.loads(value) 试解析
4. 沙箱已预注入以下常用模块,需要时直接用,不需要 import:
- re / string / unicodedata / html
- collections(Counter / defaultdict / OrderedDict)/ itertools / functools
- json / csv(解析)/ hashlib / uuid / difflib(SequenceMatcher 模糊匹配)
- ipaddress(IP 字符串校验)/ enum / dataclasses
5. 如果 value 不是字符串或为空,先 str(value) 转一下;非空才校验
6. 不要 import,需要正则就用 re 模块
只输出 def check(value) -> bool: 函数体的 Python 代码,不要任何解释、不要 Markdown 代码块。"""
......
......@@ -23,14 +23,35 @@ import builtins as _bi
import datetime
import decimal
import math
import re as _re
import time as _time
from typing import Any, Optional
# 沙箱允许的 import 模块白名单(覆盖 __import__ 实际调用 + `import x` 语句)
# 主要为了让 datetime.datetime.strptime 这种「内部 lazy-import」能跑
# 2026-08-21:扩到「常用纯计算 / 数据结构类库」以方便 AI 生成校验函数(json 校验、unicode
# 字符分类、字符频次分析 等)。**严格排除 I/O、网络、进程、动态执行类模块**——
# 准则:「只能改内存里的 Python 对象,不能碰进程外的东西」。
# - 排除:os / sys / pathlib / shutil / socket / urllib.request / subprocess /
# threading / multiprocessing / pickle / ctypes / importlib / inspect / ...
# - 例外:urllib.parse 不放(import urllib 后 urllib.request 也能通过属性访问拿到,副作用太大)
# - 例外:open 已不在 _SAFE_BUILTIN_NAMES(连文件读都不允许)
_SAFE_IMPORT_NAMES = frozenset({
"datetime", "time", "_strptime", "math", "decimal", "re",
# 数据 / 时间
"datetime", "time", "_strptime",
# 数学 / 数值
"math", "decimal", "fractions", "statistics", "numbers", "operator",
# 字符串 / 编码
"re", "string", "unicodedata", "html",
# 数据结构
"collections", "collections.abc", "itertools", "functools",
# 序列化 / 二进制
"json", "csv", "struct", "array",
# 编码 / 哈希
"hashlib", "base64", "binascii",
# 实用工具
"uuid", "difflib", "enum", "dataclasses", "typing",
# 其他纯计算
"copy", "pprint", "reprlib", "ipaddress",
})
......@@ -57,7 +78,10 @@ def _safe_import(name, globals=None, locals=None, fromlist=(), level=0):
# 故意不放:open / eval / exec / compile / globals / locals /
# vars / getattr / setattr / delattr / breakpoint / input /
# memoryview / exit / quit / help / object / super / property /
# __build_class__(防 type(name, (), {}) 动态建类)
# __build_class__ 已放(2026-08-21):class X: / @dataclasses.dataclass /
# enum.Enum 子类 都靠它。风险评估:__build_class__ 是 `class X:` 语法
# 用的;动态建类走 `type('X', (), {})`(不经过 __build_class__),
# 也不危险 —— 即使能建类,新类也在沙箱 namespace 里跑不出去。
# __import__ 用 _safe_import 替换(白名单版)
_SAFE_BUILTIN_NAMES = {
# 类型 / 构造
......@@ -78,6 +102,9 @@ _SAFE_BUILTIN_NAMES = {
'NotImplemented', 'Ellipsis',
# 字面常量
'True', 'False', 'None',
# 类定义(2026-08-21:加进来,让 class X: / @dataclasses.dataclass / enum.Enum 能用;
# 风险评估见上面注释)
'__build_class__',
}
_SAFE_BUILTINS: dict[str, Any] = {
name: getattr(_bi, name) for name in _SAFE_BUILTIN_NAMES
......@@ -85,13 +112,79 @@ _SAFE_BUILTINS: dict[str, Any] = {
# 用白名单版 __import__ 替换掉真 __import__
_SAFE_BUILTINS['__import__'] = _safe_import
# 预注入到 exec 命名空间的安全模块(用户代码可以直接用 datetime 等,不需要 import)
# 预注入到 exec 命名空间的安全模块(用户代码可以直接用 json 等,不需要 import)
# 与 _SAFE_IMPORT_NAMES 同源 —— 凡是「对用户代码有用」的都在这里也挂一份
# 例外:_strptime 是 datetime 内部实现细节,不暴露给用户
import re as _re
import string as _string
import unicodedata as _unicodedata
import html as _html
import collections as _collections
import collections.abc as _collections_abc
import itertools as _itertools
import functools as _functools
import statistics as _statistics
import fractions as _fractions
import numbers as _numbers
import operator as _operator
import json as _json
import csv as _csv
import struct as _struct
import array as _array
import hashlib as _hashlib
import base64 as _base64
import binascii as _binascii
import uuid as _uuid
import difflib as _difflib
import enum as _enum
import dataclasses as _dataclasses
import typing as _typing
import copy as _copy
import pprint as _pprint
import reprlib as _reprlib
import ipaddress as _ipaddress
_PRELOAD_MODULES: dict[str, Any] = {
"datetime": datetime,
"time": _time,
"math": math,
"decimal": decimal,
"re": _re,
# 数据 / 时间
"datetime": datetime,
"time": _time,
# 数学 / 数值
"math": math,
"decimal": decimal,
"fractions": _fractions,
"statistics": _statistics,
"numbers": _numbers,
"operator": _operator,
# 字符串 / 编码
"re": _re,
"string": _string,
"unicodedata": _unicodedata,
"html": _html,
# 数据结构
"collections": _collections,
"collections.abc": _collections_abc,
"itertools": _itertools,
"functools": _functools,
# 序列化 / 二进制
"json": _json,
"csv": _csv,
"struct": _struct,
"array": _array,
# 编码 / 哈希
"hashlib": _hashlib,
"base64": _base64,
"binascii": _binascii,
# 实用工具
"uuid": _uuid,
"difflib": _difflib,
"enum": _enum,
"dataclasses": _dataclasses,
"typing": _typing,
# 其他
"copy": _copy,
"pprint": _pprint,
"reprlib": _reprlib,
"ipaddress": _ipaddress,
}
......@@ -109,6 +202,14 @@ def _exec_user_function(code: str, value: Any) -> bool:
"""
namespace: dict[str, Any] = {
"__builtins__": _SAFE_BUILTINS,
# 2026-08-21:enum.Enum / dataclasses.dataclass 的 metaclass 会查 __name__/__qualname__,
# 真 `class X:` 语句由 __build_class__ 自动注入;AI 生成的代码偶尔会写
# `@dataclasses.dataclass class P: ...` 这种也靠这个。
# 用 "__main__" 而不是 "<user_rule>":dataclasses._is_type 内部会
# `sys.modules.get(cls.__module__).__dict__`,__module__ 从 __name__ 推导,
# 必须指向 sys.modules 里真实存在的条目。__main__ 是 Python 启动时一定在 sys.modules 里的。
"__name__": "__main__",
"__qualname__": "user_rule",
}
namespace.update(_PRELOAD_MODULES) # 预注入模块
......
......@@ -88,6 +88,14 @@
常用操作:<code>len(value)</code> / <code>value.startswith(...)</code> /
<code>value.endswith(...)</code> / <code>... in value</code>;
需要正则时直接用 <code>re</code> 模块(已预注入)。
<br />
<strong>已预注入的常用模块</strong>(不需要 import,直接用):
<code>re</code> / <code>string</code>(字符集常量)/ <code>unicodedata</code>(字符分类)/
<code>html</code> / <code>collections</code>(Counter / defaultdict)/
<code>itertools</code> / <code>functools</code> /
<code>json</code>(JSON 解析)/ <code>csv</code> /
<code>hashlib</code> / <code>uuid</code> / <code>difflib</code>(SequenceMatcher 模糊匹配)/
<code>ipaddress</code>(IP 字符串校验)/ <code>enum</code> / <code>dataclasses</code>。
</span>
<span v-else>
需要数学常量时用 math 模块;不需要 import。
......
......@@ -351,4 +351,144 @@ def test_rule_types_constant():
# ── 未知类型 ────────────────────────────────────────────
def test_unknown_rule_type_raises():
with pytest.raises(RuleRunError, match="未知的规则类型"):
run_rule("weird", None, None, "x")
\ No newline at end of file
run_rule("weird", None, None, "x")
# ── 扩展沙箱 import 白名单(2026-08-21)────────────────
# 目标:让 AI 生成的校验函数能用 json / unicode 分类 / 字符频次 等实用工具,
# 同时锁死 I/O / 网络 / 进程类模块。
class TestSandboxImports:
"""每个被允许的模块各跑一条 smoke test,确认既已 preload(无需 import)
又能通过 import 语句引入。"""
@pytest.mark.parametrize("code, value, expected", [
# json:解析合法 JSON 字符串
("def check(value):\n obj = json.loads(value)\n return isinstance(obj, dict)",
'{"a": 1}', True),
# csv:解析 CSV 行
("def check(value):\n rows = list(csv.reader(value.splitlines()))\n return len(rows) == 2",
"a,b\n1,2", True),
# string:digits 常量
("def check(value):\n return all(c in string.digits for c in value)",
"12345", True),
# collections:Counter('abba' 有 2 个 a / 2 个 b)
("def check(value):\n c = collections.Counter(value)\n return c['a'] == 2",
"abba", True),
# collections.abc:Iterable 判断
("def check(value):\n return isinstance(value, collections.abc.Iterable)",
[1, 2, 3], True),
# itertools:chain
("def check(value):\n return list(itertools.chain(value, [4])) == [1, 2, 3, 4]",
[1, 2, 3], True),
# functools:reduce
("def check(value):\n return functools.reduce(lambda a, b: a + b, value) == 6",
[1, 2, 3], True),
# statistics:mean
("def check(value):\n return statistics.mean(value) == 3.0",
[1, 2, 3, 6], True),
# unicodedata:中文是 Lo 类
("def check(value):\n return unicodedata.category('中') == 'Lo'",
"x", True),
# html:escape
("def check(value):\n return html.escape(value) == '&lt;a&gt;'",
"<a>", True),
# hashlib:sha256 hex
("def check(value):\n return hashlib.sha256(value.encode()).hexdigest()[:6] == '2cf24d'",
"hello", True),
# base64:b64encode
("def check(value):\n return base64.b64encode(value.encode()).decode() == 'aGVsbG8='",
"hello", True),
# uuid:UUID 解析
("def check(value):\n u = uuid.UUID(value)\n return str(u) == value",
"12345678-1234-5678-1234-567812345678", True),
# difflib:SequenceMatcher.ratio
("def check(value):\n return difflib.SequenceMatcher(None, value, 'hello').ratio() > 0.8",
"hello", True),
# enum:定义枚举
("def check(value):\n class Color(enum.Enum):\n RED = 'r'\n return Color.RED.value == 'r'",
"x", True),
# dataclasses:定义 dataclass
("def check(value):\n @dataclasses.dataclass\n class P:\n x: int\n return P(x=1).x == 1",
"x", True),
# struct:pack/unpack
("def check(value):\n n = struct.unpack('>I', value)[0]\n return n == 1",
b'\x00\x00\x00\x01', True),
# array:array 模块
("def check(value):\n a = array.array('i', [1, 2, 3])\n return sum(a) == 6",
"x", True),
# fractions:Fraction
("def check(value):\n return fractions.Fraction('1/3') + fractions.Fraction('1/3') == fractions.Fraction('2/3')",
"x", True),
# numbers:Number ABC
("def check(value):\n return isinstance(1, numbers.Number)",
"x", True),
# operator:add / itemgetter
("def check(value):\n return operator.add(1, 2) == 3",
"x", True),
# copy:deepcopy
("def check(value):\n a = [[1]]\n b = copy.deepcopy(a)\n a[0][0] = 9\n return b[0][0] == 1",
"x", True),
# pprint:pformat 不打印,纯格式化
("def check(value):\n s = pprint.pformat({'a': 1, 'b': [1, 2]})\n return 'a' in s and 'b' in s",
"x", True),
# reprlib:repr
("def check(value):\n return isinstance(reprlib.repr(list(range(100))), str)",
"x", True),
# typing:Any
("def check(value):\n return typing.Any is not None",
"x", True),
# ipaddress:IPv4 校验
("def check(value):\n ip = ipaddress.ip_address(value)\n return ip.version == 4",
"192.168.1.1", True),
# binascii:hexlify
("def check(value):\n return binascii.hexlify(value.encode()).decode() == '616263'",
"abc", True),
])
def test_module_preloaded_and_works(self, code, value, expected):
"""每个模块都已在沙箱 namespace 里 preload,用户代码不需要 import 直接用。"""
assert run_rule("string", None, code, value) is expected
def test_explicit_import_works(self):
"""用户也可以显式 import —— _safe_import 不会拦白名单内的模块。"""
code = (
"import json\n"
"def check(value):\n"
" obj = json.loads(value)\n"
" return obj['k'] == 1\n"
)
assert run_rule("string", None, code, '{"k": 1}') is True
def test_explicit_from_import_works(self):
"""from x import y 也走 _safe_import(白名单要拦的是模块名,不是属性)。"""
code = (
"from collections import Counter\n"
"def check(value):\n"
" return Counter(value)['x'] == 2\n"
)
assert run_rule("string", None, code, "xax") is True
# ── 危险模块必须仍然被拒 ────────────────────────────
@pytest.mark.parametrize("dangerous", [
"os", "sys", "subprocess", "socket", "pickle", "shelve",
"ctypes", "importlib", "inspect", "pathlib", "shutil",
"threading", "multiprocessing", "signal", "code", "pdb",
])
def test_dangerous_module_import_blocked(self, dangerous):
code = (
f"import {dangerous}\n"
"def check(value):\n"
" return True\n"
)
with pytest.raises(RuleRunError, match="not allowed"):
run_rule("string", None, code, "x")
def test_urllib_blocked_too(self):
"""urllib 故意不放(import urllib 后 urllib.request 也能属性访问拿到,副作用太大)。
业务上字符串校验用不到网络,json / html escape 已经够覆盖。"""
code = (
"import urllib.parse\n"
"def check(value):\n"
" return True\n"
)
with pytest.raises(RuleRunError, match="not allowed"):
run_rule("string", None, code, "x")
\ No newline at end of file
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment