OPC/mediamtx
2
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

rtmp: inform about authentication failures (#5657) (#6072)

Browse Source 
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.
This commit is contained in:
Alessandro Ros
2026-08-13 09:05:39 +00:00
committed by GitHub
parent 20048e7c11
commit 145381e81d
11 changed files with 103 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+1 -1
View File
@@ -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
go.sum
+2 -2
View File
@@ -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=
internal/core/forward_test.go
+1 -1
View File
@@ -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
internal/forward/manager_test.go
+1 -1
View File
@@ -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)
internal/forward/rtmp/dest_test.go
+1 -1
View File
@@ -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
}
internal/protocols/rtmp/from_stream_test.go
+3 -3
View File
@@ -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)
internal/servers/rtmp/conn.go
+61 -23
View File
@@ -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{
res1, err := c.pathManager.FindPathConf(defs.PathFindPathConfReq{
Author: c,
Desc: &description.Session{Medias: medias},
UseRTPPackets: false,
ReplaceNTP: true,
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{})
internal/servers/rtmp/server.go
+1
View File
@@ -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)
}
internal/servers/rtmp/server_test.go
+23 -30
View File
@@ -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&param=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&param=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
},
internal/servers/srt/conn.go
+4 -4
View File
@@ -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)
internal/staticsources/rtmp/source_test.go
+4 -4
View File
@@ -103,14 +103,14 @@ 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
}
}
err = conn.Accept()
} else {
err = conn.AcceptConn()
require.NoError(t, err)
}
w := &gortmplib.Writer{
Conn: conn,