How to use custom JSON encoder with PubSub managers after pickle deprecation?

[allen0099]

Following the fix in #1502, pickle has been replaced with engineio.json to address the RCE vulnerability.
However, this change prevents serialization of custom types (such as Arrow timestamps or other custom objects).

Would it be possible to use the JSON encoder from the server instance instead? This would allow custom encoders configured on the server to be respected by the manager.

For example, in AsyncRedisManager:

class AsyncRedisManager(AsyncPubSubManager):
    async def _publish(self, data):  # pragma: no cover
        _, error = self._get_redis_module_and_error()
        for retries_left in range(1, -1, -1):  # 2 attempts
            try:
                if not self.connected:
                    self._redis_connect()
-                return await self.redis.publish(
-                    self.channel, json.dumps(data))
+                return await self.redis.publish(
+                    self.channel, self.server.packet_class.json.dumps(data))
            except error as exc:
                if retries_left > 0:
                    self._get_logger().error(
                        'Cannot publish to redis... '
                        'retrying',
                        extra={"redis_exception": str(exc)})
                    self.connected = False
                else:
                    self._get_logger().error(
                        'Cannot publish to redis... '
                        'giving up',
                        extra={"redis_exception": str(exc)})

                    break

Questions:

• Is this approach safe and architecturally sound?
• Are there any potential issues with accessing self.server.packet_class.json from the manager?
• Would there be a better way to support custom JSON encoders while maintaining security?

[miguelgrinberg]

I'm not sure I understand what the problem is, because packet_class.json is set to engineio.json. To use a custom JSON encoder you have to pass it when you declare your server in the json argument. All parts of the library then will use it. If you have found a situation in which the custom JSON encoder that you set in the server is not used, that's a bug so let me know.

[allen0099]

Thank you for the clarification! I did pass a custom JSON encoder to the server via the json argument. However, I've found that the AsyncRedisManager doesn't seem to use it.

I've created a minimal reproduction case to demonstrate the issue:

minimal reproduction

import decimal
from collections.abc import Callable
from typing import Any

import orjson
from fastapi import FastAPI
from pydantic import BaseModel
from socketio import ASGIApp
from socketio import AsyncNamespace
from socketio import AsyncRedisManager
from socketio import AsyncServer


class SampleModel(BaseModel):
    id: int
    name: str
    value: decimal.Decimal


class SampleNamespace(AsyncNamespace):
    """Lightweight namespace for testing purposes."""

    async def on_connect(self, sid: str, environ: dict[str, Any]) -> bool:
        return True

    async def on_disconnect(self, sid: str) -> None:
        pass

    async def on_ping(self, sid: str, data: Any | None = None) -> dict[str, Any]:  # noqa: ARG002
        return {"status": "ok"}


class CustomJsonEncoder:
    @classmethod
    def _default(cls, obj: Any) -> Any:
        # Test Decimal, since json doesn't support it natively
        if isinstance(obj, decimal.Decimal):
            return str(obj)
        raise TypeError

    @classmethod
    def loads(cls, __obj: bytes | bytearray | memoryview | str) -> Any:
        return orjson.loads(__obj)

    @classmethod
    def dumps(
        cls,
        __obj: Any,
        separators: tuple[str, str] = (",", ":"),  # noqa: ARG003
        default: Callable[[Any], Any] | None = None,
        option: int | None = None,
    ) -> str:
        serializer = default or cls._default
        return orjson.dumps(__obj, default=serializer, option=option).decode()


app = FastAPI()

# Set up Socket.IO server with Redis manager and custom JSON encoder
manager = AsyncRedisManager("redis://localhost:6379/0")
server = AsyncServer(
    async_mode="asgi",
    client_manager=manager,
    json=CustomJsonEncoder,
    namespaces=["/sample"],
)

# Register the sample namespace
server.register_namespace(SampleNamespace())

app.mount("/socket.io", ASGIApp(server, other_asgi_app=app))


@app.get("/socket-test")
async def socket_test():
    # A simple emit to check socket.io functionality
    await server.emit("ping", "pong", namespace="/sample")

    # Send a sample model
    sample = SampleModel(
        id=1,
        name="Test",
        value=decimal.Decimal("123.45"),
    )
    # Emit the sample model, which includes a Decimal field
    await server.emit("sample", sample.model_dump(), namespace="/sample")

error stacks

  File "C:\xxx\main.py", line 86, in socket_test
    await server.emit("sample", sample.model_dump(), namespace="/sample")
  File "C:\xxx\.venv\Lib\site-packages\socketio\async_server.py", line 179, in emit
    await self.manager.emit(event, data, namespace, room=room,
                            skip_sid=skip_sid, callback=callback,
                            ignore_queue=ignore_queue)
  File "C:\xxx\.venv\Lib\site-packages\socketio\async_pubsub_manager.py", line 82, in emit
    await self._publish(message)  # notify other hosts
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\xxx\.venv\Lib\site-packages\socketio\async_redis_manager.py", line 120, in _publish
    self.channel, json.dumps(data))
                  ~~~~~~~~~~^^^^^^

The code uses json.dumps(data) directly (which references engineio.json), rather than using the custom JSON encoder configured on the server.

I believe this might be a bug, but please let me know if I'm misusing something or if there's a different approach I should take.

[miguelgrinberg]

Thanks. You are correct, this is a bug and I will fix it. Added #1549 to track it.

[miguelgrinberg]

@allen0099 Could you try the main branch of this repository and confirm that custom JSON settings now apply to the pubsub managers as well? Thanks!

[allen0099]

@miguelgrinberg Thank you for your fixing, the managers now works correctly with custom JSON encoder.

[miguelgrinberg]

Awesome. I will get a release out soon. Thanks!