fix(disagg): use JSON instead of pickle for P2P ZMQ requests (#4804)

[Anai-Guo]

Motivation

Fixes #4804.

The disaggregated-serving P2P connector deserializes peer-supplied bytes with pickle:

# lmdeploy/pytorch/disagg/conn/engine_conn.py
req = await self.p2p_receiver[remote_engine_id].recv_pyobj()   # = pickle.loads()
if isinstance(req, DistServeCacheFreeRequest):                 # check runs *after* loads

recv_pyobj() calls pickle.loads() on the raw frame, so the isinstance guard runs only after any __reduce__ payload has already executed. As analysed in the issue, the reachable path is not the bound PUSH socket but the HTTP endpoint: an unauthenticated POST /distserve/p2p_connect supplies zmq_address, the engine's PULL socket connect()s to that caller-chosen address, and handle_zmq_recv then pickle.loads() whatever that address sends → remote code execution on the engine node (AuthenticationMiddleware is only installed when api_keys is set, and the /distserve/* routes carry no auth dependency).

Modification

The connector's only wire message is DistServeCacheFreeRequest, a pydantic model with two scalar fields (remote_engine_id: str, remote_session_id: int). Both send and receive live in the same class and upgrade together, so switching the codec has no protocol-mismatch risk:

• zmq_send: send_pyobj(...) → send_json(req.model_dump())
• handle_zmq_recv: recv_pyobj() → recv_json() + DistServeCacheFreeRequest.model_validate(...)

recv_json() decodes with json.loads (no code execution) and model_validate enforces the schema before the payload is used, making the old isinstance check redundant. Malformed or off-schema messages are now logged and skipped rather than raising out of the bare while True in the detached task (which previously surfaced only as a task-destroyed warning).

No wire-format compatibility break beyond the coordinated both-ends change here.

Checklist

• Pre-commit / code style aligned with the existing file
• Reproduced the deserialization path against main
• Change is confined to engine_conn.py

🤖 Generated with Claude Code

[Copilot]

Pull request overview

This PR mitigates a remote code execution risk in the disaggregated-serving P2P (ZMQ) connector by removing pickle-based deserialization and replacing it with JSON + schema validation in EngineP2PConnection.

Changes:

• Switch P2P ZMQ message sending from send_pyobj() (pickle) to send_json() using DistServeCacheFreeRequest.model_dump().
• Switch receiving from recv_pyobj() (pickle.loads) to recv_json() followed by DistServeCacheFreeRequest.model_validate(...).
• Add error handling to log and skip malformed/off-schema messages rather than crashing the receive loop.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[caikun-pjlab]

LGTM