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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
