SQLAlchemy · SQL injection

SQL injection in SQLAlchemy

SQLAlchemy's expression language turns Python values into bound parameters automatically, so injection in SQLAlchemy code lives in the constructs whose job is to accept SQL text exactly as written.

Class
SQL injection in Python
CWE
CWE-89
database toolkit
SQLAlchemy documentation
On this page

How it arises in SQLAlchemy

select(User).where(User.name == name) never puts name in the statement. The compiler emits a placeholder and sends the value separately. The constructs below are the exceptions, and each exists for a good reason.

  • text() builds a TextClause "representing a textual SQL string directly". It supports named parameters written as :name, supplied at execution or with TextClause.bindparams(). A text() that receives an f-string has no parameters left to bind.
  • literal_column() produces a column expression that, in the documentation's words, renders exactly as stated. It is meant for SQL expressions the library has no construct for. Fed with a request value, it is a concatenation point.
  • column() stores a name that is assumed to be part of a table and may be quoted as such. Quoting is not an allowlist: the caller still chooses which column.
  • Textual fragments inside ORM queries, for example .where(text(...)), .order_by(text(...)) and .group_by(text(...)). These mix freely with safe constructs, so a query can be 95% expression language and still injectable in its last clause.
  • Connection.exec_driver_sql(), which hands a string straight to the DBAPI driver.

Dynamic filtering is the usual trigger. An endpoint accepts a column name and an operator, and the shortest implementation is a formatted string inside text().

Vulnerable and fixed

from sqlalchemy import select, text
from sqlalchemy.orm import Session

from models import Order


def search_orders(session: Session, status: str, sort: str, direction: str):
    stmt = (
        select(Order)
        .where(text(f"status = '{status}'"))
        .order_by(text(f"{sort} {direction}"))
    )
    return session.scalars(stmt).all()

Stay in the expression language. The column is fetched from a dictionary of mapped attributes, and the direction is a method call chosen in Python, so no request string is rendered into SQL:

from sqlalchemy import select
from sqlalchemy.orm import Session

from models import Order

SORTABLE = {"created": Order.created_at, "total": Order.total}


def search_orders(session: Session, status: str, sort: str, direction: str):
    column = SORTABLE.get(sort, Order.created_at)
    ordering = column.desc() if direction == "desc" else column.asc()
    stmt = select(Order).where(Order.status == status).order_by(ordering)
    return session.scalars(stmt).all()

When textual SQL is the right tool, for example a hand-tuned report, keep the string constant and pass the values: session.execute(text("SELECT ... WHERE status = :status"), {"status": status}).

What SQLAlchemy already does for you

Nearly everything. Comparison operators, in_(), like(), insert().values() and ORM attribute assignment all produce bound parameters without the developer asking. Identifiers that come from mapped classes are quoted by the dialect. text() has first-class parameter support. The textual constructs are clearly named, which makes them easy to search for in review.

What es-python adds

Review finds the text(f"...") you can see. es-python reports the ones you cannot: a fragment built three calls away, a helper that wraps literal_column(), a query builder in a dependency. It follows data read from the network, a file or another source you switch on, and when that data is part of a textual construct at execution it records a sql injection event with the sqli impact tag, the value and the Python stack.

Statements built entirely from the expression language carry their values as parameters, so they are silent. That makes the event list a short, accurate inventory: every row is a place where outside data became SQL text. Some of those rows will be deliberate and guarded by an allowlist. When EyalSec can prove the outside characters could not change the statement, the event is suppressed; otherwise you mark the call site with a rule.

Second-order flows are covered too. With the db_rows source on, a value read from one table and later formatted into a text() for another is reported, although no request carried it that day. sql injection is one of the labels the Block backend network attacks template can block.

Review checklist

  1. Search for text(, literal_column(, exec_driver_sql( and string arguments to order_by().
  2. Every text() should be a constant with :name parameters.
  3. Resolve column names through a dictionary of mapped attributes.
  4. Choose asc() or desc() in Python; do not render a direction string.
  5. Keep report SQL in constants, not in builders that accept fragments.

References

See it on your own SQLAlchemy code

es-python runs your application unchanged and reports when request data reaches the operation, with where the data came from and the line that used it. A live demo walks through it on a real application.

EyalSec Pricing Docs Security Research Vulnerabilities Security glossary About Contact עברית Login Book a live demo