Implement Digest Authorization Support

eng/code-quality-reports/src/main/resources/checkstyle/checkstyle-suppressions.xml

@@ -202,6 +202,9 @@
   <!-- Report AMQP retry attempts in static withRetry method. -->
   <suppress checks="com.azure.tools.checkstyle.checks.GoodLoggingCheck" files="com.azure.core.amqp.implementation.RetryUtil.java"/>
 
+  <!-- Throws a non-runtime exception -->
+  <suppress checks="com.azure.tools.checkstyle.checks.ThrowFromClientLogger" files="com.azure.core.http.netty.implementation.ProxyAuthenticationHandler.java"/>
+
   <!-- Event Hubs uses AMQP, which does not contain an HTTP response. Returning PagedResponse and Response does not apply. -->
   <suppress checks="com.azure.tools.checkstyle.checks.ServiceClientCheck" files="com.azure.messaging.eventhubs.EventHubClient.java"/>
   <suppress checks="com.azure.tools.checkstyle.checks.ServiceClientCheck" files="com.azure.messaging.eventhubs.EventHubAsyncClient.java"/>

sdk/core/azure-core-http-netty/pom.xml

@@ -138,12 +138,24 @@
       <version>5.4.2</version> <!-- {x-version-update;org.junit.jupiter:junit-jupiter-engine;external_dependency} -->
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.junit.jupiter</groupId>
+      <artifactId>junit-jupiter-params</artifactId>
+      <version>5.4.2</version> <!-- {x-version-update;org.junit.jupiter:junit-jupiter-params;external_dependency} -->
+      <scope>test</scope>
+    </dependency>
 
     <dependency>
       <groupId>com.github.tomakehurst</groupId>
       <artifactId>wiremock-standalone</artifactId>
       <version>2.24.1</version> <!-- {x-version-update;com.github.tomakehurst:wiremock-standalone;external_dependency} -->
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-core</artifactId>
+      <version>3.0.0</version><!-- {x-version-update;org.mockito:mockito-core;external_dependency} -->
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 </project>

sdk/core/azure-core-http-netty/src/main/java/com/azure/core/http/netty/NettyAsyncHttpClient.java

@@ -9,68 +9,110 @@
 import com.azure.core.http.HttpRequest;
 import com.azure.core.http.HttpResponse;
 import com.azure.core.http.ProxyOptions;
+import com.azure.core.http.netty.implementation.ProxyExceptionHandler;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.handler.codec.http.HttpMethod;
+import io.netty.handler.proxy.ProxyHandler;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.netty.ByteBufFlux;
 import reactor.netty.Connection;
 import reactor.netty.NettyOutbound;
+import reactor.netty.NettyPipeline;
+import reactor.netty.channel.BootstrapHandlers;
 import reactor.netty.http.client.HttpClientRequest;
 import reactor.netty.http.client.HttpClientResponse;
+import reactor.netty.tcp.TcpClient;
 
 import java.nio.ByteBuffer;
 import java.nio.charset.Charset;
 import java.util.Objects;
 import java.util.function.BiFunction;
+import java.util.function.Supplier;
 
 /**
- * This class provides a Netty-based implementation for the {@link HttpClient} interface. Creating an instance of
- * this class can be achieved by using the {@link NettyAsyncHttpClientBuilder} class, which offers Netty-specific API
- * for features such as {@link NettyAsyncHttpClientBuilder#nioEventLoopGroup(NioEventLoopGroup) thread pooling},
- * {@link NettyAsyncHttpClientBuilder#wiretap(boolean) wiretapping},
- * {@link NettyAsyncHttpClientBuilder#proxy(ProxyOptions) setProxy configuration}, and much more.
+ * This class provides a Netty-based implementation for the {@link HttpClient} interface. Creating an instance of this
+ * class can be achieved by using the {@link NettyAsyncHttpClientBuilder} class, which offers Netty-specific API for
+ * features such as {@link NettyAsyncHttpClientBuilder#nioEventLoopGroup(NioEventLoopGroup) thread pooling}, {@link
+ * NettyAsyncHttpClientBuilder#wiretap(boolean) wiretapping}, {@link NettyAsyncHttpClientBuilder#proxy(ProxyOptions)
+ * setProxy configuration}, and much more.
  *
  * @see HttpClient
  * @see NettyAsyncHttpClientBuilder
  */
 class NettyAsyncHttpClient implements HttpClient {
+    private final NioEventLoopGroup eventLoopGroup;
+    private final Supplier<ProxyHandler> proxyHandlerSupplier;
+
     final reactor.netty.http.client.HttpClient nettyClient;
 
     /**
      * Creates default NettyAsyncHttpClient.
      */
     NettyAsyncHttpClient() {
-        this(reactor.netty.http.client.HttpClient.create());
+        this(reactor.netty.http.client.HttpClient.create(), null, null);
     }
 
     /**
      * Creates NettyAsyncHttpClient with provided http client.
      *
      * @param nettyClient the reactor-netty http client
+     * @param eventLoopGroup {@link NioEventLoopGroup} that processes requests.
+     * @param proxyHandlerSupplier Supplier that returns the {@link ProxyHandler} that connects to the configured
+     * proxy.
      */
-    NettyAsyncHttpClient(reactor.netty.http.client.HttpClient nettyClient) {
+    NettyAsyncHttpClient(reactor.netty.http.client.HttpClient nettyClient, NioEventLoopGroup eventLoopGroup,
+        Supplier<ProxyHandler> proxyHandlerSupplier) {
         this.nettyClient = nettyClient;
+        this.eventLoopGroup = eventLoopGroup;
+        this.proxyHandlerSupplier = proxyHandlerSupplier;
     }
 
-    /** {@inheritDoc} */
+    /**
+     * {@inheritDoc}
+     */
     @Override
     public Mono<HttpResponse> send(final HttpRequest request) {
         Objects.requireNonNull(request.getHttpMethod(), "'request.getHttpMethod()' cannot be null.");
         Objects.requireNonNull(request.getUrl(), "'request.getUrl()' cannot be null.");
         Objects.requireNonNull(request.getUrl().getProtocol(), "'request.getUrl().getProtocol()' cannot be null.");
 
         return nettyClient
+            .tcpConfiguration(this::configureTcpClient)
             .request(HttpMethod.valueOf(request.getHttpMethod().toString()))
             .uri(request.getUrl().toString())
             .send(bodySendDelegate(request))
             .responseConnection(responseDelegate(request))
             .single();
     }
 
+    /*
+     * Configures the underlying TcpClient that sends the request.
+     */
+    private TcpClient configureTcpClient(TcpClient tcpClient) {
+        if (eventLoopGroup != null) {
+            tcpClient = tcpClient.runOn(eventLoopGroup);
+        }
+
+        ProxyHandler proxyHandler = (proxyHandlerSupplier == null) ? null : proxyHandlerSupplier.get();
+        if (proxyHandler != null) {
+            /*
+             * Configure the request Channel to be initialized with a ProxyHandler. The ProxyHandler is the first
+             * operation in the pipeline as it needs to handle sending a CONNECT request to the proxy before any request
+             * data is sent.
+             */
+            tcpClient = tcpClient.bootstrap(bootstrap -> BootstrapHandlers
+                .updateConfiguration(bootstrap, NettyPipeline.ProxyHandler, (connectionObserver, channel) ->
+                    channel.pipeline().addFirst(NettyPipeline.ProxyHandler, proxyHandler)
+                        .addLast("azure.proxy.exceptionHandler", new ProxyExceptionHandler())));
+        }
+
+        return tcpClient;
+    }
+
     /**
      * Delegate to send the request content.
      *
@@ -111,7 +153,7 @@ static class ReactorNettyHttpResponse extends HttpResponse {
         private final Connection reactorNettyConnection;
 
         ReactorNettyHttpResponse(HttpClientResponse reactorNettyResponse, Connection reactorNettyConnection,
-                                 HttpRequest httpRequest) {
+            HttpRequest httpRequest) {
             super(httpRequest);
             this.reactorNettyResponse = reactorNettyResponse;
             this.reactorNettyConnection = reactorNettyConnection;

sdk/core/azure-core-http-netty/src/main/java/com/azure/core/http/netty/NettyAsyncHttpClientBuilder.java

@@ -3,14 +3,20 @@
 
 package com.azure.core.http.netty;
 
+import com.azure.core.util.AuthorizationChallengeHandler;
 import com.azure.core.http.ProxyOptions;
+import com.azure.core.http.netty.implementation.ChallengeHolder;
+import com.azure.core.http.netty.implementation.ProxyAuthenticationHandler;
 import com.azure.core.util.logging.ClientLogger;
 import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.handler.proxy.ProxyHandler;
+import io.netty.handler.proxy.Socks4ProxyHandler;
+import io.netty.handler.proxy.Socks5ProxyHandler;
 import reactor.netty.http.client.HttpClient;
 import reactor.netty.resources.ConnectionProvider;
-import reactor.netty.tcp.ProxyProvider;
 
 import java.util.Objects;
+import java.util.concurrent.atomic.AtomicReference;
 
 /**
  * Builder class responsible for creating instances of {@link NettyAsyncHttpClient}.
@@ -23,6 +29,8 @@
  * @see HttpClient
  */
 public class NettyAsyncHttpClientBuilder {
+    private static final String INVALID_PROXY_MESSAGE = "Unknown Proxy type '%s' in use. Not configuring Netty proxy.";
+
     private final ClientLogger logger = new ClientLogger(NettyAsyncHttpClientBuilder.class);
 
     private final HttpClient baseHttpClient;
@@ -61,54 +69,25 @@ public NettyAsyncHttpClientBuilder(HttpClient nettyHttpClient) {
      */
     public com.azure.core.http.HttpClient build() {
         HttpClient nettyHttpClient;
-        if (this.connectionProvider != null) {
-            if (this.baseHttpClient != null) {
-                throw logger.logExceptionAsError(new IllegalStateException("connectionProvider cannot be set on an "
-                    + "existing reactor netty HttpClient."));
-            }
+        if (this.baseHttpClient != null) {
+            nettyHttpClient = baseHttpClient;
+        } else if (this.connectionProvider != null) {
             nettyHttpClient = HttpClient.create(this.connectionProvider);
         } else {
-            nettyHttpClient = this.baseHttpClient == null ? HttpClient.create() : this.baseHttpClient;
+            nettyHttpClient = HttpClient.create();
         }
+
         nettyHttpClient = nettyHttpClient
             .port(port)
-            .wiretap(enableWiretap)
-            .tcpConfiguration(tcpConfig -> {
-                if (nioEventLoopGroup != null) {
-                    tcpConfig = tcpConfig.runOn(nioEventLoopGroup);
-                }
-
-                if (proxyOptions != null) {
-                    ProxyProvider.Proxy nettyProxy;
-                    switch (proxyOptions.getType()) {
-                        case HTTP:
-                            nettyProxy = ProxyProvider.Proxy.HTTP;
-                            break;
-                        case SOCKS4:
-                            nettyProxy = ProxyProvider.Proxy.SOCKS4;
-                            break;
-                        case SOCKS5:
-                            nettyProxy = ProxyProvider.Proxy.SOCKS5;
-                            break;
-                        default:
-                            throw logger.logExceptionAsError(new IllegalStateException(
-                                String.format("Unknown Proxy type '%s' in use. Not configuring Netty proxy.",
-                                    proxyOptions.getType())));
-                    }
-                    if (proxyOptions.getUsername() != null) {
-                        // Netty supports only Basic proxy authentication and we default to it.
-                        return tcpConfig.proxy(ts -> ts.type(nettyProxy)
-                                .address(proxyOptions.getAddress())
-                                .username(proxyOptions.getUsername())
-                                .password(userName -> proxyOptions.getPassword())
-                                .build());
-                    } else {
-                        return tcpConfig.proxy(ts -> ts.type(nettyProxy).address(proxyOptions.getAddress()));
-                    }
-                }
-                return tcpConfig;
-            });
-        return new NettyAsyncHttpClient(nettyHttpClient);
+            .wiretap(enableWiretap);
+
+        AuthorizationChallengeHandler challengeHandler = (proxyOptions == null)
+            ? null
+            : new AuthorizationChallengeHandler(proxyOptions.getUsername(), proxyOptions.getPassword());
+        AtomicReference<ChallengeHolder> proxyChallengeHolder = new AtomicReference<>();
+
+        return new NettyAsyncHttpClient(nettyHttpClient, nioEventLoopGroup,
+            () -> getProxyHandler(challengeHandler, proxyChallengeHolder));
     }
 
     /**
@@ -122,6 +101,7 @@ public NettyAsyncHttpClientBuilder connectionProvider(ConnectionProvider connect
         this.connectionProvider = connectionProvider;
         return this;
     }
+
     /**
      * Sets the {@link ProxyOptions proxy options} that the client will use.
      *
@@ -174,4 +154,28 @@ public NettyAsyncHttpClientBuilder nioEventLoopGroup(NioEventLoopGroup nioEventL
         this.nioEventLoopGroup = nioEventLoopGroup;
         return this;
     }
+
+    /*
+     * Creates a proxy handler based on the passed ProxyOptions.
+     */
+    private ProxyHandler getProxyHandler(AuthorizationChallengeHandler challengeHandler,
+        AtomicReference<ChallengeHolder> proxyChallengeHolder) {
+        if (proxyOptions == null) {
+            return null;
+        }
+
+        switch (proxyOptions.getType()) {
+            case HTTP:
+                return new ProxyAuthenticationHandler(proxyOptions.getAddress(), challengeHandler,
+                    proxyChallengeHolder);
+            case SOCKS4:
+                return new Socks4ProxyHandler(proxyOptions.getAddress(), proxyOptions.getUsername());
+            case SOCKS5:
+                return new Socks5ProxyHandler(proxyOptions.getAddress(), proxyOptions.getUsername(),
+                    proxyOptions.getPassword());
+            default:
+                throw logger.logExceptionAsError(new IllegalStateException(
+                    String.format(INVALID_PROXY_MESSAGE, proxyOptions.getType())));
+        }
+    }
 }

sdk/core/azure-core-http-netty/src/main/java/com/azure/core/http/netty/implementation/ChallengeHolder.java

@@ -0,0 +1,44 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.core.http.netty.implementation;
+
+import com.azure.core.annotation.Immutable;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Model class that contains the authentication challenges returned by a server.
+ */
+@Immutable
+public final class ChallengeHolder {
+    private final boolean hasBasicChallenge;
+    private final List<Map<String, String>> digestChallenges;
+
+    /**
+     * Creates a {@link ChallengeHolder} which contains the parsed authentication digest challenges and a flag
+     * indicating if basic authorization is accepted.
+     *
+     * @param hasBasicChallenge Flag indicating if basic authorization is accepted.
+     * @param digestChallenges Parsed digest challenges.
+     */
+    public ChallengeHolder(boolean hasBasicChallenge, List<Map<String, String>> digestChallenges) {
+        this.hasBasicChallenge = hasBasicChallenge;
+        this.digestChallenges = digestChallenges;
+    }
+
+    /**
+     * @return Flag indicating if basic authorization is accepted.
+     */
+    public boolean hasBasicChallenge() {
+        return hasBasicChallenge;
+    }
+
+    /**
+     * @return The parsed digest challenges.
+     */
+    public List<Map<String, String>> getDigestChallenges() {
+        return digestChallenges;
+    }
+}