Injection Flaws in Fastapi

Injection flaws in FastAPI applications typically occur when untrusted data is sent to an interpreter as part of a command or query. FastAPI's async nature and dependency injection system create specific injection vectors that developers must guard against.

The most common injection flaw in FastAPI is SQL injection through database operations. Consider this vulnerable endpoint:

@app.get("/users/")
async def get_user(username: str):
    query = f"SELECT * FROM users WHERE username = '{username}'"
    results = await database.fetch(query)
    return results

An attacker can exploit this with username=admin' OR '1'='1, causing the query to return all users. FastAPI's automatic query parameter parsing doesn't prevent this—it's the query construction that's vulnerable.

Another FastAPI-specific injection vector is through dependency injection. When dependencies accept raw input without validation:

async def get_user_by_id(user_id: str = Depends()):
    query = f"SELECT * FROM users WHERE id = {user_id}"
    return await database.fetch(query)

Even with type hints, if the dependency doesn't validate that user_id is an integer, injection remains possible.

Command injection can occur when FastAPI endpoints execute shell commands with user input:

@app.post("/execute/")
async def execute_command(command: str):
    result = await asyncio.create_subprocess_shell(
        f"echo {command} | /usr/bin/process",
        stdout=asyncio.subprocess.PIPE
    )
    return await result.stdout.read()

Here, an attacker could inject additional commands through the command parameter.

FastAPI's background tasks can also introduce injection risks when they execute operations based on user input without proper sanitization:

@app.post("/background-task/")
async def create_task(data: dict):
    await app.state.background_tasks.add_task(
        process_data, data['query']
    )
    return {"status": "queued"}

If process_data constructs database queries or executes commands using the data['query'] value directly, injection becomes possible.

Detecting injection flaws in FastAPI requires both static analysis and dynamic testing. Static analysis tools can examine your codebase for dangerous patterns, but dynamic testing is essential for finding runtime vulnerabilities.

For SQL injection detection, look for these patterns in your FastAPI code:

# Vulnerable patterns
query = f"SELECT * FROM table WHERE id = {user_input}"
query = "SELECT * FROM table WHERE name = '" + user_input + "'"
query = "SELECT * FROM table WHERE name = %s" % user_input

Safe patterns use parameterized queries:

# Safe patterns
query = "SELECT * FROM table WHERE id = $1"
results = await database.fetch(query, user_input)

# Using asyncpg directly
query = "SELECT * FROM table WHERE name = $1"
results = await conn.fetch(query, user_input)

middleBrick's dynamic scanning specifically tests FastAPI endpoints for injection vulnerabilities by sending crafted payloads to all parameters and examining responses for signs of successful exploitation. The scanner tests:

• SQL injection payloads across all query and path parameters
• Command injection attempts through text parameters
• Template injection in endpoints that might render user input
• XPath injection in XML processing endpoints

For example, middleBrick would test your FastAPI endpoint with payloads like:

# Test payloads sent by middleBrick
payloads = [
    "' OR '1'='1",
    "'; DROP TABLE users; --",
    "",
    "../../../../etc/passwd",
    "$(whoami)",
    "“ OR 1=1 -- -"
]

The scanner analyzes HTTP responses for indicators like database error messages, unexpected data exposure, or changes in response structure that suggest successful injection.

middleBrick's OpenAPI analysis also examines your FastAPI's auto-generated spec to understand parameter types and automatically generate appropriate injection test cases for each endpoint's expected input format.

FastAPI provides several native mechanisms to prevent injection flaws. The most critical is using parameterized queries with your database driver:

from databases import Database
from fastapi import FastAPI

app = FastAPI()

async def get_user_by_username(username: str):
    query = "SELECT * FROM users WHERE username = $1"
    return await database.fetch_one(query, (username,))

For SQLAlchemy with FastAPI:

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

async def get_user_by_username(username: str, session: AsyncSession):
    stmt = select(User).where(User.username == username)
    result = await session.execute(stmt)
    return result.scalar_one_or_none()

FastAPI's dependency injection system can enforce input validation before database operations:

from pydantic import BaseModel
from fastapi import Depends

class UserIdParams(BaseModel):
    user_id: int

async def validate_user_id(params: UserIdParams = Depends()):
    # Pydantic automatically validates user_id is an integer
    return params.user_id

For command injection prevention, use the asyncio.create_subprocess_exec function instead of create_subprocess_shell, which avoids shell interpretation:

@app.post("/safe-execute/")
async def safe_execute(command: str):
    # Split command into arguments - no shell interpretation
    proc = await asyncio.create_subprocess_exec(
        "/usr/bin/process", command,
        stdout=asyncio.subprocess.PIPE
    )
    result = await proc.stdout.read()
    return result

FastAPI's Pydantic models provide automatic input validation that prevents many injection vectors:

from pydantic import Field, constr

class UserQuery(BaseModel):
    username: constr(min_length=3, max_length=50, strip_whitespace=True)
    email: constr(regex=r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$")

For NoSQL injection prevention in FastAPI applications using MongoDB:

from motor.motor_asyncio import AsyncIOMotorClient

async def get_user(username: str):
    # Use $regex with escaped input instead of raw queries
    pattern = re.compile(re.escape(username), re.IGNORECASE)
    return await db.users.find_one({"username": {"$regex": pattern}})

middleBrick's CLI tool can be integrated into your FastAPI development workflow to continuously scan for injection vulnerabilities:

# Scan your FastAPI app before deployment
middlebrick scan http://localhost:8000/docs

# Add to your CI/CD pipeline
middlebrick scan --threshold B http://staging-api.example.com/docs

The scanner provides specific findings with the exact parameters and payloads that triggered vulnerabilities, making remediation straightforward.