Fix multipart binary composed-schema matching

Author: p1c2u

openapi_core/deserializing/media_types/deserializers.py

@@ -1,5 +1,7 @@
+from dataclasses import dataclass
 from typing import TYPE_CHECKING
 from typing import Any
+from typing import Iterator
 from typing import Mapping
 from typing import Optional
 from xml.etree.ElementTree import ParseError
@@ -23,6 +25,7 @@
 from openapi_core.schema.protocols import SuportsGetAll
 from openapi_core.schema.protocols import SuportsGetList
 from openapi_core.schema.schemas import get_properties
+from openapi_core.validation.schemas.exceptions import ValidateError
 from openapi_core.validation.schemas.validators import SchemaValidator
 
 if TYPE_CHECKING:
@@ -63,6 +66,12 @@ def get_deserializer_callable(
         return self.media_type_deserializers[mimetype]
 
 
+@dataclass(frozen=True)
+class FormMediaSchemaMatch:
+    schema: SchemaPath
+    decoded_candidate: Mapping[str, Any]
+
+
 class MediaTypeDeserializer:
     def __init__(
         self,
@@ -97,7 +106,7 @@ def deserialize(self, value: bytes) -> Any:
         ):
             return deserialized
 
-        # decode multipart request bodies if schema provided
+        # Decode form-media bodies only when a schema is available.
         if self.schema is not None:
             return self.decode(deserialized)
 
@@ -126,43 +135,30 @@ def evolve(
             schema=schema,
             schema_validator=schema_validator,
             schema_caster=schema_caster,
+            encoding=self.encoding,
+            **self.parameters,
         )
 
     def decode(
         self, location: Mapping[str, Any], schema_only: bool = False
     ) -> Mapping[str, Any]:
-        # schema is required for multipart
+        # Form-media decoding always needs a schema to resolve properties.
         assert self.schema is not None
         properties: dict[str, Any] = {}
 
-        # For urlencoded/multipart, use caster for oneOf/anyOf detection if validator available
+        # For form media, select composed branches from decoded candidates.
         if self.schema_validator is not None:
-            one_of_schema = self.schema_validator.get_one_of_schema(
-                location, caster=self.schema_caster
-            )
-            if one_of_schema is not None:
-                one_of_properties = self.evolve(one_of_schema).decode(
-                    location, schema_only=True
-                )
-                properties.update(one_of_properties)
+            one_of_match = self.get_form_media_one_of_match(location)
+            if one_of_match is not None:
+                properties.update(one_of_match.decoded_candidate)
 
-            any_of_schemas = self.schema_validator.iter_any_of_schemas(
-                location, caster=self.schema_caster
-            )
-            for any_of_schema in any_of_schemas:
-                any_of_properties = self.evolve(any_of_schema).decode(
-                    location, schema_only=True
-                )
-                properties.update(any_of_properties)
+            any_of_matches = self.iter_form_media_any_of_matches(location)
+            for any_of_match in any_of_matches:
+                properties.update(any_of_match.decoded_candidate)
 
-            all_of_schemas = self.schema_validator.iter_all_of_schemas(
-                location
-            )
-            for all_of_schema in all_of_schemas:
-                all_of_properties = self.evolve(all_of_schema).decode(
-                    location, schema_only=True
-                )
-                properties.update(all_of_properties)
+            all_of_matches = self.iter_form_media_all_of_matches(location)
+            for all_of_match in all_of_matches:
+                properties.update(all_of_match.decoded_candidate)
 
         for prop_name, prop_schema in get_properties(self.schema).items():
             try:
@@ -179,6 +175,70 @@ def decode(
 
         return properties
 
+    def get_form_media_one_of_match(
+        self,
+        location: Mapping[str, Any],
+    ) -> Optional[FormMediaSchemaMatch]:
+        if self.schema is None or "oneOf" not in self.schema:
+            return None
+
+        for subschema in self.schema / "oneOf":
+            match = self.get_form_media_schema_match(subschema, location)
+            if match is not None:
+                return match
+
+        return None
+
+    def iter_form_media_any_of_matches(
+        self,
+        location: Mapping[str, Any],
+    ) -> list[FormMediaSchemaMatch]:
+        if self.schema is None or "anyOf" not in self.schema:
+            return []
+
+        return list(self.iter_form_media_schema_matches("anyOf", location))
+
+    def iter_form_media_all_of_matches(
+        self,
+        location: Mapping[str, Any],
+    ) -> list[FormMediaSchemaMatch]:
+        if self.schema is None or "allOf" not in self.schema:
+            return []
+
+        return list(self.iter_form_media_schema_matches("allOf", location))
+
+    def iter_form_media_schema_matches(
+        self,
+        keyword: str,
+        location: Mapping[str, Any],
+    ) -> Iterator[FormMediaSchemaMatch]:
+        assert self.schema is not None
+
+        for subschema in self.schema / keyword:
+            if keyword == "allOf" and "type" not in subschema:
+                continue
+            match = self.get_form_media_schema_match(subschema, location)
+            if match is not None:
+                yield match
+
+    def get_form_media_schema_match(
+        self,
+        subschema: SchemaPath,
+        location: Mapping[str, Any],
+    ) -> Optional[FormMediaSchemaMatch]:
+        assert self.schema_validator is not None
+
+        deserializer = self.evolve(subschema)
+        decoded_candidate = deserializer.decode(location, schema_only=True)
+        validator = self.schema_validator.evolve(subschema)
+
+        try:
+            validator.validate(decoded_candidate)
+        except ValidateError:
+            return None
+
+        return FormMediaSchemaMatch(subschema, decoded_candidate)
+
     def decode_property(
         self,
         prop_name: str,

openapi_core/unmarshalling/unmarshallers.py

@@ -117,7 +117,7 @@ def _get_param_or_header_and_schema(
     def _get_content_and_schema(
         self, raw: Any, content: SchemaPath, mimetype: Optional[str] = None
     ) -> Tuple[Any, Optional[SchemaPath]]:
-        casted, schema = super()._get_content_and_schema(
+        casted, schema = self._get_content_schema_value_and_schema(
             raw, content, mimetype
         )
         if schema is None:

openapi_core/validation/schemas/validators.py

@@ -20,6 +20,9 @@
 log = logging.getLogger(__name__)
 
 
+_MISSING = object()
+
+
 class SchemaValidator:
     def __init__(
         self,
@@ -33,7 +36,8 @@ def __contains__(self, schema_format: str) -> bool:
         return schema_format in self.validator.format_checker.checkers
 
     def validate(self, value: Any) -> None:
-        errors_iter = self.validator.iter_errors(value)
+        validation_value = self.get_binary_validation_value(value)
+        errors_iter = self.validator.iter_errors(validation_value)
         errors = tuple(errors_iter)
         if errors:
             schema_type = (self.schema / "type").read_str_or_list("any")
@@ -93,6 +97,8 @@ def get_primitive_type(self, value: Any) -> Optional[str]:
             schema_types = sorted(self.validator.TYPE_CHECKER._type_checkers)
         assert isinstance(schema_types, list)
         for schema_type in schema_types:
+            if self.accepts_binary_string_value(schema_type, value):
+                return schema_type
             result = self.type_validator(value, type_override=schema_type)
             if not result:
                 continue
@@ -104,6 +110,158 @@ def get_primitive_type(self, value: Any) -> Optional[str]:
         # OpenAPI 3.0: None is not a primitive type so None value will not find any type
         return None
 
+    def accepts_binary_string_value(
+        self, schema_type: Optional[str], value: Any
+    ) -> bool:
+        if schema_type != "string" or not isinstance(value, bytes):
+            return False
+
+        schema_format = (self.schema / "format").read_str(None)
+        return schema_format in ("binary", "byte")
+
+    def get_binary_validation_value(self, value: Any) -> Any:
+        # OpenAPI binary and byte string values are represented as bytes,
+        # but jsonschema validates string schemas against text values.
+        if self.accepts_binary_string_value(
+            (self.schema / "type").read_str(None), value
+        ):
+            return self.decode_binary_string_value(value)
+
+        normalized = value
+
+        for keyword in ["oneOf", "anyOf", "allOf"]:
+            if keyword not in self.schema:
+                continue
+            for subschema in self.schema / keyword:
+                branch_value = self.evolve(
+                    subschema
+                ).get_binary_validation_value(value)
+                normalized = self.merge_binary_validation_value(
+                    normalized, branch_value
+                )
+
+        if isinstance(normalized, dict):
+            return self.get_binary_validation_mapping_value(normalized)
+
+        if isinstance(normalized, list) and "items" in self.schema:
+            return self.get_binary_validation_array_value(normalized)
+
+        return normalized
+
+    def decode_binary_string_value(self, value: bytes) -> str:
+        try:
+            return value.decode("utf-8")
+        except UnicodeDecodeError:
+            return value.decode("ASCII", errors="surrogateescape")
+
+    def get_binary_validation_mapping_value(self, value: Any) -> Any:
+        normalized = value
+
+        if "properties" in self.schema:
+            for prop_name, prop_schema in (self.schema / "properties").items():
+                if prop_name not in value:
+                    continue
+                prop_value = self.evolve(
+                    prop_schema
+                ).get_binary_validation_value(value[prop_name])
+                if prop_value is value[prop_name]:
+                    continue
+                if normalized is value:
+                    normalized = dict(value)
+                normalized[prop_name] = prop_value
+
+        additional_properties = self.schema.get("additionalProperties", True)
+        if additional_properties in (True, False):
+            return normalized
+
+        property_names = set()
+        if "properties" in self.schema:
+            property_names = set((self.schema / "properties").keys())
+        additional_validator = self.evolve(
+            self.schema / "additionalProperties"
+        )
+        for prop_name, prop_value in value.items():
+            if prop_name in property_names:
+                continue
+            normalized_prop_value = (
+                additional_validator.get_binary_validation_value(prop_value)
+            )
+            if normalized_prop_value is prop_value:
+                continue
+            if normalized is value:
+                normalized = dict(value)
+            normalized[prop_name] = normalized_prop_value
+
+        return normalized
+
+    def get_binary_validation_array_value(self, value: Any) -> Any:
+        item_validator = self.evolve(self.schema / "items")
+        normalized = None
+
+        for idx, item in enumerate(value):
+            normalized_item = item_validator.get_binary_validation_value(item)
+            if normalized_item is item:
+                continue
+            if normalized is None:
+                normalized = list(value)
+            normalized[idx] = normalized_item
+
+        if normalized is None:
+            return value
+
+        return normalized
+
+    def merge_binary_validation_value(
+        self, value: Any, normalized_value: Any
+    ) -> Any:
+        if normalized_value is value:
+            return value
+
+        if isinstance(value, dict) and isinstance(normalized_value, dict):
+            merged_dict = value
+            for key, normalized_item in normalized_value.items():
+                item = value.get(key, _MISSING)
+                if item is _MISSING:
+                    if merged_dict is value:
+                        merged_dict = dict(value)
+                    merged_dict[key] = normalized_item
+                    continue
+
+                merged_item = self.merge_binary_validation_value(
+                    item, normalized_item
+                )
+                if merged_item is item:
+                    continue
+                if merged_dict is value:
+                    merged_dict = dict(value)
+                merged_dict[key] = merged_item
+
+            return merged_dict
+
+        if isinstance(value, list) and isinstance(normalized_value, list):
+            if len(value) != len(normalized_value):
+                return normalized_value
+
+            merged_list = None
+            for idx, (item, normalized_item) in enumerate(
+                zip(value, normalized_value)
+            ):
+                merged_item = self.merge_binary_validation_value(
+                    item, normalized_item
+                )
+                if merged_item is item:
+                    continue
+                if merged_list is None:
+                    merged_list = list(value)
+                merged_list[idx] = merged_item
+
+            if merged_list is None:
+                return value
+
+            return merged_list
+
+        return normalized_value
+
     def iter_valid_schemas(self, value: Any) -> Iterator[SchemaPath]:
         yield self.schema
 

tests/integration/unmarshalling/test_request_unmarshaller.py

@@ -1,6 +1,5 @@
 import json
 from base64 import b64encode
-from email.generator import _make_boundary
 
 import pytest
 
@@ -469,16 +468,10 @@ def test_request_body_with_object_default(self):
         assert result.errors == []
         assert result.body == {"tags": []}
 
-    @pytest.mark.xfail(
-        reason=(
-            "multipart composed-schema branch selection is not binary-aware"
-        ),
-        strict=True,
-    )
     def test_request_body_multipart_oneof_binary_field(self):
         from openapi_core import OpenAPI
 
-        boundary = _make_boundary()
+        boundary = "testboundary"
         spec = OpenAPI.from_dict(
             {
                 "openapi": "3.1.0",