Add ensureCustomSerialization to ensure that headers are serialized correctly with multiple transport hops

src/main/java/org/opensearch/security/configuration/ClusterInfoHolder.java

@@ -29,6 +29,7 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
+import org.opensearch.Version;
 import org.opensearch.cluster.ClusterChangedEvent;
 import org.opensearch.cluster.ClusterStateListener;
 import org.opensearch.cluster.node.DiscoveryNode;
@@ -67,6 +68,17 @@
         return initialized;
     }
 
+    public Version getMinNodeVersion() {
+        if (nodes == null) {
+            if (log.isDebugEnabled()) {
+                log.debug("Cluster Info Holder not initialized yet for 'nodes'");
+            }
+            return null;
+        }
+
+        return nodes.getMinNodeVersion();
+    }
+
     public Boolean hasNode(DiscoveryNode node) {
         if (nodes == null) {
             if (log.isDebugEnabled()) {

src/main/java/org/opensearch/security/filter/SecurityFilter.java

@@ -185,7 +185,7 @@ private <Request extends ActionRequest, Response extends ActionResponse> void ap
             }
 
             if (threadContext.getTransient(ConfigConstants.USE_JDK_SERIALIZATION) == null) {
-                threadContext.putTransient(ConfigConstants.USE_JDK_SERIALIZATION, false);
+                threadContext.putTransient(ConfigConstants.USE_JDK_SERIALIZATION, true);
             }
 
             final ComplianceConfig complianceConfig = auditLog.getComplianceConfig();

src/main/java/org/opensearch/security/support/Base64Helper.java

@@ -35,11 +35,11 @@ public static String serializeObject(final Serializable object, final boolean us
     }
 
     public static String serializeObject(final Serializable object) {
-        return serializeObject(object, false);
+        return serializeObject(object, true);
     }
 
     public static Serializable deserializeObject(final String string) {
-        return deserializeObject(string, false);
+        return deserializeObject(string, true);
     }
 
     public static Serializable deserializeObject(final String string, final boolean useJDKDeserialization) {
@@ -69,4 +69,28 @@ public static String ensureJDKSerialized(final String string) {
         // If we see an exception now, we want the caller to see it -
         return Base64Helper.serializeObject(serializable, true);
     }
+
+    /**
+     * Ensures that the returned string is custom serialized.
+     *
+     * If the supplied string is a JDK serialized representation, will deserialize it and further serialize using
+     * custom, otherwise returns the string as is.
+     *
+     * @param string original string, can be JDK or custom serialized
+     * @return custom serialized string
+     */
+    public static String ensureCustomSerialized(final String string) {
+        Serializable serializable;
+        try {
+            serializable = Base64Helper.deserializeObject(string, true);
+        } catch (Exception e) {
+            // We received an exception when de-serializing the given string. It is probably custom serialized.
+            // Try to deserialize using custom
+            Base64Helper.deserializeObject(string, false);
+            // Since we could deserialize the object using custom, the string is already custom serialized, return as is
+            return string;
+        }
+        // If we see an exception now, we want the caller to see it -
+        return Base64Helper.serializeObject(serializable, false);
+    }
 }

src/main/java/org/opensearch/security/transport/SecurityInterceptor.java

@@ -39,6 +39,7 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
+import org.opensearch.Version;
 import org.opensearch.action.admin.cluster.shards.ClusterSearchShardsAction;
 import org.opensearch.action.admin.cluster.shards.ClusterSearchShardsResponse;
 import org.opensearch.action.get.GetRequest;
@@ -231,13 +232,22 @@
             }
 
             try {
-                if (serializationFormat == SerializationFormat.JDK) {
-                    Map<String, String> jdkSerializedHeaders = new HashMap<>();
-                    HeaderHelper.getAllSerializedHeaderNames()
-                        .stream()
-                        .filter(k -> headerMap.get(k) != null)
-                        .forEach(k -> jdkSerializedHeaders.put(k, Base64Helper.ensureJDKSerialized(headerMap.get(k))));
-                    headerMap.putAll(jdkSerializedHeaders);
+                if (clusterInfoHolder.getMinNodeVersion() == null || clusterInfoHolder.getMinNodeVersion().before(Version.V_2_14_0)) {
+                    if (serializationFormat == SerializationFormat.JDK) {
+                        Map<String, String> jdkSerializedHeaders = new HashMap<>();
+                        HeaderHelper.getAllSerializedHeaderNames()
+                            .stream()
+                            .filter(k -> headerMap.get(k) != null)
+                            .forEach(k -> jdkSerializedHeaders.put(k, Base64Helper.ensureJDKSerialized(headerMap.get(k))));
+                        headerMap.putAll(jdkSerializedHeaders);
+                    } else if (serializationFormat == SerializationFormat.CustomSerializer_2_11) {
+                        Map<String, String> customSerializedHeaders = new HashMap<>();
+                        HeaderHelper.getAllSerializedHeaderNames()
+                            .stream()
+                            .filter(k -> headerMap.get(k) != null)
+                            .forEach(k -> customSerializedHeaders.put(k, Base64Helper.ensureCustomSerialized(headerMap.get(k))));
+                        headerMap.putAll(customSerializedHeaders);
+                    }
                 }
                 getThreadContext().putHeader(headerMap);
             } catch (IllegalArgumentException iae) {

src/test/java/org/opensearch/security/support/Base64HelperTest.java

@@ -53,6 +53,15 @@ public void testEnsureJDKSerialized() {
         assertThat(Base64Helper.ensureJDKSerialized(customSerialized), is(jdkSerialized));
     }
 
+    @Test
+    public void testEnsureCustomSerialized() {
+        String test = "string";
+        String jdkSerialized = Base64Helper.serializeObject(test, true);
+        String customSerialized = Base64Helper.serializeObject(test, false);
+        assertThat(Base64Helper.ensureCustomSerialized(jdkSerialized), is(customSerialized));
+        assertThat(Base64Helper.ensureCustomSerialized(customSerialized), is(customSerialized));
+    }
+
     @Test
     public void testDuplicatedItemSizes() {
         var largeObject = new HashMap<String, Object>();