Browse Source

🔒️ Add `strict_content_type` checking for JSON requests

pull/14978/head
Sebastián Ramírez 5 months ago
parent
commit
06438a3157
  1. 24
      fastapi/applications.py
  2. 36
      fastapi/routing.py

24
fastapi/applications.py

@ -840,6 +840,29 @@ class FastAPI(Starlette):
""" """
), ),
] = None, ] = 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[ **extra: Annotated[
Any, Any,
Doc( Doc(
@ -974,6 +997,7 @@ class FastAPI(Starlette):
include_in_schema=include_in_schema, include_in_schema=include_in_schema,
responses=responses, responses=responses,
generate_unique_id_function=generate_unique_id_function, generate_unique_id_function=generate_unique_id_function,
strict_content_type=strict_content_type,
) )
self.exception_handlers: dict[ self.exception_handlers: dict[
Any, Callable[[Request, Any], Response | Awaitable[Response]] Any, Callable[[Request, Any], Response | Awaitable[Response]]

36
fastapi/routing.py

@ -329,6 +329,7 @@ def get_request_handler(
response_model_exclude_none: bool = False, response_model_exclude_none: bool = False,
dependency_overrides_provider: Any | None = None, dependency_overrides_provider: Any | None = None,
embed_body_fields: bool = False, embed_body_fields: bool = False,
strict_content_type: bool = True,
) -> Callable[[Request], Coroutine[Any, Any, Response]]: ) -> Callable[[Request], Coroutine[Any, Any, Response]]:
assert dependant.call is not None, "dependant.call must be a function" assert dependant.call is not None, "dependant.call must be a function"
is_coroutine = dependant.is_coroutine_callable is_coroutine = dependant.is_coroutine_callable
@ -370,7 +371,8 @@ def get_request_handler(
json_body: Any = Undefined json_body: Any = Undefined
content_type_value = request.headers.get("content-type") content_type_value = request.headers.get("content-type")
if not content_type_value: if not content_type_value:
json_body = await request.json() if not strict_content_type:
json_body = await request.json()
else: else:
message = email.message.Message() message = email.message.Message()
message["content-type"] = content_type_value message["content-type"] = content_type_value
@ -599,6 +601,7 @@ class APIRoute(routing.Route):
openapi_extra: dict[str, Any] | None = None, openapi_extra: dict[str, Any] | None = None,
generate_unique_id_function: Callable[["APIRoute"], str] generate_unique_id_function: Callable[["APIRoute"], str]
| DefaultPlaceholder = Default(generate_unique_id), | DefaultPlaceholder = Default(generate_unique_id),
strict_content_type: bool = True,
) -> None: ) -> None:
self.path = path self.path = path
self.endpoint = endpoint self.endpoint = endpoint
@ -625,6 +628,7 @@ class APIRoute(routing.Route):
self.callbacks = callbacks self.callbacks = callbacks
self.openapi_extra = openapi_extra self.openapi_extra = openapi_extra
self.generate_unique_id_function = generate_unique_id_function self.generate_unique_id_function = generate_unique_id_function
self.strict_content_type = strict_content_type
self.tags = tags or [] self.tags = tags or []
self.responses = responses or {} self.responses = responses or {}
self.name = get_name(endpoint) if name is None else name self.name = get_name(endpoint) if name is None else name
@ -713,6 +717,7 @@ class APIRoute(routing.Route):
response_model_exclude_none=self.response_model_exclude_none, response_model_exclude_none=self.response_model_exclude_none,
dependency_overrides_provider=self.dependency_overrides_provider, dependency_overrides_provider=self.dependency_overrides_provider,
embed_body_fields=self._embed_body_fields, embed_body_fields=self._embed_body_fields,
strict_content_type=self.strict_content_type,
) )
def matches(self, scope: Scope) -> tuple[Match, Scope]: def matches(self, scope: Scope) -> tuple[Match, Scope]:
@ -963,6 +968,29 @@ class APIRouter(routing.Router):
""" """
), ),
] = Default(generate_unique_id), ] = 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/).
"""
),
] = True,
) -> None: ) -> None:
# Determine the lifespan context to use # Determine the lifespan context to use
if lifespan is None: if lifespan is None:
@ -1009,6 +1037,7 @@ class APIRouter(routing.Router):
self.route_class = route_class self.route_class = route_class
self.default_response_class = default_response_class self.default_response_class = default_response_class
self.generate_unique_id_function = generate_unique_id_function self.generate_unique_id_function = generate_unique_id_function
self.strict_content_type = strict_content_type
def route( def route(
self, self,
@ -1059,6 +1088,7 @@ class APIRouter(routing.Router):
openapi_extra: dict[str, Any] | None = None, openapi_extra: dict[str, Any] | None = None,
generate_unique_id_function: Callable[[APIRoute], str] generate_unique_id_function: Callable[[APIRoute], str]
| DefaultPlaceholder = Default(generate_unique_id), | DefaultPlaceholder = Default(generate_unique_id),
strict_content_type: bool | None = None,
) -> None: ) -> None:
route_class = route_class_override or self.route_class route_class = route_class_override or self.route_class
responses = responses or {} responses = responses or {}
@ -1105,6 +1135,9 @@ class APIRouter(routing.Router):
callbacks=current_callbacks, callbacks=current_callbacks,
openapi_extra=openapi_extra, openapi_extra=openapi_extra,
generate_unique_id_function=current_generate_unique_id, generate_unique_id_function=current_generate_unique_id,
strict_content_type=strict_content_type
if strict_content_type is not None
else self.strict_content_type,
) )
self.routes.append(route) self.routes.append(route)
@ -1480,6 +1513,7 @@ class APIRouter(routing.Router):
callbacks=current_callbacks, callbacks=current_callbacks,
openapi_extra=route.openapi_extra, openapi_extra=route.openapi_extra,
generate_unique_id_function=current_generate_unique_id, generate_unique_id_function=current_generate_unique_id,
strict_content_type=route.strict_content_type,
) )
elif isinstance(route, routing.Route): elif isinstance(route, routing.Route):
methods = list(route.methods or []) methods = list(route.methods or [])

Loading…
Cancel
Save