Dennis-Mircea commented on code in PR #1180:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/1180#discussion_r3803824813
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/artifact/HttpArtifactFetcher.java:
##########
@@ -19,45 +19,149 @@
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");
+ // 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);
Review Comment:
Shouldn't we have here a download size cap as well? A malicious URL, or a
redirect to one, can serve an enormous body and fill the operator's ephemeral
storage.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/artifact/HttpArtifactFetcher.java:
##########
@@ -19,45 +19,149 @@
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();
Review Comment:
The connection here resolves the host again when it connects
(`getResponseCode()` below), independent of the addresses `validateJarURI`
already checked with `getAllByName`. So a rebinding DNS answer can hand the
socket a different, internal address than the one that was validated. In
practice the JVM DNS cache usually makes the two lookups agree, so the window
is narrow unless caching is disabled, but pinning the validated IP for the
connection (with the Host header preserved) is what closes it fully.
Also, shouldn't we have a timeout here? A jarURI (or a redirect target)
pointing at a server that accepts the socket and then stalls hangs the fetch
forever, and this runs on the reconcile thread, so it ties up a reconcile
worker indefinitely.
--
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]
