From 145381e81d0f5dd71b12dbb3e35c5f636fad8c53 Mon Sep 17 00:00:00 2001 From: Alessandro Ros Date: Thu, 13 Aug 2026 11:05:39 +0200 Subject: [PATCH] rtmp: inform about authentication failures (#5657) (#6072) Reply with NetStream.Play.Failed or NetStream.Publish.Unauthorized when a client is not authorized to play or publish. This makes clients like OBS to stop recreating the connection in case of authentication failures. --- go.mod | 2 +- go.sum | 4 +- internal/core/forward_test.go | 2 +- internal/forward/manager_test.go | 2 +- internal/forward/rtmp/dest_test.go | 2 +- internal/protocols/rtmp/from_stream_test.go | 6 +- internal/servers/rtmp/conn.go | 86 +++++++++++++++------ internal/servers/rtmp/server.go | 1 + internal/servers/rtmp/server_test.go | 53 ++++++------- internal/servers/srt/conn.go | 8 +- internal/staticsources/rtmp/source_test.go | 8 +- 11 files changed, 103 insertions(+), 71 deletions(-) diff --git a/go.mod b/go.mod index b738d129..200ced35 100644 --- a/go.mod +++ b/go.mod @@ -11,7 +11,7 @@ require ( github.com/alecthomas/kong v1.16.1 github.com/asticode/go-astits v1.15.0 github.com/bluenviron/gohlslib/v2 v2.4.2 - github.com/bluenviron/gortmplib v1.0.0 + github.com/bluenviron/gortmplib v1.0.1-0.20260813085012-9c63c23d6928 github.com/bluenviron/gortsplib/v5 v5.6.3 github.com/bluenviron/mediacommon/v2 v2.9.2 github.com/datarhei/gosrt v0.11.1-0.20260811092139-103c469da1f1 diff --git a/go.sum b/go.sum index a3f2001f..eb065093 100644 --- a/go.sum +++ b/go.sum @@ -35,8 +35,8 @@ github.com/benburkert/openpgp v0.0.0-20160410205803-c2471f86866c h1:8XZeJrs4+ZYh github.com/benburkert/openpgp v0.0.0-20160410205803-c2471f86866c/go.mod h1:x1vxHcL/9AVzuk5HOloOEPrtJY0MaalYr78afXZ+pWI= github.com/bluenviron/gohlslib/v2 v2.4.2 h1:WU3FwwCn8hRIVe4DRBLfX95igUC+qU1bYxQ4+bP5588= github.com/bluenviron/gohlslib/v2 v2.4.2/go.mod h1:oWuUkEnWqDJ1FpC+i+NXef8/y8DE76IlAzryzhyDqwE= -github.com/bluenviron/gortmplib v1.0.0 h1:dNwvN2pofs1fiFlP9Z4cSAb7YxrxoxaGxrdd4k+it9Y= -github.com/bluenviron/gortmplib v1.0.0/go.mod h1:S/uI6mb4EmyxpZEL8W4AwfrADYeFj+xUl4FjpdTsZEE= +github.com/bluenviron/gortmplib v1.0.1-0.20260813085012-9c63c23d6928 h1:HFCBYT6R8WRIQfypSbmMoMdPBWJeJfv3l4YUNjqcNCk= +github.com/bluenviron/gortmplib v1.0.1-0.20260813085012-9c63c23d6928/go.mod h1:S/uI6mb4EmyxpZEL8W4AwfrADYeFj+xUl4FjpdTsZEE= github.com/bluenviron/gortsplib/v5 v5.6.3 h1:OXvHthQZ9fZbLh6r3Go2wuF4XQ4/QW4WTIM2f4bv/W4= github.com/bluenviron/gortsplib/v5 v5.6.3/go.mod h1:kzHgUtvl8NWNsQ5Vsez6Vuugk6ItFT4ByCPm8J/kbSQ= github.com/bluenviron/mediacommon/v2 v2.9.2 h1:jvYeBjvhHKFOBRMTMm4hvrSjyOlCelkOkx6708DidQM= diff --git a/internal/core/forward_test.go b/internal/core/forward_test.go index 24f21ae1..6406b213 100644 --- a/internal/core/forward_test.go +++ b/internal/core/forward_test.go @@ -89,7 +89,7 @@ func handleRTMPForwardConn(nconn net.Conn, received chan<- [][]byte, serverErr c return } - acceptConnErr := conn.Accept() + acceptConnErr := conn.AcceptConn() if acceptConnErr != nil { serverErr <- acceptConnErr return diff --git a/internal/forward/manager_test.go b/internal/forward/manager_test.go index 545eb8ea..3c3f8d7b 100644 --- a/internal/forward/manager_test.go +++ b/internal/forward/manager_test.go @@ -34,7 +34,7 @@ func TestManager(t *testing.T) { err2 = sc.Initialize() require.NoError(t, err2) - err2 = sc.Accept() + err2 = sc.AcceptConn() require.NoError(t, err2) require.Equal(t, true, sc.Publish) diff --git a/internal/forward/rtmp/dest_test.go b/internal/forward/rtmp/dest_test.go index 79f834dd..4d030746 100644 --- a/internal/forward/rtmp/dest_test.go +++ b/internal/forward/rtmp/dest_test.go @@ -42,7 +42,7 @@ func TestDest(t *testing.T) { serverErr <- initializeErr return } - if publishAcceptErr := conn.Accept(); publishAcceptErr != nil { + if publishAcceptErr := conn.AcceptConn(); publishAcceptErr != nil { serverErr <- publishAcceptErr return } diff --git a/internal/protocols/rtmp/from_stream_test.go b/internal/protocols/rtmp/from_stream_test.go index 93430d9a..2c9d3118 100644 --- a/internal/protocols/rtmp/from_stream_test.go +++ b/internal/protocols/rtmp/from_stream_test.go @@ -669,7 +669,7 @@ func TestFromStream(t *testing.T) { err = conn.Initialize() require.NoError(t, err) - err = conn.Accept() + err = conn.AcceptConn() require.NoError(t, err) r := &stream.Reader{Parent: test.NilLogger} @@ -812,7 +812,7 @@ func TestFromStreamLegacyClientMultipleTracks(t *testing.T) { err = conn.Initialize() require.NoError(t, err) - err = conn.Accept() + err = conn.AcceptConn() require.NoError(t, err) // Simulate a legacy client by clearing the FourCcList @@ -911,7 +911,7 @@ func TestFromStreamSkipUnsupportedTracks(t *testing.T) { err = conn.Initialize() require.NoError(t, err) - err = conn.Accept() + err = conn.AcceptConn() require.NoError(t, err) err = FromStream(desc, desc, r, conn, nil, 0, nil) diff --git a/internal/servers/rtmp/conn.go b/internal/servers/rtmp/conn.go index 2932864b..fbe8b585 100644 --- a/internal/servers/rtmp/conn.go +++ b/internal/servers/rtmp/conn.go @@ -135,7 +135,7 @@ func (c *conn) runReader() error { return err } - err = conn.Accept() + err = conn.AcceptConn() if err != nil { return err } @@ -171,6 +171,18 @@ func (c *conn) runRead() error { EnableAskCredentials: false, }, }) + if err != nil { + if _, ok := errors.AsType[*auth.Error](err); ok { + rejectErr := c.rconn.RejectAction() + if rejectErr != nil { + return rejectErr + } + } + + return err + } + + err = c.rconn.AcceptAction() if err != nil { return err } @@ -233,26 +245,8 @@ func (c *conn) runPublish() error { pathName := strings.TrimLeft(c.rconn.URL.Path, "/") query := c.rconn.URL.Query() - r := &gortmplib.Reader{ - Conn: c.rconn, - } - err := r.Initialize() - if err != nil { - return err - } - - var subStream *stream.SubStream - - medias, err := rtmp.ToStream(r, &subStream) - if err != nil { - return err - } - - res, err := c.pathManager.AddPublisher(defs.PathAddPublisherReq{ - Author: c, - Desc: &description.Session{Medias: medias}, - UseRTPPackets: false, - ReplaceNTP: true, + res1, err := c.pathManager.FindPathConf(defs.PathFindPathConfReq{ + Author: c, AccessRequest: defs.PathAccessRequest{ Name: pathName, Query: c.rconn.URL.RawQuery, @@ -269,18 +263,62 @@ func (c *conn) runPublish() error { }, }) if err != nil { + if _, ok := errors.AsType[*auth.Error](err); ok { + rejectErr := c.rconn.RejectAction() + if rejectErr != nil { + return rejectErr + } + } + return err } - defer res.Path.RemovePublisher(defs.PathRemovePublisherReq{Author: c}) + err = c.rconn.AcceptAction() + if err != nil { + return err + } - subStream = res.SubStream + r := &gortmplib.Reader{ + Conn: c.rconn, + } + err = r.Initialize() + if err != nil { + return err + } + + var subStream *stream.SubStream + + medias, err := rtmp.ToStream(r, &subStream) + if err != nil { + return err + } + + res2, err := c.pathManager.AddPublisher(defs.PathAddPublisherReq{ + Author: c, + Desc: &description.Session{Medias: medias}, + UseRTPPackets: false, + ReplaceNTP: true, + ConfToCompare: res1.Conf, + AccessRequest: defs.PathAccessRequest{ + Name: pathName, + Query: c.rconn.URL.RawQuery, + Publish: true, + SkipAuth: true, + }, + }) + if err != nil { + return err + } + + defer res2.Path.RemovePublisher(defs.PathRemovePublisherReq{Author: c}) + + subStream = res2.SubStream c.mutex.Lock() c.state = defs.APIRTMPConnStatePublish c.pathName = pathName c.query = c.rconn.URL.RawQuery - c.user = res.User + c.user = res1.User c.mutex.Unlock() c.nconn.SetWriteDeadline(time.Time{}) diff --git a/internal/servers/rtmp/server.go b/internal/servers/rtmp/server.go index 4694bf47..9d11a253 100644 --- a/internal/servers/rtmp/server.go +++ b/internal/servers/rtmp/server.go @@ -64,6 +64,7 @@ type serverMetrics interface { } type serverPathManager interface { + FindPathConf(req defs.PathFindPathConfReq) (*defs.PathFindPathConfRes, error) AddPublisher(req defs.PathAddPublisherReq) (*defs.PathAddPublisherRes, error) AddReader(req defs.PathAddReaderReq) (*defs.PathAddReaderRes, error) } diff --git a/internal/servers/rtmp/server_test.go b/internal/servers/rtmp/server_test.go index e07937f7..d98cab29 100644 --- a/internal/servers/rtmp/server_test.go +++ b/internal/servers/rtmp/server_test.go @@ -58,18 +58,25 @@ func TestAuthError(t *testing.T) { "rtmp://127.0.0.1:1939/teststream?user=myuser&pass=mypass", } { func() { - called := make(chan struct{}) pathManager := &test.PathManager{} + if ca.publish { - pathManager.AddPublisherImpl = func(_ defs.PathAddPublisherReq) (*defs.PathAddPublisherRes, error) { - close(called) + pathManager.FindPathConfImpl = func(_ defs.PathFindPathConfReq) (*defs.PathFindPathConfRes, error) { return nil, &auth.Error{Wrapped: fmt.Errorf("auth error")} } + pathManager.AddPublisherImpl = func(_ defs.PathAddPublisherReq) (*defs.PathAddPublisherRes, error) { + return nil, fmt.Errorf("should not be called") + } } else { pathManager.AddReaderImpl = func(_ defs.PathAddReaderReq) (*defs.PathAddReaderRes, error) { - close(called) return nil, &auth.Error{Wrapped: fmt.Errorf("auth error")} } + pathManager.FindPathConfImpl = func(_ defs.PathFindPathConfReq) (*defs.PathFindPathConfRes, error) { + return nil, fmt.Errorf("should not be called") + } + pathManager.AddPublisherImpl = func(_ defs.PathAddPublisherReq) (*defs.PathAddPublisherRes, error) { + return nil, fmt.Errorf("should not be called") + } } s := &Server{ @@ -96,32 +103,12 @@ func TestAuthError(t *testing.T) { Publish: ca.publish, } err = conn.Initialize(context.Background()) - require.NoError(t, err) - defer conn.Close() - - if ca.publish { - track := &gortmplib.Track{Codec: &codecs.H264{ - SPS: test.FormatH264.SPS, - PPS: test.FormatH264.PPS, - }} - w := &gortmplib.Writer{ - Conn: conn, - Tracks: []*gortmplib.Track{track}, - } - err = w.Initialize() - require.NoError(t, err) - _ = w.WriteH264(track, 2*time.Second, 2*time.Second, [][]byte{{5, 2, 3, 4}}) - } else { - r := &gortmplib.Reader{Conn: conn} - err = r.Initialize() - require.Error(t, err) + if !ca.publish { + require.ErrorContains(t, err, "NetStream.Play.Failed") + return } - select { - case <-called: - case <-time.After(2 * time.Second): - t.Fatal("auth callback not reached") - } + require.ErrorContains(t, err, "NetStream.Publish.Unauthorized") }() } }) @@ -159,12 +146,19 @@ func TestServerPublish(t *testing.T) { n := 0 pathManager := &test.PathManager{ - AddPublisherImpl: func(req defs.PathAddPublisherReq) (*defs.PathAddPublisherRes, error) { + FindPathConfImpl: func(req defs.PathFindPathConfReq) (*defs.PathFindPathConfRes, error) { require.Equal(t, "teststream", req.AccessRequest.Name) require.Equal(t, "user=myuser&pass=mypass¶m=value", req.AccessRequest.Query) require.Equal(t, "myuser", req.AccessRequest.Credentials.User) require.Equal(t, "mypass", req.AccessRequest.Credentials.Pass) + return &defs.PathFindPathConfRes{User: req.AccessRequest.Credentials.User}, nil + }, + AddPublisherImpl: func(req defs.PathAddPublisherReq) (*defs.PathAddPublisherRes, error) { + require.Equal(t, "teststream", req.AccessRequest.Name) + require.Equal(t, "user=myuser&pass=mypass¶m=value", req.AccessRequest.Query) + require.True(t, req.AccessRequest.SkipAuth) + strm = &stream.Stream{ OrigDesc: req.Desc, WriteQueueSize: 512, @@ -210,7 +204,6 @@ func TestServerPublish(t *testing.T) { return &defs.PathAddPublisherRes{ Path: &dummyPath{}, - User: req.AccessRequest.Credentials.User, SubStream: subStream, }, nil }, diff --git a/internal/servers/srt/conn.go b/internal/servers/srt/conn.go index 27c4f279..46aedc41 100644 --- a/internal/servers/srt/conn.go +++ b/internal/servers/srt/conn.go @@ -152,9 +152,9 @@ func (c *conn) runPublish(streamID *streamID) error { }, }) if err != nil { - if terr, ok := errors.AsType[*auth.Error](err); ok { + if _, ok := errors.AsType[*auth.Error](err); ok { c.connReq.Reject(srt.REJ_PEER) - return terr + return err } c.connReq.Reject(srt.REJ_PEER) @@ -278,9 +278,9 @@ func (c *conn) runRead(streamID *streamID) error { }, }) if err != nil { - if terr, ok := errors.AsType[*auth.Error](err); ok { + if _, ok := errors.AsType[*auth.Error](err); ok { c.connReq.Reject(srt.REJ_PEER) - return terr + return err } c.connReq.Reject(srt.REJ_PEER) diff --git a/internal/staticsources/rtmp/source_test.go b/internal/staticsources/rtmp/source_test.go index 986b60c4..ff3a2a66 100644 --- a/internal/staticsources/rtmp/source_test.go +++ b/internal/staticsources/rtmp/source_test.go @@ -103,15 +103,15 @@ func TestSource(t *testing.T) { require.NoError(t, err) if auth == "auth" { - err = conn.CheckCredentials("myuser", "mypass") + err = conn.AcceptConnIfCredentialsMatch("myuser", "mypass") if err != nil { continue } + } else { + err = conn.AcceptConn() + require.NoError(t, err) } - err = conn.Accept() - require.NoError(t, err) - w := &gortmplib.Writer{ Conn: conn, Tracks: []*gortmplib.Track{