From 4a24e4e97b2a5f8a75c39267222b5e268692433f Mon Sep 17 00:00:00 2001 From: Sahana Bogar Date: Sun, 18 Oct 2026 14:21:45 +0530 Subject: [PATCH 1/4] do not serve rtmp on ServerOptions.internal_port --- src/brpc/policy/rtmp_protocol.cpp | 7 ++++ test/brpc_rtmp_unittest.cpp | 65 +++++++++++++++++++++++++++++++ 2 files changed, 72 insertions(+) diff --git a/src/brpc/policy/rtmp_protocol.cpp b/src/brpc/policy/rtmp_protocol.cpp index 41ed2422d8..bf46922f78 100644 --- a/src/brpc/policy/rtmp_protocol.cpp +++ b/src/brpc/policy/rtmp_protocol.cpp @@ -3482,6 +3482,13 @@ ParseResult ParseRtmpMessage(butil::IOBuf* source, Socket *socket, bool read_eof // to avoid the confusion from root. return MakeParseError(PARSE_ERROR_TRY_OTHERS); } + if (IsInternalPort(*server, socket->local_side())) { + // ServerOptions.internal_port serves builtin and Tabbed services + // only and an RtmpService is neither. RTMP is handled right here + // without a Controller to reject the request with, so behave as + // if rtmp was not enabled on this port at all. + return MakeParseError(PARSE_ERROR_TRY_OTHERS); + } if (read_eof) { // No former RtmpContext when we read an EOF return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA); diff --git a/test/brpc_rtmp_unittest.cpp b/test/brpc_rtmp_unittest.cpp index 36e996b063..b1249382c7 100644 --- a/test/brpc_rtmp_unittest.cpp +++ b/test/brpc_rtmp_unittest.cpp @@ -1286,6 +1286,71 @@ TEST(RtmpTest, reject_unauthenticated_connection) { server.Join(); } +class RejectAllAuthenticator : public brpc::Authenticator { +public: + int GenerateCredential(std::string*) const override { return 0; } + int VerifyCredential(const std::string&, const butil::EndPoint&, + brpc::AuthContext*) const override { return -1; } +}; + +// Returns a port nothing is listening on, or -1. +static int PickUnusedPort() { + butil::fd_guard sockfd(butil::tcp_listen(butil::EndPoint(butil::IP_ANY, 0))); + if (sockfd < 0) { + return -1; + } + butil::EndPoint point; + if (butil::get_local_side(sockfd, &point) != 0) { + return -1; + } + return point.port; +} + +TEST(RtmpTest, not_served_on_internal_port) { + PlayingDummyService rtmp_service; + RejectAllAuthenticator auth; + brpc::Server server; + brpc::ServerOptions server_opt; + server_opt.rtmp_service = &rtmp_service; + server_opt.auth = &auth; + bool started = false; + for (int i = 0; i < 10 && !started; ++i) { + server_opt.internal_port = PickUnusedPort(); + started = (server_opt.internal_port >= 0 && + 0 == server.Start("127.0.0.1:0", &server_opt)); + } + ASSERT_TRUE(started); + + butil::EndPoint internal_point; + ASSERT_EQ(0, butil::str2endpoint( + "127.0.0.1", server_opt.internal_port, &internal_point)); + butil::fd_guard fd(butil::tcp_connect(internal_point, nullptr)); + ASSERT_GE(fd, 0); + timeval tv = { 2, 0 }; + ASSERT_EQ(0, setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv))); + + // Builtin services on internal_port pass authentication without any + // credential, which latches the connection as authenticated. + const std::string http_req = "GET /version HTTP/1.1\r\n\r\n"; + ASSERT_EQ((ssize_t)http_req.size(), + write(fd, http_req.data(), http_req.size())); + char resp[1024]; + ASSERT_GT(read(fd, resp, sizeof(resp)), 0); + ASSERT_EQ(0, memcmp(resp, "HTTP/1.1 200", 12)); + + // internal_port serves builtin and Tabbed services only, an RTMP + // handshake on the same connection must not reach rtmp_service. + char c0c1[1 + 1536]; + memset(c0c1, 0, sizeof(c0c1)); + c0c1[0] = 0x03; + ASSERT_EQ((ssize_t)sizeof(c0c1), write(fd, c0c1, sizeof(c0c1))); + char s0 = 0; + ASSERT_NE(1, read(fd, &s0, 1)); + + server.Stop(0); + server.Join(); +} + TEST(RtmpTest, successfully_play_streams) { PlayingDummyService rtmp_service; brpc::Server server; From eb7fbffa560fa6f1dca0604a7d631da1adab1bff Mon Sep 17 00:00:00 2001 From: Sahana Bogar Date: Mon, 19 Oct 2026 01:37:55 +0530 Subject: [PATCH 2/4] read the whole status prefix before comparing it in not_served_on_internal_port --- test/brpc_rtmp_unittest.cpp | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/test/brpc_rtmp_unittest.cpp b/test/brpc_rtmp_unittest.cpp index b1249382c7..fafa347bad 100644 --- a/test/brpc_rtmp_unittest.cpp +++ b/test/brpc_rtmp_unittest.cpp @@ -1334,9 +1334,18 @@ TEST(RtmpTest, not_served_on_internal_port) { const std::string http_req = "GET /version HTTP/1.1\r\n\r\n"; ASSERT_EQ((ssize_t)http_req.size(), write(fd, http_req.data(), http_req.size())); + // The reply may arrive in several segments, read until the status + // line prefix is complete. + const char status_prefix[] = "HTTP/1.1 200"; + const size_t prefix_len = sizeof(status_prefix) - 1; char resp[1024]; - ASSERT_GT(read(fd, resp, sizeof(resp)), 0); - ASSERT_EQ(0, memcmp(resp, "HTTP/1.1 200", 12)); + size_t nread = 0; + while (nread < prefix_len) { + const ssize_t n = read(fd, resp + nread, sizeof(resp) - nread); + ASSERT_GT(n, 0); + nread += n; + } + ASSERT_EQ(0, memcmp(resp, status_prefix, prefix_len)); // internal_port serves builtin and Tabbed services only, an RTMP // handshake on the same connection must not reach rtmp_service. From 2ff1a52b9cc4cac46ef8ea06fba39693df6525f1 Mon Sep 17 00:00:00 2001 From: Sahana Bogar Date: Mon, 19 Oct 2026 13:56:36 +0530 Subject: [PATCH 3/4] require EOF after C0C1 in not_served_on_internal_port --- test/brpc_rtmp_unittest.cpp | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/test/brpc_rtmp_unittest.cpp b/test/brpc_rtmp_unittest.cpp index fafa347bad..0a48f703fd 100644 --- a/test/brpc_rtmp_unittest.cpp +++ b/test/brpc_rtmp_unittest.cpp @@ -1348,13 +1348,18 @@ TEST(RtmpTest, not_served_on_internal_port) { ASSERT_EQ(0, memcmp(resp, status_prefix, prefix_len)); // internal_port serves builtin and Tabbed services only, an RTMP - // handshake on the same connection must not reach rtmp_service. + // handshake on the same connection must not reach rtmp_service. No + // protocol accepts the bytes there, so the server closes the connection + // and read() sees EOF. S0 (1 byte) means rtmp_service answered and -1 + // is the receive timeout, i.e. the connection was left open; both fail. char c0c1[1 + 1536]; memset(c0c1, 0, sizeof(c0c1)); c0c1[0] = 0x03; ASSERT_EQ((ssize_t)sizeof(c0c1), write(fd, c0c1, sizeof(c0c1))); char s0 = 0; - ASSERT_NE(1, read(fd, &s0, 1)); + const ssize_t n = read(fd, &s0, 1); + const int saved_errno = errno; + ASSERT_EQ(0, n) << "n=" << n << " errno=" << saved_errno; server.Stop(0); server.Join(); From e30072028cc24f6ddc8b9e9a6dffd428e86bfe2a Mon Sep 17 00:00:00 2001 From: Sahana Bogar Date: Tue, 20 Oct 2026 13:14:35 +0530 Subject: [PATCH 4/4] write fully and retry EINTR in not_served_on_internal_port --- test/brpc_rtmp_unittest.cpp | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) diff --git a/test/brpc_rtmp_unittest.cpp b/test/brpc_rtmp_unittest.cpp index 0a48f703fd..56737500b0 100644 --- a/test/brpc_rtmp_unittest.cpp +++ b/test/brpc_rtmp_unittest.cpp @@ -30,6 +30,8 @@ #include "butil/time.h" #include "butil/macros.h" #include "butil/fd_guard.h" +#include "butil/file_util.h" +#include "butil/posix/eintr_wrapper.h" #include "brpc/policy/rtmp_protocol.h" #include "brpc/socket.h" #include "brpc/acceptor.h" @@ -1331,9 +1333,10 @@ TEST(RtmpTest, not_served_on_internal_port) { // Builtin services on internal_port pass authentication without any // credential, which latches the connection as authenticated. - const std::string http_req = "GET /version HTTP/1.1\r\n\r\n"; - ASSERT_EQ((ssize_t)http_req.size(), - write(fd, http_req.data(), http_req.size())); + const std::string http_req = + "GET /version HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n"; + ASSERT_EQ((int)http_req.size(), butil::WriteFileDescriptor( + fd, http_req.data(), http_req.size())); // The reply may arrive in several segments, read until the status // line prefix is complete. const char status_prefix[] = "HTTP/1.1 200"; @@ -1341,7 +1344,8 @@ TEST(RtmpTest, not_served_on_internal_port) { char resp[1024]; size_t nread = 0; while (nread < prefix_len) { - const ssize_t n = read(fd, resp + nread, sizeof(resp) - nread); + const ssize_t n = + HANDLE_EINTR(read(fd, resp + nread, sizeof(resp) - nread)); ASSERT_GT(n, 0); nread += n; } @@ -1355,9 +1359,10 @@ TEST(RtmpTest, not_served_on_internal_port) { char c0c1[1 + 1536]; memset(c0c1, 0, sizeof(c0c1)); c0c1[0] = 0x03; - ASSERT_EQ((ssize_t)sizeof(c0c1), write(fd, c0c1, sizeof(c0c1))); + ASSERT_EQ((int)sizeof(c0c1), + butil::WriteFileDescriptor(fd, c0c1, sizeof(c0c1))); char s0 = 0; - const ssize_t n = read(fd, &s0, 1); + const ssize_t n = HANDLE_EINTR(read(fd, &s0, 1)); const int saved_errno = errno; ASSERT_EQ(0, n) << "n=" << n << " errno=" << saved_errno;