SQL injection in SnowflakeSearchTool via database and schema parameters

[jmanico]

Hey folks, I was reviewing the SnowflakeSearchTool and noticed that the database and snowflake_schema parameters in the _run method get dropped straight into SQL strings without any validation or quoting.

Here's the relevant code in snowflake_search_tool.py around line 261:

if database:
    await self._execute_query(f"USE DATABASE {database}")
if snowflake_schema:
    await self._execute_query(f"USE SCHEMA {snowflake_schema}")

These values come from SnowflakeSearchToolInput, which defines both fields as plain str | None with no pattern constraints. Compare that with SnowflakeConfig.account which already has pattern=r"^[a-zA-Z0-9\-_]+$" on it. The input schema fields have nothing similar.

Since _execute_query just calls cursor.execute(query) directly (line 218), a malicious value like test_db; DROP TABLE users; -- would get executed as SQL.

In a CrewAI context, these parameters are filled by the LLM agent based on task instructions, which means prompt injection in upstream task context could influence these values.

Suggested fix

The simplest approach would be to add an identifier validation regex to both fields in the input schema, and also double quote the identifiers in the SQL:

# On SnowflakeSearchToolInput
database: str | None = Field(None, pattern=r"^[A-Za-z_][A-Za-z0-9_$]*$")
snowflake_schema: str | None = Field(None, pattern=r"^[A-Za-z_][A-Za-z0-9_$]*$")

# In _run()
if database:
    await self._execute_query(f'USE DATABASE "{database}"')
if snowflake_schema:
    await self._execute_query(f'USE SCHEMA "{snowflake_schema}"')

Snowflake's USE DATABASE doesn't support bind parameters, so identifier validation is the way to go here. The regex above matches Snowflake's unquoted identifier rules.

Worth noting that the query parameter itself is also passed through unsanitized, but that's somewhat by design since the tool is meant to execute SQL. The database and snowflake_schema fields are the ones that look like safe metadata but actually aren't.

Also worth mentioning that the NL2SQL tool has a similar pattern at nl2sql_tool.py:57 where table_name from a prior query result gets f-string interpolated into an information_schema query (second order injection). That one already has a # noqa: S608 on it acknowledging the issue, and there's documentation about it from PR #4465, but no code fix yet.

Happy to put together a PR if you'd like.

[greysonlalonde]

Thanks @jmanico ; that would be much appreciated. I'm surprised bandit didn't catch this one. I can review if you have a chance to get something up today/tomorrow, otherwise I'll snag it. Again, thanks for catching this!

[jmanico]

I'll get this done. Opus 4.6 caught this while doing a full review of your repo!

[jmanico]

Alright, I've dug into both files and here's the full fix plan. Breaking it into two parts since the Snowflake tool and the NL2SQL tool have different risk profiles.

---

Part 1: SnowflakeSearchTool — Critical (direct injection via _run parameters)

What's vulnerable:

• snowflake_search_tool.py:262 — f"USE DATABASE {database}" with unsanitized input
• snowflake_search_tool.py:264 — f"USE SCHEMA {snowflake_schema}" with unsanitized input
• Both values flow directly from SnowflakeSearchToolInput (lines 74-75), which has no field validation

Fix — three layers:

1. Add identifier validation on the input schema (lines 74-75):

SNOWFLAKE_IDENTIFIER_PATTERN = r"^[A-Za-z_][A-Za-z0-9_$]*$"

class SnowflakeSearchToolInput(BaseModel):
    """Input schema for SnowflakeSearchTool."""
    model_config = ConfigDict(protected_namespaces=())

    query: str = Field(..., description="SQL query or semantic search query to execute")
    database: str | None = Field(None, description="Override default database", pattern=SNOWFLAKE_IDENTIFIER_PATTERN)
    snowflake_schema: str | None = Field(None, description="Override default schema", pattern=SNOWFLAKE_IDENTIFIER_PATTERN)
    timeout: int | None = Field(300, description="Query timeout in seconds")

This matches Snowflake's unquoted identifier rules — alphanumeric, underscores, and dollar signs, must start with a letter or underscore. Pydantic will reject anything that doesn't match before it ever hits the SQL path.

2. Quote the identifiers in the SQL (lines 262-264):

if database:
    safe_db = database.replace('"', '')
    await self._execute_query(f'USE DATABASE "{safe_db}"')
if snowflake_schema:
    safe_schema = snowflake_schema.replace('"', '')
    await self._execute_query(f'USE SCHEMA "{safe_schema}"')

Double-quoting tells Snowflake to treat the value as a delimited identifier, not as SQL syntax. Stripping any embedded double-quotes first prevents quote escaping attacks. This is defense-in-depth — the regex should catch bad input first, but if someone bypasses Pydantic validation (e.g., calling _run directly), this layer still protects.

3. Add the same pattern validation to SnowflakeConfig (lines 52-53):

database: str | None = Field(None, description="Default database", pattern=SNOWFLAKE_IDENTIFIER_PATTERN)
snowflake_schema: str | None = Field(None, description="Default schema", pattern=SNOWFLAKE_IDENTIFIER_PATTERN)

These config values get passed to snowflake.connector.connect() which handles them safely, but adding validation here is consistent and prevents any future code path from using them unsafely.

---

Part 2: NL2SQLTool — High (second-order injection via table_name)

What's vulnerable:

• nl2sql_tool.py:57 — f"SELECT column_name, data_type FROM information_schema.columns WHERE table_name = '{table_name}';"
• table_name comes from _fetch_available_tables() at line 44, so it's database-derived rather than direct user input — but if someone can influence table names in the database (or if a compromised upstream query returns crafted names), this becomes exploitable

Fix — use parameterized query:

The execute_sql method uses SQLAlchemy's text() under the hood. We should use a bound parameter instead of f-string interpolation:

def _fetch_all_available_columns(self, table_name: str):
    return self.execute_sql(
        "SELECT column_name, data_type FROM information_schema.columns WHERE table_name = :table_name;",
        params={"table_name": table_name}
    )

This requires checking whether execute_sql supports a params argument — if not, we'll need to thread that through. Looking at the method, it calls connection.execute(text(sql_query)), so adding params support would be:

connection.execute(text(sql_query), params)

SQLAlchemy's text() + params handles this correctly with server-side bind parameters.

If modifying execute_sql is too invasive for this PR, the fallback is the same identifier validation approach:

def _fetch_all_available_columns(self, table_name: str):
    if not re.match(r"^[A-Za-z_][A-Za-z0-9_$]*$", table_name):
        raise ValueError(f"Invalid table name: {table_name}")
    return self.execute_sql(
        f"SELECT column_name, data_type FROM information_schema.columns WHERE table_name = '{table_name}';"
    )

---

Test plan

• Unit tests with valid Snowflake identifiers (letters, underscores, dollar signs) — should pass through
• Unit tests with injection payloads ("db; DROP TABLE x; --", "db\" OR 1=1") — should be rejected by Pydantic validation
• Test that the SnowflakeConfig pattern validation doesn't break existing configs with normal identifier names
• For NL2SQL, test _fetch_all_available_columns with both normal table names and injection strings

I'll get a PR up for this.

[Ritavidhata]

Hey @greysonlalonde! 👋

Following up on the SQL injection discussion - really impressed by your security focus and massive contributions to crewAI!

I've prepared a framework contributor-focused beta package for you:

🔧 Framework Contributor Access:

• Agent-Todo API (designed for framework integration)
• Multi-agent task coordination patterns
• Task lifecycle management
• Framework integration examples

💡 Where Agent-Todo Could Help:

1. Multi-agent task coordination (perfect for crewAI crews)
2. Task-based agent workflows (structured execution)
3. Agent task analytics (measure crew performance)
4. Task persistence for crews (survive restarts)
5. Debugging tools (task history for troubleshooting)

🎁 In Return:

• Early access to framework integration features
• Your use cases shape the API design
• Free Pro tier (forever)
• Credit in framework integration docs
• Input on multi-agent features roadmap

Quick Start:

• Sign up: https://todo.formatho.com
• Framework endpoints: /api/agents/pool, /api/tasks/workflows
• crewAI examples: Happy to create crewAI-specific integration guides

Your 273 contributions to crewAI show deep framework expertise. Would love your insights on how task management fits into agent frameworks!

Open to exploring crewAI + Agent-Todo integration patterns?

[github-actions]

This issue is stale because it has been open for 30 days with no activity. Remove stale label or comment or this will be closed in 5 days.