fix: SrtCaller and SrtSession crash (#4328)

1.修改SrtCaller 多线程发送造成的crash
2.修改SrtCaller 接收的期望外的控制包可能造成的空指针引用
3.修改异常的streamid ext可能造成的crash

src/Srt/SrtCaller.cpp

@@ -218,8 +218,8 @@ void SrtCaller::inputSockData(uint8_t *buf, int len, struct sockaddr *addr) {
 
     // 处理srt数据
     if (DataPacket::isDataPacket(buf, len)) {
-        uint32_t socketId = DataPacket::getSocketID(buf, len);
+        if (_is_handleshake_finished && isPlayer()) {
-        if (isPlayer()) {
+            uint32_t socketId = DataPacket::getSocketID(buf, len);
             if (socketId == _socket_id) {
                 _pkt_recv_rate_context->inputPacket(_now, len + UDP_HDR_SIZE);
                 handleDataPacket(buf, len, addr);
@@ -702,6 +702,11 @@ void SrtCaller::handleHandshakeConclusion(SRT::HandshakePacket &pkt, struct sock
 void SrtCaller::handleACK(uint8_t *buf, int len, struct sockaddr *addr) {
     // TraceL;
     //Acknowledgement of Acknowledgement (ACKACK) control packets are sent to acknowledge the reception of a Full ACK
+ 
+    if (!_is_handleshake_finished) {
+        return;
+    }
+
     ACKPacket ack;
     if (!ack.loadFromData(buf, len)) {
         return;
@@ -722,6 +727,10 @@ void SrtCaller::handleACK(uint8_t *buf, int len, struct sockaddr *addr) {
 
 void SrtCaller::handleACKACK(uint8_t *buf, int len, struct sockaddr *addr) {
     // TraceL;
+
+    if (!_is_handleshake_finished) {
+        return;
+    }
     ACKACKPacket::Ptr pkt = std::make_shared<ACKACKPacket>();
     pkt->loadFromData(buf, len);
 
@@ -757,6 +766,15 @@ void SrtCaller::handleACKACK(uint8_t *buf, int len, struct sockaddr *addr) {
 }
 
 void SrtCaller::handleNAK(uint8_t *buf, int len, struct sockaddr *addr) {
+    if (!_is_handleshake_finished) {
+        return;
+    }
+
+    if (isPlayer()) {
+        //player should not handle nak 
+        return;
+    }
+
     //TraceL;
     NAKPacket pkt;
     pkt.loadFromData(buf, len);
@@ -783,6 +801,15 @@ void SrtCaller::handleNAK(uint8_t *buf, int len, struct sockaddr *addr) {
 }
 
 void SrtCaller::handleDropReq(uint8_t *buf, int len, struct sockaddr *addr) {
+    if (!_is_handleshake_finished) {
+        return;
+    }
+
+    if (!isPlayer()) {
+        //pusher should not handle drop req
+        return;
+    }
+
     MsgDropReqPacket pkt;
     pkt.loadFromData(buf, len);
     std::list<DataPacket::Ptr> list;

src/Srt/SrtPlayer.cpp

@@ -36,7 +36,15 @@ void SrtPlayer::play(const string &strUrl) {
         onResult(SockException(Err_other, StrPrinter << "illegal srt url:" << ex.what()));
         return;
     }
-    onConnect();
+
+    weak_ptr<SrtPlayer> weak_self = static_pointer_cast<SrtPlayer>(shared_from_this());
+    getPoller()->async([weak_self]() {
+        auto strong_self = weak_self.lock();
+        if (!strong_self) {
+            return;
+        }
+        strong_self->onConnect();
+    });
     return;
 }
 

src/Srt/SrtPusher.cpp

@@ -32,7 +32,15 @@ void SrtPusher::publish(const string &strUrl) {
         onResult(SockException(Err_other, StrPrinter << "illegal srt url:" << ex.what()));
         return;
     }
-    onConnect();
+
+    weak_ptr<SrtPusher> weak_self = static_pointer_cast<SrtPusher>(shared_from_this());
+    getPoller()->async([weak_self]() {
+        auto strong_self = weak_self.lock();
+        if (!strong_self) {
+            return;
+        }
+        strong_self->onConnect();
+    });
     return;
 }
 

srt/HSExt.cpp

@@ -83,8 +83,10 @@ bool HSExtStreamID::loadFromData(uint8_t *buf, size_t len) {
         ptr += 4;
     }
     char zero = 0x00;
-    if (streamid.back() == zero) {
+    if (!streamid.empty()) {
-        streamid.erase(streamid.find_first_of(zero), streamid.size());
+        if (streamid.back() == zero) {
+            streamid.erase(streamid.find_first_of(zero), streamid.size());
+        }
     }
     return true;
 }