1 /*
2 * Copyright (c) 2015, 2016, 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;
27
28 import java.io.IOException;
29 import java.nio.ByteBuffer;
30 import java.util.ArrayList;
31 import java.util.Arrays;
32 import java.util.List;
33 import java.util.concurrent.Semaphore;
34 import java.util.concurrent.CompletableFuture;
35 import java.util.function.Consumer;
36 import java.util.function.Supplier;
37
38 import static javax.net.ssl.SSLEngineResult.Status.*;
39 import javax.net.ssl.*;
40
41 import jdk.incubator.http.internal.common.AsyncWriteQueue;
42 import jdk.incubator.http.internal.common.ByteBufferPool;
43 import jdk.incubator.http.internal.common.ByteBufferReference;
44 import jdk.incubator.http.internal.common.Log;
45 import jdk.incubator.http.internal.common.Queue;
46 import jdk.incubator.http.internal.common.Utils;
47 import static javax.net.ssl.SSLEngineResult.HandshakeStatus.*;
48 import jdk.incubator.http.internal.common.ExceptionallyCloseable;
49
50 /**
51 * Asynchronous wrapper around SSLEngine. send and receive is fully non
52 * blocking. When handshaking is required, a thread is created to perform
53 * the handshake and application level sends do not take place during this time.
54 *
55 * Is implemented using queues and functions operating on the receiving end
56 * of each queue.
57 *
58 * Application writes to:
59 * ||
60 * \/
61 * appOutputQ
62 * ||
63 * \/
64 * appOutputQ read by "upperWrite" method which does SSLEngine.wrap
65 * and does async write to PlainHttpConnection
66 *
67 * Reading side is as follows
68 * --------------------------
69 *
70 * "upperRead" method reads off channelInputQ and calls SSLEngine.unwrap and
71 * when decrypted data is returned, it is passed to the user's Consumer<ByteBuffer>
72 * /\
73 * ||
74 * channelInputQ
75 * /\
76 * ||
77 * "asyncReceive" method puts buffers into channelInputQ. It is invoked from
78 * OP_READ events from the selector.
79 *
80 * Whenever handshaking is required, the doHandshaking() method is called
81 * which creates a thread to complete the handshake. It takes over the
82 * channelInputQ from upperRead, and puts outgoing packets on channelOutputQ.
83 * Selector events are delivered to asyncReceive and lowerWrite as normal.
84 *
85 * Errors
86 *
87 * Any exception thrown by the engine or channel, causes all Queues to be closed
88 * the channel to be closed, and the error is reported to the user's
89 * Consumer<Throwable>
90 */
91 class AsyncSSLDelegate implements ExceptionallyCloseable, AsyncConnection {
92
93 // outgoing buffers put in this queue first and may remain here
94 // while SSL handshaking happening.
95 final AsyncWriteQueue appOutputQ = new AsyncWriteQueue(this::upperWrite);
96
97 // Bytes read into this queue before being unwrapped. Backup on this
98 // Q should only happen when the engine is stalled due to delegated tasks
99 final Queue<ByteBufferReference> channelInputQ;
100
101 // input occurs through the read() method which is expected to be called
102 // when the selector signals some data is waiting to be read. All incoming
103 // handshake data is handled in this method, which means some calls to
104 // read() may return zero bytes of user data. This is not a sign of spinning,
105 // just that the handshake mechanics are being executed.
106
107 final SSLEngine engine;
108 final SSLParameters sslParameters;
109 final HttpConnection lowerOutput;
110 final HttpClientImpl client;
111 final String serverName;
112 // should be volatile to provide proper synchronization(visibility) action
113 volatile Consumer<ByteBufferReference> asyncReceiver;
114 volatile Consumer<Throwable> errorHandler;
115 volatile boolean connected = false;
116
117 // Locks.
118 final Object reader = new Object();
119 // synchronizing handshake state
120 final Semaphore handshaker = new Semaphore(1);
121 final String[] alpn;
122
123 // alpn[] may be null. upcall is callback which receives incoming decoded bytes off socket
124
125 AsyncSSLDelegate(HttpConnection lowerOutput, HttpClientImpl client, String[] alpn, String sname)
126 {
127 SSLContext context = client.sslContext();
128 this.serverName = sname;
129 engine = context.createSSLEngine();
130 engine.setUseClientMode(true);
131 SSLParameters sslp = client.sslParameters()
132 .orElseGet(context::getSupportedSSLParameters);
133 sslParameters = Utils.copySSLParameters(sslp);
134 if (alpn != null) {
135 Log.logSSL("AsyncSSLDelegate: Setting application protocols: " + Arrays.toString(alpn));
136 sslParameters.setApplicationProtocols(alpn);
137 } else {
138 Log.logSSL("AsyncSSLDelegate: no applications set!");
139 }
140 if (serverName != null) {
141 SNIHostName sn = new SNIHostName(serverName);
142 sslParameters.setServerNames(List.of(sn));
143 }
144 logParams(sslParameters);
145 engine.setSSLParameters(sslParameters);
146 this.lowerOutput = lowerOutput;
147 this.client = client;
148 this.channelInputQ = new Queue<>();
149 this.channelInputQ.registerPutCallback(this::upperRead);
150 this.alpn = alpn;
151 }
152
153 @Override
154 public void writeAsync(ByteBufferReference[] src) throws IOException {
155 appOutputQ.put(src);
156 }
157
158 @Override
159 public void writeAsyncUnordered(ByteBufferReference[] buffers) throws IOException {
160 appOutputQ.putFirst(buffers);
161 }
162
163 @Override
164 public void flushAsync() throws IOException {
165 if (appOutputQ.flush()) {
166 lowerOutput.flushAsync();
167 }
168 }
169
170 SSLEngine getEngine() {
171 return engine;
172 }
173
174 @Override
175 public void closeExceptionally(Throwable t) {
176 Utils.close(t, appOutputQ, channelInputQ, lowerOutput);
177 }
178
179 @Override
180 public void close() {
181 Utils.close(appOutputQ, channelInputQ, lowerOutput);
182 }
183
184 // The code below can be uncommented to shake out
185 // the implementation by inserting random delays and trigger
186 // handshake in the SelectorManager thread (upperRead)
187 // static final java.util.Random random =
188 // new java.util.Random(System.currentTimeMillis());
189
190 /**
191 * Attempts to wrap buffers from appOutputQ and place them on the
192 * channelOutputQ for writing. If handshaking is happening, then the
193 * process stalls and last buffers taken off the appOutputQ are put back
194 * into it until handshaking completes.
195 *
196 * This same method is called to try and resume output after a blocking
197 * handshaking operation has completed.
198 */
199 private void upperWrite(ByteBufferReference[] refs, AsyncWriteQueue delayCallback) {
200 // currently delayCallback is not used. Use it when it's needed to execute handshake in another thread.
201 try {
202 ByteBuffer[] buffers = ByteBufferReference.toBuffers(refs);
203 int bytes = Utils.remaining(buffers);
204 while (bytes > 0) {
205 EngineResult r = wrapBuffers(buffers);
206 int bytesProduced = r.bytesProduced();
207 int bytesConsumed = r.bytesConsumed();
208 bytes -= bytesConsumed;
209 if (bytesProduced > 0) {
210 lowerOutput.writeAsync(new ByteBufferReference[]{r.destBuffer});
211 }
212
213 // The code below can be uncommented to shake out
214 // the implementation by inserting random delays and trigger
215 // handshake in the SelectorManager thread (upperRead)
216
217 // int sleep = random.nextInt(100);
218 // if (sleep > 20) {
219 // Thread.sleep(sleep);
220 // }
221
222 // handshaking is happening or is needed
223 if (r.handshaking()) {
224 Log.logTrace("Write: needs handshake");
225 doHandshakeNow("Write");
226 }
227 }
228 ByteBufferReference.clear(refs);
229 } catch (Throwable t) {
230 closeExceptionally(t);
231 errorHandler.accept(t);
232 }
233 }
234
235 // Connecting at this level means the initial handshake has completed.
236 // This means that the initial SSL parameters are available including
237 // ALPN result.
238 void connect() throws IOException, InterruptedException {
239 doHandshakeNow("Init");
240 connected = true;
241 }
242
243 boolean connected() {
244 return connected;
245 }
246
247 private void startHandshake(String tag) {
248 Runnable run = () -> {
249 try {
250 doHandshakeNow(tag);
251 } catch (Throwable t) {
252 Log.logTrace("{0}: handshake failed: {1}", tag, t);
253 closeExceptionally(t);
254 errorHandler.accept(t);
255 }
256 };
257 client.executor().execute(run);
258 }
259
260 private void doHandshakeNow(String tag)
261 throws IOException, InterruptedException
262 {
263 handshaker.acquire();
264 try {
265 channelInputQ.disableCallback();
266 lowerOutput.flushAsync();
267 Log.logTrace("{0}: Starting handshake...", tag);
268 doHandshakeImpl();
269 Log.logTrace("{0}: Handshake completed", tag);
270 // don't unblock the channel here, as we aren't sure yet, whether ALPN
271 // negotiation succeeded. Caller will call enableCallback() externally
272 } finally {
273 handshaker.release();
274 }
275 }
276
277 public void enableCallback() {
278 channelInputQ.enableCallback();
279 }
280
281 /**
282 * Executes entire handshake in calling thread.
283 * Returns after handshake is completed or error occurs
284 */
285 private void doHandshakeImpl() throws IOException {
286 engine.beginHandshake();
287 while (true) {
288 SSLEngineResult.HandshakeStatus status = engine.getHandshakeStatus();
289 switch(status) {
290 case NEED_TASK: {
291 List<Runnable> tasks = obtainTasks();
292 for (Runnable task : tasks) {
293 task.run();
294 }
295 } break;
296 case NEED_WRAP:
297 handshakeWrapAndSend();
298 break;
299 case NEED_UNWRAP: case NEED_UNWRAP_AGAIN:
300 handshakeReceiveAndUnWrap();
301 break;
302 case FINISHED:
303 return;
304 case NOT_HANDSHAKING:
305 return;
306 default:
307 throw new InternalError("Unexpected Handshake Status: "
308 + status);
309 }
310 }
311 }
312
313 // acknowledge a received CLOSE request from peer
314 void doClosure() throws IOException {
315 //while (!wrapAndSend(emptyArray))
316 //;
317 }
318
319 List<Runnable> obtainTasks() {
320 List<Runnable> l = new ArrayList<>();
321 Runnable r;
322 while ((r = engine.getDelegatedTask()) != null) {
323 l.add(r);
324 }
325 return l;
326 }
327
328 @Override
329 public void setAsyncCallbacks(Consumer<ByteBufferReference> asyncReceiver,
330 Consumer<Throwable> errorReceiver,
331 Supplier<ByteBufferReference> readBufferSupplier) {
332 this.asyncReceiver = asyncReceiver;
333 this.errorHandler = errorReceiver;
334 // readBufferSupplier is not used,
335 // because of AsyncSSLDelegate has its own appBufferPool
336 }
337
338 @Override
339 public void startReading() {
340 // maybe this class does not need to implement AsyncConnection
341 }
342
343 @Override
344 public void stopAsyncReading() {
345 // maybe this class does not need to implement AsyncConnection
346 }
347
348
349 static class EngineResult {
350 final SSLEngineResult result;
351 final ByteBufferReference destBuffer;
352
353
354 // normal result
355 EngineResult(SSLEngineResult result) {
356 this(result, null);
357 }
358
359 EngineResult(SSLEngineResult result, ByteBufferReference destBuffer) {
360 this.result = result;
361 this.destBuffer = destBuffer;
362 }
363
364 boolean handshaking() {
365 SSLEngineResult.HandshakeStatus s = result.getHandshakeStatus();
366 return s != FINISHED && s != NOT_HANDSHAKING;
367 }
368
369 int bytesConsumed() {
370 return result.bytesConsumed();
371 }
372
373 int bytesProduced() {
374 return result.bytesProduced();
375 }
376
377 SSLEngineResult.HandshakeStatus handshakeStatus() {
378 return result.getHandshakeStatus();
379 }
380
381 SSLEngineResult.Status status() {
382 return result.getStatus();
383 }
384 }
385
386 EngineResult handshakeWrapAndSend() throws IOException {
387 EngineResult r = wrapBuffer(Utils.EMPTY_BYTEBUFFER);
388 if (r.bytesProduced() > 0) {
389 lowerOutput.writeAsync(new ByteBufferReference[]{r.destBuffer});
390 lowerOutput.flushAsync();
391 }
392 return r;
393 }
394
395 // called during handshaking. It blocks until a complete packet
396 // is available, unwraps it and returns.
397 void handshakeReceiveAndUnWrap() throws IOException {
398 ByteBufferReference ref = channelInputQ.take();
399 while (true) {
400 // block waiting for input
401 EngineResult r = unwrapBuffer(ref.get());
402 SSLEngineResult.Status status = r.status();
403 if (status == BUFFER_UNDERFLOW) {
404 // wait for another buffer to arrive
405 ByteBufferReference ref1 = channelInputQ.take();
406 ref = combine (ref, ref1);
407 continue;
408 }
409 // OK
410 // theoretically possible we could receive some user data
411 if (r.bytesProduced() > 0) {
412 asyncReceiver.accept(r.destBuffer);
413 } else {
414 r.destBuffer.clear();
415 }
416 // it is also possible that a delegated task could be needed
417 // even though they are handled in the calling function
418 if (r.handshakeStatus() == NEED_TASK) {
419 obtainTasks().stream().forEach((task) -> task.run());
420 }
421
422 if (!ref.get().hasRemaining()) {
423 ref.clear();
424 return;
425 }
426 }
427 }
428
429 EngineResult wrapBuffer(ByteBuffer src) throws SSLException {
430 ByteBuffer[] bufs = new ByteBuffer[1];
431 bufs[0] = src;
432 return wrapBuffers(bufs);
433 }
434
435 private final ByteBufferPool netBufferPool = new ByteBufferPool();
436 private final ByteBufferPool appBufferPool = new ByteBufferPool();
437
438 /**
439 * provides buffer of sslEngine@getPacketBufferSize().
440 * used for encrypted buffers after wrap or before unwrap.
441 * @return ByteBufferReference
442 */
443 public ByteBufferReference getNetBuffer() {
444 return netBufferPool.get(engine.getSession().getPacketBufferSize());
445 }
446
447 /**
448 * provides buffer of sslEngine@getApplicationBufferSize().
449 * @return ByteBufferReference
450 */
451 private ByteBufferReference getAppBuffer() {
452 return appBufferPool.get(engine.getSession().getApplicationBufferSize());
453 }
454
455 EngineResult wrapBuffers(ByteBuffer[] src) throws SSLException {
456 ByteBufferReference dst = getNetBuffer();
457 while (true) {
458 SSLEngineResult sslResult = engine.wrap(src, dst.get());
459 switch (sslResult.getStatus()) {
460 case BUFFER_OVERFLOW:
461 // Shouldn't happen. We allocated buffer with packet size
462 // get it again if net buffer size was changed
463 dst = getNetBuffer();
464 break;
465 case CLOSED:
466 case OK:
467 dst.get().flip();
468 return new EngineResult(sslResult, dst);
469 case BUFFER_UNDERFLOW:
470 // Shouldn't happen. Doesn't returns when wrap()
471 // underflow handled externally
472 return new EngineResult(sslResult);
473 default:
474 assert false;
475 }
476 }
477 }
478
479 EngineResult unwrapBuffer(ByteBuffer srcbuf) throws IOException {
480 ByteBufferReference dst = getAppBuffer();
481 while (true) {
482 SSLEngineResult sslResult = engine.unwrap(srcbuf, dst.get());
483 switch (sslResult.getStatus()) {
484 case BUFFER_OVERFLOW:
485 // may happen only if app size buffer was changed.
486 // get it again if app buffer size changed
487 dst = getAppBuffer();
488 break;
489 case CLOSED:
490 doClosure();
491 throw new IOException("Engine closed");
492 case BUFFER_UNDERFLOW:
493 dst.clear();
494 return new EngineResult(sslResult);
495 case OK:
496 dst.get().flip();
497 return new EngineResult(sslResult, dst);
498 }
499 }
500 }
501
502 /**
503 * Asynchronous read input. Call this when selector fires.
504 * Unwrap done in upperRead because it also happens in
505 * doHandshake() when handshake taking place
506 */
507 public void asyncReceive(ByteBufferReference buffer) {
508 try {
509 channelInputQ.put(buffer);
510 } catch (Throwable t) {
511 closeExceptionally(t);
512 errorHandler.accept(t);
513 }
514 }
515
516 private ByteBufferReference pollInput() throws IOException {
517 return channelInputQ.poll();
518 }
519
520 private ByteBufferReference pollInput(ByteBufferReference next) throws IOException {
521 return next == null ? channelInputQ.poll() : next;
522 }
523
524 public void upperRead() {
525 ByteBufferReference src;
526 ByteBufferReference next = null;
527 synchronized (reader) {
528 try {
529 src = pollInput();
530 if (src == null) {
531 return;
532 }
533 while (true) {
534 EngineResult r = unwrapBuffer(src.get());
535 switch (r.result.getStatus()) {
536 case BUFFER_UNDERFLOW:
537 // Buffer too small. Need to combine with next buf
538 next = pollInput(next);
539 if (next == null) {
540 // no data available.
541 // push buffer back until more data available
542 channelInputQ.pushback(src);
543 return;
544 } else {
545 src = shift(src, next);
546 if (!next.get().hasRemaining()) {
547 next.clear();
548 next = null;
549 }
550 }
551 break;
552 case OK:
553 // check for any handshaking work
554 if (r.handshaking()) {
555 // handshaking is happening or is needed
556 // so we put the buffer back on Q to process again
557 // later.
558 Log.logTrace("Read: needs handshake");
559 channelInputQ.pushback(src);
560 startHandshake("Read");
561 return;
562 }
563 asyncReceiver.accept(r.destBuffer);
564 }
565 if (src.get().hasRemaining()) {
566 continue;
567 }
568 src.clear();
569 src = pollInput(next);
570 next = null;
571 if (src == null) {
572 return;
573 }
574 }
575 } catch (Throwable t) {
576 closeExceptionally(t);
577 errorHandler.accept(t);
578 }
579 }
580 }
581
582 ByteBufferReference shift(ByteBufferReference ref1, ByteBufferReference ref2) {
583 ByteBuffer buf1 = ref1.get();
584 if (buf1.capacity() < engine.getSession().getPacketBufferSize()) {
585 ByteBufferReference newRef = getNetBuffer();
586 ByteBuffer newBuf = newRef.get();
587 newBuf.put(buf1);
588 buf1 = newBuf;
589 ref1.clear();
590 ref1 = newRef;
591 } else {
592 buf1.compact();
593 }
594 ByteBuffer buf2 = ref2.get();
595 Utils.copy(buf2, buf1, Math.min(buf1.remaining(), buf2.remaining()));
596 buf1.flip();
597 return ref1;
598 }
599
600
601 ByteBufferReference combine(ByteBufferReference ref1, ByteBufferReference ref2) {
602 ByteBuffer buf1 = ref1.get();
603 ByteBuffer buf2 = ref2.get();
604 int avail1 = buf1.capacity() - buf1.remaining();
605 if (buf2.remaining() < avail1) {
606 buf1.compact();
607 buf1.put(buf2);
608 buf1.flip();
609 ref2.clear();
610 return ref1;
611 }
612 int newsize = buf1.remaining() + buf2.remaining();
613 ByteBuffer newbuf = ByteBuffer.allocate(newsize); // getting rid of buffer pools
614 newbuf.put(buf1);
615 newbuf.put(buf2);
616 newbuf.flip();
617 ref1.clear();
618 ref2.clear();
619 return ByteBufferReference.of(newbuf);
620 }
621
622 SSLParameters getSSLParameters() {
623 return sslParameters;
624 }
625
626 static void logParams(SSLParameters p) {
627 if (!Log.ssl()) {
628 return;
629 }
630
631 if (p == null) {
632 Log.logSSL("SSLParameters: Null params");
633 return;
634 }
635
636 final StringBuilder sb = new StringBuilder("SSLParameters:");
637 final List<Object> params = new ArrayList<>();
638 if (p.getCipherSuites() != null) {
639 for (String cipher : p.getCipherSuites()) {
640 sb.append("\n cipher: {")
641 .append(params.size()).append("}");
642 params.add(cipher);
643 }
644 }
645
646 // SSLParameters.getApplicationProtocols() can't return null
647 // JDK 8 EXCL START
648 for (String approto : p.getApplicationProtocols()) {
649 sb.append("\n application protocol: {")
650 .append(params.size()).append("}");
651 params.add(approto);
652 }
653 // JDK 8 EXCL END
654
655 if (p.getProtocols() != null) {
656 for (String protocol : p.getProtocols()) {
657 sb.append("\n protocol: {")
658 .append(params.size()).append("}");
659 params.add(protocol);
660 }
661 }
662
663 if (p.getServerNames() != null) {
664 for (SNIServerName sname : p.getServerNames()) {
665 sb.append("\n server name: {")
666 .append(params.size()).append("}");
667 params.add(sname.toString());
668 }
669 }
670 sb.append('\n');
671
672 Log.logSSL(sb.toString(), params.toArray());
673 }
674
675 String getSessionInfo() {
676 StringBuilder sb = new StringBuilder();
677 String application = engine.getApplicationProtocol();
678 SSLSession sess = engine.getSession();
679 String cipher = sess.getCipherSuite();
680 String protocol = sess.getProtocol();
681 sb.append("Handshake complete alpn: ")
682 .append(application)
683 .append(", Cipher: ")
684 .append(cipher)
685 .append(", Protocol: ")
686 .append(protocol);
687 return sb.toString();
688 }
689 }