This is an automated email from the ASF dual-hosted git repository.
chenBright 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 ca370c28 Fix RTMP abort message deleting the chunk stream being parsed
(#3452)
ca370c28 is described below
commit ca370c28a905a387e482ee6deae446ac82a6464a
Author: Weibing Wang <[email protected]>
AuthorDate: Sun Aug 16 01:08:30 2026 +0800
Fix RTMP abort message deleting the chunk stream being parsed (#3452)
* Fix RTMP abort message deleting the chunk stream being parsed
* Address review comments on RTMP abort regression test
---
src/brpc/policy/rtmp_protocol.cpp | 7 ++++-
test/brpc_rtmp_unittest.cpp | 59 +++++++++++++++++++++++++++++++++++++++
2 files changed, 65 insertions(+), 1 deletion(-)
diff --git a/src/brpc/policy/rtmp_protocol.cpp
b/src/brpc/policy/rtmp_protocol.cpp
index 62322018..c5ece81a 100644
--- a/src/brpc/policy/rtmp_protocol.cpp
+++ b/src/brpc/policy/rtmp_protocol.cpp
@@ -1869,7 +1869,12 @@ bool RtmpChunkStream::OnAbortMessage(
RTMP_ERROR(socket, mh) << "Invalid chunk_stream_id=" << cs_id;
return false;
}
- connection_context()->ClearChunkStream(cs_id);
+ // Do not delete the chunk stream that is currently being parsed (i.e.
+ // the one running this Feed). Clearing it here would free `this' while
+ // Feed() still holds and later touches it, causing a use-after-free.
+ if (cs_id != _cs_id) {
+ connection_context()->ClearChunkStream(cs_id);
+ }
return true;
}
diff --git a/test/brpc_rtmp_unittest.cpp b/test/brpc_rtmp_unittest.cpp
index 61a63744..286f8a87 100644
--- a/test/brpc_rtmp_unittest.cpp
+++ b/test/brpc_rtmp_unittest.cpp
@@ -28,6 +28,8 @@
#include <google/protobuf/io/zero_copy_stream_impl_lite.h>
#include "butil/time.h"
#include "butil/macros.h"
+#include "butil/fd_guard.h"
+#include "brpc/policy/rtmp_protocol.h"
#include "brpc/socket.h"
#include "brpc/acceptor.h"
#include "brpc/server.h"
@@ -805,6 +807,63 @@ TEST(RtmpTest, flv_reader_rejects_zero_datasize_audio_tag)
{
ASSERT_EQ(before, buf.size());
}
+// A crafted Abort message that names the chunk stream currently being parsed
+// (itself) used to make ClearChunkStream delete the RtmpChunkStream while its
+// Feed() is still running, which caused a heap-use-after-free right after
+// OnMessage() returned.
+TEST(RtmpTest, abort_message_naming_own_chunk_stream) {
+ int pipe_fds[2];
+ ASSERT_EQ(0, pipe(pipe_fds));
+ butil::fd_guard guard0(pipe_fds[0]); // read end, closed by this guard
+ butil::fd_guard guard1(pipe_fds[1]); // write end, handed over to Socket
+
+ brpc::SocketId id;
+ brpc::SocketOptions options;
+ options.fd = guard1.release(); // Socket takes ownership of the fd
+ ASSERT_EQ(0, brpc::Socket::Create(options, &id));
+ brpc::SocketUniquePtr sock;
+ ASSERT_EQ(0, brpc::Socket::Address(id, &sock));
+
+ brpc::policy::RtmpContext ctx(NULL, NULL);
+ ctx.SetState(sock->remote_side(),
+ brpc::policy::RtmpContext::STATE_RECEIVED_C2);
+
+ // fmt0 chunk on chunk stream 2 carrying an Abort message (type 2) whose
+ // payload is the same chunk stream id (2).
+ std::string chunk;
+ chunk.push_back((char)0x02); // basic header: fmt=0, cs_id=2
+ chunk.append(3, '\0'); // timestamp = 0
+ chunk.push_back('\0'); // message_length (3 bytes) = 4
+ chunk.push_back('\0');
+ chunk.push_back((char)0x04);
+ chunk.push_back((char)0x02); // message_type = Abort
+ chunk.append(4, '\0'); // stream_id = 0 (little endian)
+ chunk.push_back('\0'); // payload: cs_id = 2 (big endian)
+ chunk.push_back('\0');
+ chunk.push_back('\0');
+ chunk.push_back((char)0x02);
+
+ butil::IOBuf buf;
+ buf.append(chunk);
+ ASSERT_EQ(brpc::PARSE_OK, ctx.Feed(&buf, sock.get()).error());
+
+ // A following type-1 chunk inherits the message header from the previous
+ // message on the same stream. If the abort had wrongly deleted the chunk
+ // stream, the freshly recreated stream would have no last header and this
+ // chunk would be rejected; instead it must be parsed successfully.
+ std::string cont;
+ cont.push_back((char)0x42); // basic header: fmt=1, cs_id=2
+ cont.append(3, '\0'); // timestamp delta = 0
+ cont.push_back('\0'); // message_length (3 bytes) = 4
+ cont.push_back('\0');
+ cont.push_back((char)0x04);
+ cont.push_back((char)0x03); // message_type = Ack
+ cont.append(4, '\0'); // payload: bytes_received = 0
+ butil::IOBuf buf2;
+ buf2.append(cont);
+ ASSERT_EQ(brpc::PARSE_OK, ctx.Feed(&buf2, sock.get()).error());
+}
+
TEST(RtmpTest, successfully_play_streams) {
PlayingDummyService rtmp_service;
brpc::Server server;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
