| summary |
ormar is vulnerable to SQL Injection through aggregate functions min() and max()
# Report of SQL Injection Vulnerability in Ormar ORM
## A SQL Injection attack can be achieved by passing a crafted string to the min() or max() aggregate functions.
## Brief description
When performing aggregate queries, Ormar ORM constructs SQL expressions by passing user-supplied column names directly into `sqlalchemy.text()` without any validation or sanitization. The `min()` and `max()` methods in the `QuerySet` class accept arbitrary string input as the column parameter. While `sum()` and `avg()` are partially protected by an `is_numeric` type check that rejects non-existent fields, `min()` and `max()` skip this validation entirely. As a result, an attacker-controlled string is embedded as raw SQL inside the aggregate function call. Any unauthorized user can exploit this vulnerability to read the entire database contents, including tables unrelated to the queried model, by injecting a subquery as the column parameter.
## Affected versions
```
0.9.9 - 0.12.2
0.20.0b1 - 0.22.0 (latest)
```
The vulnerable `SelectAction.get_text_clause()` method and the `min()`/`max()` aggregate functions were introduced together in commit `ff9d412` (March 12, 2021) and first released in version **0.9.9**. The vulnerable code has never been modified since — `get_text_clause()` is identical in every subsequent version through the latest **0.21.0**.
Versions prior to 0.9.9 do not contain the `min()`/`max()` aggregate feature and are not affected.
The following uses the latest ormar 0.21.0 as an example to illustrate the attack.
## Vulnerability details
When performing an aggregate query, the `QuerySet.max()` method (line 721, `queryset.py`) passes user input to `_query_aggr_function()`. This method creates a `SelectAction` object for each column name. The column string is split by `__` and the last part becomes `self.field_name` — with no validation against the model's actual fields.
The critical vulnerability is in `SelectAction.get_text_clause()` (line 41-43, `select_action.py`), which directly passes `self.field_name` into `sqlalchemy.text()`:
```python
#select_action.py line 41-43
def get_text_clause(self) -> sqlalchemy.sql.expression.TextClause:
alias = f"{self.table_prefix}_" if self.table_prefix else ""
return sqlalchemy.text(f"{alias}{self.field_name}") # unsanitised user input!
```
The `apply_func()` method then wraps this raw text clause inside `func.max()`, producing SQL like `max(<attacker_input>)`. Since `sqlalchemy.text()` treats its argument as literal SQL, any subquery or SQL expression injected through the column name will be executed by the database engine.
The `_query_aggr_function()` method (line 704-719, `queryset.py`) only validates field types for `sum` and `avg`, leaving `min` and `max` completely unprotected:
```python
#queryset.py line 704-719
async def _query_aggr_function(self, func_name: str, columns: List) -> Any:
func = getattr(sqlalchemy.func, func_name)
select_actions = [
SelectAction(select_str=column, model_cls=self.model) for column in columns
]
if func_name in ["sum", "avg"]: # <-- only sum/avg are checked!
if any(not x.is_numeric for x in select_actions):
raise QueryDefinitionError(...)
select_columns = [x.apply_func(func, use_label=True) for x in select_actions]
expr = self.build_select_expression().alias(f"subquery_for_{func_name}")
expr = sqlalchemy.select(*select_columns).select_from(expr)
result = await self.database.fetch_one(expr)
return dict(result) if len(result) > 1 else result[0]
```
To reproduce the attack, you can follow the steps below, using a FastAPI application with SQLite as an example. |