From 72016a09e789038309ca40d16bcad4b7a7d7373e Mon Sep 17 00:00:00 2001 From: MrlixiangWE <102979255+MrlixiangWE@users.noreply.github.com> Date: Tue, 11 Aug 2026 22:55:28 +0800 Subject: [PATCH] fix: avoid reusing released APIServer response buffers --- .../apiserver/http/handler/KillHandler.java | 14 +++--- .../apiserver/http/handler/PubHandler.java | 17 ++++--- .../http/handler/KillHandlerTest.java | 32 +++++++----- .../http/handler/PubHandlerTest.java | 49 ++++++++++++++++++- 4 files changed, 84 insertions(+), 28 deletions(-) diff --git a/bifromq-apiserver/src/main/java/org/apache/bifromq/apiserver/http/handler/KillHandler.java b/bifromq-apiserver/src/main/java/org/apache/bifromq/apiserver/http/handler/KillHandler.java index 34f7a7dec..017a4d53a 100644 --- a/bifromq-apiserver/src/main/java/org/apache/bifromq/apiserver/http/handler/KillHandler.java +++ b/bifromq-apiserver/src/main/java/org/apache/bifromq/apiserver/http/handler/KillHandler.java @@ -31,7 +31,6 @@ import static org.apache.bifromq.apiserver.http.handler.utils.HeaderUtils.getHeader; import com.google.common.base.Strings; -import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.handler.codec.http.DefaultFullHttpResponse; import io.netty.handler.codec.http.FullHttpRequest; @@ -47,6 +46,7 @@ import io.swagger.v3.oas.annotations.responses.ApiResponses; import jakarta.ws.rs.DELETE; import jakarta.ws.rs.Path; +import java.nio.charset.StandardCharsets; import java.util.Map; import java.util.Set; import java.util.concurrent.CompletableFuture; @@ -65,10 +65,8 @@ final class KillHandler extends TenantAwareHandler { private static final String SERVER_REDIRECT_VALUE_TEMP_USE = "temp_use"; private static final Set SERVER_REDIRECT_VALUES = Set.of(SERVER_REDIRECT_VALUE_NO, SERVER_REDIRECT_VALUE_MOVE, SERVER_REDIRECT_VALUE_TEMP_USE); - private static final ByteBuf INVALID_SERVER_REDIRECT = - Unpooled.wrappedBuffer("Invalid server redirect value".getBytes()); - private static final ByteBuf TOO_LONG_SERVER_REFERENCE = - Unpooled.wrappedBuffer("Server reference exceeds 65535 bytes".getBytes()); + private static final String INVALID_SERVER_REDIRECT = "Invalid server redirect value"; + private static final String TOO_LONG_SERVER_REFERENCE = "Server reference exceeds 65535 bytes"; private final ISessionDictClient sessionDictClient; KillHandler(ISettingProvider settingProvider, ISessionDictClient sessionDictClient) { @@ -119,11 +117,13 @@ public CompletableFuture handle(@Parameter(hidden = true) long Map clientMeta = getClientMeta(req); if (serverRedirect != null && !SERVER_REDIRECT_VALUES.contains(serverRedirect)) { return CompletableFuture.completedFuture( - new DefaultFullHttpResponse(req.protocolVersion(), BAD_REQUEST, INVALID_SERVER_REDIRECT)); + new DefaultFullHttpResponse(req.protocolVersion(), BAD_REQUEST, + Unpooled.copiedBuffer(INVALID_SERVER_REDIRECT, StandardCharsets.UTF_8))); } if (serverReference != null && serverReference.length() > MAX_SERVER_REFERENCE_LENGTH) { return CompletableFuture.completedFuture( - new DefaultFullHttpResponse(req.protocolVersion(), BAD_REQUEST, TOO_LONG_SERVER_REFERENCE)); + new DefaultFullHttpResponse(req.protocolVersion(), BAD_REQUEST, + Unpooled.copiedBuffer(TOO_LONG_SERVER_REFERENCE, StandardCharsets.UTF_8))); } ServerRedirection serverRedirection = buildServerRedirection(serverRedirect, serverReference); diff --git a/bifromq-apiserver/src/main/java/org/apache/bifromq/apiserver/http/handler/PubHandler.java b/bifromq-apiserver/src/main/java/org/apache/bifromq/apiserver/http/handler/PubHandler.java index 8efa9da83..c0c2f332e 100644 --- a/bifromq-apiserver/src/main/java/org/apache/bifromq/apiserver/http/handler/PubHandler.java +++ b/bifromq-apiserver/src/main/java/org/apache/bifromq/apiserver/http/handler/PubHandler.java @@ -30,7 +30,6 @@ import static org.apache.bifromq.apiserver.http.handler.utils.HeaderUtils.getHeader; import com.google.protobuf.ByteString; -import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.handler.codec.http.DefaultFullHttpResponse; import io.netty.handler.codec.http.FullHttpRequest; @@ -46,6 +45,7 @@ import io.swagger.v3.oas.annotations.responses.ApiResponses; import jakarta.ws.rs.POST; import jakarta.ws.rs.Path; +import java.nio.charset.StandardCharsets; import java.util.Map; import java.util.Optional; import java.util.concurrent.CompletableFuture; @@ -61,9 +61,9 @@ @Path("/pub") final class PubHandler extends TenantAwareHandler { - private static final ByteBuf UNACCEPTED_TOPIC = Unpooled.wrappedBuffer("Unaccepted Topic".getBytes()); - private static final ByteBuf INVALID_QOS = Unpooled.wrappedBuffer("Invalid QoS".getBytes()); - private static final ByteBuf INVALID_EXPIRY_SECONDS = Unpooled.wrappedBuffer("Invalid expiry seconds".getBytes()); + private static final String UNACCEPTED_TOPIC = "Unaccepted Topic"; + private static final String INVALID_QOS = "Invalid QoS"; + private static final String INVALID_EXPIRY_SECONDS = "Invalid expiry seconds"; private final IDistClient distClient; private final ISettingProvider settingProvider; @@ -114,15 +114,18 @@ public CompletableFuture handle(@Parameter(hidden = true) long .orElse(Integer.MAX_VALUE); if (!TopicUtil.checkTopicFilter(topic, tenantId, settingProvider)) { return CompletableFuture.completedFuture( - new DefaultFullHttpResponse(req.protocolVersion(), FORBIDDEN, UNACCEPTED_TOPIC)); + new DefaultFullHttpResponse(req.protocolVersion(), FORBIDDEN, + Unpooled.copiedBuffer(UNACCEPTED_TOPIC, StandardCharsets.UTF_8))); } if (qos < 0 || qos > 2) { return CompletableFuture.completedFuture( - new DefaultFullHttpResponse(req.protocolVersion(), BAD_REQUEST, INVALID_QOS)); + new DefaultFullHttpResponse(req.protocolVersion(), BAD_REQUEST, + Unpooled.copiedBuffer(INVALID_QOS, StandardCharsets.UTF_8))); } if (expirySeconds <= 0) { return CompletableFuture.completedFuture( - new DefaultFullHttpResponse(req.protocolVersion(), BAD_REQUEST, INVALID_EXPIRY_SECONDS)); + new DefaultFullHttpResponse(req.protocolVersion(), BAD_REQUEST, + Unpooled.copiedBuffer(INVALID_EXPIRY_SECONDS, StandardCharsets.UTF_8))); } String clientType = getHeader(HEADER_CLIENT_TYPE, req, true); Map clientMeta = getClientMeta(req); diff --git a/bifromq-apiserver/src/test/java/org/apache/bifromq/apiserver/http/handler/KillHandlerTest.java b/bifromq-apiserver/src/test/java/org/apache/bifromq/apiserver/http/handler/KillHandlerTest.java index d1f2c0995..1f20146cb 100644 --- a/bifromq-apiserver/src/test/java/org/apache/bifromq/apiserver/http/handler/KillHandlerTest.java +++ b/bifromq-apiserver/src/test/java/org/apache/bifromq/apiserver/http/handler/KillHandlerTest.java @@ -35,12 +35,12 @@ import static org.mockito.Mockito.when; import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertThrows; -import static org.testng.Assert.assertTrue; import io.netty.handler.codec.http.DefaultFullHttpRequest; import io.netty.handler.codec.http.FullHttpResponse; import io.netty.handler.codec.http.HttpMethod; import io.netty.handler.codec.http.HttpResponseStatus; +import java.nio.charset.StandardCharsets; import java.util.concurrent.CompletableFuture; import org.apache.bifromq.sessiondict.client.ISessionDictClient; import org.apache.bifromq.sessiondict.rpc.proto.KillReply; @@ -254,12 +254,7 @@ public void invalidServerRedirect() { long reqId = 123; String tenantId = "bifromq_dev"; - KillHandler handler = new KillHandler(settingProvider, sessionDictClient); - handler.handle(reqId, tenantId, req); - FullHttpResponse response = handler.handle(reqId, tenantId, req).join(); - assertEquals(response.protocolVersion(), req.protocolVersion()); - assertEquals(response.status(), HttpResponseStatus.BAD_REQUEST); - assertTrue(response.content().readableBytes() > 0); + assertRepeatedValidationResponse(reqId, tenantId, req, "Invalid server redirect value"); } @Test @@ -275,15 +270,26 @@ public void tooLongServerReference() { long reqId = 123; String tenantId = "bifromq_dev"; - KillHandler handler = new KillHandler(settingProvider, sessionDictClient); - handler.handle(reqId, tenantId, req); - FullHttpResponse response = handler.handle(reqId, tenantId, req).join(); - assertEquals(response.protocolVersion(), req.protocolVersion()); - assertEquals(response.status(), HttpResponseStatus.BAD_REQUEST); - assertTrue(response.content().readableBytes() > 0); + assertRepeatedValidationResponse(reqId, tenantId, req, "Server reference exceeds 65535 bytes"); } private DefaultFullHttpRequest buildRequest() { return buildRequest(HttpMethod.DELETE); } + + private void assertRepeatedValidationResponse(long reqId, String tenantId, DefaultFullHttpRequest req, + String expectedContent) { + KillHandler handler = new KillHandler(settingProvider, sessionDictClient); + FullHttpResponse firstResponse = handler.handle(reqId, tenantId, req).join(); + firstResponse.release(); + + FullHttpResponse secondResponse = handler.handle(reqId + 1, tenantId, req).join(); + try { + assertEquals(secondResponse.protocolVersion(), req.protocolVersion()); + assertEquals(secondResponse.status(), HttpResponseStatus.BAD_REQUEST); + assertEquals(secondResponse.content().toString(StandardCharsets.UTF_8), expectedContent); + } finally { + secondResponse.release(); + } + } } diff --git a/bifromq-apiserver/src/test/java/org/apache/bifromq/apiserver/http/handler/PubHandlerTest.java b/bifromq-apiserver/src/test/java/org/apache/bifromq/apiserver/http/handler/PubHandlerTest.java index c51f4472c..748f91782 100644 --- a/bifromq-apiserver/src/test/java/org/apache/bifromq/apiserver/http/handler/PubHandlerTest.java +++ b/bifromq-apiserver/src/test/java/org/apache/bifromq/apiserver/http/handler/PubHandlerTest.java @@ -51,6 +51,7 @@ import io.netty.handler.codec.http.HttpMethod; import io.netty.handler.codec.http.HttpResponseStatus; import io.netty.handler.codec.http.HttpVersion; +import java.nio.charset.StandardCharsets; import java.util.concurrent.CompletableFuture; import org.mockito.Mock; import org.testng.annotations.Test; @@ -128,11 +129,42 @@ public void pubWithWrongQoS() { } @Test - public void pubWithWrongExpirySeconds() { + public void repeatedInvalidQoSResponseHasReadableContent() { DefaultFullHttpRequest req = buildRequest(); req.headers().set(HEADER_TOPIC.header, "/greeting"); req.headers().set(HEADER_CLIENT_TYPE.header, "admin_team"); req.headers().set(HEADER_QOS.header, "3"); + + assertRepeatedValidationResponse(req, HttpResponseStatus.BAD_REQUEST, "Invalid QoS"); + } + + @Test + public void repeatedUnacceptedTopicResponseHasReadableContent() { + DefaultFullHttpRequest req = buildRequest(); + req.headers().set(HEADER_TOPIC.header, "a".repeat(256)); + req.headers().set(HEADER_CLIENT_TYPE.header, "admin_team"); + req.headers().set(HEADER_QOS.header, "1"); + + assertRepeatedValidationResponse(req, HttpResponseStatus.FORBIDDEN, "Unaccepted Topic"); + } + + @Test + public void repeatedInvalidExpiryResponseHasReadableContent() { + DefaultFullHttpRequest req = buildRequest(); + req.headers().set(HEADER_TOPIC.header, "/greeting"); + req.headers().set(HEADER_CLIENT_TYPE.header, "admin_team"); + req.headers().set(HEADER_QOS.header, "1"); + req.headers().set(HEADER_EXPIRY_SECONDS.header, "0"); + + assertRepeatedValidationResponse(req, HttpResponseStatus.BAD_REQUEST, "Invalid expiry seconds"); + } + + @Test + public void pubWithWrongExpirySeconds() { + DefaultFullHttpRequest req = buildRequest(); + req.headers().set(HEADER_TOPIC.header, "/greeting"); + req.headers().set(HEADER_CLIENT_TYPE.header, "admin_team"); + req.headers().set(HEADER_QOS.header, "1"); req.headers().set(HEADER_CLIENT_META_PREFIX + "age", "4"); req.headers().set(HEADER_EXPIRY_SECONDS.header, "0"); long reqId = 123; @@ -195,4 +227,19 @@ public void tooLargeRequest() { private DefaultFullHttpRequest buildRequest() { return buildRequest(HttpMethod.POST); } + + private void assertRepeatedValidationResponse(DefaultFullHttpRequest req, HttpResponseStatus expectedStatus, + String expectedContent) { + PubHandler handler = new PubHandler(settingProvider, distClient); + FullHttpResponse firstResponse = handler.handle(123, "bifromq_dev", req).join(); + firstResponse.release(); + + FullHttpResponse secondResponse = handler.handle(124, "bifromq_dev", req).join(); + try { + assertEquals(secondResponse.status(), expectedStatus); + assertEquals(secondResponse.content().toString(StandardCharsets.UTF_8), expectedContent); + } finally { + secondResponse.release(); + } + } }