1 files changed, 320 insertions, 0 deletions
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/ClientDelegate.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/ClientDelegate.java
new file mode 100644
index 0000000000..c8b7ad2a5e
--- /dev/null
+++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/ClientDelegate.java
@@ -0,0 +1,320 @@
+/*
+ *
+ * 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.qpid.transport;
+
+import org.ietf.jgss.GSSContext;
+import org.ietf.jgss.GSSException;
+import org.ietf.jgss.GSSManager;
+import org.ietf.jgss.GSSName;
+import org.ietf.jgss.Oid;
+
+import org.apache.qpid.security.UsernamePasswordCallbackHandler;
+import static org.apache.qpid.transport.Connection.State.OPEN;
+import static org.apache.qpid.transport.Connection.State.RESUMING;
+import org.apache.qpid.transport.util.Logger;
+
+import javax.security.sasl.Sasl;
+import javax.security.sasl.SaslClient;
+import javax.security.sasl.SaslException;
+import java.lang.management.ManagementFactory;
+import java.lang.management.RuntimeMXBean;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+
+/**
+ * ClientDelegate
+ *
+ */
+
+public class ClientDelegate extends ConnectionDelegate
+{
+    private static final Logger log = Logger.get(ClientDelegate.class);
+
+    private static final String KRB5_OID_STR = "1.2.840.113554.1.2.2";
+    protected static final Oid KRB5_OID;
+
+    static
+    {
+        Oid oid;
+        try
+        {
+            oid = new Oid(KRB5_OID_STR);
+        }
+        catch (GSSException ignore)
+        {
+            oid = null;
+        }
+
+        KRB5_OID = oid;
+    }
+
+    private List<String> clientMechs;
+    private ConnectionSettings conSettings;
+
+    public ClientDelegate(ConnectionSettings settings)
+    {
+        this.conSettings = settings;
+        this.clientMechs = Arrays.asList(settings.getSaslMechs().split(" "));
+    }
+
+    public void init(Connection conn, ProtocolHeader hdr)
+    {
+        if (!(hdr.getMajor() == 0 && hdr.getMinor() == 10))
+        {
+            conn.exception(new ProtocolVersionException(hdr.getMajor(), hdr.getMinor()));
+        }
+    }
+
+    @Override
+    public void connectionStart(Connection conn, ConnectionStart start)
+    {
+        Map<String,Object> clientProperties = new HashMap<String,Object>();
+
+        if(this.conSettings.getClientProperties() != null)
+        {
+            clientProperties.putAll(this.conSettings.getClientProperties());
+        }
+
+        clientProperties.put("qpid.session_flow", 1);
+        clientProperties.put("qpid.client_pid",getPID());
+        clientProperties.put("qpid.client_process",
+                System.getProperty("qpid.client_process","Qpid Java Client"));
+
+        List<Object> brokerMechs = start.getMechanisms();
+        if (brokerMechs == null || brokerMechs.isEmpty())
+        {
+            conn.connectionStartOk
+                (clientProperties, null, null, conn.getLocale());
+            return;
+        }
+
+        List<String> choosenMechs = new ArrayList<String>();
+        for (String mech:clientMechs)
+        {
+            if (brokerMechs.contains(mech))
+            {
+                choosenMechs.add(mech);
+            }
+        }
+
+        if (choosenMechs.size() == 0)
+        {
+            conn.exception(new ConnectionException("The following SASL mechanisms " +
+                    clientMechs.toString()  +
+                    " specified by the client are not supported by the broker"));
+            return;
+        }
+
+        String[] mechs = new String[choosenMechs.size()];
+        choosenMechs.toArray(mechs);
+
+        conn.setServerProperties(start.getServerProperties());
+
+        try
+        {
+            Map<String,Object> saslProps = new HashMap<String,Object>();
+            if (conSettings.isUseSASLEncryption())
+            {
+                saslProps.put(Sasl.QOP, "auth-conf");
+            }
+            UsernamePasswordCallbackHandler handler =
+                new UsernamePasswordCallbackHandler();
+            handler.initialise(conSettings.getUsername(), conSettings.getPassword());
+            SaslClient sc = Sasl.createSaslClient
+                (mechs, null, conSettings.getSaslProtocol(), conSettings.getSaslServerName(), saslProps, handler);
+            conn.setSaslClient(sc);
+
+            byte[] response = sc.hasInitialResponse() ?
+                sc.evaluateChallenge(new byte[0]) : null;
+            conn.connectionStartOk
+                (clientProperties, sc.getMechanismName(), response,
+                 conn.getLocale());
+        }
+        catch (SaslException e)
+        {
+            conn.exception(e);
+        }
+    }
+
+    @Override
+    public void connectionSecure(Connection conn, ConnectionSecure secure)
+    {
+        SaslClient sc = conn.getSaslClient();
+        try
+        {
+            byte[] response = sc.evaluateChallenge(secure.getChallenge());
+            conn.connectionSecureOk(response);
+        }
+        catch (SaslException e)
+        {
+            conn.exception(e);
+        }
+    }
+
+    @Override
+    public void connectionTune(Connection conn, ConnectionTune tune)
+    {
+        int hb_interval = calculateHeartbeatInterval(conSettings.getHeartbeatInterval(),
+                                                     tune.getHeartbeatMin(),
+                                                     tune.getHeartbeatMax()
+                                                     );
+        conn.connectionTuneOk(tune.getChannelMax(),
+                              tune.getMaxFrameSize(),
+                              hb_interval);
+        // The idle timeout is twice the heartbeat amount (in milisecs)
+        conn.setIdleTimeout(hb_interval*1000*2);
+
+        int channelMax = tune.getChannelMax();
+        //0 means no implied limit, except available server resources
+        //(or that forced by protocol limitations [0xFFFF])
+        conn.setChannelMax(channelMax == 0 ? Connection.MAX_CHANNEL_MAX : channelMax);
+
+        conn.connectionOpen(conSettings.getVhost(), null, Option.INSIST);
+    }
+
+    @Override
+    public void connectionOpenOk(Connection conn, ConnectionOpenOk ok)
+    {
+        SaslClient sc = conn.getSaslClient();
+        if (sc != null)
+        {
+            if (sc.getMechanismName().equals("GSSAPI"))
+            {
+                String id = getKerberosUser();
+                if (id != null)
+                {
+                    conn.setUserID(id);
+                }
+            }
+            else if (sc.getMechanismName().equals("EXTERNAL"))
+            {
+                if (conn.getSecurityLayer() != null)
+                {
+                    conn.setUserID(conn.getSecurityLayer().getUserID());
+                }
+            }
+        }
+        
+        if (conn.isConnectionResuming())
+        {
+            conn.setState(RESUMING);
+        }
+        else
+        {
+            conn.setState(OPEN);
+        }
+    }
+
+    @Override
+    public void connectionRedirect(Connection conn, ConnectionRedirect redir)
+    {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public void connectionHeartbeat(Connection conn, ConnectionHeartbeat hearbeat)
+    {
+        conn.connectionHeartbeat();
+    }
+
+    /**
+     * Currently the spec specified the min and max for heartbeat using secs
+     */
+    private int calculateHeartbeatInterval(int heartbeat,int min, int max)
+    {
+        int i = heartbeat;
+        if (i == 0)
+        {
+            log.info("Idle timeout is 0 sec. Heartbeats are disabled.");
+            return 0; // heartbeats are disabled.
+        }
+        else if (i >= min && i <= max)
+        {
+            return i;
+        }
+        else
+        {
+            log.info("The broker does not support the configured connection idle timeout of %s sec," +
+                     " using the brokers max supported value of %s sec instead.", i,max);
+            return max;
+        }
+    }
+
+    private int getPID()
+    {
+        RuntimeMXBean rtb = ManagementFactory.getRuntimeMXBean();
+        String processName = rtb.getName();
+        if (processName != null && processName.indexOf('@')>0)
+        {
+            try
+            {
+                return Integer.parseInt(processName.substring(0,processName.indexOf('@')));
+            }
+            catch(Exception e)
+            {
+                log.warn("Unable to get the client PID due to error",e);
+                return -1;
+            }
+        }
+        else
+        {
+            log.warn("Unable to get the client PID due to unsupported format : " + processName);
+            return -1;
+        }
+
+    }
+
+    private String getKerberosUser()
+    {
+        log.debug("Obtaining userID from kerberos");
+        String service = conSettings.getSaslProtocol() + "@" + conSettings.getSaslServerName();
+        GSSManager manager = GSSManager.getInstance();
+
+        try
+        {
+            GSSName acceptorName = manager.createName(service,
+                GSSName.NT_HOSTBASED_SERVICE, KRB5_OID);
+
+            GSSContext secCtx = manager.createContext(acceptorName,
+                                                      KRB5_OID,
+                                                      null,
+                                                      GSSContext.INDEFINITE_LIFETIME);
+
+            secCtx.initSecContext(new byte[0], 0, 1);
+
+            if (secCtx.getSrcName() != null)
+            {
+                return secCtx.getSrcName().toString();
+            }
+
+        }
+        catch (GSSException e)
+        {
+            log.warn("Unable to retrieve userID from Kerberos due to error",e);
+        }
+
+        return null;
+    }
+}