Upgrade datarhei/gosrt

go.mod

@@ -5,7 +5,7 @@ go 1.18
 require (
 	github.com/99designs/gqlgen v0.17.12
 	github.com/atrox/haikunatorgo/v2 v2.0.1
-	github.com/datarhei/gosrt v0.1.2
+	github.com/datarhei/gosrt v0.2.1-0.20220817065416-5d1fd7a090e5
 	github.com/datarhei/joy4 v0.0.0-20220728180719-f752080f4a36
 	github.com/go-playground/validator/v10 v10.11.0
 	github.com/gobwas/glob v0.2.3
@@ -24,7 +24,7 @@ require (
 	github.com/swaggo/swag v1.8.3
 	github.com/vektah/gqlparser/v2 v2.4.6
 	github.com/xeipuuv/gojsonschema v1.2.0
-	golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d
+	golang.org/x/crypto v0.0.0-20220722155217-630584e8d5aa
 	golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4
 )
 

go.sum

@@ -74,8 +74,8 @@ github.com/cpuguy83/go-md2man/v2 v2.0.0-20190314233015-f79a8a8ca69d/go.mod h1:ma
 github.com/cpuguy83/go-md2man/v2 v2.0.1 h1:r/myEWzV9lfsM1tFLgDyu0atFtJ1fXn261LKYj/3DxU=
 github.com/cpuguy83/go-md2man/v2 v2.0.1/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
-github.com/datarhei/gosrt v0.1.2 h1:rGOP2Xkbi52z4tLzBwCBw2TKt7BrfTO2LmEVY+yWf1M=
-github.com/datarhei/gosrt v0.1.2/go.mod h1:IftDbZGIIC9OvQO5on5ZpU0iB/JX/PFOqGXORbwHYQM=
+github.com/datarhei/gosrt v0.2.1-0.20220817065416-5d1fd7a090e5 h1:lTmd1w7UpULXD/B6LfRu4QJHDILq9W5atyLP+RCkoMA=
+github.com/datarhei/gosrt v0.2.1-0.20220817065416-5d1fd7a090e5/go.mod h1:wyoTu+DG45XRuCgEq/y+R8nhZCrJbOyQKn+SwNrNVZ8=
 github.com/datarhei/joy4 v0.0.0-20220728180719-f752080f4a36 h1:ppjcv7wazy4d7vANREERXkSAUnhV/nfT2a+13u4ZijQ=
 github.com/datarhei/joy4 v0.0.0-20220728180719-f752080f4a36/go.mod h1:Jcw/6jZDQQmPx8A7INEkXmuEF7E9jjBbSTfVSLwmiQw=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -387,9 +387,8 @@ golang.org/x/crypto v0.0.0-20210817164053-32db794688a5/go.mod h1:GvvjBRRGRdwPK5y
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.0.0-20211215153901-e495a2d5b3d3/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/crypto v0.0.0-20220411220226-7b82a4e95df4/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
-golang.org/x/crypto v0.0.0-20220525230936-793ad666bf5e/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
-golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d h1:sK3txAijHtOK88l68nt020reeT1ZdKLIYetKl95FzVY=
-golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
+golang.org/x/crypto v0.0.0-20220722155217-630584e8d5aa h1:zuSxTR4o9y82ebqCUJYNGJbGPo6sKVl54f/TVDObg1c=
+golang.org/x/crypto v0.0.0-20220722155217-630584e8d5aa/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190306152737-a1d7652674e8/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190510132918-efd6b22b2522/go.mod h1:ZjyILWgesfNpC6sMxTJOJm9Kp84zZh5NQWvqDGG3Qr8=

vendor/github.com/datarhei/gosrt/Makefile

@@ -5,7 +5,7 @@ all: build
 
 ## test: Run all tests
 test:
-	go test -race -coverprofile=/dev/null -timeout 15s -v ./...
+	go test -race -coverprofile=/dev/null -timeout 60s -v ./...
 
 ## vet: Analyze code for potential errors
 vet:
@@ -43,7 +43,7 @@ server:
 
 ## coverage: Generate code coverage analysis
 coverage:
-	go test -race -coverprofile=cover.out -timeout 15s -v ./...
+	go test -race -coverprofile=cover.out -timeout 60s -v ./...
 	go tool cover -html=cover.out -o cover.html
 
 ## commit: Prepare code for commit (vet, fmt, test)

vendor/github.com/datarhei/gosrt/config.go

@@ -218,17 +218,17 @@ func DefaultConfig() Config {
 
 // UnmarshalURL takes a SRT URL and parses out the configuration. A SRT URL is
 // srt://[host]:[port]?[key1]=[value1]&[key2]=[value2]...
-func (c *Config) UnmarshalURL(addr string) error {
+func (c *Config) UnmarshalURL(addr string) (string, error) {
 	u, err := url.Parse(addr)
 	if err != nil {
-		return err
+		return "", err
 	}
 
 	if u.Scheme != "srt" {
-		return fmt.Errorf("the URL doesn't seem to be an srt:// URL")
+		return "", fmt.Errorf("the URL doesn't seem to be an srt:// URL")
 	}
 
-	return c.UnmarshalQuery(u.RawQuery)
+	return u.Host, c.UnmarshalQuery(u.RawQuery)
 }
 
 // UnmarshalQuery parses a query string and interprets it as a configuration

vendor/github.com/datarhei/gosrt/connection.go

@@ -93,6 +93,7 @@ type srtConn struct {
 
 	config Config
 
+	cryptoLock             sync.Mutex
 	crypto                 crypto.Crypto
 	keyBaseEncryption      packet.PacketEncryption
 	kmPreAnnounceCountdown uint64
@@ -453,6 +454,7 @@ func (c *srtConn) pop(p packet.Packet) {
 	p.Header().DestinationSocketId = c.peerSocketId
 
 	if !p.Header().IsControlPacket {
+		c.cryptoLock.Lock()
 		if c.crypto != nil {
 			p.Header().KeyBaseEncryptionFlag = c.keyBaseEncryption
 			c.crypto.EncryptOrDecryptPayload(p.Data(), p.Header().KeyBaseEncryptionFlag, p.Header().PacketSequenceNumber.Val())
@@ -483,6 +485,7 @@ func (c *srtConn) pop(p packet.Packet) {
 				c.crypto.GenerateSEK(c.keyBaseEncryption.Opposite())
 			}
 		}
+		c.cryptoLock.Unlock()
 
 		c.log("data:send:dump", func() string { return p.Dump() })
 	}
@@ -614,6 +617,7 @@ func (c *srtConn) handlePacket(p packet.Packet) {
 
 		c.log("data:recv:dump", func() string { return p.Dump() })
 
+		c.cryptoLock.Lock()
 		if c.crypto != nil {
 			if header.KeyBaseEncryptionFlag != 0 {
 				if err := c.crypto.EncryptOrDecryptPayload(p.Data(), header.KeyBaseEncryptionFlag, header.PacketSequenceNumber.Val()); err != nil {
@@ -625,6 +629,7 @@ func (c *srtConn) handlePacket(p packet.Packet) {
 				c.statistics.byteRecvUndecrypt += p.Len()
 			}
 		}
+		c.cryptoLock.Unlock()
 
 		// Put the packet into receive congestion control
 		c.recv.Push(p)
@@ -759,8 +764,11 @@ func (c *srtConn) handleKMRequest(p packet.Packet) {
 
 	c.statistics.pktRecvKM++
 
+	c.cryptoLock.Lock()
+
 	if c.crypto == nil {
 		c.log("control:recv:KM:error", func() string { return "connection is not encrypted" })
+		c.cryptoLock.Unlock()
 		return
 	}
 
@@ -769,6 +777,7 @@ func (c *srtConn) handleKMRequest(p packet.Packet) {
 	if err := p.UnmarshalCIF(cif); err != nil {
 		c.statistics.pktRecvInvalid++
 		c.log("control:recv:KM:error", func() string { return fmt.Sprintf("invalid KM: %s", err) })
+		c.cryptoLock.Unlock()
 		return
 	}
 
@@ -779,15 +788,23 @@ func (c *srtConn) handleKMRequest(p packet.Packet) {
 		c.log("control:recv:KM:error", func() string {
 			return "invalid KM. wants to reset the key that is already in use"
 		})
+		c.cryptoLock.Unlock()
 		return
 	}
 
 	if err := c.crypto.UnmarshalKM(cif, c.config.Passphrase); err != nil {
 		c.statistics.pktRecvInvalid++
 		c.log("control:recv:KM:error", func() string { return fmt.Sprintf("invalid KM: %s", err) })
+		c.cryptoLock.Unlock()
 		return
 	}
 
+	// Switch the keys
+	c.keyBaseEncryption = c.keyBaseEncryption.Opposite()
+
+	c.cryptoLock.Unlock()
+
+	// Send KM Response
 	p.Header().SubType = packet.EXTTYPE_KMRSP
 
 	c.statistics.pktSentKM++
@@ -801,6 +818,9 @@ func (c *srtConn) handleKMResponse(p packet.Packet) {
 
 	c.statistics.pktRecvKM++
 
+	c.cryptoLock.Lock()
+	defer c.cryptoLock.Unlock()
+
 	if c.crypto == nil {
 		c.log("control:recv:KM:error", func() string { return "connection is not encrypted" })
 		return

vendor/github.com/datarhei/gosrt/dial.go

@@ -66,7 +66,8 @@ type connResponse struct {
 // The address is of the form "host:port".
 //
 // Example:
-//  Dial("srt", "127.0.0.1:3000", DefaultConfig())
+//
+//	Dial("srt", "127.0.0.1:3000", DefaultConfig())
 //
 // In case of an error the returned Conn is nil and the error is non-nil.
 func Dial(network, address string, config Config) (Conn, error) {
@@ -663,6 +664,17 @@ func (dl *dialer) Read(p []byte) (n int, err error) {
 	return dl.conn.Read(p)
 }
 
+func (dl *dialer) readPacket() (packet.Packet, error) {
+	if err := dl.checkConnection(); err != nil {
+		return nil, err
+	}
+
+	dl.connLock.RLock()
+	defer dl.connLock.RUnlock()
+
+	return dl.conn.readPacket()
+}
+
 func (dl *dialer) Write(p []byte) (n int, err error) {
 	if err := dl.checkConnection(); err != nil {
 		return 0, err
@@ -674,6 +686,17 @@ func (dl *dialer) Write(p []byte) (n int, err error) {
 	return dl.conn.Write(p)
 }
 
+func (dl *dialer) writePacket(p packet.Packet) error {
+	if err := dl.checkConnection(); err != nil {
+		return err
+	}
+
+	dl.connLock.RLock()
+	defer dl.connLock.RUnlock()
+
+	return dl.conn.writePacket(p)
+}
+
 func (dl *dialer) SetDeadline(t time.Time) error      { return dl.conn.SetDeadline(t) }
 func (dl *dialer) SetReadDeadline(t time.Time) error  { return dl.conn.SetReadDeadline(t) }
 func (dl *dialer) SetWriteDeadline(t time.Time) error { return dl.conn.SetWriteDeadline(t) }

vendor/github.com/datarhei/gosrt/listen.go

@@ -157,7 +157,8 @@ type listener struct {
 // The address has the form "host:port".
 //
 // Examples:
-//  Listen("srt", "127.0.0.1:3000", DefaultConfig())
+//
+//	Listen("srt", "127.0.0.1:3000", DefaultConfig())
 //
 // In case of an error, the returned Listener is nil and the error is non-nil.
 func Listen(network, address string, config Config) (Listener, error) {
@@ -177,36 +178,45 @@ func Listen(network, address string, config Config) (Listener, error) {
 		config: config,
 	}
 
-	raddr, err := net.ResolveUDPAddr("udp", address)
+	lc := net.ListenConfig{
+		Control: func(network, address string, c syscall.RawConn) error {
+			var opErr error
+			err := c.Control(func(fd uintptr) {
+				// Set REUSEPORT
+				opErr = syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, syscall.SO_REUSEPORT, 1)
+				if opErr != nil {
+					return
+				}
+
+				// Set TOS
+				if config.IPTOS > 0 {
+					opErr = syscall.SetsockoptInt(int(fd), syscall.IPPROTO_IP, syscall.IP_TOS, config.IPTOS)
+					if opErr != nil {
+						return
+					}
+				}
+
+				// Set TTL
+				if config.IPTTL > 0 {
+					opErr = syscall.SetsockoptInt(int(fd), syscall.IPPROTO_IP, syscall.IP_TTL, config.IPTTL)
+					if opErr != nil {
+						return
+					}
+				}
+			})
+			if err != nil {
+				return err
+			}
+			return opErr
+		},
+	}
+
+	lp, err := lc.ListenPacket(context.Background(), "udp", address)
 	if err != nil {
-		return nil, fmt.Errorf("listen: unable to resolve address: %w", err)
+		return nil, fmt.Errorf("listen: %w", err)
 	}
 
-	pc, err := net.ListenUDP("udp", raddr)
-	if err != nil {
-		return nil, fmt.Errorf("listen: failed listening: %w", err)
-	}
-
-	file, err := pc.File()
-	if err != nil {
-		return nil, err
-	}
-
-	// Set TOS
-	if config.IPTOS > 0 {
-		err = syscall.SetsockoptInt(int(file.Fd()), syscall.IPPROTO_IP, syscall.IP_TOS, config.IPTOS)
-		if err != nil {
-			return nil, fmt.Errorf("listen: failed setting socket option TOS: %w", err)
-		}
-	}
-
-	// Set TTL
-	if config.IPTTL > 0 {
-		err = syscall.SetsockoptInt(int(file.Fd()), syscall.IPPROTO_IP, syscall.IP_TTL, config.IPTTL)
-		if err != nil {
-			return nil, fmt.Errorf("listen: failed setting socket option TTL: %w", err)
-		}
-	}
+	pc := lp.(*net.UDPConn)
 
 	ln.pc = pc
 	ln.addr = pc.LocalAddr()

vendor/github.com/datarhei/gosrt/pubsub.go

@@ -12,7 +12,7 @@ import (
 // PubSub is a publish/subscriber service for SRT connections.
 type PubSub interface {
 	// Publish accepts a SRT connection where it reads from. It blocks
-	// until the connection closes. The returned error indicated why it
+	// until the connection closes. The returned error indicates why it
 	// stopped. There can be only one publisher.
 	Publish(c Conn) error
 
@@ -23,6 +23,11 @@ type PubSub interface {
 	Subscribe(c Conn) error
 }
 
+type packetReadWriter interface {
+	readPacket() (packet.Packet, error)
+	writePacket(p packet.Packet) error
+}
+
 // pubSub is an implementation of the PubSub interface
 type pubSub struct {
 	incoming      chan packet.Packet
@@ -102,28 +107,30 @@ func (pb *pubSub) Publish(c Conn) error {
 
 	var p packet.Packet
 	var err error
-	conn, ok := c.(*srtConn)
+	conn, ok := c.(packetReadWriter)
 	if !ok {
 		err := fmt.Errorf("the provided connection is not a SRT connection")
 		pb.logger.Print("pubsub:error", 0, 1, func() string { return err.Error() })
 		return err
 	}
 
-	pb.logger.Print("pubsub:publish", conn.SocketId(), 1, func() string { return "new publisher" })
+	socketId := c.SocketId()
+
+	pb.logger.Print("pubsub:publish", socketId, 1, func() string { return "new publisher" })
 
 	pb.publish = true
 
 	for {
 		p, err = conn.readPacket()
 		if err != nil {
-			pb.logger.Print("pubsub:error", conn.SocketId(), 1, func() string { return err.Error() })
+			pb.logger.Print("pubsub:error", socketId, 1, func() string { return err.Error() })
 			break
 		}
 
 		select {
 		case pb.incoming <- p:
 		default:
-			pb.logger.Print("pubsub:error", conn.SocketId(), 1, func() string { return "incoming queue is full" })
+			pb.logger.Print("pubsub:error", socketId, 1, func() string { return "incoming queue is full" })
 		}
 	}
 
@@ -135,7 +142,7 @@ func (pb *pubSub) Publish(c Conn) error {
 func (pb *pubSub) Subscribe(c Conn) error {
 	l := make(chan packet.Packet, 1024)
 	socketId := c.SocketId()
-	conn, ok := c.(*srtConn)
+	conn, ok := c.(packetReadWriter)
 	if !ok {
 		err := fmt.Errorf("the provided connection is not a SRT connection")
 		pb.logger.Print("pubsub:error", 0, 1, func() string { return err.Error() })

vendor/golang.org/x/crypto/AUTHORS

@@ -1,3 +0,0 @@
-# This source code refers to The Go Authors for copyright purposes.
-# The master list of authors is in the main Go distribution,
-# visible at https://tip.golang.org/AUTHORS.

vendor/golang.org/x/crypto/CONTRIBUTORS

@@ -1,3 +0,0 @@
-# This source code was written by the Go contributors.
-# The master list of contributors is in the main Go distribution,
-# visible at https://tip.golang.org/CONTRIBUTORS.

vendor/modules.txt

@@ -45,7 +45,7 @@ github.com/cespare/xxhash/v2
 # github.com/cpuguy83/go-md2man/v2 v2.0.1
 ## explicit; go 1.11
 github.com/cpuguy83/go-md2man/v2/md2man
-# github.com/datarhei/gosrt v0.1.2
+# github.com/datarhei/gosrt v0.2.1-0.20220817065416-5d1fd7a090e5
 ## explicit; go 1.16
 github.com/datarhei/gosrt
 github.com/datarhei/gosrt/internal/circular
@@ -281,7 +281,7 @@ github.com/xrash/smetrics
 # github.com/yusufpapurcu/wmi v1.2.2
 ## explicit; go 1.16
 github.com/yusufpapurcu/wmi
-# golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d
+# golang.org/x/crypto v0.0.0-20220722155217-630584e8d5aa
 ## explicit; go 1.17
 golang.org/x/crypto/acme
 golang.org/x/crypto/acme/autocert