ARTEMIS-3915 Support PROXY Protocol

This commit implements support the PROXY protocol so that the broker
will be able to determine a client's original IP address despite the
client's connection coming through a reverse proxy (e.g. HAProxy,
ngingx, etc.). Changes include:

 - A new Netty handler to extract relevant details out of PROXY Protocol
   messages and make them available to the broker
 - A new Netty handler to enforce the acceptor's PROXY Protocol config
 - A new chapter in the user manual
 - Updated logging to use this new data
 - Expose this new data via management
 - Disambiguate some variables names related to SOCKS proxy support
 - Embedded Netty-based PROXY server implementation for testing

Author: jbertram

artemis-commons/pom.xml

@@ -79,6 +79,10 @@
          <groupId>io.netty</groupId>
          <artifactId>netty-common</artifactId>
       </dependency>
+      <dependency>
+         <groupId>io.netty</groupId>
+         <artifactId>netty-transport</artifactId>
+      </dependency>
       <dependency>
          <groupId>commons-beanutils</groupId>
          <artifactId>commons-beanutils</artifactId>

artemis-commons/src/main/java/org/apache/activemq/artemis/utils/ProxyProtocolUtil.java

@@ -0,0 +1,60 @@
+/*
+ * 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.activemq.artemis.utils;
+
+import io.netty.channel.Channel;
+import io.netty.util.AttributeKey;
+
+public class ProxyProtocolUtil {
+   public static final AttributeKey<String> PROXY_PROTOCOL_SOURCE_ADDRESS = AttributeKey.valueOf("proxyProtocolSourceAddress");
+   public static final AttributeKey<String> PROXY_PROTOCOL_DESTINATION_ADDRESS = AttributeKey.valueOf("proxyProtocolDestinationAddress");
+   public static final AttributeKey<String> PROXY_PROTOCOL_VERSION = AttributeKey.valueOf("proxyProtocolVersion");
+
+   /**
+    * {@return a string representation of the remote address of this Channel taking into account whether the PROXY
+    * protocol was used by the remote client}
+    */
+   public static String getRemoteAddress(Channel channel) {
+      String addressFromAttribute = getAddressFromAttribute(PROXY_PROTOCOL_SOURCE_ADDRESS, channel);
+      if (addressFromAttribute != null) {
+         return addressFromAttribute;
+      } else {
+         return SocketAddressUtil.toString(channel.remoteAddress());
+      }
+   }
+
+   /**
+    * {@return a string representation of the proxy address of this Channel if and only if the PROXY protocol was used
+    * by the remote client; otherwise null}
+    */
+   public static String getProxyAddress(Channel channel) {
+      return getAddressFromAttribute(PROXY_PROTOCOL_DESTINATION_ADDRESS, channel);
+   }
+
+   private static String getAddressFromAttribute(AttributeKey<String> addressKey, Channel channel) {
+      String address = channel.attr(addressKey).get();
+      if (address != null && !address.isEmpty()) {
+         return address;
+      } else {
+         return null;
+      }
+   }
+
+   public static String getProxyProtocolVersion(Channel channel) {
+      return channel.attr(PROXY_PROTOCOL_VERSION).get() == null ? null : channel.attr(PROXY_PROTOCOL_VERSION).get().toString();
+   }
+}

artemis-commons/src/main/java/org/apache/activemq/artemis/utils/SocketAddressUtil.java

@@ -0,0 +1,37 @@
+/*
+ * 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.activemq.artemis.utils;
+
+import java.net.SocketAddress;
+
+public class SocketAddressUtil {
+
+   /**
+    * {@return a string representation of the {@code SocketAddress} with the leading "/" removed (if applicable)}
+    */
+   public static String toString(SocketAddress address) {
+      if (address == null) {
+         return null;
+      }
+      String result = address.toString();
+      if (result.startsWith("/")) {
+         return result.substring(1);
+      } else {
+         return result;
+      }
+   }
+}

artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnection.java

@@ -46,6 +46,7 @@
 import org.apache.activemq.artemis.utils.ConfigurationHelper;
 import org.apache.activemq.artemis.utils.Env;
 import org.apache.activemq.artemis.utils.IPV6Util;
+import org.apache.activemq.artemis.utils.SocketAddressUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -387,17 +388,8 @@ private static void flushAndWait(final Channel channel, final ChannelPromise pro
    }
 
    @Override
-   public final String getRemoteAddress() {
-      SocketAddress address = channel.remoteAddress();
-      if (address == null) {
-         return null;
-      }
-      String result = address.toString();
-      if (result.startsWith("/")) {
-         return result.substring(1);
-      } else {
-         return result;
-      }
+   public String getRemoteAddress() {
+      return SocketAddressUtil.toString(channel.remoteAddress());
    }
 
    @Override
@@ -477,7 +469,7 @@ private boolean isLocalhost(String hostname) {
 
    @Override
    public final String toString() {
-      return super.toString() + "[ID=" + getID() + ", local= " + channel.localAddress() + ", remote=" + channel.remoteAddress() + "]";
+      return super.toString() + "[ID=" + getID() + ", local= " + channel.localAddress() + ", remote=" + getRemoteAddress() + "]";
    }
 
    private void closeChannel(final Channel channel, boolean inEventLoop) {
@@ -489,4 +481,4 @@ private void closeChannel(final Channel channel, boolean inEventLoop) {
       }
    }
 
-}
+}

artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnector.java

@@ -211,19 +211,19 @@ public class NettyConnector extends AbstractConnector {
    // will be handled by the server's http server.
    private boolean httpUpgradeEnabled;
 
-   private boolean proxyEnabled;
+   private boolean socksEnabled;
 
-   private String proxyHost;
+   private String socksHost;
 
-   private int proxyPort;
+   private int socksPort;
 
-   private SocksVersion proxyVersion;
+   private SocksVersion socksVersion;
 
-   private String proxyUsername;
+   private String socksUsername;
 
-   private String proxyPassword;
+   private String socksPassword;
 
-   private boolean proxyRemoteDNS;
+   private boolean socksRemoteDNS;
 
    private boolean useServlet;
 
@@ -382,18 +382,18 @@ public NettyConnector(final Map<String, Object> configuration,
 
       httpUpgradeEnabled = ConfigurationHelper.getBooleanProperty(TransportConstants.HTTP_UPGRADE_ENABLED_PROP_NAME, TransportConstants.DEFAULT_HTTP_UPGRADE_ENABLED, configuration);
 
-      proxyEnabled = ConfigurationHelper.getBooleanProperty(TransportConstants.PROXY_ENABLED_PROP_NAME, TransportConstants.DEFAULT_PROXY_ENABLED, configuration);
-      if (proxyEnabled) {
-         proxyHost = ConfigurationHelper.getStringProperty(TransportConstants.PROXY_HOST_PROP_NAME, TransportConstants.DEFAULT_PROXY_HOST, configuration);
-         proxyPort = ConfigurationHelper.getIntProperty(TransportConstants.PROXY_PORT_PROP_NAME, TransportConstants.DEFAULT_PROXY_PORT, configuration);
+      socksEnabled = ConfigurationHelper.getBooleanProperty(TransportConstants.SOCKS_ENABLED_PROP_NAME, TransportConstants.DEFAULT_SOCKS_ENABLED, configuration);
+      if (socksEnabled) {
+         socksHost = ConfigurationHelper.getStringProperty(TransportConstants.SOCKS_HOST_PROP_NAME, TransportConstants.DEFAULT_SOCKS_HOST, configuration);
+         socksPort = ConfigurationHelper.getIntProperty(TransportConstants.SOCKS_PORT_PROP_NAME, TransportConstants.DEFAULT_SOCKS_PORT, configuration);
 
-         int socksVersionNumber = ConfigurationHelper.getIntProperty(TransportConstants.PROXY_VERSION_PROP_NAME, TransportConstants.DEFAULT_PROXY_VERSION, configuration);
-         proxyVersion = SocksVersion.valueOf((byte) socksVersionNumber);
+         int socksVersionNumber = ConfigurationHelper.getIntProperty(TransportConstants.SOCKS_VERSION_PROP_NAME, TransportConstants.DEFAULT_SOCKS_VERSION, configuration);
+         socksVersion = SocksVersion.valueOf((byte) socksVersionNumber);
 
-         proxyUsername = ConfigurationHelper.getStringProperty(TransportConstants.PROXY_USERNAME_PROP_NAME, TransportConstants.DEFAULT_PROXY_USERNAME, configuration);
-         proxyPassword = ConfigurationHelper.getStringProperty(TransportConstants.PROXY_PASSWORD_PROP_NAME, TransportConstants.DEFAULT_PROXY_PASSWORD, configuration);
+         socksUsername = ConfigurationHelper.getStringProperty(TransportConstants.SOCKS_USERNAME_PROP_NAME, TransportConstants.DEFAULT_SOCKS_USERNAME, configuration);
+         socksPassword = ConfigurationHelper.getStringProperty(TransportConstants.SOCKS_PASSWORD_PROP_NAME, TransportConstants.DEFAULT_SOCKS_PASSWORD, configuration);
 
-         proxyRemoteDNS = ConfigurationHelper.getBooleanProperty(TransportConstants.PROXY_REMOTE_DNS_PROP_NAME, TransportConstants.DEFAULT_PROXY_REMOTE_DNS, configuration);
+         socksRemoteDNS = ConfigurationHelper.getBooleanProperty(TransportConstants.SOCKS_REMOTE_DNS_PROP_NAME, TransportConstants.SOCKS_PROXY_REMOTE_DNS, configuration);
       }
 
       remotingThreads = ConfigurationHelper.getIntProperty(TransportConstants.NIO_REMOTING_THREADS_PROPNAME, -1, configuration);
@@ -656,19 +656,19 @@ public synchronized void start() {
          public void initChannel(Channel channel) throws Exception {
             final ChannelPipeline pipeline = channel.pipeline();
 
-            if (proxyEnabled && (proxyRemoteDNS || !isTargetLocalHost())) {
-               InetSocketAddress proxyAddress = new InetSocketAddress(proxyHost, proxyPort);
-               ProxyHandler proxyHandler = switch (proxyVersion) {
-                  case SOCKS5 -> new Socks5ProxyHandler(proxyAddress, proxyUsername, proxyPassword);
-                  case SOCKS4a -> new Socks4ProxyHandler(proxyAddress, proxyUsername);
+            if (socksEnabled && (socksRemoteDNS || !isTargetLocalHost())) {
+               InetSocketAddress proxyAddress = new InetSocketAddress(socksHost, socksPort);
+               ProxyHandler proxyHandler = switch (socksVersion) {
+                  case SOCKS5 -> new Socks5ProxyHandler(proxyAddress, socksUsername, socksPassword);
+                  case SOCKS4a -> new Socks4ProxyHandler(proxyAddress, socksUsername);
                   default -> throw new IllegalArgumentException("Unknown SOCKS proxy version");
                };
 
                channel.pipeline().addLast(proxyHandler);
 
-               logger.debug("Using a SOCKS proxy at {}:{}", proxyHost, proxyPort);
+               logger.debug("Using a SOCKS proxy at {}:{}", socksHost, socksPort);
 
-               if (proxyRemoteDNS) {
+               if (socksRemoteDNS) {
                   bootstrap.resolver(NoopAddressResolverGroup.INSTANCE);
                }
             }
@@ -876,7 +876,7 @@ public final Connection createConnection(Consumer<ChannelFuture> onConnect) {
 
    public NettyConnection createConnection(Consumer<ChannelFuture> onConnect, String host, int port) {
       InetSocketAddress remoteDestination;
-      if (proxyEnabled && proxyRemoteDNS) {
+      if (socksEnabled && socksRemoteDNS) {
          remoteDestination = InetSocketAddress.createUnresolved(IPV6Util.stripBracketsAndZoneID(host), port);
       } else {
          remoteDestination = new InetSocketAddress(IPV6Util.stripBracketsAndZoneID(host), port);

artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/TransportConstants.java

@@ -36,6 +36,8 @@ public class TransportConstants {
 
    public static final String SSL_ENABLED_PROP_NAME = "sslEnabled";
 
+   public static final String PROXY_PROTOCOL_ENABLED_PROP_NAME = "proxyProtocolEnabled";
+
    public static final String SSL_AUTO_RELOAD_PROP_NAME = "sslAutoReload";
 
    public static final boolean DEFAULT_SSL_AUTO_RELOAD = false;
@@ -186,26 +188,28 @@ public class TransportConstants {
 
    public static final int STOMP_DEFAULT_CONSUMER_WINDOW_SIZE = 10 * 1024; // 10K
 
-   public static final String PROXY_ENABLED_PROP_NAME = "socksEnabled";
+   public static final String SOCKS_ENABLED_PROP_NAME = "socksEnabled";
 
-   public static final String PROXY_HOST_PROP_NAME = "socksHost";
+   public static final String SOCKS_HOST_PROP_NAME = "socksHost";
 
-   public static final String PROXY_PORT_PROP_NAME = "socksPort";
+   public static final String SOCKS_PORT_PROP_NAME = "socksPort";
 
-   public static final String PROXY_VERSION_PROP_NAME = "socksVersion";
+   public static final String SOCKS_VERSION_PROP_NAME = "socksVersion";
 
-   public static final String PROXY_USERNAME_PROP_NAME = "socksUsername";
+   public static final String SOCKS_USERNAME_PROP_NAME = "socksUsername";
 
-   public static final String PROXY_PASSWORD_PROP_NAME = "socksPassword";
+   public static final String SOCKS_PASSWORD_PROP_NAME = "socksPassword";
 
-   public static final String PROXY_REMOTE_DNS_PROP_NAME = "socksRemoteDNS";
+   public static final String SOCKS_REMOTE_DNS_PROP_NAME = "socksRemoteDNS";
 
    public static final String AUTO_START = "autoStart";
 
    public static final boolean DEFAULT_AUTO_START = true;
 
    public static final boolean DEFAULT_SSL_ENABLED = false;
 
+   public static final boolean DEFAULT_PROXY_PROTOCOL_ENABLED = false;
+
    public static final String DEFAULT_SNIHOST_CONFIG = null;
 
    public static final boolean DEFAULT_USE_GLOBAL_WORKER_POOL = true;
@@ -381,19 +385,19 @@ public class TransportConstants {
     */
    public static final int DEFAULT_SHUTDOWN_TIMEOUT = parseDefaultVariable("DEFAULT_SHUTDOWN_TIMEOUT", 3_000);
 
-   public static final boolean DEFAULT_PROXY_ENABLED = false;
+   public static final boolean DEFAULT_SOCKS_ENABLED = false;
 
-   public static final String DEFAULT_PROXY_HOST = null;
+   public static final String DEFAULT_SOCKS_HOST = null;
 
-   public static final int DEFAULT_PROXY_PORT = 0;
+   public static final int DEFAULT_SOCKS_PORT = 0;
 
-   public static final byte DEFAULT_PROXY_VERSION = SocksVersion.SOCKS5.byteValue();
+   public static final byte DEFAULT_SOCKS_VERSION = SocksVersion.SOCKS5.byteValue();
 
-   public static final String DEFAULT_PROXY_USERNAME = null;
+   public static final String DEFAULT_SOCKS_USERNAME = null;
 
-   public static final String DEFAULT_PROXY_PASSWORD = null;
+   public static final String DEFAULT_SOCKS_PASSWORD = null;
 
-   public static final boolean DEFAULT_PROXY_REMOTE_DNS = false;
+   public static final boolean SOCKS_PROXY_REMOTE_DNS = false;
 
    public static final String ROUTER = "router";
 
@@ -479,6 +483,7 @@ private static int parseDefaultVariable(String variableName, int defaultValue) {
       allowableAcceptorKeys.add(TransportConstants.DISABLE_STOMP_SERVER_HEADER);
       allowableAcceptorKeys.add(TransportConstants.AUTO_START);
       allowableAcceptorKeys.add(TransportConstants.ROUTER);
+      allowableAcceptorKeys.add(TransportConstants.PROXY_PROTOCOL_ENABLED_PROP_NAME);
 
       ALLOWABLE_ACCEPTOR_KEYS = Collections.unmodifiableSet(allowableAcceptorKeys);
 
@@ -525,13 +530,13 @@ private static int parseDefaultVariable(String variableName, int defaultValue) {
       allowableConnectorKeys.add(TransportConstants.NIO_REMOTING_THREADS_PROPNAME);
       allowableConnectorKeys.add(TransportConstants.REMOTING_THREADS_PROPNAME);
       allowableConnectorKeys.add(TransportConstants.BATCH_DELAY);
-      allowableConnectorKeys.add(TransportConstants.PROXY_ENABLED_PROP_NAME);
-      allowableConnectorKeys.add(TransportConstants.PROXY_HOST_PROP_NAME);
-      allowableConnectorKeys.add(TransportConstants.PROXY_PORT_PROP_NAME);
-      allowableConnectorKeys.add(TransportConstants.PROXY_VERSION_PROP_NAME);
-      allowableConnectorKeys.add(TransportConstants.PROXY_USERNAME_PROP_NAME);
-      allowableConnectorKeys.add(TransportConstants.PROXY_PASSWORD_PROP_NAME);
-      allowableConnectorKeys.add(TransportConstants.PROXY_REMOTE_DNS_PROP_NAME);
+      allowableConnectorKeys.add(TransportConstants.SOCKS_ENABLED_PROP_NAME);
+      allowableConnectorKeys.add(TransportConstants.SOCKS_HOST_PROP_NAME);
+      allowableConnectorKeys.add(TransportConstants.SOCKS_PORT_PROP_NAME);
+      allowableConnectorKeys.add(TransportConstants.SOCKS_VERSION_PROP_NAME);
+      allowableConnectorKeys.add(TransportConstants.SOCKS_USERNAME_PROP_NAME);
+      allowableConnectorKeys.add(TransportConstants.SOCKS_PASSWORD_PROP_NAME);
+      allowableConnectorKeys.add(TransportConstants.SOCKS_REMOTE_DNS_PROP_NAME);
       allowableConnectorKeys.add(ActiveMQDefaultConfiguration.getPropMaskPassword());
       allowableConnectorKeys.add(ActiveMQDefaultConfiguration.getPropPasswordCodec());
       allowableConnectorKeys.add(TransportConstants.NETTY_CONNECT_TIMEOUT);

artemis-features/src/main/resources/features.xml

@@ -35,6 +35,7 @@
 		<bundle>mvn:io.netty/netty-buffer/${netty.version}</bundle>
 		<bundle>mvn:io.netty/netty-codec/${netty.version}</bundle>
 		<bundle>mvn:io.netty/netty-codec-socks/${netty.version}</bundle>
+		<bundle>mvn:io.netty/netty-codec-haproxy/${netty.version}</bundle>
 		<bundle>mvn:io.netty/netty-codec-http/${netty.version}</bundle>
 		<bundle>mvn:io.netty/netty-handler/${netty.version}</bundle>
 		<bundle>mvn:io.netty/netty-handler-proxy/${netty.version}</bundle>

artemis-pom/pom.xml

@@ -431,6 +431,12 @@
             <version>${netty.version}</version>
             <!-- License: Apache 2.0 -->
          </dependency>
+         <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-codec-haproxy</artifactId>
+            <version>${netty.version}</version>
+            <!-- License: Apache 2.0 -->
+         </dependency>
          <dependency>
             <groupId>io.netty</groupId>
             <artifactId>netty-codec-socks</artifactId>

artemis-server/pom.xml

@@ -106,6 +106,10 @@
          <groupId>io.netty</groupId>
          <artifactId>netty-codec-http</artifactId>
       </dependency>
+      <dependency>
+         <groupId>io.netty</groupId>
+         <artifactId>netty-codec-haproxy</artifactId>
+      </dependency>
       <dependency>
          <groupId>io.netty</groupId>
          <artifactId>netty-common</artifactId>

artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/view/ConnectionField.java

@@ -29,7 +29,9 @@ public enum ConnectionField {
    LOCAL_ADDRESS("localAddress"),
    SESSION_ID("sessionID"),
    CREATION_TIME("creationTime"),
-   IMPLEMENTATION("implementation");
+   IMPLEMENTATION("implementation"),
+   PROXY_ADDRESS("proxyAddress"),
+   PROXY_PROTOCOL_VERSION("proxyProtocolVersion");
 
    private static final Map<String, ConnectionField> lookup = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);