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..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" @@ -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;