LuciferYang commented on code in PR #11785: URL: https://github.com/apache/gravitino/pull/11785#discussion_r3468072303
########## common/src/main/java/org/apache/gravitino/utils/RemoteFileDownloader.java: ########## @@ -0,0 +1,377 @@ +/* + * 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.gravitino.utils; + +import java.io.BufferedInputStream; +import java.io.ByteArrayOutputStream; +import java.io.File; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.Socket; +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardCopyOption; +import java.util.HashMap; +import java.util.Locale; +import java.util.Map; +import java.util.Optional; +import javax.net.ssl.SSLParameters; +import javax.net.ssl.SSLSocket; +import javax.net.ssl.SSLSocketFactory; + +/** + * Downloads an {@code http} or {@code https} file while pinning the TCP connection to a + * pre-validated address. + * + * <p>This closes the DNS-rebinding TOCTOU window between SSRF validation and the actual fetch: the + * host is resolved and validated once by {@link RemoteUriValidator#resolveAndValidate}, and the + * resulting {@link InetAddress} is connected to directly here, so the hostname is never + * re-resolved. A minimal HTTP/1.0 client is used so the connection can be pinned at the socket + * level (the JDK exposes no per-connection address override for plain {@code HttpURLConnection}, + * and silently drops a {@code Host} header set via {@code setRequestProperty}). The original + * hostname is still used for the HTTP {@code Host} header and, for {@code https}, for TLS SNI and + * certificate-hostname verification, so virtual hosting and certificate validation keep working. + * Redirects are not followed, because a redirect target would re-resolve to an unvalidated address. + * + * <p>{@code ftp} is intentionally not handled here: the FTP data channel is opened to an address + * the server chooses in its {@code PASV}/{@code EPSV} reply, which cannot be pinned, so {@link + * FileFetcher} rejects {@code ftp} on the SSRF-blocking path instead. + * + * <p>Only responses framed by {@code Content-Length} or chunked transfer-encoding are accepted; a + * connection-close-delimited body is rejected because a premature close cannot be told apart from a + * complete one. Combined with streaming to a sibling temporary file that is atomically moved into + * place only on success, this means a connection that drops mid-transfer never leaves a truncated + * destination behind. Response and header sizes are bounded to prevent a malicious pinned host from + * exhausting disk or memory. + */ +final class RemoteFileDownloader { + + private static final int DEFAULT_HTTP_PORT = 80; + private static final int DEFAULT_HTTPS_PORT = 443; + private static final int MAX_HEADER_BYTES = 64 * 1024; + private static final long MAX_BODY_BYTES = 2L * 1024 * 1024 * 1024; + private static final int COPY_BUFFER_BYTES = 8192; + + private RemoteFileDownloader() {} + + /** + * Downloads {@code uri} into {@code destFile}, connecting only to {@code pinnedAddress}. + * + * @param uri the source URI; must use the {@code http} or {@code https} scheme + * @param pinnedAddress the validated address to connect to + * @param destFile the local destination file + * @param timeoutMs the connect/read timeout in milliseconds + * @throws IOException if the file cannot be fetched or the response is not successful + */ + static void download(URI uri, InetAddress pinnedAddress, File destFile, int timeoutMs) + throws IOException { + String scheme = Optional.ofNullable(uri.getScheme()).orElse("").toLowerCase(Locale.ROOT); + if (!scheme.equals("http") && !scheme.equals("https")) { + throw new IllegalArgumentException( + String.format("Pinned download does not support scheme '%s'", scheme)); + } + + Path destPath = destFile.toPath().toAbsolutePath(); + Path tempPath = + Files.createTempFile(destPath.getParent(), destPath.getFileName() + ".", ".tmp"); + try { + httpDownload(uri, pinnedAddress, tempPath, timeoutMs, scheme.equals("https")); + Files.move(tempPath, destPath, StandardCopyOption.REPLACE_EXISTING); + } catch (Throwable e) { + // Catch every throwable (incl. Error/OutOfMemoryError) so the temp file is never orphaned in + // the staging directory. Precise rethrow keeps the declared `throws IOException`. + try { + Files.deleteIfExists(tempPath); + } catch (IOException suppressed) { + e.addSuppressed(suppressed); + } + throw e; + } + } + + private static void httpDownload( + URI uri, InetAddress pinnedAddress, Path destPath, int timeoutMs, boolean tls) + throws IOException { + String host = uri.getHost(); + int port = uri.getPort() != -1 ? uri.getPort() : (tls ? DEFAULT_HTTPS_PORT : DEFAULT_HTTP_PORT); + + Socket socket = new Socket(); + try { + socket.connect(new InetSocketAddress(pinnedAddress, port), timeoutMs); + socket.setSoTimeout(timeoutMs); + if (tls) { + socket = startTls(socket, host, port); + } + + sendGetRequest(socket.getOutputStream(), requestTarget(uri), hostHeader(host, uri.getPort())); + + InputStream in = new BufferedInputStream(socket.getInputStream()); + int status = readStatusCode(in); + if (status < 200 || status >= 300) { + throw new IOException( + String.format( + "Unexpected HTTP status %d fetching pinned URI '%s'", status, SafeUri.redact(uri))); + } + Map<String, String> headers = readHeaders(in); + try (OutputStream out = Files.newOutputStream(destPath)) { + writeBody(in, headers, out); + } + } finally { + socket.close(); + } + } + + /** + * Layers TLS over an already-connected plain socket. The handshake uses {@code host} (the + * original hostname) for SNI and certificate verification ({@code HTTPS} endpoint + * identification), while the TCP connection stays pinned to the address the plain socket was + * connected to. + */ + private static Socket startTls(Socket plainSocket, String host, int port) throws IOException { + SSLSocketFactory factory = (SSLSocketFactory) SSLSocketFactory.getDefault(); + SSLSocket sslSocket = (SSLSocket) factory.createSocket(plainSocket, host, port, true); + SSLParameters parameters = sslSocket.getSSLParameters(); + parameters.setEndpointIdentificationAlgorithm("HTTPS"); + sslSocket.setSSLParameters(parameters); + sslSocket.startHandshake(); + return sslSocket; + } + + private static void sendGetRequest(OutputStream out, String target, String hostHeader) + throws IOException { + // HTTP/1.0 with "Connection: close" requests a non-persistent connection. The response must be + // framed by Content-Length or chunked encoding; an unframed (close-delimited) body is rejected + // by writeBody because its completeness cannot be verified. + String request = + "GET " + + target + + " HTTP/1.0\r\n" + + "Host: " + + hostHeader + + "\r\n" + + "Connection: close\r\n" + + "\r\n"; + out.write(request.getBytes(StandardCharsets.US_ASCII)); + out.flush(); + } + + private static int readStatusCode(InputStream in) throws IOException { + String statusLine = readLine(in); + if (statusLine == null) { + throw new IOException("Empty HTTP response from pinned host"); + } + // Format: "HTTP/1.x <code> <reason>"; tolerate extra whitespace between tokens. + String[] parts = statusLine.trim().split("\\s+", 3); + if (parts.length < 2) { + throw new IOException("Malformed HTTP status line: " + statusLine); + } + try { + return Integer.parseInt(parts[1]); + } catch (NumberFormatException e) { + throw new IOException("Malformed HTTP status line: " + statusLine, e); + } + } + + private static Map<String, String> readHeaders(InputStream in) throws IOException { + Map<String, String> headers = new HashMap<>(); + int total = 0; + String line; + while ((line = readLine(in)) != null) { + if (line.isEmpty()) { + return headers; + } + // Reject obsolete line folding (RFC 7230 §3.2.4); a leading space/tab continuation is treated + // differently by different parsers and is a smuggling vector. + if (line.charAt(0) == ' ' || line.charAt(0) == '\t') { + throw new IOException("Obsolete line folding in HTTP headers is rejected from pinned host"); + } + total += line.length(); + if (total > MAX_HEADER_BYTES) { + throw new IOException("HTTP response headers exceed the allowed size from pinned host"); + } + int colon = line.indexOf(':'); + if (colon <= 0) { + throw new IOException("Malformed HTTP header line from pinned host: " + line); + } + String name = line.substring(0, colon).trim().toLowerCase(Locale.ROOT); + String value = line.substring(colon + 1).trim(); + headers.put(name, value); + } + throw new IOException("Unexpected end of stream while reading HTTP headers from pinned host"); + } + + private static void writeBody(InputStream in, Map<String, String> headers, OutputStream out) + throws IOException { + String transferEncoding = headers.get("transfer-encoding"); + if (transferEncoding != null && transferEncoding.toLowerCase(Locale.ROOT).contains("chunked")) { Review Comment: Good catch. `writeBody` now requires the `Transfer-Encoding` to be exactly `chunked` (trimmed, case-insensitive); `gzip, chunked`, `xchunked`, etc. are rejected with an `Unsupported Transfer-Encoding` error rather than de-chunked into corrupt bytes. Added tests for both `gzip, chunked` and `xchunked`. (335f6aea3) ########## common/src/main/java/org/apache/gravitino/utils/RemoteFileDownloader.java: ########## @@ -0,0 +1,377 @@ +/* + * 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.gravitino.utils; + +import java.io.BufferedInputStream; +import java.io.ByteArrayOutputStream; +import java.io.File; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.Socket; +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardCopyOption; +import java.util.HashMap; +import java.util.Locale; +import java.util.Map; +import java.util.Optional; +import javax.net.ssl.SSLParameters; +import javax.net.ssl.SSLSocket; +import javax.net.ssl.SSLSocketFactory; + +/** + * Downloads an {@code http} or {@code https} file while pinning the TCP connection to a + * pre-validated address. + * + * <p>This closes the DNS-rebinding TOCTOU window between SSRF validation and the actual fetch: the + * host is resolved and validated once by {@link RemoteUriValidator#resolveAndValidate}, and the + * resulting {@link InetAddress} is connected to directly here, so the hostname is never + * re-resolved. A minimal HTTP/1.0 client is used so the connection can be pinned at the socket + * level (the JDK exposes no per-connection address override for plain {@code HttpURLConnection}, + * and silently drops a {@code Host} header set via {@code setRequestProperty}). The original + * hostname is still used for the HTTP {@code Host} header and, for {@code https}, for TLS SNI and + * certificate-hostname verification, so virtual hosting and certificate validation keep working. + * Redirects are not followed, because a redirect target would re-resolve to an unvalidated address. + * + * <p>{@code ftp} is intentionally not handled here: the FTP data channel is opened to an address + * the server chooses in its {@code PASV}/{@code EPSV} reply, which cannot be pinned, so {@link + * FileFetcher} rejects {@code ftp} on the SSRF-blocking path instead. + * + * <p>Only responses framed by {@code Content-Length} or chunked transfer-encoding are accepted; a + * connection-close-delimited body is rejected because a premature close cannot be told apart from a + * complete one. Combined with streaming to a sibling temporary file that is atomically moved into + * place only on success, this means a connection that drops mid-transfer never leaves a truncated + * destination behind. Response and header sizes are bounded to prevent a malicious pinned host from + * exhausting disk or memory. + */ +final class RemoteFileDownloader { + + private static final int DEFAULT_HTTP_PORT = 80; + private static final int DEFAULT_HTTPS_PORT = 443; + private static final int MAX_HEADER_BYTES = 64 * 1024; + private static final long MAX_BODY_BYTES = 2L * 1024 * 1024 * 1024; + private static final int COPY_BUFFER_BYTES = 8192; + + private RemoteFileDownloader() {} + + /** + * Downloads {@code uri} into {@code destFile}, connecting only to {@code pinnedAddress}. + * + * @param uri the source URI; must use the {@code http} or {@code https} scheme + * @param pinnedAddress the validated address to connect to + * @param destFile the local destination file + * @param timeoutMs the connect/read timeout in milliseconds + * @throws IOException if the file cannot be fetched or the response is not successful + */ + static void download(URI uri, InetAddress pinnedAddress, File destFile, int timeoutMs) + throws IOException { + String scheme = Optional.ofNullable(uri.getScheme()).orElse("").toLowerCase(Locale.ROOT); + if (!scheme.equals("http") && !scheme.equals("https")) { + throw new IllegalArgumentException( + String.format("Pinned download does not support scheme '%s'", scheme)); + } + + Path destPath = destFile.toPath().toAbsolutePath(); + Path tempPath = + Files.createTempFile(destPath.getParent(), destPath.getFileName() + ".", ".tmp"); + try { + httpDownload(uri, pinnedAddress, tempPath, timeoutMs, scheme.equals("https")); + Files.move(tempPath, destPath, StandardCopyOption.REPLACE_EXISTING); + } catch (Throwable e) { + // Catch every throwable (incl. Error/OutOfMemoryError) so the temp file is never orphaned in + // the staging directory. Precise rethrow keeps the declared `throws IOException`. + try { + Files.deleteIfExists(tempPath); + } catch (IOException suppressed) { + e.addSuppressed(suppressed); + } + throw e; + } + } + + private static void httpDownload( + URI uri, InetAddress pinnedAddress, Path destPath, int timeoutMs, boolean tls) + throws IOException { + String host = uri.getHost(); + int port = uri.getPort() != -1 ? uri.getPort() : (tls ? DEFAULT_HTTPS_PORT : DEFAULT_HTTP_PORT); + + Socket socket = new Socket(); + try { + socket.connect(new InetSocketAddress(pinnedAddress, port), timeoutMs); + socket.setSoTimeout(timeoutMs); Review Comment: You're right that the per-read `soTimeout` doesn't bound total wall-clock — on the chunked path a server dripping one byte just inside each window can stretch a transfer toward `MAX_BODY_BYTES`. Two notes, then a proposal: - The fetch is only triggered when an operator configures a catalog with a keytab/jar URI (a privileged action), so the blast radius is a single stuck worker thread per misconfiguration rather than an unauthenticated DoS — which is why it wasn't already a hard default. - The API currently carries only a per-operation timeout, so an overall budget is a new constant that has to stay generous enough not to abort a legitimately large/slow jar fetch. That said, I'm happy to add it as cheap insurance: an overall deadline checked in the copy loops, either a generous fixed constant or wired to config so it's tunable. Do you have a preference on the value / whether it should be configurable? I'll add it in this PR. ########## common/src/main/java/org/apache/gravitino/utils/RemoteFileDownloader.java: ########## @@ -0,0 +1,377 @@ +/* + * 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.gravitino.utils; + +import java.io.BufferedInputStream; +import java.io.ByteArrayOutputStream; +import java.io.File; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.Socket; +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardCopyOption; +import java.util.HashMap; +import java.util.Locale; +import java.util.Map; +import java.util.Optional; +import javax.net.ssl.SSLParameters; +import javax.net.ssl.SSLSocket; +import javax.net.ssl.SSLSocketFactory; + +/** + * Downloads an {@code http} or {@code https} file while pinning the TCP connection to a + * pre-validated address. + * + * <p>This closes the DNS-rebinding TOCTOU window between SSRF validation and the actual fetch: the + * host is resolved and validated once by {@link RemoteUriValidator#resolveAndValidate}, and the + * resulting {@link InetAddress} is connected to directly here, so the hostname is never + * re-resolved. A minimal HTTP/1.0 client is used so the connection can be pinned at the socket + * level (the JDK exposes no per-connection address override for plain {@code HttpURLConnection}, + * and silently drops a {@code Host} header set via {@code setRequestProperty}). The original + * hostname is still used for the HTTP {@code Host} header and, for {@code https}, for TLS SNI and + * certificate-hostname verification, so virtual hosting and certificate validation keep working. + * Redirects are not followed, because a redirect target would re-resolve to an unvalidated address. + * + * <p>{@code ftp} is intentionally not handled here: the FTP data channel is opened to an address + * the server chooses in its {@code PASV}/{@code EPSV} reply, which cannot be pinned, so {@link + * FileFetcher} rejects {@code ftp} on the SSRF-blocking path instead. + * + * <p>Only responses framed by {@code Content-Length} or chunked transfer-encoding are accepted; a + * connection-close-delimited body is rejected because a premature close cannot be told apart from a + * complete one. Combined with streaming to a sibling temporary file that is atomically moved into + * place only on success, this means a connection that drops mid-transfer never leaves a truncated + * destination behind. Response and header sizes are bounded to prevent a malicious pinned host from + * exhausting disk or memory. + */ +final class RemoteFileDownloader { + + private static final int DEFAULT_HTTP_PORT = 80; + private static final int DEFAULT_HTTPS_PORT = 443; + private static final int MAX_HEADER_BYTES = 64 * 1024; + private static final long MAX_BODY_BYTES = 2L * 1024 * 1024 * 1024; + private static final int COPY_BUFFER_BYTES = 8192; + + private RemoteFileDownloader() {} + + /** + * Downloads {@code uri} into {@code destFile}, connecting only to {@code pinnedAddress}. + * + * @param uri the source URI; must use the {@code http} or {@code https} scheme + * @param pinnedAddress the validated address to connect to + * @param destFile the local destination file + * @param timeoutMs the connect/read timeout in milliseconds + * @throws IOException if the file cannot be fetched or the response is not successful + */ + static void download(URI uri, InetAddress pinnedAddress, File destFile, int timeoutMs) + throws IOException { + String scheme = Optional.ofNullable(uri.getScheme()).orElse("").toLowerCase(Locale.ROOT); + if (!scheme.equals("http") && !scheme.equals("https")) { + throw new IllegalArgumentException( + String.format("Pinned download does not support scheme '%s'", scheme)); + } + + Path destPath = destFile.toPath().toAbsolutePath(); + Path tempPath = + Files.createTempFile(destPath.getParent(), destPath.getFileName() + ".", ".tmp"); + try { + httpDownload(uri, pinnedAddress, tempPath, timeoutMs, scheme.equals("https")); + Files.move(tempPath, destPath, StandardCopyOption.REPLACE_EXISTING); + } catch (Throwable e) { + // Catch every throwable (incl. Error/OutOfMemoryError) so the temp file is never orphaned in + // the staging directory. Precise rethrow keeps the declared `throws IOException`. + try { + Files.deleteIfExists(tempPath); + } catch (IOException suppressed) { + e.addSuppressed(suppressed); + } + throw e; + } + } + + private static void httpDownload( + URI uri, InetAddress pinnedAddress, Path destPath, int timeoutMs, boolean tls) + throws IOException { + String host = uri.getHost(); + int port = uri.getPort() != -1 ? uri.getPort() : (tls ? DEFAULT_HTTPS_PORT : DEFAULT_HTTP_PORT); + + Socket socket = new Socket(); + try { + socket.connect(new InetSocketAddress(pinnedAddress, port), timeoutMs); + socket.setSoTimeout(timeoutMs); + if (tls) { + socket = startTls(socket, host, port); + } + + sendGetRequest(socket.getOutputStream(), requestTarget(uri), hostHeader(host, uri.getPort())); + + InputStream in = new BufferedInputStream(socket.getInputStream()); + int status = readStatusCode(in); + if (status < 200 || status >= 300) { + throw new IOException( + String.format( + "Unexpected HTTP status %d fetching pinned URI '%s'", status, SafeUri.redact(uri))); + } + Map<String, String> headers = readHeaders(in); + try (OutputStream out = Files.newOutputStream(destPath)) { + writeBody(in, headers, out); + } + } finally { + socket.close(); + } + } + + /** + * Layers TLS over an already-connected plain socket. The handshake uses {@code host} (the + * original hostname) for SNI and certificate verification ({@code HTTPS} endpoint + * identification), while the TCP connection stays pinned to the address the plain socket was + * connected to. + */ + private static Socket startTls(Socket plainSocket, String host, int port) throws IOException { + SSLSocketFactory factory = (SSLSocketFactory) SSLSocketFactory.getDefault(); + SSLSocket sslSocket = (SSLSocket) factory.createSocket(plainSocket, host, port, true); + SSLParameters parameters = sslSocket.getSSLParameters(); + parameters.setEndpointIdentificationAlgorithm("HTTPS"); + sslSocket.setSSLParameters(parameters); + sslSocket.startHandshake(); + return sslSocket; + } + + private static void sendGetRequest(OutputStream out, String target, String hostHeader) + throws IOException { + // HTTP/1.0 with "Connection: close" requests a non-persistent connection. The response must be + // framed by Content-Length or chunked encoding; an unframed (close-delimited) body is rejected + // by writeBody because its completeness cannot be verified. + String request = Review Comment: Switched the request to HTTP/1.1 (still `Connection: close`) to bias origins toward `Content-Length`/chunked framing. Close-delimited bodies are still rejected for truncation safety (intentional — documented in the class Javadoc). Added a test asserting the request line is HTTP/1.1. (335f6aea3) ########## common/src/main/java/org/apache/gravitino/utils/RemoteFileDownloader.java: ########## @@ -0,0 +1,377 @@ +/* + * 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.gravitino.utils; + +import java.io.BufferedInputStream; +import java.io.ByteArrayOutputStream; +import java.io.File; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.Socket; +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardCopyOption; +import java.util.HashMap; +import java.util.Locale; +import java.util.Map; +import java.util.Optional; +import javax.net.ssl.SSLParameters; +import javax.net.ssl.SSLSocket; +import javax.net.ssl.SSLSocketFactory; + +/** + * Downloads an {@code http} or {@code https} file while pinning the TCP connection to a + * pre-validated address. + * + * <p>This closes the DNS-rebinding TOCTOU window between SSRF validation and the actual fetch: the + * host is resolved and validated once by {@link RemoteUriValidator#resolveAndValidate}, and the + * resulting {@link InetAddress} is connected to directly here, so the hostname is never + * re-resolved. A minimal HTTP/1.0 client is used so the connection can be pinned at the socket + * level (the JDK exposes no per-connection address override for plain {@code HttpURLConnection}, + * and silently drops a {@code Host} header set via {@code setRequestProperty}). The original + * hostname is still used for the HTTP {@code Host} header and, for {@code https}, for TLS SNI and + * certificate-hostname verification, so virtual hosting and certificate validation keep working. + * Redirects are not followed, because a redirect target would re-resolve to an unvalidated address. + * + * <p>{@code ftp} is intentionally not handled here: the FTP data channel is opened to an address + * the server chooses in its {@code PASV}/{@code EPSV} reply, which cannot be pinned, so {@link + * FileFetcher} rejects {@code ftp} on the SSRF-blocking path instead. + * + * <p>Only responses framed by {@code Content-Length} or chunked transfer-encoding are accepted; a + * connection-close-delimited body is rejected because a premature close cannot be told apart from a + * complete one. Combined with streaming to a sibling temporary file that is atomically moved into + * place only on success, this means a connection that drops mid-transfer never leaves a truncated + * destination behind. Response and header sizes are bounded to prevent a malicious pinned host from + * exhausting disk or memory. + */ +final class RemoteFileDownloader { + + private static final int DEFAULT_HTTP_PORT = 80; + private static final int DEFAULT_HTTPS_PORT = 443; + private static final int MAX_HEADER_BYTES = 64 * 1024; + private static final long MAX_BODY_BYTES = 2L * 1024 * 1024 * 1024; + private static final int COPY_BUFFER_BYTES = 8192; + + private RemoteFileDownloader() {} + + /** + * Downloads {@code uri} into {@code destFile}, connecting only to {@code pinnedAddress}. + * + * @param uri the source URI; must use the {@code http} or {@code https} scheme + * @param pinnedAddress the validated address to connect to + * @param destFile the local destination file + * @param timeoutMs the connect/read timeout in milliseconds + * @throws IOException if the file cannot be fetched or the response is not successful + */ + static void download(URI uri, InetAddress pinnedAddress, File destFile, int timeoutMs) + throws IOException { + String scheme = Optional.ofNullable(uri.getScheme()).orElse("").toLowerCase(Locale.ROOT); + if (!scheme.equals("http") && !scheme.equals("https")) { + throw new IllegalArgumentException( + String.format("Pinned download does not support scheme '%s'", scheme)); + } + + Path destPath = destFile.toPath().toAbsolutePath(); + Path tempPath = + Files.createTempFile(destPath.getParent(), destPath.getFileName() + ".", ".tmp"); + try { + httpDownload(uri, pinnedAddress, tempPath, timeoutMs, scheme.equals("https")); + Files.move(tempPath, destPath, StandardCopyOption.REPLACE_EXISTING); + } catch (Throwable e) { + // Catch every throwable (incl. Error/OutOfMemoryError) so the temp file is never orphaned in + // the staging directory. Precise rethrow keeps the declared `throws IOException`. + try { + Files.deleteIfExists(tempPath); + } catch (IOException suppressed) { + e.addSuppressed(suppressed); + } + throw e; + } + } + + private static void httpDownload( + URI uri, InetAddress pinnedAddress, Path destPath, int timeoutMs, boolean tls) + throws IOException { + String host = uri.getHost(); + int port = uri.getPort() != -1 ? uri.getPort() : (tls ? DEFAULT_HTTPS_PORT : DEFAULT_HTTP_PORT); + + Socket socket = new Socket(); + try { + socket.connect(new InetSocketAddress(pinnedAddress, port), timeoutMs); + socket.setSoTimeout(timeoutMs); + if (tls) { + socket = startTls(socket, host, port); + } + + sendGetRequest(socket.getOutputStream(), requestTarget(uri), hostHeader(host, uri.getPort())); + + InputStream in = new BufferedInputStream(socket.getInputStream()); + int status = readStatusCode(in); + if (status < 200 || status >= 300) { + throw new IOException( + String.format( + "Unexpected HTTP status %d fetching pinned URI '%s'", status, SafeUri.redact(uri))); + } + Map<String, String> headers = readHeaders(in); + try (OutputStream out = Files.newOutputStream(destPath)) { + writeBody(in, headers, out); + } + } finally { + socket.close(); + } + } + + /** + * Layers TLS over an already-connected plain socket. The handshake uses {@code host} (the + * original hostname) for SNI and certificate verification ({@code HTTPS} endpoint + * identification), while the TCP connection stays pinned to the address the plain socket was + * connected to. + */ + private static Socket startTls(Socket plainSocket, String host, int port) throws IOException { + SSLSocketFactory factory = (SSLSocketFactory) SSLSocketFactory.getDefault(); + SSLSocket sslSocket = (SSLSocket) factory.createSocket(plainSocket, host, port, true); + SSLParameters parameters = sslSocket.getSSLParameters(); + parameters.setEndpointIdentificationAlgorithm("HTTPS"); + sslSocket.setSSLParameters(parameters); + sslSocket.startHandshake(); + return sslSocket; + } + + private static void sendGetRequest(OutputStream out, String target, String hostHeader) + throws IOException { + // HTTP/1.0 with "Connection: close" requests a non-persistent connection. The response must be + // framed by Content-Length or chunked encoding; an unframed (close-delimited) body is rejected + // by writeBody because its completeness cannot be verified. + String request = + "GET " + + target + + " HTTP/1.0\r\n" + + "Host: " + + hostHeader + + "\r\n" + + "Connection: close\r\n" + + "\r\n"; + out.write(request.getBytes(StandardCharsets.US_ASCII)); + out.flush(); + } + + private static int readStatusCode(InputStream in) throws IOException { + String statusLine = readLine(in); + if (statusLine == null) { + throw new IOException("Empty HTTP response from pinned host"); + } + // Format: "HTTP/1.x <code> <reason>"; tolerate extra whitespace between tokens. + String[] parts = statusLine.trim().split("\\s+", 3); + if (parts.length < 2) { + throw new IOException("Malformed HTTP status line: " + statusLine); + } + try { + return Integer.parseInt(parts[1]); + } catch (NumberFormatException e) { + throw new IOException("Malformed HTTP status line: " + statusLine, e); + } + } + + private static Map<String, String> readHeaders(InputStream in) throws IOException { + Map<String, String> headers = new HashMap<>(); Review Comment: Fixed: conflicting duplicate `Content-Length`/`Transfer-Encoding` headers are now rejected (RFC 7230 §3.3.2/§3.3.3) instead of silently last-winning; identical duplicates are tolerated. Added tests. (335f6aea3) ########## common/src/main/java/org/apache/gravitino/utils/RemoteUriValidator.java: ########## @@ -71,19 +74,119 @@ private static boolean isUnsafeAddress(InetAddress address) { } byte[] bytes = address.getAddress(); - if (isCloudMetadataAddress(bytes)) { + if (isUnsafeIpv4Address(bytes)) { return true; } + if (bytes.length == 16) { + if (isIpv6UniqueLocalAddress(bytes)) { + return true; + } + // Several IPv6 forms embed an IPv4 address the platform checks above miss (IPv4-compatible, + // NAT64, 6to4, ISATAP; plus IPv4-mapped defensively). Re-classify the embedded IPv4 so a + // blocked address such as cloud metadata cannot be reached via an IPv6 literal. + byte[] embeddedIpv4 = embeddedIpv4(bytes); + if (embeddedIpv4 != null) { + try { + return isUnsafeAddress(InetAddress.getByAddress(embeddedIpv4)); + } catch (UnknownHostException e) { + // getByAddress only rejects a wrong-length array; a 4-byte array never reaches here. Fail + // closed if it somehow does. + return true; + } + } + } + return false; + } - return isIpv6UniqueLocalAddress(bytes); + /** + * Returns the IPv4 address embedded in a 16-byte IPv6 address whose embedded IPv4 is the literal + * connection target, or {@code null} if there is none. Covers the IPv4-compatible ({@code + * ::a.b.c.d}), NAT64 ({@code 64:ff9b::a.b.c.d}, RFC 6052), 6to4 ({@code 2002:a.b.c.d::}, RFC + * 3056) and ISATAP ({@code ::0:5efe:a.b.c.d}, RFC 5214) forms. The IPv4-mapped ({@code + * ::ffff:a.b.c.d}) form is handled defensively: the JDK normally surfaces it as a 4-byte {@link + * java.net.Inet4Address} validated by {@link #isUnsafeIpv4Address}, so this branch is a fallback. + * + * <p>Teredo ({@code 2001:0::/32}) is intentionally excluded: its embedded IPv4 is the + * client/relay identifier rather than the connection destination, and it requires a non-default + * Teredo tunnel to route at all. + */ + private static byte[] embeddedIpv4(byte[] bytes) { + boolean highTenZero = true; + for (int i = 0; i < 10; i++) { + if (bytes[i] != 0) { + highTenZero = false; + break; + } + } + if (highTenZero) { + boolean compatible = bytes[10] == 0 && bytes[11] == 0; + // Defensive: the JDK normally collapses IPv4-mapped addresses to a 4-byte Inet4Address. + boolean mapped = (bytes[10] & 0xFF) == 0xFF && (bytes[11] & 0xFF) == 0xFF; + if (compatible || mapped) { + return lowestFourBytes(bytes); + } + } + // NAT64 well-known prefix 64:ff9b::/96 (RFC 6052): the IPv4 is the low 32 bits. Review Comment: Added a comment noting the limitation: only the well-known `64:ff9b::/96` prefix is matched; network-specific prefixes (RFC 6052 §2.2, RFC 8215 `64:ff9b:1::/48`) are site-defined and routable only where deployed, so detecting them would need configuration. (335f6aea3) -- 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]
