Re: [PR] [improve][client] Add maxConnectionsPerHost and connectionMaxIdleSeconds to PulsarAdminBuilder [pulsar]

["github-advanced-security[bot] (via GitHub)" <gi...@apache.org>]

github-advanced-security[bot] commented on code in PR #22541:
URL: https://github.com/apache/pulsar/pull/22541#discussion_r1707290162


##########
pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/http/AsyncHttpConnector.java:
##########
@@ -323,7 +351,129 @@
         }
     }
 
+    public static class MaxRedirectException extends Exception {
+        public MaxRedirectException(String msg) {
+            super(msg, null, true, false);
+        }
+    }
+
     protected CompletableFuture<Response> oneShot(InetSocketAddress host, ClientRequest request) {
+        Request preparedRequest;
+        try {
+            preparedRequest = prepareRequest(host, request);
+        } catch (IOException e) {
+            return FutureUtil.failedFuture(e);
+        }
+        return executeRequest(preparedRequest);
+    }
+
+    public CompletableFuture<Response> executeRequest(Request request) {
+        return executeRequest(request, () -> new AsyncCompletionHandlerBase());
+    }
+
+    public CompletableFuture<Response> executeRequest(Request request,
+                                                       Supplier<AsyncHandler<Response>> handlerSupplier) {
+        return executeRequest(request, handlerSupplier, 0);
+    }
+
+    private CompletableFuture<Response> executeRequest(Request request,
+                                                       Supplier<AsyncHandler<Response>> handlerSupplier,
+                                                       int redirectCount) {
+        int maxRedirects = httpClient.getConfig().getMaxRedirects();
+        if (redirectCount > maxRedirects) {
+            return FutureUtil.failedFuture(
+                    new MaxRedirectException("Maximum redirect reached: " + maxRedirects + " uri:" + request.getUri()));
+        }
+        CompletableFuture<Response> responseFuture;
+        if (httpClient.getConfig().getMaxConnectionsPerHost() > 0) {
+            String hostAndPort = request.getUri().getHost() + ":" + request.getUri().getPort();
+            ConcurrencyReducer<Response> responseConcurrencyReducer = concurrencyReducers.computeIfAbsent(hostAndPort,
+                    h -> ConcurrencyReducer.create(httpClient.getConfig().getMaxConnectionsPerHost(),
+                            DEFAULT_MAX_QUEUE_SIZE_PER_HOST));
+            responseFuture = responseConcurrencyReducer.add(() -> doExecuteRequest(request, handlerSupplier));
+        } else {
+            responseFuture = doExecuteRequest(request, handlerSupplier);
+        }
+        CompletableFuture<Response> futureWithRedirect = responseFuture.thenCompose(response -> {
+            if (isRedirectStatusCode(response.getStatusCode())) {
+                return executeRedirect(request, response, handlerSupplier, redirectCount);
+            }
+            return CompletableFuture.completedFuture(response);
+        });
+        futureWithRedirect.whenComplete((response, throwable) -> {
+            // propagate cancellation or timeout to the original response future
+            responseFuture.cancel(false);
+        });
+        return futureWithRedirect;
+    }
+
+    private CompletableFuture<Response> executeRedirect(Request request, Response response,
+                                                        Supplier<AsyncHandler<Response>> handlerSupplier,
+                                                        int redirectCount) {
+        String originalMethod = request.getMethod();
+        int statusCode = response.getStatusCode();
+        boolean switchToGet = !originalMethod.equals(GET)
+                && !originalMethod.equals(OPTIONS) && !originalMethod.equals(HEAD) && (
+                statusCode == MOVED_PERMANENTLY_301 || statusCode == SEE_OTHER_303 || statusCode == FOUND_302);
+        boolean keepBody = statusCode == TEMPORARY_REDIRECT_307 || statusCode == PERMANENT_REDIRECT_308;
+        String location = response.getHeader(HttpHeaders.LOCATION);
+        Uri newUri = Uri.create(request.getUri(), location);
+        BoundRequestBuilder builder =
+                httpClient.prepareRequest(request);
+        if (switchToGet) {
+            builder.setMethod(GET);
+        }
+        builder.setUri(newUri);
+        if (keepBody) {
+            builder.setCharset(request.getCharset());
+            if (isNonEmpty(request.getFormParams())) {
+                builder.setFormParams(request.getFormParams());
+            } else if (request.getStringData() != null) {
+                builder.setBody(request.getStringData());
+            } else if (request.getByteData() != null){
+                builder.setBody(request.getByteData());
+            } else if (request.getByteBufferData() != null) {
+                builder.setBody(request.getByteBufferData());
+            } else if (request.getBodyGenerator() != null) {
+                builder.setBody(request.getBodyGenerator());
+            } else if (isNonEmpty(request.getBodyParts())) {
+                builder.setBodyParts(request.getBodyParts());
+            }
+        } else {
+            builder.resetFormParams();
+            builder.resetNonMultipartData();
+            builder.resetMultipartData();
+            io.netty.handler.codec.http.HttpHeaders headers = new DefaultHttpHeaders(false);

Review Comment:
   ## Disabled Netty HTTP header validation
   
   Request splitting or response splitting vulnerability due to header value verification being disabled.
   
   [Show more details](https://github.com/apache/pulsar/security/code-scanning/134)



-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org