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..5e2ba4be5b4b5 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ClientWithSocks5ProxyTest.java @@ -0,0 +1,191 @@ +/** + * 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 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; +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 +@Slf4j +public class ClientWithSocks5ProxyTest extends BrokerTestBase { + + private Socks5Server server; + + final String topicName = "persistent://public/default/socks5"; + + @BeforeMethod + public void setup() throws Exception { + baseSetup(); + initData(); + } + + @Override + 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) { + log.error("start socks5 server error", e); + } + } + }); + thread.setDaemon(true); + thread.start(); + } + + @AfterMethod(alwaysRun = true) + protected void cleanup() throws Exception { + internalCleanup(); + server.shutdown(); + System.clearProperty("socks5Proxy.address"); + } + + 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 + 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); + 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..7510b03bec655 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/Socks5Server.java @@ -0,0 +1,88 @@ +/** + * 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..b5181dff08439 --- /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..a6450c5bc6223 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/CommandRequestHandler.java @@ -0,0 +1,119 @@ +/** + * 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..a4370d07b1ed5 --- /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..7fa0550132a3d --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/socks5/handler/InitialRequestHandler.java @@ -0,0 +1,55 @@ +/** + * 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 (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..49db70dcacd59 --- /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-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..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 @@ -20,7 +20,12 @@ import com.fasterxml.jackson.annotation.JsonIgnore; 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.Optional; import java.util.Set; import lombok.AllArgsConstructor; import lombok.Data; @@ -105,6 +110,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; @@ -133,5 +143,26 @@ public ClientConfigurationData clone() { } } + public InetSocketAddress getSocks5ProxyAddress() { + if (Objects.nonNull(socks5ProxyAddress)) { + return socks5ProxyAddress; + } + String proxyAddress = System.getProperty("socks5Proxy.address"); + 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() { + 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 b84d8a41c5101..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 @@ -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,35 @@ 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"); + + 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"); + + 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]")); + } + System.clearProperty("socks5Proxy.address"); + } } 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 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.