Code injection in FastAPI

How it arises in FastAPI

ImportString on a request model. Pydantic provides ImportString, a type that expects a string and loads the Python object importable at that dotted path; math:cos becomes the function cos. Pydantic's documentation notes that the pointer is evaluated and imported on model instantiation. It is designed for settings. Put on a model that FastAPI builds from a request body, validation itself performs an import the client chose, before the first line of your handler.

Plugin loading by name. /run/{task} followed by importlib.import_module(f"tasks.{task}"). The prefix feels like a fence. It is not much of one: any module under that package is importable, and the import runs its top-level code.

Expression parameters. Data APIs accept a filter expression and hand it to eval() because the result should be "a Python condition". The parameter is declared expr: str, which FastAPI validates successfully for any text.

Callables from JSON. A scheduling endpoint stores {"func": "module.name", "args": [...]} and a worker resolves and calls it later. The import happens in another process, long after the request.

Vulnerable and fixed

import importlib

from fastapi import FastAPI
from pydantic import BaseModel, ImportString

app = FastAPI()


class Job(BaseModel):
    transform: ImportString
    payload: dict


@app.post("/jobs")
def create_job(job: Job):
    return {"result": job.transform(job.payload)}


@app.post("/run/{task}")
def run(task: str, payload: dict):
    module = importlib.import_module(f"tasks.{task}")
    return module.main(payload)

Declare the choice as an Enum. FastAPI then rejects an unknown name with a 422 before the handler runs, the OpenAPI schema lists the permitted values, and the mapping from name to function is a dictionary in your source:

from enum import Enum

from fastapi import FastAPI
from pydantic import BaseModel

import tasks.resize
import tasks.summarize

app = FastAPI()


class TaskName(str, Enum):
    resize = "resize"
    summarize = "summarize"


TASKS = {TaskName.resize: tasks.resize.main, TaskName.summarize: tasks.summarize.main}


class Job(BaseModel):
    transform: TaskName
    payload: dict


@app.post("/jobs")
def create_job(job: Job):
    return {"result": TASKS[job.transform](job.payload)}


@app.post("/run/{task}")
def run(task: TaskName, payload: dict):
    return TASKS[task](payload)

What FastAPI and Pydantic already do for you

The type system is the allowlist mechanism. Enum and Literal parameters are checked by the framework, documented automatically and impossible to bypass from the client. Pydantic's strict types reject values of the wrong shape. ImportString is clearly named and documented as an import; it is safe in a settings class read from your own environment. The work is choosing the narrow type on models that parse request bodies.

What es-python adds

The import triggered by a model field does not appear in your handler, and a reviewer reading the endpoint sees a typed parameter. es-python reports the operation where it happens. An import by a name that came from the request is recorded as __import__ with the rce impact tag, and the stack trace shows that it ran during body validation. A name passed to importlib.import_module() is reported where the import system looks that name up on disk, as a file-system event with the requested module path in it.

eval, exec and compile on request text are reported under those names. For the deferred case, where a worker resolves a stored callable name, the value is no longer request data when it is read back; switch on the db_rows source so values read from the database are followed too.

With data from the network these events are graded Critical. The fixed endpoints import their task modules at startup from names written in source, so there is nothing for es-python to report: the request only ever selects a dictionary key. If you add a Raise rule on __import__ for the socket source, the vulnerable model fails validation with a RuntimeError and the module is never loaded.

Review checklist

1. Search request and response models for ImportString.
2. Search for import_module(, __import__( and eval( with non-literal arguments.
3. Replace name parameters with Enum or Literal types.
4. Import plugin modules at startup and dispatch through a dictionary.
5. Review what workers do with stored function names.

References

• Reference: Pydantic documentation: Types, ImportString (checked 2026-10-12)
• Reference: FastAPI documentation: Path Parameters, predefined values (checked 2026-10-12)
• Reference: FastAPI documentation: Body, Nested Models (checked 2026-10-12)