This is an automated email from the ASF dual-hosted git repository. gyfora pushed a commit to branch release-1.16 in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
commit 57d3f341a1cfcf08856158545f91dd590eac5bca Author: Purushottam Sinha <[email protected]> AuthorDate: Wed Aug 19 13:26:21 2026 +0530 [FLINK-40400] Harden job artifact fetching in kubernetes operator (#1180) * [FLINK-40400] Harden job artifact fetching in kubernetes operator. Ensure proper artifact validation. Generated-by: Claude Code --- .../operator/artifact/ArtifactManager.java | 13 +- .../operator/artifact/HttpArtifactFetcher.java | 128 +++++++++++++- .../operator/utils/JarUriValidationUtils.java | 102 ++++++++++++ .../operator/validation/DefaultValidator.java | 59 +------ .../operator/artifact/ArtifactManagerTest.java | 184 ++++++++++++++++++++- 5 files changed, 420 insertions(+), 66 deletions(-) diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/artifact/ArtifactManager.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/artifact/ArtifactManager.java index f40ae67b..7e229573 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/artifact/ArtifactManager.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/artifact/ArtifactManager.java @@ -20,6 +20,7 @@ package org.apache.flink.kubernetes.operator.artifact; import org.apache.flink.configuration.Configuration; import org.apache.flink.kubernetes.operator.api.spec.FlinkSessionJobSpec; import org.apache.flink.kubernetes.operator.config.FlinkConfigManager; +import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions; import org.apache.flink.util.FlinkRuntimeException; import io.fabric8.kubernetes.api.model.ObjectMeta; @@ -58,7 +59,17 @@ public class ArtifactManager { createIfNotExists(targetDir); URI uri = new URI(jarURI); if ("http".equals(uri.getScheme()) || "https".equals(uri.getScheme())) { - return HttpArtifactFetcher.INSTANCE.fetch(jarURI, flinkConfiguration, targetDir); + // Take the scheme/host policy from the operator config (matching DefaultValidator); + // clone so the caller's config is not mutated. + var operatorConfig = configManager.getOperatorConfiguration(); + var fetchConfig = flinkConfiguration.clone(); + fetchConfig.set( + KubernetesOperatorConfigOptions.JAR_URI_ALLOWED_SCHEMES, + operatorConfig.getJarUriAllowedSchemes()); + fetchConfig.set( + KubernetesOperatorConfigOptions.JAR_URI_DISALLOW_RESTRICTED_HOSTS, + operatorConfig.isJarUriDisallowRestrictedHosts()); + return HttpArtifactFetcher.INSTANCE.fetch(jarURI, fetchConfig, targetDir); } else { return FileSystemBasedArtifactFetcher.INSTANCE.fetch( jarURI, flinkConfiguration, targetDir); diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/artifact/HttpArtifactFetcher.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/artifact/HttpArtifactFetcher.java index c2c6c2db..09e517c6 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/artifact/HttpArtifactFetcher.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/artifact/HttpArtifactFetcher.java @@ -19,6 +19,7 @@ package org.apache.flink.kubernetes.operator.artifact; import org.apache.flink.configuration.Configuration; import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions; +import org.apache.flink.kubernetes.operator.utils.JarUriValidationUtils; import org.apache.commons.io.FileUtils; import org.apache.commons.io.FilenameUtils; @@ -26,38 +27,141 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.File; +import java.io.IOException; import java.net.HttpURLConnection; +import java.net.MalformedURLException; import java.net.URL; import java.util.Map; -/** Download the jar from the http resource. */ +/** + * Download the jar from the http resource. The scheme allowlist and restricted-host policy are read + * from the given configuration; {@link ArtifactManager} sets them from the operator configuration + * before calling. + */ public class HttpArtifactFetcher implements ArtifactFetcher { public static final Logger LOG = LoggerFactory.getLogger(HttpArtifactFetcher.class); public static final HttpArtifactFetcher INSTANCE = new HttpArtifactFetcher(); + // Maximum number of redirects to follow before giving up. + private static final int MAX_REDIRECTS = 5; + @Override public File fetch(String uri, Configuration flinkConfiguration, File targetDir) throws Exception { var start = System.currentTimeMillis(); - URL url = new URL(uri); - HttpURLConnection conn = (HttpURLConnection) url.openConnection(); + + // Scheme allowlist and restricted-host policy, set by ArtifactManager from the operator + // configuration. + var allowedSchemes = + flinkConfiguration.get(KubernetesOperatorConfigOptions.JAR_URI_ALLOWED_SCHEMES); + var disallowRestrictedHosts = + flinkConfiguration.get( + KubernetesOperatorConfigOptions.JAR_URI_DISALLOW_RESTRICTED_HOSTS); // merged session job level header and cluster level header, session job level header take // precedence. Map<String, String> headers = flinkConfiguration.get(KubernetesOperatorConfigOptions.JAR_ARTIFACT_HTTP_HEADER); - if (headers != null) { - headers.forEach(conn::setRequestProperty); - } + // Follow redirects manually so each hop is validated against the same policy as the + // original URI. + String currentUri = uri; + URL originalUrl = null; + URL currentUrl; + HttpURLConnection conn; + int redirects = 0; + while (true) { + var validationError = + JarUriValidationUtils.validateJarURI( + currentUri, allowedSchemes, disallowRestrictedHosts); + if (validationError.isPresent()) { + throw new IOException( + "Refusing to fetch artifact from '" + + currentUri + + "': " + + validationError.get()); + } + + currentUrl = new URL(currentUri); + if (originalUrl == null) { + originalUrl = currentUrl; + } + conn = (HttpURLConnection) currentUrl.openConnection(); + conn.setInstanceFollowRedirects(false); + // Only send the configured headers to the original host; drop them on a cross-host + // redirect. + if (headers != null && originalUrl.getHost().equalsIgnoreCase(currentUrl.getHost())) { + headers.forEach(conn::setRequestProperty); + } + conn.setRequestMethod("GET"); - conn.setRequestMethod("GET"); + // Release the connection on every path except the final (non-redirect) one, whose body + // is streamed below. This covers getResponseCode() and the redirect handling throwing. + boolean keepConnection = false; + try { + int status = conn.getResponseCode(); + if (!isRedirect(status)) { + keepConnection = true; + break; + } - String fileName = FilenameUtils.getName(url.getPath()); + String location = conn.getHeaderField("Location"); + if (location == null || location.isEmpty()) { + throw new IOException( + "Received redirect (status " + + status + + ") from '" + + currentUri + + "' without a Location header"); + } + if (++redirects > MAX_REDIRECTS) { + throw new IOException( + "Too many redirects (>" + + MAX_REDIRECTS + + ") while fetching artifact from '" + + uri + + "'"); + } + URL nextUrl; + try { + nextUrl = new URL(currentUrl, location); + } catch (MalformedURLException e) { + throw new IOException( + "Refusing to follow redirect from '" + + currentUri + + "' to '" + + location + + "': " + + e.getMessage()); + } + // An HTTP fetch only follows http(s) redirects, even if other schemes (e.g. s3, + // hdfs) are in the jarURI allowlist for top-level use. + var nextScheme = nextUrl.getProtocol(); + if (!"http".equalsIgnoreCase(nextScheme) && !"https".equalsIgnoreCase(nextScheme)) { + throw new IOException( + "Refusing to follow redirect from '" + + currentUri + + "' to non-http(s) target '" + + nextUrl + + "'"); + } + currentUri = nextUrl.toString(); + } finally { + if (!keepConnection) { + conn.disconnect(); + } + } + } + + // Name the file from the original jarURI, not the redirect target, so a redirect can't + // change it (e.g. drop the .jar extension the JobManager upload requires). + String fileName = FilenameUtils.getName(originalUrl.getPath()); File targetFile = new File(targetDir, fileName); try (var inputStream = conn.getInputStream()) { FileUtils.copyToFile(inputStream, targetFile); + } finally { + conn.disconnect(); } LOG.debug( "Copied file from {} to {}, cost {} ms", @@ -66,4 +170,12 @@ public class HttpArtifactFetcher implements ArtifactFetcher { System.currentTimeMillis() - start); return targetFile; } + + private static boolean isRedirect(int status) { + return status == HttpURLConnection.HTTP_MOVED_PERM + || status == HttpURLConnection.HTTP_MOVED_TEMP + || status == HttpURLConnection.HTTP_SEE_OTHER + || status == 307 // Temporary Redirect + || status == 308; // Permanent Redirect + } } diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/JarUriValidationUtils.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/JarUriValidationUtils.java new file mode 100644 index 00000000..45b2c0f1 --- /dev/null +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/JarUriValidationUtils.java @@ -0,0 +1,102 @@ +/* + * 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.flink.kubernetes.operator.utils; + +import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions; + +import java.net.InetAddress; +import java.net.URI; +import java.net.URISyntaxException; +import java.net.UnknownHostException; +import java.util.Collection; +import java.util.Locale; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; + +/** + * Shared jarURI validation (scheme allowlist plus restricted-host checks), used both at + * admission/reconcile time and to re-validate every hop an artifact fetch is redirected through. + */ +public final class JarUriValidationUtils { + + private JarUriValidationUtils() {} + + public static Optional<String> validateJarURI( + String jarURI, Collection<String> allowedSchemes, boolean disallowRestrictedHosts) { + if (jarURI == null) { + return Optional.empty(); + } + + URI uri; + try { + uri = new URI(jarURI); + } catch (URISyntaxException e) { + return Optional.of("jarURI is not a valid URI: " + e.getMessage()); + } + + String scheme = uri.getScheme(); + if (scheme == null) { + return Optional.of("jarURI must include a scheme"); + } + + Set<String> normalizedAllowedSchemes = + allowedSchemes.stream() + .map(s -> s.toLowerCase(Locale.ROOT)) + .collect(Collectors.toSet()); + if (!normalizedAllowedSchemes.contains(scheme.toLowerCase(Locale.ROOT))) { + return Optional.of( + String.format( + "jarURI scheme '%s' is not in the allowlist %s. Configure '%s' to extend the allowlist.", + scheme, + normalizedAllowedSchemes, + KubernetesOperatorConfigOptions.JAR_URI_ALLOWED_SCHEMES.key())); + } + + if (("http".equalsIgnoreCase(scheme) || "https".equalsIgnoreCase(scheme)) + && disallowRestrictedHosts) { + String host = uri.getHost(); + if (host == null || host.isEmpty()) { + return Optional.of("jarURI must include a host for http/https schemes"); + } + InetAddress[] addresses; + try { + // Check every resolved address, not just the first, since a host can resolve to + // multiple A/AAAA records. + addresses = InetAddress.getAllByName(host); + } catch (UnknownHostException e) { + return Optional.of("jarURI host '" + host + "' cannot be resolved"); + } + for (InetAddress addr : addresses) { + if (isRestricted(addr)) { + return Optional.of( + "jarURI host '" + host + "' resolves to a restricted address"); + } + } + } + return Optional.empty(); + } + + private static boolean isRestricted(InetAddress addr) { + return addr.isLoopbackAddress() + || addr.isLinkLocalAddress() + || addr.isSiteLocalAddress() + || addr.isAnyLocalAddress() + || addr.isMulticastAddress(); + } +} diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java index 20b3c5c8..7b7186b2 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java @@ -48,6 +48,7 @@ import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptio import org.apache.flink.kubernetes.operator.exception.ReconciliationException; import org.apache.flink.kubernetes.operator.utils.FlinkStateSnapshotUtils; import org.apache.flink.kubernetes.operator.utils.IngressUtils; +import org.apache.flink.kubernetes.operator.utils.JarUriValidationUtils; import org.apache.flink.kubernetes.operator.utils.ResourceConfigUtils; import org.apache.flink.kubernetes.utils.Constants; import org.apache.flink.runtime.clusterframework.TaskExecutorProcessUtils; @@ -62,20 +63,14 @@ import org.slf4j.LoggerFactory; import javax.annotation.Nullable; -import java.net.InetAddress; -import java.net.URI; -import java.net.URISyntaxException; -import java.net.UnknownHostException; import java.util.ArrayList; import java.util.Collection; import java.util.List; -import java.util.Locale; import java.util.Map; import java.util.Optional; import java.util.Set; import java.util.regex.Matcher; import java.util.regex.Pattern; -import java.util.stream.Collectors; /** Default validator implementation for {@link FlinkDeployment}. */ public class DefaultValidator implements FlinkResourceValidator { @@ -316,56 +311,8 @@ public class DefaultValidator implements FlinkResourceValidator { @VisibleForTesting static Optional<String> validateJarURI( String jarURI, Collection<String> allowedSchemes, boolean disallowRestrictedHosts) { - if (jarURI == null) { - return Optional.empty(); - } - - URI uri; - try { - uri = new URI(jarURI); - } catch (URISyntaxException e) { - return Optional.of("jarURI is not a valid URI: " + e.getMessage()); - } - - String scheme = uri.getScheme(); - if (scheme == null) { - return Optional.of("jarURI must include a scheme"); - } - - Set<String> normalizedAllowedSchemes = - allowedSchemes.stream() - .map(s -> s.toLowerCase(Locale.ROOT)) - .collect(Collectors.toSet()); - if (!normalizedAllowedSchemes.contains(scheme.toLowerCase(Locale.ROOT))) { - return Optional.of( - String.format( - "jarURI scheme '%s' is not in the allowlist %s. Configure '%s' to extend the allowlist.", - scheme, - normalizedAllowedSchemes, - KubernetesOperatorConfigOptions.JAR_URI_ALLOWED_SCHEMES.key())); - } - - if (("http".equalsIgnoreCase(scheme) || "https".equalsIgnoreCase(scheme)) - && disallowRestrictedHosts) { - String host = uri.getHost(); - if (host == null || host.isEmpty()) { - return Optional.of("jarURI must include a host for http/https schemes"); - } - InetAddress addr; - try { - addr = InetAddress.getByName(host); - } catch (UnknownHostException e) { - return Optional.of("jarURI host '" + host + "' cannot be resolved"); - } - if (addr.isLoopbackAddress() - || addr.isLinkLocalAddress() - || addr.isSiteLocalAddress() - || addr.isAnyLocalAddress() - || addr.isMulticastAddress()) { - return Optional.of("jarURI host '" + host + "' resolves to a restricted address"); - } - } - return Optional.empty(); + return JarUriValidationUtils.validateJarURI( + jarURI, allowedSchemes, disallowRestrictedHosts); } private Optional<String> validateSessionJobJarURI(FlinkSessionJob sessionJob) { diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/artifact/ArtifactManagerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/artifact/ArtifactManagerTest.java index 422c08b1..1b785784 100644 --- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/artifact/ArtifactManagerTest.java +++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/artifact/ArtifactManagerTest.java @@ -42,6 +42,7 @@ import java.net.HttpURLConnection; import java.net.InetSocketAddress; import java.net.URL; import java.nio.file.Path; +import java.util.List; import java.util.Map; /** Test for {@link ArtifactManager}. */ @@ -53,11 +54,21 @@ public class ArtifactManagerTest { @BeforeEach public void setup() { + // The test server binds to loopback, so the operator policy must permit http + loopback. + artifactManager = artifactManagerWithPolicy(List.of("http"), false); + } + + private ArtifactManager artifactManagerWithPolicy( + List<String> allowedSchemes, boolean disallowRestrictedHosts) { Configuration configuration = new Configuration(); configuration.setString( KubernetesOperatorConfigOptions.OPERATOR_USER_ARTIFACTS_BASE_DIR, tempDir.toAbsolutePath().toString()); - artifactManager = new ArtifactManager(new FlinkConfigManager(configuration)); + configuration.set(KubernetesOperatorConfigOptions.JAR_URI_ALLOWED_SCHEMES, allowedSchemes); + configuration.set( + KubernetesOperatorConfigOptions.JAR_URI_DISALLOW_RESTRICTED_HOSTS, + disallowRestrictedHosts); + return new ArtifactManager(new FlinkConfigManager(configuration)); } @Test @@ -117,6 +128,160 @@ public class ArtifactManagerTest { } } + @Test + public void testHttpFetchFollowsRedirectToAllowedTarget() throws Exception { + HttpServer httpServer = null; + try { + httpServer = startHttpServer(); + var port = httpServer.getAddress().getPort(); + var sourceFile = mockTheJarFile(); + httpServer.createContext("/download/file.jar", new DownloadFileHttpHandler(sourceFile)); + httpServer.createContext( + "/myjob.jar", + new RedirectHttpHandler( + String.format("http://127.0.0.1:%d/download/file.jar";, port))); + + var file = + artifactManager.fetch( + String.format("http://127.0.0.1:%d/myjob.jar";, port), + new Configuration(), + tempDir.toString()); + Assertions.assertTrue(file.exists()); + // Content comes from the redirect target, but the name from the original jarURI. + Assertions.assertEquals("myjob.jar", file.getName()); + Assertions.assertEquals(sourceFile.length(), file.length()); + } finally { + if (httpServer != null) { + httpServer.stop(0); + } + } + } + + @Test + public void testHttpFetchBlocksRedirectToNonHttpScheme() throws Exception { + // An http fetch must only follow http(s) redirects. Here a JDK-recognized non-http scheme + // (ftp) is rejected cleanly. + HttpServer httpServer = null; + try { + httpServer = startHttpServer(); + var port = httpServer.getAddress().getPort(); + httpServer.createContext( + "/redirect", + new RedirectHttpHandler(String.format("ftp://127.0.0.1:%d/job.jar";, port))); + + var ex = + Assertions.assertThrows( + IOException.class, + () -> + artifactManager.fetch( + String.format("http://127.0.0.1:%d/redirect";, port), + new Configuration(), + tempDir.toString())); + Assertions.assertTrue(ex.getMessage().contains("non-http(s) target"), ex.getMessage()); + } finally { + if (httpServer != null) { + httpServer.stop(0); + } + } + } + + @Test + public void testHttpFetchBlocksRedirectToFilesystemScheme() throws Exception { + // An http server redirecting to an s3/hdfs target (a Flink filesystem scheme, not a + // java.net URL protocol) must fail closed cleanly rather than with a raw error. + HttpServer httpServer = null; + try { + httpServer = startHttpServer(); + var port = httpServer.getAddress().getPort(); + httpServer.createContext("/redirect", new RedirectHttpHandler("s3://bucket/job.jar")); + + var ex = + Assertions.assertThrows( + IOException.class, + () -> + artifactManager.fetch( + String.format("http://127.0.0.1:%d/redirect";, port), + new Configuration(), + tempDir.toString())); + Assertions.assertTrue( + ex.getMessage().contains("Refusing to follow redirect"), ex.getMessage()); + } finally { + if (httpServer != null) { + httpServer.stop(0); + } + } + } + + @Test + public void testHttpFetchBlocksTooManyRedirects() throws Exception { + HttpServer httpServer = null; + try { + httpServer = startHttpServer(); + var port = httpServer.getAddress().getPort(); + httpServer.createContext( + "/loop", + new RedirectHttpHandler(String.format("http://127.0.0.1:%d/loop";, port))); + + var ex = + Assertions.assertThrows( + IOException.class, + () -> + artifactManager.fetch( + String.format("http://127.0.0.1:%d/loop";, port), + new Configuration(), + tempDir.toString())); + Assertions.assertTrue(ex.getMessage().contains("Too many redirects"), ex.getMessage()); + } finally { + if (httpServer != null) { + httpServer.stop(0); + } + } + } + + @Test + public void testOperatorConfigControlsRestrictedHostPolicy() { + // The restricted-host policy comes from the operator config; a value set in the per-job + // config does not override it. No server is needed: the loopback host is rejected first. + var strictManager = artifactManagerWithPolicy(List.of("http"), true); + var jobConfig = + new Configuration() + .set( + KubernetesOperatorConfigOptions.JAR_URI_DISALLOW_RESTRICTED_HOSTS, + false); + + var ex = + Assertions.assertThrows( + IOException.class, + () -> + strictManager.fetch( + "http://127.0.0.1:9999/job.jar";, + jobConfig, + tempDir.toString())); + Assertions.assertTrue(ex.getMessage().contains("restricted address"), ex.getMessage()); + } + + @Test + public void testOperatorConfigControlsSchemeAllowlist() { + // The scheme allowlist comes from the operator config; a value set in the per-job config + // does not override it. + var strictManager = artifactManagerWithPolicy(List.of("https"), false); + var jobConfig = + new Configuration() + .set( + KubernetesOperatorConfigOptions.JAR_URI_ALLOWED_SCHEMES, + List.of("http")); + + var ex = + Assertions.assertThrows( + IOException.class, + () -> + strictManager.fetch( + "http://127.0.0.1:9999/job.jar";, + jobConfig, + tempDir.toString())); + Assertions.assertTrue(ex.getMessage().contains("scheme 'http'"), ex.getMessage()); + } + private HttpServer startHttpServer() throws IOException { int port = RandomUtils.nextInt(2000, 3000); HttpServer httpServer = null; @@ -160,4 +325,21 @@ public class ArtifactManagerTest { exchange.close(); } } + + /** Handler that always responds with a 302 redirect to the configured location. */ + public static class RedirectHttpHandler implements HttpHandler { + + private final String location; + + public RedirectHttpHandler(String location) { + this.location = location; + } + + @Override + public void handle(HttpExchange exchange) throws IOException { + exchange.getResponseHeaders().add("Location", location); + exchange.sendResponseHeaders(HttpURLConnection.HTTP_MOVED_TEMP, -1); + exchange.close(); + } + } }
