Feature: Support for authorizing clientId values based on userId

activemq-broker/src/main/java/org/apache/activemq/broker/Connector.java

@@ -71,6 +71,12 @@ public interface Connector extends Service {
     @Deprecated(forRemoval = true)
     int connectionCount();
 
+    /**
+     * @return true if connections accepted by this connector are protected by
+     * SSL/TLS, that is the transport negotiates TLS with each client
+     */
+    boolean isSsl();
+
     /**
      * If enabled, older connections with the same clientID are stopped
      *

activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnection.java

@@ -82,6 +82,7 @@
 import org.apache.activemq.network.NetworkBridgeFactory;
 import org.apache.activemq.network.NetworkConnector;
 import org.apache.activemq.security.MessageAuthorizationPolicy;
+import org.apache.activemq.security.SecurityContext;
 import org.apache.activemq.state.CommandVisitor;
 import org.apache.activemq.state.ConnectionState;
 import org.apache.activemq.state.ConsumerState;
@@ -1222,9 +1223,9 @@ public void stopAsync() {
         if (stopping.compareAndSet(false, true)) {
             // Let all the connection contexts know we are shutting down
             // so that in progress operations can notice and unblock.
-            List<TransportConnectionState> connectionStates = listConnectionStates();
-            for (TransportConnectionState cs : connectionStates) {
-                ConnectionContext connectionContext = cs.getContext();
+            var connectionStates = listConnectionStates();
+            for (var cs : connectionStates) {
+                var connectionContext = cs.getContext();
                 if (connectionContext != null) {
                     connectionContext.getStopping().set(true);
                 }
@@ -1298,8 +1299,8 @@ protected void doStop() throws Exception {
         // Remove all logical connection associated with this connection
         // from the broker.
         if (!broker.isStopped()) {
-            List<TransportConnectionState> connectionStates = listConnectionStates();
-            for (TransportConnectionState cs : connectionStates) {
+            var connectionStates = listConnectionStates();
+            for (var cs : connectionStates) {
                 cs.getContext().getStopping().set(true);
                 try {
                     LOG.debug("Cleaning up connection resources: {}", getRemoteAddress());
@@ -1480,9 +1481,26 @@ public Response processBrokerInfo(BrokerInfo info) throws IOException {
             setDuplexNetworkConnectorId(duplexNetworkConnectorId);
         }
 
+        // A connection that authenticated as an ordinary client and only now declares
+        // itself a network connection must have been permitted to do so. Bridges normally
+        // send BrokerInfo first, in which case authentication already saw the flag and
+        // there are no connection states here yet.
+        var connectionStates = listConnectionStates();
+        for (var cs : connectionStates) {
+            var securityContext = cs.getContext().getSecurityContext();
+            if (securityContext == null || !securityContext.isNetworkConnectionAuthorizationRequired()) {
+                LOG.debug("Network connection authorization not configured for {}", getRemoteAddress());
+                continue;
+            }
+            if (!securityContext.isNetworkConnectionAllowed()) {
+                LOG.warn("Rejecting network connection from {}: user {} is not allowed to register a network connection",
+                        getRemoteAddress(), securityContext.getUserName());
+                throw new IOException("User " + securityContext.getUserName() + " is not allowed to register a network connection");
+            }
+        }
+
         this.brokerInfo = info;
         networkConnection = true;
-        List<TransportConnectionState> connectionStates = listConnectionStates();
         for (TransportConnectionState cs : connectionStates) {
             cs.getContext().setNetworkConnection(true);
         }

activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnector.java

@@ -22,6 +22,7 @@
 import java.util.ArrayList;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.Locale;
 import java.util.Optional;
 import java.util.StringTokenizer;
 import java.util.concurrent.CopyOnWriteArrayList;
@@ -607,7 +608,7 @@
 
     @Deprecated(forRemoval = true)
     @Override
     public int connectionCount() {
         return connections.size();
     }
 
@@ -693,6 +694,28 @@
         return (server != null ? server.getMaxConnectionExceededCount() : 0l);
     }
 
+    /**
+     * Once the transport server is bound its own answer is authoritative. Before
+     * that the configured scheme decides, so the flag is also usable while a broker
+     * is still being configured: ssl, nio+ssl, auto+nio+ssl, mqtt+ssl and so on,
+     * plus https and wss.
+     */
+    @Override
+    public boolean isSsl() {
+        if (server != null) {
+            return server.isSslServer();
+        }
+        if (uri == null || uri.getScheme() == null) {
+            return false;
+        }
+        for (String part : uri.getScheme().toLowerCase(Locale.ROOT).split("\\+")) {
+            if ("ssl".equals(part) || "https".equals(part) || "wss".equals(part)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
     @Override
     public boolean isStarted() {
         return started.get();

activemq-broker/src/main/java/org/apache/activemq/broker/jmx/ConnectorView.java

@@ -86,12 +86,17 @@
         return connector.getStatistics().isEnabled();
     }
 
+    @Override
+    public boolean isSsl() {
+        return connector.isSsl();
+    }
+
     /**
      * Returns the number of current connections
      */
     @Override
     public int connectionCount() {
         return connector.connectionCount();
     }
 
     /**
@@ -227,7 +232,7 @@
      */
     @Override
     public int getConnectionCount() {
         return this.connector.connectionCount();
     }
 
     @Override

activemq-broker/src/main/java/org/apache/activemq/broker/jmx/ConnectorViewMBean.java

@@ -54,6 +54,9 @@ public interface ConnectorViewMBean extends Service {
     @MBeanInfo("Statistics gathering enabled")
     boolean isStatisticsEnabled();
 
+    @MBeanInfo("Connections are protected by SSL/TLS")
+    boolean isSsl();
+
     /**
      * Returns true if link stealing is enabled on this Connector
      *

activemq-broker/src/main/java/org/apache/activemq/security/JaasAuthenticationBroker.java

@@ -27,6 +27,10 @@
 import org.apache.activemq.broker.ConnectionContext;
 import org.apache.activemq.command.ConnectionInfo;
 import org.apache.activemq.jaas.JassCredentialCallbackHandler;
+import org.apache.activemq.broker.Connection;
+import org.apache.activemq.broker.Connector;
+import org.apache.activemq.jaas.ConnectionCallback;
+import org.apache.activemq.jaas.ConnectionPrincipal;
 
 /**
  * Logs a user in using JAAS.
@@ -55,6 +59,25 @@ public JaasSecurityContext(String userName, Subject subject) {
         public Set<Principal> getPrincipals() {
             return subject.getPrincipals();
         }
+
+        /**
+         * The login module records its decision on the ConnectionPrincipal when
+         * clientId authorization is configured; with no ConnectionPrincipal the
+         * feature is off and the connection is not restricted.
+         */
+        @Override
+        public boolean isNetworkConnectionAllowed() {
+            for (ConnectionPrincipal connection : subject.getPrincipals(ConnectionPrincipal.class)) {
+                return connection.isNetworkConnection();
+            }
+            return true;
+        }
+
+        /** a ConnectionPrincipal is only added when clientId authorization is configured */
+        @Override
+        public boolean isNetworkConnectionAuthorizationRequired() {
+            return !subject.getPrincipals(ConnectionPrincipal.class).isEmpty();
+        }
     }
 
     @Override
@@ -65,7 +88,7 @@ public void addConnection(ConnectionContext context, ConnectionInfo info) throws
             Thread.currentThread().setContextClassLoader(JaasAuthenticationBroker.class.getClassLoader());
             SecurityContext securityContext = null;
             try {
-                securityContext = authenticate(info.getUserName(), info.getPassword(), null);
+                securityContext = authenticate(info.getUserName(), info.getPassword(), connectionCallbackFrom(context, info));
                 context.setSecurityContext(securityContext);
                 securityContexts.add(securityContext);
                 super.addConnection(context, info);
@@ -83,14 +106,48 @@ public void addConnection(ConnectionContext context, ConnectionInfo info) throws
         }
     }
 
+    private ConnectionCallback connectionCallbackFrom(ConnectionContext context, ConnectionInfo info) {
+        var callback = new ConnectionCallback();
+        callback.setConnectionId(info.getConnectionId() != null ? info.getConnectionId().getValue() : null);
+        callback.setClientId(info.getClientId());
+        callback.setBrokerName(getBrokerName());
+        callback.setNetworkConnection(context.isNetworkConnection());
+        // the transport's view of the peer; ConnectionInfo.clientIp is client supplied
+        var connection = context.getConnection();
+        if (connection != null) {
+            callback.setRemoteAddress(connection.getRemoteAddress());
+        }
+        if (info.getTransportContext() instanceof X509Certificate[]) {
+            callback.setCertificates((X509Certificate[]) info.getTransportContext());
+        }
+        var connector = context.getConnector();
+        if (connector != null) {
+            callback.setSsl(connector.isSsl());
+            callback.setTransportConnectorName(connector.getName());
+        }
+        return callback;
+    }
+
     @Override
     public SecurityContext authenticate(String username, String password, X509Certificate[] certificates) throws SecurityException {
+        var connection = new ConnectionCallback();
+        connection.setCertificates(certificates);
+        return authenticate(username, password, connection);
+    }
+
+    /**
+     * Authenticates a connection, also handing the login module a description of
+     * the connection (id, clientId, broker name, network declaration, SSL, remote
+     * address, transport connector, client certificates) so clientId and network
+     * connection authorization can be applied.
+     */
+    public SecurityContext authenticate(String username, String password, ConnectionCallback connection) throws SecurityException {
         SecurityContext result = null;
-        JassCredentialCallbackHandler callback = new JassCredentialCallbackHandler(username, password);
+        var callback = new JassCredentialCallbackHandler(username, password, connection);
         try {
-            LoginContext lc = new LoginContext(jassConfiguration, callback);
+            var lc = new LoginContext(jassConfiguration, callback);
             lc.login();
-            Subject subject = lc.getSubject();
+            var subject = lc.getSubject();
 
             result = new JaasSecurityContext(username, subject);
         } catch (Exception ex) {

activemq-broker/src/main/java/org/apache/activemq/security/JaasDualAuthenticationBroker.java

@@ -24,7 +24,6 @@
 import org.apache.activemq.broker.ConnectionContext;
 import org.apache.activemq.broker.Connector;
 import org.apache.activemq.broker.EmptyBroker;
-import org.apache.activemq.broker.TransportConnector;
 import org.apache.activemq.command.ActiveMQDestination;
 import org.apache.activemq.command.ConnectionInfo;
 
@@ -111,12 +110,8 @@ public void removeConnection(ConnectionContext context, ConnectionInfo info, Thr
     }
 
     protected boolean isSSL(ConnectionContext context, ConnectionInfo info) throws Exception {
-        boolean sslCapable = false;
         Connector connector = context.getConnector();
-        if (connector instanceof TransportConnector) {
-            TransportConnector transportConnector = (TransportConnector) connector;
-            sslCapable = transportConnector.getServer().isSslServer();
-        }
+        boolean sslCapable = connector != null && connector.isSsl();
         // AMQ-5943, also check if transport context carries X509 cert
         if (!sslCapable && info.getTransportContext() instanceof X509Certificate[]) {
             sslCapable = true;

activemq-broker/src/main/java/org/apache/activemq/security/SecurityContext.java

@@ -83,6 +83,25 @@ public boolean isInOneOf(Set<?> allowedPrincipals) {
 
     public abstract Set<Principal> getPrincipals();
 
+    /**
+     * Whether the authenticated user may register this connection as a network
+     * connection. Consulted when a connection identifies itself as a network
+     * bridge after it has already been authenticated. Defaults to allowed so
+     * authentication plugins that do not make the distinction are unaffected.
+     */
+    public boolean isNetworkConnectionAllowed() {
+        return true;
+    }
+
+    /**
+     * Whether the authenticating plugin made a network connection decision for this
+     * connection. When false the broker applies no network connection restriction and
+     * {@link #isNetworkConnectionAllowed()} is not consulted.
+     */
+    public boolean isNetworkConnectionAuthorizationRequired() {
+        return false;
+    }
+
     public boolean contains(Object principal) {
         Set<Principal> principals = getPrincipals();
         return principals != null && principals.contains(principal);

activemq-http/src/main/java/org/apache/activemq/transport/https/HttpsTransportServer.java

@@ -24,6 +24,11 @@
 
 public class HttpsTransportServer extends HttpTransportServer {
 
+    @Override
+    public boolean isSslServer() {
+        return true;
+    }
+
     public HttpsTransportServer(URI uri, HttpsTransportFactory factory, SslContext context) {
         super(uri, factory);
         this.socketConnectorFactory = new SecureSocketConnectorFactory(context);