(gosec) Apply G115 fixes to x/mongo/driver/topology package

Address gosec G115 integer overflow warnings in topology:
- Add SafeConvertNumeric for wire message compression operations
- Add SafeConvertNumeric for server description fields (maxBsonObjectSize, etc.)
- Add SafeConvertNumeric for connection pool size conversions
- Use binaryutil for safe binary operations

Author: prestonvasquez

x/mongo/driver/topology/CMAP_spec_test.go

@@ -427,10 +427,8 @@ func runOperationInThread(t *testing.T, operation map[string]any, testInfo *test
 			t.Fatalf("unable to find thread to wait for: %v", threadName)
 		}
 
-		for {
-			if atomic.LoadInt32(&thread.JobsCompleted) == atomic.LoadInt32(&thread.JobsAssigned) {
-				break
-			}
+		for atomic.LoadInt32(&thread.JobsCompleted) != atomic.LoadInt32(&thread.JobsAssigned) {
+
 		}
 	case "waitForEvent":
 		var targetCount int

x/mongo/driver/topology/connection.go

@@ -13,13 +13,15 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"math"
 	"net"
 	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
 
 	"go.mongodb.org/mongo-driver/v2/internal/driverutil"
+	"go.mongodb.org/mongo-driver/v2/internal/mathutil"
 	"go.mongodb.org/mongo-driver/v2/mongo/address"
 	"go.mongodb.org/mongo-driver/v2/x/bsonx/bsoncore"
 	"go.mongodb.org/mongo-driver/v2/x/mongo/driver"
@@ -226,7 +228,6 @@ func (c *connection) connect(ctx context.Context) (err error) {
 			HTTPClient:              c.config.httpClient,
 		}
 		tlsNc, err := configureTLS(ctx, c.config.tlsConnectionSource, c.nc, c.addr, tlsConfig, ocspOpts)
-
 		if err != nil {
 			return ConnectionError{Wrapped: err, init: true, message: fmt.Sprintf("failed to configure TLS for %s", c.addr)}
 		}
@@ -427,7 +428,12 @@ func (c *connection) readWireMessage(ctx context.Context) ([]byte, error) {
 
 func (c *connection) parseWmSizeBytes(wmSizeBytes [4]byte) (int32, error) {
 	// read the length as an int32
-	size := int32(binary.LittleEndian.Uint32(wmSizeBytes[:]))
+	rawSize := binary.LittleEndian.Uint32(wmSizeBytes[:])
+	if rawSize > uint32(math.MaxInt32) {
+		return 0, fmt.Errorf("message length exceeds int32 max: %d", rawSize)
+	}
+
+	size := int32(rawSize)
 
 	if size < 4 {
 		return 0, fmt.Errorf("malformed message length: %d", size)
@@ -475,7 +481,12 @@ func (c *connection) read(ctx context.Context) (bytesRead []byte, errMsg string,
 	// reading messages from an exhaust cursor.
 	n, err := io.ReadFull(c.nc, sizeBuf[:])
 	if err != nil {
-		if l := int32(n); l == 0 && isCSOTTimeout(err) {
+		nI32, convErr := mathutil.SafeConvertNumeric[int32](n)
+		if convErr != nil {
+			return nil, "incomplete read of message header", convErr
+		}
+
+		if l := nI32; l == 0 && isCSOTTimeout(err) {
 			c.awaitRemainingBytes = &l
 		}
 		return nil, "incomplete read of message header", err
@@ -490,7 +501,12 @@ func (c *connection) read(ctx context.Context) (bytesRead []byte, errMsg string,
 
 	n, err = io.ReadFull(c.nc, dst[4:])
 	if err != nil {
-		remainingBytes := size - 4 - int32(n)
+		nI32, convErr := mathutil.SafeConvertNumeric[int32](n)
+		if convErr != nil {
+			return dst, "incomplete read of full message", convErr
+		}
+
+		remainingBytes := size - 4 - nI32
 		if remainingBytes > 0 && isCSOTTimeout(err) {
 			c.awaitRemainingBytes = &remainingBytes
 		}
@@ -586,15 +602,17 @@ func (c *connection) SetOIDCTokenGenID(genID uint64) {
 // *connection to a Handshaker.
 type initConnection struct{ *connection }
 
-var _ mnet.ReadWriteCloser = initConnection{}
-var _ mnet.Describer = initConnection{}
-var _ mnet.Streamer = initConnection{}
+var (
+	_ mnet.ReadWriteCloser = initConnection{}
+	_ mnet.Describer       = initConnection{}
+	_ mnet.Streamer        = initConnection{}
+)
 
 func (c initConnection) Description() description.Server {
 	if c.connection == nil {
 		return description.Server{}
 	}
-	return c.connection.desc
+	return c.desc
 }
 func (c initConnection) Close() error             { return nil }
 func (c initConnection) ID() string               { return c.id }
@@ -606,18 +624,23 @@ func (c initConnection) LocalAddress() address.Address {
 	}
 	return address.Address(c.nc.LocalAddr().String())
 }
+
 func (c initConnection) Write(ctx context.Context, wm []byte) error {
 	return c.writeWireMessage(ctx, wm)
 }
+
 func (c initConnection) Read(ctx context.Context) ([]byte, error) {
 	return c.readWireMessage(ctx)
 }
+
 func (c initConnection) SetStreaming(streaming bool) {
 	c.setStreaming(streaming)
 }
+
 func (c initConnection) CurrentlyStreaming() bool {
 	return c.getCurrentlyStreaming()
 }
+
 func (c initConnection) SupportsStreaming() bool {
 	return c.canStream
 }
@@ -639,11 +662,13 @@ type Connection struct {
 	mu sync.RWMutex
 }
 
-var _ mnet.ReadWriteCloser = (*Connection)(nil)
-var _ mnet.Describer = (*Connection)(nil)
-var _ mnet.Compressor = (*Connection)(nil)
-var _ mnet.Pinner = (*Connection)(nil)
-var _ driver.Expirable = (*Connection)(nil)
+var (
+	_ mnet.ReadWriteCloser = (*Connection)(nil)
+	_ mnet.Describer       = (*Connection)(nil)
+	_ mnet.Compressor      = (*Connection)(nil)
+	_ mnet.Pinner          = (*Connection)(nil)
+	_ driver.Expirable     = (*Connection)(nil)
+)
 
 // WriteWireMessage handles writing a wire message to the underlying connection.
 func (c *Connection) Write(ctx context.Context, wm []byte) error {
@@ -684,7 +709,13 @@ func (c *Connection) CompressWireMessage(src, dst []byte) ([]byte, error) {
 	}
 	idx, dst := wiremessage.AppendHeaderStart(dst, reqid, respto, wiremessage.OpCompressed)
 	dst = wiremessage.AppendCompressedOriginalOpCode(dst, origcode)
-	dst = wiremessage.AppendCompressedUncompressedSize(dst, int32(len(rem)))
+
+	remI32, err := mathutil.SafeConvertNumeric[int32](len(rem))
+	if err != nil {
+		return nil, err
+	}
+
+	dst = wiremessage.AppendCompressedUncompressedSize(dst, remI32)
 	dst = wiremessage.AppendCompressedCompressorID(dst, c.connection.compressor)
 	opts := driver.CompressionOpts{
 		Compressor: c.connection.compressor,
@@ -696,7 +727,11 @@ func (c *Connection) CompressWireMessage(src, dst []byte) ([]byte, error) {
 		return nil, err
 	}
 	dst = wiremessage.AppendCompressedCompressedMessage(dst, compressed)
-	return bsoncore.UpdateLength(dst, idx, int32(len(dst[idx:]))), nil
+	length, err := mathutil.SafeConvertNumeric[int32](len(dst[idx:]))
+	if err != nil {
+		return nil, err
+	}
+	return bsoncore.UpdateLength(dst, idx, length), nil
 }
 
 // Description returns the server description of the server this connection is connected to.

x/mongo/driver/topology/connection_errors_test.go

@@ -5,7 +5,6 @@
 // a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
 
 //go:build go1.13
-// +build go1.13
 
 package topology
 

x/mongo/driver/topology/errors.go

@@ -16,6 +16,7 @@ import (
 	"strings"
 	"time"
 
+	"go.mongodb.org/mongo-driver/v2/internal/mathutil"
 	"go.mongodb.org/mongo-driver/v2/x/mongo/driver/description"
 )
 
@@ -118,13 +119,24 @@ func (w WaitQueueTimeoutError) Error() string {
 
 	msg := fmt.Sprintf("%s; total connections: %d, maxPoolSize: %d, ", errorMsg, w.totalConnections, w.maxPoolSize)
 	if pinnedConnections := w.pinnedConnections; pinnedConnections != nil {
-		openConnectionCount := uint64(w.totalConnections) -
+		var totcalConnectionsWarning string
+
+		totalConnections, err := mathutil.SafeConvertNumeric[uint64](w.totalConnections)
+		if err != nil {
+			totcalConnectionsWarning = fmt.Sprintf("[WARNING]: totalConnections is negative (%d); this may indicate a bug in the driver. ",
+				w.totalConnections)
+			totalConnections = 0
+		}
+
+		openConnectionCount := totalConnections -
 			pinnedConnections.cursorConnections -
 			pinnedConnections.transactionConnections
-		msg += fmt.Sprintf("connections in use by cursors: %d, connections in use by transactions: %d, connections in use by other operations: %d, ",
+
+		msg += fmt.Sprintf("connections in use by cursors: %d, connections in use by transactions: %d, connections in use by other operations: %d%s, ",
 			pinnedConnections.cursorConnections,
 			pinnedConnections.transactionConnections,
 			openConnectionCount,
+			totcalConnectionsWarning,
 		)
 	}
 	msg += fmt.Sprintf("idle connections: %d, wait duration: %s", w.availableConnections, w.waitDuration.String())

x/mongo/driver/topology/fsm.go

@@ -125,7 +125,7 @@ func (f *fsm) apply(s description.Server) (description.Topology, description.Ser
 		SetName: f.SetName,
 	}
 
-	f.Topology.SessionTimeoutMinutes = serverTimeoutMinutes
+	f.SessionTimeoutMinutes = serverTimeoutMinutes
 
 	if _, ok := f.findServer(s.Addr); !ok {
 		return f.Topology, s
@@ -157,7 +157,7 @@ func (f *fsm) apply(s description.Server) (description.Topology, description.Ser
 					SupportedWireVersions.Min,
 					MinSupportedMongoDBVersion,
 				)
-				f.Topology.CompatibilityErr = f.compatibilityErr
+				f.CompatibilityErr = f.compatibilityErr
 				return f.Topology, s
 			}
 
@@ -169,7 +169,7 @@ func (f *fsm) apply(s description.Server) (description.Topology, description.Ser
 					server.WireVersion.Min,
 					SupportedWireVersions.Max,
 				)
-				f.Topology.CompatibilityErr = f.compatibilityErr
+				f.CompatibilityErr = f.compatibilityErr
 				return f.Topology, s
 			}
 		}

x/mongo/driver/topology/pool.go

@@ -10,6 +10,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"math"
 	"net"
 	"sync"
 	"sync/atomic"
@@ -18,6 +19,7 @@ import (
 	"go.mongodb.org/mongo-driver/v2/bson"
 	"go.mongodb.org/mongo-driver/v2/event"
 	"go.mongodb.org/mongo-driver/v2/internal/logger"
+	"go.mongodb.org/mongo-driver/v2/internal/mathutil"
 	"go.mongodb.org/mongo-driver/v2/mongo/address"
 	"go.mongodb.org/mongo-driver/v2/x/mongo/driver"
 )
@@ -158,7 +160,6 @@ func logPoolMessage(pool *pool, msg string, keysAndValues ...any) {
 			ServerHost: host,
 			ServerPort: port,
 		}, keysAndValues...)...)
-
 }
 
 type reason struct {
@@ -241,7 +242,7 @@ func newPool(config poolConfig, connOpts ...ConnectionOption) *pool {
 	var ctx context.Context
 	ctx, pool.cancelBackgroundCtx = context.WithCancel(context.Background())
 
-	for i := 0; i < int(pool.maxConnecting); i++ {
+	for i := uint64(0); i < pool.maxConnecting; i++ {
 		pool.backgroundDone.Add(1)
 		go pool.createConnections(ctx, pool.backgroundDone)
 	}
@@ -1357,7 +1358,16 @@ func (p *pool) maintain(ctx context.Context, wg *sync.WaitGroup) {
 		// the number of connections requested to max 10 at a time to prevent overshooting
 		// minPoolSize in case other checkOut() calls are requesting new connections, too.
 		total := p.totalConnectionCount()
-		n := int(p.minSize) - total - len(wantConns)
+
+		// Since this is a forced mod 10 operation, we can safely ignore overflows.
+		minSize, err := mathutil.SafeConvertNumeric[int](p.minSize)
+		if err != nil {
+			// Ignore overflow here because this is only used to drive pool growth
+			// hints.
+			minSize = math.MaxInt
+		}
+
+		n := minSize - total - len(wantConns)
 		if n > 10 {
 			n = 10
 		}

x/mongo/driver/topology/server_test.go

@@ -5,7 +5,6 @@
 // a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
 
 //go:build go1.13
-// +build go1.13
 
 package topology
 

x/mongo/driver/topology/tls_connection_source_1_16.go

@@ -5,7 +5,6 @@
 // a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
 
 //go:build !go1.17
-// +build !go1.17
 
 package topology
 

x/mongo/driver/topology/tls_connection_source_1_17.go

@@ -5,7 +5,6 @@
 // a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
 
 //go:build go1.17
-// +build go1.17
 
 package topology
 

x/mongo/driver/topology/topology.go

@@ -772,7 +772,7 @@ func (t *Topology) pollSRVRecords(hosts string) {
 			return
 		}
 		topoKind := t.Description().Kind
-		if !(topoKind == description.Unknown || topoKind == description.TopologyKindSharded) {
+		if topoKind != description.Unknown && topoKind != description.TopologyKindSharded {
 			break
 		}