This is an automated email from the ASF dual-hosted git repository.
wwbmmm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/brpc.git
The following commit(s) were added to refs/heads/master by this push:
new e98d2e745 do not serve rtmp on ServerOptions.internal_port (#3622)
e98d2e745 is described below
commit e98d2e7458f957d8d970826f7ed87d0b301d39fa
Author: sahvx655-wq <[email protected]>
AuthorDate: Sun Oct 25 15:18:35 2026 +0530
do not serve rtmp on ServerOptions.internal_port (#3622)
* do not serve rtmp on ServerOptions.internal_port
* read the whole status prefix before comparing it in
not_served_on_internal_port
* require EOF after C0C1 in not_served_on_internal_port
* write fully and retry EINTR in not_served_on_internal_port
---
src/brpc/policy/rtmp_protocol.cpp | 7 ++++
test/brpc_rtmp_unittest.cpp | 84 +++++++++++++++++++++++++++++++++++++++
2 files changed, 91 insertions(+)
diff --git a/src/brpc/policy/rtmp_protocol.cpp
b/src/brpc/policy/rtmp_protocol.cpp
index 41ed2422d..bf46922f7 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 36e996b06..56737500b 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"
@@ -1286,6 +1288,88 @@ 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\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";
+ const size_t prefix_len = sizeof(status_prefix) - 1;
+ char resp[1024];
+ size_t nread = 0;
+ while (nread < prefix_len) {
+ const ssize_t n =
+ HANDLE_EINTR(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. 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((int)sizeof(c0c1),
+ butil::WriteFileDescriptor(fd, c0c1, sizeof(c0c1)));
+ char s0 = 0;
+ const ssize_t n = HANDLE_EINTR(read(fd, &s0, 1));
+ const int saved_errno = errno;
+ ASSERT_EQ(0, n) << "n=" << n << " errno=" << saved_errno;
+
+ server.Stop(0);
+ server.Join();
+}
+
TEST(RtmpTest, successfully_play_streams) {
PlayingDummyService rtmp_service;
brpc::Server server;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
