# PYTHON-FLASK-SEC-003: Flask SQL Injection via Tainted String

> **Severity:** CRITICAL | **CWE:** CWE-89 | **OWASP:** A03:2021

- **Language:** Python
- **Category:** Flask
- **URL:** https://codepathfinder.dev/registry/python/flask/PYTHON-FLASK-SEC-003
- **Detection:** `pathfinder scan --ruleset python/PYTHON-FLASK-SEC-003 --project .`

## Description

This rule catches a specific pattern: user input from Flask request parameters
(request.args, request.form, request.get_json) gets concatenated or f-stringed
into a SQL query, then passed to cursor.execute(). The fix is straightforward --
use parameterized queries instead of string building.

What makes this rule different from a simple grep for "execute" is that it actually
traces data flow. If the user input arrives in app.py, gets assigned to a variable,
passed through a function call into db.py, and ends up in cursor.execute() three
hops later -- Code Pathfinder follows that chain and flags it. It also knows the
difference between cursor.execute(query_string) (dangerous, argument position 0)
and cursor.execute(sql, (params,)) (safe, the tuple at position 1 is the
parameterization mechanism). That distinction alone eliminates a class of false
positives that trips up most static analysis tools.


## Vulnerable Code

```python
# --- file: app.py ---
from flask import Flask, request
from db import query_user

app = Flask(__name__)


@app.route('/user')
def get_user():
    username = request.args.get('username')
    result = query_user(username)
    return str(result)

# --- file: db.py ---
import sqlite3


def get_connection():
    return sqlite3.connect('app.db')


def query_user(name):
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users WHERE name = '" + name + "'")
    return cursor.fetchall()
```

## Secure Code

```python
from flask import Flask, request
import sqlite3

app = Flask(__name__)

@app.route('/users/search')
def search_users():
    username = request.args.get('username')
    conn = sqlite3.connect('app.db')
    cursor = conn.cursor()
    # SAFE: Parameterized query -- the database driver handles escaping
    cursor.execute("SELECT * FROM users WHERE name = ?", (username,))
    return {'results': cursor.fetchall()}

# Or just use SQLAlchemy
from sqlalchemy import select
result = db.session.execute(select(User).filter_by(name=username))

```

## Detection Rule (Python SDK)

```python
from rules.python_decorators import python_rule
from codepathfinder import calls, flows, QueryType
from codepathfinder.presets import PropagationPresets


class FlaskRequest(QueryType):
    fqns = ["flask"]
    patterns = ["*request"]


class DBCursor(QueryType):
    fqns = ["sqlite3.Cursor", "mysql.connector.cursor.MySQLCursor",
            "psycopg2.extensions.cursor", "pymysql.cursors.Cursor"]
    patterns = ["*Cursor"]
    match_subclasses = True


@python_rule(
    id="PYTHON-FLASK-SEC-003",
    name="Flask SQL Injection via Tainted String",
    severity="CRITICAL",
    category="flask",
    cwe="CWE-89",
    tags="python,flask,sql-injection,database,OWASP-A03,CWE-89",
    message="User input flows to SQL execution without parameterization. Use parameterized queries.",
    owasp="A03:2021",
)
def detect_flask_sql_injection():
    """Detects Flask request data flowing to SQL execution."""
    return flows(
        from_sources=[
            FlaskRequest.method("get", "args", "form", "values",
                                "get_json", "cookies", "headers"),
        ],
        to_sinks=[
            DBCursor.method("execute", "executemany").tracks(0),
        ],
        sanitized_by=[
            calls("escape"),
            calls("escape_string"),
        ],
        propagates_through=PropagationPresets.standard(),
        scope="global",
    )
```

## How to Fix

- Use parameterized queries -- cursor.execute("SELECT * FROM users WHERE name = ?", (username,)) -- not string concatenation
- Switch to an ORM like SQLAlchemy if you're writing raw SQL frequently
- Validate and type-check input before it reaches the database layer
- Run database accounts with least privilege so a compromised query can't DROP tables
- Log SQL queries in production to catch injection attempts early

## Security Implications

- **Data Exfiltration:** An attacker can read your entire database. UNION-based injection pulls data
from other tables. Blind injection extracts it one bit at a time. Either way,
one vulnerable endpoint is enough to leak every row in every table.

- **Authentication Bypass:** The classic ' OR 1=1 -- in a login form. The query returns all users, the app
logs in as the first one (usually admin). No brute force needed, no credentials
required.

- **Data Manipulation:** Injected INSERT, UPDATE, or DELETE statements can modify records, create backdoor
accounts, or corrupt data. In financial applications, this means altered
transactions.

- **Remote Code Execution:** Some databases let you run OS commands: PostgreSQL has COPY, MySQL has
INTO OUTFILE, MSSQL has xp_cmdshell. SQL injection on these systems can
escalate to full server compromise.


## FAQ

**Q: How does this rule detect SQL injection across multiple Python files?**

It follows the data. If a Flask route in app.py calls request.args.get('id'),
passes that value to a helper function in db.py, and that function sticks it into
cursor.execute() via string concatenation -- Code Pathfinder traces the full chain
and flags it. Most real codebases split routes, services, and database code into
separate files. This rule handles that.


**Q: How do I run this rule in CI/CD?**

Run: pathfinder ci --ruleset python/flask/PYTHON-FLASK-SEC-003 --project .
It outputs SARIF, JSON, or CSV. If you're on GitHub, it can post inline review
comments directly on pull requests pointing to the exact lines. No dashboard
needed, no SARIF viewer -- just comments on the PR.


**Q: Does this rule flag parameterized queries as vulnerable?**

No. The .tracks(0) setting tells the engine that only the first argument to
cursor.execute() matters -- that's the SQL string. The second argument is the
parameter tuple (the safe one). So cursor.execute(sql, (user_input,)) won't
trigger a finding even though user_input is tainted. This is the most common
false positive in SQL injection scanning and this rule avoids it.


**Q: Which database drivers does this rule cover?**

SQLite (sqlite3), MySQL (mysql-connector-python, PyMySQL), and PostgreSQL
(psycopg2). The QueryType pattern *Cursor also catches cursor objects from
other drivers that follow the DB-API 2.0 convention. For Django ORM-specific
patterns like .raw() and .extra(), see PYTHON-DJANGO-SEC-002 and SEC-003.


**Q: What compliance standards require SQL injection testing?**

PCI DSS v4.0 (Requirement 6.2.4), OWASP Top 10 (A03:2021), NIST SP 800-53
(SI-10), SOC 2 Type II (CC6.1), ISO 27001 (A.14.2.5), and the EU Cyber
Resilience Act all require or recommend it. CWE-89 is #3 in the 2023 CWE Top 25.
Running this rule in CI gives you auditable evidence for compliance reviews.


**Q: Can I add my own sources or sinks to this rule?**

Yes. The rule file is plain Python. Fork it and add your custom input methods to
from_sources, your database functions to to_sinks, or your validation logic to
sanitized_by. You can version-control the modified rule alongside your app code.


**Q: How is this different from grep-based SQL injection detection?**

Grep finds the word "execute" in your codebase. Code Pathfinder determines whether
user-controlled data actually reaches that execute call. It follows variables through
assignments, function calls, and file boundaries. It knows that escaped input is safe
and that parameterized queries aren't vulnerable. Grep can't do any of that -- it
either misses real bugs or drowns you in false positives.


## References

- [CWE-89: SQL Injection](https://cwe.mitre.org/data/definitions/89.html)
- [OWASP SQL Injection](https://owasp.org/www-community/attacks/SQL_Injection)
- [OWASP SQL Injection Prevention Cheat Sheet](https://cheatsheetseries.owasp.org/cheatsheets/SQL_Injection_Prevention_Cheat_Sheet.html)
- [Python DB-API 2.0 Parameterized Queries](https://peps.python.org/pep-0249/#paramstyle)
- [Flask-SQLAlchemy Documentation](https://flask-sqlalchemy.palletsprojects.com/)
- [SQLite3 Parameterized Queries](https://docs.python.org/3/library/sqlite3.html#sqlite3-placeholders)

---

Source: https://codepathfinder.dev/registry/python/flask/PYTHON-FLASK-SEC-003
Code Pathfinder — Open source, type-aware SAST with cross-file dataflow analysis