HADOOP-19864 Followup: rejection of unregistered protocols. (#8470)

Server will immediately reject any protocols for which no
handler is registered.

---------
Contributed by Steve Loughran
Co-authored-by: Cheng Pan <pan3793@gmail.com>
Contains content written by github copilot

Author: steveloughran

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RPC.java

@@ -1095,9 +1095,22 @@ Map<ProtoNameVer, ProtoClassProtoImpl> getProtocolImplMap(RPC.RpcKind rpcKind) {
              new HashMap<ProtoNameVer, ProtoClassProtoImpl>(10));
        }
      }
-     return protocolImplMapArray.get(rpcKind.ordinal());   
+     return protocolImplMapArray.get(rpcKind.ordinal());
    }
-   
+
+    /**
+     * Returns {@code true} only if at least one protocol has been registered
+     * on this server instance for the given {@link RPC.RpcKind}.
+     * Used to reject incoming requests for unsupported RPC kinds before any
+     * deserialization of the request payload takes place.
+     * @param rpcKind the RPC kind from the incoming request header.
+     * @return {@code true} if at least one protocol is registered for this kind.
+     */
+    boolean hasRegisteredProtocols(RPC.RpcKind rpcKind) {
+      Map<ProtoNameVer, ProtoClassProtoImpl> implMap = getProtocolImplMap(rpcKind);
+      return implMap != null && !implMap.isEmpty();
+    }
+
    // Register  protocol and its impl for rpc calls
    void registerProtocolAndImpl(RpcKind rpcKind, Class<?> protocolClass, 
        Object protocolImpl) {

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java

@@ -350,13 +350,13 @@ public static Server get() {
    * after the call returns.
    */
   private static final ThreadLocal<Call> CurCall = new ThreadLocal<Call>();
-  
+
   /** @return Get the current call. */
   @VisibleForTesting
   public static ThreadLocal<Call> getCurCall() {
     return CurCall;
   }
-  
+
   /**
    * Returns the currently active RPC call's sequential ID number.  A negative
    * call ID indicates an invalid value, such as if there is no currently active
@@ -2869,28 +2869,41 @@ private void checkRpcHeaders(RpcRequestHeaderProto header)
     private void processRpcRequest(RpcRequestHeaderProto header,
         RpcWritable.Buffer buffer) throws RpcServerException,
         InterruptedException {
-      Class<? extends Writable> rpcRequestClass = 
+      // Reject requests for RPC kinds with no registered protocols on this
+      // server instance. This prevents deserialization of untrusted payloads
+      // for unsupported kinds. See HADOOP-19864.
+      if (Server.this instanceof RPC.Server server) {
+        RPC.Server server = (RPC.ServerServer.this;
+        final RPC.RpcKind kind = ProtoUtil.convert(header.getRpcKind());
+        if (!server.hasRegisteredProtocols(kind)) {
+          final String err = "No protocols registered on this server for RpcKind "
+              + header.getRpcKind()
+              + ". Rejecting request without deserialization.";
+          LOG.info("{} Client: {}", err, getHostAddress());
+          throw new FatalRpcServerException(
+              RpcErrorCodeProto.FATAL_INVALID_RPC_HEADER, err);
+        }
+      }
+      Class<? extends Writable> rpcRequestClass =
           getRpcRequestWrapper(header.getRpcKind());
       if (rpcRequestClass == null) {
-        LOG.warn("Unknown rpc kind "  + header.getRpcKind() + 
-            " from client " + getHostAddress());
-        final String err = "Unknown rpc kind in rpc header"  + 
-            header.getRpcKind();
+        LOG.warn("Unknown rpc kind {} from client {}", header.getRpcKind(), getHostAddress());
         throw new FatalRpcServerException(
-            RpcErrorCodeProto.FATAL_INVALID_RPC_HEADER, err);
+            RpcErrorCodeProto.FATAL_INVALID_RPC_HEADER,
+            "Unknown rpc kind in rpc header " + header.getRpcKind());
       }
       Writable rpcRequest;
       try { //Read the rpc request
         rpcRequest = buffer.newInstance(rpcRequestClass, conf);
       } catch (RpcServerException rse) { // lets tests inject failures.
         throw rse;
       } catch (Throwable t) { // includes runtime exception from newInstance
-        LOG.warn("Unable to read call parameters for client " +
-                 getHostAddress() + "on connection protocol " +
-            this.protocolName + " for rpcKind " + header.getRpcKind(),  t);
-        String err = "IPC server unable to read call parameters: "+ t.getMessage();
+        LOG.warn(
+            "Unable to read call parameters for client {} on connection protocol {} for rpcKind {}",
+            getHostAddress(), this.protocolName, header.getRpcKind(), t);
         throw new FatalRpcServerException(
-            RpcErrorCodeProto.FATAL_DESERIALIZING_REQUEST, err);
+            RpcErrorCodeProto.FATAL_DESERIALIZING_REQUEST,
+            "IPC server unable to read call parameters: " + t.getMessage());
       }
 
       Span span = null;

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestRPC.java

@@ -2099,6 +2099,31 @@ public void testNumTotalRequestsMetrics() throws Exception {
   }
 
 
+  /**
+   * Test that a Protobuf-only RPC server rejects requests for RpcKinds
+   * that have no registered protocols, without deserializing the payload.
+   */
+  @Test
+  @Timeout(value = 30)
+  public void testUnregisteredRpcKindRejectedWithoutDeserialization()
+      throws Exception {
+    // Standard test server: only RPC_PROTOCOL_BUFFER protocols are registered.
+    RPC.Server server = setupTestServer(conf, 1);
+    try {
+      // RPC_PROTOCOL_BUFFER has registered protocols — must be accepted.
+      assertThat(server.hasRegisteredProtocols(RPC.RpcKind.RPC_PROTOCOL_BUFFER))
+          .as("RPC_PROTOCOL_BUFFER should have registered protocols")
+          .isTrue();
+
+      // RPC_BUILTIN has no protocols registered on this server — must be rejected.
+      assertThat(server.hasRegisteredProtocols(RPC.RpcKind.RPC_BUILTIN))
+          .as("RPC_BUILTIN should have no registered protocols on a Protobuf-only server")
+          .isFalse();
+    } finally {
+      server.stop();
+    }
+  }
+
   public static void main(String[] args) throws Exception {
     new TestRPC().testCallsInternal(conf);
   }