dalelane commented on code in PR #1180: URL: https://github.com/apache/flink-kubernetes-operator/pull/1180#discussion_r3803159725
########## flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/JarUriValidationUtils.java: ########## @@ -0,0 +1,93 @@ +/* + * 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 addr; + try { + addr = InetAddress.getByName(host); Review Comment: I'm not sure how paranoid we want to be here... but we could use `InetAddress.getAllByName(host)` and check all the InetAddress that match, not just the first one. (This is about catching addresses with more than one A record, where the first one might validate, but connections resolve to a different one) Like I say... that might be overly paranoid. I think what you've got works, I'm just spit-balling about ways that someone determined could bypass it. ########## flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/artifact/HttpArtifactFetcher.java: ########## @@ -19,42 +19,135 @@ 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; 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"); + int status = conn.getResponseCode(); + if (!isRedirect(status)) { + break; + } - String fileName = FilenameUtils.getName(url.getPath()); + String location = conn.getHeaderField("Location"); Review Comment: I think there is a risk of a connection leak here if any (non-redirect) errors happen before we get to `conn.disconnect()` - e.g. getResponseCode() throws an IOException -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
