add RTSP-over-WebSocket (#891) (#898)

Author: aler9

README.md

@@ -14,7 +14,7 @@ Features:
 
 * Client
   * Support secure protocol variants (RTSPS, TLS, SRTP, MIKEY)
-  * Support RTSP-over-HTTP, RTSP-over-HTTPS
+  * Support tunneling (RTSP-over-HTTP, RTSP-over-WebSocket)
   * Query servers about available media streams
   * Read media streams from a server ("play")
     * Read streams with the UDP, UDP-multicast or TCP transport protocol
@@ -30,7 +30,7 @@ Features:
     * Pause without disconnecting from the server
 * Server
   * Support secure protocol variants (RTSPS, TLS, SRTP, MIKEY)
-  * Support RTSP-over-HTTP, RTSP-over-HTTPS
+  * Support tunneling (RTSP-over-HTTP, RTSP-over-WebSocket)
   * Handle requests from clients
   * Validate client credentials
   * Read media streams from clients ("record")

client.go

@@ -1137,13 +1137,22 @@ func (c *Client) connOpen() error {
 
 	var nconn net.Conn
 
-	if c.Tunnel == TunnelHTTP {
+	switch c.Tunnel {
+	case TunnelHTTP:
 		var err error
-		nconn, err = newClientHTTPTunnel(dialCtx, c.DialContext, addr, tlsConfig)
+		nconn, err = newClientTunnelHTTP(dialCtx, c.DialContext, addr, tlsConfig)
 		if err != nil {
 			return err
 		}
-	} else {
+
+	case TunnelWebSocket:
+		var err error
+		nconn, err = newClientTunnelWebSocket(dialCtx, c.DialContext, addr, tlsConfig)
+		if err != nil {
+			return err
+		}
+
+	default:
 		var err error
 		nconn, err = c.DialContext(dialCtx, "tcp", addr)
 		if err != nil {

client_test.go

@@ -3,6 +3,7 @@ package gortsplib
 import (
 	"bufio"
 	"bytes"
+	"context"
 	"crypto/tls"
 	"net"
 	"net/http"
@@ -593,7 +594,7 @@ func TestClientRelativeContentBase(t *testing.T) {
 	require.Equal(t, "rtsp://localhost:8554/relative-content-base", desc.BaseURL.String())
 }
 
-func TestClientHTTPTunnel(t *testing.T) {
+func TestClientTunnelHTTP(t *testing.T) {
 	for _, ca := range []string{"http", "https"} {
 		t.Run(ca, func(t *testing.T) {
 			var l net.Listener
@@ -768,8 +769,103 @@ func TestClientHTTPTunnel(t *testing.T) {
 			require.NoError(t, err)
 			defer c.Close()
 
-			_, _, err = c.Describe(u)
+			_, res, err := c.Describe(u)
 			require.NoError(t, err)
+			require.Equal(t, base.StatusOK, res.StatusCode)
+		})
+	}
+}
+
+func TestClientTunnelWebSocket(t *testing.T) {
+	for _, ca := range []string{"ws", "wss"} {
+		t.Run(ca, func(t *testing.T) {
+			var scheme string
+			if ca == "ws" {
+				scheme = "rtsp"
+			} else {
+				scheme = "rtsps"
+			}
+
+			s := &http.Server{
+				Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+					require.Equal(t, r.Header.Get("Sec-WebSocket-Protocol"), "rtsp.onvif.org")
+
+					wconn, err := upgrader.Upgrade(w, r, nil)
+					require.NoError(t, err)
+					defer wconn.Close() //nolint:errcheck
+
+					conn := conn.NewConn(bufio.NewReader(&wsReader{wc: wconn}), &wsWriter{wc: wconn})
+
+					req, err2 := conn.ReadRequest()
+					require.NoError(t, err2)
+					require.Equal(t, base.Options, req.Method)
+
+					err2 = conn.WriteResponse(&base.Response{
+						StatusCode: base.StatusOK,
+						Header: base.Header{
+							"Public": base.HeaderValue{strings.Join([]string{
+								string(base.Describe),
+							}, ", ")},
+						},
+					})
+					require.NoError(t, err2)
+
+					req, err2 = conn.ReadRequest()
+					require.NoError(t, err2)
+					require.Equal(t, base.Describe, req.Method)
+					require.Equal(t, mustParseURL(scheme+"://localhost:8554/teststream"), req.URL)
+
+					medias := []*description.Media{testH264Media}
+
+					err2 = conn.WriteResponse(&base.Response{
+						StatusCode: base.StatusOK,
+						Header: base.Header{
+							"Content-Type": base.HeaderValue{"application/sdp; charset=utf-8"},
+							"Content-Base": base.HeaderValue{"/relative-content-base"},
+						},
+						Body: mediasToSDP(medias),
+					})
+					require.NoError(t, err2)
+				}),
+			}
+
+			var ln net.Listener
+
+			if ca == "ws" {
+				var err error
+				ln, err = net.Listen("tcp", "localhost:8554")
+				require.NoError(t, err)
+			} else {
+				cert, err := tls.X509KeyPair(serverCert, serverKey)
+				require.NoError(t, err)
+
+				ln, err = tls.Listen("tcp", "localhost:8554", &tls.Config{Certificates: []tls.Certificate{cert}})
+				require.NoError(t, err)
+				defer ln.Close()
+			}
+
+			go s.Serve(ln)
+			defer s.Shutdown(context.Background())
+
+			u, err := base.ParseURL(scheme + "://localhost:8554/teststream")
+			require.NoError(t, err)
+
+			c := Client{
+				Scheme: u.Scheme,
+				Host:   u.Host,
+				Tunnel: TunnelWebSocket,
+				TLSConfig: &tls.Config{
+					InsecureSkipVerify: true,
+				},
+			}
+
+			err = c.Start()
+			require.NoError(t, err)
+			defer c.Close()
+
+			_, res, err := c.Describe(u)
+			require.NoError(t, err)
+			require.Equal(t, base.StatusOK, res.StatusCode)
 		})
 	}
 }

client_http_tunnel.go renamed to client_tunnel_http.go

@@ -14,53 +14,53 @@ import (
 	"github.com/google/uuid"
 )
 
-type clientHTTPTunnel struct {
+type clientTunnelHTTP struct {
 	readChan  net.Conn
 	readBuf   *bufio.Reader
 	writeChan net.Conn
 }
 
-func (c *clientHTTPTunnel) Read(p []byte) (n int, err error) {
+func (c *clientTunnelHTTP) Read(p []byte) (n int, err error) {
 	return c.readBuf.Read(p)
 }
 
-func (c *clientHTTPTunnel) Write(p []byte) (n int, err error) {
+func (c *clientTunnelHTTP) Write(p []byte) (n int, err error) {
 	return c.writeChan.Write([]byte(base64.StdEncoding.EncodeToString(p)))
 }
 
-func (c *clientHTTPTunnel) Close() error {
+func (c *clientTunnelHTTP) Close() error {
 	c.readChan.Close()
 	c.writeChan.Close()
 	return nil
 }
 
-func (c *clientHTTPTunnel) LocalAddr() net.Addr {
+func (c *clientTunnelHTTP) LocalAddr() net.Addr {
 	return c.readChan.LocalAddr()
 }
 
-func (c *clientHTTPTunnel) RemoteAddr() net.Addr {
+func (c *clientTunnelHTTP) RemoteAddr() net.Addr {
 	return c.readChan.RemoteAddr()
 }
 
-func (c *clientHTTPTunnel) SetDeadline(_ time.Time) error {
+func (c *clientTunnelHTTP) SetDeadline(_ time.Time) error {
 	panic("unimplemented")
 }
 
-func (c *clientHTTPTunnel) SetReadDeadline(t time.Time) error {
+func (c *clientTunnelHTTP) SetReadDeadline(t time.Time) error {
 	return c.readChan.SetReadDeadline(t)
 }
 
-func (c *clientHTTPTunnel) SetWriteDeadline(t time.Time) error {
+func (c *clientTunnelHTTP) SetWriteDeadline(t time.Time) error {
 	return c.writeChan.SetWriteDeadline(t)
 }
 
-func newClientHTTPTunnel(
+func newClientTunnelHTTP(
 	ctx context.Context,
 	dialContext func(ctx context.Context, network, address string) (net.Conn, error),
 	addr string,
 	tlsConfig *tls.Config,
 ) (net.Conn, error) {
-	c := &clientHTTPTunnel{}
+	c := &clientTunnelHTTP{}
 
 	var err error
 	c.readChan, err = dialContext(ctx, "tcp", addr)

client_tunnel_websocket.go

@@ -0,0 +1,81 @@
+package gortsplib
+
+import (
+	"context"
+	"crypto/tls"
+	"io"
+	"net"
+	"time"
+
+	"github.com/gorilla/websocket"
+)
+
+type clientTunnelWebSocket struct {
+	wconn *websocket.Conn
+	r     io.Reader
+	w     io.Writer
+}
+
+func (tu *clientTunnelWebSocket) Read(b []byte) (int, error) {
+	return tu.r.Read(b)
+}
+
+func (tu *clientTunnelWebSocket) Write(b []byte) (int, error) {
+	return tu.w.Write(b)
+}
+
+func (tu *clientTunnelWebSocket) Close() error {
+	return tu.wconn.Close()
+}
+
+func (tu *clientTunnelWebSocket) LocalAddr() net.Addr {
+	return tu.wconn.LocalAddr()
+}
+
+func (tu *clientTunnelWebSocket) RemoteAddr() net.Addr {
+	return tu.wconn.RemoteAddr()
+}
+
+func (tu *clientTunnelWebSocket) SetDeadline(_ time.Time) error {
+	return nil
+}
+
+func (tu *clientTunnelWebSocket) SetReadDeadline(t time.Time) error {
+	return tu.wconn.SetReadDeadline(t)
+}
+
+func (tu *clientTunnelWebSocket) SetWriteDeadline(t time.Time) error {
+	return tu.wconn.SetWriteDeadline(t)
+}
+
+func newClientTunnelWebSocket(
+	ctx context.Context,
+	dialContext func(ctx context.Context, network, address string) (net.Conn, error),
+	addr string,
+	tlsConfig *tls.Config,
+) (net.Conn, error) {
+	c := &clientTunnelWebSocket{}
+
+	var ur string
+	if tlsConfig != nil {
+		ur = "wss"
+	} else {
+		ur = "ws"
+	}
+	ur += "://" + addr + "/"
+
+	var err error
+	c.wconn, _, err = (&websocket.Dialer{
+		NetDialContext:  dialContext,
+		TLSClientConfig: tlsConfig,
+		Subprotocols:    []string{"rtsp.onvif.org"},
+	}).DialContext(ctx, ur, nil) //nolint:bodyclose
+	if err != nil {
+		return nil, err
+	}
+
+	c.r = &wsReader{wc: c.wconn}
+	c.w = &wsWriter{wc: c.wconn}
+
+	return c, nil
+}

conn_transport.go

@@ -7,6 +7,7 @@ type Tunnel int
 const (
 	TunnelNone Tunnel = iota
 	TunnelHTTP
+	TunnelWebSocket
 )
 
 // ConnTransport contains details about the transport of a connection.

go.mod

@@ -5,6 +5,7 @@ go 1.24.0
 require (
 	github.com/bluenviron/mediacommon/v2 v2.4.2
 	github.com/google/uuid v1.6.0
+	github.com/gorilla/websocket v1.5.3
 	github.com/pion/rtcp v1.2.15
 	github.com/pion/rtp v1.8.22
 	github.com/pion/sdp/v3 v3.0.16

go.sum

@@ -9,6 +9,8 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
+github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/pion/logging v0.2.4 h1:tTew+7cmQ+Mc1pTBLKH2puKsOvhm32dROumOZ655zB8=
 github.com/pion/logging v0.2.4/go.mod h1:DffhXTKYdNZU+KtJ5pyQDjvOAh/GsNSyv1lbkFbe3so=
 github.com/pion/randutil v0.1.0 h1:CFG1UdESneORglEsnimhUjf33Rwjubwj6xfiOXBa3mA=

internal/teste2e/client_vs_server_test.go

@@ -132,6 +132,22 @@ func TestClientVsServer(t *testing.T) {
 			readerProto:     "tcp",
 			readerTunnel:    "http",
 		},
+		{
+			publisherScheme: "rtsp",
+			publisherProto:  "tcp",
+			publisherTunnel: "websocket",
+			readerScheme:    "rtsp",
+			readerProto:     "udp",
+			readerTunnel:    "none",
+		},
+		{
+			publisherScheme: "rtsp",
+			publisherProto:  "tcp",
+			publisherTunnel: "none",
+			readerScheme:    "rtsp",
+			readerProto:     "tcp",
+			readerTunnel:    "websocket",
+		},
 	} {
 		t.Run(strings.Join([]string{
 			ca.publisherScheme,
@@ -166,9 +182,12 @@ func TestClientVsServer(t *testing.T) {
 			}
 
 			var publisherTunnel gortsplib.Tunnel
-			if ca.publisherTunnel == "http" {
+			switch ca.publisherTunnel {
+			case "http":
 				publisherTunnel = gortsplib.TunnelHTTP
-			} else {
+			case "websocket":
+				publisherTunnel = gortsplib.TunnelWebSocket
+			default:
 				publisherTunnel = gortsplib.TunnelNone
 			}
 
@@ -192,9 +211,12 @@ func TestClientVsServer(t *testing.T) {
 			time.Sleep(1 * time.Second)
 
 			var readerTunnel gortsplib.Tunnel
-			if ca.readerTunnel == "http" {
+			switch ca.readerTunnel {
+			case "http":
 				readerTunnel = gortsplib.TunnelHTTP
-			} else {
+			case "websocket":
+				readerTunnel = gortsplib.TunnelWebSocket
+			default:
 				readerTunnel = gortsplib.TunnelNone
 			}
 

server.go

@@ -420,7 +420,7 @@ func (s *Server) runInner() error {
 					sc := &ServerConn{
 						s:      s,
 						nconn:  nconn,
-						isHTTP: true,
+						tunnel: TunnelHTTP,
 					}
 					sc.initialize()
 					s.conns[sc] = struct{}{}