Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 45 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [2.6.2] - 2026-??-??

- Fix [#561](https://github.com/Neoteroi/BlackSheep/issues/561).
- Change `FromBodyBinder.binder_types` from a constructor-local variable to a
class attribute, defaulting to `[JSONBinder]` only. Users can configure
additional formats by setting `FromBodyBinder.binder_types` (e.g.
`FromBodyBinder.binder_types = [JSONBinder, FormBinder]`) before the
application starts.
- Add support for baking OpenAPI Specification files to disk, to support running
with `PYTHONOPTIMIZE=2` (or `-OO`) where docstrings are stripped and cannot be
used to enrich OpenAPI Documentation automatically.
Expand Down Expand Up @@ -76,6 +81,46 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
)
```

- Fix [#514](https://github.com/Neoteroi/BlackSheep/issues/514) — add support for
**multiple request body formats** on a single endpoint.
- New `MultiFormatBodyBinder`: holds an ordered list of inner `BodyBinder`s and
dispatches `get_value` to the first whose `matches_content_type` returns `True`.
Returns HTTP **415 Unsupported Media Type** when the binder is required and no
inner binder matches the request's `Content-Type`.
- **Union annotation syntax** — the normalization layer detects a union of
body-binder `BoundValue` types and automatically builds a
`MultiFormatBodyBinder`:

```python
async def create_item(data: FromJSON[Item] | FromForm[Item]) -> Item: ...
# optional variant
async def create_item(data: FromJSON[Item] | FromForm[Item] | None) -> Item: ...
```

- New **`FromBody[T]`** convenience type: equivalent to
`FromJSON[T] | FromForm[T]`, useful when you want to accept both structured
formats without being explicit:

```python
async def create_item(data: FromBody[Item]) -> Item: ...
```

- New **`FromXML[T]`** and **`XMLBinder`**: parse `application/xml` / `text/xml`
request bodies using [`defusedxml`](https://github.com/tiran/defusedxml),
protecting against **XXE injection**, **entity expansion (billion laughs)**,
and **DTD-based attacks**. Security exceptions propagate unmodified so the
application can distinguish attack attempts from ordinary malformed input.
Install the extra with `pip install blacksheep[xml]`.
- All new types compose freely:

```python
async def create_item(data: FromJSON[Item] | FromXML[Item] | FromForm[Item]): ...
```

- **OpenAPI Specification** is generated correctly for all combinations: each
accepted content type appears as a separate entry under `requestBody.content`,
all referencing the same schema.

## [2.6.1] - 2026-02-22 :cat:

- Fix missing escaping in `multipart/form-data` filenames and content-disposition headers.
Expand Down
2 changes: 2 additions & 0 deletions blacksheep/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
from .server.authorization import allow_anonymous as allow_anonymous
from .server.authorization import auth as auth
from .server.bindings import ClientInfo as ClientInfo
from .server.bindings import FromBody as FromBody
from .server.bindings import FromBytes as FromBytes
from .server.bindings import FromCookie as FromCookie
from .server.bindings import FromFiles as FromFiles
Expand All @@ -41,6 +42,7 @@
from .server.bindings import FromRoute as FromRoute
from .server.bindings import FromServices as FromServices
from .server.bindings import FromText as FromText
from .server.bindings import FromXML as FromXML
from .server.bindings import ServerInfo as ServerInfo
from .server.responses import ContentDispositionType as ContentDispositionType
from .server.responses import FileInput as FileInput
Expand Down
4 changes: 4 additions & 0 deletions blacksheep/exceptions.pxd
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,10 @@ cdef class InvalidOperation(Exception):
pass


cdef class UnsupportedMediaType(HTTPException):
pass


cdef class FailedRequestError(HTTPException):
cdef public str data

Expand Down
5 changes: 5 additions & 0 deletions blacksheep/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,11 @@ def __init__(self, message=None):
super().__init__(409, message or "Conflict")


class UnsupportedMediaType(HTTPException):
def __init__(self, message=None):
super().__init__(415, message or "Unsupported media type")


class RangeNotSatisfiable(HTTPException):
def __init__(self):
super().__init__(416, "Range not satisfiable")
Expand Down
6 changes: 6 additions & 0 deletions blacksheep/exceptions.pyx
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,12 @@ cdef class Conflict(HTTPException):
super().__init__(409, message or "Conflict")


cdef class UnsupportedMediaType(HTTPException):

def __init__(self, message=None):
super().__init__(415, message or "Unsupported media type")


cdef class RangeNotSatisfiable(HTTPException):

def __init__(self):
Expand Down
199 changes: 198 additions & 1 deletion blacksheep/server/bindings/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@
from rodi import CannotResolveTypeException, ContainerProtocol

from blacksheep.contents import FormPart
from blacksheep.exceptions import BadRequest
from blacksheep.exceptions import BadRequest, UnsupportedMediaType
from blacksheep.messages import Request
from blacksheep.server.bindings.converters import class_converters, converters
from blacksheep.server.routing import Router, URLResolver
Expand Down Expand Up @@ -199,6 +199,34 @@ async def create_profile(data: FromForm[UserProfile]):
default_value_type = dict


class FromXML(BoundValue[T]):
"""
A parameter obtained from an XML request body (``application/xml`` or ``text/xml``).

Requires ``defusedxml`` for safe parsing against common XML attacks (XXE, entity
expansion, DTD injection). Install it with::

pip install blacksheep[xml]

The root element tag is ignored; its child elements are mapped to model fields by
name. All standard field-type coercions (int, float, datetime, UUID …) apply via
the same converter chain used by JSON and form binders.
"""

default_value_type = dict


class FromBody(BoundValue[T]):
"""
A parameter obtained from the request body, accepting multiple content types.
By default accepts application/json and application/x-www-form-urlencoded /
multipart/form-data. For explicit format control use a union annotation such as
``FromJSON[T] | FromForm[T]``, which is handled identically at runtime.
"""

default_value_type = dict


class FromFiles(BoundValue[list[FormPart]]):
"""
A parameter obtained from multipart/form-data files.
Expand Down Expand Up @@ -580,6 +608,175 @@ async def read_data(self, request: Request) -> Any:
return await request.text()


def _element_to_dict(element) -> dict:
"""Recursively convert an ElementTree element to a plain dict.

- Strips XML namespaces from tag names.
- Multiple sibling elements with the same tag are collected into a list.
- Attributes of the element are merged into the dict.
"""
result: dict = {}

# Include element attributes first so child tags can override them if needed
for attr_name, attr_value in element.attrib.items():
if "}" in attr_name:
attr_name = attr_name.split("}", 1)[1]
result[attr_name] = attr_value

for child in element:
tag = child.tag
if "}" in tag:
tag = tag.split("}", 1)[1]

child_value = _element_to_dict(child) if len(child) > 0 else child.text

if tag in result:
existing = result[tag]
if not isinstance(existing, list):
result[tag] = [existing]
result[tag].append(child_value)
else:
result[tag] = child_value

return result


class XMLBinder(BodyBinder):
"""
Extracts a model from an XML request body (``application/xml`` or ``text/xml``).

Uses ``defusedxml`` to protect against XXE injection, entity expansion (billion
laughs), and DTD-based attacks. Install the extra with::

pip install blacksheep[xml]

The root element tag is ignored; its direct children are mapped to model fields by
name. The same type-coercion converters used by the JSON and form binders apply,
so ``int``, ``float``, ``datetime``, ``UUID``, and other annotated field types are
automatically coerced from their string representation.
"""

handle = FromXML

@property
def content_type(self) -> str:
return "application/xml;text/xml"

def matches_content_type(self, request: Request) -> bool:
return request.declares_content_type(
b"application/xml"
) or request.declares_content_type(b"text/xml")

async def read_data(self, request: Request) -> Any:
raw = await request.read()
if not raw:
return None
return self._parse_xml(raw)

@staticmethod
def _parse_xml(content: bytes) -> dict:
try:
import defusedxml.ElementTree as _ET # type: ignore[import]
from defusedxml import DefusedXmlException # type: ignore[import]
except ImportError as exc:
raise ImportError(
"defusedxml is required for safe XML parsing. "
"Install it with: pip install blacksheep[xml]"
) from exc

try:
root = _ET.fromstring(content)
except DefusedXmlException:
raise # security violations (XXE, entity expansion, DTD) propagate as-is
except Exception as exc:
raise InvalidRequestBody(f"Invalid XML: {exc}") from exc

return _element_to_dict(root)


class MultiFormatBodyBinder(BodyBinder):
"""
A BodyBinder that accepts multiple content types by delegating to a list of inner
BodyBinders. The first inner binder whose ``matches_content_type`` returns True is
used to read and parse the request body.

Instances are created automatically by the normalization layer when a union of
body-binder annotations is used (e.g. ``FromJSON[T] | FromForm[T]``), or when
``FromBody[T]`` is used.
"""

def __init__(
self,
inner_binders: "list[BodyBinder]",
expected_type=None,
name: str = "body",
implicit: bool = False,
required: bool = False,
):
# Pass a no-op converter so BodyBinder.__init__ doesn't build one unnecessarily;
# get_value is fully overridden and never calls self.converter.
super().__init__(
expected_type
or (inner_binders[0].expected_type if inner_binders else object),
name,
implicit,
required,
converter=lambda data: data,
)
self.inner_binders = inner_binders

@property
def content_type(self) -> str:
return ";".join(b.content_type for b in self.inner_binders)

def matches_content_type(self, request: Request) -> bool:
return any(b.matches_content_type(request) for b in self.inner_binders)

async def read_data(self, request: Request) -> Any: # pragma: no cover
raise NotImplementedError()

async def get_value(self, request: Request) -> Any:
if request.method in self._excluded_methods:
return None
for binder in self.inner_binders:
if binder.matches_content_type(request):
return await binder.get_value(request)
if self.required:
if not request.has_body():
raise MissingBodyError()
raise UnsupportedMediaType(
f"None of the supported content types matched the request. "
f"Accepted: {', '.join(b.content_type for b in self.inner_binders)}"
)
return None


class FromBodyBinder(MultiFormatBodyBinder):
"""
Binder for ``FromBody[T]``. By default accepts only JSON bodies.
To support additional formats, configure the ``binder_types`` class attribute::

FromBodyBinder.binder_types = [JSONBinder, FormBinder]
"""

handle = FromBody
binder_types: list[type[BodyBinder]] = [JSONBinder]

def __init__(
self,
expected_type,
name: str = "body",
implicit: bool = False,
required: bool = False,
converter=None,
):
inner = [
binder_type(expected_type, name, implicit, required)
for binder_type in self.binder_types
]
super().__init__(inner, expected_type, name, implicit, required)


class BytesBinder(Binder):
handle = FromBytes

Expand Down
Loading
Loading