x

libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py

@@ -329,8 +329,9 @@ class SqliteSaver(BaseCheckpointSaver[str]):
         FROM checkpoints
         {where}
         ORDER BY checkpoint_id DESC"""
-        if limit:
-            query += f" LIMIT {limit}"
+        if limit is not None:
+            query += " LIMIT ?"
+            param_values = (*param_values, limit)
         with self.cursor(transaction=False) as cur, closing(self.conn.cursor()) as wcur:
             cur.execute(query, param_values)
             for (

libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/aio.py

@@ -425,8 +425,9 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
         FROM checkpoints
         {where}
         ORDER BY checkpoint_id DESC"""
-        if limit:
-            query += f" LIMIT {limit}"
+        if limit is not None:
+            query += " LIMIT ?"
+            params = (*params, limit)
         async with (
             self.lock,
             self.conn.execute(query, params) as cur,

libs/checkpoint-sqlite/tests/test_aiosqlite.py

@@ -113,4 +113,78 @@ class TestAsyncSqliteSaver:
                 search_results_5[1].config["configurable"]["checkpoint_ns"],
             } == {"", "inner"}
 
-            # TODO: test before and limit params
+            # Test limit param
+            search_results_6 = [
+                c
+                async for c in saver.alist(
+                    {"configurable": {"thread_id": "thread-2"}}, limit=1
+                )
+            ]
+            assert len(search_results_6) == 1
+            assert search_results_6[0].config["configurable"]["thread_id"] == "thread-2"
+
+            # Test before param
+            search_results_7 = [
+                c async for c in saver.alist(None, before=search_results_5[1].config)
+            ]
+            assert len(search_results_7) == 1
+            assert search_results_7[0].config["configurable"]["thread_id"] == "thread-1"
+
+    async def test_limit_parameter_sql_injection_prevention(self) -> None:
+        """Test that the limit parameter properly uses parameterized queries to prevent SQL injection."""
+        async with AsyncSqliteSaver.from_conn_string(":memory:") as saver:
+            # Setup: Create multiple checkpoints
+            for i in range(5):
+                config: RunnableConfig = {
+                    "configurable": {
+                        "thread_id": f"thread-{i}",
+                        "checkpoint_ns": "",
+                    }
+                }
+                checkpoint = empty_checkpoint()
+                metadata: CheckpointMetadata = {"index": i}
+                await saver.aput(config, checkpoint, metadata, {})
+
+            # Test that limit works correctly with valid integer
+            results = [c async for c in saver.alist(None, limit=2)]
+            assert len(results) == 2
+
+            # Test that limit=0 returns no results
+            results = [c async for c in saver.alist(None, limit=0)]
+            assert len(results) == 0
+
+            # Test that limit=None returns all results
+            results = [c async for c in saver.alist(None, limit=None)]
+            assert len(results) == 5
+
+            # Test explicit SQL injection attempt via limit parameter
+            # Even if type checking is bypassed and a malicious string is passed,
+            # the parameterized query will treat it as a value, not SQL code
+            # This would cause an error (can't convert string to int for LIMIT),
+            # which is the correct secure behavior
+            malicious_limits = [
+                "1; DROP TABLE checkpoints; --",
+                "1 OR 1=1",
+                "999999 UNION SELECT * FROM checkpoints",
+            ]
+
+            for malicious_limit in malicious_limits:
+                # The parameterized query should safely reject non-integer limits
+                # or convert them in a way that prevents SQL injection
+                try:
+                    # Bypass type checking by casting
+                    results = [
+                        c
+                        async for c in saver.alist(None, limit=malicious_limit)  # type: ignore
+                    ]
+                    # If it doesn't raise an error, it should at least not execute the injection
+                    # SQLite's parameter binding will try to convert the string to an integer
+                    # which will either fail or treat it as 0
+                except Exception:
+                    # Expected: SQLite should reject invalid limit values
+                    pass
+
+            # Verify the checkpoints table still exists and has all data
+            # (would have been dropped if injection succeeded)
+            results = [c async for c in saver.alist(None, limit=None)]
+            assert len(results) == 5

libs/checkpoint-sqlite/tests/test_sqlite.py

@@ -243,3 +243,59 @@ class TestSqliteSaver:
 
             with pytest.raises(ValueError, match="Invalid filter key"):
                 list(saver.list(None, filter={malicious_key: "dummy"}))
+
+    def test_limit_parameter_sql_injection_prevention(self) -> None:
+        """Test that the limit parameter properly uses parameterized queries to prevent SQL injection."""
+        with SqliteSaver.from_conn_string(":memory:") as saver:
+            # Setup: Create multiple checkpoints
+            for i in range(5):
+                config: RunnableConfig = {
+                    "configurable": {
+                        "thread_id": f"thread-{i}",
+                        "checkpoint_ns": "",
+                    }
+                }
+                checkpoint = empty_checkpoint()
+                metadata: CheckpointMetadata = {"index": i}
+                saver.put(config, checkpoint, metadata, {})
+
+            # Test that limit works correctly with valid integer
+            results = list(saver.list(None, limit=2))
+            assert len(results) == 2
+
+            # Test that limit=0 returns no results
+            results = list(saver.list(None, limit=0))
+            assert len(results) == 0
+
+            # Test that limit=None returns all results
+            results = list(saver.list(None, limit=None))
+            assert len(results) == 5
+
+            # Test explicit SQL injection attempt via limit parameter
+            # Even if type checking is bypassed and a malicious string is passed,
+            # the parameterized query will treat it as a value, not SQL code
+            # This would cause an error (can't convert string to int for LIMIT),
+            # which is the correct secure behavior
+            malicious_limits = [
+                "1; DROP TABLE checkpoints; --",
+                "1 OR 1=1",
+                "999999 UNION SELECT * FROM checkpoints",
+            ]
+
+            for malicious_limit in malicious_limits:
+                # The parameterized query should safely reject non-integer limits
+                # or convert them in a way that prevents SQL injection
+                try:
+                    # Bypass type checking by casting
+                    results = list(saver.list(None, limit=malicious_limit))  # type: ignore
+                    # If it doesn't raise an error, it should at least not execute the injection
+                    # SQLite's parameter binding will try to convert the string to an integer
+                    # which will either fail or treat it as 0
+                except Exception:
+                    # Expected: SQLite should reject invalid limit values
+                    pass
+
+            # Verify the checkpoints table still exists and has all data
+            # (would have been dropped if injection succeeded)
+            results = list(saver.list(None, limit=None))
+            assert len(results) == 5