(flink) branch master updated: [FLINK-40332][table] Prevent path traversal in FileCatalogStore catalog names

[gaborgsomogyi]

This is an automated email from the ASF dual-hosted git repository.

gaborgsomogyi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/master by this push:
     new cf2df298f22 [FLINK-40332][table] Prevent path traversal in 
FileCatalogStore catalog names
cf2df298f22 is described below

commit cf2df298f22f18e9d104e80597ca69b05006bc4c
Author: Gabor Somogyi <gabor_somog...@apple.com>
AuthorDate: Wed Aug 5 17:11:09 2026 +0200

    [FLINK-40332][table] Prevent path traversal in FileCatalogStore catalog 
names
---
 .../flink/table/catalog/FileCatalogStore.java      | 23 +++++++++++++++-
 .../flink/table/catalog/FileCatalogStoreTest.java  | 32 ++++++++++++++++++++++
 2 files changed, 54 insertions(+), 1 deletion(-)

diff --git 
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/FileCatalogStore.java
 
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/FileCatalogStore.java
index 50763a828ea..0e1fb69ee38 100644
--- 
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/FileCatalogStore.java
+++ 
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/FileCatalogStore.java
@@ -260,6 +260,27 @@ public class FileCatalogStore extends AbstractCatalogStore 
{
     }
 
     private Path getCatalogPath(String catalogName) {
-        return new Path(catalogStorePath, catalogName + FILE_EXTENSION);
+        Path catalogPath;
+        try {
+            catalogPath = new Path(catalogStorePath, catalogName + 
FILE_EXTENSION);
+        } catch (Exception e) {
+            // e.g. catalogName embeds its own scheme-qualified URI (like 
"file:///etc/passwd"),
+            // which Path may reject outright while merging it against 
catalogStorePath.
+            throw new CatalogException(String.format("Invalid catalog name 
'%s'.", catalogName), e);
+        }
+
+        // catalogName is caller-supplied and may try to escape 
catalogStorePath, e.g. via ".."
+        // segments. Path's own resolution above already fully normalizes the 
result (RFC 3986
+        // dot-segment removal), so checking that the *resolved* path's parent 
is still
+        // catalogStorePath is sufficient to reject every variant of escape, 
without needing to
+        // inspect catalogName itself.
+        if (!catalogStorePath.equals(catalogPath.getParent())) {
+            throw new CatalogException(
+                    String.format(
+                            "Invalid catalog name '%s'. It resolves to '%s', 
which is outside of "
+                                    + "the catalog store directory '%s'.",
+                            catalogName, catalogPath, catalogStorePath));
+        }
+        return catalogPath;
     }
 }
diff --git 
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/FileCatalogStoreTest.java
 
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/FileCatalogStoreTest.java
index 011ce881bb8..e22f3c7d89c 100644
--- 
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/FileCatalogStoreTest.java
+++ 
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/FileCatalogStoreTest.java
@@ -106,6 +106,38 @@ class FileCatalogStoreTest {
         assertThat(storedCatalogs.contains(DUMMY)).isTrue();
     }
 
+    @Test
+    void testStoreCatalogRejectsPathTraversal() throws Exception {
+        CatalogStore catalogStore = initCatalogStore();
+        catalogStore.open();
+
+        // A malicious catalog name that, if not validated, resolves outside 
of the
+        // catalog store directory: 
tempDir/dummy-catalog-store/../escaped.yaml ->
+        // tempDir/escaped.yaml.
+        String maliciousName = "../escaped";
+        File escapedFile = tempDir.resolve("escaped" + 
FileCatalogStore.FILE_EXTENSION).toFile();
+
+        assertThatThrownBy(() -> catalogStore.storeCatalog(maliciousName, 
DUMMY_CATALOG))
+                .isInstanceOf(CatalogException.class);
+        assertThat(escapedFile).doesNotExist();
+    }
+
+    @Test
+    void testStoreCatalogRejectsAbsoluteSchemeOverride() throws Exception {
+        CatalogStore catalogStore = initCatalogStore();
+        catalogStore.open();
+
+        // A catalog name that embeds its own absolute file:// URI. Per RFC 
3986 §5.3, resolving
+        // an absolute reference against a base URI discards the base 
entirely, so if this isn't
+        // rejected, the catalog store directory is bypassed altogether.
+        File escapedFile = tempDir.resolve("escaped" + 
FileCatalogStore.FILE_EXTENSION).toFile();
+        String maliciousName = "file://" + 
escapedFile.getAbsolutePath().replace(".yaml", "");
+
+        assertThatThrownBy(() -> catalogStore.storeCatalog(maliciousName, 
DUMMY_CATALOG))
+                .isInstanceOf(CatalogException.class);
+        assertThat(escapedFile).doesNotExist();
+    }
+
     @Test
     void testRemoveExisting() {
         CatalogStore catalogStore = initCatalogStore();