tabish121 commented on code in PR #5908: URL: https://github.com/apache/activemq-artemis/pull/5908#discussion_r2379854291
########## artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/HAProxyMessageEnforcer.java: ########## @@ -0,0 +1,73 @@ +/* + * 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.core.remoting.impl.netty; + +import io.netty.buffer.ByteBuf; +import io.netty.channel.ChannelHandler; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelInboundHandlerAdapter; +import org.apache.activemq.artemis.core.server.ActiveMQServerLogger; +import org.apache.activemq.artemis.utils.SocketAddressUtil; + +/** + * This Netty handler enforces the presence or absence of PROXY protocol messages. It verifies conformity and then + * removes itself from the pipeline. + * <p> + * If the incoming message protocol does not align with the enforcer's requirements the connection is closed. + */ [email protected] +public class HAProxyMessageEnforcer extends ChannelInboundHandlerAdapter { + + public static final HAProxyMessageEnforcer PROXY_PROTOCOL_REQUIRED = new HAProxyMessageEnforcer(true); + + public static final HAProxyMessageEnforcer PROXY_PROTOCOL_REJECTED = new HAProxyMessageEnforcer(false); + + final boolean requireHaProxyMessage; + + HAProxyMessageEnforcer(boolean requireHaProxyMessage) { + this.requireHaProxyMessage = requireHaProxyMessage; + } + + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { + ByteBuf in = (ByteBuf) msg; + if (in.readableBytes() < 4) { Review Comment: I've realized there is a bug here now that this is acting as a shared static handler that likely means you would want to revert to using new instances with this being a ByteToMessageDecoder subclass. The problem would arise on both proxy enabled and not enabled cases but likely is more of an issue in the non-proxy enabled case where the needed bytes don't arrive in the first packet leaded to dropping the first bytes sent on the connection. I created a test case to add to HAProxyTest that shows this: ``` diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/proxyprotocol/HAProxyTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/proxyprotocol/HAProxyTest.java index f877dea06b..f15878897e 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/proxyprotocol/HAProxyTest.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/proxyprotocol/HAProxyTest.java @@ -51,6 +51,8 @@ import org.apache.activemq.artemis.tests.integration.jms.multiprotocol.Multiprot import org.apache.activemq.artemis.utils.RandomUtil; import org.apache.activemq.artemis.utils.Wait; import org.apache.activemq.transport.netty.NettyHAProxyServer; +import org.apache.qpid.protonj2.test.driver.ProtonTestClient; +import org.apache.qpid.protonj2.test.driver.codec.security.SaslCode; import org.eclipse.paho.mqttv5.client.IMqttToken; import org.eclipse.paho.mqttv5.client.MqttCallback; import org.eclipse.paho.mqttv5.client.MqttClient; @@ -61,6 +63,7 @@ import org.eclipse.paho.mqttv5.common.MqttException; import org.eclipse.paho.mqttv5.common.MqttMessage; import org.eclipse.paho.mqttv5.common.packet.MqttProperties; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -90,7 +93,7 @@ public class HAProxyTest extends MultiprotocolJMSClientTestSupport { protected ActiveMQServer createServer() throws Exception { server = createServer(false, createDefaultNettyConfig() .clearAcceptorConfigurations() - .addAcceptorConfiguration("standard", "tcp://127.0.0.1:" + BROKER_STANDARD_PORT + "?protocols=CORE") + .addAcceptorConfiguration("standard", "tcp://127.0.0.1:" + BROKER_STANDARD_PORT + "?protocols=CORE,AMQP") .addAcceptorConfiguration("proxyEnabled", "tcp://127.0.0.1:" + BROKER_PROXY_PORT + "?proxyProtocolEnabled=true") .addAcceptorConfiguration("proxyAndSslEnabled", "tcp://127.0.0.1:" + BROKER_PROXY_SSL_PORT + "?proxyProtocolEnabled=true;sslEnabled=true;protocols=CORE,AMQP,MQTT,OPENWIRE;supportAdvisory=false;suppressInternalManagementObjects=true;keyStorePath=server-keystore.jks;keyStorePassword=securepass")); @@ -331,6 +334,37 @@ public class HAProxyTest extends MultiprotocolJMSClientTestSupport { producer.disconnect(); } + @Test + @Timeout(30) + public void testBrokerHandlesOutOfOrderDeliveryIdInTransfer() throws Exception { + try (ProtonTestClient receivingPeer = new ProtonTestClient()) { + receivingPeer.connect("localhost", BROKER_STANDARD_PORT); + receivingPeer.waitForScriptToComplete(5, TimeUnit.SECONDS); + + receivingPeer.expectSASLHeader(); + receivingPeer.expectSaslMechanisms().withSaslServerMechanism("ANONYMOUS"); + receivingPeer.remoteSaslInit().withMechanism("ANONYMOUS").queue(); + receivingPeer.expectSaslOutcome().withCode(SaslCode.OK); + receivingPeer.remoteAMQPHeader().queue(); + receivingPeer.expectAMQPHeader(); + + receivingPeer.remoteBytes().withBytes(new byte[] {'A', 'M'}).now(); + receivingPeer.remoteBytes().withBytes(new byte[] {'Q', 'P', 3, 1, 0, 0}).later(10); + + receivingPeer.waitForScriptToComplete(5, TimeUnit.SECONDS); + + // Broker response after SASL anonymous connect + receivingPeer.expectOpen(); + receivingPeer.expectBegin(); + + // Create basic connection with session + receivingPeer.remoteOpen().withContainerId("test-sender").now(); + receivingPeer.remoteBegin().withNextOutgoingId(100).now(); + + receivingPeer.waitForScriptToComplete(); + } + } + private int startV1Proxy() { return startProxy(HAProxyProtocolVersion.V1, BROKER_PROXY_PORT); } ``` The AMQP Header arrives in fragments which leads the initial bit to get dropped leading to the broker never completing the open phase of the AMQP connection. This causes me to ask why the enforcer is needed in the negative case and if there's some other simple changes in the protocol handler that could negate that need as it would be nice to not add more overhead to connections that never expect to have a proxy header as I'd assume is how most folks operate. I'd guess in the majority of cases the connection establishment would fail if the proxy message arrives at the front of the connection request. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected] For further information, visit: https://activemq.apache.org/contact
