[ADDED] Config options to disable MQTT QOS2 support

server/mqtt.go

@@ -345,6 +345,14 @@ type mqtt struct {
 	asm  *mqttAccountSessionManager // quick reference to account session manager, immutable after processConnect()
 	sess *mqttSession               // quick reference to session, immutable after processConnect()
 	cid  string                     // client ID
+
+	// rejectQoS2Pub tells the MQTT client to not accept QoS2 PUBLISH, instead
+	// error and terminate the connection.
+	rejectQoS2Pub bool
+
+	// downgradeQOS2Sub tells the MQTT client to downgrade QoS2 SUBSCRIBE
+	// requests to QoS1.
+	downgradeQoS2Sub bool
 }
 
 type mqttPending struct {
@@ -475,7 +483,11 @@ func (s *Server) createMQTTClient(conn net.Conn, ws *websocket) *client {
 	}
 	now := time.Now()
 
-	c := &client{srv: s, nc: conn, mpay: maxPay, msubs: maxSubs, start: now, last: now, mqtt: &mqtt{}, ws: ws}
+	mqtt := &mqtt{
+		rejectQoS2Pub:    opts.MQTT.rejectQoS2Pub,
+		downgradeQoS2Sub: opts.MQTT.downgradeQoS2Sub,
+	}
+	c := &client{srv: s, nc: conn, mpay: maxPay, msubs: maxSubs, start: now, last: now, mqtt: mqtt, ws: ws}
 	c.headers = true
 	c.mqtt.pp = &mqttPublish{}
 	// MQTT clients don't send NATS CONNECT protocols. So make it an "echo"
@@ -2226,6 +2238,10 @@ func (as *mqttAccountSessionManager) processSubs(sess *mqttSession, c *client,
 		if f.qos > 2 {
 			f.qos = 2
 		}
+		if c.mqtt.downgradeQoS2Sub && f.qos == 2 {
+			c.Warnf("Downgrading subscription QoS2 to QoS1 for %q, as configured", f.filter)
+			f.qos = 1
+		}
 		subject := f.filter
 		sid := subject
 
@@ -2329,6 +2345,10 @@ func (as *mqttAccountSessionManager) serializeRetainedMsgsForSub(sess *mqttSessi
 		if qos > sub.mqtt.qos {
 			qos = sub.mqtt.qos
 		}
+		if c.mqtt.rejectQoS2Pub && qos == 2 {
+			c.Warnf("Rejecting retained message with QoS2 for subscription %q, as configured", sub.subject)
+			continue
+		}
 		if qos > 0 {
 			pi = sess.trackPublishRetained()
 
@@ -3001,6 +3021,10 @@ func (c *client) mqttParseConnect(r *mqttReader, pl int, hasMappings bool) (byte
 		hasWill = true
 	}
 
+	if c.mqtt.rejectQoS2Pub && hasWill && wqos == 2 {
+		return 0, nil, fmt.Errorf("server does not accept QoS2 for Will messages")
+	}
+
 	// Spec [MQTT-3.1.2-19]
 	hasUser := cp.flags&mqttConnFlagUsernameFlag != 0
 	// Spec [MQTT-3.1.2-21]
@@ -3385,6 +3409,11 @@ func (c *client) mqttParsePub(r *mqttReader, pl int, pp *mqttPublish, hasMapping
 	if qos > 2 {
 		return fmt.Errorf("QoS=%v is invalid in MQTT", qos)
 	}
+
+	if c.mqtt.rejectQoS2Pub && qos == 2 {
+		return fmt.Errorf("QoS=2 is disabled for PUBLISH messages")
+	}
+
 	// Keep track of where we are when starting to read the variable header
 	start := r.pos
 

server/mqtt_test.go

@@ -592,6 +592,28 @@ func TestMQTTParseOptions(t *testing.T) {
 				}
 				return nil
 			}, ""},
+		{"reject_qos2_publish",
+			`
+			mqtt {
+				reject_qos2_publish: true
+			}
+			`, func(o *MQTTOpts) error {
+				if !o.rejectQoS2Pub {
+					return fmt.Errorf("Invalid: expected rejectQoS2Pub to be set")
+				}
+				return nil
+			}, ""},
+		{"downgrade_qos2_subscribe",
+			`
+			mqtt {
+				downgrade_qos2_subscribe: true
+			}
+			`, func(o *MQTTOpts) error {
+				if !o.downgradeQoS2Sub {
+					return fmt.Errorf("Invalid: expected downgradeQoS2Sub to be set")
+				}
+				return nil
+			}, ""},
 	} {
 		t.Run(test.name, func(t *testing.T) {
 			conf := createConfFile(t, []byte(test.content))
@@ -1928,6 +1950,27 @@ func TestMQTTSubAck(t *testing.T) {
 	testMQTTSub(t, 1, mc, r, subs, expected)
 }
 
+func TestMQTTQoS2SubDowngrade(t *testing.T) {
+	o := testMQTTDefaultOptions()
+	o.MQTT.downgradeQoS2Sub = true
+	s := testMQTTRunServer(t, o)
+	defer testMQTTShutdownServer(s)
+
+	mc, r := testMQTTConnect(t, &mqttConnInfo{cleanSess: true}, o.MQTT.Host, o.MQTT.Port)
+	defer mc.Close()
+	testMQTTCheckConnAck(t, r, mqttConnAckRCConnectionAccepted, false)
+
+	subs := []*mqttFilter{
+		{filter: "bar", qos: 1},
+		{filter: "baz", qos: 2},
+	}
+	expected := []byte{
+		1,
+		1,
+	}
+	testMQTTSub(t, 1, mc, r, subs, expected)
+}
+
 func testMQTTFlush(t testing.TB, c net.Conn, bw *bufio.Writer, r *mqttReader) {
 	t.Helper()
 	w := &mqttWriter{}
@@ -2160,6 +2203,28 @@ func TestMQTTPublish(t *testing.T) {
 	testMQTTPublish(t, mcp, mpr, 2, false, false, "foo", 2, []byte("msg"))
 }
 
+func TestMQTTQoS2PubReject(t *testing.T) {
+	o := testMQTTDefaultOptions()
+	o.MQTT.rejectQoS2Pub = true
+	s := testMQTTRunServer(t, o)
+	defer testMQTTShutdownServer(s)
+
+	nc := natsConnect(t, s.ClientURL())
+	defer nc.Close()
+
+	mcp, mpr := testMQTTConnect(t, &mqttConnInfo{cleanSess: true}, o.MQTT.Host, o.MQTT.Port)
+	defer mcp.Close()
+	testMQTTCheckConnAck(t, mpr, mqttConnAckRCConnectionAccepted, false)
+
+	testMQTTPublish(t, mcp, mpr, 1, false, false, "foo", 1, []byte("msg"))
+
+	testMQTTPublishNoAcks(t, mcp, 2, false, false, "foo", 2, []byte("msg"))
+	_, err := mpr.readByte("failed attempt")
+	if err == nil || !strings.Contains(err.Error(), "error reading failed attempt: EOF") {
+		t.Fatalf("Expected error about QoS 2 publish rejected, got %v", err)
+	}
+}
+
 func TestMQTTSub(t *testing.T) {
 	o := testMQTTDefaultOptions()
 	s := testMQTTRunServer(t, o)

server/opts.go

@@ -562,6 +562,14 @@ type MQTTOpts struct {
 
 	// Snapshot of configured TLS options.
 	tlsConfigOpts *TLSConfigOpts
+
+	// rejectQoS2Pub tells the MQTT client to not accept QoS2 PUBLISH, instead
+	// error and terminate the connection.
+	rejectQoS2Pub bool
+
+	// downgradeQOS2Sub tells the MQTT client to downgrade QoS2 SUBSCRIBE
+	// requests to QoS1.
+	downgradeQoS2Sub bool
 }
 
 type netResolver interface {
@@ -4631,6 +4639,11 @@ func parseMQTT(v interface{}, o *Options, errors *[]error, warnings *[]error) er
 		case "consumer_inactive_threshold", "consumer_auto_cleanup":
 			o.MQTT.ConsumerInactiveThreshold = parseDuration("consumer_inactive_threshold", tk, mv, errors, warnings)
 
+		case "reject_qos2_publish":
+			o.MQTT.rejectQoS2Pub, _ = mv.(bool)
+		case "downgrade_qos2_subscribe":
+			o.MQTT.downgradeQoS2Sub, _ = mv.(bool)
+
 		default:
 			if !tk.IsUsedVariable() {
 				err := &unknownConfigFieldErr{