ARTEMIS-5914 Bypass link creation in remote federation when possible

Bypass link creation from the target broker back to the source when remote
federation is configured if the outcome is known to be a failure due to the
federation user not having access to send into the queue on the remote.

Author: tabish121

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/federation/AMQPFederation.java

@@ -537,6 +537,18 @@ synchronized void processRemoteQueueAdded(String addressName, String queueName)
       });
    }
 
+   /**
+    * {@return true if this is the federation source instance, false if it is the target}
+    */
+   abstract boolean isFederationSource();
+
+   /**
+    * {@return true if this is the federation target instance, false if it is the source}
+    */
+   final boolean isFederationTarget() {
+      return !isFederationSource();
+   }
+
    /**
     * Error signaling API that must be implemented by the specific federation implementation to handle error when
     * creating a federation resource such as an outgoing receiver link.

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/federation/AMQPFederationAddressPolicyManager.java

@@ -32,18 +32,21 @@
 
 import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
 import org.apache.activemq.artemis.api.core.ActiveMQException;
+import org.apache.activemq.artemis.api.core.ActiveMQSecurityException;
 import org.apache.activemq.artemis.api.core.RoutingType;
 import org.apache.activemq.artemis.api.core.SimpleString;
 import org.apache.activemq.artemis.core.filter.Filter;
 import org.apache.activemq.artemis.core.postoffice.Binding;
 import org.apache.activemq.artemis.core.postoffice.QueueBinding;
 import org.apache.activemq.artemis.core.postoffice.impl.DivertBinding;
+import org.apache.activemq.artemis.core.security.CheckType;
 import org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
 import org.apache.activemq.artemis.core.server.impl.AddressInfo;
 import org.apache.activemq.artemis.core.server.plugin.ActiveMQServerAddressPlugin;
 import org.apache.activemq.artemis.core.transaction.Transaction;
 import org.apache.activemq.artemis.protocol.amqp.federation.FederationConsumerInfo;
 import org.apache.activemq.artemis.protocol.amqp.federation.FederationReceiveFromAddressPolicy;
+import org.apache.activemq.artemis.protocol.amqp.logger.ActiveMQAMQPProtocolLogger;
 import org.apache.activemq.artemis.protocol.amqp.proton.AMQPSessionContext;
 import org.apache.activemq.artemis.utils.CompositeAddress;
 import org.slf4j.Logger;
@@ -251,6 +254,23 @@ private void checkBindingForMatch(Binding binding) {
                return;
             }
 
+            // Target brokers which have been sent remote federation policies might not have write
+            // access via the logged in user to the address with demand which we are attempting to
+            // federate messages to so instead of creating a receiver that will fail when the remote
+            // routes a message to it we can just omit creating the link in the first place.
+            if (federation.isFederationTarget()) {
+               try {
+                  session.getSessionSPI().check(addressInfo.getName(), CheckType.SEND, federation.getConnectionContext().getSecurityAuth());
+               } catch (ActiveMQSecurityException e) {
+                  ActiveMQAMQPProtocolLogger.LOGGER.federationTargetSkippedAddressFederation(
+                     addressInfo.getName().toString(), "User does not have send permission to configured address.");
+                  return;
+               } catch (Exception ex) {
+                  logger.warn("Caught unknown exception from security check on address:{} send permissions: cannot federate:", addressInfo.getName(), ex);
+                  return;
+               }
+            }
+
             createOrUpdateFederatedAddressConsumerForBinding(addressInfo, queueBinding);
          } else {
             reactIfQueueBindingMatchesAnyDivertTarget(queueBinding);
@@ -289,6 +309,23 @@ private void reactIfAnyQueueBindingMatchesDivertTarget(DivertBinding divertBindi
          return;
       }
 
+      // Target brokers which have been sent remote federation policies might not have write access
+      // via the logged in user to the address this divert is attached to which means we don't need
+      // to track this divert. Since we don't add the divert to the tracking map future demand on
+      // divert that would otherwise match the address includes won't trigger federation attempts.
+      if (federation.isFederationTarget()) {
+         try {
+            session.getSessionSPI().check(addressInfo.getName(), CheckType.SEND, federation.getConnectionContext().getSecurityAuth());
+         } catch (ActiveMQSecurityException e) {
+            ActiveMQAMQPProtocolLogger.LOGGER.federationTargetSkippedAddressFederation(
+               addressInfo.getName().toString(), "User does not have send permission to configured address.");
+            return;
+         } catch (Exception ex) {
+            logger.warn("Caught unknown exception from security check on address:{} send permissions: cannot federate:", addressInfo.getName(), ex);
+            return;
+         }
+      }
+
       // We only need to check if we've never seen the divert before, afterwards we will
       // be checking it any time a new QueueBinding is added instead.
       if (!divertsTracking.containsKey(divertBinding)) {
@@ -337,15 +374,15 @@ private void reactIfQueueBindingMatchesAnyDivertTarget(QueueBinding queueBinding
 
       final SimpleString queueAddress = queueBinding.getAddress();
 
-      divertsTracking.entrySet().forEach((e) -> {
-         final SimpleString forwardAddress = e.getKey().getDivert().getForwardAddress();
-         final DivertBinding divertBinding = e.getKey();
+      divertsTracking.entrySet().forEach((divert) -> {
+         final SimpleString forwardAddress = divert.getKey().getDivert().getForwardAddress();
+         final DivertBinding divertBinding = divert.getKey();
 
          // Check matched diverts to see if the QueueBinding address matches the address or
          // addresses (composite diverts) of the Divert and if so then we can check if we need
          // to create demand on the source address on the remote if we haven't done so already.
 
-         if (!e.getValue().contains(queueBinding) && isAddressInDivertForwards(queueAddress, forwardAddress)) {
+         if (!divert.getValue().contains(queueBinding) && isAddressInDivertForwards(queueAddress, forwardAddress)) {
             // The plugin can block the demand totally here either based on the divert itself
             // or the queue that's attached to the divert.
             if (isPluginBlockingFederationConsumerCreate(divertBinding.getDivert(), queueBinding.getQueue())) {
@@ -360,7 +397,7 @@ private void reactIfQueueBindingMatchesAnyDivertTarget(QueueBinding queueBinding
 
             // Each divert that forwards to the address the queue is bound to we add demand
             // in the diverts tracker.
-            e.getValue().add(queueBinding);
+            divert.getValue().add(queueBinding);
 
             final AddressInfo addressInfo = server.getPostOffice().getAddressInfo(divertBinding.getAddress());
 

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/federation/AMQPFederationQueuePolicyManager.java

@@ -28,9 +28,12 @@
 
 import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
 import org.apache.activemq.artemis.api.core.ActiveMQException;
+import org.apache.activemq.artemis.api.core.ActiveMQSecurityException;
+import org.apache.activemq.artemis.api.core.SimpleString;
 import org.apache.activemq.artemis.core.filter.Filter;
 import org.apache.activemq.artemis.core.postoffice.Binding;
 import org.apache.activemq.artemis.core.postoffice.QueueBinding;
+import org.apache.activemq.artemis.core.security.CheckType;
 import org.apache.activemq.artemis.core.server.Queue;
 import org.apache.activemq.artemis.core.server.ServerConsumer;
 import org.apache.activemq.artemis.core.server.ServerSession;
@@ -41,6 +44,7 @@
 import org.apache.activemq.artemis.protocol.amqp.federation.FederationReceiveFromQueuePolicy;
 import org.apache.activemq.artemis.protocol.amqp.proton.AMQPSessionContext;
 import org.apache.activemq.artemis.protocol.amqp.federation.FederationConsumerInfo.Role;
+import org.apache.activemq.artemis.protocol.amqp.logger.ActiveMQAMQPProtocolLogger;
 import org.apache.activemq.artemis.utils.CompositeAddress;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -164,8 +168,9 @@ private void checkQueueForMatch(Queue queue) {
 
    private void reactIfConsumerMatchesPolicy(ServerConsumer consumer) {
       final String queueName = consumer.getQueue().getName().toString();
+      final SimpleString addressName = consumer.getQueueAddress();
 
-      if (testIfQueueMatchesPolicy(consumer.getQueueAddress().toString(), queueName)) {
+      if (testIfQueueMatchesPolicy(addressName.toString(), queueName)) {
          final boolean federationConsumer = isFederationConsumer(consumer);
 
          // We should ignore federation consumers from remote peers but configuration does allow
@@ -175,6 +180,23 @@ private void reactIfConsumerMatchesPolicy(ServerConsumer consumer) {
             return;
          }
 
+         // Target brokers which have been sent remote federation policies might not have write
+         // access via the logged in user to the queue with demand which we are attempting to
+         // federate messages to so instead of creating a receiver that will fail when the remote
+         // routes a message to it we can just omit creating the link in the first place.
+         if (federation.isFederationTarget()) {
+            try {
+               session.getSessionSPI().check(addressName, consumer.getQueue().getName(), CheckType.SEND, federation.getConnectionContext().getSecurityAuth());
+            } catch (ActiveMQSecurityException e) {
+               ActiveMQAMQPProtocolLogger.LOGGER.federationTargetSkippedQueueFederation(
+                  queueName, "User does not have send permission to configured queue.");
+               return;
+            } catch (Exception ex) {
+               logger.warn("Caught unknown exception from security check on queue:{} send permissions: cannot federate:", queueName, ex);
+               return;
+            }
+         }
+
          logger.trace("Federation Policy matched on consumer for binding: {}", consumer.getBinding());
 
          final AMQPFederationQueueConsumerManager entry;

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/federation/AMQPFederationSource.java

@@ -294,6 +294,11 @@ protected void signalError(Exception cause) {
       brokerConnection.runtimeError(cause);
    }
 
+   @Override
+   boolean isFederationSource() {
+      return true;
+   }
+
    @Override
    void registerFederationManagement() throws Exception {
       AMQPFederationManagementSupport.registerFederationSource(this);

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/federation/AMQPFederationTarget.java

@@ -142,6 +142,11 @@ protected void signalError(Exception cause) {
       connection.close(new ErrorCondition(condition, description));
    }
 
+   @Override
+   boolean isFederationSource() {
+      return false;
+   }
+
    @Override
    void registerFederationManagement() throws Exception {
       if (brokerConnection.isManagable()) {

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/logger/ActiveMQAMQPProtocolLogger.java

@@ -82,4 +82,10 @@ public interface ActiveMQAMQPProtocolLogger {
 
    @LogMessage(id = 111012, value = "Acknowledgement retry failed for {} on address {}, queueID={}", level = LogMessage.Level.WARN)
    void ackRetryFailed(Object ackRetryInformation, Object address, long queueID);
+
+   @LogMessage(id = 111013, value = "AMQP Federation target skipped federation of address {}, reason={}", level = LogMessage.Level.WARN)
+   void federationTargetSkippedAddressFederation(String address, String reason);
+
+   @LogMessage(id = 111014, value = "AMQP Federation target skipped federation of queue {}, reason={}", level = LogMessage.Level.WARN)
+   void federationTargetSkippedQueueFederation(String address, String reason);
 }