fix(shared_poll): validate refresh proxy data against publication_data_format

Data returned by a shared poll refresh proxy reaches subscribers without ever
being checked against the channel publication_data_format, so a namespace
configured with json/json_object could still deliver payloads clients are not
able to parse.

Validate the refresh result in the shared poll dispatch and reject the whole
batch when any item does not match: the data comes from the application
backend, so a mismatch is a developer mistake and must be loud rather than
partially applied. The log line names the channel, the offending key and the
expected format.

Items without data are skipped - a refresh response carries no data both for
removals and for keys reported as unchanged, and the dispatch can't tell
"unchanged" from "changed to empty" (only the shared poll manager knows the
current version of a key). So json/json_object still catch malformed payloads
while the empty-data rule of the default format does not apply here.

Author: FZambia

internal/client/handler.go

@@ -177,7 +177,14 @@ func (h *Handler) Setup() error {
 			if !ok {
 				return centrifuge.SharedPollResult{}, centrifuge.ErrorInternal
 			}
-			return handler(ctx, event)
+			result, err := handler(ctx, event)
+			if err != nil {
+				return centrifuge.SharedPollResult{}, err
+			}
+			if err := validateSharedPollRefreshData(event.Channel, chOpts.PublicationDataFormat, result.Items); err != nil {
+				return centrifuge.SharedPollResult{}, err
+			}
+			return result, nil
 		})
 	}
 
@@ -979,6 +986,32 @@ func (h *Handler) OnPublish(c Client, e centrifuge.PublishEvent, publishProxyHan
 	return centrifuge.PublishReply{Result: &result}, err
 }
 
+// validateSharedPollRefreshData checks data returned by a shared poll refresh
+// proxy against the channel publication_data_format. Data comes from the
+// application backend, so a format mismatch is a configuration/implementation
+// mistake rather than untrusted input – reject the entire batch loudly instead
+// of delivering payloads subscribers are not able to parse.
+//
+// Items without data are skipped: a refresh response carries no data both for
+// removals and for keys the backend reports as unchanged, and this layer can't
+// tell "unchanged" from "changed to empty" (only the shared poll manager knows
+// the current version of a key). So the empty-data rule of the default format
+// is not applied here, while json/json_object still catch malformed payloads.
+func validateSharedPollRefreshData(channel string, format string, items []centrifuge.SharedPollRefreshItem) error {
+	for _, item := range items {
+		if item.Removed || len(item.Data) == 0 {
+			continue
+		}
+		if err := config.ValidatePublicationData(item.Data, format); err != nil {
+			log.Error().Err(err).Str("channel", channel).Str("key", item.Key).
+				Str("publication_data_format", format).
+				Msg("shared poll refresh data does not match channel publication_data_format, rejecting the whole refresh batch – fix the data returned by the shared poll refresh proxy")
+			return centrifuge.ErrorBadRequest
+		}
+	}
+	return nil
+}
+
 // OnMapPublish ...
 func (h *Handler) OnMapPublish(c Client, e centrifuge.MapPublishEvent, mapPublishProxyHandler proxy.MapPublishHandlerFunc) (centrifuge.MapPublishReply, error) {
 	cfg := h.cfgContainer.Config()

internal/client/handler_test.go

@@ -1826,3 +1826,51 @@ func TestOnMapPublishDataFormat(t *testing.T) {
 		})
 	}
 }
+
+// TestValidateSharedPollRefreshData ensures a shared poll refresh batch is
+// rejected as a whole when the backend returns data not matching the channel
+// publication_data_format, while removals and data-less "unchanged" items pass.
+func TestValidateSharedPollRefreshData(t *testing.T) {
+	const ch = "sp:test"
+
+	t.Run("valid batch accepted", func(t *testing.T) {
+		err := validateSharedPollRefreshData(ch, configtypes.PublicationDataFormatJSONObject, []centrifuge.SharedPollRefreshItem{
+			{Key: "k1", Data: []byte(`{"v":1}`), Version: 1},
+			{Key: "k2", Data: []byte(`{"v":2}`), Version: 1},
+		})
+		require.NoError(t, err)
+	})
+
+	t.Run("one bad item rejects whole batch", func(t *testing.T) {
+		err := validateSharedPollRefreshData(ch, configtypes.PublicationDataFormatJSONObject, []centrifuge.SharedPollRefreshItem{
+			{Key: "k1", Data: []byte(`{"v":1}`), Version: 1},
+			{Key: "k2", Data: []byte(`[1,2]`), Version: 1},
+			{Key: "k3", Data: []byte(`{"v":3}`), Version: 1},
+		})
+		require.Equal(t, centrifuge.ErrorBadRequest, err)
+	})
+
+	t.Run("invalid json rejected", func(t *testing.T) {
+		err := validateSharedPollRefreshData(ch, configtypes.PublicationDataFormatJSON, []centrifuge.SharedPollRefreshItem{
+			{Key: "k1", Data: []byte(`not json`), Version: 1},
+		})
+		require.Equal(t, centrifuge.ErrorBadRequest, err)
+	})
+
+	t.Run("removals and data-less items skipped", func(t *testing.T) {
+		// Removals carry no data, and an unchanged key may come back without
+		// data — neither must trip the empty-data rule of the default format.
+		err := validateSharedPollRefreshData(ch, "", []centrifuge.SharedPollRefreshItem{
+			{Key: "k1", Removed: true},
+			{Key: "k2", Version: 7},
+		})
+		require.NoError(t, err)
+	})
+
+	t.Run("binary format allows anything", func(t *testing.T) {
+		err := validateSharedPollRefreshData(ch, configtypes.PublicationDataFormatBinary, []centrifuge.SharedPollRefreshItem{
+			{Key: "k1", Data: []byte{0x00, 0x01, 0xff}, Version: 1},
+		})
+		require.NoError(t, err)
+	})
+}