[spark] branch branch-3.3 updated: [SPARK-38830][CORE] Warn on corrupted block messages

[do...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

dongjoon pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/branch-3.3 by this push:
     new 82581684c64 [SPARK-38830][CORE] Warn on corrupted block messages
82581684c64 is described below

commit 82581684c64dcd388d757ebde8ec0b9eb2a30816
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Sun Apr 10 17:44:39 2022 -0700

    [SPARK-38830][CORE] Warn on corrupted block messages
    
    ### What changes were proposed in this pull request?
    
    This PR aims to warn when `NettyBlockRpcServer` received a corrupted block RPC message or under attack.
    
    - `IllegalArgumentException`: When the type is unknown/invalid when decoding. This fails at Spark layer.
    - `NegativeArraySizeException`: When the size read is negative. This fails at Spark layer during buffer creation.
    - `IndexOutOfBoundsException`: When the data field isn't matched with the size. This fails at Netty later.
    
    ### Why are the changes needed?
    
    When the RPC messages are corrupted or the servers are under attack, Spark shows `IndexOutOfBoundsException` due to the failure from `Decoder`. Instead of `Exception`, we had better ignore the message with a directional warning message.
    ```
    java.lang.IndexOutOfBoundsException:
        readerIndex(5) + length(602416) exceeds writerIndex(172):
    UnpooledUnsafeDirectByteBuf(ridx: 5, widx: 172, cap: 172/172)
        at io.netty.buffer.AbstractByteBuf.checkReadableBytes0(AbstractByteBuf.java:1477)
        at io.netty.buffer.AbstractByteBuf.checkReadableBytes(AbstractByteBuf.java:1463)
        at io.netty.buffer.UnpooledDirectByteBuf.readBytes(UnpooledDirectByteBuf.java:316)
        at io.netty.buffer.AbstractByteBuf.readBytes(AbstractByteBuf.java:904)
        at org.apache.spark.network.protocol.Encoders$Strings.decode(Encoders.java:45)
        at org.apache.spark.network.shuffle.protocol.UploadBlock.decode(UploadBlock.java:112)
        at org.apache.spark.network.shuffle.protocol.BlockTransferMessage$Decoder.fromByteBuffer(BlockTransferMessage.java:71)
        at org.apache.spark.network.netty.NettyBlockRpcServer.receive(NettyBlockRpcServer.scala:53)
        at org.apache.spark.network.server.TransportRequestHandler.processRpcRequest(TransportRequestHandler.java:161)
        at org.apache.spark.network.server.TransportRequestHandler.handle(TransportRequestHandler.java:109)
        at org.apache.spark.network.server.TransportChannelHandler.channelRead0(TransportChannelHandler.java:140)
        at org.apache.spark.network.server.TransportChannelHandler.channelRead0(TransportChannelHandler.java:53)
    
    ```
    
    ### Does this PR introduce _any_ user-facing change?
    
    Yes, but this clarify the log messages from exceptions, `IndexOutOfBoundsException`.
    
    ### How was this patch tested?
    
    Pass the CIs with newly added test suite.
    
    Closes #36116 from dongjoon-hyun/SPARK-38830.
    
    Authored-by: Dongjoon Hyun <do...@apache.org>
    Signed-off-by: Dongjoon Hyun <do...@apache.org>
    (cherry picked from commit 8ac06474f8cfa8e5619f817aaeea29a77ec8a2a4)
    Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
 .../spark/network/netty/NettyBlockRpcServer.scala  | 18 ++++++-
 .../network/netty/NettyBlockRpcServerSuite.scala   | 59 ++++++++++++++++++++++
 2 files changed, 76 insertions(+), 1 deletion(-)

diff --git a/core/src/main/scala/org/apache/spark/network/netty/NettyBlockRpcServer.scala b/core/src/main/scala/org/apache/spark/network/netty/NettyBlockRpcServer.scala
index 81c878d17c6..f2a1fe49fcf 100644
--- a/core/src/main/scala/org/apache/spark/network/netty/NettyBlockRpcServer.scala
+++ b/core/src/main/scala/org/apache/spark/network/netty/NettyBlockRpcServer.scala
@@ -50,7 +50,23 @@ class NettyBlockRpcServer(
       client: TransportClient,
       rpcMessage: ByteBuffer,
       responseContext: RpcResponseCallback): Unit = {
-    val message = BlockTransferMessage.Decoder.fromByteBuffer(rpcMessage)
+    val message = try {
+      BlockTransferMessage.Decoder.fromByteBuffer(rpcMessage)
+    } catch {
+      case e: IllegalArgumentException if e.getMessage.startsWith("Unknown message type") =>
+        logWarning(s"This could be a corrupted RPC message (capacity: ${rpcMessage.capacity()}) " +
+          s"from ${client.getSocketAddress}. Please use `spark.authenticate.*` configurations " +
+          "in case of security incidents.")
+        throw e
+
+      case _: IndexOutOfBoundsException | _: NegativeArraySizeException =>
+        // Netty may throw non-'IOException's for corrupted buffers. In this case,
+        // we ignore the entire message with warnings because we cannot trust any contents.
+        logWarning(s"Ignored a corrupted RPC message (capacity: ${rpcMessage.capacity()}) " +
+          s"from ${client.getSocketAddress}. Please use `spark.authenticate.*` configurations " +
+          "in case of security incidents.")
+        return
+    }
     logTrace(s"Received request: $message")
 
     message match {
diff --git a/core/src/test/scala/org/apache/spark/network/netty/NettyBlockRpcServerSuite.scala b/core/src/test/scala/org/apache/spark/network/netty/NettyBlockRpcServerSuite.scala
new file mode 100644
index 00000000000..54e83e7bda5
--- /dev/null
+++ b/core/src/test/scala/org/apache/spark/network/netty/NettyBlockRpcServerSuite.scala
@@ -0,0 +1,59 @@
+/*
+ * 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.spark.network.netty
+
+import java.nio.ByteBuffer
+
+import org.mockito.Mockito.mock
+
+import org.apache.spark.{SparkConf, SparkFunSuite}
+import org.apache.spark.network.client.TransportClient
+import org.apache.spark.serializer.JavaSerializer
+
+class NettyBlockRpcServerSuite extends SparkFunSuite {
+
+  test("SPARK-38830: Rethrow IllegalArgumentException due to `Unknown message type`") {
+    val serializer = new JavaSerializer(new SparkConf)
+    val server = new NettyBlockRpcServer("enhanced-rpc-server", serializer, null)
+    val bytes = Array[Byte](100.toByte)
+    val message = ByteBuffer.wrap(bytes)
+    val client = mock(classOf[TransportClient])
+    val m = intercept[IllegalArgumentException] {
+      server.receive(client, message)
+    }.getMessage
+    assert(m.startsWith("Unknown message type: 100"))
+  }
+
+  test("SPARK-38830: Warn and ignore NegativeArraySizeException due to the corruption") {
+    val serializer = new JavaSerializer(new SparkConf)
+    val server = new NettyBlockRpcServer("enhanced-rpc-server", serializer, null)
+    val bytes = Array[Byte](0.toByte, 0xFF.toByte, 0xFF.toByte, 0xFF.toByte, 0xFF.toByte)
+    val message = ByteBuffer.wrap(bytes)
+    val client = mock(classOf[TransportClient])
+    server.receive(client, message)
+  }
+
+  test("SPARK-38830: Warn and ignore IndexOutOfBoundsException due to the corruption") {
+    val serializer = new JavaSerializer(new SparkConf)
+    val server = new NettyBlockRpcServer("enhanced-rpc-server", serializer, null)
+    val bytes = Array[Byte](1.toByte)
+    val message = ByteBuffer.wrap(bytes)
+    val client = mock(classOf[TransportClient])
+    server.receive(client, message)
+  }
+}


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org