[Enhancement] Add structural limits for deserialization filter

common/src/main/java/org/opensearch/sql/common/setting/Settings.java

@@ -53,6 +53,9 @@ public enum Key {
     QUERY_MEMORY_LIMIT("plugins.query.memory_limit"),
     QUERY_SIZE_LIMIT("plugins.query.size_limit"),
     MAX_EXPRESSION_DEPTH("plugins.query.max_expression_depth"),
+    DESERIALIZATION_MAX_DEPTH("plugins.query.deserialization.max_depth"),
+    DESERIALIZATION_MAX_REFS("plugins.query.deserialization.max_refs"),
+    DESERIALIZATION_MAX_BYTES("plugins.query.deserialization.max_bytes"),
     QUERY_BUCKET_SIZE("plugins.query.buckets"),
     SEARCH_MAX_BUCKETS("search.max_buckets"),
     ENCYRPTION_MASTER_KEY("plugins.query.datasources.encryption.masterkey"),

core/src/main/java/org/opensearch/sql/executor/pagination/PlanSerializer.java

@@ -17,7 +17,7 @@
 import java.util.zip.Deflater;
 import java.util.zip.GZIPInputStream;
 import java.util.zip.GZIPOutputStream;
-import lombok.RequiredArgsConstructor;
+import org.opensearch.sql.common.setting.Settings;
 import org.opensearch.sql.exception.NoCursorException;
 import org.opensearch.sql.planner.SerializablePlan;
 import org.opensearch.sql.planner.physical.PhysicalPlan;
@@ -28,12 +28,23 @@
  * This class is entry point to paged requests. It is responsible to cursor serialization and
  * deserialization.
  */
-@RequiredArgsConstructor
 public class PlanSerializer {
   public static final String CURSOR_PREFIX = "n:";
 
   private final StorageEngine engine;
 
+  /** Cluster settings supplying deserialization structural limits; null falls back to defaults. */
+  private final Settings settings;
+
+  public PlanSerializer(StorageEngine engine) {
+    this(engine, null);
+  }
+
+  public PlanSerializer(StorageEngine engine, Settings settings) {
+    this.engine = engine;
+    this.settings = settings;
+  }
+
   /** Converts a physical plan tree to a cursor. */
   public Cursor convertToCursor(PhysicalPlan plan) {
     try {
@@ -89,14 +100,17 @@ protected Serializable deserialize(String code) {
           new GZIPInputStream(new ByteArrayInputStream(HashCode.fromString(code).asBytes()));
       ObjectInputStream objectInput =
           new CursorDeserializationStream(new ByteArrayInputStream(gzip.readAllBytes()));
+      String additionalPatterns =
+          "org.opensearch.sql.planner.physical.*;"
+              + "org.opensearch.sql.opensearch.storage.scan.*;"
+              + "org.opensearch.sql.opensearch.data.type.*;"
+              + "org.opensearch.sql.executor.pagination.*;"
+              + "org.opensearch.sql.executor.QueryType;"
+              + "org.opensearch.sql.utils.*;";
       objectInput.setObjectInputFilter(
-          DeserializationFilterUtil.createFilter(
-              "org.opensearch.sql.planner.physical.*;"
-                  + "org.opensearch.sql.opensearch.storage.scan.*;"
-                  + "org.opensearch.sql.opensearch.data.type.*;"
-                  + "org.opensearch.sql.executor.pagination.*;"
-                  + "org.opensearch.sql.executor.QueryType;"
-                  + "org.opensearch.sql.utils.*;"));
+          settings == null
+              ? DeserializationFilterUtil.createFilter(additionalPatterns)
+              : DeserializationFilterUtil.createFilter(settings, additionalPatterns));
       return (Serializable) objectInput.readObject();
     } catch (Exception e) {
       throw new IllegalStateException("Failed to deserialize object", e);

core/src/main/java/org/opensearch/sql/planner/DefaultImplementor.java

@@ -5,6 +5,7 @@
 
 package org.opensearch.sql.planner;
 
+import org.opensearch.sql.common.setting.Settings;
 import org.opensearch.sql.executor.pagination.PlanSerializer;
 import org.opensearch.sql.planner.logical.LogicalAggregation;
 import org.opensearch.sql.planner.logical.LogicalCloseCursor;
@@ -57,6 +58,17 @@
  */
 public class DefaultImplementor<C> extends LogicalPlanNodeVisitor<PhysicalPlan, C> {
 
+  /** Cluster settings supplying deserialization structural limits; null falls back to defaults. */
+  private final Settings settings;
+
+  public DefaultImplementor() {
+    this(null);
+  }
+
+  public DefaultImplementor(Settings settings) {
+    this.settings = settings;
+  }
+
   @Override
   public PhysicalPlan visitRareTopN(LogicalRareTopN node, C context) {
     return new RareTopNOperator(
@@ -164,7 +176,7 @@ public PhysicalPlan visitRelation(LogicalRelation node, C context) {
 
   @Override
   public PhysicalPlan visitFetchCursor(LogicalFetchCursor plan, C context) {
-    return new PlanSerializer(plan.getEngine()).convertToPlan(plan.getCursor());
+    return new PlanSerializer(plan.getEngine(), settings).convertToPlan(plan.getCursor());
   }
 
   @Override

core/src/main/java/org/opensearch/sql/planner/Planner.java

@@ -6,8 +6,8 @@
 package org.opensearch.sql.planner;
 
 import java.util.List;
-import lombok.RequiredArgsConstructor;
 import org.apache.calcite.rel.RelNode;
+import org.opensearch.sql.common.setting.Settings;
 import org.opensearch.sql.planner.logical.LogicalPlan;
 import org.opensearch.sql.planner.logical.LogicalPlanNodeVisitor;
 import org.opensearch.sql.planner.logical.LogicalRelation;
@@ -17,11 +17,22 @@
 import org.opensearch.sql.storage.read.TableScanBuilder;
 
 /** Planner that plans and chooses the optimal physical plan. */
-@RequiredArgsConstructor
 public class Planner {
 
   private final LogicalPlanOptimizer logicalOptimizer;
 
+  /** Cluster settings supplying deserialization structural limits; null falls back to defaults. */
+  private final Settings settings;
+
+  public Planner(LogicalPlanOptimizer logicalOptimizer) {
+    this(logicalOptimizer, null);
+  }
+
+  public Planner(LogicalPlanOptimizer logicalOptimizer, Settings settings) {
+    this.logicalOptimizer = logicalOptimizer;
+    this.settings = settings;
+  }
+
   /**
    * Generate optimal physical plan for logical plan. If no table involved, translate logical plan
    * to physical by default implementor.<br>
@@ -33,7 +44,7 @@ public class Planner {
   public PhysicalPlan plan(LogicalPlan plan) {
     Table table = findTable(plan);
     if (table == null) {
-      return plan.accept(new DefaultImplementor<>(), null);
+      return plan.accept(new DefaultImplementor<>(settings), null);
     }
     LogicalPlan optimized = table.optimize(optimize(plan));
     // Give scan builders a chance to reject shapes that push-down alone cannot express safely

core/src/main/java/org/opensearch/sql/utils/DeserializationFilterUtil.java

@@ -8,6 +8,7 @@
 import java.io.ObjectInputFilter;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.opensearch.sql.common.setting.Settings;
 
 /** Utility class for creating deserialization filters with logging. */
 public class DeserializationFilterUtil {
@@ -46,6 +47,15 @@ public class DeserializationFilterUtil {
           + "java.time.**;"
           + "com.google.common.collect.**;";
 
+  /**
+   * Default structural limits on the deserialized object graph, used when a setting is unset or no
+   * {@link Settings} is available (serialize-only call sites and tests).
+   */
+  public static final int DEFAULT_MAX_DEPTH = 20;
+
+  public static final int DEFAULT_MAX_REFS = 1000;
+  public static final int DEFAULT_MAX_BYTES = 15000;
+
   /**
    * Creates a logging filter that wraps the provided filter and logs rejected classes.
    *
@@ -55,21 +65,53 @@ public class DeserializationFilterUtil {
   public static ObjectInputFilter createLoggingFilter(ObjectInputFilter filter) {
     return info -> {
       ObjectInputFilter.Status status = filter.checkInput(info);
-      if (status == ObjectInputFilter.Status.REJECTED && info.serialClass() != null) {
-        LOG.warn("Deserialization filter rejected class: {}", info.serialClass().getName());
+      if (status == ObjectInputFilter.Status.REJECTED) {
+        if (info.serialClass() != null) {
+          LOG.warn("Deserialization filter rejected class: {}", info.serialClass().getName());
+        } else {
+          LOG.warn(
+              "Deserialization filter rejected: depth={}, refs={}, bytes={}",
+              info.depth(),
+              info.references(),
+              info.streamBytes());
+        }
       }
       return status;
     };
   }
 
   /**
-   * Creates a filter with the base allowlist plus additional patterns.
+   * Creates a filter with the base allowlist, the built-in default structural limits, and
+   * additional patterns. Used by serialize-only call sites and tests that have no {@link Settings}.
    *
    * @param additionalPatterns Additional patterns to append to the base allowlist.
-   * @return A logging filter with the combined allowlist.
+   * @return A logging filter with the combined allowlist and default structural limits.
    */
   public static ObjectInputFilter createFilter(String additionalPatterns) {
-    String fullPattern = BASE_ALLOWLIST + additionalPatterns + "!*";
+    return createFilter(DEFAULT_MAX_DEPTH, DEFAULT_MAX_REFS, DEFAULT_MAX_BYTES, additionalPatterns);
+  }
+
+  /**
+   * Creates a filter with the base allowlist, the structural limits from the {@code
+   * plugins.query.deserialization.*} cluster settings, and additional patterns.
+   *
+   * @param settings cluster settings supplying the structural limits (must be non-null)
+   * @param additionalPatterns Additional patterns to append to the base allowlist.
+   * @return A logging filter with the combined allowlist and configured structural limits.
+   */
+  public static ObjectInputFilter createFilter(Settings settings, String additionalPatterns) {
+    return createFilter(
+        settings.getSettingValue(Settings.Key.DESERIALIZATION_MAX_DEPTH),
+        settings.getSettingValue(Settings.Key.DESERIALIZATION_MAX_REFS),
+        settings.getSettingValue(Settings.Key.DESERIALIZATION_MAX_BYTES),
+        additionalPatterns);
+  }
+
+  private static ObjectInputFilter createFilter(
+      int maxDepth, int maxRefs, int maxBytes, String additionalPatterns) {
+    String structuralLimits =
+        String.format("maxdepth=%d;maxrefs=%d;maxbytes=%d;", maxDepth, maxRefs, maxBytes);
+    String fullPattern = BASE_ALLOWLIST + additionalPatterns + structuralLimits + "!*";
     return createLoggingFilter(ObjectInputFilter.Config.createFilter(fullPattern));
   }
 }

core/src/test/java/org/opensearch/sql/utils/DeserializationFilterUtilTest.java

@@ -0,0 +1,130 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.utils;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.io.ObjectInputFilter;
+import java.util.List;
+import java.util.Map;
+import org.junit.jupiter.api.DisplayNameGeneration;
+import org.junit.jupiter.api.DisplayNameGenerator;
+import org.junit.jupiter.api.Test;
+import org.opensearch.sql.common.setting.Settings;
+
+@DisplayNameGeneration(DisplayNameGenerator.ReplaceUnderscores.class)
+public class DeserializationFilterUtilTest {
+
+  @Test
+  void allowlisted_class_is_allowed() {
+    ObjectInputFilter filter = DeserializationFilterUtil.createFilter("");
+    assertEquals(
+        ObjectInputFilter.Status.ALLOWED,
+        filter.checkInput(info(String.class, /*depth*/ 1, /*refs*/ 1, /*bytes*/ 100)));
+  }
+
+  @Test
+  void disallowed_class_is_rejected() {
+    ObjectInputFilter filter = DeserializationFilterUtil.createFilter("");
+    assertEquals(
+        ObjectInputFilter.Status.REJECTED,
+        filter.checkInput(info(java.net.URL.class, /*depth*/ 1, /*refs*/ 1, /*bytes*/ 100)));
+  }
+
+  @Test
+  void null_settings_fall_back_to_default_limits() {
+    // depth 21 > DEFAULT_MAX_DEPTH (20); refs 1001 > DEFAULT_MAX_REFS (1000);
+    // bytes 15001 > DEFAULT_MAX_BYTES (15000).
+    ObjectInputFilter filter = DeserializationFilterUtil.createFilter("");
+    assertEquals(
+        ObjectInputFilter.Status.REJECTED,
+        filter.checkInput(info(/*class*/ null, /*depth*/ 21, /*refs*/ 1, /*bytes*/ 100)));
+    assertEquals(
+        ObjectInputFilter.Status.REJECTED,
+        filter.checkInput(info(/*class*/ null, /*depth*/ 1, /*refs*/ 1001, /*bytes*/ 100)));
+    assertEquals(
+        ObjectInputFilter.Status.REJECTED,
+        filter.checkInput(info(/*class*/ null, /*depth*/ 1, /*refs*/ 1, /*bytes*/ 15001)));
+  }
+
+  @Test
+  void limits_are_read_from_settings() {
+    // Settings override the defaults; the filter must enforce the configured values.
+    Settings settings = settingsWith(/*depth*/ 5, /*refs*/ 10, /*bytes*/ 100);
+    ObjectInputFilter filter = DeserializationFilterUtil.createFilter(settings, "");
+    assertEquals(
+        ObjectInputFilter.Status.REJECTED,
+        filter.checkInput(info(/*class*/ null, /*depth*/ 6, /*refs*/ 1, /*bytes*/ 50)));
+    assertEquals(
+        ObjectInputFilter.Status.REJECTED,
+        filter.checkInput(info(/*class*/ null, /*depth*/ 1, /*refs*/ 11, /*bytes*/ 50)));
+    assertEquals(
+        ObjectInputFilter.Status.REJECTED,
+        filter.checkInput(info(/*class*/ null, /*depth*/ 1, /*refs*/ 1, /*bytes*/ 101)));
+    assertEquals(
+        ObjectInputFilter.Status.ALLOWED,
+        filter.checkInput(info(String.class, /*depth*/ 1, /*refs*/ 1, /*bytes*/ 50)));
+  }
+
+  @Test
+  void additional_pattern_is_honored() {
+    // Patterns passed in should extend the base allowlist.
+    ObjectInputFilter filter = DeserializationFilterUtil.createFilter("java.net.URI;");
+    assertEquals(
+        ObjectInputFilter.Status.ALLOWED,
+        filter.checkInput(info(java.net.URI.class, /*depth*/ 1, /*refs*/ 1, /*bytes*/ 100)));
+  }
+
+  private static Settings settingsWith(int depth, int refs, int bytes) {
+    Map<Settings.Key, Object> values =
+        Map.of(
+            Settings.Key.DESERIALIZATION_MAX_DEPTH, depth,
+            Settings.Key.DESERIALIZATION_MAX_REFS, refs,
+            Settings.Key.DESERIALIZATION_MAX_BYTES, bytes);
+    return new Settings() {
+      @Override
+      @SuppressWarnings("unchecked")
+      public <T> T getSettingValue(Settings.Key key) {
+        return (T) values.get(key);
+      }
+
+      @Override
+      public List<?> getSettings() {
+        return List.of();
+      }
+    };
+  }
+
+  private static ObjectInputFilter.FilterInfo info(
+      Class<?> cls, long depth, long refs, long bytes) {
+    return new ObjectInputFilter.FilterInfo() {
+      @Override
+      public Class<?> serialClass() {
+        return cls;
+      }
+
+      @Override
+      public long arrayLength() {
+        return -1;
+      }
+
+      @Override
+      public long depth() {
+        return depth;
+      }
+
+      @Override
+      public long references() {
+        return refs;
+      }
+
+      @Override
+      public long streamBytes() {
+        return bytes;
+      }
+    };
+  }
+}