Command injection in FastAPI

How it arises in FastAPI

The async subprocess API has two entry points. asyncio.create_subprocess_exec(program, *args) starts a program with separate arguments. asyncio.create_subprocess_shell(cmd) runs a shell command string. The second is one call shorter when you want a pipeline or a redirect, and the Python documentation attaches a warning to it: it is the application's responsibility to quote whitespace and special characters to avoid shell injection.

Background tasks. Media and document services accept an upload, return 202, and convert the file in BackgroundTasks.add_task() or a worker. UploadFile.filename is whatever the client sent. The command is built after the response, in a function no request handler test exercises.

Validated is not the same as safe. A Pydantic model guarantees that host is a string and perhaps that it is under 255 characters. It says nothing about ; or a leading -. Because the handler signature looks strict, reviewers relax.

Service-to-service calls. FastAPI is popular for internal APIs. "Only our other services call this" is true until one of them forwards a user's value unchanged.

Vulnerable and fixed

A diagnostics endpoint for a network team:

import asyncio

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Probe(BaseModel):
    host: str
    count: int = 3


@app.post("/diag/ping")
async def ping(probe: Probe):
    proc = await asyncio.create_subprocess_shell(
        f"ping -c {probe.count} {probe.host}",
        stdout=asyncio.subprocess.PIPE,
    )
    out, _ = await proc.communicate()
    return {"output": out.decode()}

count is an int, so only host is dangerous. The fix uses the exec form, constrains the field with a pattern that cannot start with a dash, bounds the count, and separates options from the operand:

import asyncio

from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI()


class Probe(BaseModel):
    host: str = Field(pattern=r"^[A-Za-z0-9][A-Za-z0-9.-]{0,252}$")
    count: int = Field(default=3, ge=1, le=10)


@app.post("/diag/ping")
async def ping(probe: Probe):
    proc = await asyncio.create_subprocess_exec(
        "ping", "-c", str(probe.count), "--", probe.host,
        stdout=asyncio.subprocess.PIPE,
    )
    out, _ = await proc.communicate()
    return {"output": out.decode()}

What FastAPI and Python already do for you

Type validation removes whole parameters from the problem: an int, an Enum or an IPvAnyAddress field cannot carry shell syntax. Field(pattern=...) puts the allowlist in the model, where the generated OpenAPI schema documents it for callers. On the Python side, the exec form passes arguments without any shell, and shlex.quote() is documented for the cases where a shell string cannot be avoided.

What es-python adds

Async code is where following a value by reading gets hard: the request is parsed in one coroutine, queued, and used in another after several awaits. es-python follows the data itself, so the hop through add_task() or an asyncio.Queue does not lose it. At the moment a process is created with request data in its command, it records the event with the trace of the coroutine that did it.

What you see for the vulnerable endpoint is a subprocess shell event: a shell is about to run a string that contains bytes from the network. It is graded Critical and tagged command injection in the events list. After the fix, the service still runs ping with a request value, so a base subprocess exec event remains without the shell label. If the host began with a dash you would also get argument injection; the pattern above rules that out, and the absence of that label on real traffic is the confirmation.

With the Block network code execution template, or any Raise rule on the shell labels, the coroutine gets a RuntimeError at the call and the shell never starts; your exception handler turns that into a 500 like any other error.

Review checklist

1. Search for create_subprocess_shell, shell=True and os.system.
2. Give every string field that reaches a command a pattern.
3. Review background task functions and workers, not only handlers.
4. Add -- before operands.
5. Do not treat internal callers as a validation layer.

References

• Reference: Python documentation: asyncio Subprocesses (checked 2026-10-12)
• Reference: FastAPI documentation: Background Tasks (checked 2026-10-12)
• Reference: FastAPI documentation: Request Files (checked 2026-10-12)
• Reference: Pydantic documentation: Fields (checked 2026-10-12)