First cut at adding SASL into thrift to authenticate connections to Blur.
Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/8674d9dc Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/8674d9dc Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/8674d9dc Branch: refs/heads/master Commit: 8674d9dcfe0456c07592ef9fd7ad344de0726cde Parents: b95467d Author: Aaron McCurry <[email protected]> Authored: Sat Jan 31 21:24:08 2015 -0500 Committer: Aaron McCurry <[email protected]> Committed: Sat Jan 31 21:24:08 2015 -0500 ---------------------------------------------------------------------- .../server/ControllerServerEventHandler.java | 27 +- .../blur/server/ShardServerEventHandler.java | 24 +- .../blur/thrift/ThriftBlurControllerServer.java | 13 +- .../blur/thrift/ThriftBlurShardServer.java | 17 +- .../org/apache/blur/thrift/ThriftServer.java | 102 +++- .../org/apache/blur/ExternalThriftServer.java | 5 +- .../test/java/org/apache/blur/MiniCluster.java | 6 +- .../thrift_0_9_0/server/TThreadPoolServer.java | 2 +- .../java/org/apache/blur/thrift/BlurClient.java | 5 + .../java/org/apache/blur/thrift/ClientPool.java | 61 +- .../AnonymousAuthenticationProviderImpl.java | 36 ++ .../blur/thrift/sasl/AuthenticationType.java | 21 + .../sasl/CustomAuthenticationProviderImpl.java | 56 ++ .../sasl/PasswordAuthenticationProvider.java | 17 + .../blur/thrift/sasl/PlainSaslServer.java | 165 ++++++ .../org/apache/blur/thrift/sasl/SaslHelper.java | 203 +++++++ .../blur/thrift/sasl/TSaslClientTransport.java | 109 ++++ .../blur/thrift/sasl/TSaslServerTransport.java | 231 ++++++++ .../apache/blur/thrift/sasl/TSaslTransport.java | 567 +++++++++++++++++++ .../org/apache/blur/thrift/util/SaslClient.java | 70 +++ .../org/apache/blur/utils/BlurConstants.java | 3 + 21 files changed, 1656 insertions(+), 84 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/8674d9dc/blur-core/src/main/java/org/apache/blur/server/ControllerServerEventHandler.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/server/ControllerServerEventHandler.java b/blur-core/src/main/java/org/apache/blur/server/ControllerServerEventHandler.java index 7ca983d..f17719e 100644 --- a/blur-core/src/main/java/org/apache/blur/server/ControllerServerEventHandler.java +++ b/blur-core/src/main/java/org/apache/blur/server/ControllerServerEventHandler.java @@ -31,6 +31,7 @@ import org.apache.blur.log.LogFactory; import org.apache.blur.thirdparty.thrift_0_9_0.protocol.TProtocol; import org.apache.blur.thirdparty.thrift_0_9_0.server.ServerContext; import org.apache.blur.thirdparty.thrift_0_9_0.server.TServerEventHandler; +import org.apache.blur.thirdparty.thrift_0_9_0.transport.TSocket; import org.apache.blur.thirdparty.thrift_0_9_0.transport.TTransport; import com.yammer.metrics.Metrics; @@ -39,7 +40,8 @@ import com.yammer.metrics.core.Meter; import com.yammer.metrics.core.MetricName; /** - * {@link ConrtollerServerContext} is the session manager for the controller servers. + * {@link ConrtollerServerContext} is the session manager for the controller + * servers. */ public class ControllerServerEventHandler implements TServerEventHandler { @@ -64,13 +66,24 @@ public class ControllerServerEventHandler implements TServerEventHandler { } @Override - public ServerContext createContext(TProtocol input, TProtocol output, Object selectionKeyObject) { + public ServerContext createContext(TProtocol input, TProtocol output, Object remoteInstance) { LOG.debug("Client connected"); - SelectionKey selectionKey = (SelectionKey) selectionKeyObject; - SocketChannel channel = (SocketChannel) selectionKey.channel(); - Socket socket = channel.socket(); - SocketAddress remoteSocketAddress = socket.getRemoteSocketAddress(); - SocketAddress localSocketAddress = socket.getLocalSocketAddress(); + SocketAddress remoteSocketAddress; + SocketAddress localSocketAddress; + if (remoteInstance instanceof SelectionKey) { + SelectionKey selectionKey = (SelectionKey) remoteInstance; + SocketChannel channel = (SocketChannel) selectionKey.channel(); + Socket socket = channel.socket(); + remoteSocketAddress = socket.getRemoteSocketAddress(); + localSocketAddress = socket.getLocalSocketAddress(); + } else if (remoteInstance instanceof TSocket) { + TSocket tSocket = (TSocket) remoteInstance; + Socket socket = tSocket.getSocket(); + remoteSocketAddress = socket.getRemoteSocketAddress(); + localSocketAddress = socket.getLocalSocketAddress(); + } else { + throw new RuntimeException("Cannot track remote connection off [" + remoteInstance + "]"); + } _connectionMeter.mark(); _connections.incrementAndGet(); return new ControllerServerContext(localSocketAddress, remoteSocketAddress); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/8674d9dc/blur-core/src/main/java/org/apache/blur/server/ShardServerEventHandler.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/server/ShardServerEventHandler.java b/blur-core/src/main/java/org/apache/blur/server/ShardServerEventHandler.java index 55fb8d4..3886ce1 100644 --- a/blur-core/src/main/java/org/apache/blur/server/ShardServerEventHandler.java +++ b/blur-core/src/main/java/org/apache/blur/server/ShardServerEventHandler.java @@ -31,6 +31,7 @@ import org.apache.blur.log.LogFactory; import org.apache.blur.thirdparty.thrift_0_9_0.protocol.TProtocol; import org.apache.blur.thirdparty.thrift_0_9_0.server.ServerContext; import org.apache.blur.thirdparty.thrift_0_9_0.server.TServerEventHandler; +import org.apache.blur.thirdparty.thrift_0_9_0.transport.TSocket; import org.apache.blur.thirdparty.thrift_0_9_0.transport.TTransport; import com.yammer.metrics.Metrics; @@ -65,13 +66,24 @@ public class ShardServerEventHandler implements TServerEventHandler { } @Override - public ServerContext createContext(TProtocol input, TProtocol output, Object selectionKeyObject) { + public ServerContext createContext(TProtocol input, TProtocol output, Object remoteInstance) { LOG.debug("Client connected"); - SelectionKey selectionKey = (SelectionKey) selectionKeyObject; - SocketChannel channel = (SocketChannel) selectionKey.channel(); - Socket socket = channel.socket(); - SocketAddress remoteSocketAddress = socket.getRemoteSocketAddress(); - SocketAddress localSocketAddress = socket.getLocalSocketAddress(); + SocketAddress remoteSocketAddress; + SocketAddress localSocketAddress; + if (remoteInstance instanceof SelectionKey) { + SelectionKey selectionKey = (SelectionKey) remoteInstance; + SocketChannel channel = (SocketChannel) selectionKey.channel(); + Socket socket = channel.socket(); + remoteSocketAddress = socket.getRemoteSocketAddress(); + localSocketAddress = socket.getLocalSocketAddress(); + } else if (remoteInstance instanceof TSocket) { + TSocket tSocket = (TSocket) remoteInstance; + Socket socket = tSocket.getSocket(); + remoteSocketAddress = socket.getRemoteSocketAddress(); + localSocketAddress = socket.getLocalSocketAddress(); + } else { + throw new RuntimeException("Cannot track remote connection off [" + remoteInstance + "]"); + } _connectionMeter.mark(); _connections.incrementAndGet(); return new ShardServerContext(localSocketAddress, remoteSocketAddress); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/8674d9dc/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurControllerServer.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurControllerServer.java b/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurControllerServer.java index 2c7764a..538316a 100644 --- a/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurControllerServer.java +++ b/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurControllerServer.java @@ -67,7 +67,7 @@ import org.apache.blur.metrics.ReporterSetup; import org.apache.blur.server.ControllerServerEventHandler; import org.apache.blur.thirdparty.thrift_0_9_0.protocol.TJSONProtocol; import org.apache.blur.thirdparty.thrift_0_9_0.server.TServlet; -import org.apache.blur.thirdparty.thrift_0_9_0.transport.TNonblockingServerSocket; +import org.apache.blur.thirdparty.thrift_0_9_0.transport.TServerTransport; import org.apache.blur.thrift.generated.Blur; import org.apache.blur.thrift.generated.Blur.Iface; import org.apache.blur.trace.Trace; @@ -109,11 +109,9 @@ public class ThriftBlurControllerServer extends ThriftServer { if (configBindPort == 0) { instanceBindPort = 0; } - TNonblockingServerSocket tNonblockingServerSocket = ThriftServer.getTNonblockingServerSocket(bindAddress, - instanceBindPort); - if (configBindPort == 0) { - instanceBindPort = tNonblockingServerSocket.getServerSocket().getLocalPort(); - } + TServerTransport serverTransport = ThriftServer.getTServerTransport(bindAddress, instanceBindPort, + configuration); + instanceBindPort = ThriftServer.getBindingPort(serverTransport); LOG.info("Controller Server using index [{0}] bind address [{1}]", serverIndex, bindAddress + ":" + instanceBindPort); @@ -197,7 +195,7 @@ public class ThriftBlurControllerServer extends ThriftServer { final ThriftBlurControllerServer server = new ThriftBlurControllerServer(); server.setNodeName(nodeName); - server.setServerTransport(tNonblockingServerSocket); + server.setServerTransport(serverTransport); server.setThreadCount(threadCount); server.setEventHandler(eventHandler); server.setIface(iface); @@ -205,6 +203,7 @@ public class ThriftBlurControllerServer extends ThriftServer { server.setMaxReadBufferBytes(configuration.getLong(BLUR_CONTROLLER_THRIFT_MAX_READ_BUFFER_BYTES, Long.MAX_VALUE)); server.setSelectorThreads(configuration.getInt(BLUR_CONTROLLER_THRIFT_SELECTOR_THREADS, 2)); server.setMaxFrameSize(configuration.getInt(BLUR_THRIFT_MAX_FRAME_SIZE, BLUR_THRIFT_DEFAULT_MAX_FRAME_SIZE)); + server.setConfiguration(configuration); int configGuiPort = Integer.parseInt(configuration.get(BLUR_GUI_CONTROLLER_PORT)); int instanceGuiPort = configGuiPort + serverIndex; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/8674d9dc/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java b/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java index cbdd2ce..eb6be75 100644 --- a/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java +++ b/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java @@ -93,7 +93,7 @@ import org.apache.blur.store.BlockCacheDirectoryFactoryV2; import org.apache.blur.store.buffer.BufferStore; import org.apache.blur.thirdparty.thrift_0_9_0.protocol.TJSONProtocol; import org.apache.blur.thirdparty.thrift_0_9_0.server.TServlet; -import org.apache.blur.thirdparty.thrift_0_9_0.transport.TNonblockingServerSocket; +import org.apache.blur.thirdparty.thrift_0_9_0.transport.TServerTransport; import org.apache.blur.thrift.generated.Blur; import org.apache.blur.thrift.generated.Blur.Iface; import org.apache.blur.trace.Trace; @@ -112,7 +112,6 @@ import sun.misc.VM; public class ThriftBlurShardServer extends ThriftServer { - private static final Log LOG = LogFactory.getLog(ThriftBlurShardServer.class); private static final boolean enableJsonReporter = false; private static final long _64MB = 64 * 1024 * 1024; @@ -148,11 +147,8 @@ public class ThriftBlurShardServer extends ThriftServer { if (configBindPort == 0) { instanceBindPort = 0; } - TNonblockingServerSocket tNonblockingServerSocket = ThriftServer.getTNonblockingServerSocket(bindAddress, - instanceBindPort); - if (configBindPort == 0) { - instanceBindPort = tNonblockingServerSocket.getServerSocket().getLocalPort(); - } + TServerTransport serverTransport = ThriftServer.getTServerTransport(bindAddress, instanceBindPort, configuration); + instanceBindPort = ThriftServer.getBindingPort(serverTransport); Set<Entry<String, String>> set = configuration.getProperties().entrySet(); for (Entry<String, String> e : set) { @@ -304,7 +300,7 @@ public class ThriftBlurShardServer extends ThriftServer { final ThriftBlurShardServer server = new ThriftBlurShardServer(); server.setNodeName(nodeName); - server.setServerTransport(tNonblockingServerSocket); + server.setServerTransport(serverTransport); server.setThreadCount(threadCount); server.setIface(iface); server.setEventHandler(eventHandler); @@ -312,6 +308,7 @@ public class ThriftBlurShardServer extends ThriftServer { server.setMaxReadBufferBytes(configuration.getLong(BLUR_SHARD_THRIFT_MAX_READ_BUFFER_BYTES, Long.MAX_VALUE)); server.setSelectorThreads(configuration.getInt(BLUR_SHARD_THRIFT_SELECTOR_THREADS, 2)); server.setMaxFrameSize(configuration.getInt(BLUR_THRIFT_MAX_FRAME_SIZE, BLUR_THRIFT_DEFAULT_MAX_FRAME_SIZE)); + server.setConfiguration(configuration); // This will shutdown the server when the correct path is set in zk BlurShutdown shutdown = new BlurShutdown() { @@ -319,8 +316,8 @@ public class ThriftBlurShardServer extends ThriftServer { public void shutdown() { ThreadWatcher threadWatcher = ThreadWatcher.instance(); quietClose(makeCloseable(hdfsKeyValueTimer), makeCloseable(indexImporterTimer), blockCacheDirectoryFactory, - commandManager, traceStorage, server, shardServer, indexManager, indexServer, threadWatcher, - clusterStatus, zooKeeper, httpServer); + commandManager, traceStorage, server, shardServer, indexManager, indexServer, threadWatcher, clusterStatus, + zooKeeper, httpServer); } }; server.setShutdown(shutdown); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/8674d9dc/blur-core/src/main/java/org/apache/blur/thrift/ThriftServer.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/thrift/ThriftServer.java b/blur-core/src/main/java/org/apache/blur/thrift/ThriftServer.java index 94967ec..a46666d 100644 --- a/blur-core/src/main/java/org/apache/blur/thrift/ThriftServer.java +++ b/blur-core/src/main/java/org/apache/blur/thrift/ThriftServer.java @@ -24,6 +24,7 @@ import static org.apache.blur.metrics.MetricsConstants.ORG_APACHE_BLUR; import static org.apache.blur.metrics.MetricsConstants.SYSTEM; import static org.apache.blur.utils.BlurConstants.BLUR_HDFS_TRACE_PATH; import static org.apache.blur.utils.BlurConstants.BLUR_HOME; +import static org.apache.blur.utils.BlurConstants.BLUR_SECURITY_SASL_ENABLED; import static org.apache.blur.utils.BlurConstants.BLUR_ZOOKEEPER_TRACE_PATH; import java.io.BufferedReader; @@ -38,7 +39,6 @@ import java.lang.management.OperatingSystemMXBean; import java.lang.reflect.Method; import java.net.InetAddress; import java.net.InetSocketAddress; -import java.net.ServerSocket; import java.net.UnknownHostException; import java.util.UUID; import java.util.concurrent.ExecutorService; @@ -49,13 +49,21 @@ import org.apache.blur.log.Log; import org.apache.blur.log.LogFactory; import org.apache.blur.manager.indexserver.BlurServerShutDown.BlurShutdown; import org.apache.blur.thirdparty.thrift_0_9_0.protocol.TBinaryProtocol; +import org.apache.blur.thirdparty.thrift_0_9_0.protocol.TCompactProtocol; import org.apache.blur.thirdparty.thrift_0_9_0.server.TServer; import org.apache.blur.thirdparty.thrift_0_9_0.server.TServerEventHandler; +import org.apache.blur.thirdparty.thrift_0_9_0.server.TThreadPoolServer; +import org.apache.blur.thirdparty.thrift_0_9_0.server.TThreadPoolServer.Args; import org.apache.blur.thirdparty.thrift_0_9_0.transport.TFramedTransport; import org.apache.blur.thirdparty.thrift_0_9_0.transport.TNonblockingServerSocket; +import org.apache.blur.thirdparty.thrift_0_9_0.transport.TNonblockingServerTransport; +import org.apache.blur.thirdparty.thrift_0_9_0.transport.TServerSocket; +import org.apache.blur.thirdparty.thrift_0_9_0.transport.TServerTransport; import org.apache.blur.thirdparty.thrift_0_9_0.transport.TTransportException; import org.apache.blur.thrift.generated.Blur; import org.apache.blur.thrift.generated.Blur.Iface; +import org.apache.blur.thrift.sasl.SaslHelper; +import org.apache.blur.thrift.sasl.TSaslServerTransport; import org.apache.blur.thrift.server.TThreadedSelectorServer; import org.apache.blur.thrift.server.TThreadedSelectorServer.Args.AcceptPolicy; import org.apache.blur.trace.LogTraceStorage; @@ -80,11 +88,12 @@ public class ThriftServer { private ExecutorService _queryExexutorService; private ExecutorService _mutateExecutorService; private TServerEventHandler _eventHandler; - private TNonblockingServerSocket _serverTransport; + private TServerTransport _serverTransport; private int _acceptQueueSizePerThread = 4; private long _maxReadBufferBytes = Long.MAX_VALUE; private int _selectorThreads = 2; private int _maxFrameSize = 16384000; + private BlurConfiguration _configuration; public int getMaxFrameSize() { return _maxFrameSize; @@ -94,11 +103,11 @@ public class ThriftServer { _maxFrameSize = maxFrameSize; } - public TNonblockingServerSocket getServerTransport() { + public TServerTransport getServerTransport() { return _serverTransport; } - public void setServerTransport(TNonblockingServerSocket serverTransport) { + public void setServerTransport(TServerTransport serverTransport) { _serverTransport = serverTransport; } @@ -267,38 +276,38 @@ public class ThriftServer { return 0; } - public void start() throws TTransportException { + public void start() throws TTransportException, IOException { _executorService = Executors.newThreadPool("thrift-processors", _threadCount); Blur.Processor<Blur.Iface> processor = new Blur.Processor<Blur.Iface>(_iface); - TThreadedSelectorServer.Args args = new TThreadedSelectorServer.Args(_serverTransport); - args.processor(processor); - args.executorService(_executorService); - args.transportFactory(new TFramedTransport.Factory(_maxFrameSize)); - args.protocolFactory(new TBinaryProtocol.Factory(true, true)); - args.selectorThreads = _selectorThreads; - args.maxReadBufferBytes = _maxReadBufferBytes; - args.acceptQueueSizePerThread(_acceptQueueSizePerThread); - args.acceptPolicy(AcceptPolicy.FAIR_ACCEPT); - - _server = new TThreadedSelectorServer(args); - _server.setServerEventHandler(_eventHandler); - LOG.info("Starting server [{0}]", _nodeName); - _server.serve(); - } - - public static TNonblockingServerSocket getTNonblockingServerSocket(String bindAddress, int bindPort) - throws TTransportException { - InetSocketAddress bindInetSocketAddress = getBindInetSocketAddress(bindAddress, bindPort); - return new TNonblockingServerSocket(bindInetSocketAddress); - } + if (SaslHelper.isSaslEnabled(_configuration)) { + TSaslServerTransport.Factory saslTransportFactory = SaslHelper.getTSaslServerTransportFactory(_configuration); + Args args = new TThreadPoolServer.Args(_serverTransport); + args.executorService(_executorService); + args.processor(processor); + args.protocolFactory(new TCompactProtocol.Factory()); + args.transportFactory(saslTransportFactory); - public int getLocalPort() { - ServerSocket serverSocket = _serverTransport.getServerSocket(); - if (serverSocket == null) { - return 0; + _server = new TThreadPoolServer(args); + _server.setServerEventHandler(_eventHandler); + } else { + TThreadedSelectorServer.Args args = new TThreadedSelectorServer.Args( + (TNonblockingServerTransport) _serverTransport); + args.processor(processor); + args.executorService(_executorService); + args.transportFactory(new TFramedTransport.Factory(_maxFrameSize)); + args.protocolFactory(new TBinaryProtocol.Factory(true, true)); + args.selectorThreads = _selectorThreads; + args.maxReadBufferBytes = _maxReadBufferBytes; + args.acceptQueueSizePerThread(_acceptQueueSizePerThread); + args.acceptPolicy(AcceptPolicy.FAIR_ACCEPT); + + _server = new TThreadedSelectorServer(args); + _server.setServerEventHandler(_eventHandler); } - return serverSocket.getLocalPort(); + + LOG.info("Starting server [{0}]", _nodeName); + _server.serve(); } public static InetSocketAddress getBindInetSocketAddress(String bindAddress, int bindPort) { @@ -395,4 +404,35 @@ public class ThriftServer { public void setSelectorThreads(int selectorThreads) { _selectorThreads = selectorThreads; } + + public BlurConfiguration getConfiguration() { + return _configuration; + } + + public void setConfiguration(BlurConfiguration configuration) { + this._configuration = configuration; + } + + public static TServerTransport getTServerTransport(String bindAddress, int bindPort, BlurConfiguration configuration) + throws TTransportException { + InetSocketAddress bindInetSocketAddress = getBindInetSocketAddress(bindAddress, bindPort); + if (SaslHelper.isSaslEnabled(configuration)) { + return new TServerSocket(bindInetSocketAddress); + } else { + return new TNonblockingServerSocket(bindInetSocketAddress); + } + } + + public static int getBindingPort(TServerTransport serverTransport) { + if (serverTransport instanceof TNonblockingServerSocket) { + TNonblockingServerSocket nonblockingServerSocket = (TNonblockingServerSocket) serverTransport; + return nonblockingServerSocket.getServerSocket().getLocalPort(); + } else if (serverTransport instanceof TServerSocket) { + TServerSocket serverSocket = (TServerSocket) serverTransport; + return serverSocket.getServerSocket().getLocalPort(); + } else { + throw new RuntimeException("Server Transport [" + serverTransport + "] not supported."); + } + } + } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/8674d9dc/blur-core/src/test/java/org/apache/blur/ExternalThriftServer.java ---------------------------------------------------------------------- diff --git a/blur-core/src/test/java/org/apache/blur/ExternalThriftServer.java b/blur-core/src/test/java/org/apache/blur/ExternalThriftServer.java index d8547d3..f56884a 100644 --- a/blur-core/src/test/java/org/apache/blur/ExternalThriftServer.java +++ b/blur-core/src/test/java/org/apache/blur/ExternalThriftServer.java @@ -17,6 +17,7 @@ package org.apache.blur; import java.io.File; +import java.io.IOException; import java.util.Map; import java.util.Map.Entry; import java.util.TreeMap; @@ -69,6 +70,8 @@ public class ExternalThriftServer { thriftServer.start(); } catch (TTransportException e) { e.printStackTrace(); + } catch (IOException e) { + e.printStackTrace(); } } }).start(); @@ -78,7 +81,7 @@ public class ExternalThriftServer { } catch (InterruptedException e) { throw new RuntimeException(e); } - int localPort = thriftServer.getLocalPort(); + int localPort = ThriftServer.getBindingPort(thriftServer.getServerTransport()); if (localPort == 0) { continue; } else { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/8674d9dc/blur-core/src/test/java/org/apache/blur/MiniCluster.java ---------------------------------------------------------------------- diff --git a/blur-core/src/test/java/org/apache/blur/MiniCluster.java b/blur-core/src/test/java/org/apache/blur/MiniCluster.java index 542cc5f..601548b 100644 --- a/blur-core/src/test/java/org/apache/blur/MiniCluster.java +++ b/blur-core/src/test/java/org/apache/blur/MiniCluster.java @@ -587,7 +587,7 @@ public class MiniCluster { public void kill() { ZooKeeper zk = null; try { - int shardPort = thriftServer.getServerTransport().getServerSocket().getLocalPort(); + int shardPort = ThriftServer.getBindingPort(thriftServer.getServerTransport()); String nodeNameHostname = ThriftServer.getNodeName(configuration, BLUR_SHARD_HOSTNAME); String nodeName = nodeNameHostname + ":" + shardPort; zk = new ZooKeeperClient(getZkConnectionString(), 30000, new Watcher() { @@ -621,6 +621,8 @@ public class MiniCluster { } catch (TTransportException e) { LOG.error(e); throw new RuntimeException(e); + } catch (IOException e) { + throw new RuntimeException(e); } } @@ -632,7 +634,7 @@ public class MiniCluster { } catch (InterruptedException e) { return; } - int localPort = thriftServer.getLocalPort(); + int localPort = ThriftServer.getBindingPort(thriftServer.getServerTransport()); if (localPort == 0) { continue; } else { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/8674d9dc/blur-thrift/src/main/java/org/apache/blur/thirdparty/thrift_0_9_0/server/TThreadPoolServer.java ---------------------------------------------------------------------- diff --git a/blur-thrift/src/main/java/org/apache/blur/thirdparty/thrift_0_9_0/server/TThreadPoolServer.java b/blur-thrift/src/main/java/org/apache/blur/thirdparty/thrift_0_9_0/server/TThreadPoolServer.java index 07cd950..ff6a659 100644 --- a/blur-thrift/src/main/java/org/apache/blur/thirdparty/thrift_0_9_0/server/TThreadPoolServer.java +++ b/blur-thrift/src/main/java/org/apache/blur/thirdparty/thrift_0_9_0/server/TThreadPoolServer.java @@ -193,7 +193,7 @@ public class TThreadPoolServer extends TServer { eventHandler = getEventHandler(); if (eventHandler != null) { - connectionContext = eventHandler.createContext(inputProtocol, outputProtocol, null); + connectionContext = eventHandler.createContext(inputProtocol, outputProtocol, client_); } // we check stopped_ first to make sure we're not supposed to be shutting // down. this is necessary for graceful shutdown. http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/8674d9dc/blur-thrift/src/main/java/org/apache/blur/thrift/BlurClient.java ---------------------------------------------------------------------- diff --git a/blur-thrift/src/main/java/org/apache/blur/thrift/BlurClient.java b/blur-thrift/src/main/java/org/apache/blur/thrift/BlurClient.java index ad7b013..3987bef 100644 --- a/blur-thrift/src/main/java/org/apache/blur/thrift/BlurClient.java +++ b/blur-thrift/src/main/java/org/apache/blur/thrift/BlurClient.java @@ -296,4 +296,9 @@ public class BlurClient { return getClient(blurConfiguration); } + public static void init(BlurConfiguration configuration) { + _defaultBlurConfiguration = configuration; + ClientPool clientPool = BlurClientManager.getClientPool(); + clientPool.setBlurConfiguration(_defaultBlurConfiguration); + } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/8674d9dc/blur-thrift/src/main/java/org/apache/blur/thrift/ClientPool.java ---------------------------------------------------------------------- diff --git a/blur-thrift/src/main/java/org/apache/blur/thrift/ClientPool.java b/blur-thrift/src/main/java/org/apache/blur/thrift/ClientPool.java index 68ac9b4..9cafd96 100644 --- a/blur-thrift/src/main/java/org/apache/blur/thrift/ClientPool.java +++ b/blur-thrift/src/main/java/org/apache/blur/thrift/ClientPool.java @@ -18,8 +18,8 @@ package org.apache.blur.thrift; */ import static org.apache.blur.utils.BlurConstants.BLUR_CLIENTPOOL_CLIENT_CLEAN_FREQUENCY; -import static org.apache.blur.utils.BlurConstants.BLUR_CLIENTPOOL_CLIENT_STALE_THRESHOLD; import static org.apache.blur.utils.BlurConstants.BLUR_CLIENTPOOL_CLIENT_MAX_CONNECTIONS_PER_HOST; +import static org.apache.blur.utils.BlurConstants.BLUR_CLIENTPOOL_CLIENT_STALE_THRESHOLD; import static org.apache.blur.utils.BlurConstants.BLUR_THRIFT_MAX_FRAME_SIZE; import java.io.IOException; @@ -42,12 +42,15 @@ import org.apache.blur.log.Log; import org.apache.blur.log.LogFactory; import org.apache.blur.thirdparty.thrift_0_9_0.TException; import org.apache.blur.thirdparty.thrift_0_9_0.protocol.TBinaryProtocol; +import org.apache.blur.thirdparty.thrift_0_9_0.protocol.TCompactProtocol; import org.apache.blur.thirdparty.thrift_0_9_0.protocol.TProtocol; import org.apache.blur.thirdparty.thrift_0_9_0.transport.TFramedTransport; import org.apache.blur.thirdparty.thrift_0_9_0.transport.TSocket; +import org.apache.blur.thirdparty.thrift_0_9_0.transport.TTransport; import org.apache.blur.thirdparty.thrift_0_9_0.transport.TTransportException; import org.apache.blur.thrift.generated.Blur.Client; import org.apache.blur.thrift.generated.SafeClientGen; +import org.apache.blur.thrift.sasl.SaslHelper; public class ClientPool { @@ -59,26 +62,34 @@ public class ClientPool { private static final long _idleTimeBeforeClosingClient; private static final long _clientPoolCleanFrequency; private static final Timer _master; + private static final BlurConfiguration _configurationFromClassPath; + private BlurConfiguration _configuration = _configurationFromClassPath; static { try { - BlurConfiguration config = new BlurConfiguration(); - int maxConnectionsPerHost = config.getInt(BLUR_CLIENTPOOL_CLIENT_MAX_CONNECTIONS_PER_HOST, 64); + _configurationFromClassPath = new BlurConfiguration(); + int maxConnectionsPerHost = _configurationFromClassPath.getInt(BLUR_CLIENTPOOL_CLIENT_MAX_CONNECTIONS_PER_HOST, + 64); if (maxConnectionsPerHost < 1) { LOG.fatal("Max connections per host cannot be less than 1 current value [{0}] using 1.", maxConnectionsPerHost); maxConnectionsPerHost = 1; } _maxConnectionsPerHost = maxConnectionsPerHost; - _idleTimeBeforeClosingClient = TimeUnit.SECONDS.toNanos(config - .getLong(BLUR_CLIENTPOOL_CLIENT_STALE_THRESHOLD, 30)); - _clientPoolCleanFrequency = TimeUnit.SECONDS.toMillis(config.getLong(BLUR_CLIENTPOOL_CLIENT_CLEAN_FREQUENCY, 10)); - _maxFrameSize = config.getInt(BLUR_THRIFT_MAX_FRAME_SIZE, 16384000); + _idleTimeBeforeClosingClient = TimeUnit.SECONDS.toNanos(_configurationFromClassPath.getLong( + BLUR_CLIENTPOOL_CLIENT_STALE_THRESHOLD, 30)); + _clientPoolCleanFrequency = TimeUnit.SECONDS.toMillis(_configurationFromClassPath.getLong( + BLUR_CLIENTPOOL_CLIENT_CLEAN_FREQUENCY, 10)); + _maxFrameSize = _configurationFromClassPath.getInt(BLUR_THRIFT_MAX_FRAME_SIZE, 16384000); } catch (Exception e) { throw new RuntimeException(e); } _master = checkAndRemoveStaleClients(); } + public void setBlurConfiguration(BlurConfiguration configuration) { + _configuration = configuration; + } + public static void close() { _master.cancel(); _master.purge(); @@ -246,21 +257,32 @@ public class ClientPool { public Client newClient(Connection connection) throws TTransportException, IOException { String host = connection.getHost(); int port = connection.getPort(); - TSocket trans; + + TProtocol proto; Socket socket; - if (connection.isProxy()) { - Proxy proxy = new Proxy(Type.SOCKS, new InetSocketAddress(connection.getProxyHost(), connection.getProxyPort())); - socket = new Socket(proxy); + int timeout = connection.getTimeout(); + if (SaslHelper.isSaslEnabled(_configuration)) { + if (connection.isProxy()) { + throw new IOException("Proxy connections are not allowed when SASL is enabled."); + } + TSocket transport = new TSocket(host, port, timeout); + TTransport tSaslClientTransport = SaslHelper.getTSaslClientTransport(_configuration, transport); + tSaslClientTransport.open(); + socket = transport.getSocket(); + proto = new TCompactProtocol(tSaslClientTransport); } else { - socket = new Socket(); + if (connection.isProxy()) { + Proxy proxy = new Proxy(Type.SOCKS, new InetSocketAddress(connection.getProxyHost(), connection.getProxyPort())); + socket = new Socket(proxy); + } else { + socket = new Socket(); + } + socket.setTcpNoDelay(true); + socket.setSoTimeout(timeout); + socket.connect(new InetSocketAddress(host, port), timeout); + TSocket trans = new TSocket(socket); + proto = new TBinaryProtocol(new TFramedTransport(trans, _maxFrameSize)); } - int timeout = connection.getTimeout(); - socket.setTcpNoDelay(true); - socket.setSoTimeout(timeout); - socket.connect(new InetSocketAddress(host, port), timeout); - trans = new TSocket(socket); - - TProtocol proto = new TBinaryProtocol(new TFramedTransport(trans, _maxFrameSize)); return new WeightedClient(proto, getIdentifer(socket)); } @@ -291,4 +313,5 @@ public class ClientPool { LOG.error("Error during closing of client [{0}].", client); } } + } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/8674d9dc/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/AnonymousAuthenticationProviderImpl.java ---------------------------------------------------------------------- diff --git a/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/AnonymousAuthenticationProviderImpl.java b/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/AnonymousAuthenticationProviderImpl.java new file mode 100644 index 0000000..c314774 --- /dev/null +++ b/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/AnonymousAuthenticationProviderImpl.java @@ -0,0 +1,36 @@ +/** + * 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.blur.thrift.sasl; + +import java.net.InetSocketAddress; + +import javax.security.sasl.AuthenticationException; + +import org.apache.blur.BlurConfiguration; + +public class AnonymousAuthenticationProviderImpl extends PasswordAuthenticationProvider { + + public AnonymousAuthenticationProviderImpl(BlurConfiguration configuration) { + super(configuration); + } + + @Override + public void authenticate(String username, String password, InetSocketAddress address) throws AuthenticationException { + System.out.println("Username [" + username + "] Password [" + password + "]"); + } + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/8674d9dc/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/AuthenticationType.java ---------------------------------------------------------------------- diff --git a/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/AuthenticationType.java b/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/AuthenticationType.java new file mode 100644 index 0000000..079e4bb --- /dev/null +++ b/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/AuthenticationType.java @@ -0,0 +1,21 @@ +/** + * 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.blur.thrift.sasl; + +public enum AuthenticationType { + ANONYMOUS, CUSTOM +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/8674d9dc/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/CustomAuthenticationProviderImpl.java ---------------------------------------------------------------------- diff --git a/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/CustomAuthenticationProviderImpl.java b/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/CustomAuthenticationProviderImpl.java new file mode 100644 index 0000000..7090fa9 --- /dev/null +++ b/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/CustomAuthenticationProviderImpl.java @@ -0,0 +1,56 @@ +/** + * 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.blur.thrift.sasl; + +import java.lang.reflect.Constructor; +import java.net.InetSocketAddress; + +import javax.security.sasl.AuthenticationException; + +import org.apache.blur.BlurConfiguration; +import org.apache.blur.log.Log; +import org.apache.blur.log.LogFactory; + +public class CustomAuthenticationProviderImpl extends PasswordAuthenticationProvider { + + private static final Log LOG = LogFactory.getLog(CustomAuthenticationProviderImpl.class); + private static final String BLUR_SECUTIRY_SASL_CUSTOM_CLASS = "blur.security.sasl.CUSTOM.class"; + + private final PasswordAuthenticationProvider _provider; + + @SuppressWarnings("unchecked") + public CustomAuthenticationProviderImpl(BlurConfiguration configuration) { + super(configuration); + String className = configuration.get(BLUR_SECUTIRY_SASL_CUSTOM_CLASS); + LOG.info("Custom provider using class [{0}]", className); + try { + Class<? extends PasswordAuthenticationProvider> clazz = (Class<? extends PasswordAuthenticationProvider>) Class + .forName(className); + Constructor<? extends PasswordAuthenticationProvider> constructor = clazz + .getConstructor(new Class[] { BlurConfiguration.class }); + _provider = constructor.newInstance(configuration); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + @Override + public void authenticate(String username, String password, InetSocketAddress address) throws AuthenticationException { + _provider.authenticate(username, password, address); + } + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/8674d9dc/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/PasswordAuthenticationProvider.java ---------------------------------------------------------------------- diff --git a/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/PasswordAuthenticationProvider.java b/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/PasswordAuthenticationProvider.java new file mode 100644 index 0000000..304050b --- /dev/null +++ b/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/PasswordAuthenticationProvider.java @@ -0,0 +1,17 @@ +package org.apache.blur.thrift.sasl; + +import java.net.InetSocketAddress; + +import javax.security.sasl.AuthenticationException; + +import org.apache.blur.BlurConfiguration; + +public abstract class PasswordAuthenticationProvider { + + public PasswordAuthenticationProvider(BlurConfiguration configuration) { + + } + + public abstract void authenticate(String username, String password, InetSocketAddress address) throws AuthenticationException; + +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/8674d9dc/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/PlainSaslServer.java ---------------------------------------------------------------------- diff --git a/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/PlainSaslServer.java b/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/PlainSaslServer.java new file mode 100644 index 0000000..01debd9 --- /dev/null +++ b/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/PlainSaslServer.java @@ -0,0 +1,165 @@ +/** + * 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.blur.thrift.sasl; + +import java.io.IOException; +import java.util.ArrayDeque; +import java.util.Deque; +import java.util.Map; + +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.sasl.AuthorizeCallback; +import javax.security.sasl.SaslException; +import javax.security.sasl.SaslServer; +import javax.security.sasl.SaslServerFactory; + + +/** + * PlainSaslServer. + * + * This code originated in the Apache Hive Project. + * + * Sun JDK only provides PLAIN client and not server. This class implements the Plain SASL server + * conforming to RFC #4616 (http://www.ietf.org/rfc/rfc4616.txt) + */ +public class PlainSaslServer implements SaslServer { + private String _user; + private String _passwd; + private String _authzId; + private final CallbackHandler _handler; + + PlainSaslServer(CallbackHandler handler, String authMethodStr) throws SaslException { + _handler = handler; + } + + public String getMechanismName() { + return "PLAIN"; + } + + public byte[] evaluateResponse(byte[] response) throws SaslException { + try { + // parse the response + // message = [authzid] UTF8NUL authcid UTF8NUL passwd' + + Deque<String> tokenList = new ArrayDeque<String>(); + StringBuilder messageToken = new StringBuilder(); + for (byte b : response) { + if (b == 0) { + tokenList.addLast(messageToken.toString()); + messageToken = new StringBuilder(); + } else { + messageToken.append((char)b); + } + } + tokenList.addLast(messageToken.toString()); + + // validate response + if ((tokenList.size() < 2) || (tokenList.size() > 3)) { + throw new SaslException("Invalid message format"); + } + _passwd = tokenList.removeLast(); + _user = tokenList.removeLast(); + // optional authzid + if (!tokenList.isEmpty()) { + _authzId = tokenList.removeLast(); + } else { + _authzId = _user; + } + if (_user == null || _user.isEmpty()) { + throw new SaslException("No user name provide"); + } + if (_passwd == null || _passwd.isEmpty()) { + throw new SaslException("No password name provide"); + } + + NameCallback nameCallback = new NameCallback("User"); + nameCallback.setName(_user); + PasswordCallback pcCallback = new PasswordCallback("Password", false); + pcCallback.setPassword(_passwd.toCharArray()); + AuthorizeCallback acCallback = new AuthorizeCallback(_user, _authzId); + + Callback[] cbList = new Callback[] {nameCallback, pcCallback, acCallback}; + _handler.handle(cbList); + if (!acCallback.isAuthorized()) { + throw new SaslException("Authentication failed"); + } + } catch (IllegalStateException eL) { + throw new SaslException("Invalid message format", eL); + } catch (IOException eI) { + throw new SaslException("Error validating the login", eI); + } catch (UnsupportedCallbackException eU) { + throw new SaslException("Error validating the login", eU); + } + return null; + } + + public boolean isComplete() { + return _user != null; + } + + public String getAuthorizationID() { + return _user; + } + + public byte[] unwrap(byte[] incoming, int offset, int len) { + throw new UnsupportedOperationException(); + } + + public byte[] wrap(byte[] outgoing, int offset, int len) { + throw new UnsupportedOperationException(); + } + + public Object getNegotiatedProperty(String propName) { + return null; + } + + public void dispose() {} + + public static class SaslPlainServerFactory implements SaslServerFactory { + + public SaslServer createSaslServer( + String mechanism, String protocol, String serverName, Map<String,?> props, CallbackHandler cbh) + { + if ("PLAIN".equals(mechanism)) { + try { + return new PlainSaslServer(cbh, protocol); + } catch (SaslException e) { + return null; + } + } + return null; + } + + public String[] getMechanismNames(Map<String, ?> props) { + return new String[] { "PLAIN" }; + } + } + + @SuppressWarnings("serial") + public static class SaslPlainProvider extends java.security.Provider { + public SaslPlainProvider() { + super("BlurSaslPlain", 1.0, "Blur Plain SASL provider"); + put("SaslServerFactory.PLAIN", SaslPlainServerFactory.class.getName()); + } + } +} + + http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/8674d9dc/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/SaslHelper.java ---------------------------------------------------------------------- diff --git a/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/SaslHelper.java b/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/SaslHelper.java new file mode 100644 index 0000000..343c22f --- /dev/null +++ b/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/SaslHelper.java @@ -0,0 +1,203 @@ +/** + * 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.blur.thrift.sasl; + +import static org.apache.blur.utils.BlurConstants.BLUR_SECURITY_SASL_ENABLED; +import static org.apache.blur.utils.BlurConstants.BLUR_SECURITY_SASL_TYPE; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; + +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.sasl.AuthorizeCallback; + +import org.apache.blur.BlurConfiguration; +import org.apache.blur.log.Log; +import org.apache.blur.log.LogFactory; +import org.apache.blur.thirdparty.thrift_0_9_0.transport.TTransport; + +public class SaslHelper { + + private static final Log LOG = LogFactory.getLog(SaslHelper.class); + + private static final String BLUR_SECURITY_SASL_PLAIN_PASSWORD = "blur.security.sasl.plain.password"; + private static final String BLUR_SECURITY_SASL_PLAIN_USERNAME = "blur.security.sasl.plain.username"; + private static final String CUSTOM = "CUSTOM"; + private static final String PLAIN = "PLAIN"; + + static { + java.security.Security.addProvider(new PlainSaslServer.SaslPlainProvider()); + } + + public static TTransport getTSaslClientTransport(BlurConfiguration configuration, TTransport transport) + throws IOException { + AuthenticationType type = getValueOf(configuration.get(BLUR_SECURITY_SASL_TYPE)); + switch (type) { + case ANONYMOUS: + case CUSTOM: + return getPlainTSaslClientTransport(type, configuration, transport); + default: + throw new IOException("Type [" + type + "] not supported."); + } + } + + private static TTransport getPlainTSaslClientTransport(AuthenticationType type, BlurConfiguration configuration, + TTransport transport) throws IOException { + final String username; + final String password; + switch (type) { + case ANONYMOUS: { + username = "anonymous"; + password = "anonymous"; + break; + } + case CUSTOM: { + username = configuration.get(BLUR_SECURITY_SASL_PLAIN_USERNAME); + password = configuration.get(BLUR_SECURITY_SASL_PLAIN_PASSWORD); + break; + } + default: + throw new IOException("Type [" + type + "] not supported."); + } + if (username == null) { + throw new IOException("Username cannot be null set property [" + BLUR_SECURITY_SASL_PLAIN_USERNAME + + "] in the BlurConfiguration."); + } + if (password == null) { + throw new IOException("Password cannot be null set property [" + BLUR_SECURITY_SASL_PLAIN_PASSWORD + + "] in the BlurConfiguration."); + } + final String authorizationId = null; + final String serverName = null; + Map<String, String> props = new HashMap<String, String>(); + CallbackHandler cbh = new CallbackHandler() { + @Override + public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException { + for (int i = 0; i < callbacks.length; i++) { + if (callbacks[i] instanceof NameCallback) { + NameCallback nameCallback = (NameCallback) callbacks[i]; + nameCallback.setName(username); + } else if (callbacks[i] instanceof PasswordCallback) { + PasswordCallback passCallback = (PasswordCallback) callbacks[i]; + passCallback.setPassword(password.toCharArray()); + } else { + throw new UnsupportedCallbackException(callbacks[i]); + } + } + } + + }; + return new TSaslClientTransport(PLAIN, authorizationId, CUSTOM, serverName, props, cbh, transport); + } + + public static TSaslServerTransport.Factory getTSaslServerTransportFactory(BlurConfiguration configuration) + throws IOException { + AuthenticationType type = getValueOf(configuration.get(BLUR_SECURITY_SASL_TYPE)); + LOG.info("Setting SASL Server with authentication type [{0}]", type); + switch (type) { + case ANONYMOUS: + case CUSTOM: + return getPlainTSaslServerTransportFactory(type, configuration); + default: + throw new IOException("Type [" + type + "] not supported."); + } + } + + private static AuthenticationType getValueOf(String t) throws IOException { + if (t == null) { + return null; + } + String upperCase = t.toUpperCase(); + AuthenticationType[] values = AuthenticationType.values(); + for (AuthenticationType type : values) { + if (type.name().equals(upperCase)) { + return type; + } + } + throw new IOException("Type [" + t + "] not found. Choose from [" + Arrays.asList(AuthenticationType.values()) + + "]"); + } + + private static TSaslServerTransport.Factory getPlainTSaslServerTransportFactory(final AuthenticationType type, + final BlurConfiguration configuration) throws IOException { + TSaslServerTransport.Factory saslTransportFactory = new TSaslServerTransport.Factory(); + final String serverName = null; + final Map<String, String> props = new HashMap<String, String>(); + final PasswordAuthenticationProvider provider = getAuthenticationProvider(type, configuration); + CallbackHandler cbh = new CallbackHandler() { + + @Override + public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException { + String username = null; + String password = null; + AuthorizeCallback ac = null; + for (int i = 0; i < callbacks.length; i++) { + if (callbacks[i] instanceof NameCallback) { + NameCallback nc = (NameCallback) callbacks[i]; + username = nc.getName(); + } else if (callbacks[i] instanceof PasswordCallback) { + PasswordCallback pc = (PasswordCallback) callbacks[i]; + password = new String(pc.getPassword()); + } else if (callbacks[i] instanceof AuthorizeCallback) { + ac = (AuthorizeCallback) callbacks[i]; + } else { + throw new UnsupportedCallbackException(callbacks[i]); + } + } + InetSocketAddress address = TSaslTransport._currentConnection.get(); + provider.authenticate(username, password, address); + if (ac != null) { + ac.setAuthorized(true); + } + } + }; + saslTransportFactory.addServerDefinition(PLAIN, CUSTOM, serverName, props, cbh); + return saslTransportFactory; + } + + public static PasswordAuthenticationProvider getAuthenticationProvider(AuthenticationType type, + BlurConfiguration configuration) throws IOException { + LOG.info("Setting SASL Server with password authentication provider [{0}]", type); + switch (type) { + case ANONYMOUS: + return new AnonymousAuthenticationProviderImpl(configuration); + case CUSTOM: + return new CustomAuthenticationProviderImpl(configuration); + default: + throw new IOException("Unsupported authentication method [" + type + "]"); + } + } + + public static void setPlainUsername(BlurConfiguration configuration, String username) { + configuration.set(BLUR_SECURITY_SASL_PLAIN_USERNAME, username); + } + + public static void setPlainPassword(BlurConfiguration configuration, String password) { + configuration.set(BLUR_SECURITY_SASL_PLAIN_PASSWORD, password); + } + + public static boolean isSaslEnabled(BlurConfiguration configuration) { + return configuration.getBoolean(BLUR_SECURITY_SASL_ENABLED, false); + } +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/8674d9dc/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/TSaslClientTransport.java ---------------------------------------------------------------------- diff --git a/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/TSaslClientTransport.java b/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/TSaslClientTransport.java new file mode 100644 index 0000000..5407d63 --- /dev/null +++ b/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/TSaslClientTransport.java @@ -0,0 +1,109 @@ +/* + * 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.blur.thrift.sasl; + +import java.util.Map; + +import javax.security.auth.callback.CallbackHandler; +import javax.security.sasl.Sasl; +import javax.security.sasl.SaslClient; +import javax.security.sasl.SaslException; + +import org.apache.blur.thirdparty.thrift_0_9_0.transport.TTransport; +import org.apache.blur.thirdparty.thrift_0_9_0.transport.TTransportException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Wraps another Thrift <code>TTransport</code>, but performs SASL client + * negotiation on the call to <code>open()</code>. This class will wrap ensuing + * communication over it, if a SASL QOP is negotiated with the other party. + */ +public class TSaslClientTransport extends TSaslTransport { + + private static final Logger LOGGER = LoggerFactory.getLogger(TSaslClientTransport.class); + + /** + * The name of the mechanism this client supports. + */ + private final String mechanism; + + /** + * Uses the given <code>SaslClient</code>. + * + * @param saslClient + * The <code>SaslClient</code> to use for the subsequent SASL + * negotiation. + * @param transport + * Transport underlying this one. + */ + public TSaslClientTransport(SaslClient saslClient, TTransport transport) { + super(saslClient, transport); + mechanism = saslClient.getMechanismName(); + } + + /** + * Creates a <code>SaslClient</code> using the given SASL-specific parameters. + * See the Java documentation for <code>Sasl.createSaslClient</code> for the + * details of the parameters. + * + * @param transport + * The underlying Thrift transport. + * @throws SaslException + */ + public TSaslClientTransport(String mechanism, String authorizationId, String protocol, + String serverName, Map<String, String> props, CallbackHandler cbh, TTransport transport) + throws SaslException { + super(Sasl.createSaslClient(new String[] { mechanism }, authorizationId, protocol, serverName, + props, cbh), transport); + this.mechanism = mechanism; + } + + + @Override + protected SaslRole getRole() { + return SaslRole.CLIENT; + } + + /** + * Performs the client side of the initial portion of the Thrift SASL + * protocol. Generates and sends the initial response to the server, including + * which mechanism this client wants to use. + */ + @Override + protected void handleSaslStartMessage() throws TTransportException, SaslException { + SaslClient saslClient = getSaslClient(); + + byte[] initialResponse = new byte[0]; + if (saslClient.hasInitialResponse()) + initialResponse = saslClient.evaluateChallenge(initialResponse); + + LOGGER.debug("Sending mechanism name {} and initial response of length {}", mechanism, + initialResponse.length); + + byte[] mechanismBytes = mechanism.getBytes(); + sendSaslMessage(NegotiationStatus.START, + mechanismBytes); + // Send initial response + sendSaslMessage(saslClient.isComplete() ? NegotiationStatus.COMPLETE : NegotiationStatus.OK, + initialResponse); + underlyingTransport.flush(); + } +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/8674d9dc/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/TSaslServerTransport.java ---------------------------------------------------------------------- diff --git a/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/TSaslServerTransport.java b/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/TSaslServerTransport.java new file mode 100644 index 0000000..e05e121 --- /dev/null +++ b/blur-thrift/src/main/java/org/apache/blur/thrift/sasl/TSaslServerTransport.java @@ -0,0 +1,231 @@ +/* + * 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.blur.thrift.sasl; +import java.lang.ref.WeakReference; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.WeakHashMap; + +import javax.security.auth.callback.CallbackHandler; +import javax.security.sasl.Sasl; +import javax.security.sasl.SaslException; +import javax.security.sasl.SaslServer; + +import org.apache.blur.thirdparty.thrift_0_9_0.transport.TTransport; +import org.apache.blur.thirdparty.thrift_0_9_0.transport.TTransportException; +import org.apache.blur.thirdparty.thrift_0_9_0.transport.TTransportFactory; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Wraps another Thrift <code>TTransport</code>, but performs SASL server + * negotiation on the call to <code>open()</code>. This class will wrap ensuing + * communication over it, if a SASL QOP is negotiated with the other party. + */ +public class TSaslServerTransport extends TSaslTransport { + + private static final Logger LOGGER = LoggerFactory.getLogger(TSaslServerTransport.class); + + /** + * Mapping from SASL mechanism name -> all the parameters required to + * instantiate a SASL server. + */ + private Map<String, TSaslServerDefinition> serverDefinitionMap = new HashMap<String, TSaslServerDefinition>(); + + /** + * Contains all the parameters used to define a SASL server implementation. + */ + private static class TSaslServerDefinition { + public String mechanism; + public String protocol; + public String serverName; + public Map<String, String> props; + public CallbackHandler cbh; + + public TSaslServerDefinition(String mechanism, String protocol, String serverName, + Map<String, String> props, CallbackHandler cbh) { + this.mechanism = mechanism; + this.protocol = protocol; + this.serverName = serverName; + this.props = props; + this.cbh = cbh; + } + } + + /** + * Uses the given underlying transport. Assumes that addServerDefinition is + * called later. + * + * @param transport + * Transport underlying this one. + */ + public TSaslServerTransport(TTransport transport) { + super(transport); + } + + /** + * Creates a <code>SaslServer</code> using the given SASL-specific parameters. + * See the Java documentation for <code>Sasl.createSaslServer</code> for the + * details of the parameters. + * + * @param transport + * The underlying Thrift transport. + */ + public TSaslServerTransport(String mechanism, String protocol, String serverName, + Map<String, String> props, CallbackHandler cbh, TTransport transport) { + super(transport); + addServerDefinition(mechanism, protocol, serverName, props, cbh); + } + + private TSaslServerTransport(Map<String, TSaslServerDefinition> serverDefinitionMap, TTransport transport) { + super(transport); + this.serverDefinitionMap.putAll(serverDefinitionMap); + } + + /** + * Add a supported server definition to this transport. See the Java + * documentation for <code>Sasl.createSaslServer</code> for the details of the + * parameters. + */ + public void addServerDefinition(String mechanism, String protocol, String serverName, + Map<String, String> props, CallbackHandler cbh) { + serverDefinitionMap.put(mechanism, new TSaslServerDefinition(mechanism, protocol, serverName, + props, cbh)); + } + + @Override + protected SaslRole getRole() { + return SaslRole.SERVER; + } + + /** + * Performs the server side of the initial portion of the Thrift SASL protocol. + * Receives the initial response from the client, creates a SASL server using + * the mechanism requested by the client (if this server supports it), and + * sends the first challenge back to the client. + */ + @Override + protected void handleSaslStartMessage() throws TTransportException, SaslException { + SaslResponse message = receiveSaslMessage(); + + LOGGER.debug("Received start message with status {}", message.status); + if (message.status != NegotiationStatus.START) { + sendAndThrowMessage(NegotiationStatus.ERROR, "Expecting START status, received " + message.status); + } + + // Get the mechanism name. + String mechanismName = new String(message.payload); + TSaslServerDefinition serverDefinition = serverDefinitionMap.get(mechanismName); + LOGGER.debug("Received mechanism name '{}'", mechanismName); + + if (serverDefinition == null) { + sendAndThrowMessage(NegotiationStatus.BAD, "Unsupported mechanism type " + mechanismName); + } + SaslServer saslServer = Sasl.createSaslServer(serverDefinition.mechanism, + serverDefinition.protocol, serverDefinition.serverName, serverDefinition.props, + serverDefinition.cbh); + setSaslServer(saslServer); + } + + /** + * <code>TTransportFactory</code> to create + * <code>TSaslServerTransports<c/ode>. Ensures that a given + * underlying <code>TTransport</code> instance receives the same + * <code>TSaslServerTransport</code>. This is kind of an awful hack to work + * around the fact that Thrift is designed assuming that + * <code>TTransport</code> instances are stateless, and thus the existing + * <code>TServers</code> use different <code>TTransport</code> instances for + * input and output. + */ + public static class Factory extends TTransportFactory { + + /** + * This is the implementation of the awful hack described above. + * <code>WeakHashMap</code> is used to ensure that we don't leak memory. + */ + private static Map<TTransport, WeakReference<TSaslServerTransport>> transportMap = + Collections.synchronizedMap(new WeakHashMap<TTransport, WeakReference<TSaslServerTransport>>()); + + /** + * Mapping from SASL mechanism name -> all the parameters required to + * instantiate a SASL server. + */ + private Map<String, TSaslServerDefinition> serverDefinitionMap = new HashMap<String, TSaslServerDefinition>(); + + /** + * Create a new Factory. Assumes that <code>addServerDefinition</code> will + * be called later. + */ + public Factory() { + super(); + } + + /** + * Create a new <code>Factory</code>, initially with the single server + * definition given. You may still call <code>addServerDefinition</code> + * later. See the Java documentation for <code>Sasl.createSaslServer</code> + * for the details of the parameters. + */ + public Factory(String mechanism, String protocol, String serverName, + Map<String, String> props, CallbackHandler cbh) { + super(); + addServerDefinition(mechanism, protocol, serverName, props, cbh); + } + + /** + * Add a supported server definition to the transports created by this + * factory. See the Java documentation for + * <code>Sasl.createSaslServer</code> for the details of the parameters. + */ + public void addServerDefinition(String mechanism, String protocol, String serverName, + Map<String, String> props, CallbackHandler cbh) { + serverDefinitionMap.put(mechanism, new TSaslServerDefinition(mechanism, protocol, serverName, + props, cbh)); + } + + /** + * Get a new <code>TSaslServerTransport</code> instance, or reuse the + * existing one if a <code>TSaslServerTransport</code> has already been + * created before using the given <code>TTransport</code> as an underlying + * transport. This ensures that a given underlying transport instance + * receives the same <code>TSaslServerTransport</code>. + */ + @Override + public TTransport getTransport(TTransport base) { + WeakReference<TSaslServerTransport> ret = transportMap.get(base); + if (ret == null || ret.get() == null) { + LOGGER.debug("transport map does not contain key", base); + ret = new WeakReference<TSaslServerTransport>(new TSaslServerTransport(serverDefinitionMap, base)); + try { + ret.get().open(); + } catch (TTransportException e) { + LOGGER.debug("failed to open server transport", e); + throw new RuntimeException(e); + } + transportMap.put(base, ret); // No need for putIfAbsent(). + // Concurrent calls to getTransport() will pass in different TTransports. + } else { + LOGGER.debug("transport map does contain key {}", base); + } + return ret.get(); + } + } +}
