From b133d52043a5f97c019d796a27ded81523120428 Mon Sep 17 00:00:00 2001 From: Kishor Patil Date: Wed, 2 Mar 2016 22:46:48 -0600 Subject: [PATCH 1/2] Remove SimpleTransportPlugin and configurations --- conf/defaults.yaml | 4 +- .../src/jvm/org/apache/storm/Config.java | 14 -- .../security/auth/SimpleTransportPlugin.java | 161 ------------------ .../security/auth/ThriftConnectionType.java | 13 +- .../testing/SingleUserSimpleTransport.java | 37 ---- .../apache/storm/security/auth/auth_test.clj | 25 +-- .../storm/security/auth/nimbus_auth_test.clj | 12 +- 7 files changed, 24 insertions(+), 242 deletions(-) delete mode 100644 storm-core/src/jvm/org/apache/storm/security/auth/SimpleTransportPlugin.java delete mode 100644 storm-core/src/jvm/org/apache/storm/testing/SingleUserSimpleTransport.java diff --git a/conf/defaults.yaml b/conf/defaults.yaml index 98171615000..d2f634a9e32 100644 --- a/conf/defaults.yaml +++ b/conf/defaults.yaml @@ -39,7 +39,7 @@ storm.exhibitor.port: 8080 storm.exhibitor.poll.uripath: "/exhibitor/v1/cluster/list" storm.cluster.mode: "distributed" # can be distributed or local storm.local.mode.zmq: false -storm.thrift.transport: "org.apache.storm.security.auth.SimpleTransportPlugin" +storm.thrift.transport: "org.apache.storm.security.auth.plain.PlainSaslTransportPlugin" storm.principal.tolocal: "org.apache.storm.security.auth.DefaultPrincipalToLocal" storm.group.mapping.service: "org.apache.storm.security.auth.ShellBasedGroupsMapping" storm.group.mapping.service.params: null @@ -62,7 +62,6 @@ storm.health.check.timeout.ms: 5000 nimbus.seeds : ["localhost"] nimbus.thrift.port: 6627 nimbus.thrift.threads: 64 -nimbus.thrift.max_buffer_size: 1048576 nimbus.childopts: "-Xmx1024m" nimbus.task.timeout.secs: 30 nimbus.supervisor.timeout.secs: 60 @@ -102,7 +101,6 @@ logs.users: null drpc.port: 3772 drpc.worker.threads: 64 -drpc.max_buffer_size: 1048576 drpc.queue.size: 128 drpc.invocations.port: 3773 drpc.invocations.threads: 64 diff --git a/storm-core/src/jvm/org/apache/storm/Config.java b/storm-core/src/jvm/org/apache/storm/Config.java index 6ea8b0f5d22..eb0c1d4949d 100644 --- a/storm-core/src/jvm/org/apache/storm/Config.java +++ b/storm-core/src/jvm/org/apache/storm/Config.java @@ -549,13 +549,6 @@ public class Config extends HashMap { @isString public static final String NIMBUS_DAEMON_USER = "nimbus.daemon.user"; - /** - * The maximum buffer size thrift should use when reading messages. - */ - @isInteger - @isPositiveNumber - public static final String NIMBUS_THRIFT_MAX_BUFFER_SIZE = "nimbus.thrift.max_buffer_size"; - /** * This parameter is used by the storm-deploy project to configure the * jvm options for the nimbus daemon. @@ -1088,13 +1081,6 @@ public class Config extends HashMap { @isPositiveNumber public static final String DRPC_WORKER_THREADS = "drpc.worker.threads"; - /** - * The maximum buffer size thrift should use when reading messages for DRPC. - */ - @isNumber - @isPositiveNumber - public static final String DRPC_MAX_BUFFER_SIZE = "drpc.max_buffer_size"; - /** * DRPC thrift server queue size */ diff --git a/storm-core/src/jvm/org/apache/storm/security/auth/SimpleTransportPlugin.java b/storm-core/src/jvm/org/apache/storm/security/auth/SimpleTransportPlugin.java deleted file mode 100644 index 339bae9fe7b..00000000000 --- a/storm-core/src/jvm/org/apache/storm/security/auth/SimpleTransportPlugin.java +++ /dev/null @@ -1,161 +0,0 @@ -/** - * 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.storm.security.auth; - -import java.io.IOException; -import java.security.Principal; -import java.net.InetAddress; -import java.net.Socket; -import java.net.UnknownHostException; -import java.util.Map; -import java.util.HashSet; -import java.util.concurrent.ThreadPoolExecutor; -import java.util.concurrent.ArrayBlockingQueue; -import java.util.concurrent.TimeUnit; - -import javax.security.auth.login.Configuration; -import javax.security.auth.Subject; -import org.apache.thrift.TException; -import org.apache.thrift.TProcessor; -import org.apache.thrift.protocol.TBinaryProtocol; -import org.apache.thrift.protocol.TProtocol; -import org.apache.thrift.server.THsHaServer; -import org.apache.thrift.server.TServer; -import org.apache.thrift.transport.TFramedTransport; -import org.apache.thrift.transport.TMemoryInputTransport; -import org.apache.thrift.transport.TNonblockingServerSocket; -import org.apache.thrift.transport.TSocket; -import org.apache.thrift.transport.TTransport; -import org.apache.thrift.transport.TTransportException; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * Simple transport for Thrift plugin. - * - * This plugin is designed to be backward compatible with existing Storm code. - */ -public class SimpleTransportPlugin implements ITransportPlugin { - protected ThriftConnectionType type; - protected Map storm_conf; - protected Configuration login_conf; - private static final Logger LOG = LoggerFactory.getLogger(SimpleTransportPlugin.class); - - @Override - public void prepare(ThriftConnectionType type, Map storm_conf, Configuration login_conf) { - this.type = type; - this.storm_conf = storm_conf; - this.login_conf = login_conf; - } - - @Override - public TServer getServer(TProcessor processor) throws IOException, TTransportException { - int port = type.getPort(storm_conf); - TNonblockingServerSocket serverTransport = new TNonblockingServerSocket(port); - int numWorkerThreads = type.getNumThreads(storm_conf); - int maxBufferSize = type.getMaxBufferSize(storm_conf); - Integer queueSize = type.getQueueSize(storm_conf); - - THsHaServer.Args server_args = new THsHaServer.Args(serverTransport). - processor(new SimpleWrapProcessor(processor)). - maxWorkerThreads(numWorkerThreads). - protocolFactory(new TBinaryProtocol.Factory(false, true, maxBufferSize, -1)); - - if (queueSize != null) { - server_args.executorService(new ThreadPoolExecutor(numWorkerThreads, numWorkerThreads, - 60, TimeUnit.SECONDS, new ArrayBlockingQueue(queueSize))); - } - - //construct THsHaServer - return new THsHaServer(server_args); - } - - /** - * Connect to the specified server via framed transport - * @param transport The underlying Thrift transport. - * @param serverHost unused. - * @param asUser unused. - */ - @Override - public TTransport connect(TTransport transport, String serverHost, String asUser) throws TTransportException { - int maxBufferSize = type.getMaxBufferSize(storm_conf); - //create a framed transport - TTransport conn = new TFramedTransport(transport, maxBufferSize); - - //connect - conn.open(); - LOG.debug("Simple client transport has been established"); - - return conn; - } - - /** - * @return the subject that will be used for all connections - */ - protected Subject getDefaultSubject() { - return null; - } - - /** - * Processor that populate simple transport info into ReqContext, and then invoke a service handler - */ - private class SimpleWrapProcessor implements TProcessor { - final TProcessor wrapped; - - SimpleWrapProcessor(TProcessor wrapped) { - this.wrapped = wrapped; - } - - public boolean process(final TProtocol inProt, final TProtocol outProt) throws TException { - //populating request context - ReqContext req_context = ReqContext.context(); - - TTransport trans = inProt.getTransport(); - if (trans instanceof TMemoryInputTransport) { - try { - req_context.setRemoteAddress(InetAddress.getLocalHost()); - } catch (UnknownHostException e) { - throw new RuntimeException(e); - } - } else if (trans instanceof TSocket) { - TSocket tsocket = (TSocket)trans; - //remote address - Socket socket = tsocket.getSocket(); - req_context.setRemoteAddress(socket.getInetAddress()); - } - - //anonymous user - Subject s = getDefaultSubject(); - if (s == null) { - final String user = (String)storm_conf.get("debug.simple.transport.user"); - if (user != null) { - HashSet principals = new HashSet<>(); - principals.add(new Principal() { - public String getName() { return user; } - public String toString() { return user; } - }); - s = new Subject(true, principals, new HashSet<>(), new HashSet<>()); - } - } - req_context.setSubject(s); - - //invoke service handler - return wrapped.process(inProt, outProt); - } - } -} diff --git a/storm-core/src/jvm/org/apache/storm/security/auth/ThriftConnectionType.java b/storm-core/src/jvm/org/apache/storm/security/auth/ThriftConnectionType.java index 27db1430742..1addefbaa4c 100644 --- a/storm-core/src/jvm/org/apache/storm/security/auth/ThriftConnectionType.java +++ b/storm-core/src/jvm/org/apache/storm/security/auth/ThriftConnectionType.java @@ -27,25 +27,23 @@ */ public enum ThriftConnectionType { NIMBUS(Config.NIMBUS_THRIFT_TRANSPORT_PLUGIN, Config.NIMBUS_THRIFT_PORT, Config.NIMBUS_QUEUE_SIZE, - Config.NIMBUS_THRIFT_THREADS, Config.NIMBUS_THRIFT_MAX_BUFFER_SIZE), + Config.NIMBUS_THRIFT_THREADS), DRPC(Config.DRPC_THRIFT_TRANSPORT_PLUGIN, Config.DRPC_PORT, Config.DRPC_QUEUE_SIZE, - Config.DRPC_WORKER_THREADS, Config.DRPC_MAX_BUFFER_SIZE), + Config.DRPC_WORKER_THREADS), DRPC_INVOCATIONS(Config.DRPC_INVOCATIONS_THRIFT_TRANSPORT_PLUGIN, Config.DRPC_INVOCATIONS_PORT, null, - Config.DRPC_INVOCATIONS_THREADS, Config.DRPC_MAX_BUFFER_SIZE); + Config.DRPC_INVOCATIONS_THREADS); private final String _transConf; private final String _portConf; private final String _qConf; private final String _threadsConf; - private final String _buffConf; ThriftConnectionType(String transConf, String portConf, String qConf, - String threadsConf, String buffConf) { + String threadsConf) { _transConf = transConf; _portConf = portConf; _qConf = qConf; _threadsConf = threadsConf; - _buffConf = buffConf; } public String getTransportPlugin(Map conf) { @@ -71,7 +69,4 @@ public int getNumThreads(Map conf) { return Utils.getInt(conf.get(_threadsConf)); } - public int getMaxBufferSize(Map conf) { - return Utils.getInt(conf.get(_buffConf)); - } } diff --git a/storm-core/src/jvm/org/apache/storm/testing/SingleUserSimpleTransport.java b/storm-core/src/jvm/org/apache/storm/testing/SingleUserSimpleTransport.java deleted file mode 100644 index 0196a23aab5..00000000000 --- a/storm-core/src/jvm/org/apache/storm/testing/SingleUserSimpleTransport.java +++ /dev/null @@ -1,37 +0,0 @@ -/** - * 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.storm.testing; - -import org.apache.storm.security.auth.SimpleTransportPlugin; -import javax.security.auth.Subject; -import java.security.Principal; -import java.util.HashSet; - - -public class SingleUserSimpleTransport extends SimpleTransportPlugin { - @Override - protected Subject getDefaultSubject() { - HashSet principals = new HashSet(); - principals.add(new Principal() { - public String getName() { return "user"; } - public String toString() { return "user"; } - }); - return new Subject(true, principals, new HashSet(), new HashSet()); - } -} diff --git a/storm-core/test/clj/org/apache/storm/security/auth/auth_test.clj b/storm-core/test/clj/org/apache/storm/security/auth/auth_test.clj index 27f5816329b..f37778faae8 100644 --- a/storm-core/test/clj/org/apache/storm/security/auth/auth_test.clj +++ b/storm-core/test/clj/org/apache/storm/security/auth/auth_test.clj @@ -31,7 +31,8 @@ (:import [org.apache.storm.utils NimbusClient ConfigUtils]) (:import [org.apache.storm.security.auth.authorizer SimpleWhitelistAuthorizer SimpleACLAuthorizer]) (:import [org.apache.storm.security.auth AuthUtils ThriftServer ThriftClient ShellBasedGroupsMapping - ReqContext SimpleTransportPlugin KerberosPrincipalToLocal ThriftConnectionType]) + ReqContext KerberosPrincipalToLocal ThriftConnectionType]) + (:import [org.apache.storm.security.auth.plain PlainSaslTransportPlugin]) (:use [org.apache.storm util config]) (:use [org.apache.storm.daemon common]) (:use [org.apache.storm testing]) @@ -159,9 +160,9 @@ (deftest Simple-authentication-test (let [a-port (available-port)] - (with-server [a-port nil nil "org.apache.storm.security.auth.SimpleTransportPlugin" nil] + (with-server [a-port nil nil "org.apache.storm.security.auth.plain.PlainSaslTransportPlugin" nil] (let [storm-conf (merge (clojurify-structure (ConfigUtils/readStormConfig)) - {STORM-THRIFT-TRANSPORT-PLUGIN "org.apache.storm.security.auth.SimpleTransportPlugin"}) + {STORM-THRIFT-TRANSPORT-PLUGIN "org.apache.storm.security.auth.plain.PlainSaslTransportPlugin"}) client (NimbusClient. storm-conf "localhost" a-port nimbus-timeout) nimbus_client (.getClient client)] (.activate nimbus_client "security_auth_test_topology") @@ -179,9 +180,9 @@ (let [a-port (available-port)] (with-server [a-port nil "org.apache.storm.security.auth.authorizer.SimpleWhitelistAuthorizer" - "org.apache.storm.testing.SingleUserSimpleTransport" nil] + "org.apache.storm.security.auth.plain.PlainSaslTransportPlugin" nil] (let [storm-conf (merge (clojurify-structure (ConfigUtils/readStormConfig)) - {STORM-THRIFT-TRANSPORT-PLUGIN "org.apache.storm.testing.SingleUserSimpleTransport"}) + {STORM-THRIFT-TRANSPORT-PLUGIN "org.apache.storm.security.auth.plain.PlainSaslTransportPlugin"}) client (NimbusClient. storm-conf "localhost" a-port nimbus-timeout) nimbus_client (.getClient client)] (testing "(Negative authorization) Authorization plugin should reject client request" @@ -193,9 +194,9 @@ (let [a-port (available-port)] (with-server [a-port nil "org.apache.storm.security.auth.authorizer.SimpleWhitelistAuthorizer" - "org.apache.storm.testing.SingleUserSimpleTransport" {SimpleWhitelistAuthorizer/WHITELIST_USERS_CONF ["user"]}] + "org.apache.storm.security.auth.plain.PlainSaslTransportPlugin" {SimpleWhitelistAuthorizer/WHITELIST_USERS_CONF ["user"]}] (let [storm-conf (merge (clojurify-structure (ConfigUtils/readStormConfig)) - {STORM-THRIFT-TRANSPORT-PLUGIN "org.apache.storm.testing.SingleUserSimpleTransport"}) + {STORM-THRIFT-TRANSPORT-PLUGIN "org.apache.storm.security.auth.plain.PlainSaslTransportPlugin"}) client (NimbusClient. storm-conf "localhost" a-port nimbus-timeout) nimbus_client (.getClient client)] (testing "(Positive authorization) Authorization plugin should accept client request" @@ -336,9 +337,9 @@ (let [a-port (available-port)] (with-server [a-port nil "org.apache.storm.security.auth.authorizer.NoopAuthorizer" - "org.apache.storm.security.auth.SimpleTransportPlugin" nil] + "org.apache.storm.security.auth.plain.PlainSaslTransportPlugin" nil] (let [storm-conf (merge (clojurify-structure (ConfigUtils/readStormConfig)) - {STORM-THRIFT-TRANSPORT-PLUGIN "org.apache.storm.security.auth.SimpleTransportPlugin"}) + {STORM-THRIFT-TRANSPORT-PLUGIN "org.apache.storm.security.auth.plain.PlainSaslTransportPlugin"}) client (NimbusClient. storm-conf "localhost" a-port nimbus-timeout) nimbus_client (.getClient client)] (testing "(Positive authorization) Authorization plugin should accept client request" @@ -349,9 +350,9 @@ (let [a-port (available-port)] (with-server [a-port nil "org.apache.storm.security.auth.authorizer.DenyAuthorizer" - "org.apache.storm.security.auth.SimpleTransportPlugin" nil] + "org.apache.storm.security.auth.plain.PlainSaslTransportPlugin" nil] (let [storm-conf (merge (clojurify-structure (ConfigUtils/readStormConfig)) - {STORM-THRIFT-TRANSPORT-PLUGIN "org.apache.storm.security.auth.SimpleTransportPlugin" + {STORM-THRIFT-TRANSPORT-PLUGIN "org.apache.storm.security.auth.plain.PlainSaslTransportPlugin" Config/NIMBUS_THRIFT_PORT a-port Config/NIMBUS_TASK_TIMEOUT_SECS nimbus-timeout}) client (NimbusClient. storm-conf "localhost" a-port nimbus-timeout) @@ -378,7 +379,7 @@ (.close client)) (let [storm-conf (merge (clojurify-structure (ConfigUtils/readStormConfig)) - {STORM-THRIFT-TRANSPORT-PLUGIN "org.apache.storm.security.auth.SimpleTransportPlugin" + {STORM-THRIFT-TRANSPORT-PLUGIN "org.apache.storm.security.auth.plain.PlainSaslTransportPlugin" STORM-NIMBUS-RETRY-TIMES 0}) client (NimbusClient. storm-conf "localhost" a-port nimbus-timeout) nimbus_client (.getClient client)] diff --git a/storm-core/test/clj/org/apache/storm/security/auth/nimbus_auth_test.clj b/storm-core/test/clj/org/apache/storm/security/auth/nimbus_auth_test.clj index 307296aa3eb..d8063e33e74 100644 --- a/storm-core/test/clj/org/apache/storm/security/auth/nimbus_auth_test.clj +++ b/storm-core/test/clj/org/apache/storm/security/auth/nimbus_auth_test.clj @@ -56,9 +56,9 @@ (deftest Simple-authentication-test (let [port (available-port)] - (with-test-cluster [port nil nil "org.apache.storm.security.auth.SimpleTransportPlugin"] + (with-test-cluster [port nil nil "org.apache.storm.security.auth.plain.PlainSaslTransportPlugin"] (let [storm-conf (merge (clojurify-structure (ConfigUtils/readStormConfig)) - {STORM-THRIFT-TRANSPORT-PLUGIN "org.apache.storm.security.auth.SimpleTransportPlugin" + {STORM-THRIFT-TRANSPORT-PLUGIN "org.apache.storm.security.auth.plain.PlainSaslTransportPlugin" STORM-NIMBUS-RETRY-TIMES 0}) client (NimbusClient. storm-conf "localhost" port nimbus-timeout) nimbus_client (.getClient client)] @@ -71,9 +71,9 @@ (let [port (available-port)] (with-test-cluster [port nil "org.apache.storm.security.auth.authorizer.NoopAuthorizer" - "org.apache.storm.security.auth.SimpleTransportPlugin"] + "org.apache.storm.security.auth.plain.PlainSaslTransportPlugin"] (let [storm-conf (merge (clojurify-structure (ConfigUtils/readStormConfig)) - {STORM-THRIFT-TRANSPORT-PLUGIN "org.apache.storm.security.auth.SimpleTransportPlugin" + {STORM-THRIFT-TRANSPORT-PLUGIN "org.apache.storm.security.auth.plain.PlainSaslTransportPlugin" STORM-NIMBUS-RETRY-TIMES 0}) client (NimbusClient. storm-conf "localhost" port nimbus-timeout) nimbus_client (.getClient client)] @@ -86,9 +86,9 @@ (let [port (available-port)] (with-test-cluster [port nil "org.apache.storm.security.auth.authorizer.DenyAuthorizer" - "org.apache.storm.security.auth.SimpleTransportPlugin"] + "org.apache.storm.security.auth.plain.PlainSaslTransportPlugin"] (let [storm-conf (merge (clojurify-structure (ConfigUtils/readStormConfig)) - {STORM-THRIFT-TRANSPORT-PLUGIN "org.apache.storm.security.auth.SimpleTransportPlugin" + {STORM-THRIFT-TRANSPORT-PLUGIN "org.apache.storm.security.auth.plain.PlainSaslTransportPlugin" Config/NIMBUS_THRIFT_PORT port STORM-NIMBUS-RETRY-TIMES 0}) client (NimbusClient. storm-conf "localhost" port nimbus-timeout) From 93d6b0c340561b81f08cbf8d7eaa3dd290476d6c Mon Sep 17 00:00:00 2001 From: Kishor Patil Date: Wed, 1 Mar 2017 21:27:37 +0000 Subject: [PATCH 2/2] Remove reference to SimpleTransportPlugin --- .../java/org/apache/storm/daemon/drpc/DRPCServerTest.java | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/storm-drpc-server/src/test/java/org/apache/storm/daemon/drpc/DRPCServerTest.java b/storm-drpc-server/src/test/java/org/apache/storm/daemon/drpc/DRPCServerTest.java index b2df441e96e..84e0904e91c 100644 --- a/storm-drpc-server/src/test/java/org/apache/storm/daemon/drpc/DRPCServerTest.java +++ b/storm-drpc-server/src/test/java/org/apache/storm/daemon/drpc/DRPCServerTest.java @@ -36,7 +36,7 @@ import org.apache.storm.drpc.DRPCInvocationsClient; import org.apache.storm.generated.DRPCExecutionException; import org.apache.storm.generated.DRPCRequest; -import org.apache.storm.security.auth.SimpleTransportPlugin; +import org.apache.storm.security.auth.plain.PlainSaslTransportPlugin; import org.apache.storm.utils.DRPCClient; import org.apache.storm.utils.Utils; import org.junit.AfterClass; @@ -72,10 +72,9 @@ private Map getConf(int drpcPort, int invocationsPort, Integer h Map conf = new HashMap<>(); conf.put(Config.DRPC_PORT, drpcPort); conf.put(Config.DRPC_INVOCATIONS_PORT, invocationsPort); - conf.put(Config.STORM_THRIFT_TRANSPORT_PLUGIN, SimpleTransportPlugin.class.getName()); + conf.put(Config.STORM_THRIFT_TRANSPORT_PLUGIN, PlainSaslTransportPlugin.class.getName()); conf.put(Config.DRPC_WORKER_THREADS, 5); conf.put(Config.DRPC_INVOCATIONS_THREADS, 5); - conf.put(Config.DRPC_MAX_BUFFER_SIZE, 1048576); conf.put(Config.STORM_NIMBUS_RETRY_TIMES, 2); conf.put(Config.STORM_NIMBUS_RETRY_INTERVAL, 10); conf.put(Config.STORM_NIMBUS_RETRY_INTERVAL_CEILING, 100);