[GEODE-10535] Secure Session Deserialization with Application-Level Security Model using ObjectInputFilter (JEP 290) (#7966)

* Add application-level security using ObjectInputFilter (JEP 290)

- Implement per-application deserialization filtering using standard JEP 290 API
- Add ObjectInputFilter parameter to ClassLoaderObjectInputStream constructor
- Update GemfireHttpSession to read filter configuration from ServletContext
- Add comprehensive security tests covering RCE and DoS prevention
- Add 52 tests validating gadget chain blocking and resource limits
- Add example configuration in session-testing-war web.xml

This provides application-level security isolation, allowing each web application
to define its own deserialization policy independent of cluster configuration.

* Add ObjectInputFilter security documentation for HTTP Session Management

- Add comprehensive security guide for configuring deserialization protection
- Document JEP 290 ObjectInputFilter pattern syntax and examples
- Include best practices, troubleshooting, and migration guidance
- Add navigation link in HTTP Session Management chapter overview

* Address PR review feedback: cache filter, add null check, add logging

- Implement filter caching using double-checked locking with volatile fields to eliminate race conditions and improve performance
- Add null check before setObjectInputFilter() for defensive programming
- Add INFO logging when filter is configured and WARN logging when not configured to improve security visibility

Addresses review comments by @sboorlagadda on PR #7966

Author: JinwooHwang

extensions/geode-modules-session-internal/src/main/java/org/apache/geode/modules/session/internal/filter/GemfireHttpSession.java

@@ -20,6 +20,7 @@
 import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
+import java.io.ObjectInputFilter;
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
 import java.util.Collections;
@@ -78,6 +79,13 @@ public class GemfireHttpSession implements HttpSession, DataSerializable, Delta
 
   private ServletContext context;
 
+  /**
+   * Cached ObjectInputFilter to avoid recreating on every deserialization.
+   * Initialized lazily on first use with double-checked locking.
+   */
+  private volatile ObjectInputFilter cachedFilter;
+  private volatile boolean filterLogged = false;
+
   /**
    * A session becomes invalid if it is explicitly invalidated or if it expires.
    */
@@ -107,6 +115,34 @@ public DataSerializable newInstance() {
     });
   }
 
+  /**
+   * Gets or creates the cached ObjectInputFilter. Uses double-checked locking to avoid
+   * unnecessary synchronization after initialization.
+   *
+   * @return the cached ObjectInputFilter, or null if no filter is configured
+   */
+  private ObjectInputFilter getOrCreateFilter() {
+    if (cachedFilter == null && !filterLogged) {
+      synchronized (this) {
+        if (cachedFilter == null && !filterLogged) {
+          String filterPattern = getServletContext()
+              .getInitParameter("serializable-object-filter");
+
+          if (filterPattern != null) {
+            cachedFilter = ObjectInputFilter.Config.createFilter(filterPattern);
+            LOG.info("ObjectInputFilter configured with pattern: {}", filterPattern);
+          } else {
+            LOG.warn("No ObjectInputFilter configured. Session deserialization is not protected " +
+                "against malicious payloads. Configure 'serializable-object-filter' in web.xml " +
+                "to enable deserialization security.");
+          }
+          filterLogged = true;
+        }
+      }
+    }
+    return cachedFilter;
+  }
+
   /**
    * Constructor used for de-serialization
    */
@@ -144,8 +180,11 @@ public Object getAttribute(String name) {
           oos.writeObject(obj);
           oos.close();
 
+          // Get or create cached filter for secure deserialization
+          ObjectInputFilter filter = getOrCreateFilter();
+
           ObjectInputStream ois = new ClassLoaderObjectInputStream(
-              new ByteArrayInputStream(baos.toByteArray()), loader);
+              new ByteArrayInputStream(baos.toByteArray()), loader, filter);
           tmpObj = ois.readObject();
         } catch (IOException | ClassNotFoundException e) {
           LOG.error("Exception while recreating attribute '" + name + "'", e);

extensions/geode-modules/src/main/java/org/apache/geode/modules/util/ClassLoaderObjectInputStream.java

@@ -16,16 +16,43 @@
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.io.ObjectInputFilter;
 import java.io.ObjectInputStream;
 import java.io.ObjectStreamClass;
 
 /**
  * This class is used when session attributes need to be reconstructed with a new classloader.
+ * It now supports ObjectInputFilter for secure deserialization.
  */
 public class ClassLoaderObjectInputStream extends ObjectInputStream {
 
   private final ClassLoader loader;
 
+  /**
+   * Constructs a ClassLoaderObjectInputStream with an ObjectInputFilter for secure deserialization.
+   *
+   * @param in the input stream to read from
+   * @param loader the ClassLoader to use for class resolution
+   * @param filter the ObjectInputFilter to validate deserialized classes (required for security)
+   * @throws IOException if an I/O error occurs
+   */
+  public ClassLoaderObjectInputStream(InputStream in, ClassLoader loader, ObjectInputFilter filter)
+      throws IOException {
+    super(in);
+    this.loader = loader;
+    if (filter != null) {
+      setObjectInputFilter(filter);
+    }
+  }
+
+  /**
+   * Legacy constructor for backward compatibility.
+   *
+   * @deprecated Use
+   *             {@link #ClassLoaderObjectInputStream(InputStream, ClassLoader, ObjectInputFilter)}
+   *             with a filter for secure deserialization
+   */
+  @Deprecated
   public ClassLoaderObjectInputStream(InputStream in, ClassLoader loader) throws IOException {
     super(in);
     this.loader = loader;

extensions/geode-modules/src/test/java/org/apache/geode/modules/util/ClassLoaderObjectInputStreamTest.java

@@ -21,6 +21,8 @@
 import java.io.ByteArrayOutputStream;
 import java.io.File;
 import java.io.IOException;
+import java.io.InvalidClassException;
+import java.io.ObjectInputFilter;
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
 import java.io.Serializable;
@@ -162,4 +164,142 @@ File getTempFile() {
       return null;
     }
   }
+
+  @Test
+  public void filterRejectsUnauthorizedClasses() throws Exception {
+    // Arrange: Create filter that only allows java.lang and java.util classes
+    ObjectInputFilter filter = ObjectInputFilter.Config.createFilter("java.lang.*;java.util.*;!*");
+    TestSerializable testObject = new TestSerializable("test");
+    byte[] serializedData = serialize(testObject);
+
+    // Act & Assert: Deserialization should be rejected by filter
+    assertThatThrownBy(() -> {
+      try (ClassLoaderObjectInputStream ois = new ClassLoaderObjectInputStream(
+          new ByteArrayInputStream(serializedData),
+          Thread.currentThread().getContextClassLoader(),
+          filter)) {
+        ois.readObject();
+      }
+    }).isInstanceOf(InvalidClassException.class);
+  }
+
+  @Test
+  public void filterAllowsAuthorizedClasses() throws Exception {
+    // Arrange: Create filter that allows this test class package
+    ObjectInputFilter filter = ObjectInputFilter.Config.createFilter(
+        "java.lang.*;java.util.*;org.apache.geode.modules.util.**;!*");
+    TestSerializable testObject = new TestSerializable("test data");
+    byte[] serializedData = serialize(testObject);
+
+    // Act: Deserialize with filter
+    Object deserialized;
+    try (ClassLoaderObjectInputStream ois = new ClassLoaderObjectInputStream(
+        new ByteArrayInputStream(serializedData),
+        Thread.currentThread().getContextClassLoader(),
+        filter)) {
+      deserialized = ois.readObject();
+    }
+
+    // Assert: Object should be successfully deserialized
+    assertThat(deserialized).isInstanceOf(TestSerializable.class);
+    assertThat(((TestSerializable) deserialized).getData()).isEqualTo("test data");
+  }
+
+  @Test
+  public void nullFilterAllowsAllClasses() throws Exception {
+    // Arrange: Null filter means no filtering (backward compatibility)
+    TestSerializable testObject = new TestSerializable("unfiltered data");
+    byte[] serializedData = serialize(testObject);
+
+    // Act: Deserialize with null filter
+    Object deserialized;
+    try (ClassLoaderObjectInputStream ois = new ClassLoaderObjectInputStream(
+        new ByteArrayInputStream(serializedData),
+        Thread.currentThread().getContextClassLoader(),
+        null)) {
+      deserialized = ois.readObject();
+    }
+
+    // Assert: Object should be successfully deserialized
+    assertThat(deserialized).isInstanceOf(TestSerializable.class);
+    assertThat(((TestSerializable) deserialized).getData()).isEqualTo("unfiltered data");
+  }
+
+  @Test
+  public void deprecatedConstructorStillWorks() throws Exception {
+    // Arrange: Use deprecated constructor without filter
+    TestSerializable testObject = new TestSerializable("legacy code");
+    byte[] serializedData = serialize(testObject);
+
+    // Act: Deserialize using deprecated constructor
+    Object deserialized;
+    try (ClassLoaderObjectInputStream ois = new ClassLoaderObjectInputStream(
+        new ByteArrayInputStream(serializedData),
+        Thread.currentThread().getContextClassLoader())) {
+      deserialized = ois.readObject();
+    }
+
+    // Assert: Object should be successfully deserialized (backward compatibility)
+    assertThat(deserialized).isInstanceOf(TestSerializable.class);
+    assertThat(((TestSerializable) deserialized).getData()).isEqualTo("legacy code");
+  }
+
+  @Test
+  public void filterEnforcesResourceLimits() throws Exception {
+    // Arrange: Create filter with very low depth limit
+    ObjectInputFilter filter = ObjectInputFilter.Config.createFilter("maxdepth=2;*");
+    NestedSerializable nested = new NestedSerializable(
+        new NestedSerializable(
+            new NestedSerializable(null))); // Depth of 3
+    byte[] serializedData = serialize(nested);
+
+    // Act & Assert: Should reject due to depth limit
+    assertThatThrownBy(() -> {
+      try (ClassLoaderObjectInputStream ois = new ClassLoaderObjectInputStream(
+          new ByteArrayInputStream(serializedData),
+          Thread.currentThread().getContextClassLoader(),
+          filter)) {
+        ois.readObject();
+      }
+    }).isInstanceOf(InvalidClassException.class);
+  }
+
+  /**
+   * Helper method to serialize an object to byte array
+   */
+  private byte[] serialize(Object obj) throws IOException {
+    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+    try (ObjectOutputStream oos = new ObjectOutputStream(baos)) {
+      oos.writeObject(obj);
+    }
+    return baos.toByteArray();
+  }
+
+  /**
+   * Test class for serialization testing
+   */
+  static class TestSerializable implements Serializable {
+    private static final long serialVersionUID = 1L;
+    private final String data;
+
+    TestSerializable(String data) {
+      this.data = data;
+    }
+
+    String getData() {
+      return data;
+    }
+  }
+
+  /**
+   * Nested test class for depth limit testing
+   */
+  static class NestedSerializable implements Serializable {
+    private static final long serialVersionUID = 1L;
+    private final NestedSerializable nested;
+
+    NestedSerializable(NestedSerializable nested) {
+      this.nested = nested;
+    }
+  }
 }