This closes #1449

artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/CertificateUtil.java

@@ -20,6 +20,7 @@ package org.apache.activemq.artemis.core.remoting;
 import io.netty.channel.ChannelHandler;
 import io.netty.handler.ssl.SslHandler;
 import org.apache.activemq.artemis.core.remoting.impl.netty.NettyConnection;
+import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
 import org.apache.activemq.artemis.spi.core.remoting.Connection;
 
 import javax.net.ssl.SSLPeerUnverifiedException;
@@ -28,18 +29,23 @@ import java.security.Principal;
 
 public class CertificateUtil {
 
-   public static X509Certificate[] getCertsFromConnection(Connection connection) {
+   public static X509Certificate[] getCertsFromConnection(RemotingConnection remotingConnection) {
       X509Certificate[] certificates = null;
-      if (connection instanceof NettyConnection) {
+      if (remotingConnection != null) {
-         certificates = org.apache.activemq.artemis.utils.CertificateUtil.getCertsFromChannel(((NettyConnection) connection).getChannel());
+         Connection transportConnection = remotingConnection.getTransportConnection();
+         if (transportConnection instanceof NettyConnection) {
+            certificates = org.apache.activemq.artemis.utils.CertificateUtil.getCertsFromChannel(((NettyConnection) transportConnection).getChannel());
+         }
       }
       return certificates;
    }
 
-   public static Principal getPeerPrincipalFromConnection(Connection connection) {
+   public static Principal getPeerPrincipalFromConnection(RemotingConnection remotingConnection) {
       Principal result = null;
-      if (connection instanceof NettyConnection) {
+      if (remotingConnection != null) {
-         NettyConnection nettyConnection = (NettyConnection) connection;
+         Connection transportConnection = remotingConnection.getTransportConnection();
+         if (transportConnection instanceof NettyConnection) {
+            NettyConnection nettyConnection = (NettyConnection) transportConnection;
             ChannelHandler channelHandler = nettyConnection.getChannel().pipeline().get("ssl");
             if (channelHandler != null && channelHandler instanceof SslHandler) {
                SslHandler sslHandler = (SslHandler) channelHandler;
@@ -49,6 +55,7 @@ public class CertificateUtil {
                }
             }
          }
+      }
 
       return result;
    }

artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/TransportConfigurationUtil.java

@@ -27,9 +27,6 @@ import org.apache.activemq.artemis.core.remoting.impl.netty.NettyConnectorFactor
 import org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants;
 import org.apache.activemq.artemis.utils.ClassloadingUtil;
 
-import javax.security.auth.login.AppConfigurationEntry;
-import javax.security.auth.login.Configuration;
-
 /**
  * Stores static mappings of class names to ConnectorFactory instances to act as a central repo for ConnectorFactory
  * objects.
@@ -99,28 +96,4 @@ public class TransportConfigurationUtil {
       return false;
    }
 
-   public static Configuration kerb5Config(String principal, boolean initiator) {
-      final Map<String, String> krb5LoginModuleOptions = new HashMap<>();
-      krb5LoginModuleOptions.put("isInitiator", String.valueOf(initiator));
-      krb5LoginModuleOptions.put("principal", principal);
-      krb5LoginModuleOptions.put("useKeyTab", "true");
-      krb5LoginModuleOptions.put("storeKey", "true");
-      krb5LoginModuleOptions.put("doNotPrompt", "true");
-      krb5LoginModuleOptions.put("renewTGT", "true");
-      krb5LoginModuleOptions.put("refreshKrb5Config", "true");
-      krb5LoginModuleOptions.put("useTicketCache", "true");
-      String ticketCache = System.getenv("KRB5CCNAME");
-      if (ticketCache != null) {
-         krb5LoginModuleOptions.put("ticketCache", ticketCache);
-      }
-      return new Configuration() {
-         @Override
-         public AppConfigurationEntry[] getAppConfigurationEntry(String name) {
-            return new AppConfigurationEntry[]{
-               new AppConfigurationEntry("com.sun.security.auth.module.Krb5LoginModule",
-                       AppConfigurationEntry.LoginModuleControlFlag.REQUIRED,
-                       krb5LoginModuleOptions)};
-         }
-      };
-   }
 }

artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnector.java

@@ -98,7 +98,6 @@ import org.apache.activemq.artemis.api.core.ActiveMQException;
 import org.apache.activemq.artemis.core.client.ActiveMQClientLogger;
 import org.apache.activemq.artemis.core.client.ActiveMQClientMessageBundle;
 import org.apache.activemq.artemis.core.protocol.core.impl.ActiveMQClientProtocolManager;
-import org.apache.activemq.artemis.core.remoting.impl.TransportConfigurationUtil;
 import org.apache.activemq.artemis.core.remoting.impl.ssl.SSLSupport;
 import org.apache.activemq.artemis.core.server.ActiveMQComponent;
 import org.apache.activemq.artemis.spi.core.remoting.AbstractConnector;
@@ -523,18 +522,8 @@ public class NettyConnector extends AbstractConnector {
             if (sslEnabled && !useServlet) {
 
                Subject subject = null;
-               if (kerb5Config != null && kerb5Config.length() > 0) {
+               if (kerb5Config != null) {
-
+                  LoginContext loginContext = new LoginContext(kerb5Config);
-                  LoginContext loginContext = null;
-                  if (Character.isUpperCase(kerb5Config.charAt(0))) {
-                     // use as login.config scope
-                     loginContext = new LoginContext(kerb5Config);
-                  } else {
-                     // inline keytab config using kerb5Config as principal
-                     loginContext = new LoginContext("", null, null,
-                             TransportConfigurationUtil.kerb5Config(kerb5Config, true));
-                  }
-
                   loginContext.login();
                   subject = loginContext.getSubject();
                   verifyHost = true;

artemis-core-client/src/main/java/org/apache/activemq/artemis/spi/core/protocol/AbstractRemotingConnection.java

@@ -32,6 +32,8 @@ import org.apache.activemq.artemis.spi.core.remoting.Connection;
 import org.apache.activemq.artemis.spi.core.remoting.ReadyListener;
 import org.jboss.logging.Logger;
 
+import javax.security.auth.Subject;
+
 public abstract class AbstractRemotingConnection implements RemotingConnection {
 
    private static final Logger logger = Logger.getLogger(AbstractRemotingConnection.class);
@@ -219,4 +221,9 @@ public abstract class AbstractRemotingConnection implements RemotingConnection {
    public boolean isSupportsFlowControl() {
       return true;
    }
+
+   @Override
+   public Subject getSubject() {
+      return null;
+   }
 }

artemis-core-client/src/main/java/org/apache/activemq/artemis/spi/core/protocol/RemotingConnection.java

@@ -27,6 +27,8 @@ import org.apache.activemq.artemis.spi.core.remoting.BufferHandler;
 import org.apache.activemq.artemis.spi.core.remoting.Connection;
 import org.apache.activemq.artemis.spi.core.remoting.ReadyListener;
 
+import javax.security.auth.Subject;
+
 /**
  * A RemotingConnection is a connection between a client and a server.
  *
@@ -206,4 +208,10 @@ public interface RemotingConnection extends BufferHandler {
     * @return
     */
    boolean isSupportsFlowControl();
+
+   /**
+    * the possibly null identity associated with this connection
+    * @return
+    */
+   Subject getSubject();
 }

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPConnectionCallback.java

@@ -43,6 +43,7 @@ import org.apache.activemq.artemis.protocol.amqp.proton.AmqpSupport;
 import org.apache.activemq.artemis.protocol.amqp.proton.transaction.ProtonTransactionImpl;
 import org.apache.activemq.artemis.protocol.amqp.proton.handler.ExtCapability;
 import org.apache.activemq.artemis.protocol.amqp.sasl.AnonymousServerSASL;
+import org.apache.activemq.artemis.protocol.amqp.sasl.GSSAPIServerSASL;
 import org.apache.activemq.artemis.protocol.amqp.sasl.PlainSASL;
 import org.apache.activemq.artemis.protocol.amqp.sasl.SASLResult;
 import org.apache.activemq.artemis.protocol.amqp.sasl.ServerSASL;
@@ -76,6 +77,8 @@ public class AMQPConnectionCallback implements FailureListener, CloseListener {
 
    private ActiveMQServer server;
 
+   private final String[] saslMechanisms;
+
    public AMQPConnectionCallback(ProtonProtocolManager manager,
                                  Connection connection,
                                  Executor closeExecutor,
@@ -84,25 +87,40 @@ public class AMQPConnectionCallback implements FailureListener, CloseListener {
       this.connection = connection;
       this.closeExecutor = closeExecutor;
       this.server = server;
+      saslMechanisms = manager.getSaslMechanisms();
    }
 
-   public ServerSASL[] getSASLMechnisms() {
+   public String[] getSaslMechanisms() {
-
+      return saslMechanisms;
-      ServerSASL[] result;
-
-      if (isSupportsAnonymous()) {
-         result = new ServerSASL[]{new PlainSASL(manager.getServer().getSecurityStore()), new AnonymousServerSASL()};
-      } else {
-         result = new ServerSASL[]{new PlainSASL(manager.getServer().getSecurityStore())};
    }
 
+   public ServerSASL getServerSASL(final String mechanism) {
+      ServerSASL result = null;
+      switch (mechanism) {
+         case PlainSASL.NAME:
+            result = new PlainSASL(server.getSecurityStore());
+            break;
+
+         case AnonymousServerSASL.NAME:
+            result = new AnonymousServerSASL();
+            break;
+
+         case GSSAPIServerSASL.NAME:
+            GSSAPIServerSASL gssapiServerSASL = new GSSAPIServerSASL();
+            gssapiServerSASL.setLoginConfigScope(manager.getSaslLoginConfigScope());
+            result = gssapiServerSASL;
+            break;
+
+         default:
+            break;
+      }
       return result;
    }
 
    public boolean isSupportsAnonymous() {
       boolean supportsAnonymous = false;
       try {
-         manager.getServer().getSecurityStore().authenticate(null, null, null);
+         server.getSecurityStore().authenticate(null, null, null);
          supportsAnonymous = true;
       } catch (Exception e) {
          // authentication failed so no anonymous support

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/ActiveMQProtonRemotingConnection.java

@@ -25,10 +25,13 @@ import org.apache.activemq.artemis.api.core.SimpleString;
 import org.apache.activemq.artemis.core.client.ActiveMQClientLogger;
 import org.apache.activemq.artemis.protocol.amqp.proton.AMQPConnectionContext;
 import org.apache.activemq.artemis.protocol.amqp.proton.AmqpSupport;
+import org.apache.activemq.artemis.protocol.amqp.sasl.SASLResult;
 import org.apache.activemq.artemis.spi.core.protocol.AbstractRemotingConnection;
 import org.apache.activemq.artemis.spi.core.remoting.Connection;
 import org.apache.qpid.proton.amqp.transport.ErrorCondition;
 
+import javax.security.auth.Subject;
+
 /**
  * This is a Server's Connection representation used by ActiveMQ Artemis.
  */
@@ -148,4 +151,13 @@ public class ActiveMQProtonRemotingConnection extends AbstractRemotingConnection
    public void killMessage(SimpleString nodeID) {
       //unsupported
    }
+
+   @Override
+   public Subject getSubject() {
+      SASLResult saslResult = amqpConnection.getSASLResult();
+      if (saslResult != null) {
+         return saslResult.getSubject();
+      }
+      return null;
+   }
 }

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/ProtonProtocolManager.java

@@ -36,6 +36,7 @@ import org.apache.activemq.artemis.core.server.management.NotificationListener;
 import org.apache.activemq.artemis.jms.client.ActiveMQDestination;
 import org.apache.activemq.artemis.protocol.amqp.proton.AMQPConnectionContext;
 import org.apache.activemq.artemis.protocol.amqp.proton.AMQPConstants;
+import org.apache.activemq.artemis.protocol.amqp.sasl.MechanismFinder;
 import org.apache.activemq.artemis.spi.core.protocol.AbstractProtocolManager;
 import org.apache.activemq.artemis.spi.core.protocol.ConnectionEntry;
 import org.apache.activemq.artemis.spi.core.protocol.ProtocolManagerFactory;
@@ -63,6 +64,12 @@ public class ProtonProtocolManager extends AbstractProtocolManager<AMQPMessage,
 
    private int amqpLowCredits = 30;
 
+   private int initialRemoteMaxFrameSize = 4 * 1024;
+
+   private String[] saslMechanisms = MechanismFinder.getKnownMechanisms();
+
+   private String saslLoginConfigScope = "amqp-sasl-gssapi";
+
    /*
    * used when you want to treat senders as a subscription on an address rather than consuming from the actual queue for
    * the address. This can be changed on the acceptor.
@@ -197,6 +204,23 @@ public class ProtonProtocolManager extends AbstractProtocolManager<AMQPMessage,
       this.maxFrameSize = maxFrameSize;
    }
 
+   public String[] getSaslMechanisms() {
+      return saslMechanisms;
+   }
+
+   public void setSaslMechanisms(String[] saslMechanisms) {
+      this.saslMechanisms = saslMechanisms;
+   }
+
+   public String getSaslLoginConfigScope() {
+      return saslLoginConfigScope;
+   }
+
+   public void setSaslLoginConfigScope(String saslLoginConfigScope) {
+      this.saslLoginConfigScope = saslLoginConfigScope;
+   }
+
+
    @Override
    public void setAnycastPrefix(String anycastPrefix) {
       for (String prefix : anycastPrefix.split(",")) {
@@ -223,4 +247,13 @@ public class ProtonProtocolManager extends AbstractProtocolManager<AMQPMessage,
    public void invokeOutgoing(AMQPMessage message, ActiveMQProtonRemotingConnection connection) {
       super.invokeInterceptors(this.outgoingInterceptors, message, connection);
    }
+
+   public int getInitialRemoteMaxFrameSize() {
+      return initialRemoteMaxFrameSize;
+   }
+
+   public void setInitialRemoteMaxFrameSize(int initialRemoteMaxFrameSize) {
+      this.initialRemoteMaxFrameSize = initialRemoteMaxFrameSize;
+   }
+
 }

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPConnectionContext.java

@@ -34,6 +34,7 @@ import org.apache.activemq.artemis.protocol.amqp.exceptions.ActiveMQAMQPExceptio
 import org.apache.activemq.artemis.protocol.amqp.proton.handler.EventHandler;
 import org.apache.activemq.artemis.protocol.amqp.proton.handler.ExtCapability;
 import org.apache.activemq.artemis.protocol.amqp.proton.handler.ProtonHandler;
+import org.apache.activemq.artemis.protocol.amqp.sasl.AnonymousServerSASL;
 import org.apache.activemq.artemis.protocol.amqp.sasl.SASLResult;
 import org.apache.activemq.artemis.spi.core.remoting.ReadyListener;
 import org.apache.activemq.artemis.utils.ByteUtil;
@@ -103,6 +104,7 @@ public class AMQPConnectionContext extends ProtonInitializable implements EventH
          transport.setIdleTimeout(idleTimeout);
       }
       transport.setChannelMax(channelMax);
+      transport.setInitialRemoteMaxFrameSize(protocolManager.getInitialRemoteMaxFrameSize());
       transport.setMaxFrameSize(maxFrameSize);
    }
 
@@ -321,7 +323,12 @@ public class AMQPConnectionContext extends ProtonInitializable implements EventH
    @Override
    public void onAuthInit(ProtonHandler handler, Connection connection, boolean sasl) {
       if (sasl) {
-         handler.createServerSASL(connectionCallback.getSASLMechnisms());
+         // configured mech in decreasing order of preference
+         String[] mechanisms = connectionCallback.getSaslMechanisms();
+         if (mechanisms == null || mechanisms.length == 0) {
+            mechanisms = AnonymousServerSASL.ANONYMOUS_MECH;
+         }
+         handler.createServerSASL(mechanisms);
       } else {
          if (!connectionCallback.isSupportsAnonymous()) {
             connectionCallback.sendSASLSupported();
@@ -331,6 +338,11 @@ public class AMQPConnectionContext extends ProtonInitializable implements EventH
       }
    }
 
+   @Override
+   public void onSaslRemoteMechanismChosen(ProtonHandler handler, String mech) {
+      handler.setChosenMechanism(connectionCallback.getServerSASL(mech));
+   }
+
    @Override
    public void onTransport(Transport transport) {
       handler.flushBytes();

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/handler/EventHandler.java

@@ -31,6 +31,8 @@ public interface EventHandler {
 
    void onAuthInit(ProtonHandler handler, Connection connection, boolean sasl);
 
+   void onSaslRemoteMechanismChosen(ProtonHandler handler, String mech);
+
    void onInit(Connection connection) throws Exception;
 
    void onLocalOpen(Connection connection) throws Exception;

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/handler/ProtonHandler.java

@@ -18,7 +18,6 @@ package org.apache.activemq.artemis.protocol.amqp.proton.handler;
 
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.Executor;
@@ -63,12 +62,12 @@ public class ProtonHandler extends ProtonInitializable {
 
    private Sasl serverSasl;
 
+   private ServerSASL chosenMechanism;
+
    private final ReentrantLock lock = new ReentrantLock();
 
    private final long creationTime;
 
-   private Map<String, ServerSASL> saslHandlers;
-
    private SASLResult saslResult;
 
    protected volatile boolean dataReceived;
@@ -157,17 +156,10 @@ public class ProtonHandler extends ProtonInitializable {
       return this;
    }
 
-   public void createServerSASL(ServerSASL[] handlers) {
+   public void createServerSASL(String[] mechanisms) {
       this.serverSasl = transport.sasl();
-      saslHandlers = new HashMap<>();
-      String[] names = new String[handlers.length];
-      int count = 0;
-      for (ServerSASL handler : handlers) {
-         saslHandlers.put(handler.getName(), handler);
-         names[count++] = handler.getName();
-      }
       this.serverSasl.server();
-      serverSasl.setMechanisms(names);
+      serverSasl.setMechanisms(mechanisms);
    }
 
    public void flushBytes() {
@@ -292,9 +284,14 @@ public class ProtonHandler extends ProtonInitializable {
 
    protected void checkServerSASL() {
       if (serverSasl != null && serverSasl.getRemoteMechanisms().length > 0) {
-         // TODO: should we look at the first only?
+
-         ServerSASL mechanism = saslHandlers.get(serverSasl.getRemoteMechanisms()[0]);
+         if (chosenMechanism == null) {
-         if (mechanism != null) {
+            if (log.isTraceEnabled()) {
+               log.trace("SASL chosenMechanism: " + serverSasl.getRemoteMechanisms()[0]);
+            }
+            dispatchRemoteMechanismChosen(serverSasl.getRemoteMechanisms()[0]);
+         }
+         if (chosenMechanism != null) {
 
             byte[] dataSASL = new byte[serverSasl.pending()];
             serverSasl.recv(dataSASL, 0, dataSASL.length);
@@ -303,30 +300,46 @@ public class ProtonHandler extends ProtonInitializable {
                log.trace("Working on sasl::" + (dataSASL != null && dataSASL.length > 0 ? ByteUtil.bytesToHex(dataSASL, 2) : "Anonymous"));
             }
 
-            saslResult = mechanism.processSASL(dataSASL);
+            byte[] response = chosenMechanism.processSASL(dataSASL);
-
+            if (response != null) {
-            if (saslResult != null && saslResult.isSuccess()) {
+               serverSasl.send(response, 0, response.length);
-               serverSasl.done(Sasl.SaslOutcome.PN_SASL_OK);
+            }
-               serverSasl = null;
+            saslResult = chosenMechanism.result();
-               saslHandlers.clear();
+
-               saslHandlers = null;
+            if (saslResult != null) {
-            } else {
+               if (saslResult.isSuccess()) {
-               serverSasl.done(Sasl.SaslOutcome.PN_SASL_AUTH);
+                  saslComplete(Sasl.SaslOutcome.PN_SASL_OK);
+               } else {
+                  saslComplete(Sasl.SaslOutcome.PN_SASL_AUTH);
+               }
             }
-            serverSasl = null;
          } else {
             // no auth available, system error
-            serverSasl.done(Sasl.SaslOutcome.PN_SASL_SYS);
+            saslComplete(Sasl.SaslOutcome.PN_SASL_SYS);
          }
       }
    }
 
+   private void saslComplete(Sasl.SaslOutcome saslOutcome) {
+      serverSasl.done(saslOutcome);
+      serverSasl = null;
+      if (chosenMechanism != null) {
+         chosenMechanism.done();
+      }
+   }
+
    private void dispatchAuth(boolean sasl) {
       for (EventHandler h : handlers) {
          h.onAuthInit(this, getConnection(), sasl);
       }
    }
 
+   private void dispatchRemoteMechanismChosen(final String mech) {
+      for (EventHandler h : handlers) {
+         h.onSaslRemoteMechanismChosen(this, mech);
+      }
+   }
+
    private void dispatch() {
       Event ev;
 
@@ -376,4 +389,8 @@ public class ProtonHandler extends ProtonInitializable {
       this.connection.open();
       flush();
    }
+
+   public void setChosenMechanism(ServerSASL chosenMechanism) {
+      this.chosenMechanism = chosenMechanism;
+   }
 }

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/sasl/AnonymousServerSASL.java

@@ -18,17 +18,29 @@ package org.apache.activemq.artemis.protocol.amqp.sasl;
 
 public class AnonymousServerSASL implements ServerSASL {
 
+   public static final String NAME = "ANONYMOUS";
+   public static final String[] ANONYMOUS_MECH = new String[] {NAME};
+
    public AnonymousServerSASL() {
    }
 
    @Override
    public String getName() {
-      return "ANONYMOUS";
+      return NAME;
    }
 
    @Override
-   public SASLResult processSASL(byte[] bytes) {
+   public byte[] processSASL(byte[] bytes) {
+      return null;
+   }
+
+   @Override
+   public SASLResult result() {
       return new PlainSASLResult(true, null, null);
    }
+
+   @Override
+   public void done() {
+   }
 }
 

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/sasl/GSSAPISASLResult.java

@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.artemis.protocol.amqp.sasl;
+
+import javax.security.auth.Subject;
+import java.security.Principal;
+
+public class GSSAPISASLResult implements SASLResult {
+
+   private final boolean success;
+   private final Subject identity = new Subject();
+   private String user;
+
+
+   public GSSAPISASLResult(boolean success, Principal peer) {
+      this.success = success;
+      if (success) {
+         this.identity.getPrivateCredentials().add(peer);
+         this.user = peer.getName();
+      }
+   }
+
+   @Override
+   public String getUser() {
+      return user;
+   }
+
+   @Override
+   public Subject getSubject() {
+      return identity;
+   }
+
+   @Override
+   public boolean isSuccess() {
+      return success;
+   }
+}

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/sasl/GSSAPIServerSASL.java

@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.artemis.protocol.amqp.sasl;
+
+import org.jboss.logging.Logger;
+
+import javax.security.auth.Subject;
+import javax.security.auth.callback.Callback;
+import javax.security.auth.callback.CallbackHandler;
+import javax.security.auth.callback.UnsupportedCallbackException;
+import javax.security.auth.kerberos.KerberosPrincipal;
+import javax.security.auth.login.LoginContext;
+import javax.security.sasl.AuthorizeCallback;
+import javax.security.sasl.Sasl;
+import javax.security.sasl.SaslException;
+import javax.security.sasl.SaslServer;
+import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
+import java.util.HashMap;
+
+/*
+ * delegate the the jdk GSSAPI support
+ */
+public class GSSAPIServerSASL implements ServerSASL {
+   private static final Logger log = Logger.getLogger(GSSAPIServerSASL.class);
+
+   public static final String NAME = "GSSAPI";
+   private String loginConfigScope;
+   private SaslServer saslServer;
+   private Subject jaasId;
+   private SASLResult result;
+
+   @Override
+   public String getName() {
+      return NAME;
+   }
+
+   @Override
+   public byte[] processSASL(byte[] bytes) {
+      try {
+         if (jaasId == null) {
+            // populate subject with acceptor private credentials
+            LoginContext loginContext = new LoginContext(loginConfigScope);
+            loginContext.login();
+            jaasId = loginContext.getSubject();
+         }
+
+         if (saslServer == null) {
+            saslServer = Subject.doAs(jaasId, (PrivilegedExceptionAction<SaslServer>) () -> Sasl.createSaslServer(NAME, null, null, new HashMap<String, String>(), new CallbackHandler() {
+               @Override
+               public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException {
+                  for (Callback callback : callbacks) {
+                     if (callback instanceof AuthorizeCallback) {
+                        AuthorizeCallback authorizeCallback = (AuthorizeCallback) callback;
+                        // only ok to authenticate as self
+                        authorizeCallback.setAuthorized(authorizeCallback.getAuthenticationID().equals(authorizeCallback.getAuthorizationID()));
+                     }
+                  }
+               }
+            }));
+         }
+
+         byte[] challenge = Subject.doAs(jaasId, (PrivilegedExceptionAction<byte[]>) () -> saslServer.evaluateResponse(bytes));
+         if (saslServer.isComplete()) {
+            result = new GSSAPISASLResult(true, new KerberosPrincipal(saslServer.getAuthorizationID()));
+         }
+         return challenge;
+
+      } catch (Exception outOfHere) {
+         log.info("Error on sasl input: " + outOfHere.toString(), outOfHere);
+         result = new GSSAPISASLResult(false, null);
+      }
+      return null;
+   }
+
+   @Override
+   public SASLResult result() {
+      return result;
+   }
+
+   @Override
+   public void done() {
+      if (saslServer != null) {
+         try {
+            saslServer.dispose();
+         } catch (SaslException error) {
+            log.debug("Exception on sasl dispose", error);
+         }
+      }
+   }
+
+   public String getLoginConfigScope() {
+      return loginConfigScope;
+   }
+
+   public void setLoginConfigScope(String loginConfigScope) {
+      this.loginConfigScope = loginConfigScope;
+   }
+
+}

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/sasl/MechanismFinder.java

@@ -0,0 +1,27 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.activemq.artemis.protocol.amqp.sasl;
+
+public class MechanismFinder {
+
+   public static String[] KNOWN_MECHANISMS = new String[]{PlainSASL.NAME, AnonymousServerSASL.NAME};
+
+   public static String[] getKnownMechanisms() {
+      return KNOWN_MECHANISMS;
+   }
+}

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/sasl/PlainSASLResult.java

@@ -16,6 +16,8 @@
  */
 package org.apache.activemq.artemis.protocol.amqp.sasl;
 
+import javax.security.auth.Subject;
+
 public class PlainSASLResult implements SASLResult {
 
    private boolean success;
@@ -28,6 +30,11 @@ public class PlainSASLResult implements SASLResult {
       this.password = password;
    }
 
+   @Override
+   public Subject getSubject() {
+      return null;
+   }
+
    @Override
    public String getUser() {
       return user;
@@ -41,4 +48,5 @@ public class PlainSASLResult implements SASLResult {
    public boolean isSuccess() {
       return success;
    }
+
 }

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/sasl/SASLResult.java

@@ -16,9 +16,13 @@
  */
 package org.apache.activemq.artemis.protocol.amqp.sasl;
 
+import javax.security.auth.Subject;
+
 public interface SASLResult {
 
    String getUser();
 
+   Subject getSubject();
+
    boolean isSuccess();
 }

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/sasl/ServerSASL.java

@@ -20,5 +20,9 @@ public interface ServerSASL {
 
    String getName();
 
-   SASLResult processSASL(byte[] bytes);
+   byte[] processSASL(byte[] bytes);
+
+   SASLResult result();
+
+   void done();
 }

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/sasl/ServerSASLPlain.java

@@ -19,6 +19,7 @@ package org.apache.activemq.artemis.protocol.amqp.sasl;
 public class ServerSASLPlain implements ServerSASL {
 
    public static final String NAME = "PLAIN";
+   private SASLResult result = null;
 
    @Override
    public String getName() {
@@ -26,7 +27,7 @@ public class ServerSASLPlain implements ServerSASL {
    }
 
    @Override
-   public SASLResult processSASL(byte[] data) {
+   public byte[] processSASL(byte[] data) {
       String username = null;
       String password = null;
       String bytes = new String(data);
@@ -47,7 +48,18 @@ public class ServerSASLPlain implements ServerSASL {
 
       boolean success = authenticate(username, password);
 
-      return new PlainSASLResult(success, username, password);
+      result = new PlainSASLResult(success, username, password);
+
+      return null;
+   }
+
+   @Override
+   public SASLResult result() {
+      return result;
+   }
+
+   @Override
+   public void done() {
    }
 
    /**

artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/protocol/amqp/sasl/PlainSASLTest.java

@@ -27,7 +27,8 @@ public class PlainSASLTest {
       byte[] bytesResult = plainSASL.getBytes();
 
       ServerSASLPlain serverSASLPlain = new ServerSASLPlain();
-      PlainSASLResult result = (PlainSASLResult) serverSASLPlain.processSASL(bytesResult);
+      serverSASLPlain.processSASL(bytesResult);
+      PlainSASLResult result = (PlainSASLResult) serverSASLPlain.result();
       Assert.assertEquals("user-me", result.getUser());
       Assert.assertEquals("password-secret", result.getPassword());
    }

artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTConnection.java

@@ -31,6 +31,8 @@ import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
 import org.apache.activemq.artemis.spi.core.remoting.Connection;
 import org.apache.activemq.artemis.spi.core.remoting.ReadyListener;
 
+import javax.security.auth.Subject;
+
 public class MQTTConnection implements RemotingConnection {
 
    private final Connection transportConnection;
@@ -226,4 +228,9 @@ public class MQTTConnection implements RemotingConnection {
    public boolean isSupportsFlowControl() {
       return false;
    }
+
+   @Override
+   public Subject getSubject() {
+      return null;
+   }
 }

artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireProtocolManager.java

@@ -459,7 +459,7 @@ public class OpenWireProtocolManager implements ProtocolManager<Interceptor>, Cl
    }
 
    public void validateUser(String login, String passcode, OpenWireConnection connection) throws Exception {
-      server.getSecurityStore().authenticate(login, passcode, connection.getTransportConnection());
+      server.getSecurityStore().authenticate(login, passcode, connection);
    }
 
    public void sendBrokerInfo(OpenWireConnection connection) throws Exception {

artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/StompConnection.java

@@ -51,6 +51,8 @@ import org.apache.activemq.artemis.utils.ConfigurationHelper;
 import org.apache.activemq.artemis.utils.ExecutorFactory;
 import org.apache.activemq.artemis.utils.VersionLoader;
 
+import javax.security.auth.Subject;
+
 import static org.apache.activemq.artemis.core.protocol.stomp.ActiveMQStompProtocolMessageBundle.BUNDLE;
 
 public final class StompConnection implements RemotingConnection {
@@ -560,7 +562,7 @@ public final class StompConnection implements RemotingConnection {
       manager.sendReply(this, frame);
    }
 
-   public boolean validateUser(final String login, final String pass, final Connection connection) {
+   public boolean validateUser(final String login, final String pass, final RemotingConnection connection) {
       this.valid = manager.validateUser(login, pass, connection);
       if (valid) {
          this.login = login;
@@ -779,4 +781,9 @@ public final class StompConnection implements RemotingConnection {
    public boolean isSupportsFlowControl() {
       return false;
    }
+
+   @Override
+   public Subject getSubject() {
+      return null;
+   }
 }

artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/StompProtocolManager.java

@@ -320,16 +320,16 @@ public class StompProtocolManager extends AbstractProtocolManager<StompFrame, St
       return "activemq";
    }
 
-   public boolean validateUser(String login, String passcode, Connection connection) {
+   public boolean validateUser(String login, String passcode, RemotingConnection remotingConnection) {
       boolean validated = true;
 
       ActiveMQSecurityManager sm = server.getSecurityManager();
 
       if (sm != null && server.getConfiguration().isSecurityEnabled()) {
          if (sm instanceof ActiveMQSecurityManager3) {
-            validated = ((ActiveMQSecurityManager3) sm).validateUser(login, passcode, connection) != null;
+            validated = ((ActiveMQSecurityManager3) sm).validateUser(login, passcode, remotingConnection) != null;
          } else if (sm instanceof ActiveMQSecurityManager2) {
-            validated = ((ActiveMQSecurityManager2) sm).validateUser(login, passcode, CertificateUtil.getCertsFromConnection(connection));
+            validated = ((ActiveMQSecurityManager2) sm).validateUser(login, passcode, CertificateUtil.getCertsFromConnection(remotingConnection));
          } else {
             validated = sm.validateUser(login, passcode);
          }

artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/v10/StompFrameHandlerV10.java

@@ -52,7 +52,7 @@ public class StompFrameHandlerV10 extends VersionedStompFrameHandler implements
       String clientID = headers.get(Stomp.Headers.Connect.CLIENT_ID);
       String requestID = headers.get(Stomp.Headers.Connect.REQUEST_ID);
 
-      if (connection.validateUser(login, passcode, connection.getTransportConnection())) {
+      if (connection.validateUser(login, passcode, connection)) {
          connection.setClientID(clientID);
          connection.setValid(true);
 

artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/v11/StompFrameHandlerV11.java

@@ -68,7 +68,7 @@ public class StompFrameHandlerV11 extends VersionedStompFrameHandler implements
       String requestID = headers.get(Stomp.Headers.Connect.REQUEST_ID);
 
       try {
-         if (connection.validateUser(login, passcode, connection.getTransportConnection())) {
+         if (connection.validateUser(login, passcode, connection)) {
             connection.setClientID(clientID);
             connection.setValid(true);
 

artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptor.java

@@ -71,7 +71,6 @@ import org.apache.activemq.artemis.api.core.management.CoreNotificationType;
 import org.apache.activemq.artemis.core.client.impl.ClientSessionFactoryImpl;
 import org.apache.activemq.artemis.core.protocol.ProtocolHandler;
 import org.apache.activemq.artemis.core.remoting.impl.AbstractAcceptor;
-import org.apache.activemq.artemis.core.remoting.impl.TransportConfigurationUtil;
 import org.apache.activemq.artemis.core.remoting.impl.ssl.SSLSupport;
 import org.apache.activemq.artemis.core.security.ActiveMQPrincipal;
 import org.apache.activemq.artemis.core.server.ActiveMQComponent;
@@ -442,17 +441,9 @@ public class NettyAcceptor extends AbstractAcceptor {
          throw ise;
       }
       Subject subject = null;
-      if (kerb5Config != null && kerb5Config.length() > 0) {
+      if (kerb5Config != null) {
-         LoginContext loginContext = null;
+         LoginContext loginContext = new LoginContext(kerb5Config);
-         if (Character.isUpperCase(kerb5Config.charAt(0))) {
-            // use as login.config scope
-            loginContext = new LoginContext(kerb5Config);
-         } else {
-            loginContext = new LoginContext("", null, null,
-                    TransportConfigurationUtil.kerb5Config(kerb5Config, false));
-         }
          loginContext.login();
-
          subject = loginContext.getSubject();
       }
 

artemis-server/src/main/java/org/apache/activemq/artemis/core/security/SecurityStore.java

@@ -17,11 +17,11 @@
 package org.apache.activemq.artemis.core.security;
 
 import org.apache.activemq.artemis.api.core.SimpleString;
-import org.apache.activemq.artemis.spi.core.remoting.Connection;
+import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
 
 public interface SecurityStore {
 
-   String authenticate(String user, String password, Connection transportConnection) throws Exception;
+   String authenticate(String user, String password, RemotingConnection remotingConnection) throws Exception;
 
    void check(SimpleString address, CheckType checkType, SecurityAuth session) throws Exception;
 

artemis-server/src/main/java/org/apache/activemq/artemis/core/security/impl/SecurityStoreImpl.java

@@ -34,7 +34,7 @@ import org.apache.activemq.artemis.core.server.management.Notification;
 import org.apache.activemq.artemis.core.server.management.NotificationService;
 import org.apache.activemq.artemis.core.settings.HierarchicalRepository;
 import org.apache.activemq.artemis.core.settings.HierarchicalRepositoryChangeListener;
-import org.apache.activemq.artemis.spi.core.remoting.Connection;
+import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
 import org.apache.activemq.artemis.spi.core.security.ActiveMQSecurityManager;
 import org.apache.activemq.artemis.spi.core.security.ActiveMQSecurityManager2;
 import org.apache.activemq.artemis.spi.core.security.ActiveMQSecurityManager3;
@@ -104,7 +104,7 @@ public class SecurityStoreImpl implements SecurityStore, HierarchicalRepositoryC
    @Override
    public String authenticate(final String user,
                               final String password,
-                              Connection connection) throws Exception {
+                              RemotingConnection connection) throws Exception {
       if (securityEnabled) {
 
          if (managementClusterUser.equals(user)) {
@@ -185,7 +185,7 @@ public class SecurityStoreImpl implements SecurityStore, HierarchicalRepositoryC
          final boolean validated;
          if (securityManager instanceof ActiveMQSecurityManager3) {
             final ActiveMQSecurityManager3 securityManager3 = (ActiveMQSecurityManager3) securityManager;
-            validated = securityManager3.validateUserAndRole(user, session.getPassword(), roles, checkType, saddress, session.getRemotingConnection().getTransportConnection()) != null;
+            validated = securityManager3.validateUserAndRole(user, session.getPassword(), roles, checkType, saddress, session.getRemotingConnection()) != null;
          } else if (securityManager instanceof ActiveMQSecurityManager2) {
             final ActiveMQSecurityManager2 securityManager2 = (ActiveMQSecurityManager2) securityManager;
             validated = securityManager2.validateUserAndRole(user, session.getPassword(), roles, checkType, saddress, session.getRemotingConnection());

artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java

@@ -1394,7 +1394,7 @@ public class ActiveMQServerImpl implements ActiveMQServer {
       String validatedUser = "";
 
       if (securityStore != null) {
-         validatedUser = securityStore.authenticate(username, password, connection.getTransportConnection());
+         validatedUser = securityStore.authenticate(username, password, connection);
       }
 
       checkSessionLimit(validatedUser);

artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/security/ActiveMQJAASSecurityManager.java

@@ -29,7 +29,7 @@ import java.util.Set;
 import org.apache.activemq.artemis.core.config.impl.SecurityConfiguration;
 import org.apache.activemq.artemis.core.security.CheckType;
 import org.apache.activemq.artemis.core.security.Role;
-import org.apache.activemq.artemis.spi.core.remoting.Connection;
+import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
 import org.apache.activemq.artemis.spi.core.security.jaas.JaasCallbackHandler;
 import org.apache.activemq.artemis.spi.core.security.jaas.RolePrincipal;
 import org.apache.activemq.artemis.spi.core.security.jaas.UserPrincipal;
@@ -88,9 +88,9 @@ public class ActiveMQJAASSecurityManager implements ActiveMQSecurityManager3 {
    }
 
    @Override
-   public String validateUser(final String user, final String password, Connection connection) {
+   public String validateUser(final String user, final String password, RemotingConnection remotingConnection) {
       try {
-         return getUserFromSubject(getAuthenticatedSubject(user, password, connection));
+         return getUserFromSubject(getAuthenticatedSubject(user, password, remotingConnection));
       } catch (LoginException e) {
          if (logger.isDebugEnabled()) {
             logger.debug("Couldn't validate user", e);
@@ -121,10 +121,10 @@ public class ActiveMQJAASSecurityManager implements ActiveMQSecurityManager3 {
                                      final Set<Role> roles,
                                      final CheckType checkType,
                                      final String address,
-                                     final Connection connection) {
+                                     final RemotingConnection remotingConnection) {
       Subject localSubject;
       try {
-         localSubject = getAuthenticatedSubject(user, password, connection);
+         localSubject = getAuthenticatedSubject(user, password, remotingConnection);
       } catch (LoginException e) {
          if (logger.isDebugEnabled()) {
             logger.debug("Couldn't validate user", e);
@@ -170,7 +170,7 @@ public class ActiveMQJAASSecurityManager implements ActiveMQSecurityManager3 {
 
    private Subject getAuthenticatedSubject(final String user,
                                            final String password,
-                                           final Connection connection) throws LoginException {
+                                           final RemotingConnection remotingConnection) throws LoginException {
       LoginContext lc;
       ClassLoader currentLoader = Thread.currentThread().getContextClassLoader();
       ClassLoader thisLoader = this.getClass().getClassLoader();
@@ -178,10 +178,10 @@ public class ActiveMQJAASSecurityManager implements ActiveMQSecurityManager3 {
          if (thisLoader != currentLoader) {
             Thread.currentThread().setContextClassLoader(thisLoader);
          }
-         if (certificateConfigurationName != null && certificateConfigurationName.length() > 0 && getCertsFromConnection(connection) != null) {
+         if (certificateConfigurationName != null && certificateConfigurationName.length() > 0 && getCertsFromConnection(remotingConnection) != null) {
-            lc = new LoginContext(certificateConfigurationName, null, new JaasCallbackHandler(user, password, connection), certificateConfiguration);
+            lc = new LoginContext(certificateConfigurationName, null, new JaasCallbackHandler(user, password, remotingConnection), certificateConfiguration);
          } else {
-            lc = new LoginContext(configurationName, null, new JaasCallbackHandler(user, password, connection), configuration);
+            lc = new LoginContext(configurationName, null, new JaasCallbackHandler(user, password, remotingConnection), configuration);
          }
          lc.login();
          return lc.getSubject();

artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/security/ActiveMQSecurityManager3.java

@@ -20,7 +20,7 @@ import java.util.Set;
 
 import org.apache.activemq.artemis.core.security.CheckType;
 import org.apache.activemq.artemis.core.security.Role;
-import org.apache.activemq.artemis.spi.core.remoting.Connection;
+import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
 
 /**
  * Used to validate whether a user is authorized to connect to the
@@ -40,9 +40,10 @@ public interface ActiveMQSecurityManager3 extends ActiveMQSecurityManager {
     *
     * @param user     the user
     * @param password the users password
+    * @param remotingConnection
     * @return the name of the validated user or null if the user isn't validated
     */
-   String validateUser(String user, String password, Connection connection);
+   String validateUser(String user, String password, RemotingConnection remotingConnection);
 
    /**
     * Determine whether the given user is valid and whether they have
@@ -56,7 +57,7 @@ public interface ActiveMQSecurityManager3 extends ActiveMQSecurityManager {
     * @param roles      the user's roles
     * @param checkType  which permission to validate
     * @param address    the address for which to perform authorization
-    * @param connection the user's connection
+    * @param remotingConnection the user's connection
     * @return the name of the validated user or null if the user isn't validated
     */
    String validateUserAndRole(String user,
@@ -64,5 +65,5 @@ public interface ActiveMQSecurityManager3 extends ActiveMQSecurityManager {
                               Set<Role> roles,
                               CheckType checkType,
                               String address,
-                              Connection connection);
+                              RemotingConnection remotingConnection);
 }

artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/security/jaas/JaasCallbackHandler.java

@@ -16,14 +16,17 @@
  */
 package org.apache.activemq.artemis.spi.core.security.jaas;
 
-import org.apache.activemq.artemis.spi.core.remoting.Connection;
+import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
 
+import javax.security.auth.Subject;
 import javax.security.auth.callback.Callback;
 import javax.security.auth.callback.CallbackHandler;
 import javax.security.auth.callback.NameCallback;
 import javax.security.auth.callback.PasswordCallback;
 import javax.security.auth.callback.UnsupportedCallbackException;
+import javax.security.auth.kerberos.KerberosPrincipal;
 import java.io.IOException;
+import java.security.Principal;
 
 import static org.apache.activemq.artemis.core.remoting.CertificateUtil.getCertsFromConnection;
 import static org.apache.activemq.artemis.core.remoting.CertificateUtil.getPeerPrincipalFromConnection;
@@ -35,12 +38,12 @@ public class JaasCallbackHandler implements CallbackHandler {
 
    private final String username;
    private final String password;
-   final Connection connection;
+   final RemotingConnection remotingConnection;
 
-   public JaasCallbackHandler(String username, String password, Connection connection) {
+   public JaasCallbackHandler(String username, String password, RemotingConnection remotingConnection) {
       this.username = username;
       this.password = password;
-      this.connection = connection;
+      this.remotingConnection = remotingConnection;
    }
 
    @Override
@@ -63,11 +66,19 @@ public class JaasCallbackHandler implements CallbackHandler {
          } else if (callback instanceof CertificateCallback) {
             CertificateCallback certCallback = (CertificateCallback) callback;
 
-            certCallback.setCertificates(getCertsFromConnection(connection));
+            certCallback.setCertificates(getCertsFromConnection(remotingConnection));
-         } else if (callback instanceof Krb5SslCallback) {
+         } else if (callback instanceof Krb5Callback) {
-            Krb5SslCallback krb5SslCallback = (Krb5SslCallback) callback;
+            Krb5Callback krb5Callback = (Krb5Callback) callback;
 
-            krb5SslCallback.setPeerPrincipal(getPeerPrincipalFromConnection(connection));
+            Subject peerSubject = remotingConnection.getSubject();
+            if (peerSubject != null) {
+               for (Principal principal : peerSubject.getPrivateCredentials(KerberosPrincipal.class)) {
+                  krb5Callback.setPeerPrincipal(principal);
+                  return;
+               }
+            }
+
+            krb5Callback.setPeerPrincipal(getPeerPrincipalFromConnection(remotingConnection));
          } else {
             throw new UnsupportedCallbackException(callback);
          }

artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/security/jaas/Krb5Callback.java

@@ -20,9 +20,9 @@ import javax.security.auth.callback.Callback;
 import java.security.Principal;
 
 /**
- * A Callback for SSL kerberos peer principal.
+ * A Callback for kerberos peer principal.
  */
-public class Krb5SslCallback implements Callback {
+public class Krb5Callback implements Callback {
 
    Principal peerPrincipal;
 

artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/security/jaas/Krb5LoginModule.java

@@ -31,11 +31,11 @@ import java.util.List;
 import java.util.Map;
 
 /**
- * populate a subject with kerberos and UserPrincipal from SSLContext peerPrincipal
+ * populate a subject with kerberos credential from the handler
  */
-public class Krb5SslLoginModule implements LoginModule {
+public class Krb5LoginModule implements LoginModule {
 
-   private static final Logger logger = Logger.getLogger(Krb5SslLoginModule.class);
+   private static final Logger logger = Logger.getLogger(Krb5LoginModule.class);
 
    private Subject subject;
    private final List<Principal> principals = new LinkedList<>();
@@ -55,7 +55,7 @@ public class Krb5SslLoginModule implements LoginModule {
    public boolean login() throws LoginException {
       Callback[] callbacks = new Callback[1];
 
-      callbacks[0] = new Krb5SslCallback();
+      callbacks[0] = new Krb5Callback();
       try {
          callbackHandler.handle(callbacks);
       } catch (IOException ioe) {
@@ -63,7 +63,7 @@ public class Krb5SslLoginModule implements LoginModule {
       } catch (UnsupportedCallbackException uce) {
          throw new LoginException(uce.getMessage() + " not available to obtain information from user");
       }
-      principals.add(((Krb5SslCallback)callbacks[0]).getPeerPrincipal());
+      principals.add(((Krb5Callback)callbacks[0]).getPeerPrincipal());
       if (!principals.isEmpty()) {
          loginSucceeded = true;
       }

docs/user-manual/en/security.md

@@ -649,6 +649,31 @@ like the following:
 The simplest way to make the login configuration available to JAAS is to add the directory containing the file,
 `login.config`, to your CLASSPATH.
 
+### Kerberos Authentication
+
+The [Krb5LoginModule](https://docs.oracle.com/javase/7/docs/jre/api/security/jaas/spec/com/sun/security/auth/module/Krb5LoginModule.html)
+can be used with JAAS to authenticate using the Kerberos protocol.
+
+Using SASL over [AMQP](using-AMQP.md), Kerberos authentication is supported using the `GSSAPI` SASL mechanism. With SASL doing Kerberos
+authentication, TLS can be used to provide integrity and confidentially to the communications channel in the normal way.
+The `GSSAPI` SASL mechanism must be enabled on the amqp acceptor by adding it to the `saslMechanisms` list url parameter:
+`saslMechanisms="GSSAPI<,PLAIN, etc>`.
+
+By default the server will use a JAAS login configuration scope named `amqp-sasl-gssapi` to obtain acceptor Kerberos
+credentials. An alternative configuration scope can be specified on the amqp acceptor url using the parameter: `saslLoginConfigScope=<some other scope>`.
+
+On the server, the Kerberos authenticated Peer Principal can be associated with a JAAS Subject as an Apache ActiveMQ Artemis UserPrincipal
+using the Apache ActiveMQ Artemis Krb5LoginModule login module. The [PropertiesLoginModule](#propertiesloginmodule) can be used to map
+the peer principal to a role.
+Note: the Kerberos Peer Principal does not exist as an Apache ActiveMQ Artemis user.
+
+    org.apache.activemq.artemis.spi.core.security.jaas.Krb5LoginModule optional;
+
+The legacy [rfc2712](http://www.ietf.org/rfc/rfc2712.txt) defines TLS Kerberos cipher suites that can be used by TLS to negotiate
+Kerberos authentication. The cypher suites offered by rfc2712 are dated and insecure and rfc2712 has been superseded by
+SASL GSSAPI. However, for clients that don't support SASL (core client), using TLS can provide Kerberos authentication
+over an *unsecure* channel.
+
 
 ## Changing the username/password for clustering
 

tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/JMSConnectionWithSecurityTest.java

@@ -36,6 +36,11 @@ public class JMSConnectionWithSecurityTest extends JMSClientTestSupport {
       return true;
    }
 
+   @Override
+   protected String getJmsConnectionURIOptions() {
+      return "amqp.saslMechanisms=PLAIN";
+   }
+
    @Test(timeout = 10000)
    public void testNoUserOrPassword() throws Exception {
       Connection connection = null;

tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/JMSSaslGssapiTest.java

@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.artemis.tests.integration.amqp;
+
+import javax.jms.Connection;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import java.io.File;
+import java.net.URI;
+import java.net.URL;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.activemq.artemis.core.security.Role;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.spi.core.security.ActiveMQJAASSecurityManager;
+import org.apache.activemq.artemis.utils.RandomUtil;
+import org.apache.hadoop.minikdc.MiniKdc;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class JMSSaslGssapiTest extends JMSClientTestSupport {
+
+   static {
+      String path = System.getProperty("java.security.auth.login.config");
+      if (path == null) {
+         URL resource = JMSSaslGssapiTest.class.getClassLoader().getResource("login.config");
+         if (resource != null) {
+            path = resource.getFile();
+            System.setProperty("java.security.auth.login.config", path);
+         }
+      }
+   }
+   MiniKdc kdc = null;
+
+   @Before
+   public void setUpKerberos() throws Exception {
+      kdc = new MiniKdc(MiniKdc.createConf(), temporaryFolder.newFolder("kdc"));
+      kdc.start();
+
+      // hard coded match, default_keytab_name in minikdc-krb5.conf template
+      File userKeyTab = new File("target/test.krb5.keytab");
+      kdc.createPrincipal(userKeyTab, "client", "amqp/localhost");
+
+      java.util.logging.Logger logger = java.util.logging.Logger.getLogger("javax.security.sasl");
+      logger.setLevel(java.util.logging.Level.FINEST);
+      logger.addHandler(new java.util.logging.ConsoleHandler());
+      for (java.util.logging.Handler handler: logger.getHandlers()) {
+         handler.setLevel(java.util.logging.Level.FINEST);
+      }
+
+   }
+
+   @After
+   public void stopKerberos() throws Exception {
+      if (kdc != null) {
+         kdc.stop();
+      }
+   }
+
+   @Override
+   protected boolean isSecurityEnabled() {
+      return true;
+   }
+
+   @Override
+   protected void configureBrokerSecurity(ActiveMQServer server) {
+      server.getConfiguration().setSecurityEnabled(isSecurityEnabled());
+      ActiveMQJAASSecurityManager securityManager = (ActiveMQJAASSecurityManager) server.getSecurityManager();
+      securityManager.setConfigurationName("Krb5Plus");
+      securityManager.setConfiguration(null);
+
+      final String roleName = "ALLOW_ALL";
+      Role role = new Role(roleName, true, true, true, true, true, true, true, true, true, true);
+      Set<Role> roles = new HashSet<>();
+      roles.add(role);
+      server.getSecurityRepository().addMatch(getQueueName().toString(), roles);
+
+   }
+
+   @Override
+   protected String getJmsConnectionURIOptions() {
+      return "amqp.saslMechanisms=GSSAPI";
+   }
+
+   @Override
+   protected URI getBrokerQpidJMSConnectionURI() {
+
+      try {
+         int port = AMQP_PORT;
+
+         // match the sasl.service <the host name>
+         String uri = "amqp://localhost:" + port;
+
+         if (!getJmsConnectionURIOptions().isEmpty()) {
+            uri = uri + "?" + getJmsConnectionURIOptions();
+         }
+
+         return new URI(uri);
+      } catch (Exception e) {
+         throw new RuntimeException();
+      }
+   }
+
+   @Override
+   protected void configureAMQPAcceptorParameters(Map<String, Object> params) {
+      params.put("saslMechanisms", "GSSAPI");
+      params.put("saslLoginConfigScope", "amqp-sasl-gssapi");
+   }
+
+   @Test(timeout = 600000)
+   public void testConnection() throws Exception {
+      Connection connection = createConnection("client", null);
+      connection.start();
+
+      try {
+         Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+         javax.jms.Queue queue = session.createQueue(getQueueName());
+         MessageConsumer consumer = session.createConsumer(queue);
+         MessageProducer producer = session.createProducer(queue);
+
+         final String text = RandomUtil.randomString();
+         producer.send(session.createTextMessage(text));
+
+         TextMessage m = (TextMessage) consumer.receive(1000);
+         assertNotNull(m);
+         assertEquals(text, m.getText());
+
+      } finally {
+         connection.close();
+      }
+   }
+}

tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/security/SecurityTest.java

@@ -54,7 +54,6 @@ import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
 import org.apache.activemq.artemis.core.server.impl.AddressInfo;
 import org.apache.activemq.artemis.core.settings.HierarchicalRepository;
 import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
-import org.apache.activemq.artemis.spi.core.remoting.Connection;
 import org.apache.activemq.artemis.spi.core.security.ActiveMQJAASSecurityManager;
 import org.apache.activemq.artemis.spi.core.security.ActiveMQSecurityManager;
 import org.apache.activemq.artemis.spi.core.security.ActiveMQSecurityManager2;
@@ -1936,7 +1935,7 @@ public class SecurityTest extends ActiveMQTestBase {
          @Override
          public String validateUser(final String username,
                                     final String password,
-                                    final Connection connection) {
+                                    final RemotingConnection remotingConnection) {
             if ((username.equals("foo") || username.equals("bar") || username.equals("all")) && password.equals("frobnicate")) {
                return username;
             } else {
@@ -1960,9 +1959,9 @@ public class SecurityTest extends ActiveMQTestBase {
                                            final Set<Role> requiredRoles,
                                            final CheckType checkType,
                                            final String address,
-                                           final Connection connection) {
+                                           final RemotingConnection connection) {
 
-            if (!(connection instanceof InVMConnection)) {
+            if (!(connection.getTransportConnection() instanceof InVMConnection)) {
                return null;
             }
 

tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/ssl/CoreClientOverOneWaySSLKerb5Test.java

@@ -88,7 +88,7 @@ public class CoreClientOverOneWaySSLKerb5Test extends ActiveMQTestBase {
       tc.getParams().put(TransportConstants.SSL_ENABLED_PROP_NAME, true);
       tc.getParams().put(TransportConstants.ENABLED_CIPHER_SUITES_PROP_NAME, getSuitableCipherSuite());
       tc.getParams().put(TransportConstants.SNIHOST_PROP_NAME, SNI_HOST); // static service name rather than dynamic machine name
-      tc.getParams().put(TransportConstants.SSL_KRB5_CONFIG_PROP_NAME, "client"); // lower case used as principal with default keytab
+      tc.getParams().put(TransportConstants.SSL_KRB5_CONFIG_PROP_NAME, "core-tls-krb5-client");
       final ServerLocator locator = addServerLocator(ActiveMQClient.createServerLocatorWithoutHA(tc));
 
       ClientSessionFactory sf = null;
@@ -171,7 +171,7 @@ public class CoreClientOverOneWaySSLKerb5Test extends ActiveMQTestBase {
 
       params.put(TransportConstants.SSL_ENABLED_PROP_NAME, true);
       params.put(TransportConstants.ENABLED_CIPHER_SUITES_PROP_NAME, getSuitableCipherSuite());
-      params.put(TransportConstants.SSL_KRB5_CONFIG_PROP_NAME, SERVICE_PRINCIPAL);
+      params.put(TransportConstants.SSL_KRB5_CONFIG_PROP_NAME, "core-tls-krb5-server");
 
       ConfigurationImpl config = createBasicConfig().addAcceptorConfiguration(new TransportConfiguration(NETTY_ACCEPTOR_FACTORY, params, "nettySSL"));
       config.setPopulateValidatedUser(true); // so we can verify the kerb5 id is present
@@ -179,7 +179,7 @@ public class CoreClientOverOneWaySSLKerb5Test extends ActiveMQTestBase {
 
       config.addAcceptorConfiguration(new TransportConfiguration(INVM_ACCEPTOR_FACTORY));
 
-      ActiveMQSecurityManager securityManager = new ActiveMQJAASSecurityManager("Krb5SslPlus");
+      ActiveMQSecurityManager securityManager = new ActiveMQJAASSecurityManager("Krb5Plus");
       server = addServer(ActiveMQServers.newActiveMQServer(config, ManagementFactory.getPlatformMBeanServer(), securityManager, false));
       HierarchicalRepository<Set<Role>> securityRepository = server.getSecurityRepository();
 

tests/integration-tests/src/test/resources/login.config

@@ -138,9 +138,9 @@ DualAuthenticationPropertiesLogin {
         org.apache.activemq.jaas.properties.role="dual-authentication-roles.properties";
 };
 
-Krb5SslPlus {
+Krb5Plus {
 
-    org.apache.activemq.artemis.spi.core.security.jaas.Krb5SslLoginModule optional
+    org.apache.activemq.artemis.spi.core.security.jaas.Krb5LoginModule optional
         debug=true;
 
     org.apache.activemq.artemis.spi.core.security.jaas.PropertiesLoginModule optional
@@ -148,3 +148,32 @@ Krb5SslPlus {
         org.apache.activemq.jaas.properties.user="dual-authentication-users.properties"
         org.apache.activemq.jaas.properties.role="dual-authentication-roles.properties";
 };
+
+core-tls-krb5-server {
+    com.sun.security.auth.module.Krb5LoginModule required
+    isInitiator=false
+    storeKey=true
+    useKeyTab=true
+    principal="host/sni.host"
+    debug=true;
+};
+
+core-tls-krb5-client {
+    com.sun.security.auth.module.Krb5LoginModule required
+    principal="client"
+    useKeyTab=true;
+};
+
+amqp-sasl-gssapi {
+    com.sun.security.auth.module.Krb5LoginModule required
+    isInitiator=false
+    storeKey=true
+    useKeyTab=true
+    principal="amqp/localhost"
+    debug=true;
+};
+
+amqp-jms-client {
+    com.sun.security.auth.module.Krb5LoginModule required
+    useKeyTab=true;
+};