You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "spuru9 (via GitHub)" <gi...@apache.org> on 2026/08/20 15:15:37 UTC
[PR] [FLINK-40448] Pin resolved address during operator artifact fetch to prevent DNS rebinding [flink-kubernetes-operator]
spuru9 opened a new pull request, #1188:
URL: https://github.com/apache/flink-kubernetes-operator/pull/1188
## What is the purpose of the change
The operator validates a `FlinkSessionJob`'s `jarURI` host against the restricted-host policy, but the subsequent HTTP fetch resolved the hostname again when opening the connection. A hostname that passed validation could therefore resolve to a different (e.g. internal) address at connect time — a DNS-rebinding / TOCTOU gap.
This change resolves the host once during validation and pins the connection to that resolved address for each redirect hop, so the address that was checked is the address that is connected to. Split out of #1186 (FLINK-40402) per review, as it is independent of the artifact-fetch timeout / size-limit work.
## Brief change log
- `JarUriValidationUtils#validateAndResolve` returns the resolved address alongside the existing scheme / restricted-host validation.
- `HttpArtifactFetcher` connects to the resolved address (as an IP literal) for each redirect hop instead of re-resolving the hostname.
- The original hostname is preserved for the `Host` request header and, over HTTPS, for TLS SNI and certificate verification, so pinning does not break virtual hosting or the TLS handshake.
## Verifying this change
This change added tests and can be verified as follows:
- `ArtifactManagerTest#testHttpFetchPreservesOriginalHostnameWhenConnectingToPinnedAddress` — asserts the original hostname is sent as the `Host` header while the connection is pinned to the resolved address.
- `ArtifactManagerTest#testHttpsFetchPinsAddressAndPreservesHostnameForTls` — asserts HTTPS SNI / certificate verification use the original hostname (cert SAN = `localhost`), not the pinned IP literal.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changes to the `CustomResourceDescriptors`: no
- Core observer or reconciler logic that is regularly executed: yes (the artifact fetch performed during `FlinkSessionJob` reconciliation)
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? not applicable
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (Claude Code)
Generated-by: Claude Code
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40448][operator] Pin resolved address during operator artifact fetch to prevent DNS rebinding [flink-kubernetes-operator]
Posted by "Copilot (via GitHub) via issues" <is...@flink.apache.org>.
Copilot commented on code in PR #1188:
URL: https://github.com/apache/flink-kubernetes-operator/pull/1188#discussion_r4530385209
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/artifact/HttpArtifactFetcher.java:
##########
@@ -264,4 +287,124 @@ private static boolean isRedirect(int status) {
|| status == 307 // Temporary Redirect
|| status == 308; // Permanent Redirect
}
+
+ @VisibleForTesting
+ static Optional<InetAddress> validateHop(
+ String uri, Collection<String> allowedSchemes, boolean disallowRestrictedHosts)
+ throws IOException {
+ var validation =
+ JarUriValidationUtils.validateAndResolve(
+ uri, allowedSchemes, disallowRestrictedHosts);
+ if (validation.getError().isPresent()) {
+ throw new IOException(
+ "Refusing to fetch artifact from '"
+ + uri
+ + "': "
+ + validation.getError().get());
+ }
+ return validation.getCheckedAddress();
+ }
+
+ private static boolean isDirect(String uri) {
+ var selector = ProxySelector.getDefault();
+ return selector == null
+ || selector.select(URI.create(uri)).stream()
+ .allMatch(proxy -> proxy.type() == Proxy.Type.DIRECT);
Review Comment:
When a ProxySelector returns [HTTP proxy, DIRECT], this predicate disables pinning for the entire list, but HttpURLConnection can fall back to an unpinned direct connection if the proxy fails. That fallback resolves the hostname again and leaves the restricted-host policy vulnerable to DNS rebinding. Handle proxy attempts explicitly so every DIRECT attempt is pinned (or fail closed), and add a regression test for proxy failure followed by DIRECT.
##########
flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/artifact/HttpArtifactFetcherTest.java:
##########
@@ -0,0 +1,422 @@
+/*
+ * 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.artifact;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
+import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
+
+import com.sun.net.httpserver.Headers;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+import com.sun.net.httpserver.HttpServer;
+import com.sun.net.httpserver.HttpsConfigurator;
+import com.sun.net.httpserver.HttpsExchange;
+import com.sun.net.httpserver.HttpsServer;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import javax.net.ssl.ExtendedSSLSession;
+import javax.net.ssl.HttpsURLConnection;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SNIHostName;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLHandshakeException;
+import javax.net.ssl.TrustManagerFactory;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.Proxy;
+import java.net.ProxySelector;
+import java.net.ServerSocket;
+import java.net.SocketAddress;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.security.KeyStore;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link HttpArtifactFetcher}. */
+class HttpArtifactFetcherTest {
+
+ private static final byte[] ARTIFACT = "artifact-bytes".getBytes(StandardCharsets.UTF_8);
+ private static final byte[] LOOPBACK_BYTES = {127, 0, 0, 1};
+ private static final char[] KEYSTORE_PASSWORD = "changeit".toCharArray();
+ private static final String REJECTED_HOST = "rejected.invalid";
+
+ @TempDir Path tempDir;
+
+ private final List<RecordedRequest> requests = new CopyOnWriteArrayList<>();
+ private final List<HttpServer> servers = new ArrayList<>();
+ private final RecordingValidator validator = new RecordingValidator();
+ private final HttpArtifactFetcher fetcher = new HttpArtifactFetcher(validator);
+ private HttpServer server;
+ private int port;
+
+ @BeforeEach
+ void startServer() throws IOException {
+ server = start(HttpServer.create(new InetSocketAddress(loopback(), 0), 0));
+ server.createContext("/job.jar", serveArtifact(requests));
+ port = server.getAddress().getPort();
+ }
+
+ @AfterEach
+ void stopServers() {
+ servers.forEach(s -> s.stop(0));
+ }
+
+ @Test
+ void connectsToCheckedAddressAndSendsOriginalHost() throws Exception {
+ var file = fetch("http://artifact.invalid:" + port + "/job.jar", Map.of());
+
+ assertThat(file).hasBinaryContent(ARTIFACT);
+ assertThat(requests).hasSize(1);
+ assertThat(requests.get(0).header("Host")).isEqualTo("artifact.invalid:" + port);
+ }
+
+ @Test
+ void validatesAndPinsEachRedirectHopAndDropsConfiguredHeadersOnHostChange() throws Exception {
+ var start = "http://artifact.invalid:" + port + "/start";
+ var target = "http://other.invalid:" + port + "/job.jar";
+ server.createContext("/start", exchange -> redirect(exchange, target));
+
+ fetch(start, Map.of("X-Token", "secret"));
+
+ assertThat(validator.validatedUris).containsExactly(start, target);
+ assertThat(requests).hasSize(2);
+ assertThat(requests.get(0).header("Host")).isEqualTo("artifact.invalid:" + port);
+ assertThat(requests.get(0).header("X-Token")).isEqualTo("secret");
+ assertThat(requests.get(1).header("Host")).isEqualTo("other.invalid:" + port);
+ assertThat(requests.get(1).header("X-Token")).isNull();
+ }
+
+ @Test
+ void refusesRedirectToRejectedHost() {
+ var target = "http://" + REJECTED_HOST + ":" + port + "/job.jar";
+ server.createContext("/start", exchange -> redirect(exchange, target));
+
+ assertThatThrownBy(() -> fetch("http://artifact.invalid:" + port + "/start", Map.of()))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Refusing to fetch artifact from '" + target + "'");
+ assertThat(requests).hasSize(1);
+ }
+
+ @Test
+ void configuredHeadersCannotOverrideHost() throws Exception {
+ fetch(
+ "http://artifact.invalid:" + port + "/job.jar",
+ Map.of("Host", "other.invalid", "X-Token", "secret"));
+
+ assertThat(requests).hasSize(1);
+ assertThat(requests.get(0).header("Host")).isEqualTo("artifact.invalid:" + port);
+ assertThat(requests.get(0).header("X-Token")).isEqualTo("secret");
+ }
+
+ @Test
+ void verifiesCertificateAndSendsSniForOriginalHostname() throws Exception {
+ var httpsPort = startHttpsServer().getAddress().getPort();
+
+ var file =
+ fetchTrustingTestCertificate("https://artifact.invalid:" + httpsPort + "/job.jar");
+
+ assertThat(file).hasBinaryContent(ARTIFACT);
+ assertThat(requests).hasSize(1);
+ assertThat(requests.get(0).header("Host")).isEqualTo("artifact.invalid:" + httpsPort);
+ assertThat(requests.get(0).sniHostnames()).containsExactly("artifact.invalid");
+ }
+
+ @Test
+ void rejectsCertificateIssuedForAnotherHostname() throws Exception {
+ var httpsPort = startHttpsServer().getAddress().getPort();
+
+ assertThatThrownBy(
+ () ->
+ fetchTrustingTestCertificate(
+ "https://other.invalid:" + httpsPort + "/job.jar"))
+ .isInstanceOf(SSLHandshakeException.class)
+ .hasMessageContaining("other.invalid");
+ assertThat(requests).isEmpty();
+ }
+
+ @Test
+ void doesNotPinWhenFetchingThroughProxy() throws Exception {
+ var proxyRequests = new CopyOnWriteArrayList<RecordedRequest>();
+ var proxyServer = start(HttpServer.create(new InetSocketAddress(loopback(), 0), 0));
+ proxyServer.createContext("/job.jar", serveArtifact(proxyRequests));
+ var proxy = new Proxy(Proxy.Type.HTTP, proxyServer.getAddress());
+
+ withProxySelector(
+ proxy, () -> fetch("http://artifact.invalid:" + port + "/job.jar", Map.of()));
+
+ assertThat(requests).isEmpty();
+ assertThat(proxyRequests).hasSize(1);
+ assertThat(proxyRequests.get(0).uri())
+ .hasToString("http://artifact.invalid:" + port + "/job.jar");
+ }
+
+ @Test
+ void doesNotPinHttpsWhenFetchingThroughProxy() throws Exception {
+ try (var proxySocket = new ServerSocket(0, 1, loopback())) {
+ var proxyPort = proxySocket.getLocalPort();
+ var firstLine = CompletableFuture.supplyAsync(() -> refuseTunnel(proxySocket));
+ var proxy = new Proxy(Proxy.Type.HTTP, new InetSocketAddress(loopback(), proxyPort));
+
+ assertThatThrownBy(
+ () ->
+ withProxySelector(
+ proxy,
+ () ->
+ fetch(
+ "https://artifact.invalid:"
+ + proxyPort
+ + "/job.jar",
+ Map.of())))
+ .isInstanceOf(IOException.class);
+ assertThat(firstLine.get(10, TimeUnit.SECONDS))
+ .isEqualTo("CONNECT artifact.invalid:" + proxyPort + " HTTP/1.1");
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"http://192.0.2.1/job.jar", "https://192.0.2.1/job.jar"})
+ void pinnedConnectionsHonourConnectTimeout(String uri) {
+ var operatorConfig = new Configuration();
+ operatorConfig.set(
+ KubernetesOperatorConfigOptions.JAR_URI_ALLOWED_SCHEMES, List.of("http", "https"));
+ operatorConfig.set(
+ KubernetesOperatorConfigOptions.JAR_FETCH_SOCKET_TIMEOUT, Duration.ofSeconds(1));
+ var fetchStart = System.currentTimeMillis();
+
+ assertThatThrownBy(
+ () ->
+ HttpArtifactFetcher.INSTANCE.fetch(
+ uri,
+ new Configuration(),
+ FlinkOperatorConfiguration.fromConfiguration(
+ operatorConfig),
+ tempDir.toFile()))
+ .isInstanceOf(IOException.class)
+ .satisfies(
+ e -> assertThat(e.getMessage()).containsIgnoringCase("connect timed out"));
Review Comment:
192.0.2.1 is a documentation address, not a guaranteed TCP black hole. An IPv6-only environment or a network that actively rejects this traffic can immediately return 'Network is unreachable' or 'Connection refused', failing this assertion even when the connect timeout is correctly applied. Please replace the external-network dependency with a controllable socket/connector fixture that verifies the timeout deterministically.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40448][operator] Pin resolved address during operator artifact fetch to prevent DNS rebinding [flink-kubernetes-operator]
Posted by "Copilot (via GitHub) via issues" <is...@flink.apache.org>.
Copilot commented on code in PR #1188:
URL: https://github.com/apache/flink-kubernetes-operator/pull/1188#discussion_r4535637439
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/artifact/HttpArtifactFetcher.java:
##########
@@ -264,4 +295,123 @@ private static boolean isRedirect(int status) {
|| status == 307 // Temporary Redirect
|| status == 308; // Permanent Redirect
}
+
+ @VisibleForTesting
+ static Optional<InetAddress> validateHop(
+ String uri, Collection<String> allowedSchemes, boolean disallowRestrictedHosts)
+ throws IOException {
+ var validation =
+ JarUriValidationUtils.validateAndResolve(
+ uri, allowedSchemes, disallowRestrictedHosts);
+ if (validation.getError().isPresent()) {
+ throw new IOException(
+ "Refusing to fetch artifact from '"
+ + uri
+ + "': "
+ + validation.getError().get());
+ }
+ return validation.getCheckedAddress();
+ }
+
+ private static Proxy selectProxy(String uri) {
+ var selector = ProxySelector.getDefault();
+ var proxies = selector == null ? List.<Proxy>of() : selector.select(URI.create(uri));
+ return proxies.isEmpty() ? Proxy.NO_PROXY : proxies.get(0);
+ }
+
+ private static HttpURLConnection openPinnedConnection(URL url, InetAddress checkedAddress)
+ throws IOException {
+ // An address from getAllByName() keeps the hostname, which the JDK would resolve again
+ // wherever it connects by host string, so connect by the raw IP only.
+ var address = InetAddress.getByAddress(checkedAddress.getAddress());
+ if ("https".equalsIgnoreCase(url.getProtocol())) {
+ // The URL keeps the hostname, so Host, SNI and the certificate check still use it; only
+ // the TCP socket goes to the checked address.
+ var conn = (HttpsURLConnection) url.openConnection(Proxy.NO_PROXY);
+ conn.setSSLSocketFactory(
+ new PinnedSSLSocketFactory(conn.getSSLSocketFactory(), address));
+ return conn;
+ }
+ // Plain HTTP has no socket hook, so the request goes to the address as if it were a proxy:
+ // in absolute form, which servers must accept (RFC 9112, section 3.2.2), with Host still
+ // taken from the URL.
+ var port = url.getPort() == -1 ? url.getDefaultPort() : url.getPort();
+ return (HttpURLConnection)
+ url.openConnection(
+ new Proxy(Proxy.Type.HTTP, new InetSocketAddress(address, port)));
Review Comment:
Treating the artifact origin as a proxy enables proxy authentication even when the proxy selector chose DIRECT. If the JVM's default `Authenticator` supplies proxy credentials, a tenant-controlled artifact host can return a `407` challenge and receive them in `Proxy-Authorization`. Isolate this synthetic-proxy connection from proxy authentication and cached proxy credentials, or use a transport that pins the address without treating the origin as a proxy, while preserving origin authentication and real-proxy behavior.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org