Intercepting Publish on server does not return the correct status code to the client

[grendo]

I am trying to put an authorization functionality around publishing topics. The code below will stop the topic being published but the client gets a Success status instead of NotAuthorized

Is there anyway for the client to know the topic was not rejected by the server or is this a bug ?

Below is a test to show the issue

    [Test]
    public async Task TestCancelPublish()
    {
        var testTopic = "test/topic";
        var mqttFactory = new MqttFactory();
        var mqttServerOptions = new MqttServerOptionsBuilder().WithDefaultEndpoint().Build();
        using (var mqttServer = mqttFactory.CreateMqttServer(mqttServerOptions))
        {
            string clientId = Guid.NewGuid().ToString();
            await mqttServer.StartAsync();
            bool messageInterceptedOnServer = false;
            mqttServer.InterceptingPublishAsync += async (args) =>
            {
                // intercept server message and set reason code to NotAuthorized

                args.Response.ReasonCode = MqttPubAckReasonCode.NotAuthorized;
                args.Response.ReasonString = "Not Authorized";
                args.ProcessPublish = false;
                messageInterceptedOnServer = true;
            };

            using (var mqttClient = mqttFactory.CreateMqttClient())
            {
                var mqttClientOptions = new MqttClientOptionsBuilder()
                    .WithTcpServer("localhost")
                    .WithClientId(clientId)
                    .Build();
                await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);

                // publish the message
                var res = await mqttClient.PublishStringAsync(testTopic, "hello");
                while (messageInterceptedOnServer == false)
                {
                    await Task.Delay(300);
                }
                
                // This Fails
                // response status code says Success, I would expect NotAuthorized ???
                Assert.AreEqual(res.ReasonCode, MqttClientPublishReasonCode.NotAuthorized);
                await mqttClient.DisconnectAsync();
            }

            await mqttServer.StopAsync();
        }
    }

[chkr1011]

You must change a few things in order to get it running:

1. You must use protocol version 5 which can be set in the client options builder
2. Do not set ProcessPublish to false because then the message will basically treated as never sent
3. You must use QoS 1 so that the client will receive a response. QoS 0 will not receive and response from the server
4. When using protocol version 5 you must have a subscriber. Otherwise you will get a "NotMatchingSubscribers" response

[grendo]

I adjusted the unit test based on the guide above and it is now working

       [Test]
    public async Task TestCancelPublish()
    {
        var testTopic = "test/topic";
        var mqttFactory = new MqttFactory();
        var mqttServerOptions = new MqttServerOptionsBuilder()
            .WithDefaultEndpoint()
            .Build();
        using (var mqttServer = mqttFactory.CreateMqttServer(mqttServerOptions))
        {
            await mqttServer.StartAsync();
            bool messageInterceptedOnServer = false;
            
            mqttServer.InterceptingPublishAsync += async (args) =>
            {
                // intercept server message and set reason code to NotAuthorized
                args.Response.ReasonCode = MqttPubAckReasonCode.NotAuthorized;
                args.Response.ReasonString = "Not Authorized";
                messageInterceptedOnServer = true;
            };

            using (var client1 = mqttFactory.CreateMqttClient())
            {
                var client1Options = new MqttClientOptionsBuilder()
                    .WithTcpServer("localhost")
                    .WithClientId(Guid.NewGuid().ToString())
                    .WithProtocolVersion(MQTTnet.Formatter.MqttProtocolVersion.V500)
                    .Build();
                
                await client1.ConnectAsync(client1Options, CancellationToken.None);

                using(var client2 = mqttFactory.CreateMqttClient())
                {
                    var client2Options = new MqttClientOptionsBuilder()
                        .WithTcpServer("localhost")
                        .WithClientId(Guid.NewGuid().ToString())
                        .WithProtocolVersion(MQTTnet.Formatter.MqttProtocolVersion.V500)
                        .Build();
                    await client2.ConnectAsync(client2Options, CancellationToken.None);
                    await client2.SubscribeAsync(testTopic,MqttQualityOfServiceLevel.AtLeastOnce);

                    // publish the message
                    var res = await client1.PublishStringAsync(testTopic, "hello",MqttQualityOfServiceLevel.AtLeastOnce);

                    while (messageInterceptedOnServer == false)
                    {
                        await Task.Delay(300);
                    }

                    await client2.DisconnectAsync();
                    // Now get not authorized
                    Assert.AreEqual(res.ReasonCode, MqttClientPublishReasonCode.NotAuthorized);

                }
                await client1.DisconnectAsync();
            }

            await mqttServer.StopAsync();
        }
    }

It might not be Mqtt standard but only sending NoMatchingSubscribers if you are authorized in the MqttClientSessionsManager class would be a cleaner solution for me. If you have retained messages on I think it is ok that you do not have any subscribers.

eg
if (matchingSubscribersCount == 0 && reasonCode != (int)MqttPubAckReasonCode.NotAuthorized)
{
reasonCode = (int)MqttPubAckReasonCode.NoMatchingSubscribers;
await FireApplicationMessageNotConsumedEvent(applicationMessage, senderId).ConfigureAwait(false);
}

[chkr1011]

That is a good idea. But would check for success instead. So that it will never return NoMatchingSubscribers only if it is basically a success and no error etc.