1 /*

     2  * Copyright (c) 2015, 2018, Oracle and/or its affiliates. All rights reserved.

     3  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

     4  *

     5  * This code is free software; you can redistribute it and/or modify it

     6  * under the terms of the GNU General Public License version 2 only, as

     7  * published by the Free Software Foundation.  Oracle designates this

     8  * particular file as subject to the "Classpath" exception as provided

     9  * by Oracle in the LICENSE file that accompanied this code.

    10  *

    11  * This code is distributed in the hope that it will be useful, but WITHOUT

    12  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

    13  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License

    14  * version 2 for more details (a copy is included in the LICENSE file that

    15  * accompanied this code).

    16  *

    17  * You should have received a copy of the GNU General Public License version

    18  * 2 along with this work; if not, write to the Free Software Foundation,

    19  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

    20  *

    21  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA

    22  * or visit www.oracle.com if you need additional information or have any

    23  * questions.

    24  */

    25 

    26 package jdk.incubator.http.internal;

    27 

    28 import java.io.IOException;

    29 import java.lang.System.Logger.Level;

    30 import java.net.InetSocketAddress;

    31 import java.net.ProxySelector;

    32 import java.net.URI;

    33 import java.net.URISyntaxException;

    34 import java.net.URLPermission;

    35 import java.security.AccessControlContext;

    36 import java.util.List;

    37 import java.util.Map;

    38 import java.util.concurrent.CompletableFuture;

    39 import java.util.concurrent.Executor;

    40 import java.util.function.Function;

    41 import jdk.incubator.http.HttpClient;

    42 import jdk.incubator.http.HttpHeaders;

    43 import jdk.incubator.http.HttpResponse;

    44 import jdk.incubator.http.HttpTimeoutException;

    45 import jdk.incubator.http.internal.common.MinimalFuture;

    46 import jdk.incubator.http.internal.common.Utils;

    47 import jdk.incubator.http.internal.common.Log;

    48 

    49 import static jdk.incubator.http.internal.common.Utils.permissionForProxy;

    50 

    51 /**

    52  * One request/response exchange (handles 100/101 intermediate response also).

    53  * depth field used to track number of times a new request is being sent

    54  * for a given API request. If limit exceeded exception is thrown.

    55  *

    56  * Security check is performed here:

    57  * - uses AccessControlContext captured at API level

    58  * - checks for appropriate URLPermission for request

    59  * - if permission allowed, grants equivalent SocketPermission to call

    60  * - in case of direct HTTP proxy, checks additionally for access to proxy

    61  *    (CONNECT proxying uses its own Exchange, so check done there)

    62  *

    63  */

    64 final class Exchange<T> {

    65 

    66     static final boolean DEBUG = Utils.DEBUG; // Revisit: temporary dev flag.

    67     final System.Logger  debug = Utils.getDebugLogger(this::dbgString, DEBUG);

    68 

    69     final HttpRequestImpl request;

    70     final HttpClientImpl client;

    71     volatile ExchangeImpl<T> exchImpl;

    72     volatile CompletableFuture<? extends ExchangeImpl<T>> exchangeCF;

    73     volatile CompletableFuture<Void> bodyIgnored;

    74 

    75     // used to record possible cancellation raised before the exchImpl

    76     // has been established.

    77     private volatile IOException failed;

    78     final AccessControlContext acc;

    79     final MultiExchange<T> multi;

    80     final Executor parentExecutor;

    81     boolean upgrading; // to HTTP/2

    82     final PushGroup<T> pushGroup;

    83     final String dbgTag;

    84 

    85     Exchange(HttpRequestImpl request, MultiExchange<T> multi) {

    86         this.request = request;

    87         this.upgrading = false;

    88         this.client = multi.client();

    89         this.multi = multi;

    90         this.acc = multi.acc;

    91         this.parentExecutor = multi.executor;

    92         this.pushGroup = multi.pushGroup;

    93         this.dbgTag = "Exchange";

    94     }

    95 

    96     /* If different AccessControlContext to be used  */

    97     Exchange(HttpRequestImpl request,

    98              MultiExchange<T> multi,

    99              AccessControlContext acc)

   100     {

   101         this.request = request;

   102         this.acc = acc;

   103         this.upgrading = false;

   104         this.client = multi.client();

   105         this.multi = multi;

   106         this.parentExecutor = multi.executor;

   107         this.pushGroup = multi.pushGroup;

   108         this.dbgTag = "Exchange";

   109     }

   110 

   111     PushGroup<T> getPushGroup() {

   112         return pushGroup;

   113     }

   114 

   115     Executor executor() {

   116         return parentExecutor;

   117     }

   118 

   119     public HttpRequestImpl request() {

   120         return request;

   121     }

   122 

   123     HttpClientImpl client() {

   124         return client;

   125     }

   126 

   127 

   128     public CompletableFuture<T> readBodyAsync(HttpResponse.BodyHandler<T> handler) {

   129         // If we received a 407 while establishing the exchange

   130         // there will be no body to read: bodyIgnored will be true,

   131         // and exchImpl will be null (if we were trying to establish

   132         // an HTTP/2 tunnel through an HTTP/1.1 proxy)

   133         if (bodyIgnored != null) return MinimalFuture.completedFuture(null);

   134 

   135         // The connection will not be returned to the pool in the case of WebSocket

   136         return exchImpl.readBodyAsync(handler, !request.isWebSocket(), parentExecutor)

   137                 .whenComplete((r,t) -> exchImpl.completed());

   138     }

   139 

   140     /**

   141      * Called after a redirect or similar kind of retry where a body might

   142      * be sent but we don't want it. Should send a RESET in h2. For http/1.1

   143      * we can consume small quantity of data, or close the connection in

   144      * other cases.

   145      */

   146     public CompletableFuture<Void> ignoreBody() {

   147         if (bodyIgnored != null) return bodyIgnored;

   148         return exchImpl.ignoreBody();

   149     }

   150 

   151     /**

   152      * Called when a new exchange is created to replace this exchange.

   153      * At this point it is guaranteed that readBody/readBodyAsync will

   154      * not be called.

   155      */

   156     public void released() {

   157         ExchangeImpl<?> impl = exchImpl;

   158         if (impl != null) impl.released();

   159         // Don't set exchImpl to null here. We need to keep

   160         // it alive until it's replaced by a Stream in wrapForUpgrade.

   161         // Setting it to null here might get it GC'ed too early, because

   162         // the Http1Response is now only weakly referenced by the Selector.

   163     }

   164 

   165     public void cancel() {

   166         // cancel can be called concurrently before or at the same time

   167         // that the exchange impl is being established.

   168         // In that case we won't be able to propagate the cancellation

   169         // right away

   170         if (exchImpl != null) {

   171             exchImpl.cancel();

   172         } else {

   173             // no impl - can't cancel impl yet.

   174             // call cancel(IOException) instead which takes care

   175             // of race conditions between impl/cancel.

   176             cancel(new IOException("Request cancelled"));

   177         }

   178     }

   179 

   180     public void cancel(IOException cause) {

   181         // If the impl is non null, propagate the exception right away.

   182         // Otherwise record it so that it can be propagated once the

   183         // exchange impl has been established.

   184         ExchangeImpl<?> impl = exchImpl;

   185         if (impl != null) {

   186             // propagate the exception to the impl

   187             debug.log(Level.DEBUG, "Cancelling exchImpl: %s", exchImpl);

   188             impl.cancel(cause);

   189         } else {

   190             // no impl yet. record the exception

   191             failed = cause;

   192             // now call checkCancelled to recheck the impl.

   193             // if the failed state is set and the impl is not null, reset

   194             // the failed state and propagate the exception to the impl.

   195             checkCancelled();

   196         }

   197     }

   198 

   199     // This method will raise an exception if one was reported and if

   200     // it is possible to do so. If the exception can be raised, then

   201     // the failed state will be reset. Otherwise, the failed state

   202     // will persist until the exception can be raised and the failed state

   203     // can be cleared.

   204     // Takes care of possible race conditions.

   205     private void checkCancelled() {

   206         ExchangeImpl<?> impl = null;

   207         IOException cause = null;

   208         CompletableFuture<? extends ExchangeImpl<T>> cf = null;

   209         if (failed != null) {

   210             synchronized(this) {

   211                 cause = failed;

   212                 impl = exchImpl;

   213                 cf = exchangeCF;

   214             }

   215         }

   216         if (cause == null) return;

   217         if (impl != null) {

   218             // The exception is raised by propagating it to the impl.

   219             debug.log(Level.DEBUG, "Cancelling exchImpl: %s", impl);

   220             impl.cancel(cause);

   221             failed = null;

   222         } else {

   223             Log.logTrace("Exchange: request [{0}/timeout={1}ms] no impl is set."

   224                          + "\n\tCan''t cancel yet with {2}",

   225                          request.uri(),

   226                          request.timeout().isPresent() ?

   227                          // calling duration.toMillis() can throw an exception.

   228                          // this is just debugging, we don't care if it overflows.

   229                          (request.timeout().get().getSeconds() * 1000

   230                           + request.timeout().get().getNano() / 1000000) : -1,

   231                          cause);

   232             if (cf != null) cf.completeExceptionally(cause);

   233         }

   234     }

   235 

   236     public void h2Upgrade() {

   237         upgrading = true;

   238         request.setH2Upgrade(client.client2());

   239     }

   240 

   241     synchronized IOException getCancelCause() {

   242         return failed;

   243     }

   244 

   245     // get/set the exchange impl, solving race condition issues with

   246     // potential concurrent calls to cancel() or cancel(IOException)

   247     private CompletableFuture<? extends ExchangeImpl<T>>

   248     establishExchange(HttpConnection connection) {

   249         if (debug.isLoggable(Level.DEBUG)) {

   250             debug.log(Level.DEBUG,

   251                     "establishing exchange for %s,%n\t proxy=%s",

   252                     request,

   253                     request.proxy());

   254         }

   255         // check if we have been cancelled first.

   256         Throwable t = getCancelCause();

   257         checkCancelled();

   258         if (t != null) {

   259             return MinimalFuture.failedFuture(t);

   260         }

   261 

   262         CompletableFuture<? extends ExchangeImpl<T>> cf, res;

   263         cf = ExchangeImpl.get(this, connection);

   264         // We should probably use a VarHandle to get/set exchangeCF

   265         // instead - as we need CAS semantics.

   266         synchronized (this) { exchangeCF = cf; };

   267         res = cf.whenComplete((r,x) -> {

   268             synchronized(Exchange.this) {

   269                 if (exchangeCF == cf) exchangeCF = null;

   270             }

   271         });

   272         checkCancelled();

   273         return res.thenCompose((eimpl) -> {

   274                     // recheck for cancelled, in case of race conditions

   275                     exchImpl = eimpl;

   276                     IOException tt = getCancelCause();

   277                     checkCancelled();

   278                     if (tt != null) {

   279                         return MinimalFuture.failedFuture(tt);

   280                     } else {

   281                         // Now we're good to go. Because exchImpl is no longer

   282                         // null cancel() will be able to propagate directly to

   283                         // the impl after this point ( if needed ).

   284                         return MinimalFuture.completedFuture(eimpl);

   285                     } });

   286     }

   287 

   288     // Completed HttpResponse will be null if response succeeded

   289     // will be a non null responseAsync if expect continue returns an error

   290 

   291     public CompletableFuture<Response> responseAsync() {

   292         return responseAsyncImpl(null);

   293     }

   294 

   295     CompletableFuture<Response> responseAsyncImpl(HttpConnection connection) {

   296         SecurityException e = checkPermissions();

   297         if (e != null) {

   298             return MinimalFuture.failedFuture(e);

   299         } else {

   300             return responseAsyncImpl0(connection);

   301         }

   302     }

   303 

   304     // check whether the headersSentCF was completed exceptionally with

   305     // ProxyAuthorizationRequired. If so the Response embedded in the

   306     // exception is returned. Otherwise we proceed.

   307     private CompletableFuture<Response> checkFor407(ExchangeImpl<T> ex, Throwable t,

   308                                                     Function<ExchangeImpl<T>,CompletableFuture<Response>> andThen) {

   309         t = Utils.getCompletionCause(t);

   310         if (t instanceof ProxyAuthenticationRequired) {

   311             bodyIgnored = MinimalFuture.completedFuture(null);

   312             Response proxyResponse = ((ProxyAuthenticationRequired)t).proxyResponse;

   313             Response syntheticResponse = new Response(request, this,

   314                     proxyResponse.headers, proxyResponse.statusCode,

   315                     proxyResponse.version, true);

   316             return MinimalFuture.completedFuture(syntheticResponse);

   317         } else if (t != null) {

   318             return MinimalFuture.failedFuture(t);

   319         } else {

   320             return andThen.apply(ex);

   321         }

   322     }

   323 

   324     // After sending the request headers, if no ProxyAuthorizationRequired

   325     // was raised and the expectContinue flag is on, we need to wait

   326     // for the 100-Continue response

   327     private CompletableFuture<Response> expectContinue(ExchangeImpl<T> ex) {

   328         assert request.expectContinue();

   329         return ex.getResponseAsync(parentExecutor)

   330                 .thenCompose((Response r1) -> {

   331             Log.logResponse(r1::toString);

   332             int rcode = r1.statusCode();

   333             if (rcode == 100) {

   334                 Log.logTrace("Received 100-Continue: sending body");

   335                 CompletableFuture<Response> cf =

   336                         exchImpl.sendBodyAsync()

   337                                 .thenCompose(exIm -> exIm.getResponseAsync(parentExecutor));

   338                 cf = wrapForUpgrade(cf);

   339                 cf = wrapForLog(cf);

   340                 return cf;

   341             } else {

   342                 Log.logTrace("Expectation failed: Received {0}",

   343                         rcode);

   344                 if (upgrading && rcode == 101) {

   345                     IOException failed = new IOException(

   346                             "Unable to handle 101 while waiting for 100");

   347                     return MinimalFuture.failedFuture(failed);

   348                 }

   349                 return exchImpl.readBodyAsync(this::ignoreBody, false, parentExecutor)

   350                         .thenApply(v ->  r1);

   351             }

   352         });

   353     }

   354 

   355     // After sending the request headers, if no ProxyAuthorizationRequired

   356     // was raised and the expectContinue flag is off, we can immediately

   357     // send the request body and proceed.

   358     private CompletableFuture<Response> sendRequestBody(ExchangeImpl<T> ex) {

   359         assert !request.expectContinue();

   360         CompletableFuture<Response> cf = ex.sendBodyAsync()

   361                 .thenCompose(exIm -> exIm.getResponseAsync(parentExecutor));

   362         cf = wrapForUpgrade(cf);

   363         cf = wrapForLog(cf);

   364         return cf;

   365     }

   366 

   367     CompletableFuture<Response> responseAsyncImpl0(HttpConnection connection) {

   368         Function<ExchangeImpl<T>, CompletableFuture<Response>> after407Check;

   369         bodyIgnored = null;

   370         if (request.expectContinue()) {

   371             request.addSystemHeader("Expect", "100-Continue");

   372             Log.logTrace("Sending Expect: 100-Continue");

   373             // wait for 100-Continue before sending body

   374             after407Check = this::expectContinue;

   375         } else {

   376             // send request body and proceed.

   377             after407Check = this::sendRequestBody;

   378         }

   379         // The ProxyAuthorizationRequired can be triggered either by

   380         // establishExchange (case of HTTP/2 SSL tunelling through HTTP/1.1 proxy

   381         // or by sendHeaderAsync (case of HTTP/1.1 SSL tunelling through HTTP/1.1 proxy

   382         // Therefore we handle it with a call to this checkFor407(...) after these

   383         // two places.

   384         Function<ExchangeImpl<T>, CompletableFuture<Response>> afterExch407Check =

   385                 (ex) -> ex.sendHeadersAsync()

   386                         .handle((r,t) -> this.checkFor407(r, t, after407Check))

   387                         .thenCompose(Function.identity());

   388         return establishExchange(connection)

   389                 .handle((r,t) -> this.checkFor407(r,t, afterExch407Check))

   390                 .thenCompose(Function.identity());

   391     }

   392 

   393     private CompletableFuture<Response> wrapForUpgrade(CompletableFuture<Response> cf) {

   394         if (upgrading) {

   395             return cf.thenCompose(r -> checkForUpgradeAsync(r, exchImpl));

   396         }

   397         return cf;

   398     }

   399 

   400     private CompletableFuture<Response> wrapForLog(CompletableFuture<Response> cf) {

   401         if (Log.requests()) {

   402             return cf.thenApply(response -> {

   403                 Log.logResponse(response::toString);

   404                 return response;

   405             });

   406         }

   407         return cf;

   408     }

   409 

   410     HttpResponse.BodySubscriber<T> ignoreBody(int status, HttpHeaders hdrs) {

   411         return HttpResponse.BodySubscriber.replace(null);

   412     }

   413 

   414     // if this response was received in reply to an upgrade

   415     // then create the Http2Connection from the HttpConnection

   416     // initialize it and wait for the real response on a newly created Stream

   417 

   418     private CompletableFuture<Response>

   419     checkForUpgradeAsync(Response resp,

   420                          ExchangeImpl<T> ex) {

   421 

   422         int rcode = resp.statusCode();

   423         if (upgrading && (rcode == 101)) {

   424             Http1Exchange<T> e = (Http1Exchange<T>)ex;

   425             // check for 101 switching protocols

   426             // 101 responses are not supposed to contain a body.

   427             //    => should we fail if there is one?

   428             debug.log(Level.DEBUG, "Upgrading async %s", e.connection());

   429             return e.readBodyAsync(this::ignoreBody, false, parentExecutor)

   430                 .thenCompose((T v) -> {// v is null

   431                     debug.log(Level.DEBUG, "Ignored body");

   432                     // we pass e::getBuffer to allow the ByteBuffers to accumulate

   433                     // while we build the Http2Connection

   434                     return Http2Connection.createAsync(e.connection(),

   435                                                  client.client2(),

   436                                                  this, e::drainLeftOverBytes)

   437                         .thenCompose((Http2Connection c) -> {

   438                             boolean cached = c.offerConnection();

   439                             Stream<T> s = c.getStream(1);

   440 

   441                             if (s == null) {

   442                                 // s can be null if an exception occurred

   443                                 // asynchronously while sending the preface.

   444                                 Throwable t = c.getRecordedCause();

   445                                 IOException ioe;

   446                                 if (t != null) {

   447                                     if (!cached)

   448                                         c.close();

   449                                     ioe = new IOException("Can't get stream 1: " + t, t);

   450                                 } else {

   451                                     ioe = new IOException("Can't get stream 1");

   452                                 }

   453                                 return MinimalFuture.failedFuture(ioe);

   454                             }

   455                             exchImpl.released();

   456                             Throwable t;

   457                             // There's a race condition window where an external

   458                             // thread (SelectorManager) might complete the

   459                             // exchange in timeout at the same time where we're

   460                             // trying to switch the exchange impl.

   461                             // 'failed' will be reset to null after

   462                             // exchImpl.cancel() has completed, so either we

   463                             // will observe failed != null here, or we will

   464                             // observe e.getCancelCause() != null, or the

   465                             // timeout exception will be routed to 's'.

   466                             // Either way, we need to relay it to s.

   467                             synchronized (this) {

   468                                 exchImpl = s;

   469                                 t = failed;

   470                             }

   471                             // Check whether the HTTP/1.1 was cancelled.

   472                             if (t == null) t = e.getCancelCause();

   473                             // if HTTP/1.1 exchange was timed out, don't

   474                             // try to go further.

   475                             if (t instanceof HttpTimeoutException) {

   476                                  s.cancelImpl(t);

   477                                  return MinimalFuture.failedFuture(t);

   478                             }

   479                             debug.log(Level.DEBUG, "Getting response async %s", s);

   480                             return s.getResponseAsync(null);

   481                         });}

   482                 );

   483         }

   484         return MinimalFuture.completedFuture(resp);

   485     }

   486 

   487     private URI getURIForSecurityCheck() {

   488         URI u;

   489         String method = request.method();

   490         InetSocketAddress authority = request.authority();

   491         URI uri = request.uri();

   492 

   493         // CONNECT should be restricted at API level

   494         if (method.equalsIgnoreCase("CONNECT")) {

   495             try {

   496                 u = new URI("socket",

   497                              null,

   498                              authority.getHostString(),

   499                              authority.getPort(),

   500                              null,

   501                              null,

   502                              null);

   503             } catch (URISyntaxException e) {

   504                 throw new InternalError(e); // shouldn't happen

   505             }

   506         } else {

   507             u = uri;

   508         }

   509         return u;

   510     }

   511 

   512     /**

   513      * Returns the security permission required for the given details.

   514      * If method is CONNECT, then uri must be of form "scheme://host:port"

   515      */

   516     private static URLPermission permissionForServer(URI uri,

   517                                                      String method,

   518                                                      Map<String, List<String>> headers) {

   519         if (method.equals("CONNECT")) {

   520             return new URLPermission(uri.toString(), "CONNECT");

   521         } else {

   522             return Utils.permissionForServer(uri, method, headers.keySet().stream());

   523         }

   524     }

   525 

   526     /**

   527      * Performs the necessary security permission checks required to retrieve

   528      * the response. Returns a security exception representing the denied

   529      * permission, or null if all checks pass or there is no security manager.

   530      */

   531     private SecurityException checkPermissions() {

   532         String method = request.method();

   533         SecurityManager sm = System.getSecurityManager();

   534         if (sm == null || method.equals("CONNECT")) {

   535             // tunneling will have a null acc, which is fine. The proxy

   536             // permission check will have already been preformed.

   537             return null;

   538         }

   539 

   540         HttpHeaders userHeaders = request.getUserHeaders();

   541         URI u = getURIForSecurityCheck();

   542         URLPermission p = permissionForServer(u, method, userHeaders.map());

   543 

   544         try {

   545             assert acc != null;

   546             sm.checkPermission(p, acc);

   547         } catch (SecurityException e) {

   548             return e;

   549         }

   550         ProxySelector ps = client.proxySelector();

   551         if (ps != null) {

   552             if (!method.equals("CONNECT")) {

   553                 // a non-tunneling HTTP proxy. Need to check access

   554                 URLPermission proxyPerm = permissionForProxy(request.proxy());

   555                 if (proxyPerm != null) {

   556                     try {

   557                         sm.checkPermission(proxyPerm, acc);

   558                     } catch (SecurityException e) {

   559                         return e;

   560                     }

   561                 }

   562             }

   563         }

   564         return null;

   565     }

   566 

   567     HttpClient.Version version() {

   568         return multi.version();

   569     }

   570 

   571     String dbgString() {

   572         return dbgTag;

   573     }

   574 }