Repository: nats-io/nats.go
Pull request #1857 (2025-04-16 14:32:23)

[FIXED] Bump golang.org/x/crypto to fix vulnerability
Fixes https://www.cve.org/CVERecord?id=CVE-2025-22869

Signed-off-by: Piotr Piotrowski <piotr@synadia.com>


Patch:
diff --git a/go.mod b/go.mod
index 2dc5942..4afd008 100644
--- a/go.mod
+++ b/go.mod
@@ -4,11 +4,11 @@ go 1.23.0
 
 require (
 	github.com/klauspost/compress v1.18.0
-	github.com/nats-io/nkeys v0.4.9
+	github.com/nats-io/nkeys v0.4.11
 	github.com/nats-io/nuid v1.0.1
 )
 
 require (
-	golang.org/x/crypto v0.31.0 // indirect
-	golang.org/x/sys v0.28.0 // indirect
+	golang.org/x/crypto v0.37.0 // indirect
+	golang.org/x/sys v0.32.0 // indirect
 )
diff --git a/go.sum b/go.sum
index 2620b62..0fb272f 100644
--- a/go.sum
+++ b/go.sum
@@ -1,10 +1,10 @@
 github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
 github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
-github.com/nats-io/nkeys v0.4.9 h1:qe9Faq2Gxwi6RZnZMXfmGMZkg3afLLOtrU+gDZJ35b0=
-github.com/nats-io/nkeys v0.4.9/go.mod h1:jcMqs+FLG+W5YO36OX6wFIFcmpdAns+w1Wm6D3I/evE=
+github.com/nats-io/nkeys v0.4.11 h1:q44qGV008kYd9W1b1nEBkNzvnWxtRSQ7A8BoqRrcfa0=
+github.com/nats-io/nkeys v0.4.11/go.mod h1:szDimtgmfOi9n25JpfIdGw12tZFYXqhGxjhVxsatHVE=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
-golang.org/x/crypto v0.31.0 h1:ihbySMvVjLAeSH1IbfcRTkD/iNscyz8rGzjF/E5hV6U=
-golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk=
-golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA=
-golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/crypto v0.37.0 h1:kJNSjF/Xp7kU0iB2Z+9viTPMW4EqqsrywMXLJOOsXSE=
+golang.org/x/crypto v0.37.0/go.mod h1:vg+k43peMZ0pUMhYmVAWysMK35e6ioLh3wB8ZCAfbVc=
+golang.org/x/sys v0.32.0 h1:s77OFDvIQeibCmezSnk/q6iAfkdiQaJi4VzroCFrN20=
+golang.org/x/sys v0.32.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
diff --git a/js.go b/js.go
index 4c956f1..132e69a 100644
--- a/js.go
+++ b/js.go
@@ -3145,12 +3145,11 @@ func (sub *Subscription) Fetch(batch int, opts ...PullOpt) ([]*Msg, error) {
 		go func() {
 			select {
 			case <-ctx.Done():
-				return
 			case <-connStatusChanged:
 				disconnected.Store(true)
 				cancel()
-				return
 			}
+			nc.RemoveStatusListener(connStatusChanged)
 		}()
 		err = sendReq()
 		for err == nil && len(msgs) < batch {
@@ -3404,12 +3403,11 @@ func (sub *Subscription) FetchBatch(batch int, opts ...PullOpt) (MessageBatch, e
 	go func() {
 		select {
 		case <-ctx.Done():
-			return
 		case <-connStatusChanged:
 			disconnected.Store(true)
 			cancel()
-			return
 		}
+		nc.RemoveStatusListener(connStatusChanged)
 	}()
 	requestBatch := batch - len(result.msgs)
 	req := nextRequest{
diff --git a/kv.go b/kv.go
index 3559f0a..dc401d5 100644
--- a/kv.go
+++ b/kv.go
@@ -834,6 +834,10 @@ func (kv *kvs) PurgeDeletes(opts ...PurgeOpt) error {
 		b  strings.Builder
 	)
 	// Do actual purges here.
+	purgeOpts := []JSOpt{}
+	if o.ctx != nil {
+		purgeOpts = append(purgeOpts, Context(o.ctx))
+	}
 	for _, entry := range deleteMarkers {
 		b.WriteString(kv.pre)
 		b.WriteString(entry.Key())
@@ -842,7 +846,7 @@ func (kv *kvs) PurgeDeletes(opts ...PurgeOpt) error {
 		if olderThan > 0 && entry.Created().After(limit) {
 			pr.Keep = 1
 		}
-		if err := kv.js.purgeStream(kv.stream, &pr); err != nil {
+		if err := kv.js.purgeStream(kv.stream, &pr, purgeOpts...); err != nil {
 			return err
 		}
 		b.Reset()
diff --git a/nats.go b/nats.go
index 75d8020..ba7a683 100644
--- a/nats.go
+++ b/nats.go
@@ -572,7 +572,7 @@ type Conn struct {
 	pongs         []chan struct{}
 	scratch       [scratchSize]byte
 	status        Status
-	statListeners map[Status][]chan Status
+	statListeners map[Status]map[chan Status]struct{}
 	initc         bool // true if the connection is performing the initial connect
 	err           error
 	ps            *parseState
@@ -5913,18 +5913,39 @@ func (nc *Conn) StatusChanged(statuses ...Status) chan Status {
 	return ch
 }
 
+// RemoveStatusListener removes a status change listener.
+// If the channel is not closed, it will be closed.
+// Listeners will be removed automatically on status change
+// as well, but this is a way to remove them manually.
+func (nc *Conn) RemoveStatusListener(ch chan (Status)) {
+	nc.mu.Lock()
+	defer nc.mu.Unlock()
+
+	select {
+	case <-ch:
+	default:
+		close(ch)
+	}
+
+	for _, listeners := range nc.statListeners {
+		for l := range listeners {
+			delete(listeners, l)
+		}
+	}
+}
+
 // registerStatusChangeListener registers a channel waiting for a specific status change event.
 // Status change events are non-blocking - if no receiver is waiting for the status change,
 // it will not be sent on the channel. Closed channels are ignored.
 // The lock should be held entering.
 func (nc *Conn) registerStatusChangeListener(status Status, ch chan Status) {
 	if nc.statListeners == nil {
-		nc.statListeners = make(map[Status][]chan Status)
+		nc.statListeners = make(map[Status]map[chan Status]struct{})
 	}
 	if _, ok := nc.statListeners[status]; !ok {
-		nc.statListeners[status] = make([]chan Status, 0)
+		nc.statListeners[status] = make(map[chan Status]struct{})
 	}
-	nc.statListeners[status] = append(nc.statListeners[status], ch)
+	nc.statListeners[status][ch] = struct{}{}
 }
 
 // sendStatusEvent sends connection status event to all channels.
@@ -5932,20 +5953,18 @@ func (nc *Conn) registerStatusChangeListener(status Status, ch chan Status) {
 // will not block. Lock should be held entering.
 func (nc *Conn) sendStatusEvent(s Status) {
 Loop:
-	for i := 0; i < len(nc.statListeners[s]); i++ {
+	for ch := range nc.statListeners[s] {
 		// make sure channel is not closed
 		select {
-		case <-nc.statListeners[s][i]:
+		case <-ch:
 			// if chan is closed, remove it
-			nc.statListeners[s][i] = nc.statListeners[s][len(nc.statListeners[s])-1]
-			nc.statListeners[s] = nc.statListeners[s][:len(nc.statListeners[s])-1]
-			i--
+			delete(nc.statListeners[s], ch)
 			continue Loop
 		default:
 		}
 		// only send event if someone's listening
 		select {
-		case nc.statListeners[s][i] <- s:
+		case ch <- s:
 		default:
 		}
 	}


Test patch:
diff --git a/go_test.mod b/go_test.mod
index 5673259..66a91fb 100644
--- a/go_test.mod
+++ b/go_test.mod
@@ -6,11 +6,10 @@ require (
 	github.com/golang/protobuf v1.4.2
 	github.com/klauspost/compress v1.18.0
 	github.com/nats-io/jwt v1.2.2
-	github.com/nats-io/nats-server/v2 v2.11.0
-	github.com/nats-io/nkeys v0.4.10
+	github.com/nats-io/nats-server/v2 v2.11.1
+	github.com/nats-io/nkeys v0.4.11
 	github.com/nats-io/nuid v1.0.1
 	go.uber.org/goleak v1.3.0
-	golang.org/x/text v0.23.0
 	google.golang.org/protobuf v1.23.0
 )
 
@@ -18,7 +17,8 @@ require (
 	github.com/google/go-tpm v0.9.3 // indirect
 	github.com/minio/highwayhash v1.0.3 // indirect
 	github.com/nats-io/jwt/v2 v2.7.3 // indirect
-	golang.org/x/crypto v0.36.0 // indirect
-	golang.org/x/sys v0.31.0 // indirect
+	go.uber.org/automaxprocs v1.6.0 // indirect
+	golang.org/x/crypto v0.37.0 // indirect
+	golang.org/x/sys v0.32.0 // indirect
 	golang.org/x/time v0.11.0 // indirect
 )
diff --git a/go_test.sum b/go_test.sum
index 7f6017f..9792fb9 100644
--- a/go_test.sum
+++ b/go_test.sum
@@ -26,30 +26,32 @@ github.com/nats-io/jwt/v2 v2.7.3 h1:6bNPK+FXgBeAqdj4cYQ0F8ViHRbi7woQLq4W29nUAzE=
 github.com/nats-io/jwt/v2 v2.7.3/go.mod h1:GvkcbHhKquj3pkioy5put1wvPxs78UlZ7D/pY+BgZk4=
 github.com/nats-io/nats-server/v2 v2.11.0 h1:fdwAT1d6DZW/4LUz5rkvQUe5leGEwjjOQYntzVRKvjE=
 github.com/nats-io/nats-server/v2 v2.11.0/go.mod h1:leXySghbdtXSUmWem8K9McnJ6xbJOb0t9+NQ5HTRZjI=
+github.com/nats-io/nats-server/v2 v2.11.1 h1:LwdauqMqMNhTxTN3+WFTX6wGDOKntHljgZ+7gL5HCnk=
+github.com/nats-io/nats-server/v2 v2.11.1/go.mod h1:leXySghbdtXSUmWem8K9McnJ6xbJOb0t9+NQ5HTRZjI=
 github.com/nats-io/nkeys v0.2.0/go.mod h1:XdZpAbhgyyODYqjTawOnIOI7VlbKSarI9Gfy1tqEu/s=
-github.com/nats-io/nkeys v0.4.10 h1:glmRrpCmYLHByYcePvnTBEAwawwapjCPMjy2huw20wc=
-github.com/nats-io/nkeys v0.4.10/go.mod h1:OjRrnIKnWBFl+s4YK5ChQfvHP2fxqZexrKJoVVyWB3U=
+github.com/nats-io/nkeys v0.4.11 h1:q44qGV008kYd9W1b1nEBkNzvnWxtRSQ7A8BoqRrcfa0=
+github.com/nats-io/nkeys v0.4.11/go.mod h1:szDimtgmfOi9n25JpfIdGw12tZFYXqhGxjhVxsatHVE=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk=
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
+go.uber.org/automaxprocs v1.6.0 h1:O3y2/QNTOdbF+e/dpXNNW7Rx2hZ4sTIPyybbxyNqTUs=
+go.uber.org/automaxprocs v1.6.0/go.mod h1:ifeIMSnPZuznNm6jmdzmU3/bfk01Fe2fotchwEFJ8r8=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20200323165209-0ec3e9974c59/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
-golang.org/x/crypto v0.36.0 h1:AnAEvhDddvBdpY+uR+MyHmuZzzNqXSe/GvuDeob5L34=
-golang.org/x/crypto v0.36.0/go.mod h1:Y4J0ReaxCR1IMaabaSMugxJES1EpwhBHhv2bDHklZvc=
+golang.org/x/crypto v0.37.0 h1:kJNSjF/Xp7kU0iB2Z+9viTPMW4EqqsrywMXLJOOsXSE=
+golang.org/x/crypto v0.37.0/go.mod h1:vg+k43peMZ0pUMhYmVAWysMK35e6ioLh3wB8ZCAfbVc=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
-golang.org/x/sys v0.31.0 h1:ioabZlmFYtWhL+TRYpcnNlLwhyxaM9kWTDEmfnprqik=
-golang.org/x/sys v0.31.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
+golang.org/x/sys v0.32.0 h1:s77OFDvIQeibCmezSnk/q6iAfkdiQaJi4VzroCFrN20=
+golang.org/x/sys v0.32.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
-golang.org/x/text v0.23.0 h1:D71I7dUrlY+VX0gQShAThNGHFxZ13dGLBHQLVl1mJlY=
-golang.org/x/text v0.23.0/go.mod h1:/BLNzu4aZCJ1+kcD0DNRotWKage4q2rGVAg4o22unh4=
 golang.org/x/time v0.11.0 h1:/bpjEDfN9tkoN/ryeYHnv5hcMlc8ncjMcM4XBk5NWV0=
 golang.org/x/time v0.11.0/go.mod h1:CDIdPxbZBQxdj6cxyCIdrNogrJKMJ7pr37NYpMcMDSg=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
diff --git a/test/conn_test.go b/test/conn_test.go
index 9e141e3..63bc21f 100644
--- a/test/conn_test.go
+++ b/test/conn_test.go
@@ -3059,6 +3059,64 @@ func TestConnStatusChangedEvents(t *testing.T) {
 	})
 }
 
+func TestRemoveStatusListener(t *testing.T) {
+	t.Run("with channel not closed", func(t *testing.T) {
+		s := RunDefaultServer()
+		defer s.Shutdown()
+		nc, err := nats.Connect(s.ClientURL())
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		defer nc.Close()
+		statusCh := nc.StatusChanged()
+		done := make(chan struct{})
+		go func() {
+			_, ok := <-statusCh
+			if !ok {
+				done <- struct{}{}
+				return
+			}
+		}()
+		time.Sleep(50 * time.Millisecond)
+
+		nc.RemoveStatusListener(statusCh)
+
+		select {
+		case <-done:
+		case <-time.After(time.Second):
+			t.Fatal("Expected to receive done signal")
+		}
+	})
+	t.Run("with channel closed", func(t *testing.T) {
+		s := RunDefaultServer()
+		defer s.Shutdown()
+		nc, err := nats.Connect(s.ClientURL())
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		defer nc.Close()
+		statusCh := nc.StatusChanged()
+		done := make(chan struct{})
+		go func() {
+			_, ok := <-statusCh
+			if !ok {
+				done <- struct{}{}
+				return
+			}
+		}()
+		time.Sleep(50 * time.Millisecond)
+
+		close(statusCh)
+		nc.RemoveStatusListener(statusCh)
+
+		select {
+		case <-done:
+		case <-time.After(time.Second):
+			t.Fatal("Expected to receive done signal")
+		}
+	})
+}
+
 func TestTLSHandshakeFirst(t *testing.T) {
 	s, opts := RunServerWithConfig("./configs/tls.conf")
 	defer s.Shutdown()