[common] Redact sensitive configuration values in logs and exceptions

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 extends RESTResponse> T get(
             String path, Class<T> 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 extends RESTResponse> T get(
             Class<T> 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 extends RESTResponse> T post(
             RESTRequest body,
             Class<T> 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 extends RESTResponse> T delete(String path, RESTAuthFunction restAuthF
     @Override
     public <T extends RESTResponse> 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,21 +139,31 @@ private <T extends RESTResponse> T exec(HttpUriRequestBase request, Class<T> 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 {
                             throw new RESTException("response body is null.");
                         }
                     });
         } 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);
     }
 }

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,13 +35,15 @@
 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;
 import org.apache.hc.core5.util.Timeout;
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.util.function.Function;
 
 /** Utils for {@link HttpClientBuilder}. */
 public class HttpClientUtils {
@@ -86,8 +91,8 @@
     }
 
     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 @@
             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 @@
             }
             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 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> T newRequest(String uri, Function<String, T> 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);
     }

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<String, String> 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;

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<String, String> 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<String, String> headers) throws
 
     public String get(String url, Map<String, String> queryParams, Map<String, String> 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.");
         }
     }
 

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.");
         }
     }
 

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;
     }
 }

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());
         }
     }
 }