How DB-GPT Handles SQL Injection Prevention in Generative Business Intelligence (GBI)

DB-GPT prevents SQL injection in its Generative Business Intelligence (GBI) feature through a multi-layered defense strategy that includes syntactic sanitization, query parameterization, operation whitelisting, API-key authentication, and connection isolation.

DB-GPT's Generative Business Intelligence (GBI) capability allows large language models (LLMs) to generate SQL statements from natural language queries. Because generated SQL executes directly against user databases, the framework implements rigorous SQL injection prevention measures to protect against malicious payloads and unauthorized data access.

SQL Sanitization and Parameterization

The first line of defense resides in the sanitize_sql helper function implemented in packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/editor/api_editor_v1.py. This utility processes every SQL statement—whether generated by an LLM or provided by a user—before it reaches the database engine.

Dangerous Pattern Detection

The sanitization routine performs several normalization steps: it strips SQL comments, collapses excess whitespace, and rejects multiple statements (detecting semicolons followed by non-comment text). The function maintains a curated blocklist of dangerous patterns including INTO OUTFILE, LOAD DATA, SYSTEM, EXEC, and DROP DATABASE. For DuckDB connections, it adds additional restrictions against COPY, EXPORT, IMPORT, INSTALL, and PRAGMA operations.

Query Parameterization

After passing syntactic checks, the function replaces literal string values with placeholder parameters (e.g., :param_0). It returns a tuple containing a boolean safety flag, the parameterized SQL string, and a mapping dictionary of placeholder-to-original-value mappings. This ensures that downstream execution via conn.query_ex(sql, params=..., timeout=30) treats inputs as bound parameters rather than executable code, effectively neutralizing injection attempts through string escaping or command chaining.

Operation Whitelisting

DB-GPT enforces a strict operation whitelist after sanitization. The system only permits statements beginning with SELECT, CREATE TABLE, INSERT INTO, UPDATE, DELETE FROM, or ALTER TABLE. Any attempt to execute DROP, TRUNCATE, GRANT, or other administrative commands results in immediate rejection with a descriptive error message. This constraint limits the blast radius of potential vulnerabilities by ensuring that even bypassed sanitization cannot destroy schema objects or modify permissions.

API-Key Based Access Control

All HTTP endpoints serving GBI functionality utilize the check_api_key dependency defined in packages/dbgpt-serve/src/dbgpt_serve/utils/_template_files/default_serve_template/api/endpoints.py. When api_keys are configured in the server settings, every incoming request must present a valid Bearer token in the authorization header. Requests lacking valid credentials receive an HTTP 401 response. If no API keys are defined, the endpoint operates in open mode—suitable for development environments but not production deployments.

Runtime Safeguards

Beyond static analysis and authentication, DB-GPT implements runtime protections to mitigate availability risks and lateral movement.

Connection Isolation via ConnectorManager

The framework accesses databases through the ConnectorManager (CFG.local_db_manager), instantiated in packages/dbgpt-core/src/dbgpt/_private/config.py. This manager provisions isolated connector instances for each database, ensuring that compromised queries cannot access connection pools or credentials belonging to other data sources. Each connector operates within its own process or thread boundary, containing potential execution flaws.

Query Timeouts

Every database interaction invoked through query_ex (implemented in packages/dbgpt-core/src/dbgpt/datasource/rdbms/base.py) requires an explicit timeout parameter, typically set to 30 seconds. Long-running or resource-exhausting queries terminate automatically, preventing denial-of-service attacks that attempt to lock tables or consume excessive CPU/memory through cartesian products or infinite loops.

Implementation Example

The following pattern demonstrates the complete security flow from LLM generation to safe execution:

from dbgpt._private.config import Config
from dbgpt_app.openapi.api_v1.editor.api_editor_v1 import sanitize_sql

CFG = Config()

def execute_gbi_query(nl_query: str, db_name: str):
    # 1. Generate SQL via LLM (simplified)

    llm_response = llm.generate_sql(nl_query)
    
    # 2. Sanitize and parameterize

    is_safe, safe_sql, params = sanitize_sql(llm_response, db_type="sqlite")
    if not is_safe:
        raise ValueError(f"Rejected unsafe SQL: {safe_sql}")
    
    # 3. Execute via isolated connector with timeout

    connector = CFG.local_db_manager.get_connector(db_name)
    columns, rows = connector.query_ex(safe_sql, params=params, timeout=30)
    
    return columns, rows

For FastAPI endpoints, the security layers combine as shown below:

from fastapi import Depends, Body
from dbgpt_serve.utils._template_files.default_serve_template.api.endpoints import check_api_key

@router.post(
    "/gbi/run_sql",
    dependencies=[Depends(check_api_key)],  # API-key enforcement

)
async def run_gbi_sql(payload: dict = Body()):
    db_name = payload["db_name"]
    raw_sql = payload["sql"]
    
    conn = CFG.local_db_manager.get_connector(db_name)
    is_safe, safe_sql, params = sanitize_sql(
        raw_sql, 
        getattr(conn, "db_type", "")
    )
    
    if not is_safe:
        return {"error": f"Unsafe SQL detected: {safe_sql}"}
    
    cols, data = conn.query_ex(safe_sql, params=params, timeout=30)
    return {"columns": cols, "rows": data}

Summary

• Syntactic Sanitization: The sanitize_sql function in api_editor_v1.py strips comments, normalizes whitespace, and blocks dangerous SQL patterns before execution.
• Parameterized Queries: All literals convert to bound parameters (:param_0), ensuring database drivers treat values as data, not code.
• Operation Whitelisting: Only SELECT, CREATE TABLE, INSERT, UPDATE, DELETE, and ALTER TABLE statements execute; destructive commands fail immediately.
• API-Key Gateway: The check_api_key dependency enforces bearer token validation on all GBI endpoints when configured.
• Defense in Depth: ConnectorManager isolation and mandatory 30-second timeouts prevent resource exhaustion and lateral movement.

Frequently Asked Questions

How does DB-GPT's sanitize_sql function prevent SQL injection?

According to the DB-GPT source code, sanitize_sql prevents SQL injection by first normalizing the input—removing comments and extra whitespace—then rejecting multiple statements and dangerous keywords like INTO OUTFILE or SYSTEM. It subsequently extracts all literal values, replaces them with named placeholders, and returns a parameterized query that the database driver executes as a prepared statement, ensuring user input never parses as executable SQL syntax.

What SQL operations are blocked in DB-GPT GBI?

DB-GPT explicitly blocks DROP, TRUNCATE, GRANT, and administrative commands while maintaining a whitelist of SELECT, CREATE TABLE, INSERT INTO, UPDATE, DELETE FROM, and ALTER TABLE. For DuckDB specifically, the system additionally blocks COPY, EXPORT, IMPORT, INSTALL, and PRAGMA statements to prevent file system access and configuration tampering.

Does DB-GPT require API keys for GBI endpoints?

API keys are required when the server configuration defines an api_keys list. In this mode, the check_api_key dependency validates Bearer tokens against the configured list, rejecting unauthorized requests with HTTP 401. If no API keys are configured, endpoints remain open, though this mode is intended only for development environments.

How does DB-GPT prevent denial-of-service attacks from long-running queries?

The framework enforces a mandatory timeout parameter (defaulting to 30 seconds) on every query_ex call in packages/dbgpt-core/src/dbgpt/datasource/rdbms/base.py. Database connectors abort queries exceeding this threshold, preventing attackers from locking resources or exhausting server capacity through computationally expensive SQL constructions.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}