diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/HttpClient.java b/paimon-api/src/main/java/org/apache/paimon/rest/HttpClient.java index 2477ff9db69b..311a1a2d4abe 100644 --- a/paimon-api/src/main/java/org/apache/paimon/rest/HttpClient.java +++ b/paimon-api/src/main/java/org/apache/paimon/rest/HttpClient.java @@ -24,6 +24,7 @@ import org.apache.paimon.rest.exceptions.RESTException; import org.apache.paimon.rest.interceptor.LoggingInterceptor; import org.apache.paimon.rest.responses.ErrorResponse; +import org.apache.paimon.utils.SensitiveConfigUtils; import org.apache.paimon.utils.StringUtils; import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.JsonProcessingException; @@ -63,7 +64,7 @@ public HttpClient(String uri) { public T get( String path, Class responseType, RESTAuthFunction restAuthFunction) { Header[] authHeaders = getHeaders(path, "GET", "", restAuthFunction); - HttpGet httpGet = new HttpGet(getRequestUrl(path, null)); + HttpGet httpGet = HttpClientUtils.newHttpGet(getRequestUrl(path, null)); httpGet.setHeaders(authHeaders); return exec(httpGet, responseType); } @@ -75,7 +76,7 @@ public T get( Class responseType, RESTAuthFunction restAuthFunction) { Header[] authHeaders = getHeaders(path, queryParams, "GET", "", restAuthFunction); - HttpGet httpGet = new HttpGet(getRequestUrl(path, queryParams)); + HttpGet httpGet = HttpClientUtils.newHttpGet(getRequestUrl(path, queryParams)); httpGet.setHeaders(authHeaders); return exec(httpGet, responseType); } @@ -92,7 +93,7 @@ public T post( RESTRequest body, Class responseType, RESTAuthFunction restAuthFunction) { - HttpPost httpPost = new HttpPost(getRequestUrl(path, null)); + HttpPost httpPost = HttpClientUtils.newHttpPost(getRequestUrl(path, null)); String encodedBody = RESTUtil.encodedBody(body); if (encodedBody != null) { httpPost.setEntity(new StringEntity(encodedBody, ContentType.APPLICATION_JSON)); @@ -110,7 +111,7 @@ public T delete(String path, RESTAuthFunction restAuthF @Override public T delete( String path, RESTRequest body, RESTAuthFunction restAuthFunction) { - HttpDelete httpDelete = new HttpDelete(getRequestUrl(path, null)); + HttpDelete httpDelete = HttpClientUtils.newHttpDelete(getRequestUrl(path, null)); String encodedBody = RESTUtil.encodedBody(body); if (encodedBody != null) { httpDelete.setEntity(new StringEntity(encodedBody, ContentType.APPLICATION_JSON)); @@ -138,12 +139,21 @@ private T exec(HttpUriRequestBase request, Class res } catch (JsonProcessingException e) { // ignore exception } - error = buildErrorResponse(error, responseBodyStr, response.getCode()); + error = buildErrorResponse(error, response.getCode()); errorHandler.accept(error, extractRequestId(response)); } if (responseType != null && responseBodyStr != null) { - return RESTApi.fromJson(responseBodyStr, responseType); + try { + return RESTApi.fromJson(responseBodyStr, responseType); + } catch (JsonProcessingException e) { + // Do not surface the body or Jackson's source snippet: a + // successful response (e.g. a token response) may contain + // credentials. Report only the target type. + throw new RESTException( + "Failed to parse REST response into %s", + responseType.getName()); + } } else if (responseType == null) { return null; } else { @@ -151,8 +161,9 @@ private T exec(HttpUriRequestBase request, Class res } }); } catch (IOException e) { + // No cause: a redirect/protocol error message can echo the target URL (a signed URL). throw new RESTException( - e, "Error occurred while processing %s request", request.getMethod()); + "Error occurred while processing %s request", request.getMethod()); } } @@ -225,21 +236,23 @@ private static Header[] getHeaders( .toArray(Header[]::new); } - private static ErrorResponse buildErrorResponse( - ErrorResponse error, String responseBodyStr, int errorCode) { - if (error == null || error.getMessage() == null || error.getMessage().isEmpty()) { - String resourceType = - (error != null && error.getResourceType() != null) - ? error.getResourceType() - : ""; - String resourceName = - (error != null && error.getResourceName() != null) - ? error.getResourceName() - : ""; - String message = responseBodyStr != null ? responseBodyStr : "response body is null"; - int code = (error != null && error.getCode() != null) ? error.getCode() : errorCode; - error = new ErrorResponse(resourceType, resourceName, message, code); + private static ErrorResponse buildErrorResponse(ErrorResponse error, int errorCode) { + String resourceType = + (error != null && error.getResourceType() != null) ? error.getResourceType() : ""; + String resourceName = + (error != null && error.getResourceName() != null) ? error.getResourceName() : ""; + int code = (error != null && error.getCode() != null) ? error.getCode() : errorCode; + String message; + if (error != null && error.getMessage() != null && !error.getMessage().isEmpty()) { + // A parsed message may still embed secrets (e.g. "password=..."); redact it. + message = SensitiveConfigUtils.redactText(error.getMessage()); + } else if (error == null) { + // The body could not be parsed; it is arbitrary and is not echoed (may hold secrets). + message = "Unparseable error response body (HTTP " + code + ")."; + } else { + // Parsed as an ErrorResponse but with no message. + message = "Empty error message (HTTP " + code + ")."; } - return error; + return new ErrorResponse(resourceType, resourceName, message, code); } } diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/HttpClientUtils.java b/paimon-api/src/main/java/org/apache/paimon/rest/HttpClientUtils.java index f54ef7044db0..e4334e60e59d 100644 --- a/paimon-api/src/main/java/org/apache/paimon/rest/HttpClientUtils.java +++ b/paimon-api/src/main/java/org/apache/paimon/rest/HttpClientUtils.java @@ -20,9 +20,12 @@ import org.apache.paimon.rest.interceptor.LoggingInterceptor; import org.apache.paimon.rest.interceptor.TimingInterceptor; +import org.apache.paimon.utils.SensitiveConfigUtils; +import org.apache.hc.client5.http.classic.methods.HttpDelete; import org.apache.hc.client5.http.classic.methods.HttpGet; import org.apache.hc.client5.http.classic.methods.HttpHead; +import org.apache.hc.client5.http.classic.methods.HttpPost; import org.apache.hc.client5.http.config.RequestConfig; import org.apache.hc.client5.http.impl.classic.CloseableHttpClient; import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse; @@ -32,6 +35,7 @@ import org.apache.hc.client5.http.io.HttpClientConnectionManager; import org.apache.hc.client5.http.ssl.DefaultClientTlsStrategy; import org.apache.hc.client5.http.ssl.HttpsSupport; +import org.apache.hc.core5.http.ClassicHttpRequest; import org.apache.hc.core5.http.HttpStatus; import org.apache.hc.core5.reactor.ssl.SSLBufferMode; import org.apache.hc.core5.ssl.SSLContexts; @@ -39,6 +43,7 @@ import java.io.IOException; import java.io.InputStream; +import java.util.function.Function; /** Utils for {@link HttpClientBuilder}. */ public class HttpClientUtils { @@ -86,8 +91,8 @@ private static HttpClientConnectionManager configureConnectionManager() { } public static InputStream getAsInputStream(String uri) throws IOException { - HttpGet httpGet = new HttpGet(uri); - CloseableHttpResponse response = DEFAULT_HTTP_CLIENT.execute(httpGet); + HttpGet httpGet = newHttpGet(uri); + CloseableHttpResponse response = execute(httpGet, uri); int statusCode = response.getCode(); if (statusCode != HttpStatus.SC_OK) { try { @@ -120,7 +125,10 @@ public static boolean exists(String uri) throws IOException { return false; } throw new IOException( - "Unexpected HTTP status code: " + rangeStatusCode + " for uri: " + uri); + "Unexpected HTTP status code: " + + rangeStatusCode + + " for uri: " + + SensitiveConfigUtils.sanitizeUri(uri)); } public static boolean isNotFoundError(Throwable throwable) { @@ -136,7 +144,9 @@ public static boolean isInvalidUriException(Throwable throwable) { } if (current instanceof IllegalArgumentException && current.getMessage() != null - && current.getMessage().contains("Illegal character")) { + && (current.getMessage().contains("Illegal character") + || current.getMessage() + .startsWith(SensitiveConfigUtils.INVALID_URI_MESSAGE_PREFIX))) { return true; } current = current.getCause(); @@ -182,20 +192,61 @@ private static Integer parseStatusCodeSuffix(String statusText) { } private static int headStatusCode(String uri) throws IOException { - HttpHead httpHead = new HttpHead(uri); - try (CloseableHttpResponse response = DEFAULT_HTTP_CLIENT.execute(httpHead)) { + HttpHead httpHead = newHttpHead(uri); + try (CloseableHttpResponse response = execute(httpHead, uri)) { return response.getCode(); } } private static int getRangeStatusCode(String uri) throws IOException { - HttpGet httpGet = new HttpGet(uri); + HttpGet httpGet = newHttpGet(uri); httpGet.addHeader("Range", "bytes=0-0"); - try (CloseableHttpResponse response = DEFAULT_HTTP_CLIENT.execute(httpGet)) { + try (CloseableHttpResponse response = execute(httpGet, uri)) { return response.getCode(); } } + /** + * Executes a request, converting any execute-stage failure into an exception that carries + * neither the original message nor cause. Redirect and protocol errors (e.g. "Circular redirect + * to <Location>") echo the target URL, which for a signed URL is a credential; only the + * sanitized request URI is reported. + */ + private static CloseableHttpResponse execute(ClassicHttpRequest request, String uri) + throws IOException { + try { + return DEFAULT_HTTP_CLIENT.execute(request); + } catch (IOException | RuntimeException e) { + throw new IOException( + "HTTP request failed for uri: " + SensitiveConfigUtils.sanitizeUri(uri)); + } + } + + public static HttpGet newHttpGet(String uri) { + return newRequest(uri, HttpGet::new); + } + + public static HttpHead newHttpHead(String uri) { + return newRequest(uri, HttpHead::new); + } + + public static HttpPost newHttpPost(String uri) { + return newRequest(uri, HttpPost::new); + } + + public static HttpDelete newHttpDelete(String uri) { + return newRequest(uri, HttpDelete::new); + } + + /** A malformed URL leaks the raw URL from the constructor; sanitize it. */ + private static T newRequest(String uri, Function constructor) { + try { + return constructor.apply(uri); + } catch (RuntimeException e) { + throw SensitiveConfigUtils.invalidUri(uri); + } + } + private static RuntimeException httpError(int statusCode) { return new RuntimeException("HTTP error code: " + statusCode); } diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/RESTUtil.java b/paimon-api/src/main/java/org/apache/paimon/rest/RESTUtil.java index 8a5e760793b7..280a4aabb821 100644 --- a/paimon-api/src/main/java/org/apache/paimon/rest/RESTUtil.java +++ b/paimon-api/src/main/java/org/apache/paimon/rest/RESTUtil.java @@ -167,7 +167,10 @@ public static String encodedBody(Object body) { try { return RESTApi.toJson(body); } catch (JsonProcessingException e) { - throw new RESTException(e, "Failed to encode request body: %s", body); + // Keep only the body type: a throwing getter can surface the secret in both + // the cause chain and Jackson's message, so neither is safe to include. + throw new RESTException( + "Failed to encode request body of type %s", body.getClass().getName()); } } return null; @@ -199,7 +202,8 @@ public static String buildRequestUrl(String url, Map queryParams url = builder.build().toString(); } } catch (URISyntaxException e) { - throw new RESTException(e, "build request URL failed."); + // cause / getMessage() echo the raw URL (may carry credentials); keep only the reason. + throw new RESTException("build request URL failed: %s", e.getReason()); } return url; diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/SimpleHttpClient.java b/paimon-api/src/main/java/org/apache/paimon/rest/SimpleHttpClient.java index 04aaf3bfea64..1fc21a31eb89 100644 --- a/paimon-api/src/main/java/org/apache/paimon/rest/SimpleHttpClient.java +++ b/paimon-api/src/main/java/org/apache/paimon/rest/SimpleHttpClient.java @@ -47,7 +47,7 @@ private SimpleHttpClient() { } public String post(String url, Object body, Map headers) throws IOException { - HttpPost httpPost = new HttpPost(url); + HttpPost httpPost = HttpClientUtils.newHttpPost(url); if (headers != null) { httpPost.setHeaders( headers.entrySet().stream() @@ -68,7 +68,7 @@ public String post(String url, Object body, Map headers) throws public String get(String url, Map queryParams, Map headers) throws IOException { - HttpGet httpGet = new HttpGet(RESTUtil.buildRequestUrl(url, queryParams)); + HttpGet httpGet = HttpClientUtils.newHttpGet(RESTUtil.buildRequestUrl(url, queryParams)); if (headers != null) { httpGet.setHeaders( headers.entrySet().stream() @@ -101,8 +101,8 @@ private String exec(HttpUriRequestBase request) { return responseBodyStr; }); } catch (IOException e) { - throw new RuntimeException( - "Failed to convert HTTP response body to string, error : " + e.getMessage()); + // No message from e: a redirect/protocol error can echo the target URL (a signed URL). + throw new RuntimeException("Failed to convert HTTP response body to string."); } } diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/auth/DLFECSTokenLoader.java b/paimon-api/src/main/java/org/apache/paimon/rest/auth/DLFECSTokenLoader.java index 404cb49b8db2..56b687ee12f4 100644 --- a/paimon-api/src/main/java/org/apache/paimon/rest/auth/DLFECSTokenLoader.java +++ b/paimon-api/src/main/java/org/apache/paimon/rest/auth/DLFECSTokenLoader.java @@ -22,6 +22,8 @@ import org.apache.paimon.rest.RESTApi; import org.apache.paimon.rest.SimpleHttpClient; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.JsonProcessingException; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -66,13 +68,17 @@ private static String getRole(String url) { } private static DLFToken getToken(String url) { + String token; try { - String token = SimpleHttpClient.INSTANCE.get(url); - return RESTApi.fromJson(token, DLFToken.class); + token = SimpleHttpClient.INSTANCE.get(url); } catch (IOException e) { throw new UncheckedIOException(e); - } catch (Exception e) { - throw new RuntimeException("get token failed, error : " + e.getMessage(), e); + } + try { + return RESTApi.fromJson(token, DLFToken.class); + } catch (JsonProcessingException e) { + // The token response carries AK/SK/STS; never surface the body or Jackson's snippet. + throw new RuntimeException("Failed to parse ECS token response."); } } diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/auth/DLFLocalFileTokenLoader.java b/paimon-api/src/main/java/org/apache/paimon/rest/auth/DLFLocalFileTokenLoader.java index 199152474426..5514cd34c0bf 100644 --- a/paimon-api/src/main/java/org/apache/paimon/rest/auth/DLFLocalFileTokenLoader.java +++ b/paimon-api/src/main/java/org/apache/paimon/rest/auth/DLFLocalFileTokenLoader.java @@ -18,9 +18,12 @@ package org.apache.paimon.rest.auth; +import org.apache.paimon.annotation.VisibleForTesting; import org.apache.paimon.rest.RESTApi; import org.apache.paimon.utils.FileReadUtils; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.JsonProcessingException; + import java.io.File; /** DLF Token Loader for local file. */ @@ -43,14 +46,23 @@ public String description() { } protected static DLFToken readToken(String tokenFilePath) { + return readToken(tokenFilePath, 5); + } + + @VisibleForTesting + static DLFToken readToken(String tokenFilePath, int maxRetries) { int retry = 1; - Exception lastException = null; - while (retry <= 5) { + RuntimeException lastException = null; + while (retry <= maxRetries) { try { String tokenStr = FileReadUtils.readFileUtf8(new File(tokenFilePath)); return RESTApi.fromJson(tokenStr, DLFToken.class); + } catch (JsonProcessingException e) { + // The token file carries AK/SK/STS; never keep the body or Jackson's snippet. + lastException = new RuntimeException("Failed to parse token file."); } catch (Exception e) { - lastException = e; + lastException = + new RuntimeException("Failed to read token file: " + tokenFilePath, e); } try { Thread.sleep(retry * 1000L); @@ -60,6 +72,6 @@ protected static DLFToken readToken(String tokenFilePath) { } retry++; } - throw new RuntimeException(lastException); + throw lastException; } } diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/interceptor/LoggingInterceptor.java b/paimon-api/src/main/java/org/apache/paimon/rest/interceptor/LoggingInterceptor.java index 8422944dca5c..75898455cd75 100644 --- a/paimon-api/src/main/java/org/apache/paimon/rest/interceptor/LoggingInterceptor.java +++ b/paimon-api/src/main/java/org/apache/paimon/rest/interceptor/LoggingInterceptor.java @@ -18,6 +18,8 @@ package org.apache.paimon.rest.interceptor; +import org.apache.paimon.utils.SensitiveConfigUtils; + import org.apache.hc.core5.http.EntityDetails; import org.apache.hc.core5.http.HttpRequest; import org.apache.hc.core5.http.HttpResponse; @@ -55,10 +57,11 @@ public void process( "[rest] requestId:{} method:{} url:{} duration:{}ms", requestId, request.getMethod(), - request.getUri(), + SensitiveConfigUtils.sanitizeUri(request.getUri().toString()), durationMs); } catch (URISyntaxException e) { - LOG.warn("Failed to log rest request: {}", e.getMessage()); + // e.getMessage() echoes the raw URI (may carry credentials); log only the reason. + LOG.warn("Failed to log rest request: {}", e.getReason()); } } } diff --git a/paimon-api/src/main/java/org/apache/paimon/utils/SensitiveConfigUtils.java b/paimon-api/src/main/java/org/apache/paimon/utils/SensitiveConfigUtils.java new file mode 100644 index 000000000000..08d0338ba7f7 --- /dev/null +++ b/paimon-api/src/main/java/org/apache/paimon/utils/SensitiveConfigUtils.java @@ -0,0 +1,236 @@ +/* + * 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.paimon.utils; + +import java.net.URI; +import java.net.URISyntaxException; +import java.util.LinkedHashMap; +import java.util.Locale; +import java.util.Map; + +/** + * Redacts sensitive configuration values (passwords, secrets, tokens, access keys) before they are + * written to logs or exception messages. Only protects Paimon-produced output; third-party logs + * (Hive, Hadoop, cloud SDKs) must be handled by the runtime logging config. + */ +public class SensitiveConfigUtils { + + /** Placeholder for a fully-masked (short) sensitive value. */ + public static final String REDACTED = "******"; + + /** Values at least this long keep their last {@link #TAIL_LEN} chars to aid troubleshooting. */ + private static final int MIN_LEN_FOR_TAIL = 12; + + private static final int TAIL_LEN = 4; + + /** + * Substrings that mark a configuration key as sensitive. Keys are normalized (lower-cased with + * separators removed) before matching, so {@code fs.oss.accessKeySecret}, {@code + * fs.s3a.access.key}, {@code fs.azure.account.key.*}, {@code fs.azure.sas.*} and {@code + * fs.s3a.encryption.key} are all detected. + */ + private static final String[] SENSITIVE_KEY_PATTERNS = { + "password", + "secret", + "token", + "credential", + "accesskey", + "accountkey", + "encryptionkey", + "authorization", + "privatekey", + "apikey", + "sas" + }; + + /** + * Keys whose value is fully masked instead of keeping a trailing hint. Every true secret is + * here; only identifier-like keys ({@code accessKeyId}) keep a tail. This mirrors AWS/Azure, + * where the access-key id is loggable but the secret / token / SAS is never shown. Note {@code + * accessKeySecret} normalizes to contain {@code secret}, so it is fully masked while {@code + * accessKeyId} (only {@code accesskey}) keeps its tail. + */ + private static final String[] FULL_MASK_KEY_PATTERNS = { + "password", + "token", + "authorization", + "secret", + "credential", + "privatekey", + "encryptionkey", + "apikey", + "accountkey", + "sas" + }; + + /** + * Markers that flag free-form text (a server error message or a signed URL) as possibly + * carrying a secret. Matched after the text is normalized (lower-cased, separators removed), so + * {@code api_key}, {@code private.key}, {@code accessKey} and {@code X-Amz-Signature} all hit. + */ + private static final String[] SENSITIVE_TEXT_MARKERS = { + "password", + "secret", + "token", + "credential", + "accesskey", + "accountkey", + "encryptionkey", + "authorization", + "privatekey", + "apikey", + "signature", + "sas" + }; + + /** + * Literal markers (Azure SAS {@code sig}) that would be ambiguous once separators are removed. + */ + private static final String[] SENSITIVE_TEXT_LITERAL_MARKERS = {"sig=", "\"sig\"", "'sig'"}; + + private SensitiveConfigUtils() {} + + /** Returns whether the given configuration key is considered sensitive. */ + public static boolean isSensitive(String key) { + return matchesAny(key, SENSITIVE_KEY_PATTERNS); + } + + private static boolean matchesAny(String key, String[] patterns) { + if (key == null || key.isEmpty()) { + return false; + } + String normalized = key.toLowerCase(Locale.ROOT).replaceAll("[^a-z0-9]", ""); + for (String pattern : patterns) { + if (normalized.contains(pattern)) { + return true; + } + } + return false; + } + + /** Returns the value masked if its key is sensitive, otherwise the value unchanged. */ + public static String redactValue(String key, String value) { + return isSensitive(key) ? maskValue(value, matchesAny(key, FULL_MASK_KEY_PATTERNS)) : value; + } + + /** + * Returns a copy of the map with every sensitive value masked. Insertion order is preserved. + * Safe to pass {@code null}. + */ + public static Map redactMap(Map options) { + if (options == null) { + return null; + } + Map redacted = new LinkedHashMap<>(options.size()); + for (Map.Entry entry : options.entrySet()) { + String key = entry.getKey(); + redacted.put(key, redactValue(key, entry.getValue())); + } + return redacted; + } + + /** + * Redacts free-form text (e.g. a server {@code ErrorResponse.message}). Arbitrary text cannot + * be masked per-secret reliably, so if any sensitive marker is present the whole text is + * replaced. Runs a plain linear substring scan (no regex backtracking / ReDoS) and is safe to + * call on attacker-controlled input. Prefer {@link #redactMap} when the keys are known. + */ + public static String redactText(String text) { + if (text == null || text.isEmpty()) { + return text; + } + String lower = text.toLowerCase(Locale.ROOT); + for (String marker : SENSITIVE_TEXT_LITERAL_MARKERS) { + if (lower.contains(marker)) { + return REDACTED; + } + } + String normalized = lower.replaceAll("[^a-z0-9]", ""); + for (String marker : SENSITIVE_TEXT_MARKERS) { + if (normalized.contains(marker)) { + return REDACTED; + } + } + return text; + } + + /** Placeholder for a URI that cannot be parsed and thus cannot be safely sanitized. */ + public static final String INVALID_URI = ""; + + /** Message prefix of the safe exception thrown for an unparseable URI. */ + public static final String INVALID_URI_MESSAGE_PREFIX = "Invalid URI: "; + + /** + * Builds a safe exception for an unparseable URI. The original exception echoes the raw URI + * (which may carry credentials), so it is neither kept as a cause nor reused as the message. + * Callers across modules share this prefix so the exception can be classified uniformly. + */ + public static IllegalArgumentException invalidUri(String uri) { + return new IllegalArgumentException(INVALID_URI_MESSAGE_PREFIX + sanitizeUri(uri)); + } + + /** + * Strips the query string and user-info from a URI so signed-URL credentials (AWS/GCS + * signature, Azure SAS {@code sig}, an embedded {@code user:password}) never reach logs or + * exceptions. Returns {@code scheme://host[:port]/path}. Fail-closed: an unparseable URI is + * replaced by {@link #INVALID_URI} instead of guessing at its structure. + */ + public static String sanitizeUri(String uri) { + if (uri == null || uri.isEmpty()) { + return uri; + } + try { + URI parsed = new URI(uri); + // Host unresolved but '@' present: credentials may hide elsewhere -> fail closed. + if (parsed.getHost() == null && uri.indexOf('@') >= 0) { + return INVALID_URI; + } + StringBuilder sb = new StringBuilder(); + if (parsed.getScheme() != null) { + sb.append(parsed.getScheme()).append("://"); + } + if (parsed.getHost() != null) { + sb.append(parsed.getHost()); + if (parsed.getPort() != -1) { + sb.append(':').append(parsed.getPort()); + } + } + if (parsed.getRawPath() != null) { + sb.append(parsed.getRawPath()); + } + return sb.length() == 0 ? INVALID_URI : sb.toString(); + } catch (URISyntaxException e) { + return INVALID_URI; + } + } + + /** + * Masks a value. Password/token keys are fully masked; other secrets keep their last {@link + * #TAIL_LEN} chars when long enough, to aid troubleshooting. + */ + private static String maskValue(String value, boolean fullMask) { + if (value == null) { + return null; + } + if (!fullMask && value.length() >= MIN_LEN_FOR_TAIL) { + return "****" + value.substring(value.length() - TAIL_LEN); + } + return REDACTED; + } +} diff --git a/paimon-api/src/test/java/org/apache/paimon/rest/HttpClientUtilsTest.java b/paimon-api/src/test/java/org/apache/paimon/rest/HttpClientUtilsTest.java index d92cb2b5013a..cb56cc1faeca 100644 --- a/paimon-api/src/test/java/org/apache/paimon/rest/HttpClientUtilsTest.java +++ b/paimon-api/src/test/java/org/apache/paimon/rest/HttpClientUtilsTest.java @@ -18,9 +18,12 @@ package org.apache.paimon.rest; +import org.apache.paimon.utils.SensitiveConfigUtils; + import com.sun.net.httpserver.HttpExchange; import com.sun.net.httpserver.HttpHandler; import com.sun.net.httpserver.HttpServer; +import org.assertj.core.api.ThrowableAssert; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -178,6 +181,51 @@ public void testGetAsInputStreamThrowsForNotFound() { .hasMessage("HTTP error code: 404"); } + @Test + public void testInvalidUriExceptionDoesNotLeakCredentials() { + String uri = "https://alice:secret@host/bad path?sig=QUERY_SECRET"; + for (ThrowableAssert.ThrowingCallable call : + new ThrowableAssert.ThrowingCallable[] { + () -> HttpClientUtils.exists(uri), () -> HttpClientUtils.getAsInputStream(uri) + }) { + assertThatThrownBy(call) + .isInstanceOf(IllegalArgumentException.class) + .hasNoCause() + .matches(e -> HttpClientUtils.isInvalidUriException(e)) + .satisfies( + e -> { + assertThat(String.valueOf(e)).doesNotContain("secret"); + assertThat(String.valueOf(e)).doesNotContain("QUERY_SECRET"); + }); + } + } + + @Test + public void testExecuteFailureDoesNotLeakRedirectLocation() { + String secretLocation = url("/redirect") + "?sig=REDIRECT_SECRET"; + registerHandler( + "/redirect", + exchange -> { + exchange.getResponseHeaders().add("Location", secretLocation); + respond(exchange, 302, new byte[0]); + }); + + for (ThrowableAssert.ThrowingCallable call : + new ThrowableAssert.ThrowingCallable[] { + () -> HttpClientUtils.getAsInputStream(url("/redirect")), + () -> HttpClientUtils.exists(url("/redirect")) + }) { + assertThatThrownBy(call) + .isInstanceOf(IOException.class) + .hasNoCause() + .satisfies( + e -> { + assertThat(String.valueOf(e)).doesNotContain("REDIRECT_SECRET"); + assertThat(e.getMessage()).doesNotContain("sig="); + }); + } + } + @Test public void testGetAsInputStreamDoesNotLeakConnectionsOnRepeatedNotFound() throws Exception { registerHandler( @@ -224,6 +272,11 @@ public void testIsInvalidUriException() { HttpClientUtils.isInvalidUriException( new IllegalArgumentException("Illegal character in path"))) .isTrue(); + // The shared invalid-URI exception must classify uniformly across modules. + assertThat( + HttpClientUtils.isInvalidUriException( + SensitiveConfigUtils.invalidUri("https://host/bad path"))) + .isTrue(); assertThat( HttpClientUtils.isInvalidUriException( new RuntimeException("HTTP error code: 404"))) diff --git a/paimon-api/src/test/java/org/apache/paimon/rest/SimpleHttpClientTest.java b/paimon-api/src/test/java/org/apache/paimon/rest/SimpleHttpClientTest.java new file mode 100644 index 000000000000..0f426918c579 --- /dev/null +++ b/paimon-api/src/test/java/org/apache/paimon/rest/SimpleHttpClientTest.java @@ -0,0 +1,55 @@ +/* + * 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.paimon.rest; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for {@link SimpleHttpClient}. */ +public class SimpleHttpClientTest { + + private static final String MALFORMED_URL = + "https://alice:secret@host/bad path?sig=QUERY_SECRET"; + + @Test + public void testGetWithMalformedUrlDoesNotLeakCredentials() { + assertMalformedUrlSanitized(() -> SimpleHttpClient.INSTANCE.get(MALFORMED_URL)); + } + + @Test + public void testPostWithMalformedUrlDoesNotLeakCredentials() { + assertMalformedUrlSanitized( + () -> SimpleHttpClient.INSTANCE.post(MALFORMED_URL, null, null)); + } + + private static void assertMalformedUrlSanitized( + org.assertj.core.api.ThrowableAssert.ThrowingCallable call) { + assertThatThrownBy(call) + .isInstanceOf(IllegalArgumentException.class) + .hasNoCause() + .matches(HttpClientUtils::isInvalidUriException) + .satisfies( + e -> { + assertThat(String.valueOf(e)).doesNotContain("secret"); + assertThat(String.valueOf(e)).doesNotContain("QUERY_SECRET"); + }); + } +} diff --git a/paimon-api/src/test/java/org/apache/paimon/rest/auth/DLFECSTokenLoaderTest.java b/paimon-api/src/test/java/org/apache/paimon/rest/auth/DLFECSTokenLoaderTest.java new file mode 100644 index 000000000000..704ff8a4501f --- /dev/null +++ b/paimon-api/src/test/java/org/apache/paimon/rest/auth/DLFECSTokenLoaderTest.java @@ -0,0 +1,80 @@ +/* + * 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.paimon.rest.auth; + +import com.sun.net.httpserver.HttpExchange; +import com.sun.net.httpserver.HttpServer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.io.OutputStream; +import java.net.InetSocketAddress; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for {@link DLFECSTokenLoader}. */ +public class DLFECSTokenLoaderTest { + + private HttpServer server; + private int port; + + @BeforeEach + public void setUp() throws Exception { + server = HttpServer.create(new InetSocketAddress(0), 0); + port = server.getAddress().getPort(); + server.start(); + } + + @AfterEach + public void tearDown() { + if (server != null) { + server.stop(0); + } + } + + @Test + public void testMalformedTokenResponseDoesNotLeakCredentials() { + String secret = "STSSECRET_AKID_9999"; + server.createContext( + "/token", + exchange -> + respond( + exchange, + ("{\"AccessKeyId\":\"akid\",\"AccessKeySecret\":\"" + + secret + + "\" INVALID_JSON") + .getBytes())); + + DLFECSTokenLoader loader = new DLFECSTokenLoader("http://127.0.0.1:" + port + "/token", ""); + + assertThatThrownBy(loader::loadToken) + .hasNoCause() + .satisfies(e -> assertThat(String.valueOf(e)).doesNotContain(secret)); + } + + private static void respond(HttpExchange exchange, byte[] body) throws IOException { + exchange.sendResponseHeaders(200, body.length); + try (OutputStream out = exchange.getResponseBody()) { + out.write(body); + } + } +} diff --git a/paimon-api/src/test/java/org/apache/paimon/rest/auth/DLFLocalFileTokenLoaderTest.java b/paimon-api/src/test/java/org/apache/paimon/rest/auth/DLFLocalFileTokenLoaderTest.java new file mode 100644 index 000000000000..b56ab4e2db59 --- /dev/null +++ b/paimon-api/src/test/java/org/apache/paimon/rest/auth/DLFLocalFileTokenLoaderTest.java @@ -0,0 +1,48 @@ +/* + * 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.paimon.rest.auth; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Files; +import java.nio.file.Path; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for {@link DLFLocalFileTokenLoader}. */ +public class DLFLocalFileTokenLoaderTest { + + @TempDir Path tempDir; + + @Test + public void testMalformedTokenFileDoesNotLeakCredentials() throws Exception { + String secret = "STSSECRET_AKID_9999"; + Path tokenFile = tempDir.resolve("token.json"); + Files.write( + tokenFile, + ("{\"AccessKeyId\":\"akid\",\"AccessKeySecret\":\"" + secret + "\" INVALID_JSON") + .getBytes()); + + assertThatThrownBy(() -> DLFLocalFileTokenLoader.readToken(tokenFile.toString(), 1)) + .hasNoCause() + .satisfies(e -> assertThat(String.valueOf(e)).doesNotContain(secret)); + } +} diff --git a/paimon-api/src/test/java/org/apache/paimon/utils/SensitiveConfigUtilsTest.java b/paimon-api/src/test/java/org/apache/paimon/utils/SensitiveConfigUtilsTest.java new file mode 100644 index 000000000000..6120a75e5774 --- /dev/null +++ b/paimon-api/src/test/java/org/apache/paimon/utils/SensitiveConfigUtilsTest.java @@ -0,0 +1,180 @@ +/* + * 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.paimon.utils; + +import org.junit.jupiter.api.Test; + +import java.util.LinkedHashMap; +import java.util.Map; + +import static org.apache.paimon.utils.SensitiveConfigUtils.REDACTED; +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link SensitiveConfigUtils}. */ +class SensitiveConfigUtilsTest { + + @Test + void testIsSensitiveDetectsCredentialKeys() { + // Access keys / secrets in various vendor spellings. + assertThat(SensitiveConfigUtils.isSensitive("fs.oss.accessKeyId")).isTrue(); + assertThat(SensitiveConfigUtils.isSensitive("fs.oss.accessKeySecret")).isTrue(); + assertThat(SensitiveConfigUtils.isSensitive("fs.s3a.access.key")).isTrue(); + assertThat(SensitiveConfigUtils.isSensitive("fs.s3a.secret.key")).isTrue(); + assertThat(SensitiveConfigUtils.isSensitive("fs.s3a.session.token")).isTrue(); + assertThat(SensitiveConfigUtils.isSensitive("security.token")).isTrue(); + assertThat(SensitiveConfigUtils.isSensitive("my.password")).isTrue(); + assertThat(SensitiveConfigUtils.isSensitive("some.credential")).isTrue(); + assertThat(SensitiveConfigUtils.isSensitive("Authorization")).isTrue(); + assertThat(SensitiveConfigUtils.isSensitive("client.private-key")).isTrue(); + assertThat(SensitiveConfigUtils.isSensitive("dlf.api-key")).isTrue(); + // Azure / S3 cloud credentials. + assertThat(SensitiveConfigUtils.isSensitive("fs.azure.account.key.acct.blob.core")) + .isTrue(); + assertThat(SensitiveConfigUtils.isSensitive("fs.azure.sas.container.acct.blob.core")) + .isTrue(); + assertThat(SensitiveConfigUtils.isSensitive("fs.s3a.encryption.key")).isTrue(); + } + + @Test + void testIsSensitiveIgnoresNonCredentialKeys() { + assertThat(SensitiveConfigUtils.isSensitive("bucket")).isFalse(); + assertThat(SensitiveConfigUtils.isSensitive("warehouse")).isFalse(); + assertThat(SensitiveConfigUtils.isSensitive("fs.oss.endpoint")).isFalse(); + assertThat(SensitiveConfigUtils.isSensitive("metastore")).isFalse(); + assertThat(SensitiveConfigUtils.isSensitive("")).isFalse(); + assertThat(SensitiveConfigUtils.isSensitive(null)).isFalse(); + } + + @Test + void testRedactValue() { + // Identifier-like access-key id keeps only its last 4 chars. + assertThat(SensitiveConfigUtils.redactValue("fs.oss.accessKeyId", "0123456789abcdef")) + .isEqualTo("****cdef"); + // Short value is fully masked. + assertThat(SensitiveConfigUtils.redactValue("password", "short")).isEqualTo(REDACTED); + // Non-sensitive key is untouched. + assertThat(SensitiveConfigUtils.redactValue("bucket", "my-bucket")).isEqualTo("my-bucket"); + } + + @Test + void testTrueSecretsAreFullyMasked() { + // True secrets are never partially revealed (AWS/Azure: the secret is never shown). + for (String key : + new String[] { + "my.password", + "security.token", + "Authorization", + "fs.s3a.secret.key", + "fs.oss.accessKeySecret", + "client.private-key", + "fs.s3a.encryption.key", + "fs.azure.account.key.acct", + "fs.azure.sas.acct", + "some.credential" + }) { + assertThat(SensitiveConfigUtils.redactValue(key, "0123456789abcdef")) + .as(key) + .isEqualTo(REDACTED); + } + // Contrast: an access-key id (an identifier) keeps its tail. + assertThat(SensitiveConfigUtils.redactValue("fs.oss.accessKeyId", "0123456789abcdef")) + .isEqualTo("****cdef"); + } + + @Test + void testRedactMapReplacesOnlySensitiveValues() { + Map options = new LinkedHashMap<>(); + options.put("warehouse", "mock://warehouse"); + options.put("fs.oss.accessKeyId", "mock-access-id-0001"); + options.put("fs.oss.accessKeySecret", "mock-secret-value"); + options.put("fs.azure.account.key.acct", "mock-azure-key-value"); + options.put("fs.oss.endpoint", "mock-endpoint.example.com"); + + Map redacted = SensitiveConfigUtils.redactMap(options); + + assertThat(redacted).containsEntry("warehouse", "mock://warehouse"); + // Access-key id keeps a tail; true secrets are fully masked; endpoint is untouched. + assertThat(redacted).containsEntry("fs.oss.accessKeyId", "****0001"); + assertThat(redacted).containsEntry("fs.oss.accessKeySecret", REDACTED); + assertThat(redacted).containsEntry("fs.azure.account.key.acct", REDACTED); + assertThat(redacted).containsEntry("fs.oss.endpoint", "mock-endpoint.example.com"); + // Original map must not be mutated. + assertThat(options).containsEntry("fs.oss.accessKeySecret", "mock-secret-value"); + // The rendered string must not contain any raw secret. + assertThat(redacted.toString()) + .doesNotContain("mock-secret-value") + .doesNotContain("mock-access-id-0001"); + } + + @Test + void testRedactMapNullSafe() { + assertThat(SensitiveConfigUtils.redactMap(null)).isNull(); + } + + @Test + void testRedactTextRedactsWholeMarkedText() { + for (String text : + new String[] { + "{\"accessKeySecret\":\"mock-secret-abcd\",\"endpoint\":\"mock\"}", + "password=mock-pass&user=alice", + "Authorization: Bearer mock-jwt-token-abcdef", + "{\"password\":\"alpha,beta\"}", + "{\"sas\":\"sv=2024-11-04&sig=REST-LEAK\"}", + "invalid token SECRET-9999 in request", + "https://x.blob.core.windows.net/c/f?sig=SECRETSIG", + "api_key=mock", + "private.key: mock", + "fs.s3a.access.key=mock", + "{\"apiKey\":\"mock\"}" + }) { + assertThat(SensitiveConfigUtils.redactText(text)).as(text).isEqualTo(REDACTED); + } + } + + @Test + void testRedactTextKeepsTextWithoutMarkers() { + assertThat(SensitiveConfigUtils.redactText(null)).isNull(); + assertThat(SensitiveConfigUtils.redactText("")).isEmpty(); + assertThat(SensitiveConfigUtils.redactText("Table not found: default.t")) + .isEqualTo("Table not found: default.t"); + } + + @Test + void testSanitizeUriDropsQueryAndUserInfo() { + assertThat(SensitiveConfigUtils.sanitizeUri("https://host:443/p?sig=x&X-Amz-Signature=y")) + .isEqualTo("https://host:443/p"); + assertThat(SensitiveConfigUtils.sanitizeUri("https://alice:secret@host/p?sig=x")) + .isEqualTo("https://host/p"); + } + + @Test + void testSanitizeUriFailsClosedOnUnparseableUri() { + for (String uri : + new String[] { + "https://alice:secret@host/bad path?sig=x", + "https://alice:se/cret@host/bad path?sig=x", + "https://alice:se@cret@host/bad path?sig=x", + "https://alice:se/cret@host/p?sig=x" + }) { + assertThat(SensitiveConfigUtils.sanitizeUri(uri)) + .as(uri) + .isEqualTo(SensitiveConfigUtils.INVALID_URI); + } + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/HadoopUtils.java b/paimon-common/src/main/java/org/apache/paimon/utils/HadoopUtils.java index a37c64dbb847..eb21b4eb8754 100644 --- a/paimon-common/src/main/java/org/apache/paimon/utils/HadoopUtils.java +++ b/paimon-common/src/main/java/org/apache/paimon/utils/HadoopUtils.java @@ -140,11 +140,12 @@ public static Configuration getHadoopConfiguration(Options options) { String newKey = key.substring(prefix.length()); String value = options.getString(key, null); result.set(newKey, value); + // Redact value: sensitive keys are masked, others printed as-is. LOG.debug( "Adding Paimon config entry for {} as {}={} to Hadoop config", key, newKey, - value); + SensitiveConfigUtils.redactValue(newKey, value)); foundHadoopConfiguration = true; } } diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/UriReaderFactory.java b/paimon-common/src/main/java/org/apache/paimon/utils/UriReaderFactory.java index a2fd2d4ad356..08f0691303e6 100644 --- a/paimon-common/src/main/java/org/apache/paimon/utils/UriReaderFactory.java +++ b/paimon-common/src/main/java/org/apache/paimon/utils/UriReaderFactory.java @@ -44,11 +44,19 @@ public UriReaderFactory(CatalogContext context) { } public UriReader create(String input) { - URI uri = URI.create(input); + URI uri = parseUri(input); UriKey key = new UriKey(uri.getScheme(), uri.getAuthority()); return readers.computeIfAbsent(key, k -> newReader(k, uri)); } + private static URI parseUri(String input) { + try { + return URI.create(input); + } catch (IllegalArgumentException e) { + throw SensitiveConfigUtils.invalidUri(input); + } + } + public boolean exists(String input) throws IOException { UriReader reader = create(input); if (reader instanceof UriReader.FileUriReader) { diff --git a/paimon-common/src/test/java/org/apache/paimon/utils/HadoopUtilsSecretLoggingTest.java b/paimon-common/src/test/java/org/apache/paimon/utils/HadoopUtilsSecretLoggingTest.java new file mode 100644 index 000000000000..7f092b8a865e --- /dev/null +++ b/paimon-common/src/test/java/org/apache/paimon/utils/HadoopUtilsSecretLoggingTest.java @@ -0,0 +1,83 @@ +/* + * 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.paimon.utils; + +import org.apache.paimon.options.Options; + +import org.apache.hadoop.conf.Configuration; +import org.apache.logging.log4j.Level; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.core.Logger; +import org.apache.logging.log4j.core.appender.OutputStreamAppender; +import org.apache.logging.log4j.core.layout.PatternLayout; +import org.junit.jupiter.api.Test; + +import java.io.ByteArrayOutputStream; +import java.nio.charset.StandardCharsets; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Verifies that {@link HadoopUtils#getHadoopConfiguration} forwards {@code hadoop.*} credentials to + * the Hadoop configuration without leaking their values into the (DEBUG) logs. + */ +class HadoopUtilsSecretLoggingTest { + + private static final String SECRET_MARKER = "UNIQUE-SECRET-MARKER-9f8e7d6c"; + + @Test + void testCredentialValueIsNotLogged() { + Options options = new Options(); + // hadoop.* keys are forwarded (prefix stripped) into the Hadoop configuration. + options.set("hadoop.fs.oss.accessKeySecret", SECRET_MARKER); + options.set("hadoop.fs.oss.endpoint", "mock-endpoint.example.com"); + + // Warm up: first Hadoop Configuration build reconfigures log4j2 and drops our appender. + HadoopUtils.getHadoopConfiguration(new Options()); + + ByteArrayOutputStream output = new ByteArrayOutputStream(); + OutputStreamAppender appender = + OutputStreamAppender.newBuilder() + .setName("hadoop-utils-secret-test") + .setTarget(output) + .setLayout(PatternLayout.newBuilder().withPattern("%level %msg%n").build()) + .build(); + Logger logger = (Logger) LogManager.getLogger(HadoopUtils.class); + Level previousLevel = logger.getLevel(); + + appender.start(); + logger.addAppender(appender); + logger.setLevel(Level.DEBUG); + try { + Configuration conf = HadoopUtils.getHadoopConfiguration(options); + + // The credential must still be forwarded so the configuration keeps working. + assertThat(conf.get("fs.oss.accessKeySecret")).isEqualTo(SECRET_MARKER); + + String logs = new String(output.toByteArray(), StandardCharsets.UTF_8); + // The key name may appear, but the secret value must never be logged. + assertThat(logs).contains("fs.oss.accessKeySecret"); + assertThat(logs).doesNotContain(SECRET_MARKER); + } finally { + logger.setLevel(previousLevel); + logger.removeAppender(appender); + appender.stop(); + } + } +} diff --git a/paimon-common/src/test/java/org/apache/paimon/utils/UriReaderFactoryTest.java b/paimon-common/src/test/java/org/apache/paimon/utils/UriReaderFactoryTest.java index 3cf1cbaf3e5d..e0679cb74fcd 100644 --- a/paimon-common/src/test/java/org/apache/paimon/utils/UriReaderFactoryTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/utils/UriReaderFactoryTest.java @@ -36,6 +36,7 @@ import java.nio.file.Files; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Test for {@link UriReaderFactory}. */ public class UriReaderFactoryTest { @@ -68,6 +69,19 @@ public void testCreateHttpUriReader() { assertThat(reader).isInstanceOf(HttpUriReader.class); } + @Test + public void testInvalidUriDoesNotLeakCredentials() { + assertThatThrownBy( + () -> factory.create("https://alice:secret@host/bad path?sig=QUERY_SECRET")) + .isInstanceOf(IllegalArgumentException.class) + .hasNoCause() + .satisfies( + e -> { + assertThat(String.valueOf(e)).doesNotContain("secret"); + assertThat(String.valueOf(e)).doesNotContain("QUERY_SECRET"); + }); + } + @Test public void testCreateHttpsUriReader() { UriReader reader = factory.create("https://example.com/file.txt"); diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/HttpClientTest.java b/paimon-core/src/test/java/org/apache/paimon/rest/HttpClientTest.java index a7424190372d..5bdc553fdf6a 100644 --- a/paimon-core/src/test/java/org/apache/paimon/rest/HttpClientTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/rest/HttpClientTest.java @@ -23,6 +23,7 @@ import org.apache.paimon.rest.auth.RESTAuthFunction; import org.apache.paimon.rest.auth.RESTAuthParameter; import org.apache.paimon.rest.exceptions.BadRequestException; +import org.apache.paimon.rest.exceptions.RESTException; import org.apache.paimon.rest.responses.ErrorResponse; import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableMap; @@ -43,6 +44,7 @@ import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertThrows; /** Test for {@link HttpClient}. */ @@ -130,6 +132,51 @@ public void testGetFail() { () -> httpClient.get(MOCK_PATH, MockRESTData.class, restAuthFunction)); } + @Test + public void testErrorResponseMessageIsRedacted() throws Exception { + // A parsed ErrorResponse whose message carries a secret must not leak it. + String secret = "SUPERSECRETVALUE9999"; + String body = + server.createResponseBody( + new ErrorResponse( + ErrorResponse.RESOURCE_TYPE_DATABASE, + "test", + "token=" + secret, + 400)); + server.enqueueResponse(body, 400); + BadRequestException e = + assertThrows( + BadRequestException.class, + () -> httpClient.get(MOCK_PATH, MockRESTData.class, restAuthFunction)); + assertFalse(e.getMessage().contains(secret)); + } + + @Test + public void testUnparseableErrorBodyIsNotEchoed() { + // A non-JSON error body cannot be reliably sanitized, so it must not be echoed at all. + String secret = "SUPERSECRETVALUE9999"; + server.enqueueResponse("token=" + secret + " <>", 400); + BadRequestException e = + assertThrows( + BadRequestException.class, + () -> httpClient.get(MOCK_PATH, MockRESTData.class, restAuthFunction)); + assertFalse(e.getMessage().contains(secret)); + } + + @Test + public void testSuccessResponseParseFailureDoesNotLeakBody() { + // A 2xx body that fails to deserialize (e.g. a token response) must not surface the + // raw body or Jackson's source snippet. + String secret = "SUPERSECRETVALUE9999"; + String body = "{\"data\": \"token=" + secret + "\" INVALID_JSON}"; + server.enqueueResponse(body, 200); + RESTException e = + assertThrows( + RESTException.class, + () -> httpClient.get(MOCK_PATH, MockRESTData.class, restAuthFunction)); + assertFalse(e.getMessage().contains(secret)); + } + @Test public void testPostSuccess() { server.enqueueResponse(mockResponseDataStr, 200); @@ -228,8 +275,8 @@ private Map getParameters(String path) { @Test public void testGetWithUnparsableJsonErrorResponse() { - // Test case for JSON response with mismatched field names that cannot be parsed as - // ErrorResponse + // A JSON body that parses to an ErrorResponse with no message must NOT be echoed (it may + // carry secrets); it is reported as an empty message, not "unparseable". String jsonWithUppercaseFields = "{\"Message\":\"Your request is denied as lack of ssl protect.\"," + "\"Code\":\"InvalidProtocol.NeedSsl\"}"; @@ -239,16 +286,19 @@ public void testGetWithUnparsableJsonErrorResponse() { httpClient.get(MOCK_PATH, MockRESTData.class, restAuthFunction); Assertions.fail("Expected exception to be thrown"); } catch (Exception e) { - Assertions.assertTrue( + Assertions.assertFalse( e.getMessage().contains("Your request is denied as lack of ssl protect") || e.getMessage().contains(jsonWithUppercaseFields), - "Error message should contain the original response body"); + "Raw response body must not be echoed"); + Assertions.assertTrue( + e.getMessage().contains("Empty error message"), + "Parsed-but-empty message must not be labelled unparseable"); } } @Test public void testPostWithNonJsonErrorResponse() { - // Test case for non-JSON response (plain text) that cannot be parsed + // A non-JSON (plain text) error body must NOT be echoed; only a generic message remains. String plainTextResponse = "Internal Server Error: Database connection failed"; server.enqueueResponse(plainTextResponse, 500); @@ -256,14 +306,45 @@ public void testPostWithNonJsonErrorResponse() { httpClient.post(MOCK_PATH, mockResponseData, MockRESTData.class, restAuthFunction); Assertions.fail("Expected exception to be thrown"); } catch (Exception e) { - // Verify that the error message contains the original plain text response - Assertions.assertTrue( + Assertions.assertFalse( e.getMessage().contains(plainTextResponse) || e.getMessage().contains("Database connection failed"), - "Error message should contain the original non-JSON response"); + "Raw response body must not be echoed"); + Assertions.assertTrue( + e.getMessage().contains("500"), "Message should carry the HTTP status"); } } + @Test + public void testGetWithMalformedUrlDoesNotLeakCredentials() { + // Malformed URL throws in the constructor before exec(); the raw URL must not leak. + String secret = "QUERY_SECRET"; + IllegalArgumentException e = + assertThrows( + IllegalArgumentException.class, + () -> + httpClient.get( + "/x?sig=" + secret + " bad", + MockRESTData.class, + restAuthFunction)); + assertFalse(e.getMessage().contains(secret)); + } + + @Test + public void testPostWithMalformedUrlDoesNotLeakCredentials() { + String secret = "QUERY_SECRET"; + IllegalArgumentException e = + assertThrows( + IllegalArgumentException.class, + () -> + httpClient.post( + "/x?sig=" + secret + " bad", + mockResponseData, + MockRESTData.class, + restAuthFunction)); + assertFalse(e.getMessage().contains(secret)); + } + @Test public void testPostSetsJsonContentType() throws Exception { server.enqueueResponse(mockResponseDataStr, 200); diff --git a/paimon-filesystems/paimon-cosn-impl/src/main/java/org/apache/paimon/cosn/COSNFileIO.java b/paimon-filesystems/paimon-cosn-impl/src/main/java/org/apache/paimon/cosn/COSNFileIO.java index 00b0bc60127d..0f0096751d9e 100644 --- a/paimon-filesystems/paimon-cosn-impl/src/main/java/org/apache/paimon/cosn/COSNFileIO.java +++ b/paimon-filesystems/paimon-cosn-impl/src/main/java/org/apache/paimon/cosn/COSNFileIO.java @@ -21,6 +21,7 @@ import org.apache.paimon.catalog.CatalogContext; import org.apache.paimon.fs.FileIO; import org.apache.paimon.options.Options; +import org.apache.paimon.utils.SensitiveConfigUtils; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; @@ -83,7 +84,7 @@ public void configure(CatalogContext context) { LOG.debug( "Adding config entry for {} as {} to Hadoop config", key, - hadoopOptions.get(key)); + SensitiveConfigUtils.redactValue(key, hadoopOptions.get(key))); } } } diff --git a/paimon-filesystems/paimon-gs-impl/src/main/java/org/apache/paimon/gs/GSFileIO.java b/paimon-filesystems/paimon-gs-impl/src/main/java/org/apache/paimon/gs/GSFileIO.java index 2226d8be32bc..d2efcad70c7b 100644 --- a/paimon-filesystems/paimon-gs-impl/src/main/java/org/apache/paimon/gs/GSFileIO.java +++ b/paimon-filesystems/paimon-gs-impl/src/main/java/org/apache/paimon/gs/GSFileIO.java @@ -21,6 +21,7 @@ import org.apache.paimon.catalog.CatalogContext; import org.apache.paimon.fs.FileIO; import org.apache.paimon.options.Options; +import org.apache.paimon.utils.SensitiveConfigUtils; import com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem; import org.apache.hadoop.conf.Configuration; @@ -70,7 +71,7 @@ public void configure(CatalogContext context) { LOG.warn( "Adding config entry for {} as {} to Hadoop config", key, - hadoopOptions.get(key)); + SensitiveConfigUtils.redactValue(key, hadoopOptions.get(key))); } } } diff --git a/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoFileIO.java b/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoFileIO.java index c5e2d13a7707..7cab897e7654 100644 --- a/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoFileIO.java +++ b/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoFileIO.java @@ -26,6 +26,7 @@ import org.apache.paimon.options.Options; import org.apache.paimon.utils.IOUtils; import org.apache.paimon.utils.Pair; +import org.apache.paimon.utils.SensitiveConfigUtils; import org.apache.paimon.utils.StringUtils; import com.aliyun.jindodata.common.JindoHadoopSystem; @@ -113,7 +114,7 @@ public void configure(CatalogContext context) { LOG.debug( "Adding config entry for {} as {} to Hadoop config", key, - hadoopOptions.get(key)); + SensitiveConfigUtils.redactValue(key, hadoopOptions.get(key))); } } } diff --git a/paimon-filesystems/paimon-obs-impl/src/main/java/org/apache/paimon/obs/OBSFileIO.java b/paimon-filesystems/paimon-obs-impl/src/main/java/org/apache/paimon/obs/OBSFileIO.java index ae11dc2f9e7e..4f425ca59065 100644 --- a/paimon-filesystems/paimon-obs-impl/src/main/java/org/apache/paimon/obs/OBSFileIO.java +++ b/paimon-filesystems/paimon-obs-impl/src/main/java/org/apache/paimon/obs/OBSFileIO.java @@ -21,6 +21,7 @@ import org.apache.paimon.catalog.CatalogContext; import org.apache.paimon.fs.FileIO; import org.apache.paimon.options.Options; +import org.apache.paimon.utils.SensitiveConfigUtils; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; @@ -99,7 +100,7 @@ public void configure(CatalogContext context) { LOG.debug( "Adding config entry for {} as {} to Hadoop config", key, - hadoopOptions.get(key)); + SensitiveConfigUtils.redactValue(key, hadoopOptions.get(key))); } } } diff --git a/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSFileIO.java b/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSFileIO.java index 410c460bb419..c397eca2f3de 100644 --- a/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSFileIO.java +++ b/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSFileIO.java @@ -26,6 +26,7 @@ import org.apache.paimon.options.Options; import org.apache.paimon.utils.IOUtils; import org.apache.paimon.utils.ReflectionUtils; +import org.apache.paimon.utils.SensitiveConfigUtils; import org.apache.paimon.utils.StringUtils; import com.aliyun.oss.OSSClient; @@ -145,7 +146,7 @@ public void configure(CatalogContext context) { LOG.debug( "Adding config entry for {} as {} to Hadoop config", key, - hadoopOptions.get(key)); + SensitiveConfigUtils.redactValue(key, hadoopOptions.get(key))); } } } diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java index c13a1bb158e4..c3e55247c34f 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java @@ -37,6 +37,7 @@ import org.apache.paimon.table.FormatTable; import org.apache.paimon.table.Table; import org.apache.paimon.utils.Preconditions; +import org.apache.paimon.utils.SensitiveConfigUtils; import org.apache.flink.api.common.RuntimeExecutionMode; import org.apache.flink.configuration.ConfigOption; @@ -289,7 +290,7 @@ static Map getDynamicConfigOptions(DynamicTableFactory.Context c LOG.info( "Loading dynamic table options for {} in table config: {}", tableName, - optionsFromTableConfig); + SensitiveConfigUtils.redactMap(optionsFromTableConfig)); } return optionsFromTableConfig; } diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkRowWrapper.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkRowWrapper.java index a6d900262b5e..5e587f9780ff 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkRowWrapper.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkRowWrapper.java @@ -33,6 +33,7 @@ import org.apache.paimon.rest.HttpClientUtils; import org.apache.paimon.types.DataTypeRoot; import org.apache.paimon.types.RowKind; +import org.apache.paimon.utils.SensitiveConfigUtils; import org.apache.paimon.utils.UriReaderFactory; import org.apache.flink.table.data.DecimalData; @@ -240,7 +241,7 @@ private boolean descriptorFileExists(int pos, BlobDescriptor descriptor) { } LOG.warn( "Failed to check blob descriptor file {} for BLOB field at position {}.", - descriptor.uri(), + SensitiveConfigUtils.sanitizeUri(descriptor.uri()), pos, e); throw new RuntimeException(e); @@ -250,7 +251,7 @@ private boolean descriptorFileExists(int pos, BlobDescriptor descriptor) { } LOG.warn( "Failed to check blob descriptor file {} for BLOB field at position {}.", - descriptor.uri(), + SensitiveConfigUtils.sanitizeUri(descriptor.uri()), pos, e); throw e; @@ -261,12 +262,12 @@ private void logMissingDescriptor(int pos, BlobDescriptor descriptor) { if (isHttpUri(descriptor.uri())) { LOG.warn( "Blob descriptor file {} returned HTTP 404, returning NULL for BLOB field at position {}.", - descriptor.uri(), + SensitiveConfigUtils.sanitizeUri(descriptor.uri()), pos); } else { LOG.warn( "Blob descriptor file {} does not exist, returning NULL for BLOB field at position {}.", - descriptor.uri(), + SensitiveConfigUtils.sanitizeUri(descriptor.uri()), pos); } } diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/ConfigRefresher.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/ConfigRefresher.java index 1d4b8deb083e..917b851ca687 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/ConfigRefresher.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/ConfigRefresher.java @@ -23,6 +23,7 @@ import org.apache.paimon.options.Options; import org.apache.paimon.schema.TableSchema; import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.utils.SensitiveConfigUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -101,8 +102,8 @@ public void tryRefresh() { refresher.refresh(table); LOG.info( "write has been refreshed due to configs changed. old options:{}, new options:{}.", - currentOptions, - newOptions); + SensitiveConfigUtils.redactMap(currentOptions), + SensitiveConfigUtils.redactMap(newOptions)); } } catch (Exception e) { throw new RuntimeException("update write failed.", e); diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkTableSource.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkTableSource.java index 66cb49798aa0..534783fb02b3 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkTableSource.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkTableSource.java @@ -37,6 +37,7 @@ import org.apache.paimon.table.Table; import org.apache.paimon.table.source.Split; import org.apache.paimon.utils.RowDataToObjectArrayConverter; +import org.apache.paimon.utils.SensitiveConfigUtils; import org.apache.flink.configuration.Configuration; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; @@ -180,7 +181,10 @@ private PartitionPredicate getPartitionPredicateWithOptions() { // In older versions of Flink, however, lookup sources will first be treated as normal // sources. So this method will also be visited by lookup tables, and the options might // cause IllegalArgumentException. In this case we ignore the filters. - LOG.info("Failed to get filter with table options {} ", table.options(), e); + LOG.info( + "Failed to get filter with table options {} ", + SensitiveConfigUtils.redactMap(table.options()), + e); return null; } } diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkRowWrapperTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkRowWrapperTest.java index 517917e4c011..3141ef706242 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkRowWrapperTest.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkRowWrapperTest.java @@ -167,7 +167,8 @@ public void testInvalidUriThrowsWhenFetchFailureDisabled() { assertThatThrownBy(() -> wrapper.isNullAt(0)) .isInstanceOf(IllegalArgumentException.class) - .hasMessageContaining("Illegal character"); + .hasMessageContaining("Invalid URI") + .hasMessageNotContaining("1304008055350781673"); } @Test diff --git a/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatWriter.java b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatWriter.java index 9de5a169d1e2..212f6d8f4a28 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatWriter.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatWriter.java @@ -37,6 +37,7 @@ import org.apache.paimon.types.RowType; import org.apache.paimon.utils.DeltaVarintCompressor; import org.apache.paimon.utils.LongArrayList; +import org.apache.paimon.utils.SensitiveConfigUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -382,7 +383,8 @@ private void writeNullElement() throws IOException { private static String blobUri(@Nullable Blob blob) { if (blob instanceof BlobRef) { - return ((BlobRef) blob).toDescriptor().uri(); + // Sanitize: a signed URL carries token/signature in its query. + return SensitiveConfigUtils.sanitizeUri(((BlobRef) blob).toDescriptor().uri()); } return "unknown"; }