Fix SQL injection in Python and FastAPI

1. Find every text(), exec_driver_sql(), cursor.execute() and order_by(text(...)) whose SQL is built with an f-string, %, .format() or + from request data: path, query and body parameters of your path operations and anything the data layer they call receives.
2. Make the SQL in text() a constant with :name placeholders and pass the values as a dict to session.execute(), or replace the raw SQL with select() and column comparisons such as Customer.name == customer.
3. For the sort column and direction, look the request value up in a dict of column objects and use order(column) with asc or desc; raise ValueError for anything else and turn it into HTTPException(status_code=400).
4. Declare query parameters with types (customer: str) so FastAPI rejects a missing value with 422 before your code runs; validation does not make a string safe to paste into SQL.
5. Register an exception handler for SQLAlchemyError that logs the error and returns a generic 500, so database error text never reaches the client.
6. Add regression tests that send this card's payloads through TestClient next to functional tests for names such as O'Brien and non-ASCII text.

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(session: Session, customer_name: str, status: str) -> list[dict]:
    sql = text(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._mapping) for row in session.execute(sql)]

def orders_for_customer(session: Session, customer_name: str, status: str) -> list[dict]:
    sql = text("""
        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 AND o.status = :status
        ORDER BY o.id""")
    rows = session.execute(sql, {"customer": customer_name, "status": status})
    return [dict(row._mapping) for row in rows]

# A bound parameter can only stand in for a value. Map the caller's choice onto
# column objects and asc()/desc(); SQLAlchemy renders the identifiers itself.
SORT_COLUMNS = {
    "created_at": Order.created_at,
    "total_cents": Order.total_cents,
    "customer": Customer.name,
}
SORT_DIRECTIONS = {"asc": asc, "desc": desc}


def search_orders(session: Session, 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")
    stmt = (
        select(Order.id, Customer.name.label("customer"), Order.status, Order.total_cents, Order.created_at)
        .join(Customer, Customer.id == Order.customer_id)
        .where(Order.status == status)
        .order_by(order(column), Order.id)
    )
    return [dict(row._mapping) for row in session.execute(stmt)]

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


def orders_for_customer(session: Session, customer_name: str, status: str) -> list[dict]:
    sql = text(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._mapping) for row in session.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", params={"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", params={"status": "shipped", "sort": "created_at", "dir": direction})
    assert res.status_code == 400

• text() is not a sanitiser. It marks a string as SQL and parses :name placeholders out of it; everything an f-string put there is already part of the statement.
• order_by("nickname") raises CompileError ("Can't resolve label reference") unless the string names a label in the query, so people reach for order_by(text(sort)), which pastes the caller's expression into ORDER BY. Use the allowlist of column objects.
• literal_column(), column() with a caller-supplied name and exec_driver_sql() with a formatted string all put text into the SQL without quoting it.
• A bound parameter can only stand in for a value: ORDER BY :sort compares a constant and SQLite silently skips the sort. Identifiers need the allowlist, and the allowlist should hold only columns the caller may see: sorting by a hidden column such as email still leaks its order.
• LIKE: a bound % still matches every row. Use column.like(value, escape="\\") or contains(value, autoescape=True) when the caller's text should match literally.
• IN lists: use column.in_(values), which binds each element (an "expanding" bind parameter), instead of joining values into the SQL.
• TestClient re-raises server exceptions by default, which hides a missing exception handler in tests; check that a database error comes back as a generic 500 when raise_server_exceptions=False.
• Starlette 1.7.0's TestClient prefers httpx2 and warns (StarletteDeprecationWarning) when only httpx is installed, although FastAPI's testing tutorial still names HTTPX; this card installs httpx2.

1. Run the tests in all three modes: IMPL=fixed must pass; IMPL=vulnerable and IMPL=wrong must fail (node ../../run-all.mjs sql-injection/python-fastapi does this for you).
2. Run semgrep --config r/python.sqlalchemy.security.audit.avoid-sqlalchemy-text.avoid-sqlalchemy-text impl/ and bandit -t B608 -r impl/; expect findings only in vulnerable.py and wrong.py.
3. Against your own endpoint, send customer=' OR '1'='1 and expect an empty list, 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-fastapifolder.

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-fastapi runs all three modes and checks the expected results.

Download the project ZIP, 20 KB

Verified

Machine

macOS 27.0 (Darwin 27.0.0, arm64)

Toolchain

• python CPython 3.14.8
• fastapi 0.142.2
• starlette 1.7.0
• pydantic 2.13.5
• sqlalchemy 2.0.54
• httpx2 2.13.1
• pytest 9.1.1
• semgrep 1.180.0
• bandit 1.9.4

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

Dependency freshness

FastAPI 0.142.2 was published on 2026-09-30; 0.142.3, 0.142.4 and 0.143.0 (2026-10-07 and 2026-10-08) are OpenTelemetry fixes and changes, not security releases, so they are under the 7-day rule. The same rule kept Pydantic on 2.13.5 (2026-08-28) rather than 2.14.0 (2026-10-08) and opentelemetry-api on 1.45.0 (2026-09-25). Starlette 1.7.0 dates from 2026-09-23, SQLAlchemy 2.0.54 from 2026-09-15 (the 2.0 series; 2.1.0 came out on 2026-09-24), httpx2 2.13.1 from 2026-09-23. The test-only iniconfig is pinned at 2.3.0 because 2.3.1 is from 2026-10-06. OSV reported no known vulnerabilities for any of the 21 pinned packages.

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. SQLAlchemy 2.0 tutorial: sending parameters with text()
4. SQLAlchemy 2.0: text() and TextClause
5. SQLAlchemy 2.0 tutorial: ORDER BY, GROUP BY, HAVING
6. FastAPI: SQL (relational) databases
7. FastAPI: dependencies with yield
8. FastAPI: handling errors (exception handlers)
9. FastAPI: testing with TestClient
10. CodeQL Python model of SQLAlchemy (TextClause as SQL construction)
11. Bandit B608: hardcoded_sql_expressions

Attack patterns:

• CAPEC-66 at MITRE