Fix SQL injection in Python and Flask

1. Find every execute() call whose SQL is built with an f-string, %, .format() or + from request data (request.args, request.form, request.json), including in helper modules.
2. Make the SQL a constant with ? (qmark) or :name (named) placeholders and pass the values as a tuple or dict in the second argument.
3. For the sort column and direction, look the request value up in a dict of fixed SQL fragments; raise ValueError for anything else and turn it into a 400 in the view.
4. Add regression tests with this card's payloads next to functional tests for names such as O'Brien and non-ASCII text.
5. Keep debug mode off in production: the debugger page shows the database error and traceback to whoever triggered it.

The three implementations share one API and one test suite; only the code below changes. The tests load the one that IMPL names.

def orders_for_customer(db: sqlite3.Connection, customer_name: str, status: str) -> list[dict]:
    sql = f"""
        SELECT o.id, o.status, o.total_cents, o.created_at
        FROM orders AS o
        JOIN customers AS c ON c.id = o.customer_id
        WHERE c.name = '{customer_name}' AND o.status = '{status}'
        ORDER BY o.id"""
    return [dict(row) for row in db.execute(sql)]

def orders_for_customer(db: sqlite3.Connection, customer_name: str, status: str) -> list[dict]:
    sql = """
        SELECT o.id, o.status, o.total_cents, o.created_at
        FROM orders AS o
        JOIN customers AS c ON c.id = o.customer_id
        WHERE c.name = ? AND o.status = ?
        ORDER BY o.id"""
    return [dict(row) for row in db.execute(sql, (customer_name, status))]

# A placeholder can only stand in for a value. Column names and ASC/DESC are
# part of the SQL text, so map the caller's choice onto fixed fragments.
SORT_COLUMNS = {
    "created_at": "o.created_at",
    "total_cents": "o.total_cents",
    "customer": "c.name",
}
SORT_DIRECTIONS = {"asc": "ASC", "desc": "DESC"}


def search_orders(db: sqlite3.Connection, status: str, sort: str, direction: str) -> list[dict]:
    column = SORT_COLUMNS.get(sort)
    order = SORT_DIRECTIONS.get(direction)
    if column is None or order is None:
        raise ValueError("unsupported sort column or direction")
    sql = f"""
        SELECT o.id, c.name AS customer, o.status, o.total_cents, o.created_at
        FROM orders AS o
        JOIN customers AS c ON c.id = o.customer_id
        WHERE o.status = ?
        ORDER BY {column} {order}, o.id"""
    return [dict(row) for row in db.execute(sql, (status,))]

def _escape_sql_string(value: str) -> str:
    return value.replace("\\", "\\\\").replace("'", "\\'")


def orders_for_customer(db: sqlite3.Connection, customer_name: str, status: str) -> list[dict]:
    sql = f"""
        SELECT o.id, o.status, o.total_cents, o.created_at
        FROM orders AS o
        JOIN customers AS c ON c.id = o.customer_id
        WHERE c.name = '{_escape_sql_string(customer_name)}' AND o.status = '{_escape_sql_string(status)}'
        ORDER BY o.id"""
    return [dict(row) for row in db.execute(sql)]

The exploit tests send the attack and check that it fails; functional tests next to them keep ordinary requests working, so a fix that rejects everything fails too.

# Exploit: each payload must stay data and leave the query's meaning unchanged.

def test_exploit_tautology_does_not_return_other_customers_orders(client):
    assert order_ids(client, "/orders", customer="' OR '1'='1", status="shipped") == []


def test_exploit_union_cannot_read_another_customers_orders(client):
    customer = "' UNION SELECT id, status, total_cents, created_at FROM orders WHERE customer_id = 2 --"
    assert order_ids(client, "/orders", customer=customer, status="shipped") == []


def test_exploit_comment_cannot_cut_off_the_status_filter(client):
    assert order_ids(client, "/orders", customer="Alice Martin'--", status="shipped") == []


def test_exploit_sort_expression_is_rejected_not_used_as_an_oracle(client):
    # Row order would reveal whether each customer's hidden email starts with "b".
    sort = "CASE WHEN substr(c.email, 1, 1) = char(98) THEN 0 ELSE 1 END"
    res = client.get("/admin/orders", query_string={"status": "shipped", "sort": sort, "dir": "asc"})
    assert res.status_code == 400


def test_exploit_direction_cannot_add_a_second_sort_expression(client):
    direction = "ASC, CASE WHEN substr(c.email, 1, 1) = char(98) THEN 0 ELSE 1 END"
    res = client.get("/admin/orders", query_string={"status": "shipped", "sort": "created_at", "dir": direction})
    assert res.status_code == 400

• executescript() accepts no parameters; never pass it a string that contains request data.
• execute() refuses more than one statement (ProgrammingError), which stops stacked queries but not UNION, boolean or comment injection. It is not a defence.
• SQLAlchemy: write text("... WHERE name = :name") and pass {"name": name} to execute(); an f-string inside text() is still injection.
• LIKE: a bound % still matches every row. Escape \, % and _ in the value and add ESCAPE '\' to the SQL.
• IN lists: sqlite3 cannot bind a list. Build the placeholders from the count, ", ".join("?" for _ in ids), never from the values, or bind a JSON array and use IN (SELECT value FROM json_each(?)).
• ORDER BY ? binds a constant, so SQLite silently skips the sort. Identifiers need the allowlist.
• Allowlist only columns the caller may see: sorting by a hidden column such as email still leaks its order.
• Bandit B608 cannot tell an allowlisted fragment from user input, so it also flags the fixed ORDER BY. Keep the check on and justify that one finding.

1. Run the tests in all three modes: IMPL=fixed must pass; IMPL=vulnerable and IMPL=wrong must fail (node ../../run-all.mjs python-flask does this for you).
2. Run semgrep --config r/python.lang.security.audit.formatted-sql-query.formatted-sql-query impl/ and bandit -t B608 -r impl/; expect findings only in vulnerable.py and wrong.py, plus Bandit's LOW-confidence note on the fixed ORDER BY.
3. Against your own endpoint, send customer=' OR '1'='1 and expect an empty result, and send a CASE expression as the sort and expect 400.

What each rule reported when it was run on this card's code, or what its source shows it covers where it was not run. A rule that exists for the weakness does not always catch it.

Download the project, install its dependencies once (this step needs the network), then run the tests against each implementation from the sql-injection/python-flaskfolder.

python3 -m venv .venv
.venv/bin/python -m pip install -r requirements.txt

# Expected: every test passes
IMPL=fixed .venv/bin/python -m pytest

# Expected: an exploit test fails
IMPL=vulnerable .venv/bin/python -m pytest

# Expected: a test fails
IMPL=wrong .venv/bin/python -m pytest

Each run writes a JUnit report to reports/junit.xml. From the top folder of the download, node run-all.mjs sql-injection/python-flask runs all three modes and checks the expected results.

Download the project ZIP, 17 KB

Verified

Machine

macOS 27.0 (Darwin 27.0.0, arm64)

Toolchain

• python CPython 3.14.8
• flask 3.1.3

Matrix

fixed: 11 of 11 tests pass; vulnerable: 7 of 11 tests fail (5 exploit tests); wrong: 6 of 11 tests fail (4 exploit tests)

Test results

fixed 11/11 pass; vulnerable 7/11 fail (5 exploit tests); wrong 6/11 fail (4 exploit tests), offline after install

Versions and results hold for the toolchain above. The card's tests are the check: run them with your own versions before relying on the fix.

References

1. OWASP SQL Injection Prevention Cheat Sheet
2. OWASP Query Parameterization Cheat Sheet
3. Python sqlite3: how to use placeholders to bind values in SQL queries
4. Flask: Using SQLite 3 with Flask
5. Flask: Testing Flask Applications
6. SQLAlchemy 2.0 tutorial: sending parameters
7. SQLite: SQL language expressions (string literals, parameters, LIKE ... ESCAPE)
8. SQLite: json_each() table-valued function
9. Bandit B608: hardcoded_sql_expressions

Attack patterns:

• CAPEC-66 at MITRE