option to specify tls min version

client.go

@@ -6,6 +6,7 @@ import (
 	"crypto/x509"
 	"errors"
 	"fmt"
+	"math"
 	"os"
 	"time"
 
@@ -54,6 +55,8 @@ type conf struct {
 	clientCertKeyPath string
 	// wether to skip the cert validity check
 	skipTLSValidation bool
+	// configure tls min version
+	tlsMinVersion uint16
 }
 
 const (
@@ -99,6 +102,21 @@ func getLabels(labelsArg sobek.Value, rt *sobek.Runtime) mqttMetricsLabels {
 	return labels
 }
 
+func tlsVersionStringToNumber(version string) (uint16, error) {
+	versionMap := map[string]uint16{
+		"TLS 1.0": tls.VersionTLS10,
+		"TLS 1.1": tls.VersionTLS11,
+		"TLS 1.2": tls.VersionTLS12,
+		"TLS 1.3": tls.VersionTLS13,
+	}
+
+	if versionNumber, ok := versionMap[version]; ok {
+		return versionNumber, nil
+	}
+
+	return 0, errors.New("unknown TLS version")
+}
+
 //nolint:nosnakecase // their choice not mine
 func (m *MqttAPI) client(c sobek.ConstructorCall) *sobek.Object {
 	serversArray := c.Argument(0)
@@ -140,7 +158,11 @@ func (m *MqttAPI) client(c sobek.ConstructorCall) *sobek.Object {
 	if timeoutValue == nil || sobek.IsUndefined(timeoutValue) {
 		common.Throw(rt, errors.New("Client requires a timeout value"))
 	}
-	clientConf.timeout = uint(timeoutValue.ToInteger())
+	timeoutIntValue := timeoutValue.ToInteger()
+	if timeoutIntValue < 0 {
+		common.Throw(rt, errors.New("negative timeout value is not allowed"))
+	}
+	clientConf.timeout = uint(timeoutIntValue)
 
 	// optional args
 	if caRootPathValue := c.Argument(6); caRootPathValue == nil || sobek.IsUndefined(caRootPathValue) {
@@ -168,12 +190,31 @@ func (m *MqttAPI) client(c sobek.ConstructorCall) *sobek.Object {
 	skipTLS := c.Argument(10)
 	clientConf.skipTLSValidation = skipTLS.ToBoolean()
 
+	if tlsMinVersionValue := c.Argument(11); tlsMinVersionValue == nil || sobek.IsUndefined(tlsMinVersionValue) {
+		clientConf.tlsMinVersion = tls.VersionTLS13
+	} else {
+		tlsMinVersion, err := tlsVersionStringToNumber(tlsMinVersionValue.String())
+		if err != nil {
+			common.Throw(m.vu.Runtime(), err)
+		} else {
+			clientConf.tlsMinVersion = tlsMinVersion
+		}
+	}
+
 	client := &client{
 		vu:      m.vu,
 		metrics: &metrics,
 		conf:    clientConf,
 		obj:     rt.NewObject(),
 	}
+
+	m.defineRuntimeMethods(client)
+
+	return client.obj
+}
+
+func (m *MqttAPI) defineRuntimeMethods(client *client) {
+	rt := m.vu.Runtime()
 	must := func(err error) {
 		if err != nil {
 			common.Throw(rt, err)
@@ -196,14 +237,18 @@ func (m *MqttAPI) client(c sobek.ConstructorCall) *sobek.Object {
 
 	must(client.obj.DefineDataProperty(
 		"close", rt.ToValue(client.Close), sobek.FLAG_FALSE, sobek.FLAG_FALSE, sobek.FLAG_TRUE))
-
-	return client.obj
 }
 
 // Connect create a connection to mqtt
 func (c *client) Connect() error {
 	opts := paho.NewClientOptions()
 
+	// check timeout value
+	timeoutValue, err := safeUintToInt64(c.conf.timeout)
+	if err != nil {
+		panic("timeout value is too large")
+	}
+
 	var tlsConfig *tls.Config
 	// Use root CA if specified
 	if len(c.conf.caRootPath) > 0 {
@@ -218,7 +263,7 @@ func (c *client) Connect() error {
 		}
 		tlsConfig = &tls.Config{
 			RootCAs:    rootCA,
-			MinVersion: tls.VersionTLS13,
+			MinVersion: c.conf.tlsMinVersion, // #nosec G402
 		}
 	}
 	// Use local cert if specified
@@ -232,7 +277,7 @@ func (c *client) Connect() error {
 		} else {
 			tlsConfig = &tls.Config{
 				Certificates: []tls.Certificate{cert},
-				MinVersion:   tls.VersionTLS13,
+				MinVersion:   c.conf.tlsMinVersion, // #nosec G402
 			}
 		}
 	}
@@ -257,7 +302,7 @@ func (c *client) Connect() error {
 	client := paho.NewClient(opts)
 	token := client.Connect()
 	rt := c.vu.Runtime()
-	if !token.WaitTimeout(time.Duration(c.conf.timeout) * time.Millisecond) {
+	if !token.WaitTimeout(time.Duration(timeoutValue) * time.Millisecond) {
 		common.Throw(rt, ErrTimeout)
 		return ErrTimeout
 	}
@@ -306,3 +351,10 @@ func (c *client) newErrorEvent(msg string) *sobek.Object {
 	must(o.DefineDataProperty("message", rt.ToValue(msg), sobek.FLAG_FALSE, sobek.FLAG_FALSE, sobek.FLAG_TRUE))
 	return o
 }
+
+func safeUintToInt64(u uint) (int64, error) {
+	if u > math.MaxInt64 {
+		return 0, errors.New("value too large to convert to int64")
+	}
+	return int64(u), nil
+}

errors.go

@@ -13,6 +13,8 @@ var (
 	ErrClient = errors.New("client is not connected")
 	// ErrTimeout operation timeout
 	ErrTimeout = errors.New("operation timeout")
+	// ErrTimeoutToLong timeout value is too large
+	ErrTimeoutToLong = errors.New("timeout value is too large")
 	// ErrSubscribe failed to subscribe to mqtt topic
 	ErrSubscribe = errors.New("subscribe failure")
 	// ErrConsumeToken consume token is invalid

publish.go

@@ -24,6 +24,15 @@ func (c *client) Publish(
 	if success == nil && failure == nil {
 		return c.publishSync(topic, qos, message, retain, timeout)
 	}
+
+	// check timeout value
+	timeoutValue, err := safeUintToInt64(timeout)
+	if err != nil {
+		rt := c.vu.Runtime()
+		common.Throw(rt, ErrTimeoutToLong)
+		return ErrTimeoutToLong
+	}
+
 	// async case
 	callback := c.vu.RegisterCallback()
 	go func() {
@@ -40,7 +49,7 @@ func (c *client) Publish(
 			return
 		}
 		token := c.pahoClient.Publish(topic, byte(qos), retain, message)
-		if !token.WaitTimeout(time.Duration(timeout) * time.Millisecond) {
+		if !token.WaitTimeout(time.Duration(timeoutValue) * time.Millisecond) {
 			callback(func() error {
 				ev := c.newErrorEvent("publish timeout")
 
@@ -94,19 +103,30 @@ func (c *client) publishSync(
 		common.Throw(rt, ErrClient)
 		return ErrClient
 	}
+
+	// check timeout value
+	timeoutValue, err := safeUintToInt64(timeout)
+	if err != nil {
+		rt := c.vu.Runtime()
+		common.Throw(rt, ErrTimeoutToLong)
+		return ErrTimeoutToLong
+	}
+
 	token := c.pahoClient.Publish(topic, byte(qos), retain, message)
 	// sync case
-	if !token.WaitTimeout(time.Duration(timeout) * time.Millisecond) {
+	if !token.WaitTimeout(time.Duration(timeoutValue) * time.Millisecond) {
 		rt := c.vu.Runtime()
 		common.Throw(rt, ErrTimeout)
 		return ErrTimeout
 	}
-	if err := token.Error(); err != nil {
+
+	if err = token.Error(); err != nil {
 		rt := c.vu.Runtime()
 		common.Throw(rt, ErrPublish)
 		return ErrPublish
 	}
-	err := c.publishMessageMetric(float64(len(message)))
+
+	err = c.publishMessageMetric(float64(len(message)))
 	if err != nil {
 		return err
 	}

subscribe.go

@@ -25,14 +25,22 @@ func (c *client) Subscribe(
 		common.Throw(rt, ErrClient)
 		return ErrClient
 	}
+
+	// check timeout value
+	timeoutValue, err := safeUintToInt64(timeout)
+	if err != nil {
+		common.Throw(rt, ErrTimeoutToLong)
+		return ErrTimeoutToLong
+	}
+
 	c.messageChan = make(chan paho.Message)
 	messageCB := func(_ paho.Client, msg paho.Message) {
 		go func(msg paho.Message) {
 			c.messageChan <- msg
 		}(msg)
 	}
 	token := c.pahoClient.Subscribe(topic, byte(qos), messageCB)
-	if !token.WaitTimeout(time.Duration(timeout) * time.Millisecond) {
+	if !token.WaitTimeout(time.Duration(timeoutValue) * time.Millisecond) {
 		common.Throw(rt, ErrTimeout)
 		return ErrTimeout
 	}
@@ -47,7 +55,7 @@ func (c *client) Subscribe(
 		}
 	}
 	c.tq = taskqueue.New(registerCallback)
-	go c.loop(c.messageChan, timeout)
+	go c.loop(c.messageChan, timeoutValue)
 	return nil
 }
 
@@ -77,7 +85,7 @@ func (c *client) receiveMessageMetric(msgLen float64) error {
 }
 
 //nolint:gocognit // todo improve this
-func (c *client) loop(messageChan <-chan paho.Message, timeout uint) {
+func (c *client) loop(messageChan <-chan paho.Message, timeout int64) {
 	ctx := c.vu.Context()
 	stop := make(chan struct{})
 	defer c.tq.Close()