Fix interception of outbound publish packets

The publish inbound interceptor usage was incorrect, the publish outbound interceptor ensures packet delivery can be prevented.

Author: djnewbould

hivemq-krb/src/main/java/uk/co/amrc/factoryplus/hivemq_auth_krb/FPKrbAuthorizer.java

@@ -84,7 +84,6 @@ public void authorizeSubscribe(
                     .subscribe(result -> {
                                 if (result) {
                                     log.info("Successfully authorized subscription client {} for topic {}.", clientUsername, topic);
-                                    provider.context.storeTopicMapping(clientUsername, topic);
                                     output.authorizeSuccessfully();
                                 } else {
                                     log.info("Subscription permission denied for user {} topic {}", clientUsername, topic);

hivemq-krb/src/main/java/uk/co/amrc/factoryplus/hivemq_auth_krb/FPKrbContext.java

@@ -16,7 +16,6 @@
 
 public class FPKrbContext {
     private final ConcurrentHashMap<String, String> sessionMap = new ConcurrentHashMap<>();
-    private final ConcurrentHashMap<String, Set<String>> topicToSubscribers = new ConcurrentHashMap<>();
     private static final @NotNull Logger log = LoggerFactory.getLogger(FPKrbMain.class);
     public  FPServiceClient fplus;
 
@@ -55,48 +54,6 @@ public void removeClientUserNameMapping(String clientId) {
         sessionMap.remove(clientId);
     }
 
-    /**
-     *
-     * @param clientId
-     * @param topic
-     */
-    public void storeTopicMapping(String clientId,String topic) {
-        topicToSubscribers
-                .computeIfAbsent(topic, k -> ConcurrentHashMap.newKeySet())
-                .add(clientId);
-    }
-
-    /**
-     *
-     * @param topic
-     * @return
-     */
-    public Set<String> getTopicMapping(String topic){
-        Set<String> subscribers = new HashSet<>();
-
-        topicToSubscribers.forEachKey(1, k-> {
-            if(matchesPermission(k, topic)){
-                // Exact match subscribers
-                Set<String> exactSubscribers = topicToSubscribers.get(k);
-                if (exactSubscribers != null) {
-                    subscribers.addAll(exactSubscribers);
-                }
-            }
-        });
-
-        return subscribers;
-    }
-
-    /**
-     *
-     * @param clientId
-     */
-    public void removeClientFromTopicMapping(String clientId){
-        for (Set<String> subscribers : topicToSubscribers.values()) {
-            subscribers.remove(clientId);
-        }
-    }
-
     private FPServiceClient startServiceClient ()
     {
         var fplus = new FPServiceClient();

hivemq-krb/src/main/java/uk/co/amrc/factoryplus/hivemq_auth_krb/interceptors/FPKrbClientInitializer.java

@@ -18,8 +18,8 @@ public FPKrbClientInitializer(FPKrbContext context) {
 
     @Override
     public void initialize(@NotNull InitializerInput initializerInput, @NotNull ClientContext clientContext) {
-        clientContext.addPublishInboundInterceptor(new FPKrbInboundPublishInterceptor(this));
         clientContext.addUnsubscribeInboundInterceptor(new FPKrbUnsubscribeInboundInterceptor(this));
         clientContext.addDisconnectInboundInterceptor(new FPKrbDisconnectInboundInterceptor(this));
+        clientContext.addPublishOutboundInterceptor(new FPKrbOutboundPublishInterceptor(this));
     }
 }

hivemq-krb/src/main/java/uk/co/amrc/factoryplus/hivemq_auth_krb/interceptors/FPKrbDisconnectInboundInterceptor.java

@@ -29,8 +29,7 @@ public void onInboundDisconnect(@NotNull DisconnectInboundInput disconnectInboun
         final CompletableFuture<?> taskFuture = Services.extensionExecutorService().submit(() -> {
             try {
                 var clientId = disconnectInboundInput.getClientInformation().getClientId();
-                initializer.context.removeClientUserNameMapping(initializer.context.getUsername(clientId));
-                initializer.context.removeClientFromTopicMapping(initializer.context.getUsername(clientId));
+                initializer.context.removeClientUserNameMapping(clientId);
             } catch (final Exception e) {
                 log.error("Disconnect inbound interception failed:", e);
             }

hivemq-krb/src/main/java/uk/co/amrc/factoryplus/hivemq_auth_krb/interceptors/FPKrbInboundPublishInterceptor.java

@@ -0,0 +1,54 @@
+package uk.co.amrc.factoryplus.hivemq_auth_krb.interceptors;
+
+import com.hivemq.extension.sdk.api.annotations.NotNull;
+import com.hivemq.extension.sdk.api.auth.parameter.TopicPermission;
+import com.hivemq.extension.sdk.api.interceptor.publish.PublishOutboundInterceptor;
+import com.hivemq.extension.sdk.api.interceptor.publish.parameter.PublishOutboundInput;
+import com.hivemq.extension.sdk.api.interceptor.publish.parameter.PublishOutboundOutput;
+import com.hivemq.extension.sdk.api.async.Async;
+import com.hivemq.extension.sdk.api.async.TimeoutFallback;
+import com.hivemq.extension.sdk.api.services.Services;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import uk.co.amrc.factoryplus.hivemq_auth_krb.FPKrbAuthenticator;
+
+import java.time.Duration;
+
+import static uk.co.amrc.factoryplus.hivemq_auth_krb.AuthUtils.getACLforPrincipal;
+import static uk.co.amrc.factoryplus.hivemq_auth_krb.AuthUtils.isPermissionAllowed;
+
+public class FPKrbOutboundPublishInterceptor implements PublishOutboundInterceptor {
+    private FPKrbClientInitializer initializer;
+    private static final @NotNull Logger log = LoggerFactory.getLogger(FPKrbAuthenticator.class);
+
+    public FPKrbOutboundPublishInterceptor(FPKrbClientInitializer initializer) {
+        this.initializer = initializer;
+    }
+
+    @Override
+    public void onOutboundPublish(@NotNull PublishOutboundInput input, @NotNull PublishOutboundOutput output) {
+        final Async<PublishOutboundOutput> async = output.async(Duration.ofSeconds(10), TimeoutFallback.FAILURE);
+        Services.extensionExecutorService().submit(() -> {
+            final String clientId = input.getClientInformation().getClientId();
+            final String username = initializer.context.getUsername(clientId);
+            final String topic = input.getPublishPacket().getTopic();
+            isPermissionAllowed(getACLforPrincipal(username, initializer.fplus), topic, TopicPermission.MqttActivity.SUBSCRIBE)
+                    .subscribe(result -> {
+                                if (result) {
+                                    log.info("Successfully intercepted outbound subscription client {} for topic {}.", username, topic);
+                                } else {
+                                    log.info("Subscription outbound intercepted! permission denied for user {} topic {}", username, topic);
+                                    output.preventPublishDelivery();
+                                    // Clean up.
+                                    initializer.context.removeClientUserNameMapping(clientId);
+                                    System.out.println("Client " + username + " was kicked due to policy violation.");
+                                }
+                                async.resume();
+                            },
+                            error ->  {
+                                log.error("Error occurred: {}", error.getMessage());
+                                async.resume();
+                            });
+        });
+    }
+}

hivemq-krb/src/main/java/uk/co/amrc/factoryplus/hivemq_auth_krb/interceptors/FPKrbOutboundPublishInterceptor.java

@@ -30,8 +30,7 @@ public void onInboundUnsubscribe(@NotNull UnsubscribeInboundInput unsubscribeInb
             try {
                 // Remove the client from our context storage.
                 log.info("Unsubscribe from {}", clientId);
-                initializer.context.removeClientUserNameMapping(initializer.context.getUsername(clientId));
-                initializer.context.removeClientFromTopicMapping(initializer.context.getUsername(clientId));
+                initializer.context.removeClientUserNameMapping(clientId);
             } catch (final Exception e) {
                 log.info("Unsubscribe inbound interception failed:", e);
             }