From c69ec3dc85b5ac9a7a8150d446e74c1665ce557d Mon Sep 17 00:00:00 2001 From: technoboy Date: Fri, 25 Jun 2021 08:53:30 +0800 Subject: [PATCH 1/9] support socks5 proxy --- .../pulsar/client/api/ClientBuilder.java | 22 +++++++++++++++ pulsar-client/pom.xml | 8 ++++++ .../pulsar/client/impl/ClientBuilderImpl.java | 19 +++++++++++++ .../pulsar/client/impl/ConnectionPool.java | 6 ++++- .../client/impl/PulsarChannelInitializer.java | 27 +++++++++++++++++++ .../impl/conf/ClientConfigurationData.java | 7 +++++ .../impl/conf/ConfigurationDataUtilsTest.java | 15 +++++++++++ 7 files changed, 103 insertions(+), 1 deletion(-) diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ClientBuilder.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ClientBuilder.java index aaa378f4780c8..2605afd85b72c 100644 --- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ClientBuilder.java +++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ClientBuilder.java @@ -18,6 +18,7 @@ */ package org.apache.pulsar.client.api; +import java.net.InetSocketAddress; import java.time.Clock; import java.util.Map; import java.util.Set; @@ -517,4 +518,25 @@ ClientBuilder authentication(String authPluginClassName, Map aut * @return */ ClientBuilder enableTransaction(boolean enableTransaction); + + /** + * Set socks5 proxy address. + * @param socks5ProxyAddress + * @return + */ + ClientBuilder socks5ProxyAddress(InetSocketAddress socks5ProxyAddress); + + /** + * Set socks5 proxy username. + * @param socks5ProxyUsername + * @return + */ + ClientBuilder socks5ProxyUsername(String socks5ProxyUsername); + + /** + * Set socks5 proxy password. + * @param socks5ProxyPassword + * @return + */ + ClientBuilder socks5ProxyPassword(String socks5ProxyPassword); } diff --git a/pulsar-client/pom.xml b/pulsar-client/pom.xml index d949830c55ced..65aef6af05a5f 100644 --- a/pulsar-client/pom.xml +++ b/pulsar-client/pom.xml @@ -70,6 +70,14 @@ io.netty netty-codec-http + + io.netty + netty-handler-proxy + + + io.netty + netty-codec-socks + io.netty diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientBuilderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientBuilderImpl.java index 4b9891447cc83..d7fa8189e95e3 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientBuilderImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientBuilderImpl.java @@ -18,6 +18,7 @@ */ package org.apache.pulsar.client.impl; +import java.net.InetSocketAddress; import java.time.Clock; import java.util.Map; import java.util.Set; @@ -324,4 +325,22 @@ public ClientBuilder enableTransaction(boolean enableTransaction) { conf.setEnableTransaction(enableTransaction); return this; } + + @Override + public ClientBuilder socks5ProxyAddress(InetSocketAddress socks5ProxyAddress) { + conf.setSocks5ProxyAddress(socks5ProxyAddress); + return this; + } + + @Override + public ClientBuilder socks5ProxyUsername(String socks5ProxyUsername) { + conf.setSocks5ProxyUsername(socks5ProxyUsername); + return this; + } + + @Override + public ClientBuilder socks5ProxyPassword(String socks5ProxyPassword) { + conf.setSocks5ProxyPassword(socks5ProxyPassword); + return this; + } } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionPool.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionPool.java index c7e4f58932a6e..eebe71a161cfc 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionPool.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionPool.java @@ -294,9 +294,13 @@ private CompletableFuture connectToAddress(InetAddress ipAddress, int p return toCompletableFuture(bootstrap.register()) .thenCompose(channel -> channelInitializerHandler .initTls(channel, sniHost != null ? sniHost : remoteAddress)) + .thenCompose(channel -> channelInitializerHandler + .initSocks5IfConfig(channel)) .thenCompose(channel -> toCompletableFuture(channel.connect(remoteAddress))); } else { - return toCompletableFuture(bootstrap.connect(remoteAddress)); + return toCompletableFuture(bootstrap.register()) + .thenCompose(channel -> channelInitializerHandler.initSocks5IfConfig(channel)) + .thenCompose(channel -> toCompletableFuture(channel.connect(remoteAddress))); } } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarChannelInitializer.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarChannelInitializer.java index e9a8bcd8e5d17..1353424b1b4ea 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarChannelInitializer.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarChannelInitializer.java @@ -24,6 +24,7 @@ import java.util.concurrent.TimeUnit; import java.util.function.Supplier; +import io.netty.handler.proxy.Socks5ProxyHandler; import org.apache.pulsar.client.api.AuthenticationDataProvider; import org.apache.pulsar.client.impl.conf.ClientConfigurationData; import org.apache.pulsar.client.util.ObjectCache; @@ -50,6 +51,9 @@ public class PulsarChannelInitializer extends ChannelInitializer @Getter private final boolean tlsEnabled; private final boolean tlsEnabledWithKeyStore; + private final InetSocketAddress socks5ProxyAddress; + private final String socks5ProxyUsername; + private final String socks5ProxyPassword; private final Supplier sslContextSupplier; private NettySSLContextAutoRefreshBuilder nettySSLContextAutoRefreshBuilder; @@ -61,6 +65,10 @@ public PulsarChannelInitializer(ClientConfigurationData conf, Supplier initTls(Channel ch, InetSocketAddress sniHost) { return initTlsFuture; } + + CompletableFuture initSocks5IfConfig(Channel ch) { + CompletableFuture initSocks5Future = new CompletableFuture<>(); + if (socks5ProxyAddress != null) { + ch.eventLoop().execute(() -> { + try { + Socks5ProxyHandler socks5ProxyHandler = new Socks5ProxyHandler(socks5ProxyAddress, socks5ProxyUsername, socks5ProxyPassword); + ch.pipeline().addFirst(socks5ProxyHandler.protocol(), socks5ProxyHandler); + initSocks5Future.complete(ch); + } catch (Throwable t) { + initSocks5Future.completeExceptionally(t); + } + }); + } else { + initSocks5Future.complete(ch); + } + + return initSocks5Future; + } } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java index 8ea35d218e1a7..735a15ddce2c1 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java @@ -20,6 +20,8 @@ import com.fasterxml.jackson.annotation.JsonIgnore; import com.google.common.collect.Sets; + +import java.net.InetSocketAddress; import java.time.Clock; import java.util.Set; import lombok.AllArgsConstructor; @@ -105,6 +107,11 @@ public class ClientConfigurationData implements Serializable, Cloneable { @JsonIgnore private Clock clock = Clock.systemDefaultZone(); + // socks5 + private InetSocketAddress socks5ProxyAddress; + private String socks5ProxyUsername; + private String socks5ProxyPassword; + public Authentication getAuthentication() { if (authentication == null) { this.authentication = AuthenticationDisabled.INSTANCE; diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/conf/ConfigurationDataUtilsTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/conf/ConfigurationDataUtilsTest.java index b84d8a41c5101..419fb756f0ec5 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/conf/ConfigurationDataUtilsTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/conf/ConfigurationDataUtilsTest.java @@ -28,6 +28,7 @@ import static org.testng.Assert.fail; import java.io.IOException; +import java.net.InetSocketAddress; import java.util.HashMap; import java.util.Map; import java.util.concurrent.TimeUnit; @@ -197,4 +198,18 @@ public void testEquals() { assertEquals(confData1, confData2); assertEquals(confData1.hashCode(), confData2.hashCode()); } + + @Test + public void testSocks5() throws PulsarClientException { + ClientConfigurationData clientConfig = new ClientConfigurationData(); + clientConfig.setServiceUrl("pulsar://unknown:6650"); + clientConfig.setSocks5ProxyAddress(new InetSocketAddress("localhost", 11080)); + clientConfig.setSocks5ProxyUsername("test"); + clientConfig.setSocks5ProxyPassword("test123"); + + PulsarClientImpl pulsarClient = new PulsarClientImpl(clientConfig); + assertEquals(pulsarClient.getConfiguration().getSocks5ProxyAddress(), new InetSocketAddress("localhost", 11080)); + assertEquals(pulsarClient.getConfiguration().getSocks5ProxyUsername(), "test"); + assertEquals(pulsarClient.getConfiguration().getSocks5ProxyPassword(), "test123"); + } } From 4c7944c49673fa3eb55deb6498cab231d9230dff Mon Sep 17 00:00:00 2001 From: technoboy Date: Fri, 25 Jun 2021 10:53:27 +0800 Subject: [PATCH 2/9] add from '-D' or System property --- .../impl/conf/ClientConfigurationData.java | 21 +++++++++++++++++++ .../impl/conf/ConfigurationDataUtilsTest.java | 8 +++++++ 2 files changed, 29 insertions(+) diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java index 735a15ddce2c1..8af39512ca8f6 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java @@ -22,7 +22,9 @@ import com.google.common.collect.Sets; import java.net.InetSocketAddress; +import java.net.URI; import java.time.Clock; +import java.util.Objects; import java.util.Set; import lombok.AllArgsConstructor; import lombok.Data; @@ -140,5 +142,24 @@ public ClientConfigurationData clone() { } } + public InetSocketAddress getSocks5ProxyAddress() { + if (Objects.nonNull(socks5ProxyAddress)) { + return socks5ProxyAddress; + } + String proxyAddress = System.getProperty("socks5Proxy.address"); + try { + URI uri = URI.create(proxyAddress); + return new InetSocketAddress(uri.getHost(), uri.getPort()); + } catch (Exception ignore) { + return null; + } + } + public String getSocks5ProxyUsername() { + return Objects.nonNull(socks5ProxyUsername) ? socks5ProxyUsername : System.getProperty("socks5Proxy.username"); + } + + public String getSocks5ProxyPassword() { + return Objects.nonNull(socks5ProxyPassword) ? socks5ProxyPassword : System.getProperty("socks5Proxy.password"); + } } diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/conf/ConfigurationDataUtilsTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/conf/ConfigurationDataUtilsTest.java index 419fb756f0ec5..cb67beca28393 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/conf/ConfigurationDataUtilsTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/conf/ConfigurationDataUtilsTest.java @@ -211,5 +211,13 @@ public void testSocks5() throws PulsarClientException { assertEquals(pulsarClient.getConfiguration().getSocks5ProxyAddress(), new InetSocketAddress("localhost", 11080)); assertEquals(pulsarClient.getConfiguration().getSocks5ProxyUsername(), "test"); assertEquals(pulsarClient.getConfiguration().getSocks5ProxyPassword(), "test123"); + + ClientConfigurationData clientConfig2 = new ClientConfigurationData(); + System.setProperty("socks5Proxy.address", "http://localhost:11080"); + System.setProperty("socks5Proxy.username", "pulsar"); + System.setProperty("socks5Proxy.password", "pulsar123"); + assertEquals(clientConfig2.getSocks5ProxyAddress(), new InetSocketAddress("localhost", 11080)); + assertEquals(clientConfig2.getSocks5ProxyUsername(), "pulsar"); + assertEquals(clientConfig2.getSocks5ProxyPassword(), "pulsar123"); } } From 873f38df1a996f7159150b36976dbb9ad51251b7 Mon Sep 17 00:00:00 2001 From: technoboy Date: Sun, 27 Jun 2021 14:45:40 +0800 Subject: [PATCH 3/9] add test cases and LICENSE --- .../impl/ClientWithSocks5ProxyTest.java | 159 ++++++++++++++++++ .../apache/pulsar/socks5/Socks5Server.java | 87 ++++++++++ .../socks5/auth/DefaultPasswordAuthImpl.java | 31 ++++ .../pulsar/socks5/auth/PasswordAuth.java | 25 +++ .../pulsar/socks5/config/Socks5Config.java | 30 ++++ .../socks5/handler/CommandRequestHandler.java | 118 +++++++++++++ .../pulsar/socks5/handler/IdleHandler.java | 36 ++++ .../socks5/handler/InitialRequestHandler.java | 56 ++++++ .../handler/PasswordAuthRequestHandler.java | 50 ++++++ pulsar-sql/presto-distribution/LICENSE | 2 + 10 files changed, 594 insertions(+) create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ClientWithSocks5ProxyTest.java create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/socks5/Socks5Server.java create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/socks5/auth/DefaultPasswordAuthImpl.java create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/socks5/auth/PasswordAuth.java create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/socks5/config/Socks5Config.java create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/CommandRequestHandler.java create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/IdleHandler.java create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/InitialRequestHandler.java create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/PasswordAuthRequestHandler.java diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ClientWithSocks5ProxyTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ClientWithSocks5ProxyTest.java new file mode 100644 index 0000000000000..7799f0a0ff21c --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ClientWithSocks5ProxyTest.java @@ -0,0 +1,159 @@ +/** + * 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.pulsar.client.impl; + +import org.apache.pulsar.broker.service.BrokerTestBase; +import org.apache.pulsar.client.admin.PulsarAdminException; +import org.apache.pulsar.client.api.ClientBuilder; +import org.apache.pulsar.client.api.Consumer; +import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.client.api.Producer; +import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.client.api.SubscriptionType; +import org.apache.pulsar.common.policies.data.TenantInfo; +import org.apache.pulsar.socks5.Socks5Server; +import org.apache.pulsar.socks5.config.Socks5Config; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; +import org.testng.internal.thread.ThreadTimeoutException; + +import java.net.InetSocketAddress; +import java.util.Collections; +import java.util.HashSet; +import java.util.Set; + +import static org.testng.Assert.assertEquals; + +@Test +public class ClientWithSocks5ProxyTest extends BrokerTestBase { + + private Socks5Server server; + + final String topicName = "persistent://public/default/socks5"; + + @BeforeMethod + public void setup() throws Exception { + baseSetup(); + initData(); + } + + protected void customizeNewPulsarClientBuilder(ClientBuilder clientBuilder) { + clientBuilder.socks5ProxyAddress(new InetSocketAddress("localhost", 11080)) + .socks5ProxyUsername("socks5") + .socks5ProxyPassword("pulsar"); + } + + private void startSocks5Server(boolean enableAuth) { + Socks5Config config = new Socks5Config(); + config.setPort(11080); + config.setEnableAuth(enableAuth); + server = new Socks5Server(config); + Thread thread = new Thread(new Runnable() { + @Override + public void run() { + try { + server.start(); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + thread.setDaemon(true); + thread.start(); + } + + @AfterMethod(alwaysRun = true) + protected void cleanup() throws Exception { + internalCleanup(); + server.shutdown(); + } + + private void initData() throws PulsarAdminException { + admin.tenants().createTenant("public", new TenantInfo() { + @Override + public Set getAdminRoles() { + return Collections.emptySet(); + } + + @Override + public Set getAllowedClusters() { + Set clusters = new HashSet<>(); + clusters.add("test"); + return clusters; + } + }); + admin.namespaces().createNamespace("public/default"); + admin.topics().createNonPartitionedTopic(topicName); + } + + @Test + public void testSendAndConsumer() throws PulsarClientException { + startSocks5Server(true); + // init consumer + final String subscriptionName = "socks5-subscription"; + Consumer consumer = pulsarClient.newConsumer() + .topic(topicName) + .subscriptionName(subscriptionName) + .subscriptionType(SubscriptionType.Shared) + .subscribe(); + + //init producer + Producer producer = pulsarClient.newProducer() + .topic(topicName) + .create(); + + String msg = "abc"; + producer.send(msg.getBytes()); + Message message = consumer.receive(); + + assertEquals(new String(message.getData()), msg); + + consumer.unsubscribe(); + } + + @Test + public void testDisableAuth() throws PulsarClientException { + startSocks5Server(false); + ClientBuilder clientBuilder = PulsarClient.builder() + .serviceUrl(pulsar.getBrokerServiceUrl()) + .socks5ProxyAddress(new InetSocketAddress("localhost", 11080)); + PulsarClient pulsarClient = replacePulsarClient(clientBuilder); + Producer producer = pulsarClient.newProducer() + .topic(topicName) + .create(); + String msg = "abc"; + producer.send(msg.getBytes()); + } + + @Test(timeOut = 5000, expectedExceptions = {ThreadTimeoutException.class}) + public void testWithErrorPassword() throws PulsarClientException { + startSocks5Server(true); + ClientBuilder clientBuilder = PulsarClient.builder() + .serviceUrl(pulsar.getBrokerServiceUrl()) + .socks5ProxyAddress(new InetSocketAddress("localhost", 11080)) + .socks5ProxyUsername("socks5") + .socks5ProxyPassword("error-password"); + PulsarClient pulsarClient = replacePulsarClient(clientBuilder); + pulsarClient.newProducer() + .topic(topicName) + .create(); + } +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/socks5/Socks5Server.java b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/Socks5Server.java new file mode 100644 index 0000000000000..787eb67818d43 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/Socks5Server.java @@ -0,0 +1,87 @@ +/** + * 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.pulsar.socks5; + +import io.netty.bootstrap.ServerBootstrap; +import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelInitializer; +import io.netty.channel.ChannelOption; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.socket.SocketChannel; +import io.netty.channel.socket.nio.NioServerSocketChannel; +import io.netty.handler.codec.socksx.v5.Socks5CommandRequestDecoder; +import io.netty.handler.codec.socksx.v5.Socks5InitialRequestDecoder; +import io.netty.handler.codec.socksx.v5.Socks5PasswordAuthRequestDecoder; +import io.netty.handler.codec.socksx.v5.Socks5ServerEncoder; +import io.netty.handler.timeout.IdleStateHandler; +import lombok.Getter; +import lombok.extern.slf4j.Slf4j; +import org.apache.pulsar.socks5.config.Socks5Config; +import org.apache.pulsar.socks5.handler.CommandRequestHandler; +import org.apache.pulsar.socks5.handler.IdleHandler; +import org.apache.pulsar.socks5.handler.InitialRequestHandler; +import org.apache.pulsar.socks5.handler.PasswordAuthRequestHandler; + +@Slf4j +public class Socks5Server { + + @Getter + private EventLoopGroup boss = new NioEventLoopGroup(); + private EventLoopGroup worker = new NioEventLoopGroup(); + + private final Socks5Config socks5Config; + + public Socks5Server(Socks5Config socks5Config) { + this.socks5Config = socks5Config; + } + + public void start() throws Exception { + ServerBootstrap bootstrap = new ServerBootstrap(); + bootstrap.group(boss, worker) + .channel(NioServerSocketChannel.class) + .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000) + .childHandler(new ChannelInitializer() { + @Override + protected void initChannel(SocketChannel ch) throws Exception { + ch.pipeline().addLast(new IdleStateHandler(90, 90, 0)); + ch.pipeline().addLast(new IdleHandler()); + ch.pipeline().addLast(Socks5ServerEncoder.DEFAULT); + ch.pipeline().addLast(new Socks5InitialRequestDecoder()); + ch.pipeline().addLast(new InitialRequestHandler(socks5Config)); + if (socks5Config.isEnableAuth()) { + ch.pipeline().addLast(new Socks5PasswordAuthRequestDecoder()); + ch.pipeline().addLast(new PasswordAuthRequestHandler()); + } + ch.pipeline().addLast(new Socks5CommandRequestDecoder()); + ch.pipeline().addLast(new CommandRequestHandler(Socks5Server.this)); + } + }); + ChannelFuture future = bootstrap.bind(socks5Config.getPort()).sync(); + if (log.isInfoEnabled()) { + log.info("bind port : {}", socks5Config.getPort()); + } + future.channel().closeFuture().sync(); + } + + public void shutdown() { + boss.shutdownGracefully(); + worker.shutdownGracefully(); + } +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/socks5/auth/DefaultPasswordAuthImpl.java b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/auth/DefaultPasswordAuthImpl.java new file mode 100644 index 0000000000000..11d9b555b087e --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/auth/DefaultPasswordAuthImpl.java @@ -0,0 +1,31 @@ +/** + * 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.pulsar.socks5.auth; + +public class DefaultPasswordAuthImpl implements PasswordAuth { + + public static final String DEFAULT_USERNAME = "socks5"; + + public static final String DEFAULT_PASSWORD = "pulsar"; + + @Override + public boolean auth(String username, String password) { + return DEFAULT_USERNAME.equalsIgnoreCase(username) && DEFAULT_PASSWORD.equalsIgnoreCase(password); + } +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/socks5/auth/PasswordAuth.java b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/auth/PasswordAuth.java new file mode 100644 index 0000000000000..d4370f2ca7023 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/auth/PasswordAuth.java @@ -0,0 +1,25 @@ +/** + * 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.pulsar.socks5.auth; + +public interface PasswordAuth { + + boolean auth(String username, String password); + +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/socks5/config/Socks5Config.java b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/config/Socks5Config.java new file mode 100644 index 0000000000000..40310e657adda --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/config/Socks5Config.java @@ -0,0 +1,30 @@ +/** + * 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.pulsar.socks5.config; + +import lombok.Data; + +@Data +public class Socks5Config { + + private boolean enableAuth; + + private int port; + +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/CommandRequestHandler.java b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/CommandRequestHandler.java new file mode 100644 index 0000000000000..8c6130572ac46 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/CommandRequestHandler.java @@ -0,0 +1,118 @@ +/** + * 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.pulsar.socks5.handler; + +import io.netty.bootstrap.Bootstrap; +import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelFutureListener; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelInboundHandlerAdapter; +import io.netty.channel.ChannelInitializer; +import io.netty.channel.ChannelOption; +import io.netty.channel.SimpleChannelInboundHandler; +import io.netty.channel.socket.SocketChannel; +import io.netty.channel.socket.nio.NioSocketChannel; +import io.netty.handler.codec.socksx.v5.DefaultSocks5CommandRequest; +import io.netty.handler.codec.socksx.v5.DefaultSocks5CommandResponse; +import io.netty.handler.codec.socksx.v5.Socks5AddressType; +import io.netty.handler.codec.socksx.v5.Socks5CommandStatus; +import io.netty.handler.codec.socksx.v5.Socks5CommandType; +import lombok.extern.slf4j.Slf4j; +import org.apache.pulsar.socks5.Socks5Server; + +@Slf4j +public class CommandRequestHandler extends SimpleChannelInboundHandler { + + private final Socks5Server socks5Server; + + public CommandRequestHandler(Socks5Server socks5Server) { + this.socks5Server = socks5Server; + } + + @Override + protected void channelRead0(final ChannelHandlerContext clientChannelContext, DefaultSocks5CommandRequest msg) throws Exception { + if (Socks5CommandType.CONNECT.equals(msg.type())) { + Bootstrap bootstrap = new Bootstrap(); + bootstrap.group(socks5Server.getBoss()) + .channel(NioSocketChannel.class) + .option(ChannelOption.TCP_NODELAY, true) + .handler(new ChannelInitializer() { + @Override + protected void initChannel(SocketChannel ch) throws Exception { + ch.pipeline().addLast(new ClientHandler(clientChannelContext)); + } + }); + ChannelFuture future = bootstrap.connect(msg.dstAddr(), msg.dstPort()); + future.addListener(new ChannelFutureListener() { + + public void operationComplete(final ChannelFuture future) throws Exception { + if (future.isSuccess()) { + if (log.isDebugEnabled()) { + log.debug("connected : {} {}", msg.dstAddr(), msg.dstPort()); + } + clientChannelContext.pipeline().addLast(new TargetHandler(future)); + clientChannelContext.writeAndFlush(new DefaultSocks5CommandResponse(Socks5CommandStatus.SUCCESS, Socks5AddressType.IPv4)); + } else { + clientChannelContext.writeAndFlush(new DefaultSocks5CommandResponse(Socks5CommandStatus.FAILURE, Socks5AddressType.IPv4)); + } + } + }); + } else { + clientChannelContext.fireChannelRead(msg); + } + } + + private static class ClientHandler extends ChannelInboundHandlerAdapter { + + private ChannelHandlerContext clientChannelContext; + + public ClientHandler(ChannelHandlerContext clientChannelContext) { + this.clientChannelContext = clientChannelContext; + } + + @Override + public void channelRead(ChannelHandlerContext ctx2, Object destMsg) throws Exception { + clientChannelContext.writeAndFlush(destMsg); + } + + @Override + public void channelInactive(ChannelHandlerContext ctx2) throws Exception { + clientChannelContext.channel().close(); + } + } + + private static class TargetHandler extends ChannelInboundHandlerAdapter { + + private ChannelFuture targetChannel; + + public TargetHandler(ChannelFuture targetChannel) { + this.targetChannel = targetChannel; + } + + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { + targetChannel.channel().writeAndFlush(msg); + } + + @Override + public void channelInactive(ChannelHandlerContext ctx) throws Exception { + targetChannel.channel().close(); + } + } +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/IdleHandler.java b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/IdleHandler.java new file mode 100644 index 0000000000000..73f6cbe621c39 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/IdleHandler.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.pulsar.socks5.handler; + +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelInboundHandlerAdapter; +import io.netty.handler.timeout.IdleStateEvent; + +public class IdleHandler extends ChannelInboundHandlerAdapter { + + @Override + public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { + if (evt instanceof IdleStateEvent) { + ctx.channel().close(); + } else { + super.userEventTriggered(ctx, evt); + } + } + +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/InitialRequestHandler.java b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/InitialRequestHandler.java new file mode 100644 index 0000000000000..2b9cc98c3ebb6 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/InitialRequestHandler.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.pulsar.socks5.handler; + +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.SimpleChannelInboundHandler; +import io.netty.handler.codec.socksx.SocksVersion; +import io.netty.handler.codec.socksx.v5.DefaultSocks5InitialRequest; +import io.netty.handler.codec.socksx.v5.DefaultSocks5InitialResponse; +import io.netty.handler.codec.socksx.v5.Socks5AuthMethod; +import lombok.extern.slf4j.Slf4j; +import org.apache.pulsar.socks5.config.Socks5Config; + +@Slf4j +public class InitialRequestHandler extends SimpleChannelInboundHandler { + + private final Socks5Config socks5Config; + + public InitialRequestHandler(final Socks5Config socks5Config) { + this.socks5Config = socks5Config; + } + + @Override + protected void channelRead0(ChannelHandlerContext ctx, DefaultSocks5InitialRequest msg) throws Exception { + if (SocksVersion.SOCKS5.equals(msg.version())) { + if (msg.decoderResult().isFailure()) { + log.warn("decode failure : {}", msg.decoderResult()); + ctx.fireChannelRead(msg); + } else { + if (SocksVersion.SOCKS5.equals(msg.version())) { + if (socks5Config.isEnableAuth()) { + ctx.writeAndFlush(new DefaultSocks5InitialResponse(Socks5AuthMethod.PASSWORD)); + } else { + ctx.writeAndFlush(new DefaultSocks5InitialResponse(Socks5AuthMethod.NO_AUTH)); + } + } + } + } + } +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/PasswordAuthRequestHandler.java b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/PasswordAuthRequestHandler.java new file mode 100644 index 0000000000000..1078f0e87e79a --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/PasswordAuthRequestHandler.java @@ -0,0 +1,50 @@ +/** + * 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.pulsar.socks5.handler; + +import io.netty.channel.ChannelFutureListener; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.SimpleChannelInboundHandler; +import io.netty.handler.codec.socksx.v5.DefaultSocks5PasswordAuthRequest; +import io.netty.handler.codec.socksx.v5.DefaultSocks5PasswordAuthResponse; +import io.netty.handler.codec.socksx.v5.Socks5PasswordAuthStatus; +import lombok.extern.slf4j.Slf4j; +import org.apache.pulsar.socks5.auth.DefaultPasswordAuthImpl; +import org.apache.pulsar.socks5.auth.PasswordAuth; + +@Slf4j +public class PasswordAuthRequestHandler extends SimpleChannelInboundHandler { + + private final PasswordAuth passwordAuth; + + public PasswordAuthRequestHandler() { + this.passwordAuth = new DefaultPasswordAuthImpl(); + } + + @Override + protected void channelRead0(ChannelHandlerContext ctx, DefaultSocks5PasswordAuthRequest msg) throws Exception { + if (passwordAuth.auth(msg.username(), msg.password())) { + ctx.writeAndFlush(new DefaultSocks5PasswordAuthResponse(Socks5PasswordAuthStatus.SUCCESS)); + } else { + ctx.writeAndFlush(new DefaultSocks5PasswordAuthResponse(Socks5PasswordAuthStatus.FAILURE)) + .addListener(ChannelFutureListener.CLOSE); + } + } + +} diff --git a/pulsar-sql/presto-distribution/LICENSE b/pulsar-sql/presto-distribution/LICENSE index 0d10346a964ae..4af9e55498fd3 100644 --- a/pulsar-sql/presto-distribution/LICENSE +++ b/pulsar-sql/presto-distribution/LICENSE @@ -238,6 +238,8 @@ The Apache Software License, Version 2.0 - netty-codec-dns-4.1.63.Final.jar - netty-codec-http-4.1.63.Final.jar - netty-codec-haproxy-4.1.63.Final.jar + - netty-codec-socks-4.1.63.Final.jar + - netty-handler-proxy-4.1.63.Final.jar - netty-common-4.1.63.Final.jar - netty-handler-4.1.63.Final.jar - netty-reactive-streams-2.0.4.jar From 4e62fd781307fb70d5f2b0204cc9867e3617e3d6 Mon Sep 17 00:00:00 2001 From: technoboy Date: Mon, 28 Jun 2021 11:41:09 +0800 Subject: [PATCH 4/9] add documentation --- site2/docs/client-libraries-java.md | 4 +++- .../versioned_docs/version-2.8.1/client-libraries-java.md | 3 +++ 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/site2/docs/client-libraries-java.md b/site2/docs/client-libraries-java.md index da25f16d2e1da..47cdba2cf1276 100644 --- a/site2/docs/client-libraries-java.md +++ b/site2/docs/client-libraries-java.md @@ -121,7 +121,9 @@ int|`connectionTimeoutMs`|Duration of waiting for a connection to a broker to be int|`requestTimeoutMs`|Maximum duration for completing a request |60000 int|`defaultBackoffIntervalNanos`| Default duration for a backoff interval | TimeUnit.MILLISECONDS.toNanos(100); long|`maxBackoffIntervalNanos`|Maximum duration for a backoff interval|TimeUnit.SECONDS.toNanos(30) - +SocketAddress|`socks5ProxyAddress`|SOCKS5 proxy address | None +String|`socks5ProxyUsername`|SOCKS5 proxy username | None +String|`socks5ProxyPassword`|SOCKS5 proxy password | None Check out the Javadoc for the {@inject: javadoc:PulsarClient:/client/org/apache/pulsar/client/api/PulsarClient} class for a full list of configurable parameters. > In addition to client-level configuration, you can also apply [producer](#configuring-producers) and [consumer](#configuring-consumers) specific configuration as described in sections below. diff --git a/site2/website/versioned_docs/version-2.8.1/client-libraries-java.md b/site2/website/versioned_docs/version-2.8.1/client-libraries-java.md index 261b2bc3dfd42..2a0b764b0a420 100644 --- a/site2/website/versioned_docs/version-2.8.1/client-libraries-java.md +++ b/site2/website/versioned_docs/version-2.8.1/client-libraries-java.md @@ -122,6 +122,9 @@ int|`connectionTimeoutMs`|Duration of waiting for a connection to a broker to be int|`requestTimeoutMs`|Maximum duration for completing a request |60000 int|`defaultBackoffIntervalNanos`| Default duration for a backoff interval | TimeUnit.MILLISECONDS.toNanos(100); long|`maxBackoffIntervalNanos`|Maximum duration for a backoff interval|TimeUnit.SECONDS.toNanos(30) +SocketAddress|`socks5ProxyAddress`|SOCKS5 proxy address | None +String|`socks5ProxyUsername`|SOCKS5 proxy username | None +String|`socks5ProxyPassword`|SOCKS5 proxy password | None Check out the Javadoc for the {@inject: javadoc:PulsarClient:/client/org/apache/pulsar/client/api/PulsarClient} class for a full list of configurable parameters. From 04b1bb78ffd5f5a8bb38591ef96e1cd93520253a Mon Sep 17 00:00:00 2001 From: technoboy Date: Wed, 30 Jun 2021 15:37:54 +0800 Subject: [PATCH 5/9] updates from suggestion --- .../pulsar/client/impl/ClientWithSocks5ProxyTest.java | 4 +++- .../pulsar/socks5/handler/InitialRequestHandler.java | 10 ++++------ .../client/impl/conf/ClientConfigurationData.java | 4 ++-- 3 files changed, 9 insertions(+), 9 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ClientWithSocks5ProxyTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ClientWithSocks5ProxyTest.java index 7799f0a0ff21c..7eb84453a6877 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ClientWithSocks5ProxyTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ClientWithSocks5ProxyTest.java @@ -18,6 +18,7 @@ */ package org.apache.pulsar.client.impl; +import lombok.extern.slf4j.Slf4j; import org.apache.pulsar.broker.service.BrokerTestBase; import org.apache.pulsar.client.admin.PulsarAdminException; import org.apache.pulsar.client.api.ClientBuilder; @@ -43,6 +44,7 @@ import static org.testng.Assert.assertEquals; @Test +@Slf4j public class ClientWithSocks5ProxyTest extends BrokerTestBase { private Socks5Server server; @@ -72,7 +74,7 @@ public void run() { try { server.start(); } catch (Exception e) { - e.printStackTrace(); + log.error("start socks5 server error", e); } } }); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/InitialRequestHandler.java b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/InitialRequestHandler.java index 2b9cc98c3ebb6..ee9dc3984a27b 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/InitialRequestHandler.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/InitialRequestHandler.java @@ -43,12 +43,10 @@ protected void channelRead0(ChannelHandlerContext ctx, DefaultSocks5InitialReque log.warn("decode failure : {}", msg.decoderResult()); ctx.fireChannelRead(msg); } else { - if (SocksVersion.SOCKS5.equals(msg.version())) { - if (socks5Config.isEnableAuth()) { - ctx.writeAndFlush(new DefaultSocks5InitialResponse(Socks5AuthMethod.PASSWORD)); - } else { - ctx.writeAndFlush(new DefaultSocks5InitialResponse(Socks5AuthMethod.NO_AUTH)); - } + if (socks5Config.isEnableAuth()) { + ctx.writeAndFlush(new DefaultSocks5InitialResponse(Socks5AuthMethod.PASSWORD)); + } else { + ctx.writeAndFlush(new DefaultSocks5InitialResponse(Socks5AuthMethod.NO_AUTH)); } } } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java index 8af39512ca8f6..fe9e0a373f872 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java @@ -150,8 +150,8 @@ public InetSocketAddress getSocks5ProxyAddress() { try { URI uri = URI.create(proxyAddress); return new InetSocketAddress(uri.getHost(), uri.getPort()); - } catch (Exception ignore) { - return null; + } catch (Exception e) { + throw new RuntimeException("Invalid config [socks5Proxy.address]", e); } } From d4ff2a90ad180992e51703a7e78840b77a40d65d Mon Sep 17 00:00:00 2001 From: technoboy Date: Wed, 30 Jun 2021 16:09:41 +0800 Subject: [PATCH 6/9] add more testcases --- .../impl/ClientWithSocks5ProxyTest.java | 29 +++++++++++++++++++ .../impl/conf/ConfigurationDataUtilsTest.java | 9 ++++++ 2 files changed, 38 insertions(+) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ClientWithSocks5ProxyTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ClientWithSocks5ProxyTest.java index 7eb84453a6877..5aa61cfceec2a 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ClientWithSocks5ProxyTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ClientWithSocks5ProxyTest.java @@ -57,6 +57,7 @@ public void setup() throws Exception { initData(); } + @Override protected void customizeNewPulsarClientBuilder(ClientBuilder clientBuilder) { clientBuilder.socks5ProxyAddress(new InetSocketAddress("localhost", 11080)) .socks5ProxyUsername("socks5") @@ -145,6 +146,34 @@ public void testDisableAuth() throws PulsarClientException { producer.send(msg.getBytes()); } + @Test + public void testSetFromSystemProperty() throws PulsarClientException { + startSocks5Server(false); + System.setProperty("socks5Proxy.address", "http://localhost:11080"); + ClientBuilder clientBuilder = PulsarClient.builder() + .serviceUrl(pulsar.getBrokerServiceUrl()); + PulsarClient pulsarClient = replacePulsarClient(clientBuilder); + Producer producer = pulsarClient.newProducer() + .topic(topicName) + .create(); + String msg = "abc"; + producer.send(msg.getBytes()); + } + + @Test(expectedExceptions = PulsarClientException.class) + public void testSetErrorProxyAddress() throws PulsarClientException { + startSocks5Server(false); + System.setProperty("socks5Proxy.address", "localhost:11080"); // with no scheme + ClientBuilder clientBuilder = PulsarClient.builder() + .serviceUrl(pulsar.getBrokerServiceUrl()); + PulsarClient pulsarClient = replacePulsarClient(clientBuilder); + Producer producer = pulsarClient.newProducer() + .topic(topicName) + .create(); + String msg = "abc"; + producer.send(msg.getBytes()); + } + @Test(timeOut = 5000, expectedExceptions = {ThreadTimeoutException.class}) public void testWithErrorPassword() throws PulsarClientException { startSocks5Server(true); diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/conf/ConfigurationDataUtilsTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/conf/ConfigurationDataUtilsTest.java index cb67beca28393..61e6f2262ad99 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/conf/ConfigurationDataUtilsTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/conf/ConfigurationDataUtilsTest.java @@ -219,5 +219,14 @@ public void testSocks5() throws PulsarClientException { assertEquals(clientConfig2.getSocks5ProxyAddress(), new InetSocketAddress("localhost", 11080)); assertEquals(clientConfig2.getSocks5ProxyUsername(), "pulsar"); assertEquals(clientConfig2.getSocks5ProxyPassword(), "pulsar123"); + + System.setProperty("socks5Proxy.address", "localhost:11080"); // invalid address, no scheme + try { + clientConfig2.getSocks5ProxyAddress(); + fail("No exception thrown."); + } catch (Exception ex) { + assertTrue(ex.getMessage().contains("Invalid config [socks5Proxy.address]")); + } + } } From 572f89486589cd8ce09bf89e0bbd975f596c84ab Mon Sep 17 00:00:00 2001 From: technoboy Date: Wed, 30 Jun 2021 17:53:54 +0800 Subject: [PATCH 7/9] fix tab indents --- .../apache/pulsar/socks5/Socks5Server.java | 77 ++++----- .../pulsar/socks5/auth/PasswordAuth.java | 2 +- .../socks5/handler/CommandRequestHandler.java | 153 +++++++++--------- .../pulsar/socks5/handler/IdleHandler.java | 20 +-- .../socks5/handler/InitialRequestHandler.java | 39 ++--- .../handler/PasswordAuthRequestHandler.java | 26 +-- 6 files changed, 160 insertions(+), 157 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/socks5/Socks5Server.java b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/Socks5Server.java index 787eb67818d43..7510b03bec655 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/socks5/Socks5Server.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/Socks5Server.java @@ -42,46 +42,47 @@ @Slf4j public class Socks5Server { - @Getter - private EventLoopGroup boss = new NioEventLoopGroup(); - private EventLoopGroup worker = new NioEventLoopGroup(); + @Getter + private EventLoopGroup boss = new NioEventLoopGroup(); + private EventLoopGroup worker = new NioEventLoopGroup(); - private final Socks5Config socks5Config; + private final Socks5Config socks5Config; - public Socks5Server(Socks5Config socks5Config) { - this.socks5Config = socks5Config; - } + public Socks5Server(Socks5Config socks5Config) { + this.socks5Config = socks5Config; + } - public void start() throws Exception { - ServerBootstrap bootstrap = new ServerBootstrap(); - bootstrap.group(boss, worker) - .channel(NioServerSocketChannel.class) - .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000) - .childHandler(new ChannelInitializer() { - @Override - protected void initChannel(SocketChannel ch) throws Exception { - ch.pipeline().addLast(new IdleStateHandler(90, 90, 0)); - ch.pipeline().addLast(new IdleHandler()); - ch.pipeline().addLast(Socks5ServerEncoder.DEFAULT); - ch.pipeline().addLast(new Socks5InitialRequestDecoder()); - ch.pipeline().addLast(new InitialRequestHandler(socks5Config)); - if (socks5Config.isEnableAuth()) { - ch.pipeline().addLast(new Socks5PasswordAuthRequestDecoder()); - ch.pipeline().addLast(new PasswordAuthRequestHandler()); - } - ch.pipeline().addLast(new Socks5CommandRequestDecoder()); - ch.pipeline().addLast(new CommandRequestHandler(Socks5Server.this)); - } - }); - ChannelFuture future = bootstrap.bind(socks5Config.getPort()).sync(); - if (log.isInfoEnabled()) { - log.info("bind port : {}", socks5Config.getPort()); - } - future.channel().closeFuture().sync(); - } + public void start() throws Exception { + ServerBootstrap bootstrap = new ServerBootstrap(); + bootstrap.group(boss, worker) + .channel(NioServerSocketChannel.class) + .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000) + .childHandler(new ChannelInitializer() { + @Override + protected void initChannel(SocketChannel ch) throws Exception { + ch.pipeline().addLast(new IdleStateHandler(90, 90, 0)); + ch.pipeline().addLast(new IdleHandler()); + ch.pipeline().addLast(Socks5ServerEncoder.DEFAULT); + ch.pipeline().addLast(new Socks5InitialRequestDecoder()); + ch.pipeline().addLast(new InitialRequestHandler(socks5Config)); + if (socks5Config.isEnableAuth()) { + ch.pipeline().addLast(new Socks5PasswordAuthRequestDecoder()); + ch.pipeline().addLast(new PasswordAuthRequestHandler()); + } + ch.pipeline().addLast(new Socks5CommandRequestDecoder()); + ch.pipeline().addLast(new CommandRequestHandler(Socks5Server.this)); + } + }); + ChannelFuture future = bootstrap.bind(socks5Config.getPort()).sync(); + if (log.isInfoEnabled()) { + log.info("bind port : {}", socks5Config.getPort()); + } + future.channel().closeFuture().sync(); + } + + public void shutdown() { + boss.shutdownGracefully(); + worker.shutdownGracefully(); + } - public void shutdown() { - boss.shutdownGracefully(); - worker.shutdownGracefully(); - } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/socks5/auth/PasswordAuth.java b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/auth/PasswordAuth.java index d4370f2ca7023..b5181dff08439 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/socks5/auth/PasswordAuth.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/auth/PasswordAuth.java @@ -20,6 +20,6 @@ public interface PasswordAuth { - boolean auth(String username, String password); + boolean auth(String username, String password); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/CommandRequestHandler.java b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/CommandRequestHandler.java index 8c6130572ac46..a6450c5bc6223 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/CommandRequestHandler.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/CommandRequestHandler.java @@ -39,80 +39,81 @@ @Slf4j public class CommandRequestHandler extends SimpleChannelInboundHandler { - private final Socks5Server socks5Server; - - public CommandRequestHandler(Socks5Server socks5Server) { - this.socks5Server = socks5Server; - } - - @Override - protected void channelRead0(final ChannelHandlerContext clientChannelContext, DefaultSocks5CommandRequest msg) throws Exception { - if (Socks5CommandType.CONNECT.equals(msg.type())) { - Bootstrap bootstrap = new Bootstrap(); - bootstrap.group(socks5Server.getBoss()) - .channel(NioSocketChannel.class) - .option(ChannelOption.TCP_NODELAY, true) - .handler(new ChannelInitializer() { - @Override - protected void initChannel(SocketChannel ch) throws Exception { - ch.pipeline().addLast(new ClientHandler(clientChannelContext)); - } - }); - ChannelFuture future = bootstrap.connect(msg.dstAddr(), msg.dstPort()); - future.addListener(new ChannelFutureListener() { - - public void operationComplete(final ChannelFuture future) throws Exception { - if (future.isSuccess()) { - if (log.isDebugEnabled()) { - log.debug("connected : {} {}", msg.dstAddr(), msg.dstPort()); - } - clientChannelContext.pipeline().addLast(new TargetHandler(future)); - clientChannelContext.writeAndFlush(new DefaultSocks5CommandResponse(Socks5CommandStatus.SUCCESS, Socks5AddressType.IPv4)); - } else { - clientChannelContext.writeAndFlush(new DefaultSocks5CommandResponse(Socks5CommandStatus.FAILURE, Socks5AddressType.IPv4)); - } - } - }); - } else { - clientChannelContext.fireChannelRead(msg); - } - } - - private static class ClientHandler extends ChannelInboundHandlerAdapter { - - private ChannelHandlerContext clientChannelContext; - - public ClientHandler(ChannelHandlerContext clientChannelContext) { - this.clientChannelContext = clientChannelContext; - } - - @Override - public void channelRead(ChannelHandlerContext ctx2, Object destMsg) throws Exception { - clientChannelContext.writeAndFlush(destMsg); - } - - @Override - public void channelInactive(ChannelHandlerContext ctx2) throws Exception { - clientChannelContext.channel().close(); - } - } - - private static class TargetHandler extends ChannelInboundHandlerAdapter { - - private ChannelFuture targetChannel; - - public TargetHandler(ChannelFuture targetChannel) { - this.targetChannel = targetChannel; - } - - @Override - public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { - targetChannel.channel().writeAndFlush(msg); - } - - @Override - public void channelInactive(ChannelHandlerContext ctx) throws Exception { - targetChannel.channel().close(); - } - } + private final Socks5Server socks5Server; + + public CommandRequestHandler(Socks5Server socks5Server) { + this.socks5Server = socks5Server; + } + + @Override + protected void channelRead0(final ChannelHandlerContext clientChannelContext, DefaultSocks5CommandRequest msg) throws Exception { + if (Socks5CommandType.CONNECT.equals(msg.type())) { + Bootstrap bootstrap = new Bootstrap(); + bootstrap.group(socks5Server.getBoss()) + .channel(NioSocketChannel.class) + .option(ChannelOption.TCP_NODELAY, true) + .handler(new ChannelInitializer() { + @Override + protected void initChannel(SocketChannel ch) throws Exception { + ch.pipeline().addLast(new ClientHandler(clientChannelContext)); + } + }); + ChannelFuture future = bootstrap.connect(msg.dstAddr(), msg.dstPort()); + future.addListener(new ChannelFutureListener() { + + public void operationComplete(final ChannelFuture future) throws Exception { + if (future.isSuccess()) { + if (log.isDebugEnabled()) { + log.debug("connected : {} {}", msg.dstAddr(), msg.dstPort()); + } + clientChannelContext.pipeline().addLast(new TargetHandler(future)); + clientChannelContext.writeAndFlush(new DefaultSocks5CommandResponse(Socks5CommandStatus.SUCCESS, Socks5AddressType.IPv4)); + } else { + clientChannelContext.writeAndFlush(new DefaultSocks5CommandResponse(Socks5CommandStatus.FAILURE, Socks5AddressType.IPv4)); + } + } + }); + } else { + clientChannelContext.fireChannelRead(msg); + } + } + + private static class ClientHandler extends ChannelInboundHandlerAdapter { + + private ChannelHandlerContext clientChannelContext; + + public ClientHandler(ChannelHandlerContext clientChannelContext) { + this.clientChannelContext = clientChannelContext; + } + + @Override + public void channelRead(ChannelHandlerContext ctx2, Object destMsg) throws Exception { + clientChannelContext.writeAndFlush(destMsg); + } + + @Override + public void channelInactive(ChannelHandlerContext ctx2) throws Exception { + clientChannelContext.channel().close(); + } + } + + private static class TargetHandler extends ChannelInboundHandlerAdapter { + + private ChannelFuture targetChannel; + + public TargetHandler(ChannelFuture targetChannel) { + this.targetChannel = targetChannel; + } + + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { + targetChannel.channel().writeAndFlush(msg); + } + + @Override + public void channelInactive(ChannelHandlerContext ctx) throws Exception { + targetChannel.channel().close(); + } + } + } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/IdleHandler.java b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/IdleHandler.java index 73f6cbe621c39..a4370d07b1ed5 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/IdleHandler.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/IdleHandler.java @@ -23,14 +23,14 @@ import io.netty.handler.timeout.IdleStateEvent; public class IdleHandler extends ChannelInboundHandlerAdapter { - - @Override - public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { - if (evt instanceof IdleStateEvent) { - ctx.channel().close(); - } else { - super.userEventTriggered(ctx, evt); - } - } - + + @Override + public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { + if (evt instanceof IdleStateEvent) { + ctx.channel().close(); + } else { + super.userEventTriggered(ctx, evt); + } + } + } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/InitialRequestHandler.java b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/InitialRequestHandler.java index ee9dc3984a27b..7fa0550132a3d 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/InitialRequestHandler.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/InitialRequestHandler.java @@ -30,25 +30,26 @@ @Slf4j public class InitialRequestHandler extends SimpleChannelInboundHandler { - private final Socks5Config socks5Config; + private final Socks5Config socks5Config; - public InitialRequestHandler(final Socks5Config socks5Config) { - this.socks5Config = socks5Config; - } + public InitialRequestHandler(final Socks5Config socks5Config) { + this.socks5Config = socks5Config; + } + + @Override + protected void channelRead0(ChannelHandlerContext ctx, DefaultSocks5InitialRequest msg) throws Exception { + if (SocksVersion.SOCKS5.equals(msg.version())) { + if (msg.decoderResult().isFailure()) { + log.warn("decode failure : {}", msg.decoderResult()); + ctx.fireChannelRead(msg); + } else { + if (socks5Config.isEnableAuth()) { + ctx.writeAndFlush(new DefaultSocks5InitialResponse(Socks5AuthMethod.PASSWORD)); + } else { + ctx.writeAndFlush(new DefaultSocks5InitialResponse(Socks5AuthMethod.NO_AUTH)); + } + } + } + } - @Override - protected void channelRead0(ChannelHandlerContext ctx, DefaultSocks5InitialRequest msg) throws Exception { - if (SocksVersion.SOCKS5.equals(msg.version())) { - if (msg.decoderResult().isFailure()) { - log.warn("decode failure : {}", msg.decoderResult()); - ctx.fireChannelRead(msg); - } else { - if (socks5Config.isEnableAuth()) { - ctx.writeAndFlush(new DefaultSocks5InitialResponse(Socks5AuthMethod.PASSWORD)); - } else { - ctx.writeAndFlush(new DefaultSocks5InitialResponse(Socks5AuthMethod.NO_AUTH)); - } - } - } - } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/PasswordAuthRequestHandler.java b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/PasswordAuthRequestHandler.java index 1078f0e87e79a..49db70dcacd59 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/PasswordAuthRequestHandler.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/PasswordAuthRequestHandler.java @@ -31,20 +31,20 @@ @Slf4j public class PasswordAuthRequestHandler extends SimpleChannelInboundHandler { - private final PasswordAuth passwordAuth; + private final PasswordAuth passwordAuth; - public PasswordAuthRequestHandler() { - this.passwordAuth = new DefaultPasswordAuthImpl(); - } + public PasswordAuthRequestHandler() { + this.passwordAuth = new DefaultPasswordAuthImpl(); + } - @Override - protected void channelRead0(ChannelHandlerContext ctx, DefaultSocks5PasswordAuthRequest msg) throws Exception { - if (passwordAuth.auth(msg.username(), msg.password())) { - ctx.writeAndFlush(new DefaultSocks5PasswordAuthResponse(Socks5PasswordAuthStatus.SUCCESS)); - } else { - ctx.writeAndFlush(new DefaultSocks5PasswordAuthResponse(Socks5PasswordAuthStatus.FAILURE)) - .addListener(ChannelFutureListener.CLOSE); - } - } + @Override + protected void channelRead0(ChannelHandlerContext ctx, DefaultSocks5PasswordAuthRequest msg) throws Exception { + if (passwordAuth.auth(msg.username(), msg.password())) { + ctx.writeAndFlush(new DefaultSocks5PasswordAuthResponse(Socks5PasswordAuthStatus.SUCCESS)); + } else { + ctx.writeAndFlush(new DefaultSocks5PasswordAuthResponse(Socks5PasswordAuthStatus.FAILURE)) + .addListener(ChannelFutureListener.CLOSE); + } + } } From 0f5f6cb66dde901a38172f0793c8971ca27ff683 Mon Sep 17 00:00:00 2001 From: technoboy Date: Thu, 1 Jul 2021 14:01:01 +0800 Subject: [PATCH 8/9] fix getSocks5ProxyAddress method --- .../client/impl/conf/ClientConfigurationData.java | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java index fe9e0a373f872..9d5b12d4099c4 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java @@ -25,6 +25,7 @@ import java.net.URI; import java.time.Clock; import java.util.Objects; +import java.util.Optional; import java.util.Set; import lombok.AllArgsConstructor; import lombok.Data; @@ -147,12 +148,14 @@ public InetSocketAddress getSocks5ProxyAddress() { return socks5ProxyAddress; } String proxyAddress = System.getProperty("socks5Proxy.address"); - try { - URI uri = URI.create(proxyAddress); - return new InetSocketAddress(uri.getHost(), uri.getPort()); - } catch (Exception e) { - throw new RuntimeException("Invalid config [socks5Proxy.address]", e); - } + return Optional.ofNullable(proxyAddress).map(address -> { + try { + URI uri = URI.create(address); + return new InetSocketAddress(uri.getHost(), uri.getPort()); + } catch (Exception e) { + throw new RuntimeException("Invalid config [socks5Proxy.address]", e); + } + }).orElse(null); } public String getSocks5ProxyUsername() { From 52c1961b592d32deeadc87381a125a073c6adddd Mon Sep 17 00:00:00 2001 From: technoboy Date: Thu, 1 Jul 2021 16:26:26 +0800 Subject: [PATCH 9/9] fix test case --- .../apache/pulsar/client/impl/ClientWithSocks5ProxyTest.java | 1 + .../pulsar/client/impl/conf/ConfigurationDataUtilsTest.java | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ClientWithSocks5ProxyTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ClientWithSocks5ProxyTest.java index 5aa61cfceec2a..5e2ba4be5b4b5 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ClientWithSocks5ProxyTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ClientWithSocks5ProxyTest.java @@ -87,6 +87,7 @@ public void run() { protected void cleanup() throws Exception { internalCleanup(); server.shutdown(); + System.clearProperty("socks5Proxy.address"); } private void initData() throws PulsarAdminException { diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/conf/ConfigurationDataUtilsTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/conf/ConfigurationDataUtilsTest.java index 61e6f2262ad99..e076a6a1afc58 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/conf/ConfigurationDataUtilsTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/conf/ConfigurationDataUtilsTest.java @@ -227,6 +227,6 @@ public void testSocks5() throws PulsarClientException { } catch (Exception ex) { assertTrue(ex.getMessage().contains("Invalid config [socks5Proxy.address]")); } - + System.clearProperty("socks5Proxy.address"); } }