fix(writer): reject directory-traversal partition paths across all writers

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/CustomAvroKeyGenerator.java

@@ -21,6 +21,7 @@
 import org.apache.hudi.common.config.TypedProperties;
 import org.apache.hudi.common.table.HoodieTableConfig;
 import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.PartitionPathEncodeUtils;
 import org.apache.hudi.common.util.collection.Pair;
 import org.apache.hudi.exception.HoodieKeyException;
 import org.apache.hudi.exception.HoodieKeyGeneratorException;
@@ -161,7 +162,7 @@ public String getPartitionPath(GenericRecord record) {
         partitionPath.append(DEFAULT_PARTITION_PATH_SEPARATOR);
       }
     }
-    return partitionPath.toString();
+    return PartitionPathEncodeUtils.validateNoPathTraversal(partitionPath.toString());
   }
 
   @Override

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/KeyGenUtils.java

@@ -266,7 +266,7 @@ public static String getRecordPartitionPath(GenericRecord record,
         partitionPath.append(DEFAULT_PARTITION_PATH_SEPARATOR);
       }
     }
-    return partitionPath.toString();
+    return PartitionPathEncodeUtils.validateNoPathTraversal(partitionPath.toString());
   }
 
   public static String getRecordKey(GenericRecord record, String recordKeyField, boolean consistentLogicalTimestampEnabled) {
@@ -294,7 +294,7 @@ public static String getPartitionPath(GenericRecord record, String partitionPath
     if (slashSeparatedDatePartitioning) {
       partitionPath = partitionPath.replace('-', '/');
     }
-    return partitionPath;
+    return PartitionPathEncodeUtils.validateNoPathTraversal(partitionPath);
   }
 
   /**

hudi-client/hudi-client-common/src/test/java/org/apache/hudi/keygen/TestKeyGenUtils.java

@@ -290,4 +290,30 @@ void testIsComplexKeyGeneratorWithSingleRecordKeyFieldEmptyRecordKeyFields() {
     tableConfig.setValue(RECORDKEY_FIELDS, "");
     assertFalse(KeyGenUtils.isComplexKeyGeneratorWithSingleRecordKeyField(tableConfig));
   }
+
+  @Test
+  void testGetPartitionPathRejectsPathTraversal() {
+    Schema stringSchema = Schema.create(Schema.Type.STRING);
+    Schema schema = Schema.createRecord("TestRecord", "doc", "test", false,
+        Arrays.asList(
+            new Schema.Field("part1", stringSchema),
+            new Schema.Field("part2", stringSchema)
+        ));
+    GenericRecord avroRecord = new GenericData.Record(schema);
+    avroRecord.put("part1", "../../../../tmp/hudi_poc_escaped_evidence");
+    avroRecord.put("part2", "safe");
+
+    // Single-field (SimpleKeyGenerator) path, with url-encoding disabled (the vulnerable default).
+    assertThrows(HoodieKeyException.class,
+        () -> KeyGenUtils.getPartitionPath(avroRecord, "part1", false, false, false, false));
+    // Multi-field (ComplexKeyGenerator) path.
+    assertThrows(HoodieKeyException.class,
+        () -> KeyGenUtils.getRecordPartitionPath(
+            avroRecord, Arrays.asList("part1", "part2"), false, false, false, false));
+
+    // A legitimate value (dots that are not a full ".." segment) must still be accepted.
+    avroRecord.put("part1", "2024-01-01");
+    assertEquals("2024-01-01",
+        KeyGenUtils.getPartitionPath(avroRecord, "part1", false, false, false, false));
+  }
 }

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/keygen/PartitionPathFormatterBase.java

@@ -17,6 +17,8 @@
 
 package org.apache.hudi.keygen;
 
+import org.apache.hudi.common.util.PartitionPathEncodeUtils;
+
 import org.apache.spark.unsafe.types.UTF8String;
 
 import java.util.List;
@@ -63,9 +65,9 @@ public final S combine(List<String> partitionPathFields, Object... partitionPath
     // and Hive-style of partitioning is not required
     if (!useHiveStylePartitioning && partitionPathParts.length == 1) {
       if (slashSeparatedDatePartitioning) {
-        return ((S) ((String) toString(partitionPathParts[0])).replace('-', '/'));
+        return validateNoPathTraversal((S) ((String) toString(partitionPathParts[0])).replace('-', '/'));
       } else {
-        return tryEncode(handleEmpty(toString(partitionPathParts[0])));
+        return validateNoPathTraversal(tryEncode(handleEmpty(toString(partitionPathParts[0]))));
       }
     }
 
@@ -89,7 +91,12 @@ public final S combine(List<String> partitionPathFields, Object... partitionPath
       }
     }
 
-    return sb.build();
+    return validateNoPathTraversal(sb.build());
+  }
+
+  private S validateNoPathTraversal(S partitionPath) {
+    PartitionPathEncodeUtils.validateNoPathTraversal(partitionPath == null ? null : partitionPath.toString());
+    return partitionPath;
   }
 
   private S tryEncode(S partitionPathPart) {

hudi-common/src/main/java/org/apache/hudi/common/util/PartitionPathEncodeUtils.java

@@ -17,6 +17,8 @@
 
 package org.apache.hudi.common.util;
 
+import org.apache.hudi.exception.HoodieKeyException;
+
 import java.util.BitSet;
 import java.util.function.Function;
 
@@ -141,4 +143,62 @@ public static String escapePartitionValue(String value) {
       return escapePathName(value);
     }
   }
+
+  /**
+   * Returns {@code true} if the given (relative) partition path contains a directory-traversal
+   * segment (a path segment equal to {@code ".."}). Such a partition path, once resolved against
+   * the table base path, can escape the base path and write Hudi-managed files into arbitrary
+   * directories reachable by the writer's credentials.
+   *
+   * <p>The check is intentionally value-content only: it tolerates {@code '.'} inside a segment
+   * (e.g. date partitions like {@code 2024.01.01}) and only rejects the standalone {@code ".."}
+   * segment. Both forward slash {@code '/'} and the platform-independent literal are treated as
+   * separators, since a partition path is always stored using forward slashes.
+   *
+   * @param partitionPath the relative partition path (or a single partition field value).
+   * @return {@code true} if a {@code ".."} traversal segment is present, {@code false} otherwise.
+   */
+  /**
+   * Validates that the given (relative) partition path does not contain a directory-traversal
+   * segment, throwing {@link HoodieKeyException} if it does. This is enforced regardless of the
+   * {@code hoodie.datasource.write.partitionpath.urlencode} setting, since url-encoding is opt-in
+   * (disabled by default) and never rejects {@code ".."}.
+   *
+   * @param partitionPath the relative partition path (or a single partition field value).
+   * @return the same {@code partitionPath} if it is safe.
+   * @throws HoodieKeyException if the partition path contains a {@code ".."} traversal segment.
+   */
+  public static String validateNoPathTraversal(String partitionPath) {
+    if (hasPathTraversal(partitionPath)) {
+      throw new HoodieKeyException("Invalid partition path \"" + partitionPath + "\": partition paths "
+          + "must not contain \"..\" path-traversal segments, which could let a record write Hudi files "
+          + "outside the table base path. This is most often caused by an unsanitized data field being "
+          + "used as the partition path. If \"..\" is a legitimate value in your data, enable "
+          + "hoodie.datasource.write.partitionpath.urlencode so partition values are escaped.");
+    }
+    return partitionPath;
+  }
+
+  public static boolean hasPathTraversal(String partitionPath) {
+    if (partitionPath == null || partitionPath.isEmpty()) {
+      return false;
+    }
+    // Normalize backslashes to forward slashes so a "..\.." style value cannot slip through on
+    // any platform. Partition paths are always persisted using forward slashes.
+    String normalized = partitionPath.replace('\\', '/');
+    int start = 0;
+    int len = normalized.length();
+    while (start <= len) {
+      int end = normalized.indexOf('/', start);
+      if (end < 0) {
+        end = len;
+      }
+      // A segment of exactly ".." is a traversal segment.
+      if (end - start == 2 && normalized.charAt(start) == '.' && normalized.charAt(start + 1) == '.') {
+        return true;
+      }
+      start = end + 1;
+    }
+    return false;
+  }
 }

hudi-common/src/test/java/org/apache/hudi/common/util/TestPartitionPathEncodeUtils.java

@@ -0,0 +1,83 @@
+/*
+ * 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.hudi.common.util;
+
+import org.apache.hudi.exception.HoodieKeyException;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TestPartitionPathEncodeUtils {
+
+  @ParameterizedTest
+  @ValueSource(strings = {
+      "..",
+      "../foo",
+      "foo/..",
+      "foo/../bar",
+      "../../../../tmp/evil",
+      "a/../../b",
+      "..\\foo",
+      "foo\\..\\bar"
+  })
+  void detectsPathTraversal(String partitionPath) {
+    assertTrue(PartitionPathEncodeUtils.hasPathTraversal(partitionPath),
+        "Expected traversal to be detected for: " + partitionPath);
+    assertThrows(HoodieKeyException.class,
+        () -> PartitionPathEncodeUtils.validateNoPathTraversal(partitionPath));
+  }
+
+  @ParameterizedTest
+  @ValueSource(strings = {
+      "2024/01/01",
+      "2024-01-01",
+      "region=us-west-2",
+      "year=2024/month=01",
+      "rider=..",
+      "rider=../evil",
+      "..foo",
+      "foo..",
+      "foo..bar",
+      "a.b.c",
+      "...",
+      "....",
+      "a/.hidden/b",
+      "."
+  })
+  void allowsLegitimatePartitionPaths(String partitionPath) {
+    assertFalse(PartitionPathEncodeUtils.hasPathTraversal(partitionPath),
+        "Expected no traversal for: " + partitionPath);
+    // validate should return the same value unchanged.
+    assertEquals(partitionPath, PartitionPathEncodeUtils.validateNoPathTraversal(partitionPath));
+  }
+
+  @Test
+  void nullAndEmptyAreSafe() {
+    assertFalse(PartitionPathEncodeUtils.hasPathTraversal(null));
+    assertFalse(PartitionPathEncodeUtils.hasPathTraversal(""));
+    assertEquals(null, PartitionPathEncodeUtils.validateNoPathTraversal(null));
+    assertEquals("", PartitionPathEncodeUtils.validateNoPathTraversal(""));
+  }
+}

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/RowDataKeyGen.java

@@ -44,6 +44,7 @@
 
 import static org.apache.hudi.common.util.PartitionPathEncodeUtils.DEFAULT_PARTITION_PATH;
 import static org.apache.hudi.common.util.PartitionPathEncodeUtils.escapePathName;
+import static org.apache.hudi.common.util.PartitionPathEncodeUtils.validateNoPathTraversal;
 import static org.apache.hudi.keygen.KeyGenerator.DEFAULT_COLUMN_VALUE_SEPARATOR;
 
 /**
@@ -172,13 +173,14 @@ public String getRecordKey(RowData rowData) {
 
   public String getPartitionPath(RowData rowData) {
     if (this.simplePartitionPath) {
-      return getPartitionPath(partitionPathFieldGetter.getFieldOrNull(rowData),
-          this.partitionPathFields[0], this.hiveStylePartitioning, this.encodePartitionPath, this.keyGenOpt);
+      return validateNoPathTraversal(getPartitionPath(partitionPathFieldGetter.getFieldOrNull(rowData),
+          this.partitionPathFields[0], this.hiveStylePartitioning, this.encodePartitionPath, this.keyGenOpt));
     } else if (this.nonPartitioned) {
       return EMPTY_PARTITION;
     } else {
       Object[] partValues = this.partitionPathProjection.projectAsValues(rowData);
-      return getRecordPartitionPath(partValues, this.partitionPathFields, this.hiveStylePartitioning, this.encodePartitionPath);
+      return validateNoPathTraversal(
+          getRecordPartitionPath(partValues, this.partitionPathFields, this.hiveStylePartitioning, this.encodePartitionPath));
     }
   }