rtmp: add new AMF0 encoder and encoder (#3069)

This improves performance, security and removes a dependency.

README.md

@@ -1970,6 +1970,7 @@ All the code in this repository is released under the [MIT License](LICENSE). Co
 |[HLS specifications](https://github.com/bluenviron/gohlslib#specifications)|HLS|
 |[RTMP](https://rtmp.veriskope.com/pdf/rtmp_specification_1.0.pdf)|RTMP|
 |[Enhanced RTMP](https://raw.githubusercontent.com/veovera/enhanced-rtmp/main/enhanced-rtmp-v1.pdf)|RTMP|
+|[Action Message Format](https://rtmp.veriskope.com/pdf/amf0-file-format-specification.pdf)|RTMP|
 |[WebRTC: Real-Time Communication in Browsers](https://www.w3.org/TR/webrtc/)|WebRTC|
 |[WebRTC HTTP Ingestion Protocol (WHIP)](https://datatracker.ietf.org/doc/draft-ietf-wish-whip/)|WebRTC|
 |[WebRTC HTTP Egress Protocol (WHEP)](https://datatracker.ietf.org/doc/draft-murillo-whep/)|WebRTC|
@@ -1987,7 +1988,6 @@ All the code in this repository is released under the [MIT License](LICENSE). Co
 * [pion/sdp (SDP library used internally)](https://github.com/pion/sdp)
 * [pion/rtp (RTP library used internally)](https://github.com/pion/rtp)
 * [pion/rtcp (RTCP library used internally)](https://github.com/pion/rtcp)
-* [notedit/rtmp (RTMP library used internally)](https://github.com/notedit/rtmp)
 * [go-astits (MPEG-TS library used internally)](https://github.com/asticode/go-astits)
 * [go-mp4 (MP4 library used internally)](https://github.com/abema/go-mp4)
 * [hls.js (browser-side HLS library used internally)](https://github.com/video-dev/hls.js)

go.mod

@@ -17,7 +17,6 @@ require (
 	github.com/gorilla/websocket v1.5.1
 	github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51
 	github.com/matthewhartstonge/argon2 v1.0.0
-	github.com/notedit/rtmp v0.0.2
 	github.com/pion/ice/v2 v2.3.11
 	github.com/pion/interceptor v0.1.25
 	github.com/pion/logging v0.2.2

go.sum

@@ -107,8 +107,6 @@ github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
 github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M=
 github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
-github.com/notedit/rtmp v0.0.2 h1:5+to4yezKATiJgnrcETu9LbV5G/QsWkOV9Ts2M/p33w=
-github.com/notedit/rtmp v0.0.2/go.mod h1:vzuE21rowz+lT1NGsWbreIvYulgBpCGnQyeTyFblUHc=
 github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A=
 github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE=
 github.com/nxadm/tail v1.4.8/go.mod h1:+ncqLTQzXmGhMZNUePPaPqPvBxHAIsmXswZKocGu+AU=

internal/protocols/rtmp/amf0/marshal.go

@@ -0,0 +1,171 @@
+// Package amf0 contains an AMF0 marshaler and unmarshaler.
+package amf0
+
+import (
+	"fmt"
+	"math"
+)
+
+// Marshal encodes AMF0 data.
+func Marshal(data []interface{}) ([]byte, error) {
+	n, err := marshalSize(data)
+	if err != nil {
+		return nil, err
+	}
+
+	buf := make([]byte, n)
+	n = 0
+
+	for _, item := range data {
+		n += marshalItem(item, buf[n:])
+	}
+
+	return buf, nil
+}
+
+func marshalSize(data []interface{}) (int, error) {
+	n := 0
+
+	for _, item := range data {
+		in, err := marshalSizeItem(item)
+		if err != nil {
+			return 0, err
+		}
+
+		n += in
+	}
+
+	return n, nil
+}
+
+func marshalSizeItem(item interface{}) (int, error) {
+	switch item := item.(type) {
+	case float64:
+		return 9, nil
+
+	case bool:
+		return 2, nil
+
+	case string:
+		return 3 + len(item), nil
+
+	case ECMAArray:
+		n := 5
+
+		for _, entry := range item {
+			en, err := marshalSizeItem(entry.Value)
+			if err != nil {
+				return 0, err
+			}
+
+			n += 2 + len(entry.Key) + en
+		}
+
+		n += 3
+
+		return n, nil
+
+	case Object:
+		n := 1
+
+		for _, entry := range item {
+			en, err := marshalSizeItem(entry.Value)
+			if err != nil {
+				return 0, err
+			}
+
+			n += 2 + len(entry.Key) + en
+		}
+
+		n += 3
+
+		return n, nil
+
+	case nil:
+		return 1, nil
+
+	default:
+		return 0, fmt.Errorf("unsupported data type: %T", item)
+	}
+}
+
+func marshalItem(item interface{}, buf []byte) int {
+	switch item := item.(type) {
+	case float64:
+		v := math.Float64bits(item)
+		buf[0] = markerNumber
+		buf[1] = byte(v >> 56)
+		buf[2] = byte(v >> 48)
+		buf[3] = byte(v >> 40)
+		buf[4] = byte(v >> 32)
+		buf[5] = byte(v >> 24)
+		buf[6] = byte(v >> 16)
+		buf[7] = byte(v >> 8)
+		buf[8] = byte(v)
+		return 9
+
+	case bool:
+		buf[0] = markerBoolean
+		if item {
+			buf[1] = 1
+		}
+		return 2
+
+	case string:
+		le := len(item)
+		buf[0] = markerString
+		buf[1] = byte(le >> 8)
+		buf[2] = byte(le)
+		copy(buf[3:], item)
+		return 3 + le
+
+	case ECMAArray:
+		le := len(item)
+		buf[0] = markerECMAArray
+		buf[1] = byte(le >> 24)
+		buf[2] = byte(le >> 16)
+		buf[3] = byte(le >> 8)
+		buf[4] = byte(le)
+		n := 5
+
+		for _, entry := range item {
+			le := len(entry.Key)
+			buf[n] = byte(le >> 8)
+			buf[n+1] = byte(le)
+			copy(buf[n+2:], entry.Key)
+			n += 2 + le
+
+			n += marshalItem(entry.Value, buf[n:])
+		}
+
+		buf[n] = 0
+		buf[n+1] = 0
+		buf[n+2] = markerObjectEnd
+
+		return n + 3
+
+	case Object:
+		buf[0] = markerObject
+		n := 1
+
+		for _, entry := range item {
+			le := len(entry.Key)
+			buf[n] = byte(le >> 8)
+			buf[n+1] = byte(le)
+			copy(buf[n+2:], entry.Key)
+			n += 2 + le
+
+			n += marshalItem(entry.Value, buf[n:])
+		}
+
+		buf[n] = 0
+		buf[n+1] = 0
+		buf[n+2] = markerObjectEnd
+
+		return n + 3
+
+	default:
+		buf[0] = markerNull
+		return 1
+	}
+}

internal/protocols/rtmp/amf0/marshal_test.go

@@ -0,0 +1,17 @@
+package amf0
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestMarshal(t *testing.T) {
+	for _, ca := range cases {
+		t.Run(ca.name, func(t *testing.T) {
+			enc, err := Marshal(ca.dec)
+			require.NoError(t, err)
+			require.Equal(t, ca.enc, enc)
+		})
+	}
+}

internal/protocols/rtmp/amf0/object.go

@@ -0,0 +1,53 @@
+package amf0
+
+// ObjectEntry is an entry of Object.
+type ObjectEntry struct {
+	Key   string
+	Value interface{}
+}
+
+// Object is an AMF0 object.
+type Object []ObjectEntry
+
+// ECMAArray is an AMF0 ECMA Array.
+type ECMAArray Object
+
+// Get returns the value corresponding to key.
+func (o Object) Get(key string) (interface{}, bool) {
+	for _, item := range o {
+		if item.Key == key {
+			return item.Value, true
+		}
+	}
+	return nil, false
+}
+
+// GetString returns the value corresponding to key, only if that is a string.
+func (o Object) GetString(key string) (string, bool) {
+	v, ok := o.Get(key)
+	if !ok {
+		return "", false
+	}
+
+	v2, ok2 := v.(string)
+	if !ok2 {
+		return "", false
+	}
+
+	return v2, ok2
+}
+
+// GetFloat64 returns the value corresponding to key, only if that is a float64.
+func (o Object) GetFloat64(key string) (float64, bool) {
+	v, ok := o.Get(key)
+	if !ok {
+		return 0, false
+	}
+
+	v2, ok2 := v.(float64)
+	if !ok2 {
+		return 0, false
+	}
+
+	return v2, ok2
+}

internal/protocols/rtmp/amf0/object_test.go

@@ -0,0 +1,28 @@
+package amf0
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestObjectGet(t *testing.T) {
+	o := Object{{Key: "testme", Value: "ok"}}
+	v, ok := o.Get("testme")
+	require.Equal(t, true, ok)
+	require.Equal(t, "ok", v)
+}
+
+func TestObjectGetString(t *testing.T) {
+	o := Object{{Key: "testme", Value: "ok"}}
+	v, ok := o.GetString("testme")
+	require.Equal(t, true, ok)
+	require.Equal(t, "ok", v)
+}
+
+func TestObjectGetFloat64(t *testing.T) {
+	o := Object{{Key: "testme", Value: float64(123)}}
+	v, ok := o.GetFloat64("testme")
+	require.Equal(t, true, ok)
+	require.Equal(t, float64(123), v)
+}

internal/protocols/rtmp/amf0/testdata/fuzz/FuzzUnmarshal/110121ffaa6941a6

@@ -0,0 +1,2 @@
+go test fuzz v1
+[]byte("\x0200")

internal/protocols/rtmp/amf0/testdata/fuzz/FuzzUnmarshal/118a6dec0931d635

@@ -0,0 +1,2 @@
+go test fuzz v1
+[]byte("\b000000")

internal/protocols/rtmp/amf0/testdata/fuzz/FuzzUnmarshal/13ba6ce8ecfbc991

@@ -0,0 +1,2 @@
+go test fuzz v1
+[]byte("\b0000\x00\x000")

internal/protocols/rtmp/amf0/testdata/fuzz/FuzzUnmarshal/399a2041db7e1ff9

@@ -0,0 +1,2 @@
+go test fuzz v1
+[]byte("\x000")

internal/protocols/rtmp/amf0/testdata/fuzz/FuzzUnmarshal/582528ddfad69eb5

@@ -0,0 +1,2 @@
+go test fuzz v1
+[]byte("0")

internal/protocols/rtmp/amf0/testdata/fuzz/FuzzUnmarshal/6df12bd1a096b953

@@ -0,0 +1,2 @@
+go test fuzz v1
+[]byte("\x03\x00\x0200")

internal/protocols/rtmp/amf0/testdata/fuzz/FuzzUnmarshal/6f9c29532ccb80bf

@@ -0,0 +1,2 @@
+go test fuzz v1
+[]byte("\b")

internal/protocols/rtmp/amf0/testdata/fuzz/FuzzUnmarshal/7fbe967ee430d9f4

@@ -0,0 +1,2 @@
+go test fuzz v1
+[]byte("\x0300")

internal/protocols/rtmp/amf0/testdata/fuzz/FuzzUnmarshal/97dc7172b48e6ffd

@@ -0,0 +1,2 @@
+go test fuzz v1
+[]byte("\x01")

internal/protocols/rtmp/amf0/testdata/fuzz/FuzzUnmarshal/aba8f2fdaa5caedb

@@ -0,0 +1,2 @@
+go test fuzz v1
+[]byte("\x03\x00\x000")

internal/protocols/rtmp/amf0/testdata/fuzz/FuzzUnmarshal/b9ef082bd4d297b2

@@ -0,0 +1,2 @@
+go test fuzz v1
+[]byte("\b0000\x00\x00")

internal/protocols/rtmp/amf0/testdata/fuzz/FuzzUnmarshal/df015416c2cf2dc6

@@ -0,0 +1,2 @@
+go test fuzz v1
+[]byte("\b0000")

internal/protocols/rtmp/amf0/testdata/fuzz/FuzzUnmarshal/e93e775e4de86c93

@@ -0,0 +1,2 @@
+go test fuzz v1
+[]byte("\b0000\x00\x06000000")

internal/protocols/rtmp/amf0/testdata/fuzz/FuzzUnmarshal/f3b49d384cddc291

@@ -0,0 +1,2 @@
+go test fuzz v1
+[]byte("\x03")

internal/protocols/rtmp/amf0/testdata/fuzz/FuzzUnmarshal/fb46839f39edbbd8

@@ -0,0 +1,2 @@
+go test fuzz v1
+[]byte("\x03\x00\x00")