--- /dev/null
+# Strict Content-Type Checking { #strict-content-type-checking }
+
+By default, **FastAPI** uses strict `Content-Type` header checking for JSON request bodies, this means that JSON requests **must** include a valid `Content-Type` header (e.g. `application/json`) in order for the body to be parsed as JSON.
+
+## CSRF Risk { #csrf-risk }
+
+This default behavior provides protection against a class of **Cross-Site Request Forgery (CSRF)** attacks in a very specific scenario.
+
+These attacks exploit the fact that browsers allow scripts to send requests without doing any CORS preflight check when they:
+
+* don't have a `Content-Type` header (e.g. using `fetch()` with a `Blob` body)
+* and don't send any authentication credentials.
+
+This type of attack is mainly relevant when:
+
+* the application is running locally (e.g. on `localhost`) or in an internal network
+* and the application doesn't have any authentication, it expects that any request from the same network can be trusted.
+
+## Example Attack { #example-attack }
+
+Imagine you build a way to run a local AI agent.
+
+It provides an API at
+
+```
+http://localhost:8000/v1/agents/multivac
+```
+
+There's also a frontend at
+
+```
+http://localhost:8000
+```
+
+/// tip
+
+Note that both have the same host.
+
+///
+
+Then using the frontend you can make the AI agent do things on your behalf.
+
+As it's running **locally**, and not in the open internet, you decide to **not have any authentication** set up, just trusting the access to the local network.
+
+Then one of your users could install it and run it locally.
+
+Then they could open a malicious website, e.g. something like
+
+```
+https://evilhackers.example.com
+```
+
+And that malicious website sends requests using `fetch()` with a `Blob` body to the local API at
+
+```
+http://localhost:8000/v1/agents/multivac
+```
+
+Even though the host of the malicious website and the local app is different, the browser won't trigger a CORS preflight request because:
+
+* It's running without any authentication, it doesn't have to send any credentials.
+* The browser thinks it's not sending JSON (because of the missing `Content-Type` header).
+
+Then the malicious website could make the local AI agent send angry messages to the user's ex-boss... or worse. 😅
+
+## Open Internet { #open-internet }
+
+If your app is in the open internet, you wouldn't "trust the network" and let anyone send privileged requests without authentication.
+
+Attackers could simply run a script to send requests to your API, no need for browser interaction, so you are probably already securing any privileged endpoints.
+
+In that case **this attack / risk doesn't apply to you**.
+
+This risk and attack is mainly relevant when the app runs on the **local network** and that is the **only assumed protection**.
+
+## Allowing Requests Without Content-Type { #allowing-requests-without-content-type }
+
+If you need to support clients that don't send a `Content-Type` header, you can disable strict checking by setting `strict_content_type=False`:
+
+{* ../../docs_src/strict_content_type/tutorial001_py310.py hl[4] *}
+
+With this setting, requests without a `Content-Type` header will have their body parsed as JSON, which is the same behavior as older versions of FastAPI.
+
+/// info
+
+This behavior and configuration was added in FastAPI 0.132.0.
+
+///
- advanced/generate-clients.md
- advanced/advanced-python-types.md
- advanced/json-base64-bytes.md
+ - advanced/strict-content-type.md
- fastapi-cli.md
- Deployment:
- deployment/index.md
--- /dev/null
+from fastapi import FastAPI
+from pydantic import BaseModel
+
+app = FastAPI(strict_content_type=False)
+
+
+class Item(BaseModel):
+ name: str
+ price: float
+
+
+@app.post("/items/")
+async def create_item(item: Item):
+ return item
"""
),
] = None,
+ strict_content_type: Annotated[
+ bool,
+ Doc(
+ """
+ Enable strict checking for request Content-Type headers.
+
+ When `True` (the default), requests with a body that do not include
+ a `Content-Type` header will **not** be parsed as JSON.
+
+ This prevents potential cross-site request forgery (CSRF) attacks
+ that exploit the browser's ability to send requests without a
+ Content-Type header, bypassing CORS preflight checks. In particular
+ applicable for apps that need to be run locally (in localhost).
+
+ When `False`, requests without a `Content-Type` header will have
+ their body parsed as JSON, which maintains compatibility with
+ certain clients that don't send `Content-Type` headers.
+
+ Read more about it in the
+ [FastAPI docs for Strict Content-Type](https://fastapi.tiangolo.com/advanced/strict-content-type/).
+ """
+ ),
+ ] = True,
**extra: Annotated[
Any,
Doc(
include_in_schema=include_in_schema,
responses=responses,
generate_unique_id_function=generate_unique_id_function,
+ strict_content_type=strict_content_type,
)
self.exception_handlers: dict[
Any, Callable[[Request, Any], Response | Awaitable[Response]]
response_model_exclude_none: bool = False,
dependency_overrides_provider: Any | None = None,
embed_body_fields: bool = False,
+ strict_content_type: bool | DefaultPlaceholder = Default(True),
) -> Callable[[Request], Coroutine[Any, Any, Response]]:
assert dependant.call is not None, "dependant.call must be a function"
is_coroutine = dependant.is_coroutine_callable
actual_response_class: type[Response] = response_class.value
else:
actual_response_class = response_class
+ if isinstance(strict_content_type, DefaultPlaceholder):
+ actual_strict_content_type: bool = strict_content_type.value
+ else:
+ actual_strict_content_type = strict_content_type
async def app(request: Request) -> Response:
response: Response | None = None
json_body: Any = Undefined
content_type_value = request.headers.get("content-type")
if not content_type_value:
- json_body = await request.json()
+ if not actual_strict_content_type:
+ json_body = await request.json()
else:
message = email.message.Message()
message["content-type"] = content_type_value
openapi_extra: dict[str, Any] | None = None,
generate_unique_id_function: Callable[["APIRoute"], str]
| DefaultPlaceholder = Default(generate_unique_id),
+ strict_content_type: bool | DefaultPlaceholder = Default(True),
) -> None:
self.path = path
self.endpoint = endpoint
self.callbacks = callbacks
self.openapi_extra = openapi_extra
self.generate_unique_id_function = generate_unique_id_function
+ self.strict_content_type = strict_content_type
self.tags = tags or []
self.responses = responses or {}
self.name = get_name(endpoint) if name is None else name
response_model_exclude_none=self.response_model_exclude_none,
dependency_overrides_provider=self.dependency_overrides_provider,
embed_body_fields=self._embed_body_fields,
+ strict_content_type=self.strict_content_type,
)
def matches(self, scope: Scope) -> tuple[Match, Scope]:
"""
),
] = Default(generate_unique_id),
+ strict_content_type: Annotated[
+ bool,
+ Doc(
+ """
+ Enable strict checking for request Content-Type headers.
+
+ When `True` (the default), requests with a body that do not include
+ a `Content-Type` header will **not** be parsed as JSON.
+
+ This prevents potential cross-site request forgery (CSRF) attacks
+ that exploit the browser's ability to send requests without a
+ Content-Type header, bypassing CORS preflight checks. In particular
+ applicable for apps that need to be run locally (in localhost).
+
+ When `False`, requests without a `Content-Type` header will have
+ their body parsed as JSON, which maintains compatibility with
+ certain clients that don't send `Content-Type` headers.
+
+ Read more about it in the
+ [FastAPI docs for Strict Content-Type](https://fastapi.tiangolo.com/advanced/strict-content-type/).
+ """
+ ),
+ ] = Default(True),
) -> None:
# Determine the lifespan context to use
if lifespan is None:
self.route_class = route_class
self.default_response_class = default_response_class
self.generate_unique_id_function = generate_unique_id_function
+ self.strict_content_type = strict_content_type
def route(
self,
openapi_extra: dict[str, Any] | None = None,
generate_unique_id_function: Callable[[APIRoute], str]
| DefaultPlaceholder = Default(generate_unique_id),
+ strict_content_type: bool | DefaultPlaceholder = Default(True),
) -> None:
route_class = route_class_override or self.route_class
responses = responses or {}
callbacks=current_callbacks,
openapi_extra=openapi_extra,
generate_unique_id_function=current_generate_unique_id,
+ strict_content_type=get_value_or_default(
+ strict_content_type, self.strict_content_type
+ ),
)
self.routes.append(route)
callbacks=current_callbacks,
openapi_extra=route.openapi_extra,
generate_unique_id_function=current_generate_unique_id,
+ strict_content_type=get_value_or_default(
+ route.strict_content_type,
+ router.strict_content_type,
+ self.strict_content_type,
+ ),
)
elif isinstance(route, routing.Route):
methods = list(route.methods or [])
--- /dev/null
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+app_default = FastAPI()
+
+
+@app_default.post("/items/")
+async def app_default_post(data: dict):
+ return data
+
+
+app_lax = FastAPI(strict_content_type=False)
+
+
+@app_lax.post("/items/")
+async def app_lax_post(data: dict):
+ return data
+
+
+client_default = TestClient(app_default)
+client_lax = TestClient(app_lax)
+
+
+def test_default_strict_rejects_no_content_type():
+ response = client_default.post("/items/", content='{"key": "value"}')
+ assert response.status_code == 422
+
+
+def test_default_strict_accepts_json_content_type():
+ response = client_default.post("/items/", json={"key": "value"})
+ assert response.status_code == 200
+ assert response.json() == {"key": "value"}
+
+
+def test_lax_accepts_no_content_type():
+ response = client_lax.post("/items/", content='{"key": "value"}')
+ assert response.status_code == 200
+ assert response.json() == {"key": "value"}
+
+
+def test_lax_accepts_json_content_type():
+ response = client_lax.post("/items/", json={"key": "value"})
+ assert response.status_code == 200
+ assert response.json() == {"key": "value"}
--- /dev/null
+from fastapi import APIRouter, FastAPI
+from fastapi.testclient import TestClient
+
+# Lax app with nested routers, inner overrides to strict
+
+app_nested = FastAPI(strict_content_type=False) # lax app
+outer_router = APIRouter(prefix="/outer") # inherits lax from app
+inner_strict = APIRouter(prefix="/strict", strict_content_type=True)
+inner_default = APIRouter(prefix="/default")
+
+
+@inner_strict.post("/items/")
+async def inner_strict_post(data: dict):
+ return data
+
+
+@inner_default.post("/items/")
+async def inner_default_post(data: dict):
+ return data
+
+
+outer_router.include_router(inner_strict)
+outer_router.include_router(inner_default)
+app_nested.include_router(outer_router)
+
+client_nested = TestClient(app_nested)
+
+
+def test_strict_inner_on_lax_app_rejects_no_content_type():
+ response = client_nested.post("/outer/strict/items/", content='{"key": "value"}')
+ assert response.status_code == 422
+
+
+def test_default_inner_inherits_lax_from_app():
+ response = client_nested.post("/outer/default/items/", content='{"key": "value"}')
+ assert response.status_code == 200
+ assert response.json() == {"key": "value"}
+
+
+def test_strict_inner_accepts_json_content_type():
+ response = client_nested.post("/outer/strict/items/", json={"key": "value"})
+ assert response.status_code == 200
+
+
+def test_default_inner_accepts_json_content_type():
+ response = client_nested.post("/outer/default/items/", json={"key": "value"})
+ assert response.status_code == 200
+
+
+# Strict app -> lax outer router -> strict inner router
+
+app_mixed = FastAPI(strict_content_type=True)
+mixed_outer = APIRouter(prefix="/outer", strict_content_type=False)
+mixed_inner = APIRouter(prefix="/inner", strict_content_type=True)
+
+
+@mixed_outer.post("/items/")
+async def mixed_outer_post(data: dict):
+ return data
+
+
+@mixed_inner.post("/items/")
+async def mixed_inner_post(data: dict):
+ return data
+
+
+mixed_outer.include_router(mixed_inner)
+app_mixed.include_router(mixed_outer)
+
+client_mixed = TestClient(app_mixed)
+
+
+def test_lax_outer_on_strict_app_accepts_no_content_type():
+ response = client_mixed.post("/outer/items/", content='{"key": "value"}')
+ assert response.status_code == 200
+ assert response.json() == {"key": "value"}
+
+
+def test_strict_inner_on_lax_outer_rejects_no_content_type():
+ response = client_mixed.post("/outer/inner/items/", content='{"key": "value"}')
+ assert response.status_code == 422
+
+
+def test_lax_outer_accepts_json_content_type():
+ response = client_mixed.post("/outer/items/", json={"key": "value"})
+ assert response.status_code == 200
+
+
+def test_strict_inner_on_lax_outer_accepts_json_content_type():
+ response = client_mixed.post("/outer/inner/items/", json={"key": "value"})
+ assert response.status_code == 200
--- /dev/null
+from fastapi import APIRouter, FastAPI
+from fastapi.testclient import TestClient
+
+app = FastAPI()
+
+router_lax = APIRouter(prefix="/lax", strict_content_type=False)
+router_strict = APIRouter(prefix="/strict", strict_content_type=True)
+router_default = APIRouter(prefix="/default")
+
+
+@router_lax.post("/items/")
+async def router_lax_post(data: dict):
+ return data
+
+
+@router_strict.post("/items/")
+async def router_strict_post(data: dict):
+ return data
+
+
+@router_default.post("/items/")
+async def router_default_post(data: dict):
+ return data
+
+
+app.include_router(router_lax)
+app.include_router(router_strict)
+app.include_router(router_default)
+
+client = TestClient(app)
+
+
+def test_lax_router_on_strict_app_accepts_no_content_type():
+ response = client.post("/lax/items/", content='{"key": "value"}')
+ assert response.status_code == 200
+ assert response.json() == {"key": "value"}
+
+
+def test_strict_router_on_strict_app_rejects_no_content_type():
+ response = client.post("/strict/items/", content='{"key": "value"}')
+ assert response.status_code == 422
+
+
+def test_default_router_inherits_strict_from_app():
+ response = client.post("/default/items/", content='{"key": "value"}')
+ assert response.status_code == 422
+
+
+def test_lax_router_accepts_json_content_type():
+ response = client.post("/lax/items/", json={"key": "value"})
+ assert response.status_code == 200
+
+
+def test_strict_router_accepts_json_content_type():
+ response = client.post("/strict/items/", json={"key": "value"})
+ assert response.status_code == 200
+
+
+def test_default_router_accepts_json_content_type():
+ response = client.post("/default/items/", json={"key": "value"})
+ assert response.status_code == 200
assert response.status_code == 200, response.text
-def test_no_content_type_is_json(client: TestClient):
+def test_no_content_type_json(client: TestClient):
response = client.post(
"/items/",
content='{"name": "Foo", "price": 50.5}',
)
- assert response.status_code == 200, response.text
- assert response.json() == {
- "name": "Foo",
- "description": None,
- "price": 50.5,
- "tax": None,
- }
+ assert response.status_code == 422, response.text
def test_wrong_headers(client: TestClient):
--- /dev/null
+import importlib
+
+import pytest
+from fastapi.testclient import TestClient
+
+
+@pytest.fixture(
+ name="client",
+ params=[
+ "tutorial001_py310",
+ ],
+)
+def get_client(request: pytest.FixtureRequest):
+ mod = importlib.import_module(f"docs_src.strict_content_type.{request.param}")
+ client = TestClient(mod.app)
+ return client
+
+
+def test_lax_post_without_content_type_is_parsed_as_json(client: TestClient):
+ response = client.post(
+ "/items/",
+ content='{"name": "Foo", "price": 50.5}',
+ )
+ assert response.status_code == 200, response.text
+ assert response.json() == {"name": "Foo", "price": 50.5}
+
+
+def test_lax_post_with_json_content_type(client: TestClient):
+ response = client.post(
+ "/items/",
+ json={"name": "Foo", "price": 50.5},
+ )
+ assert response.status_code == 200, response.text
+ assert response.json() == {"name": "Foo", "price": 50.5}
+
+
+def test_lax_post_with_text_plain_is_still_rejected(client: TestClient):
+ response = client.post(
+ "/items/",
+ content='{"name": "Foo", "price": 50.5}',
+ headers={"Content-Type": "text/plain"},
+ )
+ assert response.status_code == 422, response.text
projects / thirdparty / fastapi / fastapi.git / commitdiff
