Prompt Injection in Flask with Cockroachdb

Prompt injection becomes particularly concerning in a Flask application that uses CockroachDB when user-influenced input is incorporated into prompts sent to an LLM, and those prompts indirectly affect how database operations are constructed or interpreted. In this stack, a developer might build an endpoint that accepts natural language input, forwards it to an LLM to generate a SQL-like query or a filter expression, and then executes the result against CockroachDB. If the endpoint does not strictly separate intent from data, an attacker can craft input that changes the effective instruction set of the LLM, leading to unintended database queries or data exposure.

Consider a Flask route that asks an LLM to translate a user request into a CockroachDB query. The prompt template may include instructions such as: "Given the user query, produce a safe SELECT statement against the tenants table." If user input is concatenated into the prompt without isolation, an attacker can inject instructions or constraints that override the original intent. For example, a benign input like search=active could be transformed into search=active; DROP TABLE tenants; -- if the LLM is tricked into including destructive statements, or more commonly, the injection causes the LLM to ignore schema constraints and return rows it should not. CockroachDB’s SQL interface will execute whatever statement the LLM produces, so the injected instruction directly changes the query behavior, potentially bypassing row-level filters or exposing sensitive columns.

The LLM/AI Security checks in middleBrick specifically target this scenario by running active prompt injection probes against unauthenticated endpoints. It tests for system prompt extraction, instruction override, DAN jailbreak, data exfiltration, and cost exploitation patterns. When a Flask endpoint exposes an LLM call that references CockroachDB, these probes can reveal whether user input can change the system instructions or data access boundaries. Because CockroachDB often serves distributed workloads with multi-tenant schemas, unchecked prompt injection can lead to cross-tenant data access, violating isolation expectations even before any SQL injection is attempted.

Moreover, the interplay with CockroachDB’s concurrency and transaction model can amplify the impact. An injected prompt that encourages the LLM to generate speculative or batched statements may trigger unexpected transaction behaviors or error paths that leak schema details through error messages. The LLM might also be induced to request more columns or tables than necessary, and CockroachDB will return those if the generated SQL is syntactically valid. This increases the data exposure surface. middleBrick’s output scanning for PII, API keys, and executable code helps detect whether LLM responses inadvertently disclose sensitive information that should never have been surfaced to the attacker.

Remediation focuses on strict separation between system instructions, static policy, and user data. Never allow raw user input to modify the core prompt that governs LLM behavior, and never construct CockroachDB queries by string interpolation based on LLM output. Use parameterized statements and schema-aware validation to ensure the database only receives expected, bounded operations.

First, define a clear system prompt that is provided programmatically and never concatenated with user input. In Flask, this can be enforced by keeping the prompt in the application code and passing user data as structured parameters to the LLM. For example:

import os
from flask import Flask, request, jsonify
import anthropic  # or openai, together depending on your LLM provider

app = Flask(__name__)

SYSTEM_PROMPT = (
    "You are a helpful assistant that generates safe CockroachDB SELECT queries. "
    "Always restrict output to the columns and table specified in the schema. "
    "Never include DROP, DELETE, or ALTER statements."
)

@app.route("/query")
def generate_query():
    user_request = request.args.get("q", "")
    client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
    message = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=256,
        temperature=0.2,
        system=[{"role": "user", "content": SYSTEM_PROMPT}],
        messages=[{"role": "user", "content": f"User request: {user_request}"}]
    )
    generated_sql = message.content[0].text if message.content else ""
    # Further validation of generated_sql should happen before execution
    return jsonify({"generated_sql": generated_sql})

Second, validate and parameterize any SQL that reaches CockroachDB. Do not execute the LLM’s output verbatim. Instead, parse the intent, map it to allowed operations, and use placeholders for values. For example, if the LLM is expected to produce a SELECT with filters, extract the filter values and pass them as parameters:

import psycopg2
from flask import g

def get_db():
    return psycopg2.connect(
        host=os.getenv("COCKROACH_HOST"),
        port=os.getenv("COCKROACH_PORT"),
        user=os.getenv("COCKROACH_USER"),
        password=os.getenv("COCKROACH_PASSWORD"),
        database=os.getenv("COCKROACH_DB"),
        sslmode='require',
        connection_factory=CustomConnectionFactory()
    )

def safe_execute_select(table, allowed_columns, filters):
    col = [c for c in filters.keys() if c in allowed_columns]
    vals = [filters[c] for c in col]
    placeholders = ', '.join(['%s'] * len(col))
    column_list = ', '.join(col) if col else '*'
    query = f"SELECT {column_list} FROM {table} WHERE {col[0]} = %s" if col else f"SELECT * FROM {table}"
    with get_db() as conn:
        with conn.cursor() as cur:
            cur.execute(query, vals)
            return cur.fetchall()

Third, enforce tenant isolation explicitly. If your CockroachDB schema uses tenant_id columns, ensure the generated SQL includes this filter regardless of LLM output. Do not rely on the LLM to include it. This prevents prompt-induced omission of tenant constraints:

@app.route("/mydata")
def my_data():
    tenant_id = g.current_user.tenant_id
    user_request = request.args.get("q", "")
    # LLM generates column hints only, not the full WHERE clause
    columns = llm_suggest_columns(user_request)
    safe_cols = [c for c in columns if c in ALLOWED_COLUMNS]
    results = safe_execute_select(
        table="user_data",
        allowed_columns=safe_cols,
        filters={"tenant_id": tenant_id}
    )
    return jsonify(results)

Finally, integrate middleBrick to continuously validate your endpoints. Use the CLI to scan your Flask API and verify that prompt injection tests pass and that generated SQL patterns conform to your schema rules. The dashboard can track how remediation changes affect your security score over time, and the GitHub Action can prevent deployments if new endpoints introduce risky prompt handling around CockroachDB interactions.