This is an automated email from the ASF dual-hosted git repository.
yuqi1129 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new b5eead2397 [MINOR] improvement(core): validate remote URI host in
JobManager before download (#11354)
b5eead2397 is described below
commit b5eead2397dc7371ba71eb577be668000c0d7eca
Author: Qi Yu <[email protected]>
AuthorDate: Fri Jun 12 15:30:25 2026 +0800
[MINOR] improvement(core): validate remote URI host in JobManager before
download (#11354)
### What changes were proposed in this pull request?
Add shared remote URI validation and file-fetching utilities for
server-side HTTP/HTTPS/FTP downloads. The validator rejects loopback,
link-local, RFC 1918 private, IPv6 unique-local, multicast, unspecified,
and known cloud metadata addresses by default.
Consolidate the duplicated `FetchFileUtils` implementations into
`common` and use the shared implementation from `JobManager` and the
Hive, Iceberg, Hadoop, and Paimon Kerberos clients. Administrators can
explicitly disable unsafe-address blocking for trusted local or private
endpoints.
### Why are the changes needed?
The previous implementations passed configured or user-supplied URIs
directly to the download client without checking the resolved address.
This could allow the server to access internal services or cloud
metadata endpoints. The duplicated download implementations also made it
easy for security behavior to diverge across modules.
### Does this PR introduce _any_ user-facing change?
Yes. Remote file downloads now block unsafe addresses by default.
Deployments that intentionally fetch from trusted local or private
endpoints must disable the corresponding unsafe-address blocking
configuration.
### How was this patch tested?
- `TestJobManager` covers blocked and explicitly allowed JobManager
downloads.
- `TestRemoteUriValidator` covers loopback, link-local, RFC 1918
private, IPv6 unique-local, localhost, and cloud metadata addresses.
- `TestFetchFileUtils` covers shared local/remote fetching,
unsafe-address configuration, concurrency, and error handling.
---
.../iceberg/IcebergCatalogPropertiesMetadata.java | 23 +-
.../paimon/PaimonCatalogPropertiesMetadata.java | 18 +-
.../authentication/kerberos/FetchFileUtils.java | 58 ------
.../authentication/kerberos/KerberosClient.java | 8 +-
.../catalog/hadoop/fs/kerberos/FetchFileUtils.java | 65 ------
.../catalog/hadoop/fs/kerberos/KerberosClient.java | 4 +-
.../gravitino/hive/kerberos/FetchFileUtils.java | 102 ---------
.../gravitino/hive/kerberos/KerberosClient.java | 5 +-
.../hive/kerberos/TestFetchFileUtils.java | 119 -----------
common/build.gradle.kts | 1 +
.../org/apache/gravitino/utils/FileFetcher.java | 176 ++++++++++++++++
.../apache/gravitino/utils/RemoteUriValidator.java | 92 ++++++++
.../apache/gravitino/utils/TestFileFetcher.java | 231 +++++++++++++++++++++
.../gravitino/utils/TestRemoteUriValidator.java | 91 ++++++++
conf/gravitino-iceberg-rest-server.conf.template | 2 +
conf/gravitino.conf.template | 3 +
.../main/java/org/apache/gravitino/Configs.java | 12 ++
.../java/org/apache/gravitino/GravitinoEnv.java | 3 +
.../java/org/apache/gravitino/job/JobManager.java | 31 +--
.../org/apache/gravitino/job/TestJobManager.java | 85 ++++++++
docs/apache-hive-catalog.md | 27 ++-
docs/fileset-catalog.md | 16 +-
docs/gravitino-server-config.md | 15 ++
docs/iceberg-rest-service.md | 20 +-
docs/lakehouse-hudi-catalog.md | 16 +-
.../authentication/kerberos/FetchFileUtils.java | 65 ------
.../authentication/kerberos/KerberosClient.java | 4 +-
27 files changed, 791 insertions(+), 501 deletions(-)
diff --git
a/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalogPropertiesMetadata.java
b/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalogPropertiesMetadata.java
index 7b1466a341..bf6246a738 100644
---
a/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalogPropertiesMetadata.java
+++
b/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalogPropertiesMetadata.java
@@ -50,19 +50,16 @@ public class IcebergCatalogPropertiesMetadata extends
BaseCatalogPropertiesMetad
private static final Map<String, PropertyEntry<?>> PROPERTIES_METADATA;
public static final Map<String, String>
KERBEROS_CONFIGURATION_FOR_HIVE_BACKEND =
- ImmutableMap.of(
- KerberosConfig.PRINCIPAL_KEY,
- KerberosConfig.PRINCIPAL_KEY,
- KerberosConfig.KET_TAB_URI_KEY,
- KerberosConfig.KET_TAB_URI_KEY,
- KerberosConfig.CHECK_INTERVAL_SEC_KEY,
- KerberosConfig.CHECK_INTERVAL_SEC_KEY,
- KerberosConfig.FETCH_TIMEOUT_SEC_KEY,
- KerberosConfig.FETCH_TIMEOUT_SEC_KEY,
- AuthenticationConfig.IMPERSONATION_ENABLE_KEY,
- AuthenticationConfig.IMPERSONATION_ENABLE_KEY,
- AuthenticationConfig.AUTH_TYPE_KEY,
- AuthenticationConfig.AUTH_TYPE_KEY);
+ ImmutableMap.<String, String>builder()
+ .put(KerberosConfig.PRINCIPAL_KEY, KerberosConfig.PRINCIPAL_KEY)
+ .put(KerberosConfig.KET_TAB_URI_KEY, KerberosConfig.KET_TAB_URI_KEY)
+ .put(KerberosConfig.CHECK_INTERVAL_SEC_KEY,
KerberosConfig.CHECK_INTERVAL_SEC_KEY)
+ .put(KerberosConfig.FETCH_TIMEOUT_SEC_KEY,
KerberosConfig.FETCH_TIMEOUT_SEC_KEY)
+ .put(
+ AuthenticationConfig.IMPERSONATION_ENABLE_KEY,
+ AuthenticationConfig.IMPERSONATION_ENABLE_KEY)
+ .put(AuthenticationConfig.AUTH_TYPE_KEY,
AuthenticationConfig.AUTH_TYPE_KEY)
+ .build();
static {
List<PropertyEntry<?>> propertyEntries =
diff --git
a/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonCatalogPropertiesMetadata.java
b/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonCatalogPropertiesMetadata.java
index b82061ac36..7f64c711f7 100644
---
a/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonCatalogPropertiesMetadata.java
+++
b/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonCatalogPropertiesMetadata.java
@@ -85,17 +85,13 @@ public class PaimonCatalogPropertiesMetadata extends
BaseCatalogPropertiesMetada
.build();
private static final Map<String, PropertyEntry<?>> PROPERTIES_METADATA;
public static final Map<String, String> KERBEROS_CONFIGURATION =
- ImmutableMap.of(
- KerberosConfig.PRINCIPAL_KEY,
- KerberosConfig.PRINCIPAL_KEY,
- KerberosConfig.KEY_TAB_URI_KEY,
- KerberosConfig.KEY_TAB_URI_KEY,
- KerberosConfig.CHECK_INTERVAL_SEC_KEY,
- KerberosConfig.CHECK_INTERVAL_SEC_KEY,
- KerberosConfig.FETCH_TIMEOUT_SEC_KEY,
- KerberosConfig.FETCH_TIMEOUT_SEC_KEY,
- AuthenticationConfig.AUTH_TYPE_KEY,
- AuthenticationConfig.AUTH_TYPE_KEY);
+ ImmutableMap.<String, String>builder()
+ .put(KerberosConfig.PRINCIPAL_KEY, KerberosConfig.PRINCIPAL_KEY)
+ .put(KerberosConfig.KEY_TAB_URI_KEY, KerberosConfig.KEY_TAB_URI_KEY)
+ .put(KerberosConfig.CHECK_INTERVAL_SEC_KEY,
KerberosConfig.CHECK_INTERVAL_SEC_KEY)
+ .put(KerberosConfig.FETCH_TIMEOUT_SEC_KEY,
KerberosConfig.FETCH_TIMEOUT_SEC_KEY)
+ .put(AuthenticationConfig.AUTH_TYPE_KEY,
AuthenticationConfig.AUTH_TYPE_KEY)
+ .build();
public static final Map<String, String> S3_CONFIGURATION =
ImmutableMap.of(
diff --git
a/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/authentication/kerberos/FetchFileUtils.java
b/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/authentication/kerberos/FetchFileUtils.java
deleted file mode 100644
index 29cfdc37c5..0000000000
---
a/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/authentication/kerberos/FetchFileUtils.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
- * 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.gravitino.catalog.lakehouse.paimon.authentication.kerberos;
-
-import java.io.File;
-import java.io.IOException;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.nio.file.Files;
-import java.util.Optional;
-import org.apache.commons.io.FileUtils;
-
-public class FetchFileUtils {
-
- private FetchFileUtils() {}
-
- public static void fetchFileFromUri(String fileUri, File destFile, int
timeout)
- throws IOException {
- try {
- URI uri = new URI(fileUri);
- String scheme = Optional.ofNullable(uri.getScheme()).orElse("file");
-
- switch (scheme) {
- case "http":
- case "https":
- case "ftp":
- FileUtils.copyURLToFile(uri.toURL(), destFile, timeout * 1000,
timeout * 1000);
- break;
-
- case "file":
- Files.createSymbolicLink(destFile.toPath(), new
File(uri.getPath()).toPath());
- break;
-
- default:
- throw new IllegalArgumentException(
- String.format("Doesn't support the scheme %s", scheme));
- }
- } catch (URISyntaxException ue) {
- throw new IllegalArgumentException("The uri of file has the wrong
format", ue);
- }
- }
-}
diff --git
a/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/authentication/kerberos/KerberosClient.java
b/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/authentication/kerberos/KerberosClient.java
index d8ae6c6438..6e4737d9bb 100644
---
a/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/authentication/kerberos/KerberosClient.java
+++
b/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/authentication/kerberos/KerberosClient.java
@@ -26,6 +26,7 @@ import java.util.Map;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
+import org.apache.gravitino.utils.FileFetcher;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.security.UserGroupInformation;
import org.slf4j.Logger;
@@ -100,7 +101,12 @@ public class KerberosClient implements Closeable {
}
int fetchKeytabFileTimeout = kerberosConfig.getFetchTimeoutSec();
- FetchFileUtils.fetchFileFromUri(keyTabUri, keytabFile,
fetchKeytabFileTimeout);
+ FileFetcher.get()
+ .fetchFileFromUri(
+ keyTabUri,
+ keytabFile,
+ fetchKeytabFileTimeout * 1000,
+ null /* hadoopConf: Paimon keytab URIs never use the hdfs scheme
*/);
return keytabFile;
}
diff --git
a/catalogs/hadoop-common/src/main/java/org/apache/gravitino/catalog/hadoop/fs/kerberos/FetchFileUtils.java
b/catalogs/hadoop-common/src/main/java/org/apache/gravitino/catalog/hadoop/fs/kerberos/FetchFileUtils.java
deleted file mode 100644
index d014345c54..0000000000
---
a/catalogs/hadoop-common/src/main/java/org/apache/gravitino/catalog/hadoop/fs/kerberos/FetchFileUtils.java
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
- * 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.gravitino.catalog.hadoop.fs.kerberos;
-
-import java.io.File;
-import java.io.IOException;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.nio.file.Files;
-import java.util.Optional;
-import org.apache.commons.io.FileUtils;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-
-public class FetchFileUtils {
-
- private FetchFileUtils() {}
-
- public static void fetchFileFromUri(
- String fileUri, File destFile, int timeout, Configuration conf) throws
IOException {
- try {
- URI uri = new URI(fileUri);
- String scheme = Optional.ofNullable(uri.getScheme()).orElse("file");
-
- switch (scheme) {
- case "http":
- case "https":
- case "ftp":
- FileUtils.copyURLToFile(uri.toURL(), destFile, timeout * 1000,
timeout * 1000);
- break;
-
- case "file":
- Files.createSymbolicLink(destFile.toPath(), new
File(uri.getPath()).toPath());
- break;
-
- case "hdfs":
- FileSystem.get(conf).copyToLocalFile(new Path(uri), new
Path(destFile.toURI()));
- break;
-
- default:
- throw new IllegalArgumentException(
- String.format("Doesn't support the scheme %s", scheme));
- }
- } catch (URISyntaxException ue) {
- throw new IllegalArgumentException("The uri of file has the wrong
format", ue);
- }
- }
-}
diff --git
a/catalogs/hadoop-common/src/main/java/org/apache/gravitino/catalog/hadoop/fs/kerberos/KerberosClient.java
b/catalogs/hadoop-common/src/main/java/org/apache/gravitino/catalog/hadoop/fs/kerberos/KerberosClient.java
index eb93f2ab72..b3d3799f73 100644
---
a/catalogs/hadoop-common/src/main/java/org/apache/gravitino/catalog/hadoop/fs/kerberos/KerberosClient.java
+++
b/catalogs/hadoop-common/src/main/java/org/apache/gravitino/catalog/hadoop/fs/kerberos/KerberosClient.java
@@ -34,6 +34,7 @@ import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.StringUtils;
+import org.apache.gravitino.utils.FileFetcher;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.security.UserGroupInformation;
import org.slf4j.Logger;
@@ -122,7 +123,8 @@ public class KerberosClient implements Closeable {
}
int fetchKeytabFileTimeout = kerberosConfig.getFetchTimeoutSec();
- FetchFileUtils.fetchFileFromUri(keyTabUri, keytabFile,
fetchKeytabFileTimeout, hadoopConf);
+ FileFetcher.get()
+ .fetchFileFromUri(keyTabUri, keytabFile, fetchKeytabFileTimeout *
1000, hadoopConf);
return keytabFile;
}
diff --git
a/catalogs/hive-metastore-common/src/main/java/org/apache/gravitino/hive/kerberos/FetchFileUtils.java
b/catalogs/hive-metastore-common/src/main/java/org/apache/gravitino/hive/kerberos/FetchFileUtils.java
deleted file mode 100644
index 7149e1c82d..0000000000
---
a/catalogs/hive-metastore-common/src/main/java/org/apache/gravitino/hive/kerberos/FetchFileUtils.java
+++ /dev/null
@@ -1,102 +0,0 @@
-/*
- * 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.gravitino.hive.kerberos;
-
-import java.io.File;
-import java.io.IOException;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.nio.file.Files;
-import java.nio.file.StandardCopyOption;
-import java.util.Optional;
-import java.util.concurrent.ConcurrentHashMap;
-import org.apache.commons.io.FileUtils;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-
-public class FetchFileUtils {
-
- /**
- * Per-destination lock map used to serialize concurrent symlink creation
for the same keytab
- * file. Keyed by the normalized absolute destination path string to avoid
races caused by
- * different path spellings referring to the same file. Entries are removed
when the corresponding
- * {@link KerberosClient} is closed, so the map size is bounded by the
number of live catalogs.
- */
- private static final ConcurrentHashMap<String, Object> SYMLINK_LOCKS = new
ConcurrentHashMap<>();
-
- private FetchFileUtils() {}
-
- /**
- * Removes the per-destination lock entry for the given file. Should be
called when the keytab
- * file is deleted (e.g., on {@link KerberosClient#close()}) to prevent
unbounded map growth.
- *
- * @param destFile the keytab destination file whose lock entry should be
removed
- */
- static void removeLock(File destFile) {
-
SYMLINK_LOCKS.remove(destFile.toPath().toAbsolutePath().normalize().toString());
- }
-
- public static void fetchFileFromUri(
- String fileUri, File destFile, int timeout, Configuration conf) throws
IOException {
- try {
- URI uri = new URI(fileUri);
- String scheme = Optional.ofNullable(uri.getScheme()).orElse("file");
-
- switch (scheme) {
- case "http":
- case "https":
- case "ftp":
- FileUtils.copyURLToFile(uri.toURL(), destFile, timeout * 1000,
timeout * 1000);
- break;
-
- case "file":
- var srcPath = new File(uri.getPath()).toPath().normalize();
- var destPath = destFile.toPath().toAbsolutePath().normalize();
- Object lock = SYMLINK_LOCKS.computeIfAbsent(destPath.toString(), k
-> new Object());
- synchronized (lock) {
- // Skip if the symlink already points to the correct target.
- if (Files.isSymbolicLink(destPath)
- &&
Files.readSymbolicLink(destPath).normalize().equals(srcPath)) {
- break;
- }
- // Replace via a temporary symlink + rename to minimize the window
where the
- // keytab path is absent (which could cause loginUserFromKeytab to
fail).
- // REPLACE_EXISTING is used here; on common local filesystems
(ext4, xfs, APFS)
- // a same-directory rename is effectively atomic at the OS level.
- var tmpPath = destPath.resolveSibling(destPath.getFileName() +
".symlink.tmp");
- Files.deleteIfExists(tmpPath);
- Files.createSymbolicLink(tmpPath, srcPath);
- Files.move(tmpPath, destPath, StandardCopyOption.REPLACE_EXISTING);
- }
- break;
-
- case "hdfs":
- FileSystem.get(conf).copyToLocalFile(new Path(uri), new
Path(destFile.toURI()));
- break;
-
- default:
- throw new IllegalArgumentException(
- String.format("The scheme '%s' is not supported", scheme));
- }
- } catch (URISyntaxException ue) {
- throw new IllegalArgumentException("The uri of file has the wrong
format", ue);
- }
- }
-}
diff --git
a/catalogs/hive-metastore-common/src/main/java/org/apache/gravitino/hive/kerberos/KerberosClient.java
b/catalogs/hive-metastore-common/src/main/java/org/apache/gravitino/hive/kerberos/KerberosClient.java
index a86535184a..57ed6949e1 100644
---
a/catalogs/hive-metastore-common/src/main/java/org/apache/gravitino/hive/kerberos/KerberosClient.java
+++
b/catalogs/hive-metastore-common/src/main/java/org/apache/gravitino/hive/kerberos/KerberosClient.java
@@ -36,6 +36,7 @@ import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.hive.client.HiveClient;
+import org.apache.gravitino.utils.FileFetcher;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.thrift.DelegationTokenIdentifier;
import org.apache.hadoop.io.Text;
@@ -173,7 +174,8 @@ public class KerberosClient implements java.io.Closeable {
}
File keytabFile = new File(path);
int fetchKeytabFileTimeout = kerberosConfig.getFetchTimeoutSec();
- FetchFileUtils.fetchFileFromUri(keyTabUri, keytabFile,
fetchKeytabFileTimeout, hadoopConf);
+ FileFetcher.get()
+ .fetchFileFromUri(keyTabUri, keytabFile, fetchKeytabFileTimeout *
1000, hadoopConf);
return keytabFile;
}
@@ -189,7 +191,6 @@ public class KerberosClient implements java.io.Closeable {
}
Files.deleteIfExists(Paths.get(keytabFilePath));
- FetchFileUtils.removeLock(new File(keytabFilePath));
} catch (IOException e) {
LOG.warn("Failed to delete keytab file: {}", keytabFilePath, e);
}
diff --git
a/catalogs/hive-metastore-common/src/test/java/org/apache/gravitino/hive/kerberos/TestFetchFileUtils.java
b/catalogs/hive-metastore-common/src/test/java/org/apache/gravitino/hive/kerberos/TestFetchFileUtils.java
deleted file mode 100644
index c9a4bbec28..0000000000
---
a/catalogs/hive-metastore-common/src/test/java/org/apache/gravitino/hive/kerberos/TestFetchFileUtils.java
+++ /dev/null
@@ -1,119 +0,0 @@
-/*
- * 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.gravitino.hive.kerberos;
-
-import java.io.File;
-import java.nio.file.Files;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.CyclicBarrier;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
-import org.apache.hadoop.conf.Configuration;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.io.TempDir;
-
-public class TestFetchFileUtils {
-
- @TempDir File tempDir;
-
- @Test
- public void testLinkLocalFile() throws Exception {
- File srcFile = new File(tempDir, "source");
- Assertions.assertTrue(srcFile.createNewFile());
- File destFile = new File(tempDir, "dest");
-
- FetchFileUtils.fetchFileFromUri(srcFile.toURI().toString(), destFile, 10,
new Configuration());
- Assertions.assertTrue(Files.isSymbolicLink(destFile.toPath()));
- Assertions.assertEquals(srcFile.toPath(),
Files.readSymbolicLink(destFile.toPath()));
- }
-
- @Test
- public void testConcurrentSymlinkCreation() throws Exception {
- File srcFile = new File(tempDir, "source_concurrent");
- Assertions.assertTrue(srcFile.createNewFile());
- File destFile = new File(tempDir, "dest_concurrent");
-
- int threadCount = 10;
- CyclicBarrier barrier = new CyclicBarrier(threadCount);
- ExecutorService executor = Executors.newFixedThreadPool(threadCount);
- List<Future<?>> futures = new ArrayList<>();
- try {
- for (int i = 0; i < threadCount; i++) {
- futures.add(
- executor.submit(
- () -> {
- try {
- barrier.await(30, TimeUnit.SECONDS);
- FetchFileUtils.fetchFileFromUri(
- srcFile.toURI().toString(), destFile, 10, new
Configuration());
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }));
- }
- for (Future<?> future : futures) {
- future.get(30, TimeUnit.SECONDS);
- }
- } finally {
- executor.shutdownNow();
- }
-
- Assertions.assertTrue(Files.isSymbolicLink(destFile.toPath()));
- Assertions.assertEquals(srcFile.toPath(),
Files.readSymbolicLink(destFile.toPath()));
- }
-
- @Test
- public void testIdempotentSymlinkCreation() throws Exception {
- File srcFile = new File(tempDir, "source_idempotent");
- Assertions.assertTrue(srcFile.createNewFile());
- File destFile = new File(tempDir, "dest_idempotent");
-
- Configuration conf = new Configuration();
- String uri = srcFile.toURI().toString();
-
- FetchFileUtils.fetchFileFromUri(uri, destFile, 10, conf);
- Assertions.assertTrue(Files.isSymbolicLink(destFile.toPath()));
-
- // Second call to the same dest should succeed without error
- FetchFileUtils.fetchFileFromUri(uri, destFile, 10, conf);
- Assertions.assertTrue(Files.isSymbolicLink(destFile.toPath()));
- Assertions.assertEquals(srcFile.toPath(),
Files.readSymbolicLink(destFile.toPath()));
- }
-
- @Test
- public void testSymlinkReplacedWithDifferentTarget() throws Exception {
- File srcFileA = new File(tempDir, "source_a");
- File srcFileB = new File(tempDir, "source_b");
- Assertions.assertTrue(srcFileA.createNewFile());
- Assertions.assertTrue(srcFileB.createNewFile());
- File destFile = new File(tempDir, "dest_replace");
-
- Configuration conf = new Configuration();
-
- FetchFileUtils.fetchFileFromUri(srcFileA.toURI().toString(), destFile, 10,
conf);
- Assertions.assertEquals(srcFileA.toPath(),
Files.readSymbolicLink(destFile.toPath()));
-
- FetchFileUtils.fetchFileFromUri(srcFileB.toURI().toString(), destFile, 10,
conf);
- Assertions.assertEquals(srcFileB.toPath(),
Files.readSymbolicLink(destFile.toPath()));
- }
-}
diff --git a/common/build.gradle.kts b/common/build.gradle.kts
index b30f01f533..e267e9a118 100644
--- a/common/build.gradle.kts
+++ b/common/build.gradle.kts
@@ -30,6 +30,7 @@ dependencies {
implementation(project(":api"))
implementation(libs.commons.collections4)
+ implementation(libs.commons.io)
implementation(libs.commons.lang3)
implementation(libs.guava)
implementation(libs.jackson.annotations)
diff --git a/common/src/main/java/org/apache/gravitino/utils/FileFetcher.java
b/common/src/main/java/org/apache/gravitino/utils/FileFetcher.java
new file mode 100644
index 0000000000..7291a9232a
--- /dev/null
+++ b/common/src/main/java/org/apache/gravitino/utils/FileFetcher.java
@@ -0,0 +1,176 @@
+/*
+ * 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.gravitino.utils;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
+import java.util.Optional;
+import javax.annotation.Nullable;
+import org.apache.commons.io.FileUtils;
+
+/**
+ * Singleton that fetches a file referenced by a URI to a local destination.
Supports {@code file},
+ * {@code http}, {@code https}, {@code ftp} and {@code hdfs} schemes. This is
the single shared
+ * implementation used by the job manager and the Kerberos clients of the
Hive, Iceberg, Hadoop and
+ * Paimon catalogs.
+ *
+ * <p>The {@code hdfs} scheme is resolved reflectively against {@code
+ * org.apache.hadoop.fs.FileSystem} so that this class can live in the {@code
common} module, which
+ * does not declare a compile-time dependency on Hadoop. Callers that never
use {@code hdfs} URIs
+ * (for example the job manager and the Paimon catalog) simply pass {@code
null} for the Hadoop
+ * configuration.
+ */
+public final class FileFetcher {
+
+ /** The server configuration that controls unsafe remote URI blocking. */
+ public static final String BLOCK_UNSAFE_REMOTE_URI_CONFIG =
+ "gravitino.fetchFile.blockUnsafeRemoteUri";
+
+ private volatile boolean blockUnsafeRemoteUri = true;
+
+ private FileFetcher() {}
+
+ private static class InstanceHolder {
+ private static final FileFetcher INSTANCE = new FileFetcher();
+ }
+
+ /**
+ * Returns the singleton file fetcher.
+ *
+ * @return the singleton file fetcher
+ */
+ public static FileFetcher get() {
+ return InstanceHolder.INSTANCE;
+ }
+
+ /**
+ * Initializes the file fetcher.
+ *
+ * @param blockUnsafeRemoteUri whether to block remote URIs that resolve to
unsafe addresses
+ */
+ public void initialize(boolean blockUnsafeRemoteUri) {
+ this.blockUnsafeRemoteUri = blockUnsafeRemoteUri;
+ }
+
+ /**
+ * Fetches the file referenced by {@code fileUri} into {@code destFile}.
+ *
+ * @param fileUri the source URI; a missing scheme is treated as {@code file}
+ * @param destFile the local destination file
+ * @param timeoutMs the connect/read timeout in milliseconds, applied to
remote (http/https/ftp)
+ * downloads
+ * @param hadoopConf an {@code org.apache.hadoop.conf.Configuration}
instance, required only for
+ * the {@code hdfs} scheme; may be {@code null} when no {@code hdfs} URI
is fetched
+ * @return the absolute path of {@code destFile}
+ * @throws IOException if the file cannot be fetched
+ */
+ public String fetchFileFromUri(
+ String fileUri, File destFile, int timeoutMs, @Nullable Object
hadoopConf)
+ throws IOException {
+ try {
+ URI uri = new URI(fileUri);
+ String scheme = Optional.ofNullable(uri.getScheme()).orElse("file");
+
+ switch (scheme) {
+ case "http":
+ case "https":
+ case "ftp":
+ RemoteUriValidator.validate(
+ uri,
+ blockUnsafeRemoteUri,
+ String.format("'%s' to false", BLOCK_UNSAFE_REMOTE_URI_CONFIG));
+ FileUtils.copyURLToFile(uri.toURL(), destFile, timeoutMs, timeoutMs);
+ break;
+
+ case "file":
+ linkLocalFile(uri, destFile);
+ break;
+
+ case "hdfs":
+ copyHdfsFileToLocal(uri, destFile, Optional.ofNullable(hadoopConf));
+ break;
+
+ default:
+ throw new IllegalArgumentException(
+ String.format("The scheme '%s' is not supported", scheme));
+ }
+
+ return destFile.getAbsolutePath();
+ } catch (URISyntaxException ue) {
+ throw new IllegalArgumentException("The uri of file has the wrong
format", ue);
+ }
+ }
+
+ private synchronized void linkLocalFile(URI uri, File destFile) throws
IOException {
+ Path srcPath = new File(uri.getPath()).toPath().normalize();
+ if (!Files.exists(srcPath)) {
+ throw new IOException(
+ String.format("Source file does not exist: %s",
srcPath.toAbsolutePath()));
+ }
+
+ Path destPath = destFile.toPath().toAbsolutePath().normalize();
+ // Skip if the symlink already points to the correct target.
+ if (Files.isSymbolicLink(destPath)
+ && Files.readSymbolicLink(destPath).normalize().equals(srcPath)) {
+ return;
+ }
+ // Replace via a temporary symlink + rename to minimize the window where
the destination path
+ // is absent (which could otherwise cause a concurrent reader, e.g.
loginUserFromKeytab, to
+ // fail). REPLACE_EXISTING is used here; on common local filesystems
(ext4, xfs, APFS) a
+ // same-directory rename is effectively atomic at the OS level.
+ Path tmpPath = destPath.resolveSibling(destPath.getFileName() +
".symlink.tmp");
+ Files.deleteIfExists(tmpPath);
+ Files.createSymbolicLink(tmpPath, srcPath);
+ Files.move(tmpPath, destPath, StandardCopyOption.REPLACE_EXISTING);
+ }
+
+ /**
+ * Copies an {@code hdfs} file to the local destination reflectively, so
that this class does not
+ * require a compile-time dependency on Hadoop.
+ */
+ private void copyHdfsFileToLocal(URI uri, File destFile, Optional<Object>
hadoopConfiguration)
+ throws IOException {
+ Object configuration =
+ hadoopConfiguration.orElseThrow(
+ () ->
+ new IllegalArgumentException(
+ String.format(
+ "A Hadoop configuration is required to fetch an 'hdfs'
uri: %s", uri)));
+ try {
+ Class<?> configurationClass =
Class.forName("org.apache.hadoop.conf.Configuration");
+ Class<?> fileSystemClass =
Class.forName("org.apache.hadoop.fs.FileSystem");
+ Class<?> pathClass = Class.forName("org.apache.hadoop.fs.Path");
+
+ Object fileSystem =
+ fileSystemClass.getMethod("get", configurationClass).invoke(null,
configuration);
+ Object srcPath = pathClass.getConstructor(URI.class).newInstance(uri);
+ Object destPath =
pathClass.getConstructor(URI.class).newInstance(destFile.toURI());
+ fileSystemClass
+ .getMethod("copyToLocalFile", pathClass, pathClass)
+ .invoke(fileSystem, srcPath, destPath);
+ } catch (ReflectiveOperationException e) {
+ throw new IOException(String.format("Failed to fetch file from hdfs uri:
%s", uri), e);
+ }
+ }
+}
diff --git
a/common/src/main/java/org/apache/gravitino/utils/RemoteUriValidator.java
b/common/src/main/java/org/apache/gravitino/utils/RemoteUriValidator.java
new file mode 100644
index 0000000000..41a60fa870
--- /dev/null
+++ b/common/src/main/java/org/apache/gravitino/utils/RemoteUriValidator.java
@@ -0,0 +1,92 @@
+/*
+ * 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.gravitino.utils;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.URI;
+
+/** Validates remote URI hosts before server-side downloads. */
+public final class RemoteUriValidator {
+
+ private RemoteUriValidator() {}
+
+ /**
+ * Resolves the host in the given URI and rejects unsafe addresses when
blocking is enabled.
+ *
+ * @param uri The remote URI to validate.
+ * @param blockUnsafeAddress Whether unsafe addresses should be blocked.
+ * @param blockUnsafeAddressHint The configuration hint that disables unsafe
address blocking.
+ * @throws IOException If host resolution fails.
+ * @throws IllegalArgumentException If the URI has no host or resolves to an
unsafe address.
+ */
+ public static void validate(URI uri, boolean blockUnsafeAddress, String
blockUnsafeAddressHint)
+ throws IOException {
+ String host = uri.getHost();
+ if (host == null) {
+ throw new IllegalArgumentException("URI has no host: " + uri);
+ }
+
+ if (!blockUnsafeAddress) {
+ return;
+ }
+
+ InetAddress[] addresses = InetAddress.getAllByName(host);
+ for (InetAddress address : addresses) {
+ if (isUnsafeAddress(address)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "URI '%s' resolves to blocked address %s from the Gravitino
server side. "
+ + "Access to local, private, link-local, multicast,
unspecified, and cloud "
+ + "metadata addresses is disabled by default to prevent
SSRF. If this URI is "
+ + "trusted and this access is required, set %s.",
+ uri, address.getHostAddress(), blockUnsafeAddressHint));
+ }
+ }
+ }
+
+ private static boolean isUnsafeAddress(InetAddress address) {
+ if (address.isLoopbackAddress()
+ || address.isLinkLocalAddress()
+ || address.isSiteLocalAddress()
+ || address.isMulticastAddress()
+ || address.isAnyLocalAddress()) {
+ return true;
+ }
+
+ byte[] bytes = address.getAddress();
+ if (isCloudMetadataAddress(bytes)) {
+ return true;
+ }
+
+ return isIpv6UniqueLocalAddress(bytes);
+ }
+
+ private static boolean isCloudMetadataAddress(byte[] bytes) {
+ return bytes.length == 4
+ && (bytes[0] & 0xFF) == 100
+ && (bytes[1] & 0xFF) == 100
+ && (bytes[2] & 0xFF) == 100
+ && (bytes[3] & 0xFF) == 200;
+ }
+
+ private static boolean isIpv6UniqueLocalAddress(byte[] bytes) {
+ return bytes.length == 16 && ((bytes[0] & 0xFE) == 0xFC);
+ }
+}
diff --git
a/common/src/test/java/org/apache/gravitino/utils/TestFileFetcher.java
b/common/src/test/java/org/apache/gravitino/utils/TestFileFetcher.java
new file mode 100644
index 0000000000..0397e044ee
--- /dev/null
+++ b/common/src/test/java/org/apache/gravitino/utils/TestFileFetcher.java
@@ -0,0 +1,231 @@
+/*
+ * 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.gravitino.utils;
+
+import com.sun.net.httpserver.HttpServer;
+import java.io.File;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+public class TestFileFetcher {
+
+ // The hdfs happy path is exercised reflectively against Hadoop and is
covered by the catalog
+ // Kerberos integration tests; the common module has no Hadoop on its test
classpath, so here we
+ // only assert that an hdfs uri without a Hadoop configuration is rejected.
+ @TempDir File tempDir;
+
+ @Test
+ public void testBlockUnsafeRemoteUriConfigName() {
+ Assertions.assertEquals(
+ "gravitino.fetchFile.blockUnsafeRemoteUri",
FileFetcher.BLOCK_UNSAFE_REMOTE_URI_CONFIG);
+ }
+
+ @Test
+ public void testGetShouldReturnSingleton() {
+ Assertions.assertSame(FileFetcher.get(), FileFetcher.get());
+ }
+
+ @Test
+ public void testLinkLocalFile() throws Exception {
+ File srcFile = new File(tempDir, "source");
+ Assertions.assertTrue(srcFile.createNewFile());
+ File destFile = new File(tempDir, "dest");
+
+ FileFetcher.get().fetchFileFromUri(srcFile.toURI().toString(), destFile,
10, null);
+ Assertions.assertTrue(Files.isSymbolicLink(destFile.toPath()));
+ Assertions.assertEquals(
+ srcFile.toPath().normalize(),
Files.readSymbolicLink(destFile.toPath()).normalize());
+ }
+
+ @Test
+ public void testMissingLocalFileShouldFail() {
+ File destFile = new File(tempDir, "dest_missing");
+ String uri = new File(tempDir, "does-not-exist-" +
UUID.randomUUID()).toURI().toString();
+
+ Assertions.assertThrows(
+ IOException.class, () -> FileFetcher.get().fetchFileFromUri(uri,
destFile, 10, null));
+ }
+
+ @Test
+ public void testConcurrentSymlinkCreation() throws Exception {
+ File srcFile = new File(tempDir, "source_concurrent");
+ Assertions.assertTrue(srcFile.createNewFile());
+ File destFile = new File(tempDir, "dest_concurrent");
+
+ int threadCount = 10;
+ CyclicBarrier barrier = new CyclicBarrier(threadCount);
+ ExecutorService executor = Executors.newFixedThreadPool(threadCount);
+ List<Future<?>> futures = new ArrayList<>();
+ try {
+ for (int i = 0; i < threadCount; i++) {
+ futures.add(
+ executor.submit(
+ () -> {
+ try {
+ barrier.await(30, TimeUnit.SECONDS);
+ FileFetcher.get()
+ .fetchFileFromUri(srcFile.toURI().toString(),
destFile, 10, null);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }));
+ }
+ for (Future<?> future : futures) {
+ future.get(30, TimeUnit.SECONDS);
+ }
+ } finally {
+ executor.shutdownNow();
+ }
+
+ Assertions.assertTrue(Files.isSymbolicLink(destFile.toPath()));
+ Assertions.assertEquals(
+ srcFile.toPath().normalize(),
Files.readSymbolicLink(destFile.toPath()).normalize());
+ }
+
+ @Test
+ public void testIdempotentSymlinkCreation() throws Exception {
+ File srcFile = new File(tempDir, "source_idempotent");
+ Assertions.assertTrue(srcFile.createNewFile());
+ File destFile = new File(tempDir, "dest_idempotent");
+ String uri = srcFile.toURI().toString();
+
+ FileFetcher.get().fetchFileFromUri(uri, destFile, 10, null);
+ Assertions.assertTrue(Files.isSymbolicLink(destFile.toPath()));
+
+ // Second call to the same dest should succeed without error.
+ FileFetcher.get().fetchFileFromUri(uri, destFile, 10, null);
+ Assertions.assertTrue(Files.isSymbolicLink(destFile.toPath()));
+ Assertions.assertEquals(
+ srcFile.toPath().normalize(),
Files.readSymbolicLink(destFile.toPath()).normalize());
+ }
+
+ @Test
+ public void testSymlinkReplacedWithDifferentTarget() throws Exception {
+ File srcFileA = new File(tempDir, "source_a");
+ File srcFileB = new File(tempDir, "source_b");
+ Assertions.assertTrue(srcFileA.createNewFile());
+ Assertions.assertTrue(srcFileB.createNewFile());
+ File destFile = new File(tempDir, "dest_replace");
+
+ FileFetcher.get().fetchFileFromUri(srcFileA.toURI().toString(), destFile,
10, null);
+ Assertions.assertEquals(
+ srcFileA.toPath().normalize(),
Files.readSymbolicLink(destFile.toPath()).normalize());
+
+ FileFetcher.get().fetchFileFromUri(srcFileB.toURI().toString(), destFile,
10, null);
+ Assertions.assertEquals(
+ srcFileB.toPath().normalize(),
Files.readSymbolicLink(destFile.toPath()).normalize());
+ }
+
+ @Test
+ public void testRemoteFetchShouldBlockLocalhostByDefault() {
+ File destFile = new File(tempDir, "blocked");
+ FileFetcher.get().initialize(true);
+
+ IllegalArgumentException exception =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ FileFetcher.get()
+ .fetchFileFromUri("http://127.0.0.1:8090/keytab";,
destFile, 10, null));
+
+ Assertions.assertTrue(exception.getMessage().contains("Gravitino server
side"));
+ Assertions.assertTrue(
+
exception.getMessage().contains(FileFetcher.BLOCK_UNSAFE_REMOTE_URI_CONFIG));
+ }
+
+ @Test
+ public void testRemoteFetchShouldAllowLocalhostWhenBlockingDisabled() throws
Exception {
+ File destFile = new File(tempDir, "allowed");
+ HttpServer server = createLoopbackHttpServer("keytab");
+
+ try {
+ server.start();
+ int port = server.getAddress().getPort();
+ FileFetcher.get().initialize(false);
+ FileFetcher.get()
+ .fetchFileFromUri(
+ String.format("http://127.0.0.1:%d/keytab";, port),
+ destFile,
+ 30000 /* 30s connect/read timeout in milliseconds */,
+ null);
+
+ Assertions.assertEquals("keytab", Files.readString(destFile.toPath()));
+ } finally {
+ FileFetcher.get().initialize(true);
+ server.stop(0);
+ }
+ }
+
+ @Test
+ public void testHdfsSchemeWithoutConfShouldFail() {
+ File destFile = new File(tempDir, "dest_hdfs");
+
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> FileFetcher.get().fetchFileFromUri("hdfs://namenode/keytab",
destFile, 10, null));
+ }
+
+ @Test
+ public void testUnsupportedSchemeShouldFail() {
+ File destFile = new File(tempDir, "dest_scp");
+
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> FileFetcher.get().fetchFileFromUri("scp://host/keytab",
destFile, 10, null));
+ }
+
+ @Test
+ public void testReturnsDestinationAbsolutePath() throws Exception {
+ File srcFile = new File(tempDir, "source_return");
+ Assertions.assertTrue(srcFile.createNewFile());
+ File destFile = new File(tempDir, "dest_return");
+
+ String returned =
+ FileFetcher.get().fetchFileFromUri(srcFile.toURI().toString(),
destFile, 10, null);
+ Assertions.assertEquals(destFile.getAbsolutePath(), returned);
+ }
+
+ private HttpServer createLoopbackHttpServer(String response) throws
Exception {
+ HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1",
0), 0);
+ server.createContext(
+ "/keytab",
+ exchange -> {
+ byte[] bytes = response.getBytes(StandardCharsets.UTF_8);
+ exchange.sendResponseHeaders(200, bytes.length);
+ try (OutputStream outputStream = exchange.getResponseBody()) {
+ outputStream.write(bytes);
+ }
+ });
+ return server;
+ }
+}
diff --git
a/common/src/test/java/org/apache/gravitino/utils/TestRemoteUriValidator.java
b/common/src/test/java/org/apache/gravitino/utils/TestRemoteUriValidator.java
new file mode 100644
index 0000000000..5afd96fe82
--- /dev/null
+++
b/common/src/test/java/org/apache/gravitino/utils/TestRemoteUriValidator.java
@@ -0,0 +1,91 @@
+/*
+ * 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.gravitino.utils;
+
+import java.net.URI;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestRemoteUriValidator {
+ private static final String BLOCK_UNSAFE_ADDRESS_CONFIG =
"test.block-unsafe-address";
+
+ @Test
+ public void testRejectLocalAddressesByDefault() {
+ IllegalArgumentException exception =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ RemoteUriValidator.validate(
+ new URI("http://127.0.0.1/";), true,
BLOCK_UNSAFE_ADDRESS_CONFIG));
+ Assertions.assertTrue(exception.getMessage().contains("Gravitino server
side"));
+
Assertions.assertTrue(exception.getMessage().contains(BLOCK_UNSAFE_ADDRESS_CONFIG));
+
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ RemoteUriValidator.validate(
+ new URI("http://localhost/";), true,
BLOCK_UNSAFE_ADDRESS_CONFIG));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ RemoteUriValidator.validate(
+ new URI("http://169.254.169.254/";), true,
BLOCK_UNSAFE_ADDRESS_CONFIG));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ RemoteUriValidator.validate(
+ new URI("http://10.0.0.1/";), true,
BLOCK_UNSAFE_ADDRESS_CONFIG));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ RemoteUriValidator.validate(
+ new URI("http://172.16.0.1/";), true,
BLOCK_UNSAFE_ADDRESS_CONFIG));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ RemoteUriValidator.validate(
+ new URI("http://192.168.0.1/";), true,
BLOCK_UNSAFE_ADDRESS_CONFIG));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ RemoteUriValidator.validate(
+ new URI("http://100.100.100.200/";), true,
BLOCK_UNSAFE_ADDRESS_CONFIG));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ RemoteUriValidator.validate(
+ new URI("http://[fd00::1]/";), true,
BLOCK_UNSAFE_ADDRESS_CONFIG));
+ }
+
+ @Test
+ public void testAllowUnsafeAddressesWhenBlockingDisabled() {
+ Assertions.assertDoesNotThrow(
+ () ->
+ RemoteUriValidator.validate(
+ new URI("http://127.0.0.1/";), false,
BLOCK_UNSAFE_ADDRESS_CONFIG));
+ Assertions.assertDoesNotThrow(
+ () ->
+ RemoteUriValidator.validate(
+ new URI("http://localhost/";), false,
BLOCK_UNSAFE_ADDRESS_CONFIG));
+ Assertions.assertDoesNotThrow(
+ () ->
+ RemoteUriValidator.validate(
+ new URI("http://192.168.0.1/";), false,
BLOCK_UNSAFE_ADDRESS_CONFIG));
+ }
+}
diff --git a/conf/gravitino-iceberg-rest-server.conf.template
b/conf/gravitino-iceberg-rest-server.conf.template
index 77d93a35ae..3aed55216a 100644
--- a/conf/gravitino-iceberg-rest-server.conf.template
+++ b/conf/gravitino-iceberg-rest-server.conf.template
@@ -19,6 +19,8 @@
# THE CONFIGURATION FOR Iceberg REST SERVER
gravitino.iceberg-rest.shutdown.timeout = 3000
+# Whether to block remote file URIs that resolve to unsafe addresses.
+gravitino.fetchFile.blockUnsafeRemoteUri = true
# THE CONFIGURATION FOR Iceberg REST WEB SERVER
# The host name of the built-in web server
diff --git a/conf/gravitino.conf.template b/conf/gravitino.conf.template
index e1c9ba5788..8336246d6a 100644
--- a/conf/gravitino.conf.template
+++ b/conf/gravitino.conf.template
@@ -76,6 +76,9 @@ gravitino.cache.enableWeigher = true
# The cache implementation to use (e.g., caffeine).
gravitino.cache.implementation = caffeine
+# Whether to block remote file URIs that resolve to unsafe addresses.
+gravitino.fetchFile.blockUnsafeRemoteUri = true
+
# THE CONFIGURATION FOR authorization
# Whether Gravitino enable authorization or not
gravitino.authorization.enable = false
diff --git a/core/src/main/java/org/apache/gravitino/Configs.java
b/core/src/main/java/org/apache/gravitino/Configs.java
index 6319cdd739..134f36f2e5 100644
--- a/core/src/main/java/org/apache/gravitino/Configs.java
+++ b/core/src/main/java/org/apache/gravitino/Configs.java
@@ -29,6 +29,7 @@ import org.apache.gravitino.config.ConfigBuilder;
import org.apache.gravitino.config.ConfigConstants;
import org.apache.gravitino.config.ConfigEntry;
import org.apache.gravitino.stats.storage.JdbcPartitionStatisticStorageFactory;
+import org.apache.gravitino.utils.FileFetcher;
import org.apache.gravitino.utils.HierarchicalSchemaUtil;
public class Configs {
@@ -551,6 +552,17 @@ public class Configs {
.checkValue(value -> value > 0,
ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
.createWithDefault(5 * 60 * 1000L); // Default is 5 minutes
+ public static final ConfigEntry<Boolean> BLOCK_UNSAFE_REMOTE_URI =
+ new ConfigBuilder(FileFetcher.BLOCK_UNSAFE_REMOTE_URI_CONFIG)
+ .doc(
+ "Whether to block remote file URIs from resolving to unsafe
addresses from the "
+ + "Gravitino server side. This applies to job files and
catalog files such as "
+ + "Kerberos keytabs. This is enabled by default to prevent
SSRF. Set it to false "
+ + "only when the URI is trusted and access is required.")
+ .version(ConfigConstants.VERSION_1_3_0)
+ .booleanConf()
+ .createWithDefault(true);
+
public static final ConfigEntry<String> SCHEMA_SEPARATOR =
new ConfigBuilder("gravitino.schema.separator")
.doc(
diff --git a/core/src/main/java/org/apache/gravitino/GravitinoEnv.java
b/core/src/main/java/org/apache/gravitino/GravitinoEnv.java
index c845c43c49..700e187b89 100644
--- a/core/src/main/java/org/apache/gravitino/GravitinoEnv.java
+++ b/core/src/main/java/org/apache/gravitino/GravitinoEnv.java
@@ -103,6 +103,7 @@ import org.apache.gravitino.storage.IdGenerator;
import org.apache.gravitino.storage.RandomIdGenerator;
import org.apache.gravitino.tag.TagDispatcher;
import org.apache.gravitino.tag.TagManager;
+import org.apache.gravitino.utils.FileFetcher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -200,6 +201,7 @@ public class GravitinoEnv {
public void initializeBaseComponents(Config config) {
LOG.info("Initializing Gravitino base environment...");
this.config = config;
+ FileFetcher.get().initialize(config.get(Configs.BLOCK_UNSAFE_REMOTE_URI));
this.manageFullComponents = false;
initBaseComponents();
LOG.info("Gravitino base environment is initialized.");
@@ -213,6 +215,7 @@ public class GravitinoEnv {
public void initializeFullComponents(Config config) {
LOG.info("Initializing Gravitino full environment...");
this.config = config;
+ FileFetcher.get().initialize(config.get(Configs.BLOCK_UNSAFE_REMOTE_URI));
this.manageFullComponents = true;
initBaseComponents();
initGravitinoServerComponents();
diff --git a/core/src/main/java/org/apache/gravitino/job/JobManager.java
b/core/src/main/java/org/apache/gravitino/job/JobManager.java
index 95b90e89dc..3e9ae66fdd 100644
--- a/core/src/main/java/org/apache/gravitino/job/JobManager.java
+++ b/core/src/main/java/org/apache/gravitino/job/JobManager.java
@@ -26,7 +26,6 @@ import com.google.common.base.Preconditions;
import java.io.File;
import java.io.IOException;
import java.net.URI;
-import java.nio.file.Files;
import java.time.Instant;
import java.util.Arrays;
import java.util.List;
@@ -61,6 +60,7 @@ import org.apache.gravitino.meta.JobEntity;
import org.apache.gravitino.meta.JobTemplateEntity;
import org.apache.gravitino.metalake.MetalakeManager;
import org.apache.gravitino.storage.IdGenerator;
+import org.apache.gravitino.utils.FileFetcher;
import org.apache.gravitino.utils.NameIdentifierUtil;
import org.apache.gravitino.utils.NamespaceUtil;
import org.apache.gravitino.utils.PrincipalUtils;
@@ -801,30 +801,13 @@ public class JobManager implements JobOperationDispatcher
{
static String fetchFileFromUri(String uri, File stagingDir, int timeoutInMs)
{
try {
URI fileUri = new URI(uri);
- String scheme = Optional.ofNullable(fileUri.getScheme()).orElse("file");
File destFile = new File(stagingDir, new
File(fileUri.getPath()).getName());
-
- switch (scheme) {
- case "http":
- case "https":
- case "ftp":
- FileUtils.copyURLToFile(fileUri.toURL(), destFile, timeoutInMs,
timeoutInMs);
- break;
-
- case "file":
- java.nio.file.Path sourcePath = new File(fileUri.getPath()).toPath();
- if (!Files.exists(sourcePath)) {
- throw new IOException(
- String.format("Source file does not exist: %s",
sourcePath.toAbsolutePath()));
- }
- Files.createSymbolicLink(destFile.toPath(), sourcePath);
- break;
-
- default:
- throw new IllegalArgumentException("Unsupported scheme: " + scheme);
- }
-
- return destFile.getAbsolutePath();
+ return FileFetcher.get()
+ .fetchFileFromUri(
+ uri,
+ destFile,
+ timeoutInMs,
+ null /* hadoopConf: job file URIs never use the hdfs scheme */);
} catch (Exception e) {
throw new RuntimeException(String.format("Failed to fetch file from URI
%s", uri), e);
}
diff --git a/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
b/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
index 2dc4af741f..32348e789f 100644
--- a/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
+++ b/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
@@ -31,8 +31,13 @@ import static org.mockito.Mockito.when;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
+import com.sun.net.httpserver.HttpServer;
import java.io.File;
import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
import java.nio.file.Path;
import java.time.Instant;
import java.util.Collections;
@@ -70,6 +75,7 @@ import org.apache.gravitino.meta.SchemaVersion;
import org.apache.gravitino.metalake.MetalakeManager;
import org.apache.gravitino.storage.IdGenerator;
import org.apache.gravitino.storage.RandomIdGenerator;
+import org.apache.gravitino.utils.FileFetcher;
import org.apache.gravitino.utils.NameIdentifierUtil;
import org.apache.gravitino.utils.NamespaceUtil;
import org.awaitility.Awaitility;
@@ -901,6 +907,20 @@ public class TestJobManager {
.build();
}
+ private HttpServer createLoopbackHttpServer(String response) throws
IOException {
+ HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1",
0), 0);
+ server.createContext(
+ "/artifact.jar",
+ exchange -> {
+ byte[] bytes = response.getBytes(StandardCharsets.UTF_8);
+ exchange.sendResponseHeaders(200, bytes.length);
+ try (OutputStream outputStream = exchange.getResponseBody()) {
+ outputStream.write(bytes);
+ }
+ });
+ return server;
+ }
+
@Test
public void testFetchFileFromUriWithMissingLocalFileShouldFail() throws
IOException {
File stagingDir = new File(testStagingDir);
@@ -914,6 +934,71 @@ public class TestJobManager {
RuntimeException.class, () -> JobManager.fetchFileFromUri(uri,
stagingDir, 1000));
}
+ @Test
+ public void testFetchFileFromUriSsrfBlocked() {
+ File stagingDir = new File(testStagingDir);
+ Assertions.assertTrue(stagingDir.mkdirs() || stagingDir.exists());
+ FileFetcher.get().initialize(true);
+
+ // Loopback address
+ RuntimeException e1 =
+ Assertions.assertThrows(
+ RuntimeException.class,
+ () -> JobManager.fetchFileFromUri("http://127.0.0.1:8090/configs";,
stagingDir, 1000));
+ assertRemoteUriBlockedMessage(e1);
+
+ // AWS / GCP / Azure cloud-metadata endpoint (link-local 169.254.x.x)
+ RuntimeException e2 =
+ Assertions.assertThrows(
+ RuntimeException.class,
+ () ->
+ JobManager.fetchFileFromUri(
+ "http://169.254.169.254/latest/meta-data/";, stagingDir,
1000));
+ assertRemoteUriBlockedMessage(e2);
+
+ // RFC-1918 private range
+ RuntimeException e3 =
+ Assertions.assertThrows(
+ RuntimeException.class,
+ () -> JobManager.fetchFileFromUri("http://192.168.1.1/";,
stagingDir, 1000));
+ assertRemoteUriBlockedMessage(e3);
+
+ // Alibaba Cloud / Oracle Cloud metadata endpoint
+ RuntimeException e4 =
+ Assertions.assertThrows(
+ RuntimeException.class,
+ () -> JobManager.fetchFileFromUri("http://100.100.100.200/";,
stagingDir, 1000));
+ assertRemoteUriBlockedMessage(e4);
+ }
+
+ @Test
+ public void testFetchFileFromUriShouldAllowLocalhostWhenBlockingDisabled()
throws Exception {
+ File stagingDir = new File(testStagingDir);
+ Assertions.assertTrue(stagingDir.mkdirs() || stagingDir.exists());
+ HttpServer server = createLoopbackHttpServer("job artifact");
+
+ try {
+ server.start();
+ int port = server.getAddress().getPort();
+ FileFetcher.get().initialize(false);
+
+ String fetchedFile =
+ JobManager.fetchFileFromUri(
+ String.format("http://127.0.0.1:%d/artifact.jar";, port),
stagingDir, 1000);
+
+ Assertions.assertEquals("job artifact",
Files.readString(Path.of(fetchedFile)));
+ } finally {
+ FileFetcher.get().initialize(true);
+ server.stop(0);
+ }
+ }
+
+ private static void assertRemoteUriBlockedMessage(RuntimeException
exception) {
+
Assertions.assertTrue(exception.getCause().getMessage().contains("Gravitino
server side"));
+ Assertions.assertTrue(
+
exception.getCause().getMessage().contains(FileFetcher.BLOCK_UNSAFE_REMOTE_URI_CONFIG));
+ }
+
@Test
public void testCloseShouldShutdownExecutorsWhenJobExecutorCloseFails()
throws IOException {
JobExecutor failingJobExecutor = Mockito.mock(JobExecutor.class);
diff --git a/docs/apache-hive-catalog.md b/docs/apache-hive-catalog.md
index e76f54c19b..8bef06cf60 100644
--- a/docs/apache-hive-catalog.md
+++ b/docs/apache-hive-catalog.md
@@ -29,19 +29,19 @@ The Hive catalog supports creating, updating, and deleting
databases and tables
Besides the [common catalog
properties](./gravitino-server-config.md#catalog-properties-configuration), the
Hive catalog has the following properties:
-| Property Name | Description
| Default Value | Required
| Since Version |
-|------------------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------|------------------------------|---------------|
-| `metastore.uris` | The Hive metastore service URIs,
separate multiple addresses with commas. Such as `thrift://127.0.0.1:9083`
| (none) | Yes
| 0.2.0 |
-| `client.pool-size` | The maximum number of Hive
metastore clients in the pool for Gravitino.
| 1 | No
| 0.2.0 |
-| `gravitino.bypass.` | Property name with this prefix
passed down to the underlying HMS client for use. Such as
`gravitino.bypass.hive.metastore.failure.retries = 3` indicate 3 times of
retries upon failure of Thrift metastore calls
| (none) | No | 0.2.0 |
-| `client.pool-cache.eviction-interval-ms` | The cache pool eviction interval.
| 300000 | No
| 0.4.0 |
-| `impersonation-enable` | Enable user impersonation for
Hive catalog.
| false | No
| 0.4.0 |
-| `kerberos.principal` | The Kerberos principal for the
catalog. You should configure
`gravitino.bypass.hadoop.security.authentication`,
`gravitino.bypass.hive.metastore.kerberos.principal` and
`gravitino.bypass.hive.metastore.sasl.enabled`if you want to use Kerberos. |
(none) | required if you use kerberos | 0.4.0 |
-| `kerberos.keytab-uri` | The uri of key tab for the
catalog. Now supported protocols are `https`, `http`, `ftp`, `file`.
| (none) |
required if you use kerberos | 0.4.0 |
-| `kerberos.check-interval-sec` | The interval to check validness
of the principal
| 60 | No
| 0.4.0 |
-| `kerberos.keytab-fetch-timeout-sec` | The timeout to fetch key tab
| 60 | No
| 0.4.0 |
-| `list-all-tables` | Whether to list all tables in a
database, including non-Hive tables such as Iceberg, Paimon, and Hudi. When
false, non-Hive tables are filtered out on a best-effort basis; see the note
below for known limitations. | false | No
| 0.5.1 |
-| `default.catalog` | The default catalog name for the
Hive3 metastore backend; this configuration is ignored when using a Hive2
metastore.
| hive | No
| 1.1.0 |
+| Property Name | Description
| Default Value | Required
| Since Version |
+|------------------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|----------------|------------------------------|---------------|
+| `metastore.uris` | The Hive metastore service URIs,
separate multiple addresses with commas. Such as `thrift://127.0.0.1:9083`
| (none) | Yes
| 0.2.0 |
+| `client.pool-size` | The maximum number of Hive
metastore clients in the pool for Gravitino.
| 1 | No
| 0.2.0 |
+| `gravitino.bypass.` | Property name with this prefix
passed down to the underlying HMS client for use. Such as
`gravitino.bypass.hive.metastore.failure.retries = 3` indicate 3 times of
retries upon failure of Thrift metastore calls
| (none) | No | 0.2.0 |
+| `client.pool-cache.eviction-interval-ms` | The cache pool eviction interval.
| 300000 | No
| 0.4.0 |
+| `impersonation-enable` | Enable user impersonation for
Hive catalog.
| false | No
| 0.4.0 |
+| `kerberos.principal` | The Kerberos principal for the
catalog. You should configure
`gravitino.bypass.hadoop.security.authentication`,
`gravitino.bypass.hive.metastore.kerberos.principal` and
`gravitino.bypass.hive.metastore.sasl.enabled`if you want to use Kerberos. |
(none) | required if you use kerberos | 0.4.0 |
+| `kerberos.keytab-uri` | The uri of key tab for the
catalog. Now supported protocols are `https`, `http`, `ftp`, `file`.
| (none) |
required if you use kerberos | 0.4.0 |
+| `kerberos.check-interval-sec` | The interval to check validity of
the principal
| 60 | No
| 0.4.0 |
+| `kerberos.keytab-fetch-timeout-sec` | The timeout to fetch key tab
| 60 | No
| 0.4.0 |
+| `list-all-tables` | Whether to list all tables in a
database, including non-Hive tables such as Iceberg, Paimon, and Hudi. When
false, non-Hive tables are filtered out on a best-effort basis; see the note
below for known limitations. | false |
No | 0.5.1 |
+| `default.catalog` | The default catalog name for the
Hive3 metastore backend; this configuration is ignored when using a Hive2
metastore.
| hive | No
| 1.1.0 |
:::note
When `list-all-tables=false`, the Hive catalog removes the following on a
best-effort basis:
@@ -249,4 +249,3 @@ Refer to [Manage view metadata using
Gravitino](./manage-view-metadata-using-gra
To create a Hive catalog with S3 storage, you can refer to the [Hive catalog
with S3](./hive-catalog-with-cloud-storage.md) documentation. No special
configurations are required for the Hive catalog to work with S3 storage.
The only difference is the storage location of the files, which is in S3. Use
`location` to specify the S3 path for the database or table.
-
diff --git a/docs/fileset-catalog.md b/docs/fileset-catalog.md
index 60f9b51ee8..bb7b315fd1 100644
--- a/docs/fileset-catalog.md
+++ b/docs/fileset-catalog.md
@@ -53,14 +53,14 @@ Refer to [Credential
vending](./security/credential-vending.md) for more details
Apart from the above properties, to access fileset like HDFS fileset, you need
to configure the following extra
properties.
-| Property Name | Description
|
Default Value | Required |
Since Version |
-|----------------------------------------------------|-------------------------------------------------------------------------------------------------|---------------|-------------------------------------------------------------|---------------|
-| `authentication.impersonation-enable` | Whether to enable
impersonation for the Fileset catalog. |
`false` | No |
0.5.1 |
-| `authentication.type` | The type of
authentication for Fileset catalog, we only support `kerberos`, `simple`.
| `simple` | No
| 0.5.1 |
-| `authentication.kerberos.principal` | The principal of the
Kerberos authentication |
(none) | required if the value of `authentication.type` is Kerberos. |
0.5.1 |
-| `authentication.kerberos.keytab-uri` | The URI of The keytab
for the Kerberos authentication. |
(none) | required if the value of `authentication.type` is Kerberos. |
0.5.1 |
-| `authentication.kerberos.check-interval-sec` | The check interval of
Kerberos credential for Fileset catalog. | 60
| No | 0.5.1
|
-| `authentication.kerberos.keytab-fetch-timeout-sec` | The fetch timeout of
retrieving Kerberos keytab from `authentication.kerberos.keytab-uri`. | 60
| No |
0.5.1 |
+| Property Name | Description
| Default
Value | Required | Since
Version |
+|----------------------------------------------------|--------------------------------------------------------------------------------------------|---------------|-------------------------------------------------------------|---------------|
+| `authentication.impersonation-enable` | Whether to enable
impersonation for the Fileset catalog. |
`false` | No |
0.5.1 |
+| `authentication.type` | The type of
authentication for Fileset catalog, we only support `kerberos`, `simple`.
| `simple` | No |
0.5.1 |
+| `authentication.kerberos.principal` | The principal of the
Kerberos authentication | (none)
| required if the value of `authentication.type` is Kerberos. | 0.5.1
|
+| `authentication.kerberos.keytab-uri` | The URI of The keytab
for the Kerberos authentication. | (none)
| required if the value of `authentication.type` is Kerberos. | 0.5.1
|
+| `authentication.kerberos.check-interval-sec` | The check interval of
Kerberos credential for Fileset catalog. | 60
| No | 0.5.1
|
+| `authentication.kerberos.keytab-fetch-timeout-sec` | The fetch timeout of
retrieving Kerberos keytab from `authentication.kerberos.keytab-uri`. | 60
| No | 0.5.1
|
The `config.resources` property allows users to specify custom configuration
files.
diff --git a/docs/gravitino-server-config.md b/docs/gravitino-server-config.md
index c47e0c85f3..a2b8db5700 100644
--- a/docs/gravitino-server-config.md
+++ b/docs/gravitino-server-config.md
@@ -132,6 +132,17 @@ All cache entries are subject to a TTL (Time-To-Live)
expiration policy. By defa
- TTL can work in conjunction with both capacity and weight-based eviction;
- Expired entries will also trigger asynchronous cleanup mechanisms for
resource release and logging.
+### Job Configuration
+
+The following table lists the job configuration items:
+
+| Configuration Item | Description
| Default Value
| Required | Since Version |
+|----------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------|-------------------------------|----------|---------------|
+| `gravitino.job.stagingDir` | Directory for managing staging
files when running jobs.
|
`/tmp/gravitino/jobs/staging` | No | 1.0.0 |
+| `gravitino.job.executor` | The executor to run jobs. By
default it is `local`; users can implement their own executor and set it here.
| `local`
| No | 1.0.0 |
+| `gravitino.job.stagingDirKeepTimeInMs` | The time in milliseconds to keep
the staging files of the finished job in the job staging directory. The minimum
recommended value is 10 minutes if you are not testing. | `604800000` (7 days)
| No | 1.0.0 |
+| `gravitino.job.statusPullIntervalInMs` | The interval in milliseconds to
pull the job status from the job executor. The minimum recommended value is 1
minute if you are not testing. | `300000` (5
minutes) | No | 1.0.0 |
+
### Tree Lock Configuration
The Gravitino server uses a tree lock to ensure data consistency. The tree
lock is an in-memory lock; Gravitino currently supports only in-memory locks.
The configuration items are as follows:
@@ -295,6 +306,10 @@ appender.audit_file.strategy.delete.ifLastModified.age =
90d
Refer to [security](security/security.md) for HTTPS and authentication
configurations.
+| Configuration Item | Description
| Default Value | Required
| Since Version |
+|--------------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------|----------|---------------|
+| `gravitino.fetchFile.blockUnsafeRemoteUri` | Whether to block remote file
URIs from resolving to unsafe addresses from the Gravitino server side. This
applies to job files and catalog files such as Kerberos keytabs. Disable this
only for trusted URIs that require access to such addresses. | `true` |
No | 1.3.0 |
+
### Metrics Configuration
| Property name | Description
| Default value | Required | Since Version |
diff --git a/docs/iceberg-rest-service.md b/docs/iceberg-rest-service.md
index 1a3689092e..66900d8824 100644
--- a/docs/iceberg-rest-service.md
+++ b/docs/iceberg-rest-service.md
@@ -96,6 +96,8 @@ The `gravitino.auxService.iceberg-rest.` prefix has been
deprecated since `0.6.0
If both `gravitino.auxService.iceberg-rest.key` and
`gravitino.iceberg-rest.key` are present, `gravitino.iceberg-rest.key` takes
precedence.
The following sections use the `gravitino.iceberg-rest.` prefix.
+The server-level `gravitino.fetchFile.blockUnsafeRemoteUri` configuration
controls whether remote files such as Kerberos keytabs may resolve to unsafe
addresses. It defaults to `true`. Configure it in
`gravitino-iceberg-rest-server.conf` for standalone mode or `gravitino.conf`
for auxiliary mode.
+
### Service Configuration
#### Auxiliary Service
@@ -429,15 +431,15 @@ Refer to [HTTPS
Configuration](./security/how-to-use-https.md#apache-iceberg-res
For JDBC backend, you can use the `gravitino.iceberg-rest.jdbc-user` and
`gravitino.iceberg-rest.jdbc-password` to authenticate the JDBC connection. For
Hive backend, you can use the `gravitino.iceberg-rest.authentication.type` to
specify the authentication type, and use the
`gravitino.iceberg-rest.authentication.kerberos.principal` and
`gravitino.iceberg-rest.authentication.kerberos.keytab-uri` to authenticate the
Kerberos connection.
The detailed configuration items are as follows:
-| Configuration item |
Description
| Default value | Required
[...]
-|---------------------------------------------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------|------------------------------------------------------------------------------------------------------------------------------------------------------
[...]
-| `gravitino.iceberg-rest.authentication.type` |
The type of authentication for Iceberg rest catalog backend. This configuration
only applicable for for Hive backend, and only supports `Kerberos`, `simple`
currently. As for JDBC backend, only username/password authentication was
supported now. | `simple` | No
[...]
-| `gravitino.iceberg-rest.authentication.impersonation-enable` |
Whether to enable impersonation for the Iceberg catalog
| `false` | No
[...]
-| `gravitino.iceberg-rest.hive.metastore.sasl.enabled` |
Whether to enable SASL authentication protocol when connect to Kerberos Hive
metastore.
| `false` | No, This value should be true in most case(Some
will use SSL protocol, but it rather rare) if the value of
`gravitino.iceberg-rest.authentication.typ [...]
-| `gravitino.iceberg-rest.authentication.kerberos.principal` |
The principal of the Kerberos authentication
| (none) | required if the value of
`gravitino.iceberg-rest.authentication.type` is Kerberos.
[...]
-| `gravitino.iceberg-rest.authentication.kerberos.keytab-uri` |
The URI of The keytab for the Kerberos authentication.
| (none) | required if the value of
`gravitino.iceberg-rest.authentication.type` is Kerberos.
[...]
-| `gravitino.iceberg-rest.authentication.kerberos.check-interval-sec` |
The check interval of Kerberos credential for Iceberg catalog.
| 60 | No
[...]
-| `gravitino.iceberg-rest.authentication.kerberos.keytab-fetch-timeout-sec` |
The fetch timeout of retrieving Kerberos keytab from
`authentication.kerberos.keytab-uri`.
| 60 | No
[...]
+| Configuration item |
Description
| Default value | Required
[...]
+|---------------------------------------------------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------|-------------------------------------------------------------------------------------------------------------------------------------------------------
[...]
+| `gravitino.iceberg-rest.authentication.type` |
The type of authentication for Iceberg rest catalog backend. This configuration
only applicable for Hive backend, and only supports `Kerberos`, `simple`
currently. As for JDBC backend, only username/password authentication was
supported now. | `simple` | No
[...]
+| `gravitino.iceberg-rest.authentication.impersonation-enable` |
Whether to enable impersonation for the Iceberg catalog
| `false` | No
[...]
+| `gravitino.iceberg-rest.hive.metastore.sasl.enabled` |
Whether to enable SASL authentication protocol when connect to Kerberos Hive
metastore.
| `false` | No, This value should be true in most case(Some will
use SSL protocol, but it rather rare) if the value of
`gravitino.iceberg-rest.authentication.type [...]
+| `gravitino.iceberg-rest.authentication.kerberos.principal` |
The principal of the Kerberos authentication
| (none) | required if the value of
`gravitino.iceberg-rest.authentication.type` is Kerberos.
[...]
+| `gravitino.iceberg-rest.authentication.kerberos.keytab-uri` |
The URI of The keytab for the Kerberos authentication.
| (none) | required if the value of
`gravitino.iceberg-rest.authentication.type` is Kerberos.
[...]
+| `gravitino.iceberg-rest.authentication.kerberos.check-interval-sec` |
The check interval of Kerberos credential for Iceberg catalog.
| 60 | No
[...]
+| `gravitino.iceberg-rest.authentication.kerberos.keytab-fetch-timeout-sec` |
The fetch timeout of retrieving Kerberos keytab from
`authentication.kerberos.keytab-uri`.
| 60 | No
[...]
#### Credential Vending
diff --git a/docs/lakehouse-hudi-catalog.md b/docs/lakehouse-hudi-catalog.md
index a3baaa7bad..0b78a1dcbf 100644
--- a/docs/lakehouse-hudi-catalog.md
+++ b/docs/lakehouse-hudi-catalog.md
@@ -44,14 +44,14 @@ Tested and verified with Apache Hudi `0.15.0`.
Users can use the following properties to configure the security of the
catalog backend if needed. For example, if you are using a Kerberos Hive
catalog backend, you must set `authentication.type` to `Kerberos` and provide
`authentication.kerberos.principal` and `authentication.kerberos.keytab-uri`.
-| Property name | Description
| Default value |
Required | Since Version
|
-|----------------------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------|-------------------------------------------------------------|-------------------|
-| `authentication.type` | The type of
authentication for hudi catalog backend. This configuration only applicable for
for hms backend, and only supports `kerberos`, `simple` currently. | `simple`
| No | 1.0.0 |
-| `authentication.impersonation-enable` | Whether to enable
impersonation for the hudi catalog
| `false` |
No | 1.0.0 |
-| `authentication.kerberos.principal` | The principal of the
Kerberos authentication
| (none) |
required if the value of `authentication.type` is kerberos. | 1.0.0 |
-| `authentication.kerberos.keytab-uri` | The URI of The keytab
for the Kerberos authentication.
| (none) |
required if the value of `authentication.type` is kerberos. | 1.0.0 |
-| `authentication.kerberos.check-interval-sec` | The check interval of
Kerberos credential for hudi catalog.
| 60 | No
| 1.0.0 |
-| `authentication.kerberos.keytab-fetch-timeout-sec` | The fetch timeout of
retrieving Kerberos keytab from `authentication.kerberos.keytab-uri`.
| 60 | No
| 1.0.0 |
+| Property name | Description
| Default value |
Required | Since Version |
+|----------------------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------|----------------|--------------------------------------------------------------|----------------|
+| `authentication.type` | The type of
authentication for hudi catalog backend. This configuration only applicable for
hms backend, and only supports `kerberos`, `simple` currently. | `simple`
| No | 1.0.0
|
+| `authentication.impersonation-enable` | Whether to enable
impersonation for the hudi catalog
| `false` |
No | 1.0.0 |
+| `authentication.kerberos.principal` | The principal of the
Kerberos authentication
| (none) |
required if the value of `authentication.type` is kerberos. | 1.0.0 |
+| `authentication.kerberos.keytab-uri` | The URI of The keytab
for the Kerberos authentication.
| (none) |
required if the value of `authentication.type` is kerberos. | 1.0.0 |
+| `authentication.kerberos.check-interval-sec` | The check interval of
Kerberos credential for hudi catalog.
| 60 | No
| 1.0.0 |
+| `authentication.kerberos.keytab-fetch-timeout-sec` | The fetch timeout of
retrieving Kerberos keytab from `authentication.kerberos.keytab-uri`.
| 60 | No
| 1.0.0 |
Property name with this prefix passed down to the underlying backend client
for use. Such as
`gravitino.bypass.hive.metastore.kerberos.principal=XXXX`、`gravitino.bypass.hadoop.security.authentication=kerberos`、`gravitino.bypass.hive.metastore.sasl.enabled=ture`
And so on.
diff --git
a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/authentication/kerberos/FetchFileUtils.java
b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/authentication/kerberos/FetchFileUtils.java
deleted file mode 100644
index 96d91765a2..0000000000
---
a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/authentication/kerberos/FetchFileUtils.java
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
- * 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.gravitino.iceberg.common.authentication.kerberos;
-
-import java.io.File;
-import java.io.IOException;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.nio.file.Files;
-import java.util.Optional;
-import org.apache.commons.io.FileUtils;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-
-public class FetchFileUtils {
-
- private FetchFileUtils() {}
-
- public static void fetchFileFromUri(
- String fileUri, File destFile, int timeout, Configuration conf) throws
IOException {
- try {
- URI uri = new URI(fileUri);
- String scheme = Optional.ofNullable(uri.getScheme()).orElse("file");
-
- switch (scheme) {
- case "http":
- case "https":
- case "ftp":
- FileUtils.copyURLToFile(uri.toURL(), destFile, timeout * 1000,
timeout * 1000);
- break;
-
- case "file":
- Files.createSymbolicLink(destFile.toPath(), new
File(uri.getPath()).toPath());
- break;
-
- case "hdfs":
- FileSystem.get(conf).copyToLocalFile(new Path(uri), new
Path(destFile.toURI()));
- break;
-
- default:
- throw new IllegalArgumentException(
- String.format("Doesn't support the scheme %s", scheme));
- }
- } catch (URISyntaxException ue) {
- throw new IllegalArgumentException("The uri of file has the wrong
format", ue);
- }
- }
-}
diff --git
a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/authentication/kerberos/KerberosClient.java
b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/authentication/kerberos/KerberosClient.java
index 12b612fa82..be4536e03e 100644
---
a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/authentication/kerberos/KerberosClient.java
+++
b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/authentication/kerberos/KerberosClient.java
@@ -30,6 +30,7 @@ import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import lombok.Getter;
import org.apache.commons.lang3.StringUtils;
+import org.apache.gravitino.utils.FileFetcher;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.security.UserGroupInformation;
import org.slf4j.Logger;
@@ -120,7 +121,8 @@ public class KerberosClient implements Closeable {
// TODO: Make the configuration
int fetchKeytabFileTimeout = kerberosConfig.getFetchTimeoutSec();
- FetchFileUtils.fetchFileFromUri(keyTabUri, keytabFile,
fetchKeytabFileTimeout, hadoopConf);
+ FileFetcher.get()
+ .fetchFileFromUri(keyTabUri, keytabFile, fetchKeytabFileTimeout *
1000, hadoopConf);
return keytabFile;
}
