Updated exserver logic for dynamic topic publishing and subscribing

components/extensions/emqx-extensions/io.entgra.device.mgt.plugins.emqx.exhook/src/main/java/io/entgra/device/mgt/plugins/emqx/exhook/ExServer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2018 - 2023, Entgra (Pvt) Ltd. (http://www.entgra.io) All Rights Reserved.
+ * Copyright (c) 2018 - 2026, Entgra (Pvt) Ltd. (http://www.entgra.io) All Rights Reserved.
  *
  * Entgra (Pvt) Ltd. licenses this file to you under the Apache License,
  * Version 2.0 (the "License"); you may not use this file except
@@ -61,6 +61,11 @@ public class ExServer {
     private Server server;
     private final ExServerUtilityService utilityService;
 
+    private static final String TOPIC_SEPARATOR = ":";
+    private static final String SINGLE_LEVEL_WILDCARD = "+";
+    private static final String MULTI_LEVEL_WILDCARD = "#";
+    private static final String MULTI_LEVEL_WILDCARD_SUFFIX = TOPIC_SEPARATOR + MULTI_LEVEL_WILDCARD;
+
     public ExServer(ExServerUtilityService utilityService) {
         this.utilityService = utilityService;
     }
@@ -399,7 +404,7 @@ public void onClientCheckAcl(ClientCheckAclRequest request, StreamObserver<Value
                                 .asRuntimeException();
                 }
 
-                boolean isAuthorized = scopeList.stream().anyMatch(scope -> scope.startsWith(tempScope));
+                boolean isAuthorized = scopeList.stream().anyMatch(scope -> isTopicMatch(scope, tempScope));
                 if (!isAuthorized) {
                     logger.warn("topic=" + topic + ", requiredScope=" + tempScope);
                     throw Status.PERMISSION_DENIED.withDescription("User not authorized for requested topic").asRuntimeException();
@@ -420,6 +425,72 @@ public void onClientCheckAcl(ClientCheckAclRequest request, StreamObserver<Value
             }
         }
 
+        private boolean isTopicMatch(String scopePattern, String exactTopic) {
+            if (scopePattern.equals(exactTopic)) {
+                return true;
+            }
+
+            String[] scopeParts = scopePattern.split(TOPIC_SEPARATOR);
+            String[] topicParts = exactTopic.split(TOPIC_SEPARATOR);
+
+            boolean scopeHasMulti = scopePattern.endsWith(MULTI_LEVEL_WILDCARD_SUFFIX);
+            boolean topicHasMulti = exactTopic.endsWith(MULTI_LEVEL_WILDCARD_SUFFIX);
+
+            if (topicHasMulti) {
+                return matchMultiLevelSubscription(scopeParts, topicParts);
+            }
+
+            if (scopeHasMulti) {
+                return matchMultiLevelScope(scopeParts, topicParts);
+            }
+
+            return matchSingleLevelWildcards(scopeParts, topicParts);
+        }
+
+        /**
+         * Handles cases where the incoming subscription request contains a '#'
+         */
+        private boolean matchMultiLevelSubscription(String[] scopeParts, String[] topicParts) {
+            if (topicParts.length > scopeParts.length || !MULTI_LEVEL_WILDCARD.equals(scopeParts[scopeParts.length - 1])) {
+                return false;
+            }
+            return compareParts(scopeParts, topicParts, topicParts.length - 1);
+        }
+
+        /**
+         * Handles cases where the defined scope contains a '#'
+         */
+        private boolean matchMultiLevelScope(String[] scopeParts, String[] topicParts) {
+            if (topicParts.length < scopeParts.length - 1) {
+                return false;
+            }
+            return compareParts(scopeParts, topicParts, scopeParts.length - 1);
+        }
+
+        /**
+         * Handles standard positional matching including '+' wildcards
+         */
+        private boolean matchSingleLevelWildcards(String[] scopeParts, String[] topicParts) {
+            if (scopeParts.length != topicParts.length) {
+                return false;
+            }
+            return compareParts(scopeParts, topicParts, scopeParts.length);
+        }
+
+        /**
+         * Helper to compare individual segments of the topic
+         */
+        private boolean compareParts(String[] scopeParts, String[] topicParts, int lengthToCompare) {
+            for (int i = 0; i < lengthToCompare; i++) {
+                String scope = scopeParts[i];
+                String topic = topicParts[i];
+                if (!SINGLE_LEVEL_WILDCARD.equals(scope) && !scope.equals(topic)) {
+                    return false;
+                }
+            }
+            return true;
+        }
+
         @Override
         public void onClientSubscribe(ClientSubscribeRequest request, StreamObserver<EmptySuccess> responseObserver) {
             DEBUG("onClientSubscribe", request);