From SQLi to RCE – Exploiting LangGraph’s Checkpointer

By Yarden Porat

AI agents need memory. Frameworks like LangGraph provide it through checkpointers – persistence layers that store execution state. But what happens when that persistence layer isn’t locked down?

Key Points

• Check Point Research analyzed LangGraph, an open-source framework for stateful AI agents with over 50 million monthly downloads, and uncovered three vulnerabilities in its persistence layer.
  
• Two of them chain into remote code execution: a SQL injection in the SQLite checkpointer (CVE-2025-67644) and an unsafe msgpack deserialization (CVE-2026-28277).

• A third, parallel issue (CVE-2026-27022) introduces the same injection class into the Redis checkpointer.
  
• Who’s at risk: teams self-hosting LangGraph with the SQLite or Redis checkpointer, where the application exposes get_state_history() with a user-controlled filter. LangChain’s managed cloud service, LangSmith Deployment (formerly LangGraph Platform), runs PostgreSQL and is not vulnerable.

• LangChain patched all three issues. Users should update to langgraph-checkpoint-sqlite 3.0.1+, langgraph 1.0.10+, and langgraph-checkpoint-redis 1.0.2+.

Background

LangGraph is an open-source framework for building stateful, multi-agent AI systems with built-in persistence. It’s an extension of LangChain, with over 50 million monthly downloads according to PyPI stats.

Checkpointers are LangGraph’s persistence layer that stores execution state at each step. LangGraph supports two checkpointer implementations: SQLite and PostgreSQL.

Vulnerability #1: SQL Injection (CVE-2025-67644)

The SQLite Checkpointer Database Schema:
The SQLite checkpointer uses an internal table called checkpoints with the following structure:

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)
);

The metadata column stores additional contextual information about each checkpoint in JSON format. For example:

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

The list() Function and Filtering:

When calling the list() function on sqliteSaver (the checkpointer), the filter parameter is used to query checkpoints based on their metadata:

def list(
    self,
    config: RunnableConfig | None,
    *,
    filter: dict[str, Any] | None = None,  # Used to filter by metadata
    before: RunnableConfig | None = None,
    limit: int | None = None,
) -> Iterator[CheckpointTuple]:

The filter parameter is passed to an internal function called _metadata_predicate, which constructs the SQL WHERE clause to query checkpoints by their metadata fields.

# 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)

The Injection

The vulnerability exists in how _metadata_predicate handles the query_key from the filter dictionary.
Notice this critical line:

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

An attacker-controlled filter could provide a query_key with a ' character that will escape the JSON path string and inject arbitrary SQL code.

Injection -> Arbitrary Deserialization

To understand how SQL injection leads to arbitrary deserialization, we need to see the complete picture.
Here’s the SQL query that gets executed in list():

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

This query retrieves checkpoint data from the database, including the checkpoint’s BLOB column.
The results are then processed:

async for (
    thread_id,
    checkpoint_ns,
    checkpoint_id,
    parent_checkpoint_id,
    type,
    checkpoint,  # ← This comes directly from the SQL query results
    metadata,
) in cur:  # ← cur contains the query results
    # ... 
    yield CheckpointTuple(
        # ...
        self.serde.loads_typed((type, checkpoint)),  # ← Deserialization
        # ...
    )

The checkpoint contains serialized data, and when fetched gets deserialized.

The Attack

Using SQL injection in the WHERE clause, an attacker can inject a UNION SELECT that adds their own row to the query results:

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'', '{}' -- )
ORDER BY checkpoint_id DESC

The injected UNION SELECT returns a fake checkpoint row where the checkpoint column contains attacker-controlled serialized data. When the code loops through the query results, it deserializes this malicious checkpoint’s BLOB, giving the attacker arbitrary deserialization

Vulnerability #2: MsgPack Unsafe Deserialization (CVE-2026-28277)

Now let’s examine what happens during deserialization. The self.serde.loads_typed() function that deserializes checkpoint data looks like this:

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)

Key Takeaways

• By Yarden Porat AI agents need memory.
• Frameworks like LangGraph provide it through checkpointers persistence layers that store execution state.
• but what happens when that persistence layer isnt locked down?