Langroid has Prompt to SQL Injection, Leading to RCE

@@ -17,6 +17,29 @@ automatically extracting schemas from the database.
 
 ## Before you begin
 
+!!! warning "Security: SQLChatAgent executes LLM-generated SQL"
+    `SQLChatAgent` runs SQL produced by a language model against your
+    database. Because the LLM is influenceable by prompt injection — both
+    through the user's question and through any data the LLM reads back from
+    the database — an attacker who can shape either input can attempt to
+    coerce the agent into running dangerous SQL.
+
+    The agent defaults to **SELECT-only** execution (CVE-2026-25879) and
+    rejects queries that match known dangerous primitives
+    (e.g. PostgreSQL `COPY ... FROM PROGRAM`, MySQL `INTO OUTFILE`, SQLite
+    `load_extension`, MSSQL `xp_cmdshell`). See
+    [Security configuration](#security-configuration) below.
+
+    Independently of the agent's defenses, you should always:
+
+    - Connect using a **least-privilege database role** that does not have
+      `pg_execute_server_program`, `FILE`, `xp_cmdshell`, or other
+      privileges enabling code execution or filesystem access.
+    - Treat the agent's prompt input as untrusted if it can come from end
+      users.
+    - Avoid pointing the agent at databases that contain other tenants'
+      data.
+
 !!! note "Data Privacy Considerations"
     Since the SQLChatAgent uses the OpenAI GPT-4 as the underlying language model,
     users should be aware that database information processed by the agent may be
@@ -100,6 +123,48 @@ agent will automatically generate the file using the built-in Postgres table/col
   reduces token usage, though it may result in 1-3 additional OpenAI API calls before
   the final SQL query is generated.
 
+## Security configuration
+
+`SQLChatAgentConfig` exposes two flags that govern which SQL the agent will
+execute:
+
+* `allowed_statement_types: List[str] = ["SELECT"]` — the SQL statement
+  types the agent is permitted to run. Each query is parsed with
+  [`sqlglot`](https://github.com/tobymao/sqlglot); any statement whose top
+  level falls outside this list is rejected before it reaches the database.
+  Multi-statement queries are validated statement-by-statement, so a
+  payload like `SELECT 1; DROP TABLE t` is also rejected.
+* `allow_dangerous_operations: bool = False` — when `True`, **disables**
+  the statement-type allowlist and the dangerous-pattern blocklist. Only
+  set this if (a) the agent's prompt input is fully trusted, and (b) the
+  database connection uses a least-privilege role.
+
+In addition to the statement-type check, the agent maintains a dialect-aware
+blocklist of patterns that enable code execution, arbitrary file access, or
+other escape primitives — for example:
+
+| Dialect    | Blocked construct (examples)                                      |
+|------------|-------------------------------------------------------------------|
+| PostgreSQL | `COPY … FROM/TO PROGRAM`, `pg_read_server_files`, `lo_import/export`, `CREATE EXTENSION`, `CREATE FUNCTION` |
+| MySQL      | `INTO OUTFILE`, `INTO DUMPFILE`, `LOAD_FILE(…)`, `LOAD DATA`      |
+| SQLite     | `load_extension(…)`, `ATTACH DATABASE`                            |
+| SQL Server | `xp_cmdshell`, `sp_OACreate`, `OPENROWSET`, `BULK INSERT`         |
+
+To enable database writes from the agent, extend the allowlist explicitly
+rather than disabling all checks:
+
+```python
+config = SQLChatAgentConfig(
+    database_uri="postgresql://example.db",
+    allowed_statement_types=["SELECT", "INSERT", "UPDATE", "DELETE"],
+)
+```
+
+If you need DDL or other operations the allowlist doesn't cover, the
+recommended approach is still to keep `allow_dangerous_operations=False`
+and grant the DB role exactly the privileges you need, rather than relying
+on the agent to filter SQL.
+
 ## Putting it all together
 
 In the code below, we will allow the agent to generate the context descriptions

@@ -1,13 +1,14 @@
 """
-Agent that allows interaction with an SQL database using SQLAlchemy library. 
-The agent can execute SQL queries in the database and return the result. 
+Agent that allows interaction with an SQL database using SQLAlchemy library.
+The agent can execute SQL queries in the database and return the result.
 
 Functionality includes:
 - adding table and column context
 - asking a question about a SQL schema
 """
 
 import logging
+import re
 from typing import Any, Dict, List, Optional, Sequence, Union
 
 from rich.console import Console
@@ -24,6 +25,15 @@
 except ImportError as e:
     raise LangroidImportError(extra="sql", error=str(e))
 
+try:
+    # sqlglot is required for the statement-type allowlist enforced in
+    # `_validate_query`. Importing it at module load ensures the security
+    # guarantee cannot be silently bypassed by a partial/stale install.
+    import sqlglot
+    from sqlglot import expressions as sqlglot_exp
+except ImportError as e:
+    raise LangroidImportError(extra="sql", error=str(e))
+
 from langroid.agent.chat_agent import ChatAgent, ChatAgentConfig
 from langroid.agent.chat_document import ChatDocument
 from langroid.agent.special.sql.utils.description_extractors import (
@@ -96,6 +106,46 @@
 SQL_ERROR_MSG = "There was an error in your SQL Query"
 
 
+# Dialect-specific SQL patterns that enable code execution, arbitrary file
+# access, or other escapes from the database engine. Matched against the raw
+# query text (case-insensitive) as a defense-in-depth layer in addition to the
+# sqlglot-based statement-type allowlist.
+_DANGEROUS_SQL_PATTERNS: List["re.Pattern[str]"] = [
+    # PostgreSQL: COPY ... FROM/TO PROGRAM executes shell commands as the DB
+    # server OS user. This is the primitive used in CVE-2026-25879.
+    re.compile(r"\bcopy\b[\s\S]*\bprogram\b", re.IGNORECASE),
+    # PostgreSQL server-side filesystem access
+    re.compile(r"\bpg_read_server_files?\b", re.IGNORECASE),
+    re.compile(r"\bpg_read_binary_file\b", re.IGNORECASE),
+    re.compile(r"\bpg_ls_dir\b", re.IGNORECASE),
+    re.compile(r"\blo_(import|export)\b", re.IGNORECASE),
+    # MySQL/MariaDB filesystem
+    re.compile(r"\binto\s+(outfile|dumpfile)\b", re.IGNORECASE),
+    re.compile(r"\bload_file\s*\(", re.IGNORECASE),
+    re.compile(r"\bload\s+data\b", re.IGNORECASE),
+    # SQLite: load_extension enables loading arbitrary shared objects;
+    # ATTACH DATABASE can read/write arbitrary files.
+    re.compile(r"\bload_extension\s*\(", re.IGNORECASE),
+    re.compile(r"\battach\s+database\b", re.IGNORECASE),
+    # SQL Server: command execution and OLE automation
+    re.compile(r"\bxp_cmdshell\b", re.IGNORECASE),
+    re.compile(r"\bsp_oacreate\b", re.IGNORECASE),
+    re.compile(r"\bsp_oamethod\b", re.IGNORECASE),
+    re.compile(r"\bopenrowset\b", re.IGNORECASE),
+    re.compile(r"\bbulk\s+insert\b", re.IGNORECASE),
+    # Generic: stored-program creation and procedural language extensions
+    re.compile(
+        r"\bcreate\s+(or\s+replace\s+)?(function|procedure|trigger)\b", re.IGNORECASE
+    ),
+    re.compile(r"\bcreate\s+extension\b", re.IGNORECASE),
+]
+
+
+# Default set of SQL statement types the agent is allowed to execute when
+# `allow_dangerous_operations` is False. SELECT-only is safe for Q&A workloads.
+_DEFAULT_ALLOWED_STATEMENTS: List[str] = ["SELECT"]
+
+
 class SQLChatAgentConfig(ChatAgentConfig):
     system_message: str = DEFAULT_SQL_CHAT_SYSTEM_MESSAGE
     user_message: None | str = None
@@ -117,6 +167,22 @@ class SQLChatAgentConfig(ChatAgentConfig):
     max_result_rows: int | None = None  # limit query results to this
     max_retained_tokens: int | None = None  # limit history of query results to this
 
+    # --- Security controls (see CVE-2026-25879) ---------------------------
+    # By default, the agent only executes SELECT statements and rejects any
+    # query that matches a known dangerous pattern (e.g. PostgreSQL
+    # `COPY ... FROM PROGRAM`, MySQL `INTO OUTFILE`, SQLite `load_extension`,
+    # MSSQL `xp_cmdshell`). The LLM-generated SQL is influenceable by prompt
+    # injection — including injection via data the LLM reads back from the
+    # database — so executing it without restrictions is unsafe when the DB
+    # role has elevated privileges.
+    #
+    # To enable writes: extend allowed_statement_types, e.g. ["SELECT",
+    # "INSERT", "UPDATE", "DELETE"]. To disable all checks (only do this with
+    # a least-privilege DB role and trusted prompts): set
+    # allow_dangerous_operations=True.
+    allowed_statement_types: List[str] = list(_DEFAULT_ALLOWED_STATEMENTS)
+    allow_dangerous_operations: bool = False
+
     """
     Optional, but strongly recommended, context descriptions for tables, columns, 
     and relationships. It should be a dictionary where each key is a table name 
@@ -261,6 +327,19 @@ def _init_system_message(self) -> None:
         message = self._format_message()
         self.config.system_message = self.config.system_message.format(mode=message)
 
+        if not self.config.allow_dangerous_operations:
+            allowed = sorted(
+                t.strip().upper() for t in self.config.allowed_statement_types
+            )
+            self.config.system_message += (
+                f"\n\nIMPORTANT - SECURITY POLICY:\n"
+                f"You may ONLY issue SQL queries whose top-level statement "
+                f"type is one of: {allowed}. Any other statement type "
+                f"(e.g. DDL, COPY, EXEC, multi-statement scripts that "
+                f"include a disallowed type) will be REJECTED by the "
+                f"executor and not run.\n"
+            )
+
         if self.config.chat_mode:
             self.config.addressing_prefix = self.config.addressing_prefix or SEND_TO
             self.config.system_message += ADDRESSING_INSTRUCTION.format(
@@ -455,6 +534,86 @@ def _tool_result_llm_answer_prompt(self) -> str:
                 set to the answer or result
                 """
 
+    def _sqlglot_dialect(self) -> Optional[str]:
+        """Map the SQLAlchemy dialect name to a sqlglot dialect name."""
+        if self.engine is None:
+            return None
+        name: str = str(self.engine.dialect.name)
+        # sqlglot uses 'postgres', not 'postgresql'; 'tsql' for MSSQL.
+        mapping: Dict[str, str] = {"postgresql": "postgres", "mssql": "tsql"}
+        return mapping.get(name, name)
+
+    def _validate_query(self, query: str) -> Optional[str]:
+        """
+        Check whether `query` is permitted under the agent's security config.
+
+        Returns None if the query may be executed, otherwise an error message
+        explaining why it was rejected (to be relayed to the LLM).
+        """
+        if self.config.allow_dangerous_operations:
+            return None
+
+        for pat in _DANGEROUS_SQL_PATTERNS:
+            if pat.search(query):
+                logger.warning(
+                    "SQLChatAgent rejected query matching dangerous pattern "
+                    f"{pat.pattern!r}: {query!r}"
+                )
+                return (
+                    f"Query REJECTED for safety: it matches a pattern "
+                    f"({pat.pattern!r}) that enables code execution, "
+                    f"filesystem access, or other unsafe operations. "
+                    f"Rewrite the query without using this construct, or ask "
+                    f"the operator to set `allow_dangerous_operations=True` "
+                    f"on the SQLChatAgent config."
+                )
+
+        allowed = {t.strip().upper() for t in self.config.allowed_statement_types}
+        try:
+            statements = sqlglot.parse(query, read=self._sqlglot_dialect())
+        except Exception as e:
+            logger.warning(f"sqlglot failed to parse query {query!r}: {e}")
+            return (
+                f"Query REJECTED for safety: could not be parsed to verify it "
+                f"is a {sorted(allowed)} statement ({e}). Rewrite the query "
+                f"more simply, or ask the operator to set "
+                f"`allow_dangerous_operations=True` on the SQLChatAgent config."
+            )
+
+        kind_map = {
+            sqlglot_exp.Select: "SELECT",
+            sqlglot_exp.Insert: "INSERT",
+            sqlglot_exp.Update: "UPDATE",
+            sqlglot_exp.Delete: "DELETE",
+            sqlglot_exp.Merge: "MERGE",
+            sqlglot_exp.Create: "CREATE",
+            sqlglot_exp.Drop: "DROP",
+            sqlglot_exp.Alter: "ALTER",
+            sqlglot_exp.TruncateTable: "TRUNCATE",
+            sqlglot_exp.Command: "COMMAND",
+        }
+        for stmt in statements:
+            if stmt is None:
+                continue
+            kind = next(
+                (v for k, v in kind_map.items() if isinstance(stmt, k)),
+                type(stmt).__name__.upper(),
+            )
+            if kind not in allowed:
+                logger.warning(
+                    f"SQLChatAgent rejected {kind} statement (allowed: "
+                    f"{sorted(allowed)}): {query!r}"
+                )
+                return (
+                    f"Query REJECTED for safety: statement type {kind!r} is "
+                    f"not in the allowed list {sorted(allowed)}. Rewrite the "
+                    f"query as one of the allowed statement types, or ask "
+                    f"the operator to extend `allowed_statement_types` "
+                    f"(or set `allow_dangerous_operations=True`) on the "
+                    f"SQLChatAgent config."
+                )
+        return None
+
     def run_query(self, msg: RunQueryTool) -> str:
         """
         Handle a RunQueryTool message by executing a SQL query and returning the result.
@@ -468,6 +627,18 @@ def run_query(self, msg: RunQueryTool) -> str:
         query = msg.query
         session = self.Session
         self.used_run_query = True
+
+        rejection = self._validate_query(query)
+        if rejection is not None:
+            return f"""
+        Below is the result from your use of the TOOL `{RunQueryTool.name()}`:
+        ==== result ====
+        {rejection}
+        ================
+
+        Try a different query that complies with the policy above.
+        """
+
         try:
             logger.info(f"Executing SQL query: {query}")
 

@@ -95,6 +95,7 @@ db = [
     "psycopg2<3.0.0,>=2.9.7",
     "psycopg2-binary>=2.9.10",
     "pymysql<2.0.0,>=1.1.0",
+    "sqlglot>=23.0.0",
 ]
 
 all = [
@@ -109,6 +110,7 @@ all = [
     "sqlalchemy<3.0.0,>=2.0.19",
     "psycopg2<3.0.0,>=2.9.7",
     "pymysql<2.0.0,>=1.1.0",
+    "sqlglot>=23.0.0",
     "sentence-transformers<3.0.0,>=2.2.2",
     "torch<3.0.0,>=2.0.0",
     "transformers<5.0.0,>=4.40.1",
@@ -192,16 +194,19 @@ postgres = [
     "psycopg2<3.0.0,>=2.9.7",
     "psycopg2-binary>=2.9.10",
     "sqlalchemy<3.0.0,>=2.0.19",
+    "sqlglot>=23.0.0",
 ]
 
 mysql = [
     "pymysql<2.0.0,>=1.1.0",
+    "sqlglot>=23.0.0",
 ]
 
 sql = [
     "sqlalchemy<3.0.0,>=2.0.19",
     "pymysql<2.0.0,>=1.1.0",
     "psycopg2<3.0.0,>=2.9.7",
+    "sqlglot>=23.0.0",
 ]
 
 litellm = [

@@ -134,10 +134,14 @@ def _test_sql_chat_agent(
     use_schema_tools: bool = False,
     turns: int = 18,
     addressing_prefix: str = "",
+    allowed_statement_types: list[str] | None = None,
 ) -> None:
     """
     Test the SQLChatAgent with a uri as data source
     """
+    config_kwargs: dict = {}
+    if allowed_statement_types is not None:
+        config_kwargs["allowed_statement_types"] = allowed_statement_types
     agent_config = SQLChatAgentConfig(
         name="SQLChatAgent",
         database_session=db_session,
@@ -150,6 +154,7 @@ def _test_sql_chat_agent(
         chat_mode=False,
         use_helper=True,
         llm=OpenAIGPTConfig(supports_json_schema=json_schema),
+        **config_kwargs,
     )
     agent = SQLChatAgent(agent_config)
     task = Task(agent, interactive=False)
@@ -227,6 +232,7 @@ def test_sql_chat_db_update(
 ):
     set_global(test_settings)
     # with context descriptions:
+    # UPDATE requires extending the default SELECT-only allowlist.
     _test_sql_chat_agent(
         fn_api=fn_api,
         tools_api=tools_api,
@@ -235,6 +241,7 @@ def test_sql_chat_db_update(
         context=mock_context,
         prompt="Update Bob's sale amount to 900",
         answer="900",
+        allowed_statement_types=["SELECT", "UPDATE"],
     )
 
     _test_sql_chat_agent(
@@ -256,6 +263,7 @@ def test_sql_chat_db_update(
         context={},
         prompt="Update Bob's sale amount to 9100",
         answer="9100",
+        allowed_statement_types=["SELECT", "UPDATE"],
     )
 
     _test_sql_chat_agent(

@@ -0,0 +1,197 @@
+"""
+Unit tests for SQLChatAgent's query validator (CVE-2026-25879 mitigation).
+
+These tests exercise `_validate_query` and `run_query` directly without an
+LLM, so they don't require API credentials.
+"""
+
+import pytest
+
+from langroid.exceptions import LangroidImportError
+
+try:
+    from sqlalchemy import Column, Integer, String, create_engine
+    from sqlalchemy.ext.declarative import declarative_base
+    from sqlalchemy.orm import sessionmaker
+except ImportError as e:
+    raise LangroidImportError(extra="sql", error=str(e))
+
+from langroid.agent.special.sql.sql_chat_agent import (
+    SQLChatAgent,
+    SQLChatAgentConfig,
+)
+from langroid.agent.special.sql.utils.tools import RunQueryTool
+
+Base = declarative_base()
+
+
+class Item(Base):
+    __tablename__ = "items"
+    id = Column(Integer, primary_key=True)
+    name = Column(String, nullable=False)
+
+
+@pytest.fixture
+def session():
+    engine = create_engine("sqlite:///:memory:")
+    Base.metadata.create_all(engine)
+    Session = sessionmaker(bind=engine)
+    s = Session()
+    s.add(Item(id=1, name="a"))
+    s.commit()
+    yield s
+    s.close()
+
+
+def _make_agent(session, **cfg_kwargs):
+    cfg = SQLChatAgentConfig(
+        database_session=session,
+        llm=None,
+        use_helper=False,
+        **cfg_kwargs,
+    )
+    return SQLChatAgent(cfg)
+
+
+# ---------------------------------------------------------------------------
+# Validator unit tests
+# ---------------------------------------------------------------------------
+
+
+def test_select_allowed_by_default(session):
+    agent = _make_agent(session)
+    assert agent._validate_query("SELECT * FROM items") is None
+
+
+@pytest.mark.parametrize(
+    "query",
+    [
+        "DROP TABLE items",
+        "CREATE TABLE x (id int)",
+        "ALTER TABLE items ADD COLUMN y int",
+        "UPDATE items SET name='b' WHERE id=1",
+        "INSERT INTO items (id, name) VALUES (2, 'b')",
+        "DELETE FROM items WHERE id=1",
+        "TRUNCATE TABLE items",
+    ],
+)
+def test_non_select_blocked_by_default(session, query):
+    agent = _make_agent(session)
+    rejection = agent._validate_query(query)
+    assert rejection is not None
+    assert "REJECTED" in rejection
+
+
+def test_cve_2026_25879_poc_blocked(session):
+    """The exact reproducer from the security advisory must be rejected."""
+    agent = _make_agent(session)
+    poc = (
+        "DROP TABLE IF EXISTS log;\n"
+        "CREATE TABLE log(content text);\n"
+        "COPY log(content) FROM PROGRAM 'id';\n"
+        "SELECT * FROM log;"
+    )
+    rejection = agent._validate_query(poc)
+    assert rejection is not None
+    assert "REJECTED" in rejection
+
+
+@pytest.mark.parametrize(
+    "query",
+    [
+        # PostgreSQL: command execution
+        "COPY t FROM PROGRAM 'id'",
+        "COPY t (c) FROM PROGRAM 'whoami'",
+        # PostgreSQL: server-side file read
+        "SELECT pg_read_server_files('/etc/passwd')",
+        "SELECT pg_read_binary_file('/etc/shadow')",
+        "SELECT pg_ls_dir('/')",
+        "SELECT lo_import('/etc/passwd')",
+        # MySQL: filesystem
+        "SELECT * FROM items INTO OUTFILE '/tmp/x'",
+        "SELECT * FROM items INTO DUMPFILE '/tmp/x'",
+        "SELECT load_file('/etc/passwd')",
+        "LOAD DATA INFILE '/etc/passwd' INTO TABLE items",
+        # SQLite: arbitrary code / file access
+        "SELECT load_extension('/tmp/evil.so')",
+        "ATTACH DATABASE '/etc/passwd' AS p",
+        # MSSQL: command execution
+        "EXEC xp_cmdshell 'id'",
+        "EXEC sp_OACreate 'WScript.Shell', @s OUT",
+        "SELECT * FROM OPENROWSET('SQLNCLI', 'connstring', 'q')",
+        "BULK INSERT t FROM '/etc/passwd'",
+        # Generic: stored programs and extensions
+        "CREATE FUNCTION evil() RETURNS void AS $$ ... $$ LANGUAGE plpgsql",
+        "CREATE OR REPLACE PROCEDURE p() AS ...",
+        "CREATE EXTENSION plpython3u",
+    ],
+)
+def test_dangerous_patterns_blocked(session, query):
+    agent = _make_agent(session)
+    rejection = agent._validate_query(query)
+    assert rejection is not None
+    assert "REJECTED" in rejection
+
+
+def test_multi_statement_with_buried_drop_blocked(session):
+    agent = _make_agent(session)
+    rejection = agent._validate_query("SELECT 1; DROP TABLE items")
+    assert rejection is not None
+    assert "REJECTED" in rejection
+
+
+def test_allow_dangerous_operations_bypasses_all_checks(session):
+    agent = _make_agent(session, allow_dangerous_operations=True)
+    poc = "DROP TABLE IF EXISTS log;\n" "COPY log(content) FROM PROGRAM 'id';\n"
+    assert agent._validate_query(poc) is None
+    assert agent._validate_query("DROP TABLE items") is None
+    assert agent._validate_query("EXEC xp_cmdshell 'id'") is None
+
+
+def test_extended_allowlist_permits_writes(session):
+    agent = _make_agent(
+        session,
+        allowed_statement_types=["SELECT", "INSERT", "UPDATE", "DELETE"],
+    )
+    assert agent._validate_query("UPDATE items SET name='b' WHERE id=1") is None
+    assert agent._validate_query("INSERT INTO items VALUES (2, 'b')") is None
+    assert agent._validate_query("DELETE FROM items WHERE id=1") is None
+    # Still blocks CREATE/DROP even with writes allowed.
+    assert agent._validate_query("DROP TABLE items") is not None
+    assert agent._validate_query("CREATE TABLE x (id int)") is not None
+    # Still blocks dialect-specific dangerous primitives.
+    assert agent._validate_query("SELECT load_extension('e')") is not None
+
+
+# ---------------------------------------------------------------------------
+# Integration tests via run_query (no LLM involved)
+# ---------------------------------------------------------------------------
+
+
+def test_run_query_rejects_drop_without_executing(session):
+    agent = _make_agent(session)
+    result = agent.run_query(RunQueryTool(query="DROP TABLE items"))
+    assert "REJECTED" in result
+    # The table must still exist after the rejected call.
+    rows = session.execute(
+        __import__("sqlalchemy").text("SELECT COUNT(*) FROM items")
+    ).scalar()
+    assert rows == 1
+
+
+def test_run_query_allows_select(session):
+    agent = _make_agent(session)
+    result = agent.run_query(RunQueryTool(query="SELECT name FROM items"))
+    assert "REJECTED" not in result
+    assert "a" in result
+
+
+def test_run_query_with_dangerous_ops_allowed_runs_drop(session):
+    agent = _make_agent(session, allow_dangerous_operations=True)
+    result = agent.run_query(RunQueryTool(query="DROP TABLE items"))
+    assert "REJECTED" not in result
+    # Sanity check that the table actually got dropped.
+    with pytest.raises(Exception):
+        session.execute(
+            __import__("sqlalchemy").text("SELECT COUNT(*) FROM items")
+        ).scalar()