From cd94b21264a76c484a567566658995fd3a84db65 Mon Sep 17 00:00:00 2001 From: aoshen02 Date: Wed, 30 Sep 2026 00:46:56 +0800 Subject: [PATCH] [Bugfix][Frontend] Return 400 for malformed RL dev route bodies (#59236) Signed-off-by: aoshen02 Co-authored-by: aoshen02 Co-authored-by: Claude Opus 5.5 --- .../test_dev_route_request_bodies.py | 96 +++++++++++++++++++ vllm/entrypoints/serve/dev/rlhf/api_router.py | 37 +++---- vllm/entrypoints/serve/dev/rpc/api_router.py | 5 + 3 files changed, 115 insertions(+), 23 deletions(-) create mode 100644 tests/entrypoints/unit_tests/test_dev_route_request_bodies.py diff --git a/tests/entrypoints/unit_tests/test_dev_route_request_bodies.py b/tests/entrypoints/unit_tests/test_dev_route_request_bodies.py new file mode 100644 index 0000000000..78fe5eef5d --- /dev/null +++ b/tests/entrypoints/unit_tests/test_dev_route_request_bodies.py @@ -0,0 +1,96 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +"""Malformed bodies on RL dev routes are 400s that never reach the engine.""" + +import json +from types import SimpleNamespace + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from vllm.entrypoints.serve.dev.rlhf.api_router import router as rlhf_router +from vllm.entrypoints.serve.dev.rpc.api_router import router as rpc_router +from vllm.entrypoints.serve.exception_handling.register import init_exception_handler + +pytestmark = pytest.mark.cpu_test + + +class Engine: + def __init__(self): + self.calls: list[str] = [] + + def __getattr__(self, name): + async def call(*args, **kwargs): + self.calls.append(name) + + return call + + +@pytest.fixture +def client(): + app = FastAPI() + app.include_router(rlhf_router) + app.include_router(rpc_router) + init_exception_handler(app) + app.state.engine_client = Engine() + app.state.args = SimpleNamespace(log_error_stack=False) + return TestClient(app) + + +def post(client, path, body): + return client.post( + path, content=json.dumps(body), headers={"content-type": "application/json"} + ) + + +@pytest.mark.parametrize( + "path", + [ + "/abort_requests", + "/init_weight_transfer_engine", + "/update_weights", + "/collective_rpc", + ], +) +@pytest.mark.parametrize("body", [[], 1, "x", None]) +def test_non_object_body_is_rejected(client, path, body): + assert post(client, path, body).status_code == 400 + assert client.app.state.engine_client.calls == [] + + +@pytest.mark.parametrize( + "path,body", + [ + ("/init_weight_transfer_engine", {}), + ("/init_weight_transfer_engine", {"init_info": []}), + ("/init_weight_transfer_engine", {"init_info": 1}), + ("/update_weights", {}), + ("/update_weights", {"update_info": 1}), + ("/update_weights", {"update_info": [1, 2]}), + ("/update_weights", {"update_info": "x"}), + ], +) +def test_invalid_field_is_rejected_before_the_engine(client, path, body): + # Reaching the engine would also abort an in-progress weight update. + assert post(client, path, body).status_code == 400 + assert client.app.state.engine_client.calls == [] + + +@pytest.mark.parametrize( + "path,body,engine_call", + [ + ("/abort_requests", {"request_ids": ["a"]}, "abort"), + ( + "/init_weight_transfer_engine", + {"init_info": {}}, + "init_weight_transfer_engine", + ), + ("/update_weights", {"update_info": {"names": []}}, "update_weights"), + ("/update_weights", {"update_info": [{}, {}]}, "update_weights"), + ("/collective_rpc", {"method": "m"}, "collective_rpc"), + ], +) +def test_valid_body_reaches_the_engine(client, path, body, engine_call): + assert post(client, path, body).status_code == 200 + assert client.app.state.engine_client.calls == [engine_call] diff --git a/vllm/entrypoints/serve/dev/rlhf/api_router.py b/vllm/entrypoints/serve/dev/rlhf/api_router.py index 603fdf785c..7af2ead22c 100644 --- a/vllm/entrypoints/serve/dev/rlhf/api_router.py +++ b/vllm/entrypoints/serve/dev/rlhf/api_router.py @@ -3,13 +3,14 @@ import json from http import HTTPStatus -from typing import Annotated +from typing import Annotated, Any from fastapi import APIRouter, Body, FastAPI, HTTPException, Query, Request from fastapi.responses import JSONResponse from vllm.distributed.weight_transfer.base import ( WeightTransferInitRequest, + WeightTransferUpdatePayload, WeightTransferUpdateRequest, ) from vllm.engine.protocol import EngineClient @@ -104,6 +105,10 @@ async def abort_requests(raw_request: Request) -> JSONResponse: body = await raw_request.json() except json.JSONDecodeError as e: raise HTTPException(status_code=400, detail="Invalid JSON format") from e # noqa: B904 + if not isinstance(body, dict): + raise HTTPException( + status_code=400, detail="Request body must be a JSON object" + ) request_ids = body.get("request_ids") @@ -154,17 +159,10 @@ async def is_paused(raw_request: Request) -> JSONResponse: @router.post("/init_weight_transfer_engine") -async def init_weight_transfer_engine(raw_request: Request): - try: - body = await raw_request.json() - except json.JSONDecodeError as e: - raise HTTPException(status_code=400, detail="Invalid JSON format") from e # noqa: B904 - init_info = body.get("init_info") - if init_info is None: - raise HTTPException( - status_code=HTTPStatus.BAD_REQUEST.value, - detail="Missing 'init_info' in request body", - ) +async def init_weight_transfer_engine( + raw_request: Request, + init_info: Annotated[dict[str, Any], Body(embed=True)], +): await engine_client(raw_request).init_weight_transfer_engine( WeightTransferInitRequest(init_info=init_info) ) @@ -184,17 +182,10 @@ async def start_draft_weight_update(raw_request: Request): @router.post("/update_weights") -async def update_weights(raw_request: Request): - try: - body = await raw_request.json() - except json.JSONDecodeError as e: - raise HTTPException(status_code=400, detail="Invalid JSON format") from e # noqa: B904 - update_info = body.get("update_info") - if update_info is None: - raise HTTPException( - status_code=HTTPStatus.BAD_REQUEST.value, - detail="Missing 'update_info' in request body", - ) +async def update_weights( + raw_request: Request, + update_info: Annotated[WeightTransferUpdatePayload, Body(embed=True)], +): await engine_client(raw_request).update_weights( request=WeightTransferUpdateRequest(update_info=update_info) ) diff --git a/vllm/entrypoints/serve/dev/rpc/api_router.py b/vllm/entrypoints/serve/dev/rpc/api_router.py index 99b904c2f6..468e876508 100644 --- a/vllm/entrypoints/serve/dev/rpc/api_router.py +++ b/vllm/entrypoints/serve/dev/rpc/api_router.py @@ -29,6 +29,11 @@ async def collective_rpc(raw_request: Request): status_code=HTTPStatus.BAD_REQUEST.value, detail=f"JSON decode error: {e}", ) from e + if not isinstance(body, dict): + raise HTTPException( + status_code=HTTPStatus.BAD_REQUEST.value, + detail="Request body must be a JSON object", + ) method = body.get("method") if method is None: raise HTTPException(