Merge 020c3a5e0d into a09bba67cd

CHANGELOG.txt

@@ -13,7 +13,9 @@ Changelog
  * Fix: Preserve whitespace in comment replies (Elhussein Almasri)
  * Docs: Remove duplicate section on frontend caching proxies from performance page (Jake Howard)
  * Docs: Document `restriction_type` field on PageViewRestriction (Shlomo Markowitz)
+ * Docs: Document Wagtail's bug bounty policy (Jake Howard)
  * Maintenance: Use `DjangoJSONEncoder` instead of custom `LazyStringEncoder` to serialize Draftail config (Sage Abdullah)
+ * Maintenance: Refactor image chooser pagination to check `WAGTAILIMAGES_CHOOSER_PAGE_SIZE` at runtime (Matt Westcott)
 
 
 6.1 (01.05.2024)

docs/contributing/security.md

@@ -34,6 +34,12 @@ At any given time, the Wagtail team provides official security support for sever
 When new releases are issued for security reasons, the accompanying notice will include a list of affected versions.
 This list is comprised solely of supported versions of Wagtail: older versions may also be affected, but we do not investigate to determine that, and will not issue patches or new releases for those versions.
 
+## Bug Bounties
+
+Wagtail does not have a "Bug Bounty" program. Whilst we appreciate and accept reports from anyone, and will gladly give credit to you and/or your organisation, we aren't able to "reward" you for reporting the vulnerability.
+
+["Beg Bounties"](https://www.troyhunt.com/beg-bounties/) are ever increasing among security researchers, and it's not something we condone or support.
+
 ## How Wagtail discloses security issues
 
 Our process for taking a security issue from private discussion to public disclosure involves multiple steps.
@@ -46,8 +52,8 @@ On the day of disclosure, we will take the following steps:
 1. Apply the relevant patch(es) to Wagtail's codebase.
    The commit messages for these patches will indicate that they are for security issues, but will not describe the issue in any detail; instead, they will warn of upcoming disclosure.
 2. Issue the relevant release(s), by placing new packages on [the Python Package Index](https://pypi.org/project/wagtail/), tagging the new release(s) in Wagtail's GitHub repository and updating Wagtail's [release notes](../releases/index).
-3. Post a public entry on [Wagtail's blog](https://wagtail.org/blog/), describing the issue and its resolution in detail, pointing to the relevant patches and new releases, and crediting the reporter of the issue (if the reporter wishes to be publicly identified).
-4. Post a notice to the [Wagtail discussion board](https://github.com/wagtail/wagtail/discussions), [Slack workspace](https://wagtail.org/slack/) and Twitter feed ([\@WagtailCMS](https://twitter.com/wagtailcms)) that links to the blog post.
+3. Publish a [security advisory](https://github.com/wagtail/wagtail/security/advisories?state=published) on Wagtail's GitHub repository. This describes the issue and its resolution in detail, pointing to the relevant patches and new releases, and crediting the reporter of the issue (if the reporter wishes to be publicly identified)
+4. Post a notice to the [Wagtail discussion board](https://github.com/wagtail/wagtail/discussions), [Slack workspace](https://wagtail.org/slack/) and Twitter feed ([\@WagtailCMS](https://twitter.com/wagtailcms)) that links to the security advisory.
 
 If a reported issue is believed to be particularly time-sensitive -- due to a known exploit in the wild, for example -- the time between advance notification and public disclosure may be shortened considerably.
 

docs/releases/6.2.md

@@ -30,11 +30,13 @@ depth: 1
 
  * Remove duplicate section on frontend caching proxies from performance page (Jake Howard)
  * Document `restriction_type` field on PageViewRestriction (Shlomo Markowitz)
+ * Document Wagtail's bug bounty policy (Jake Howard)
 
 
 ### Maintenance
 
  * Use `DjangoJSONEncoder` instead of custom `LazyStringEncoder` to serialize Draftail config (Sage Abdullah)
+ * Refactor image chooser pagination to check `WAGTAILIMAGES_CHOOSER_PAGE_SIZE` at runtime (Matt Westcott)
 
 
 ## Upgrade considerations - changes affecting all projects

wagtail/admin/viewsets/base.py

@@ -17,6 +17,10 @@ class ViewSet(WagtailMenuRegisterable):
     For more information on how to use this class, see :ref:`using_base_viewset`.
     """
 
+    #: A special value that, when passed in a kwargs dict to construct a view, indicates that
+    #: the attribute should not be written and should instead be left as the view's initial value
+    UNDEFINED = object()
+
     #: A name for this viewset, used as the default URL prefix and namespace.
     name = None
 
@@ -42,12 +46,13 @@ class ViewSet(WagtailMenuRegisterable):
         in addition to any kwargs passed to this method. Items from get_common_view_kwargs will be
         filtered to only include those that are valid for the given view_class.
         """
+        merged_kwargs = self.get_common_view_kwargs()
+        merged_kwargs.update(kwargs)
         filtered_kwargs = {
             key: value
-            for key, value in self.get_common_view_kwargs().items()
-            if hasattr(view_class, key)
+            for key, value in merged_kwargs.items()
+            if hasattr(view_class, key) and value is not self.UNDEFINED
         }
-        filtered_kwargs.update(kwargs)
         return view_class.as_view(**filtered_kwargs)
 
     def inject_view_methods(self, view_class, method_names):

wagtail/admin/viewsets/chooser.py

@@ -29,7 +29,7 @@ class ChooserViewSet(ViewSet):
     )  #: Label for the 'choose' button in the chooser widget, when an item has already been chosen
     edit_item_text = _("Edit")  #: Label for the 'edit' button in the chooser widget
 
-    per_page = 10  #: Number of results to show per page
+    per_page = ViewSet.UNDEFINED  #: Number of results to show per page
 
     #: A list of URL query parameters that should be passed on unmodified as part of any links or
     #: form submissions within the chooser modal workflow.

wagtail/blocks/stream_block.py

@@ -75,398 +75,6 @@ class StreamBlockValidationError(ValidationError):
         return result
 
 
-class BaseStreamBlock(Block):
-    def __init__(self, local_blocks=None, search_index=True, **kwargs):
-        self._constructor_kwargs = kwargs
-        self.search_index = search_index
-
-        super().__init__(**kwargs)
-
-        # create a local (shallow) copy of base_blocks so that it can be supplemented by local_blocks
-        self.child_blocks = self.base_blocks.copy()
-        if local_blocks:
-            for name, block in local_blocks:
-                block.set_name(name)
-                self.child_blocks[name] = block
-
-    def empty_value(self, raw_text=None):
-        return StreamValue(self, [], raw_text=raw_text)
-
-    def sorted_child_blocks(self):
-        """Child blocks, sorted in to their groups."""
-        return sorted(
-            self.child_blocks.values(), key=lambda child_block: child_block.meta.group
-        )
-
-    def grouped_child_blocks(self):
-        """
-        The available child block types of this stream block, organised into groups according to
-        their meta.group attribute.
-        Returned as an iterable of (group_name, list_of_blocks) tuples
-        """
-        return itertools.groupby(
-            self.sorted_child_blocks(), key=lambda child_block: child_block.meta.group
-        )
-
-    def value_from_datadict(self, data, files, prefix):
-        count = int(data["%s-count" % prefix])
-        values_with_indexes = []
-        for i in range(0, count):
-            if data["%s-%d-deleted" % (prefix, i)]:
-                continue
-            block_type_name = data["%s-%d-type" % (prefix, i)]
-            try:
-                child_block = self.child_blocks[block_type_name]
-            except KeyError:
-                continue
-
-            values_with_indexes.append(
-                (
-                    int(data["%s-%d-order" % (prefix, i)]),
-                    block_type_name,
-                    child_block.value_from_datadict(
-                        data, files, "%s-%d-value" % (prefix, i)
-                    ),
-                    data.get("%s-%d-id" % (prefix, i)),
-                )
-            )
-
-        values_with_indexes.sort()
-        return StreamValue(
-            self,
-            [
-                (child_block_type_name, value, block_id)
-                for (
-                    index,
-                    child_block_type_name,
-                    value,
-                    block_id,
-                ) in values_with_indexes
-            ],
-        )
-
-    def value_omitted_from_data(self, data, files, prefix):
-        return ("%s-count" % prefix) not in data
-
-    @property
-    def required(self):
-        return self.meta.required
-
-    def clean(self, value):
-        cleaned_data = []
-        errors = {}
-        non_block_errors = ErrorList()
-        for i, child in enumerate(value):  # child is a StreamChild instance
-            try:
-                cleaned_data.append(
-                    (child.block.name, child.block.clean(child.value), child.id)
-                )
-            except ValidationError as e:
-                errors[i] = e
-
-        if self.meta.min_num is not None and self.meta.min_num > len(value):
-            non_block_errors.append(
-                ValidationError(
-                    _("The minimum number of items is %(min_num)d")
-                    % {"min_num": self.meta.min_num}
-                )
-            )
-        elif self.required and len(value) == 0:
-            non_block_errors.append(ValidationError(_("This field is required.")))
-
-        if self.meta.max_num is not None and self.meta.max_num < len(value):
-            non_block_errors.append(
-                ValidationError(
-                    _("The maximum number of items is %(max_num)d")
-                    % {"max_num": self.meta.max_num}
-                )
-            )
-
-        if self.meta.block_counts:
-            block_counts = defaultdict(int)
-            for item in value:
-                block_counts[item.block_type] += 1
-
-            for block_name, min_max in self.meta.block_counts.items():
-                block = self.child_blocks[block_name]
-                max_num = min_max.get("max_num", None)
-                min_num = min_max.get("min_num", None)
-                block_count = block_counts[block_name]
-                if min_num is not None and min_num > block_count:
-                    non_block_errors.append(
-                        ValidationError(
-                            "{}: {}".format(
-                                block.label,
-                                _("The minimum number of items is %(min_num)d")
-                                % {"min_num": min_num},
-                            )
-                        )
-                    )
-                if max_num is not None and max_num < block_count:
-                    non_block_errors.append(
-                        ValidationError(
-                            "{}: {}".format(
-                                block.label,
-                                _("The maximum number of items is %(max_num)d")
-                                % {"max_num": max_num},
-                            )
-                        )
-                    )
-
-        if errors or non_block_errors:
-            # The message here is arbitrary - outputting error messages is delegated to the child blocks,
-            # which only involves the 'params' list
-            raise StreamBlockValidationError(
-                block_errors=errors, non_block_errors=non_block_errors
-            )
-
-        return StreamValue(self, cleaned_data)
-
-    def to_python(self, value):
-        if isinstance(value, StreamValue):
-            return value
-        elif isinstance(value, str) and value:
-            try:
-                value = json.loads(value)
-            except ValueError:
-                # value is not valid JSON; most likely, this field was previously a
-                # rich text field before being migrated to StreamField, and the data
-                # was left intact in the migration. Return an empty stream instead
-                # (but keep the raw text available as an attribute, so that it can be
-                # used to migrate that data to StreamField)
-                return self.empty_value(raw_text=value)
-
-        if not value:
-            return self.empty_value()
-
-        # ensure value is a list and not some other kind of iterable
-        value = list(value)
-
-        if isinstance(value[0], dict):
-            # value is in JSONish representation - a dict with 'type' and 'value' keys.
-            # This is passed to StreamValue to be expanded lazily - but first we reject any unrecognised
-            # block types from the list
-            return StreamValue(
-                self,
-                [
-                    child_data
-                    for child_data in value
-                    if child_data["type"] in self.child_blocks
-                ],
-                is_lazy=True,
-            )
-        else:
-            # See if it looks like the standard non-smart representation of a
-            # StreamField value: a list of (block_name, value) tuples
-            try:
-                [None for (x, y) in value]
-            except (TypeError, ValueError) as exc:
-                # Give up trying to make sense of the value
-                raise TypeError(
-                    f"Cannot handle {value!r} (type {type(value)!r}) as a value of a StreamBlock"
-                ) from exc
-
-            # Test succeeded, so return as a StreamValue-ified version of that value
-            return StreamValue(
-                self,
-                [
-                    (k, self.child_blocks[k].normalize(v))
-                    for k, v in value
-                    if k in self.child_blocks
-                ],
-            )
-
-    def bulk_to_python(self, values):
-        # 'values' is a list of streams, each stream being a list of dicts with 'type', 'value' and
-        # optionally 'id'.
-        # We will iterate over these streams, constructing:
-        # 1) a set of per-child-block lists ('child_inputs'), to be sent to each child block's
-        #    bulk_to_python method in turn (giving us 'child_outputs')
-        # 2) a 'block map' of each stream, telling us the type and id of each block and the index we
-        #    need to look up in the corresponding child_outputs list to obtain its final value
-
-        child_inputs = defaultdict(list)
-        block_maps = []
-
-        for stream in values:
-            block_map = []
-            for block_dict in stream:
-                block_type = block_dict["type"]
-
-                if block_type not in self.child_blocks:
-                    # skip any blocks with an unrecognised type
-                    continue
-
-                child_input_list = child_inputs[block_type]
-                child_index = len(child_input_list)
-                child_input_list.append(block_dict["value"])
-                block_map.append((block_type, block_dict.get("id"), child_index))
-
-            block_maps.append(block_map)
-
-        # run each list in child_inputs through the relevant block's bulk_to_python
-        # to obtain child_outputs
-        child_outputs = {
-            block_type: self.child_blocks[block_type].bulk_to_python(child_input_list)
-            for block_type, child_input_list in child_inputs.items()
-        }
-
-        # for each stream, go through the block map, picking out the appropriately-indexed
-        # value from the relevant list in child_outputs
-        return [
-            StreamValue(
-                self,
-                [
-                    (block_type, child_outputs[block_type][child_index], id)
-                    for block_type, id, child_index in block_map
-                ],
-                is_lazy=False,
-            )
-            for block_map in block_maps
-        ]
-
-    def get_prep_value(self, value):
-        if not value:
-            # Falsy values (including None, empty string, empty list, and
-            # empty StreamValue) become an empty stream
-            return []
-        else:
-            # value is a StreamValue - delegate to its get_prep_value() method
-            # (which has special-case handling for lazy StreamValues to avoid useless
-            # round-trips to the full data representation and back)
-            return value.get_prep_value()
-
-    def normalize(self, value):
-        return self.to_python(value)
-
-    def get_form_state(self, value):
-        if not value:
-            return []
-        else:
-            return [
-                {
-                    "type": child.block.name,
-                    "value": child.block.get_form_state(child.value),
-                    "id": child.id,
-                }
-                for child in value
-            ]
-
-    def get_api_representation(self, value, context=None):
-        if value is None:
-            # treat None as identical to an empty stream
-            return []
-
-        return [
-            {
-                "type": child.block.name,
-                "value": child.block.get_api_representation(
-                    child.value, context=context
-                ),
-                "id": child.id,
-            }
-            for child in value  # child is a StreamChild instance
-        ]
-
-    def render_basic(self, value, context=None):
-        return format_html_join(
-            "\n",
-            '<div class="block-{1}">{0}</div>',
-            [(child.render(context=context), child.block_type) for child in value],
-        )
-
-    def get_searchable_content(self, value):
-        if not self.search_index:
-            return []
-        content = []
-
-        for child in value:
-            content.extend(child.block.get_searchable_content(child.value))
-
-        return content
-
-    def extract_references(self, value):
-        for child in value:
-            for (
-                model,
-                object_id,
-                model_path,
-                content_path,
-            ) in child.block.extract_references(child.value):
-                model_path = (
-                    f"{child.block_type}.{model_path}"
-                    if model_path
-                    else child.block_type
-                )
-                content_path = (
-                    f"{child.id}.{content_path}" if content_path else child.id
-                )
-                yield model, object_id, model_path, content_path
-
-    def get_block_by_content_path(self, value, path_elements):
-        """
-        Given a list of elements from a content path, retrieve the block at that path
-        as a BoundBlock object, or None if the path does not correspond to a valid block.
-        """
-        if path_elements:
-            id, *remaining_elements = path_elements
-            for child in value:
-                if child.id == id:
-                    return child.block.get_block_by_content_path(
-                        child.value, remaining_elements
-                    )
-        else:
-            # an empty path refers to the stream as a whole
-            return self.bind(value)
-
-    def deconstruct(self):
-        """
-        Always deconstruct StreamBlock instances as if they were plain StreamBlocks with all of the
-        field definitions passed to the constructor - even if in reality this is a subclass of StreamBlock
-        with the fields defined declaratively, or some combination of the two.
-
-        This ensures that the field definitions get frozen into migrations, rather than leaving a reference
-        to a custom subclass in the user's models.py that may or may not stick around.
-        """
-        path = "wagtail.blocks.StreamBlock"
-        args = [list(self.child_blocks.items())]
-        kwargs = self._constructor_kwargs
-        return (path, args, kwargs)
-
-    def check(self, **kwargs):
-        errors = super().check(**kwargs)
-        for name, child_block in self.child_blocks.items():
-            errors.extend(child_block.check(**kwargs))
-            errors.extend(child_block._check_name(**kwargs))
-
-        return errors
-
-    class Meta:
-        # No icon specified here, because that depends on the purpose that the
-        # block is being used for. Feel encouraged to specify an icon in your
-        # descendant block type
-        icon = "placeholder"
-        default = []
-        required = True
-        form_classname = None
-        min_num = None
-        max_num = None
-        block_counts = {}
-        collapsed = False
-
-    MUTABLE_META_ATTRIBUTES = [
-        "required",
-        "min_num",
-        "max_num",
-        "block_counts",
-        "collapsed",
-    ]
-
-
-class StreamBlock(BaseStreamBlock, metaclass=DeclarativeSubBlocksMetaclass):
-    pass
-
-
 class StreamValue(MutableSequence):
     """
     Custom type used to represent the value of a StreamBlock; behaves as a sequence of BoundBlocks
@@ -801,6 +409,400 @@ class StreamValue(MutableSequence):
         )
 
 
+class BaseStreamBlock(Block):
+    def __init__(self, local_blocks=None, search_index=True, **kwargs):
+        self._constructor_kwargs = kwargs
+        self.search_index = search_index
+
+        super().__init__(**kwargs)
+
+        # create a local (shallow) copy of base_blocks so that it can be supplemented by local_blocks
+        self.child_blocks = self.base_blocks.copy()
+        if local_blocks:
+            for name, block in local_blocks:
+                block.set_name(name)
+                self.child_blocks[name] = block
+
+    def empty_value(self, raw_text=None):
+        return self.meta.value_class(self, [], raw_text=raw_text)
+
+    def sorted_child_blocks(self):
+        """Child blocks, sorted in to their groups."""
+        return sorted(
+            self.child_blocks.values(), key=lambda child_block: child_block.meta.group
+        )
+
+    def grouped_child_blocks(self):
+        """
+        The available child block types of this stream block, organised into groups according to
+        their meta.group attribute.
+        Returned as an iterable of (group_name, list_of_blocks) tuples
+        """
+        return itertools.groupby(
+            self.sorted_child_blocks(), key=lambda child_block: child_block.meta.group
+        )
+
+    def value_from_datadict(self, data, files, prefix):
+        count = int(data["%s-count" % prefix])
+        values_with_indexes = []
+        for i in range(0, count):
+            if data["%s-%d-deleted" % (prefix, i)]:
+                continue
+            block_type_name = data["%s-%d-type" % (prefix, i)]
+            try:
+                child_block = self.child_blocks[block_type_name]
+            except KeyError:
+                continue
+
+            values_with_indexes.append(
+                (
+                    int(data["%s-%d-order" % (prefix, i)]),
+                    block_type_name,
+                    child_block.value_from_datadict(
+                        data, files, "%s-%d-value" % (prefix, i)
+                    ),
+                    data.get("%s-%d-id" % (prefix, i)),
+                )
+            )
+
+        values_with_indexes.sort()
+        return self.meta.value_class(
+            self,
+            [
+                (child_block_type_name, value, block_id)
+                for (
+                    index,
+                    child_block_type_name,
+                    value,
+                    block_id,
+                ) in values_with_indexes
+            ],
+        )
+
+    def value_omitted_from_data(self, data, files, prefix):
+        return ("%s-count" % prefix) not in data
+
+    @property
+    def required(self):
+        return self.meta.required
+
+    def clean(self, value):
+        cleaned_data = []
+        errors = {}
+        non_block_errors = ErrorList()
+        for i, child in enumerate(value):  # child is a StreamChild instance
+            try:
+                cleaned_data.append(
+                    (child.block.name, child.block.clean(child.value), child.id)
+                )
+            except ValidationError as e:
+                errors[i] = e
+
+        if self.meta.min_num is not None and self.meta.min_num > len(value):
+            non_block_errors.append(
+                ValidationError(
+                    _("The minimum number of items is %(min_num)d")
+                    % {"min_num": self.meta.min_num}
+                )
+            )
+        elif self.required and len(value) == 0:
+            non_block_errors.append(ValidationError(_("This field is required.")))
+
+        if self.meta.max_num is not None and self.meta.max_num < len(value):
+            non_block_errors.append(
+                ValidationError(
+                    _("The maximum number of items is %(max_num)d")
+                    % {"max_num": self.meta.max_num}
+                )
+            )
+
+        if self.meta.block_counts:
+            block_counts = defaultdict(int)
+            for item in value:
+                block_counts[item.block_type] += 1
+
+            for block_name, min_max in self.meta.block_counts.items():
+                block = self.child_blocks[block_name]
+                max_num = min_max.get("max_num", None)
+                min_num = min_max.get("min_num", None)
+                block_count = block_counts[block_name]
+                if min_num is not None and min_num > block_count:
+                    non_block_errors.append(
+                        ValidationError(
+                            "{}: {}".format(
+                                block.label,
+                                _("The minimum number of items is %(min_num)d")
+                                % {"min_num": min_num},
+                            )
+                        )
+                    )
+                if max_num is not None and max_num < block_count:
+                    non_block_errors.append(
+                        ValidationError(
+                            "{}: {}".format(
+                                block.label,
+                                _("The maximum number of items is %(max_num)d")
+                                % {"max_num": max_num},
+                            )
+                        )
+                    )
+
+        if errors or non_block_errors:
+            # The message here is arbitrary - outputting error messages is delegated to the child blocks,
+            # which only involves the 'params' list
+            raise StreamBlockValidationError(
+                block_errors=errors, non_block_errors=non_block_errors
+            )
+
+        return self.meta.value_class(self, cleaned_data)
+
+    def to_python(self, value):
+        if isinstance(value, self.meta.value_class):
+            return value
+        elif isinstance(value, str) and value:
+            try:
+                value = json.loads(value)
+            except ValueError:
+                # value is not valid JSON; most likely, this field was previously a
+                # rich text field before being migrated to StreamField, and the data
+                # was left intact in the migration. Return an empty stream instead
+                # (but keep the raw text available as an attribute, so that it can be
+                # used to migrate that data to StreamField)
+                return self.empty_value(raw_text=value)
+
+        if not value:
+            return self.empty_value()
+
+        # ensure value is a list and not some other kind of iterable
+        value = list(value)
+
+        if isinstance(value[0], dict):
+            # value is in JSONish representation - a dict with 'type' and 'value' keys.
+            # This is passed to StreamValue to be expanded lazily - but first we reject any unrecognised
+            # block types from the list
+            return self.meta.value_class(
+                self,
+                [
+                    child_data
+                    for child_data in value
+                    if child_data["type"] in self.child_blocks
+                ],
+                is_lazy=True,
+            )
+        else:
+            # See if it looks like the standard non-smart representation of a
+            # StreamField value: a list of (block_name, value) tuples
+            try:
+                [None for (x, y) in value]
+            except (TypeError, ValueError) as exc:
+                # Give up trying to make sense of the value
+                raise TypeError(
+                    f"Cannot handle {value!r} (type {type(value)!r}) as a value of a StreamBlock"
+                ) from exc
+
+            # Test succeeded, so return as a StreamValue-ified version of that value
+            return self.meta.value_class(
+                self,
+                [
+                    (k, self.child_blocks[k].normalize(v))
+                    for k, v in value
+                    if k in self.child_blocks
+                ],
+            )
+
+    def bulk_to_python(self, values):
+        # 'values' is a list of streams, each stream being a list of dicts with 'type', 'value' and
+        # optionally 'id'.
+        # We will iterate over these streams, constructing:
+        # 1) a set of per-child-block lists ('child_inputs'), to be sent to each child block's
+        #    bulk_to_python method in turn (giving us 'child_outputs')
+        # 2) a 'block map' of each stream, telling us the type and id of each block and the index we
+        #    need to look up in the corresponding child_outputs list to obtain its final value
+
+        child_inputs = defaultdict(list)
+        block_maps = []
+
+        for stream in values:
+            block_map = []
+            for block_dict in stream:
+                block_type = block_dict["type"]
+
+                if block_type not in self.child_blocks:
+                    # skip any blocks with an unrecognised type
+                    continue
+
+                child_input_list = child_inputs[block_type]
+                child_index = len(child_input_list)
+                child_input_list.append(block_dict["value"])
+                block_map.append((block_type, block_dict.get("id"), child_index))
+
+            block_maps.append(block_map)
+
+        # run each list in child_inputs through the relevant block's bulk_to_python
+        # to obtain child_outputs
+        child_outputs = {
+            block_type: self.child_blocks[block_type].bulk_to_python(child_input_list)
+            for block_type, child_input_list in child_inputs.items()
+        }
+
+        # for each stream, go through the block map, picking out the appropriately-indexed
+        # value from the relevant list in child_outputs
+        return [
+            self.meta.value_class(
+                self,
+                [
+                    (block_type, child_outputs[block_type][child_index], id)
+                    for block_type, id, child_index in block_map
+                ],
+                is_lazy=False,
+            )
+            for block_map in block_maps
+        ]
+
+    def get_prep_value(self, value):
+        if not value:
+            # Falsy values (including None, empty string, empty list, and
+            # empty StreamValue) become an empty stream
+            return []
+        else:
+            # value is a StreamValue - delegate to its get_prep_value() method
+            # (which has special-case handling for lazy StreamValues to avoid useless
+            # round-trips to the full data representation and back)
+            return value.get_prep_value()
+
+    def normalize(self, value):
+        return self.to_python(value)
+
+    def get_form_state(self, value):
+        if not value:
+            return []
+        else:
+            return [
+                {
+                    "type": child.block.name,
+                    "value": child.block.get_form_state(child.value),
+                    "id": child.id,
+                }
+                for child in value
+            ]
+
+    def get_api_representation(self, value, context=None):
+        if value is None:
+            # treat None as identical to an empty stream
+            return []
+
+        return [
+            {
+                "type": child.block.name,
+                "value": child.block.get_api_representation(
+                    child.value, context=context
+                ),
+                "id": child.id,
+            }
+            for child in value  # child is a StreamChild instance
+        ]
+
+    def render_basic(self, value, context=None):
+        return format_html_join(
+            "\n",
+            '<div class="block-{1}">{0}</div>',
+            [(child.render(context=context), child.block_type) for child in value],
+        )
+
+    def get_searchable_content(self, value):
+        if not self.search_index:
+            return []
+        content = []
+
+        for child in value:
+            content.extend(child.block.get_searchable_content(child.value))
+
+        return content
+
+    def extract_references(self, value):
+        for child in value:
+            for (
+                model,
+                object_id,
+                model_path,
+                content_path,
+            ) in child.block.extract_references(child.value):
+                model_path = (
+                    f"{child.block_type}.{model_path}"
+                    if model_path
+                    else child.block_type
+                )
+                content_path = (
+                    f"{child.id}.{content_path}" if content_path else child.id
+                )
+                yield model, object_id, model_path, content_path
+
+    def get_block_by_content_path(self, value, path_elements):
+        """
+        Given a list of elements from a content path, retrieve the block at that path
+        as a BoundBlock object, or None if the path does not correspond to a valid block.
+        """
+        if path_elements:
+            id, *remaining_elements = path_elements
+            for child in value:
+                if child.id == id:
+                    return child.block.get_block_by_content_path(
+                        child.value, remaining_elements
+                    )
+        else:
+            # an empty path refers to the stream as a whole
+            return self.bind(value)
+
+    def deconstruct(self):
+        """
+        Always deconstruct StreamBlock instances as if they were plain StreamBlocks with all of the
+        field definitions passed to the constructor - even if in reality this is a subclass of StreamBlock
+        with the fields defined declaratively, or some combination of the two.
+
+        This ensures that the field definitions get frozen into migrations, rather than leaving a reference
+        to a custom subclass in the user's models.py that may or may not stick around.
+        """
+        path = "wagtail.blocks.StreamBlock"
+        args = [list(self.child_blocks.items())]
+        kwargs = self._constructor_kwargs
+        return (path, args, kwargs)
+
+    def check(self, **kwargs):
+        errors = super().check(**kwargs)
+        for name, child_block in self.child_blocks.items():
+            errors.extend(child_block.check(**kwargs))
+            errors.extend(child_block._check_name(**kwargs))
+
+        return errors
+
+    class Meta:
+        # No icon specified here, because that depends on the purpose that the
+        # block is being used for. Feel encouraged to specify an icon in your
+        # descendant block type
+        icon = "placeholder"
+        default = []
+        required = True
+        form_classname = None
+        min_num = None
+        max_num = None
+        block_counts = {}
+        collapsed = False
+        value_class = StreamValue
+
+    MUTABLE_META_ATTRIBUTES = [
+        "required",
+        "min_num",
+        "max_num",
+        "block_counts",
+        "collapsed",
+        "value_class",
+    ]
+
+
+class StreamBlock(BaseStreamBlock, metaclass=DeclarativeSubBlocksMetaclass):
+    pass
+
+
 class StreamBlockAdapter(Adapter):
     js_constructor = "wagtail.blocks.StreamBlock"
 

wagtail/fields.py

@@ -6,7 +6,7 @@ from django.db import models
 from django.db.models.fields.json import KeyTransform
 from django.utils.encoding import force_str
 
-from wagtail.blocks import Block, BlockField, StreamBlock, StreamValue
+from wagtail.blocks import Block, BlockField, StreamBlock
 from wagtail.rich_text import (
     RichTextMaxLengthValidator,
     extract_references_from_rich_text,
@@ -88,7 +88,7 @@ class StreamField(models.Field):
 
         # extract kwargs that are to be passed on to the block, not handled by super
         block_opts = {}
-        for arg in ["min_num", "max_num", "block_counts", "collapsed"]:
+        for arg in ["min_num", "max_num", "block_counts", "collapsed", "value_class"]:
             if arg in kwargs:
                 block_opts[arg] = kwargs.pop(arg)
 
@@ -111,6 +111,10 @@ class StreamField(models.Field):
 
         self.stream_block.set_meta_options(block_opts)
 
+    @property
+    def value_class(self):
+        return self.stream_block.meta.value_class
+
     @property
     def json_field(self):
         return models.JSONField(encoder=DjangoJSONEncoder)
@@ -142,7 +146,7 @@ class StreamField(models.Field):
 
     def get_prep_value(self, value):
         if (
-            isinstance(value, StreamValue)
+            isinstance(value, self.value_class)
             and not (value)
             and value.raw_text is not None
         ):
@@ -151,7 +155,7 @@ class StreamField(models.Field):
             # for reverse migrations that convert StreamField data back into plain text
             # fields.)
             return value.raw_text
-        elif isinstance(value, StreamValue):
+        elif isinstance(value, self.value_class):
             # StreamValue instances must be prepared first.
             return json.dumps(
                 self.stream_block.get_prep_value(value), cls=DjangoJSONEncoder
@@ -163,7 +167,7 @@ class StreamField(models.Field):
             return self.json_field.get_prep_value(value)
 
     def get_db_prep_value(self, value, connection, prepared=False):
-        if not isinstance(value, StreamValue):
+        if not isinstance(value, self.value_class):
             # When querying with JSONField features, the rhs might not be a StreamValue.
             # As of Django 4.2, JSONField value serialisation is handled in
             # get_db_prep_value instead of get_prep_value.

wagtail/images/tests/test_admin_views.py

@@ -1681,6 +1681,22 @@ class TestImageChooserView(WagtailTestUtils, TestCase):
         response = self.get({"p": 9999})
         self.assertEqual(response.status_code, 404)
 
+    @override_settings(WAGTAILIMAGES_CHOOSER_PAGE_SIZE=4)
+    def test_chooser_page_size(self):
+        images = [
+            Image(
+                title="Test image %i" % i,
+                file=get_test_image_file(size=(1, 1)),
+            )
+            for i in range(1, 12)
+        ]
+        Image.objects.bulk_create(images)
+
+        response = self.get()
+
+        self.assertContains(response, "Page 1 of 3")
+        self.assertEqual(response.status_code, 200)
+
     def test_filter_by_tag(self):
         for i in range(0, 10):
             image = Image.objects.create(

wagtail/images/views/chooser.py

@@ -72,10 +72,15 @@ class ImageCreationFormMixin(CreationFormMixin):
 class BaseImageChooseView(BaseChooseView):
     template_name = "wagtailimages/chooser/chooser.html"
     results_template_name = "wagtailimages/chooser/results.html"
-    per_page = 12
     ordering = "-created_at"
     construct_queryset_hook_name = "construct_image_chooser_queryset"
 
+    @property
+    def per_page(self):
+        # Make per_page into a property so that we can read back WAGTAILIMAGES_CHOOSER_PAGE_SIZE
+        # at runtime.
+        return getattr(settings, "WAGTAILIMAGES_CHOOSER_PAGE_SIZE", 20)
+
     def get_object_list(self):
         return (
             permission_policy.instances_user_has_any_permission_for(
@@ -309,7 +314,6 @@ class ImageChooserViewSet(ChooserViewSet):
     preserve_url_parameters = ChooserViewSet.preserve_url_parameters + ["select_format"]
 
     icon = "image"
-    per_page = getattr(settings, "WAGTAILIMAGES_CHOOSER_PAGE_SIZE", 10)
     choose_one_text = _("Choose an image")
     create_action_label = _("Upload")
     create_action_clicked_label = _("Uploading…")

wagtail/test/testapp/migrations/0037_jsoncustomvaluestreammodel.py

@@ -0,0 +1,24 @@
+# Generated by Django 4.2.11 on 2024-04-22 08:19
+
+from django.db import migrations, models
+import wagtail.blocks
+import wagtail.fields
+import wagtail.images.blocks
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('tests', '0036_complexdefaultstreampage'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='JSONCustomValueStreamModel',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('primary_content', wagtail.fields.StreamField([('text', wagtail.blocks.CharBlock()), ('rich_text', wagtail.blocks.RichTextBlock()), ('image', wagtail.images.blocks.ImageChooserBlock())])),
+                ('secondary_content', wagtail.fields.StreamField([('text', wagtail.blocks.CharBlock()), ('rich_text', wagtail.blocks.RichTextBlock()), ('image', wagtail.images.blocks.ImageChooserBlock())])),
+            ],
+        ),
+    ]

wagtail/test/testapp/models.py

@@ -42,6 +42,7 @@ from wagtail.blocks import (
     RawHTMLBlock,
     RichTextBlock,
     StreamBlock,
+    StreamValue,
     StructBlock,
 )
 from wagtail.contrib.forms.forms import FormBuilder
@@ -1552,6 +1553,40 @@ class JSONBlockCountsStreamModel(models.Model):
     )
 
 
+class CustomStreamValue(StreamValue):
+    """
+    Used by ``CustomStreamBlock`` and ``JSONCustomValueStreamModel.primary_content`` (below)
+    to demonstrate support for custom value classes with ``StreamField`` and ``StreamBlock``.
+    """
+
+    def level_of_customness(self) -> bool:
+        return "medium"
+
+
+class CustomValueStreamBlock(StreamBlock):
+    text = CharBlock()
+    rich_text = RichTextBlock()
+    image = ImageChooserBlock()
+
+    class Meta:
+        value_class = CustomStreamValue
+
+
+class JSONCustomValueStreamModel(models.Model):
+    # `value_class` can be provided as an init kwarg to StreamField
+    primary_content = StreamField(
+        [
+            ("text", CharBlock()),
+            ("rich_text", RichTextBlock()),
+            ("image", ImageChooserBlock()),
+        ],
+        value_class=CustomStreamValue,
+    )
+
+    # `value_class` can be customised by overriding in StreamBlock.Meta
+    secondary_content = StreamField(CustomValueStreamBlock())
+
+
 class ExtendedImageChooserBlock(ImageChooserBlock):
     """
     Example of Block with custom get_api_representation method.

wagtail/tests/test_blocks.py

@@ -20,7 +20,11 @@ from wagtail.blocks.base import get_error_json_data
 from wagtail.blocks.field_block import FieldBlockAdapter
 from wagtail.blocks.list_block import ListBlockAdapter, ListBlockValidationError
 from wagtail.blocks.static_block import StaticBlockAdapter
-from wagtail.blocks.stream_block import StreamBlockAdapter, StreamBlockValidationError
+from wagtail.blocks.stream_block import (
+    StreamBlockAdapter,
+    StreamBlockValidationError,
+    StreamValue,
+)
 from wagtail.blocks.struct_block import StructBlockAdapter, StructBlockValidationError
 from wagtail.models import Page
 from wagtail.rich_text import RichText
@@ -3174,6 +3178,7 @@ class TestStreamBlock(WagtailTestUtils, SimpleTestCase):
         )
 
         self.assertEqual(list(block.child_blocks.keys()), ["heading", "paragraph"])
+        self.assertIs(block.value_class, StreamValue)
 
     def test_initialisation_with_binary_string_names(self):
         # migrations will sometimes write out names as binary strings, just to keep us on our toes
@@ -3186,6 +3191,20 @@ class TestStreamBlock(WagtailTestUtils, SimpleTestCase):
 
         self.assertEqual(list(block.child_blocks.keys()), [b"heading", b"paragraph"])
 
+    def test_initialisation_with_custom_value_class(self):
+        class CustomStreamValue(StreamValue):
+            pass
+
+        block = blocks.StreamBlock(
+            [
+                ("heading", blocks.CharBlock()),
+                ("paragraph", blocks.CharBlock()),
+            ],
+            value_class=CustomStreamValue,
+        )
+
+        self.assertIs(block.value_class, CustomStreamValue)
+
     def test_initialisation_from_subclass(self):
         class ArticleBlock(blocks.StreamBlock):
             heading = blocks.CharBlock()

wagtail/tests/test_streamfield.py

@@ -17,7 +17,9 @@ from wagtail.rich_text import RichText
 from wagtail.signal_handlers import disable_reference_index_auto_update
 from wagtail.test.testapp.models import (
     ComplexDefaultStreamPage,
+    CustomStreamValue,
     JSONBlockCountsStreamModel,
+    JSONCustomValueStreamModel,
     JSONMinMaxCountStreamModel,
     JSONStreamModel,
     StreamPage,
@@ -225,6 +227,38 @@ class TestStreamValueAccess(TestCase):
         self.assertIsInstance(fetched_body[0].value, RichText)
         self.assertEqual(fetched_body[0].value.source, "<h2>hello world</h2>")
 
+    def test_custom_value_class(self):
+        original_content = json.dumps([{"type": "text", "value": "foo"}])
+
+        obj = JSONCustomValueStreamModel.objects.create(
+            primary_content=original_content,
+            secondary_content=original_content,
+        )
+
+        # Both fields should return instances of CustomStreamValue
+        self.assertIsInstance(obj.primary_content, CustomStreamValue)
+        self.assertIsInstance(obj.secondary_content, CustomStreamValue)
+
+        # It should still be possible to update the fields using a raw dict value
+        new_content = [("rich_text", RichText("<h2>hello world</h2>"))]
+        obj.primary_content = new_content
+        obj.secondary_content = new_content
+        obj.save()
+        obj.refresh_from_db("primary_content", "secondary_content")
+
+        # CustomStreamValue is functionally equivalent to StreamValue, so the same value
+        # transformation should have taken place
+        for streamfield in ("primary_content", "secondary_content"):
+            with self.subTest(streamfield):
+                field_value = getattr(self, streamfield)
+                self.assertEqual(len(field_value), 1)
+                self.assertIsInstance(field_value[0].value, RichText)
+                self.assertEqual(field_value[0].value.source, "<h2>hello world</h2>")
+                # The value is still an instance of the custom value class
+                self.assertIsInstance(field_value, CustomStreamValue)
+                # So, we can do this...
+                self.assertEqual(field_value.level_of_customness(), "medium")
+
     def test_can_append(self):
         self.json_body.body.append(("text", "bar"))
         self.json_body.save()