implement RTMP authentication

internal/client/client.go

@@ -15,8 +15,9 @@ type Client interface {
 	IsClient()
 	IsSource()
 	Close()
-	Authenticate([]headers.AuthMethod, []interface{}, string, string,
+	Authenticate([]headers.AuthMethod,
-		*base.Request, *base.URL) error
+		string, []interface{},
+		string, string, interface{}) error
 	OnReaderFrame(int, gortsplib.StreamType, []byte)
 }
 
@@ -81,7 +82,7 @@ type AnnounceReq struct {
 	Client   Client
 	PathName string
 	Tracks   gortsplib.Tracks
-	Req      *base.Request
+	Req      interface{}
 	Res      chan AnnounceRes
 }
 

internal/clientrtmp/client.go

@@ -3,6 +3,7 @@ package clientrtmp
 import (
 	"fmt"
 	"io"
+	"net"
 	"net/url"
 	"strings"
 	"sync"
@@ -27,6 +28,23 @@ const (
 	pauseAfterAuthError = 2 * time.Second
 )
 
+func ipEqualOrInRange(ip net.IP, ips []interface{}) bool {
+	for _, item := range ips {
+		switch titem := item.(type) {
+		case net.IP:
+			if titem.Equal(ip) {
+				return true
+			}
+
+		case *net.IPNet:
+			if titem.Contains(ip) {
+				return true
+			}
+		}
+	}
+	return false
+}
+
 // Parent is implemented by clientman.ClientMan.
 type Parent interface {
 	Log(logger.Level, string, ...interface{})
@@ -90,6 +108,10 @@ func (c *Client) log(level logger.Level, format string, args ...interface{}) {
 	c.parent.Log(level, "[client %s] "+format, append([]interface{}{c.conn.NConn.RemoteAddr().String()}, args...)...)
 }
 
+func (c *Client) ip() net.IP {
+	return c.conn.NConn.RemoteAddr().(*net.TCPAddr).IP
+}
+
 func (c *Client) run() {
 	defer c.wg.Done()
 	defer c.log(logger.Info, "disconnected")
@@ -165,13 +187,13 @@ func (c *Client) run() {
 		pathName := strings.TrimPrefix(ur.Path, "/")
 
 		resc := make(chan client.AnnounceRes)
-		c.parent.OnClientAnnounce(client.AnnounceReq{c, pathName, tracks, nil, resc}) //nolint:govet
+		c.parent.OnClientAnnounce(client.AnnounceReq{c, pathName, tracks, ur.Query(), resc}) //nolint:govet
 		res := <-resc
 
 		if res.Err != nil {
 			switch res.Err.(type) {
 			case client.ErrAuthNotCritical:
-				return err
+				return res.Err
 
 			case client.ErrAuthCritical:
 				// wait some seconds to stop brute force attacks
@@ -179,10 +201,10 @@ func (c *Client) run() {
 				case <-time.After(pauseAfterAuthError):
 				case <-c.terminate:
 				}
-				return err
+				return res.Err
 
 			default:
-				return err
+				return res.Err
 			}
 		}
 
@@ -304,10 +326,33 @@ func (c *Client) run() {
 }
 
 // Authenticate performs an authentication.
-func (c *Client) Authenticate(authMethods []headers.AuthMethod, ips []interface{},
+func (c *Client) Authenticate(authMethods []headers.AuthMethod,
-	user string, pass string, req *base.Request, altURL *base.URL) error {
+	pathName string, ips []interface{},
+	user string, pass string, req interface{}) error {
+
+	// validate ip
+	if ips != nil {
+		ip := c.ip()
+
+		if !ipEqualOrInRange(ip, ips) {
+			c.log(logger.Info, "ERR: ip '%s' not allowed", ip)
+
+			return client.ErrAuthCritical{&base.Response{ //nolint:govet
+				StatusCode: base.StatusUnauthorized,
+			}}
+		}
+	}
+
+	// validate user
+	if user != "" {
+		values := req.(url.Values)
+
+		if values.Get("user") != user ||
+			values.Get("pass") != pass {
+			return client.ErrAuthCritical{nil} //nolint:govet
+		}
+	}
 
-	// TODO
 	return nil
 }
 

internal/clientrtsp/client.go

@@ -37,6 +37,23 @@ func (e ErrNoOnePublishing) Error() string {
 	return fmt.Sprintf("no one is publishing to path '%s'", e.PathName)
 }
 
+func ipEqualOrInRange(ip net.IP, ips []interface{}) bool {
+	for _, item := range ips {
+		switch titem := item.(type) {
+		case net.IP:
+			if titem.Equal(ip) {
+				return true
+			}
+
+		case *net.IPNet:
+			if titem.Contains(ip) {
+				return true
+			}
+		}
+	}
+	return false
+}
+
 // Parent is implemented by clientman.ClientMan.
 type Parent interface {
 	Log(logger.Level, string, ...interface{})
@@ -488,8 +505,10 @@ func (c *Client) run() {
 }
 
 // Authenticate performs an authentication.
-func (c *Client) Authenticate(authMethods []headers.AuthMethod, ips []interface{},
+func (c *Client) Authenticate(authMethods []headers.AuthMethod,
-	user string, pass string, req *base.Request, altURL *base.URL) error {
+	pathName string, ips []interface{},
+	user string, pass string, req interface{}) error {
+
 	// validate ip
 	if ips != nil {
 		ip := c.ip()
@@ -505,6 +524,8 @@ func (c *Client) Authenticate(authMethods []headers.AuthMethod, ips []interface{
 
 	// validate user
 	if user != "" {
+		reqRTSP := req.(*base.Request)
+
 		// reset authValidator every time the credentials change
 		if c.authValidator == nil || c.authUser != user || c.authPass != pass {
 			c.authUser = user
@@ -512,8 +533,21 @@ func (c *Client) Authenticate(authMethods []headers.AuthMethod, ips []interface{
 			c.authValidator = auth.NewValidator(user, pass, authMethods)
 		}
 
-		err := c.authValidator.ValidateHeader(req.Header["Authorization"],
+		// VLC strips the control attribute
-			req.Method, req.URL, altURL)
+		// provide an alternative URL without the control attribute
+		altURL := func() *base.URL {
+			if reqRTSP.Method != base.Setup {
+				return nil
+			}
+			return &base.URL{
+				Scheme: reqRTSP.URL.Scheme,
+				Host:   reqRTSP.URL.Host,
+				Path:   "/" + pathName + "/",
+			}
+		}()
+
+		err := c.authValidator.ValidateHeader(reqRTSP.Header["Authorization"],
+			reqRTSP.Method, reqRTSP.URL, altURL)
 		if err != nil {
 			c.authFailures++
 

internal/clientrtsp/utils.go

@@ -1,22 +0,0 @@
-package clientrtsp
-
-import (
-	"net"
-)
-
-func ipEqualOrInRange(ip net.IP, ips []interface{}) bool {
-	for _, item := range ips {
-		switch titem := item.(type) {
-		case net.IP:
-			if titem.Equal(ip) {
-				return true
-			}
-
-		case *net.IPNet:
-			if titem.Contains(ip) {
-				return true
-			}
-		}
-	}
-	return false
-}

internal/pathman/pathman.go

@@ -5,7 +5,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/aler9/gortsplib/pkg/base"
 	"github.com/aler9/gortsplib/pkg/headers"
 
 	"github.com/aler9/rtsp-simple-server/internal/client"
@@ -153,8 +152,9 @@ outer:
 				continue
 			}
 
-			err = req.Client.Authenticate(pm.authMethods, pathConf.ReadIpsParsed,
+			err = req.Client.Authenticate(pm.authMethods, req.PathName,
-				pathConf.ReadUser, pathConf.ReadPass, req.Req, nil)
+				pathConf.ReadIpsParsed, pathConf.ReadUser, pathConf.ReadPass,
+				req.Req)
 			if err != nil {
 				req.Res <- client.DescribeRes{nil, "", err} //nolint:govet
 				continue
@@ -185,9 +185,9 @@ outer:
 				continue
 			}
 
-			err = req.Client.Authenticate(pm.authMethods,
+			err = req.Client.Authenticate(pm.authMethods, req.PathName,
 				pathConf.PublishIpsParsed, pathConf.PublishUser,
-				pathConf.PublishPass, req.Req, nil)
+				pathConf.PublishPass, req.Req)
 			if err != nil {
 				req.Res <- client.AnnounceRes{nil, err} //nolint:govet
 				continue
@@ -218,23 +218,9 @@ outer:
 				continue
 			}
 
-			// VLC strips the control attribute
-			// provide an alternative URL without the control attribute
-			altURL := func() *base.URL {
-				if req.Req == nil {
-					return nil
-				}
-
-				return &base.URL{
-					Scheme: req.Req.URL.Scheme,
-					Host:   req.Req.URL.Host,
-					Path:   "/" + req.PathName + "/",
-				}
-			}()
-
 			err = req.Client.Authenticate(pm.authMethods,
-				pathConf.ReadIpsParsed, pathConf.ReadUser, pathConf.ReadPass,
+				req.PathName, pathConf.ReadIpsParsed, pathConf.ReadUser,
-				req.Req, altURL)
+				pathConf.ReadPass, req.Req)
 			if err != nil {
 				req.Res <- client.SetupPlayRes{nil, err} //nolint:govet
 				continue

main_test.go

@@ -506,6 +506,40 @@ func TestAuth(t *testing.T) {
 		defer cnt2.close()
 		require.Equal(t, 0, cnt2.wait())
 	})
+
+	t.Run("rtmp", func(t *testing.T) {
+		p, ok := testProgram("rtmpEnable: yes\n" +
+			"paths:\n" +
+			"  all:\n" +
+			"    publishUser: testuser\n" +
+			"    publishPass: testpass\n")
+		require.Equal(t, true, ok)
+		defer p.close()
+
+		cnt1, err := newContainer("ffmpeg", "source", []string{
+			"-re",
+			"-stream_loop", "-1",
+			"-i", "emptyvideo.ts",
+			"-c", "copy",
+			"-f", "flv",
+			"rtmp://" + ownDockerIP + "/test1/test2?user=testuser&pass=testpass",
+		})
+		require.NoError(t, err)
+		defer cnt1.close()
+
+		time.Sleep(1 * time.Second)
+
+		cnt2, err := newContainer("ffmpeg", "dest", []string{
+			"-rtsp_transport", "udp",
+			"-i", "rtsp://" + ownDockerIP + ":8554/test1/test2",
+			"-vframes", "1",
+			"-f", "image2",
+			"-y", "/dev/null",
+		})
+		require.NoError(t, err)
+		defer cnt2.close()
+		require.Equal(t, 0, cnt2.wait())
+	})
 }
 
 func TestAuthFail(t *testing.T) {
@@ -845,8 +879,7 @@ func TestFallback(t *testing.T) {
 }
 
 func TestRTMP(t *testing.T) {
-	p, ok := testProgram("paths:\n" +
+	p, ok := testProgram("rtmpEnable: yes\n")
-		"rtmpEnable: yes\n")
 	require.Equal(t, true, ok)
 	defer p.close()
 
@@ -856,7 +889,7 @@ func TestRTMP(t *testing.T) {
 		"-i", "emptyvideo.ts",
 		"-c", "copy",
 		"-f", "flv",
-		"rtmp://test:tast@" + ownDockerIP + ":1935/test1/test2",
+		"rtmp://" + ownDockerIP + "/test1/test2",
 	})
 	require.NoError(t, err)
 	defer cnt1.close()