[refactor] 代码整理：检查rtmp msg长度有效性

CHANGELOG.md

@@ -14,7 +14,7 @@
 - [fix] lalserver: 优雅关闭pprof和http server
 - [perf] mpegts: 优化转换mpegts的性能
 - [refactor] 将转换mpegts的代码从package hls独立出来，移动到package remux中
-- [refactor] 大幅重构logic.Group，为lalserver支持插件化做准备
+- [refactor] lalserver: 大幅重构logic.Group，为支持插件化做准备
 - [log] 支持独立设置单个pkg的日志配置 #62
 - [log] rtmp和rtsp收包时添加trace级别日志 #63
 - [log] rtmp: 优化定位问题的日志 #135

gen_tag.sh

@@ -5,7 +5,7 @@
 # 步骤：
 # 1. 提交所有代码
 # 1-. 检查配置文件中的配置文件版本号和代码中的配置文件版本号是否匹配
-# 2. 修改CHANGELOG.md
+# 2. 修改CHANGELOG.md(并手动提交CHANGELOG.md)
 # 3. 执行gen_tag.sh
 
 #set -x

go.mod

@@ -2,4 +2,4 @@ module github.com/q191201771/lal
 
 go 1.14
 
-require github.com/q191201771/naza v0.30.1
+require github.com/q191201771/naza v0.30.2

go.sum

@@ -1,2 +1,2 @@
-github.com/q191201771/naza v0.30.1 h1:8EAydcrHs+4lUjP4XBJvTlevuOzVcZIoGj5ZK6Y6Njc=
-github.com/q191201771/naza v0.30.1/go.mod h1:n+dpJjQSh90PxBwxBNuifOwQttywvSIN5TkWSSYCeBk=
+github.com/q191201771/naza v0.30.2 h1:9ZC4T5AdSgGlW9cuFGp6H0mOOXQ156HxOzkYPqrvc14=
+github.com/q191201771/naza v0.30.2/go.mod h1:n+dpJjQSh90PxBwxBNuifOwQttywvSIN5TkWSSYCeBk=

pkg/base/rtmp_t.go

@@ -132,6 +132,10 @@ func (msg RtmpMsg) Dts() uint32 {
 	return msg.Header.TimestampAbs
 }
 
+// Pts
+//
+// 注意，只有视频才能调用该函数获取pts，音频的dts和pts都直接使用 RtmpMsg.Header.TimestampAbs
+//
 func (msg RtmpMsg) Pts() uint32 {
 	return msg.Header.TimestampAbs + bele.BeUint24(msg.Payload[2:])
 }

pkg/base/version.go

@@ -16,7 +16,7 @@ import "strings"
 // 并且将这些信息打入可执行文件、日志、各协议中的标准版本字段中
 
 // LalVersion 版本，该变量由外部脚本修改维护
-const LalVersion = "v0.27.1"
+const LalVersion = "v0.28.0"
 
 var (
 	LalLibraryName = "lal"

pkg/logic/group__core_streaming.go

@@ -145,10 +145,35 @@ func (group *Group) broadcastByRtmpMsg(msg base.RtmpMsg) {
 		lrm2ft remux.LazyRtmpMsg2FlvTag
 	)
 
+	// # 数据有效性检查
 	if len(msg.Payload) == 0 {
 		Log.Warnf("[%s] msg payload length is 0. %+v", group.UniqueKey, msg.Header)
 		return
 	}
+	// TODO(chef): 暂时不打开，因为过滤掉了innertest中rtmp和flv的输出和输入就不完全相同了
+	//if msg.Header.MsgTypeId == base.RtmpTypeIdAudio {
+	//	if len(msg.Payload) <= 2 {
+	//		// 注意，ffmpeg有时会发送这几种空数据，这种情况我们直接返回，不打印日志
+	//		if bytes.Equal(msg.Payload, []byte{0xaf, 0x0}) {
+	//			// noop
+	//			return
+	//		}
+	//		Log.Errorf("[%s] invalid rtmp audio message. header=%+v, payload=%s",
+	//			group.UniqueKey, msg.Header, hex.Dump(msg.Payload))
+	//		return
+	//	}
+	//} else if msg.Header.MsgTypeId == base.RtmpTypeIdVideo {
+	//	if len(msg.Payload) <= 5 {
+	//		if bytes.Equal(msg.Payload, []byte{0x27, 0x02, 0x0, 0x0, 0x0}) ||
+	//			bytes.Equal(msg.Payload, []byte{0x17, 0x02, 0x0, 0x0, 0x0}) {
+	//			// noop
+	//			return
+	//		}
+	//		Log.Errorf("[%s] invalid rtmp video message. header=%+v, payload=%s",
+	//			group.UniqueKey, msg.Header, hex.Dump(msg.Payload))
+	//		return
+	//	}
+	//}
 
 	// # mpegts remuxer
 	if group.rtmp2MpegtsRemuxer != nil {

pkg/remux/rtmp2mpegts.go

@@ -9,7 +9,6 @@
 package remux
 
 import (
-	"bytes"
 	"encoding/hex"
 
 	"github.com/q191201771/lal/pkg/aac"
@@ -147,13 +146,7 @@ func (s *Rtmp2MpegtsRemuxer) onPop(msg base.RtmpMsg) {
 
 func (s *Rtmp2MpegtsRemuxer) feedVideo(msg base.RtmpMsg) {
 	if len(msg.Payload) <= 5 {
-		// 注意，ffmpeg有时会发送msg.Payload为 27 02 00 00 00，这种情况我们直接返回，不打印日志
-		if bytes.Equal(msg.Payload, []byte{0x27, 0x02, 0x0, 0x0, 0x0}) {
-			// noop
-			return
-		}
-		Log.Errorf("[%s] invalid video message length. header=%+v, payload=%s",
-			s.UniqueKey, msg.Header, hex.Dump(msg.Payload))
+		Log.Warnf("[%s] rtmp msg too short, ignore. header=%+v, payload=%s", s.UniqueKey, msg.Header, hex.Dump(msg.Payload))
 		return
 	}
 
@@ -285,13 +278,8 @@ func (s *Rtmp2MpegtsRemuxer) feedVideo(msg base.RtmpMsg) {
 }
 
 func (s *Rtmp2MpegtsRemuxer) feedAudio(msg base.RtmpMsg) {
-	if len(msg.Payload) < 3 {
-		// 注意，ffmpeg有时会发送msg.Payload为 af 00，这种情况我们直接返回，不打印日志
-		if bytes.Equal(msg.Payload, []byte{0xaf, 0x0}) {
-			return
-		}
-		Log.Errorf("[%s] invalid audio message length. header=%+v, payload=%s",
-			s.UniqueKey, msg.Header, hex.Dump(msg.Payload))
+	if len(msg.Payload) <= 2 {
+		Log.Warnf("[%s] rtmp msg too short, ignore. header=%+v, payload=%s", s.UniqueKey, msg.Header, hex.Dump(msg.Payload))
 		return
 	}
 	if msg.Payload[0]>>4 != base.RtmpSoundFormatAac {

pkg/remux/rtmp2rtsp.go

@@ -66,6 +66,21 @@ func NewRtmp2RtspRemuxer(onSdp OnSdp, onRtpPacket OnRtpPacket) *Rtmp2RtspRemuxer
 func (r *Rtmp2RtspRemuxer) FeedRtmpMsg(msg base.RtmpMsg) {
 	var err error
 
+	switch msg.Header.MsgTypeId {
+	case base.RtmpTypeIdMetadata:
+		return
+	case base.RtmpTypeIdAudio:
+		if len(msg.Payload) <= 5 {
+			Log.Warnf("rtmp msg too short, ignore. header=%+v, payload=%s", msg.Header, hex.Dump(msg.Payload))
+			return
+		}
+	case base.RtmpTypeIdVideo:
+		if len(msg.Payload) <= 2 {
+			Log.Warnf("rtmp msg too short, ignore. header=%+v, payload=%s", msg.Header, hex.Dump(msg.Payload))
+			return
+		}
+	}
+
 	if msg.Header.MsgTypeId == base.RtmpTypeIdMetadata {
 		// noop
 		return
@@ -163,7 +178,7 @@ func (r *Rtmp2RtspRemuxer) remux(msg base.RtmpMsg) {
 	var rtppkts []rtprtcp.RtpPacket
 	switch msg.Header.MsgTypeId {
 	case base.RtmpTypeIdAudio:
-		if len(msg.Payload)<3{
+		if len(msg.Payload) < 3 {
 			Log.Warnf("aac seq header payload too short. len=%d, payload=%s", len(msg.Payload), hex.Dump(msg.Payload))
 			return
 		}
@@ -176,7 +191,7 @@ func (r *Rtmp2RtspRemuxer) remux(msg base.RtmpMsg) {
 			})
 		}
 	case base.RtmpTypeIdVideo:
-		if len(msg.Payload)<6{
+		if len(msg.Payload) < 6 {
 			Log.Warnf("Video seq header payload too short. len=%d, payload=%s", len(msg.Payload), hex.Dump(msg.Payload))
 			return
 		}