[CELEBORN-1212] Support for Anonymous SASL Mechanism

### What changes were proposed in this pull request?
This adds support for ANONYMOUS Sasl Mechanism.

### Why are the changes needed?
The changes are needed for adding authentication to Celeborn. See [CELEBORN-1011](https://issues.apache.org/jira/browse/CELEBORN-1011).

### Does this PR introduce _any_ user-facing change?
No

### How was this patch tested?
Added UT.

Closes apache#2210 from otterc/CELEBORN-1212.

Lead-authored-by: Chandni Singh <[email protected]>
Co-authored-by: otterc <[email protected]>
Signed-off-by: zky.zhoukeyong <[email protected]>

common/src/main/java/org/apache/celeborn/common/network/sasl/CelebornSaslClient.java

@@ -58,6 +58,7 @@ public CelebornSaslClient(
       @Nullable Map<String, String> saslProps,
       @Nullable CallbackHandler authCallbackHandler) {
     Preconditions.checkNotNull(saslMechanism);
+    initializeSaslProviders();
     try {
       this.saslClient =
           Sasl.createSaslClient(

common/src/main/java/org/apache/celeborn/common/network/sasl/CelebornSaslServer.java

@@ -52,6 +52,7 @@ public CelebornSaslServer(
       @Nullable Map<String, String> saslProps,
       @Nullable CallbackHandler callbackHandler) {
     Preconditions.checkNotNull(saslMechanism);
+    initializeSaslProviders();
     try {
       this.saslServer =
           Sasl.createSaslServer(saslMechanism, null, DEFAULT_REALM, saslProps, callbackHandler);

common/src/main/java/org/apache/celeborn/common/network/sasl/SaslUtils.java

@@ -26,12 +26,16 @@
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableMap;
 
+import org.apache.celeborn.common.network.sasl.anonymous.AnonymousSaslProvider;
+
 public class SaslUtils {
   static final byte[] EMPTY_BYTE_ARRAY = new byte[0];
 
   /** Sasl Mechanisms */
   static final String DIGEST_MD5 = "DIGEST-MD5";
 
+  public static final String ANONYMOUS = "ANONYMOUS";
+
   /** Quality of protection value that does not include encryption. */
   static final String QOP_AUTH = "auth";
 
@@ -66,4 +70,8 @@ static char[] encodePassword(String password) {
         .encodeToString(password.getBytes(StandardCharsets.UTF_8))
         .toCharArray();
   }
+
+  static void initializeSaslProviders() {
+    AnonymousSaslProvider.initializeIfNeeded();
+  }
 }

common/src/main/java/org/apache/celeborn/common/network/sasl/anonymous/AnonymousSaslClientFactory.java

@@ -0,0 +1,121 @@
+/*
+ * 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.celeborn.common.network.sasl.anonymous;
+
+import static org.apache.celeborn.common.network.sasl.SaslUtils.*;
+
+import java.util.Map;
+
+import javax.security.auth.callback.CallbackHandler;
+import javax.security.sasl.SaslClient;
+import javax.security.sasl.SaslClientFactory;
+import javax.security.sasl.SaslException;
+
+import com.google.common.base.Preconditions;
+
+/**
+ * This implements the {@code SaslClientFactory} for the ANONYMOUS SASL mechanism. It allows the
+ * creation of SASL clients that can perform ANONYMOUS authentication with a remote server.
+ */
+public class AnonymousSaslClientFactory implements SaslClientFactory {
+
+  /**
+   * Creates a SASL client for the ANONYMOUS mechanism.
+   *
+   * @param mechanisms The list of SASL mechanisms.
+   * @param authorizationId The authorization ID, typically null for ANONYMOUS.
+   * @param protocol The name of the protocol being used.
+   * @param serverName The name of the server.
+   * @param props A map of properties to configure the SASL client.
+   * @param cbh A callback handler for handling challenges.
+   * @return A {@code CelebornAnonymousSaslClient} instance if ANONYMOUS is requested, or null
+   *     otherwise.
+   * @throws SaslException
+   */
+  @Override
+  public SaslClient createSaslClient(
+      String[] mechanisms,
+      String authorizationId,
+      String protocol,
+      String serverName,
+      Map<String, ?> props,
+      CallbackHandler cbh)
+      throws SaslException {
+    Preconditions.checkNotNull(mechanisms);
+    for (String mech : mechanisms) {
+      if (mech.equals(ANONYMOUS)) {
+        return new CelebornAnonymousSaslClient();
+      }
+    }
+    return null;
+  }
+
+  @Override
+  public String[] getMechanismNames(Map<String, ?> props) {
+    return new String[] {ANONYMOUS};
+  }
+
+  class CelebornAnonymousSaslClient implements SaslClient {
+
+    private boolean isCompleted = false;
+
+    @Override
+    public String getMechanismName() {
+      return ANONYMOUS;
+    }
+
+    @Override
+    public boolean hasInitialResponse() {
+      return false;
+    }
+
+    @Override
+    public byte[] evaluateChallenge(byte[] challenge) throws SaslException {
+      if (isCompleted) {
+        throw new IllegalStateException("Authentication has already completed.");
+      }
+      isCompleted = true;
+      return ANONYMOUS.getBytes();
+    }
+
+    @Override
+    public boolean isComplete() {
+      return isCompleted;
+    }
+
+    @Override
+    public byte[] unwrap(byte[] incoming, int offset, int len) throws SaslException {
+      throw new IllegalStateException("ANONYMOUS mechanism does not support wrap/unwrap");
+    }
+
+    @Override
+    public byte[] wrap(byte[] outgoing, int offset, int len) throws SaslException {
+      throw new IllegalStateException("ANONYMOUS mechanism does not support wrap/unwrap");
+    }
+
+    @Override
+    public Object getNegotiatedProperty(String propName) {
+      return null;
+    }
+
+    @Override
+    public void dispose() throws SaslException {
+      // No resources to cleanup for ANONYMOUS
+    }
+  }
+}

common/src/main/java/org/apache/celeborn/common/network/sasl/anonymous/AnonymousSaslProvider.java

@@ -0,0 +1,49 @@
+/*
+ * 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.celeborn.common.network.sasl.anonymous;
+
+import static org.apache.celeborn.common.network.sasl.SaslUtils.*;
+
+import java.security.Provider;
+import java.security.Security;
+
+/**
+ * This is a Java Security Provider that adds support for the ANONYMOUS SASL mechanism. It allows
+ * for the registration of SASL client and server factories for ANONYMOUS authentication.
+ *
+ * <p>This provider registers the necessary SASL factories to enable ANONYMOUS SASL mechanism
+ * authentication.
+ */
+public final class AnonymousSaslProvider extends Provider {
+
+  private static boolean init = false;
+
+  private AnonymousSaslProvider() {
+    super("AnonymousSasl", 1.0, "ANONYMOUS SASL MECHANISM PROVIDER");
+    put("SaslClientFactory." + ANONYMOUS, AnonymousSaslClientFactory.class.getName());
+    put("SaslServerFactory." + ANONYMOUS, AnonymousSaslServerFactory.class.getName());
+  }
+
+  public static synchronized void initializeIfNeeded() {
+    if (!init) {
+      AnonymousSaslProvider provider = new AnonymousSaslProvider();
+      Security.addProvider(provider);
+      init = true;
+    }
+  }
+}

common/src/main/java/org/apache/celeborn/common/network/sasl/anonymous/AnonymousSaslServerFactory.java

@@ -0,0 +1,115 @@
+/*
+ * 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.celeborn.common.network.sasl.anonymous;
+
+import static org.apache.celeborn.common.network.sasl.SaslUtils.*;
+
+import java.util.Map;
+
+import javax.security.auth.callback.CallbackHandler;
+import javax.security.sasl.SaslException;
+import javax.security.sasl.SaslServer;
+import javax.security.sasl.SaslServerFactory;
+
+/**
+ * This implements the {@code SaslServerFactory} interface for the ANONYMOUS SASL mechanism. It
+ * allows the creation of SASL servers that can handle ANONYMOUS authentication requests from
+ * clients.
+ */
+public class AnonymousSaslServerFactory implements SaslServerFactory {
+
+  /**
+   * Creates a SASL server for the ANONYMOUS mechanism.
+   *
+   * @param mechanism The requested SASL mechanism (e.g., ANONYMOUS).
+   * @param protocol The name of the protocol being used.
+   * @param serverName The name of the server.
+   * @param props A map of properties to configure the SASL server.
+   * @param cbh A callback handler for handling authentication callbacks.
+   * @return A {@code CelebornAnonymousSaslServer} instance if ANONYMOUS is requested, or null
+   *     otherwise.
+   * @throws SaslException
+   */
+  @Override
+  public SaslServer createSaslServer(
+      String mechanism,
+      String protocol,
+      String serverName,
+      Map<String, ?> props,
+      CallbackHandler cbh)
+      throws SaslException {
+    if (mechanism.equals(ANONYMOUS)) {
+      return new CelebornAnonymousSaslServer();
+    }
+    return null;
+  }
+
+  @Override
+  public String[] getMechanismNames(Map<String, ?> props) {
+    return new String[] {ANONYMOUS};
+  }
+
+  class CelebornAnonymousSaslServer implements SaslServer {
+    private boolean isCompleted = false;
+
+    @Override
+    public String getMechanismName() {
+      return ANONYMOUS;
+    }
+
+    @Override
+    public byte[] evaluateResponse(byte[] response) throws SaslException {
+      if (isCompleted) {
+        throw new IllegalStateException("Authentication has already completed.");
+      }
+      // Typically, we would process the response here. For ANONYMOUS, we just accept it.
+      isCompleted = true;
+      return new byte[0]; // No challenge is expected for ANONYMOUS.
+    }
+
+    @Override
+    public boolean isComplete() {
+      return isCompleted;
+    }
+
+    @Override
+    public String getAuthorizationID() {
+      return ANONYMOUS;
+    }
+
+    @Override
+    public byte[] unwrap(byte[] incoming, int offset, int len) {
+      throw new IllegalStateException("ANONYMOUS mechanism does not support wrap/unwrap");
+    }
+
+    @Override
+    public byte[] wrap(byte[] outgoing, int offset, int len) {
+      throw new IllegalStateException("ANONYMOUS mechanism does not support wrap/unwrap");
+    }
+
+    @Override
+    public Object getNegotiatedProperty(String propName) {
+      return null;
+    }
+
+    @Override
+    public void dispose() {
+      // Cleanup resources if any.
+    }
+  }
+}

common/src/test/java/org/apache/celeborn/common/network/sasl/CelebornSaslSuiteJ.java

@@ -168,6 +168,27 @@ public void testRpcHandlerDelegate() {
     verify(handler).exceptionCaught(isNull(), isNull());
   }
 
+  @Test
+  public void testAnonymous() {
+    CelebornSaslClient client = new CelebornSaslClient(ANONYMOUS, null, null);
+    CelebornSaslServer server = new CelebornSaslServer(ANONYMOUS, null, null);
+
+    assertFalse(client.isComplete());
+    assertFalse(server.isComplete());
+
+    byte[] clientMessage = client.firstToken();
+    while (!client.isComplete()) {
+      clientMessage = client.response(server.response(clientMessage));
+    }
+    assertTrue(server.isComplete());
+
+    // Disposal should invalidate
+    server.dispose();
+    assertFalse(server.isComplete());
+    client.dispose();
+    assertFalse(client.isComplete());
+  }
+
   private static class SaslTestCtx implements AutoCloseable {
 
     final TransportClient client;