Repository: mochi-mqtt/server

Issue:
When a MQTTv311 client tries to publish with qos2, and fails OnACLCheck, the client hangs forever and the connection stays open. Should we disconnect the client instead?
http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc398718042

> If a Server implementation does not authorize a PUBLISH to be performed by a Client; it has no way of informing that Client. It MUST either make a positive acknowledgement, according to the normal QoS rules, or close the Network Connection [MQTT-3.3.5-2].

I encountered this when testing my implementation and testing a `publish qos2` which `succesfully Connects (authenticates), but fails OnACLCheck`, in my case using Python Paho client 3.1.1, the publish fails but the client hangs forever on the publish. 

When I looked into the broker code it appears on failure of ACL check that it returns `nil` with no handling. 

Please let me know if I am missing anything or misinformed! 

Discussion:
@SimonMacIntyre: This case is the same using client of MQTTv5 too@mochi-co: I believe you are correct @SimonMacIntyre, this is a bug, at least for MQTTv3. If I remember, for v5 there is no provision to disconnect on ACL errors as some servers may wish to obscure their restrictions. I may be wrong though.@SimonMacIntyre: I'll take a quick peek! I see these in the v5 spec:
> The Client or Server sending the PUBACK packet MUST use one of the PUBACK Reason Codes [MQTT-3.4.2-1]
>The Client or Server sending the PUBREC packet MUST use one of the PUBREC Reason Code values. [MQTT-3.5.2-1]

Same  PUBACK reason code for both would be:
>135 | 0x87 | Not authorized | The PUBLISH is not authorized.

For qos0, it appears to be handled correctly!  (In the sense that doing nothing _is_ not hanging the client and the client thinks its 'published', of course thats the nature of qos0)





@dgduncan: So from reading that it seems like the spec for v3.1.1 should be to either ack and not publish or disconnect. A disconnect seems like the preferred solution as it makes it obvious there is a breach in the server rules; however, I can absolutely imagine a world in which a user of the server may not have control over some functionality of their MQTT client because it may be third part and a disconnection may make the server useless to them. Perhaps a capability could be in order to adjust functionality but that may just add unwanted complexity@SimonMacIntyre: Personally I lean towards the disconnect, over the ack and not publish. 

I do like the idea of in the future making that configurable in the file configuration, but maybe the disconnect is a reasonable path forward for now and if it becomes a problem for users we can explore making the server configurable on how to handle that only for v3.1.1?

@thedevop: Created a PR #292 to address this.

The server will:

| | QOS 0 | QOS 1 | QOS 2 |
|-|:-:|:-:|:-:|
|MQTT v3| Ignore | Disconnect | Disconnect
|MQTT v3.1.1| Ignore | Disconnect | Disconnect
|MQTT v5| Ignore | Puback NotAuthorized | Pubrec NotAuthorized@mochi-co: This has been fixed and released in [v2.4.0](https://github.com/mochi-mqtt/server/releases/tag/v2.4.0). Thank you @thedevop!@SimonMacIntyre: While working on https://github.com/mochi-mqtt/server/issues/465, I noticed this code in receivePacket: https://github.com/SimonMacIntyre/mochi-server/blob/0e001765aa7840e7585f6ff5ac5edaf21bea0506/server.go#L524-L524

Given the discussion in this thread and the table showing how we forcefully disconnect `v3` clients when QoS1 or QoS2 messages fail ACL checks, but forego disconnecting v5 clients, it made me wonder: doesn't the current code in receivePacket mean we would disconnect a v5 client if any error with code >= 0x80 bubbles up? I suspect this might have been intended to handle [MQTT-3.2.2-7], but that specification only applies to CONNACK packets. Furthermore, doesn't it conflict philosophically with the result of this thread and PR which was merged?

Currently, we handle custom ACK requirements and packet processing no-ops in the various `processPublish, processX` family of functions, which still makes sense. However, I'm wondering if we should remove the s.disconnectClient code from receivePacket for two reasons:

*  It shouldn't disconnect v5 clients simply because of error codes, since v5 supports graceful error handling (as seen in the graceful errors returned for v5 QoS1/2 failures), what spec might this code be adhering to? Let me know if I am missing it or misunderstanding entirely.
*  If any disconnect logic is needed, it should probably be handled in the validateConnect logic instead and not as a catch-all in `receivePacket`, maybe?

Can someone provide a gut check on this? I'm still building my understanding of how the server architecture fits together, so please bear with me.

I can remove it without adding anything elsewhere if that's the change we need. 

Solution patch:
diff --git a/packets/tpackets.go b/packets/tpackets.go
index 9e44c12..1ad1255 100644
--- a/packets/tpackets.go
+++ b/packets/tpackets.go
@@ -131,12 +131,14 @@ const (
 	TPublishSpecDenySysTopic
 	TPuback
 	TPubackMqtt5
+	TPubackMqtt5NotAuthorized
 	TPubackMalPacketID
 	TPubackMalProperties
 	TPubackUnexpectedError
 	TPubrec
 	TPubrecMqtt5
 	TPubrecMqtt5IDInUse
+	TPubrecMqtt5NotAuthorized
 	TPubrecMalPacketID
 	TPubrecMalProperties
 	TPubrecMalReasonCode
@@ -1707,41 +1709,41 @@ var TPacketData = map[byte]TPacketCases{
 			},
 		},
 		{
-            Case:    TPublishQos1Mqtt5,
-            Desc:    "mqtt v5",
-            Primary: true,
-            RawBytes: []byte{
-                Publish<<4 | 1<<1, 37, // Fixed header
-                0, 5, // Topic Name - LSB+MSB
-                'a', '/', 'b', '/', 'c', // Topic Name
-                0, 7, // Packet ID - LSB+MSB
-                // Properties
-                16, // length
-                38, // User Properties (38)
-                0, 5, 'h', 'e', 'l', 'l', 'o',
-                0, 6, 228, 184, 150, 231, 149, 140,
-                'h', 'e', 'l', 'l', 'o', ' ', 'm', 'o', 'c', 'h', 'i', // Payload
-            },
-            Packet: &Packet{
-                ProtocolVersion: 5,
-                FixedHeader: FixedHeader{
-                    Type:      Publish,
-                    Remaining: 37,
-                    Qos:       1,
-                },
-                PacketID:  7,
-                TopicName: "a/b/c",
-                Properties: Properties{
-                    User: []UserProperty{
-                        {
-                            Key: "hello",
-                            Val: "世界",
-                        },
-                    },
-                },
-                Payload: []byte("hello mochi"),
-            },
-        },
+			Case:    TPublishQos1Mqtt5,
+			Desc:    "mqtt v5",
+			Primary: true,
+			RawBytes: []byte{
+				Publish<<4 | 1<<1, 37, // Fixed header
+				0, 5, // Topic Name - LSB+MSB
+				'a', '/', 'b', '/', 'c', // Topic Name
+				0, 7, // Packet ID - LSB+MSB
+				// Properties
+				16, // length
+				38, // User Properties (38)
+				0, 5, 'h', 'e', 'l', 'l', 'o',
+				0, 6, 228, 184, 150, 231, 149, 140,
+				'h', 'e', 'l', 'l', 'o', ' ', 'm', 'o', 'c', 'h', 'i', // Payload
+			},
+			Packet: &Packet{
+				ProtocolVersion: 5,
+				FixedHeader: FixedHeader{
+					Type:      Publish,
+					Remaining: 37,
+					Qos:       1,
+				},
+				PacketID:  7,
+				TopicName: "a/b/c",
+				Properties: Properties{
+					User: []UserProperty{
+						{
+							Key: "hello",
+							Val: "世界",
+						},
+					},
+				},
+				Payload: []byte("hello mochi"),
+			},
+		},
 
 		{
 			Case:    TPublishQos1Dup,
@@ -2274,6 +2276,40 @@ var TPacketData = map[byte]TPacketCases{
 				},
 			},
 		},
+		{
+			Case:    TPubackMqtt5NotAuthorized,
+			Desc:    "QOS 1 publish not authorized mqtt5",
+			Primary: true,
+			RawBytes: []byte{
+				Puback << 4, 37, // Fixed header
+				0, 7, // Packet ID - LSB+MSB
+				ErrNotAuthorized.Code, // Reason Code
+				33,                    // Properties Length
+				31, 0, 14, 'n', 'o', 't', ' ', 'a', 'u',
+				't', 'h', 'o', 'r', 'i', 'z', 'e', 'd', // Reason String (31)
+				38, // User Properties (38)
+				0, 5, 'h', 'e', 'l', 'l', 'o',
+				0, 6, 228, 184, 150, 231, 149, 140,
+			},
+			Packet: &Packet{
+				ProtocolVersion: 5,
+				FixedHeader: FixedHeader{
+					Type:      Puback,
+					Remaining: 31,
+				},
+				PacketID:   7,
+				ReasonCode: ErrNotAuthorized.Code,
+				Properties: Properties{
+					ReasonString: ErrNotAuthorized.Reason,
+					User: []UserProperty{
+						{
+							Key: "hello",
+							Val: "世界",
+						},
+					},
+				},
+			},
+		},
 		{
 			Case:  TPubackUnexpectedError,
 			Desc:  "unexpected error",
@@ -2412,6 +2448,40 @@ var TPacketData = map[byte]TPacketCases{
 				},
 			},
 		},
+		{
+			Case:    TPubrecMqtt5NotAuthorized,
+			Desc:    "QOS 2 publish not authorized mqtt5",
+			Primary: true,
+			RawBytes: []byte{
+				Pubrec << 4, 37, // Fixed header
+				0, 7, // Packet ID - LSB+MSB
+				ErrNotAuthorized.Code, // Reason Code
+				33,                    // Properties Length
+				31, 0, 14, 'n', 'o', 't', ' ', 'a', 'u',
+				't', 'h', 'o', 'r', 'i', 'z', 'e', 'd', // Reason String (31)
+				38, // User Properties (38)
+				0, 5, 'h', 'e', 'l', 'l', 'o',
+				0, 6, 228, 184, 150, 231, 149, 140,
+			},
+			Packet: &Packet{
+				ProtocolVersion: 5,
+				FixedHeader: FixedHeader{
+					Type:      Pubrec,
+					Remaining: 31,
+				},
+				PacketID:   7,
+				ReasonCode: ErrNotAuthorized.Code,
+				Properties: Properties{
+					ReasonString: ErrNotAuthorized.Reason,
+					User: []UserProperty{
+						{
+							Key: "hello",
+							Val: "世界",
+						},
+					},
+				},
+			},
+		},
 		{
 			Case:      TPubrecMalReasonCode,
 			Desc:      "malformed reason code",
diff --git a/server.go b/server.go
index 19e0829..c9474ba 100644
--- a/server.go
+++ b/server.go
@@ -753,7 +753,21 @@ func (s *Server) processPublish(cl *Client, pk packets.Packet) error {
 	}
 
 	if !cl.Net.Inline && !s.hooks.OnACLCheck(cl, pk.TopicName, true) {
-		return nil
+		if pk.FixedHeader.Qos == 0 {
+			return nil
+		}
+
+		if cl.Properties.ProtocolVersion != 5 {
+			return s.DisconnectClient(cl, packets.ErrNotAuthorized)
+		}
+
+		if pk.FixedHeader.Qos == 1 {
+			ack := s.buildAck(pk.PacketID, packets.Puback, 0, pk.Properties, packets.ErrNotAuthorized)
+			return cl.WritePacket(ack)
+		}
+
+		ack := s.buildAck(pk.PacketID, packets.Pubrec, 0, pk.Properties, packets.ErrNotAuthorized)
+		return cl.WritePacket(ack)
 	}
 
 	pk.Origin = cl.ID



Test patch:
diff --git a/server_test.go b/server_test.go
index 4930e8a..64b110f 100644
--- a/server_test.go
+++ b/server_test.go
@@ -12,6 +12,7 @@ import (
 	"net"
 	"os"
 	"strconv"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -53,6 +54,26 @@ func (h *AllowHook) Provides(b byte) bool {
 func (h *AllowHook) OnConnectAuthenticate(cl *Client, pk packets.Packet) bool { return true }
 func (h *AllowHook) OnACLCheck(cl *Client, topic string, write bool) bool     { return true }
 
+type DenyHook struct {
+	HookBase
+}
+
+func (h *DenyHook) SetOpts(l *slog.Logger, opts *HookOptions) {
+	h.Log = l
+	h.Opts = opts
+}
+
+func (h *DenyHook) ID() string {
+	return "deny-all-auth"
+}
+
+func (h *DenyHook) Provides(b byte) bool {
+	return bytes.Contains([]byte{OnConnectAuthenticate, OnACLCheck}, []byte{b})
+}
+
+func (h *DenyHook) OnConnectAuthenticate(cl *Client, pk packets.Packet) bool { return false }
+func (h *DenyHook) OnACLCheck(cl *Client, topic string, write bool) bool     { return false }
+
 type DelayHook struct {
 	HookBase
 	DisconnectDelay time.Duration
@@ -1421,14 +1442,65 @@ func TestServerProcessPublishInvalidTopic(t *testing.T) {
 }
 
 func TestServerProcessPublishACLCheckDeny(t *testing.T) {
-	s := New(&Options{
-		Logger: logger,
-	})
-	s.Serve()
-	defer s.Close()
-	cl, _, _ := newTestClient()
-	err := s.processPublish(cl, *packets.TPacketData[packets.Publish].Get(packets.TPublishBasic).Packet)
-	require.NoError(t, err) // ACL check fails silently
+	tests := []struct {
+		name             string
+		protocolVersion  byte
+		pk               packets.Packet
+		expectErr        error
+		expectReponse    []byte
+		expectDisconnect bool
+	}{
+		{"v4_QOS0", 4, *packets.TPacketData[packets.Publish].Get(packets.TPublishBasic).Packet,
+			nil, nil, false},
+		{"v4_QOS1", 4, *packets.TPacketData[packets.Publish].Get(packets.TPublishQos1).Packet,
+			packets.ErrNotAuthorized, nil, true},
+		{"v4_QOS2", 4, *packets.TPacketData[packets.Publish].Get(packets.TPublishQos2).Packet,
+			packets.ErrNotAuthorized, nil, true},
+
+		{"v5_QOS0", 5, *packets.TPacketData[packets.Publish].Get(packets.TPublishBasicMqtt5).Packet,
+			nil, nil, false},
+		{"v5_QOS1", 5, *packets.TPacketData[packets.Publish].Get(packets.TPublishQos1Mqtt5).Packet,
+			nil, packets.TPacketData[packets.Puback].Get(packets.TPubrecMqtt5NotAuthorized).RawBytes, false},
+		{"v5_QOS2", 5, *packets.TPacketData[packets.Publish].Get(packets.TPublishQos2Mqtt5).Packet,
+			nil, packets.TPacketData[packets.Pubrec].Get(packets.TPubrecMqtt5NotAuthorized).RawBytes, false},
+	}
+
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			cc := *DefaultServerCapabilities
+			s := New(&Options{
+				Logger:       logger,
+				Capabilities: &cc,
+			})
+			s.AddHook(new(DenyHook), nil)
+			s.Serve()
+			defer s.Close()
+
+			cl, r, w := newTestClient()
+			cl.Properties.ProtocolVersion = tt.protocolVersion
+			s.Clients.Add(cl)
+
+			wg := sync.WaitGroup{}
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				err := s.processPublish(cl, tt.pk)
+				require.ErrorIs(t, err, tt.expectErr)
+				w.Close()
+			}()
+
+			buf, err := io.ReadAll(r)
+			require.NoError(t, err)
+
+			if tt.expectReponse != nil {
+				require.Equal(t, tt.expectReponse, buf)
+			}
+
+			require.Equal(t, tt.expectDisconnect, cl.Closed())
+			wg.Wait()
+		})
+	}
 }
 
 func TestServerProcessPublishOnMessageRecvRejected(t *testing.T) {