QPID-5955 : [Java Broker] Allow HTTP Management to run TCP and SSL on the same port

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1615322 13f79535-47bb-0310-9956-ffa450edef68

2 files changed, 361 insertions, 1 deletions

diff --git a/qpid/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/HttpManagement.java b/qpid/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/HttpManagement.java
index ece4ccca4f..613218f2fc 100644
--- a/qpid/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/HttpManagement.java
+++ b/qpid/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/HttpManagement.java
@@ -41,6 +41,7 @@ import org.eclipse.jetty.server.Connector;
 import org.eclipse.jetty.server.Server;
 import org.eclipse.jetty.server.SessionManager;
 import org.eclipse.jetty.server.nio.SelectChannelConnector;
+import org.eclipse.jetty.server.ssl.SslSelectChannelConnector;
 import org.eclipse.jetty.server.ssl.SslSocketConnector;
 import org.eclipse.jetty.servlet.FilterHolder;
 import org.eclipse.jetty.servlet.ServletContextHandler;
@@ -49,6 +50,7 @@ import org.eclipse.jetty.util.ssl.SslContextFactory;
 
 import org.apache.qpid.server.configuration.IllegalConfigurationException;
 import org.apache.qpid.server.logging.messages.ManagementConsoleMessages;
+import org.apache.qpid.server.management.plugin.connector.TcpAndSslSelectChannelConnector;
 import org.apache.qpid.server.management.plugin.filter.ForbiddingAuthorisationFilter;
 import org.apache.qpid.server.management.plugin.filter.RedirectingAuthorisationFilter;
 import org.apache.qpid.server.management.plugin.servlet.DefinedFileServlet;
@@ -380,7 +382,9 @@ public class HttpManagement extends AbstractPluginAdapter<HttpManagement> implem
         {
             throw new ServerScopedRuntimeException("Cannot configure port " + port.getName() + " for transport " + Transport.SSL, e);
         }
-        connector = new SslSocketConnector(factory);
+        connector = port.getTransports().contains(Transport.TCP)
+                ? new TcpAndSslSelectChannelConnector(factory)
+                : new SslSelectChannelConnector(factory);
         return connector;
     }
 
diff --git a/qpid/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/connector/TcpAndSslSelectChannelConnector.java b/qpid/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/connector/TcpAndSslSelectChannelConnector.java
new file mode 100644
index 0000000000..7bdeb7fee8
--- /dev/null
+++ b/qpid/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/connector/TcpAndSslSelectChannelConnector.java
@@ -0,0 +1,356 @@
+package org.apache.qpid.server.management.plugin.connector;
+
+import java.io.IOException;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.SocketChannel;
+
+import javax.net.ssl.SSLEngine;
+import javax.net.ssl.SSLSession;
+
+import org.eclipse.jetty.http.HttpSchemes;
+import org.eclipse.jetty.io.AsyncEndPoint;
+import org.eclipse.jetty.io.Buffer;
+import org.eclipse.jetty.io.Connection;
+import org.eclipse.jetty.io.EndPoint;
+import org.eclipse.jetty.io.nio.AsyncConnection;
+import org.eclipse.jetty.io.nio.IndirectNIOBuffer;
+import org.eclipse.jetty.io.nio.SelectChannelEndPoint;
+import org.eclipse.jetty.io.nio.SelectorManager;
+import org.eclipse.jetty.io.nio.SslConnection;
+import org.eclipse.jetty.server.AsyncHttpConnection;
+import org.eclipse.jetty.server.Request;
+import org.eclipse.jetty.server.nio.SelectChannelConnector;
+import org.eclipse.jetty.server.ssl.SslCertificates;
+import org.eclipse.jetty.util.log.Log;
+import org.eclipse.jetty.util.log.Logger;
+import org.eclipse.jetty.util.ssl.SslContextFactory;
+
+public class TcpAndSslSelectChannelConnector extends SelectChannelConnector
+{
+
+    private static final Logger LOG = Log.getLogger(TcpAndSslSelectChannelConnector.class);
+
+    private final SslContextFactory _sslContextFactory;
+
+    public TcpAndSslSelectChannelConnector(SslContextFactory factory)
+    {
+        _sslContextFactory = factory;
+        addBean(_sslContextFactory);
+        setUseDirectBuffers(false);
+        setSoLingerTime(30000);
+    }
+
+
+    @Override
+    public void customize(EndPoint endpoint, Request request) throws IOException
+    {
+        if(endpoint instanceof SslConnection.SslEndPoint)
+        {
+            request.setScheme(HttpSchemes.HTTPS);
+        }
+
+        super.customize(endpoint,request);
+
+        if(endpoint instanceof SslConnection.SslEndPoint)
+        {
+            SslConnection.SslEndPoint sslEndpoint = (SslConnection.SslEndPoint) endpoint;
+            SSLEngine sslEngine = sslEndpoint.getSslEngine();
+            SSLSession sslSession = sslEngine.getSession();
+
+            SslCertificates.customize(sslSession, endpoint, request);
+        }
+    }
+
+    /* ------------------------------------------------------------------------------- */
+    @Override
+    protected AsyncConnection newConnection(SocketChannel channel, AsyncEndPoint endpoint)
+    {
+        return new ProtocolIdentifyingConnection((ProtocolIdentifyingEndpoint) endpoint);
+    }
+
+    @Override
+    protected SelectChannelEndPoint newEndPoint(final SocketChannel channel,
+                                                final SelectorManager.SelectSet selectSet,
+                                                final SelectionKey key) throws IOException
+    {
+
+        SelectChannelEndPoint endp = new ProtocolIdentifyingEndpoint(channel,selectSet,key, getMaxIdleTime());
+        endp.setConnection(selectSet.getManager().newConnection(channel,endp, key.attachment()));
+        return endp;
+    }
+
+    /* ------------------------------------------------------------ */
+    /**
+     * @param channel A channel which if passed is used as to extract remote
+     * host and port for the purposes of SSL session caching
+     * @return A SSLEngine for a new or cached SSL Session
+     * @throws IOException if the SSLEngine cannot be created
+     */
+    protected SSLEngine createSSLEngine(SocketChannel channel) throws IOException
+    {
+        SSLEngine engine;
+        if (channel != null)
+        {
+            String peerHost = channel.socket().getInetAddress().getHostAddress();
+            int peerPort = channel.socket().getPort();
+            engine = _sslContextFactory.newSslEngine(peerHost, peerPort);
+        }
+        else
+        {
+            engine = _sslContextFactory.newSslEngine();
+        }
+
+        engine.setUseClientMode(false);
+        return engine;
+    }
+
+    @Override
+    protected void doStart() throws Exception
+    {
+        _sslContextFactory.checkKeyStore();
+        _sslContextFactory.start();
+
+        SSLEngine sslEngine = _sslContextFactory.newSslEngine();
+
+        sslEngine.setUseClientMode(false);
+
+        SSLSession sslSession = sslEngine.getSession();
+
+        if (getRequestHeaderSize()<sslSession.getApplicationBufferSize())
+            setRequestHeaderSize(sslSession.getApplicationBufferSize());
+        if (getRequestBufferSize()<sslSession.getApplicationBufferSize())
+            setRequestBufferSize(sslSession.getApplicationBufferSize());
+
+        super.doStart();
+    }
+
+    enum Protocol { UNKNOWN, TCP , SSL }
+
+    private class ProtocolIdentifyingEndpoint extends SelectChannelEndPoint
+    {
+
+        private Protocol _protocol = Protocol.UNKNOWN;
+        private Buffer _preBuffer = new IndirectNIOBuffer(6);
+
+        public ProtocolIdentifyingEndpoint(final SocketChannel channel,
+                                           final SelectorManager.SelectSet selectSet,
+                                           final SelectionKey key, final int maxIdleTime) throws IOException
+        {
+            super(channel, selectSet, key, maxIdleTime);
+        }
+
+        public Protocol getProtocol() throws IOException
+        {
+            if(_protocol == Protocol.UNKNOWN)
+            {
+                if(_preBuffer.space() != 0)
+                {
+                    super.fill(_preBuffer);
+                    _protocol = identifyFromPreBuffer();
+                }
+            }
+            return _protocol;
+        }
+
+        public SocketChannel getSocketChannel()
+        {
+            return (SocketChannel) getChannel();
+        }
+
+        private Protocol identifyFromPreBuffer()
+        {
+            if(_preBuffer.space() == 0)
+            {
+                byte[] helloBytes = _preBuffer.array();
+                if (looksLikeSSLv2ClientHello(helloBytes) || looksLikeSSLv3ClientHello(helloBytes))
+                {
+                    return Protocol.SSL;
+                }
+                else
+                {
+                    return Protocol.TCP;
+                }
+            }
+            return Protocol.UNKNOWN;
+        }
+
+        private boolean looksLikeSSLv3ClientHello(byte[] headerBytes)
+        {
+            return headerBytes[0] == 22 && // SSL Handshake
+                   (headerBytes[1] == 3 && // SSL 3.0 / TLS 1.x
+                    (headerBytes[2] == 0 || // SSL 3.0
+                     headerBytes[2] == 1 || // TLS 1.0
+                     headerBytes[2] == 2 || // TLS 1.1
+                     headerBytes[2] == 3)) && // TLS1.2
+                   (headerBytes[5] == 1); // client_hello
+        }
+
+        private boolean looksLikeSSLv2ClientHello(byte[] headerBytes)
+        {
+            return headerBytes[0] == -128 &&
+                   headerBytes[3] == 3 && // SSL 3.0 / TLS 1.x
+                   (headerBytes[4] == 0 || // SSL 3.0
+                    headerBytes[4] == 1 || // TLS 1.0
+                    headerBytes[4] == 2 || // TLS 1.1
+                    headerBytes[4] == 3);
+        }
+
+        @Override
+        public int fill(final Buffer buffer) throws IOException
+        {
+            int size = 0;
+
+            if(getProtocol() != Protocol.UNKNOWN)
+            {
+                if (_preBuffer.hasContent())
+                {
+                    size = buffer.put(_preBuffer);
+                    _preBuffer.skip(size);
+                }
+                if (buffer.space() != 0)
+                {
+                    size += super.fill(buffer);
+                }
+            }
+            return size;
+        }
+    }
+
+    private class ProtocolIdentifyingConnection implements AsyncConnection
+    {
+        private final ProtocolIdentifyingEndpoint _endpoint;
+        private AsyncConnection _delegate;
+        private final long _timestamp;
+        private IOException _exception;
+
+        private ProtocolIdentifyingConnection(final ProtocolIdentifyingEndpoint endpoint)
+        {
+            _endpoint = endpoint;
+            _timestamp = System.currentTimeMillis();
+        }
+
+        @Override
+        public void onInputShutdown() throws IOException
+        {
+            if (_delegate == null)
+            {
+                createDelegate(true);
+            }
+            _delegate.onInputShutdown();
+        }
+
+        private boolean createDelegate(boolean createPlainWhenUnknown) throws IOException
+        {
+            if(_exception != null)
+            {
+                throw _exception;
+            }
+            Protocol protocol = _endpoint.getProtocol();
+            if(protocol == Protocol.TCP || (createPlainWhenUnknown && protocol == Protocol.UNKNOWN))
+            {
+                // shutdown before enough info arrived to make a decision - just create a non-SSL connection anyway
+                _delegate = new AsyncHttpConnection(TcpAndSslSelectChannelConnector.this, _endpoint, getServer());
+                return true;
+            }
+            else if(protocol == Protocol.SSL)
+            {
+                SocketChannel channel = _endpoint.getSocketChannel();
+                SSLEngine engine = createSSLEngine(channel);
+                SslConnection connection = new SslConnection(engine, _endpoint);
+                AsyncConnection delegate = new AsyncHttpConnection(TcpAndSslSelectChannelConnector.this,
+                                                                   connection.getSslEndPoint(),
+                                                                   getServer());
+                connection.getSslEndPoint().setConnection(delegate);
+                connection.setAllowRenegotiate(_sslContextFactory.isAllowRenegotiate());
+
+                _delegate = connection;
+                return true;
+            }
+            return false;
+        }
+
+        private boolean createDelegateNoException()
+        {
+            try
+            {
+                return createDelegate(false);
+            }
+            catch (IOException e)
+            {
+                _exception = e;
+                return false;
+            }
+        }
+
+        @Override
+        public Connection handle() throws IOException
+        {
+            if(_delegate != null || createDelegate(false))
+            {
+                return _delegate.handle();
+            }
+            return this;
+        }
+
+        @Override
+        public long getTimeStamp()
+        {
+            return _timestamp;
+        }
+
+        @Override
+        public boolean isIdle()
+        {
+            if(_delegate != null || createDelegateNoException())
+            {
+                return _delegate.isIdle();
+            }
+            return false;
+        }
+
+        @Override
+        public boolean isSuspended()
+        {
+            if(_delegate != null || createDelegateNoException())
+            {
+                return _delegate.isSuspended();
+            }
+            return false;
+        }
+
+        @Override
+        public void onClose()
+        {
+            if(_delegate != null)
+            {
+                _delegate.onClose();
+            }
+        }
+
+        @Override
+        public void onIdleExpired(final long idleForMs)
+        {
+            try
+            {
+                if(_delegate != null || createDelegate(true))
+                {
+                    _delegate.onIdleExpired(idleForMs);
+                }
+            }
+            catch (IOException e)
+            {
+                LOG.ignore(e);
+
+                try
+                {
+                    _endpoint.close();
+                }
+                catch(IOException e2)
+                {
+                    LOG.ignore(e2);
+                }
+            }
+        }
+    }
+
+}