1
2
3
4
5
6
7
8
9
10
11
12
13
14 package org.eclipse.jetty.io.nio;
15
16 import java.io.IOException;
17 import java.nio.ByteBuffer;
18 import java.util.concurrent.atomic.AtomicBoolean;
19 import javax.net.ssl.SSLEngine;
20 import javax.net.ssl.SSLEngineResult;
21 import javax.net.ssl.SSLEngineResult.HandshakeStatus;
22 import javax.net.ssl.SSLException;
23 import javax.net.ssl.SSLSession;
24
25 import org.eclipse.jetty.io.AbstractConnection;
26 import org.eclipse.jetty.io.AsyncEndPoint;
27 import org.eclipse.jetty.io.Buffer;
28 import org.eclipse.jetty.io.Connection;
29 import org.eclipse.jetty.io.EndPoint;
30 import org.eclipse.jetty.util.log.Log;
31 import org.eclipse.jetty.util.log.Logger;
32 import org.eclipse.jetty.util.thread.Timeout.Task;
33
34
35
36
37
38
39
40
41
42
43 public class SslConnection extends AbstractConnection implements AsyncConnection
44 {
45 private final Logger _logger = Log.getLogger("org.eclipse.jetty.io.nio.ssl");
46
47 private static final NIOBuffer __ZERO_BUFFER=new IndirectNIOBuffer(0);
48
49 private static final ThreadLocal<SslBuffers> __buffers = new ThreadLocal<SslBuffers>();
50 private final SSLEngine _engine;
51 private final SSLSession _session;
52 private AsyncConnection _connection;
53 private final SslEndPoint _sslEndPoint;
54 private int _allocations;
55 private SslBuffers _buffers;
56 private NIOBuffer _inbound;
57 private NIOBuffer _unwrapBuf;
58 private NIOBuffer _outbound;
59 private AsyncEndPoint _aEndp;
60 private boolean _allowRenegotiate=true;
61 private boolean _handshook;
62 private boolean _ishut;
63 private boolean _oshut;
64 private final AtomicBoolean _progressed = new AtomicBoolean();
65
66
67
68
69 private static class SslBuffers
70 {
71 final NIOBuffer _in;
72 final NIOBuffer _out;
73 final NIOBuffer _unwrap;
74
75 SslBuffers(int packetSize, int appSize)
76 {
77 _in=new IndirectNIOBuffer(packetSize);
78 _out=new IndirectNIOBuffer(packetSize);
79 _unwrap=new IndirectNIOBuffer(appSize);
80 }
81 }
82
83
84 public SslConnection(SSLEngine engine,EndPoint endp)
85 {
86 this(engine,endp,System.currentTimeMillis());
87 }
88
89
90 public SslConnection(SSLEngine engine,EndPoint endp, long timeStamp)
91 {
92 super(endp,timeStamp);
93 _engine=engine;
94 _session=_engine.getSession();
95 _aEndp=(AsyncEndPoint)endp;
96 _sslEndPoint = newSslEndPoint();
97 }
98
99
100 protected SslEndPoint newSslEndPoint()
101 {
102 return new SslEndPoint();
103 }
104
105
106
107
108
109 public boolean isAllowRenegotiate()
110 {
111 return _allowRenegotiate;
112 }
113
114
115
116
117
118
119
120
121
122
123
124
125 public void setAllowRenegotiate(boolean allowRenegotiate)
126 {
127 _allowRenegotiate = allowRenegotiate;
128 }
129
130
131 private void allocateBuffers()
132 {
133 synchronized (this)
134 {
135 if (_allocations++==0)
136 {
137 if (_buffers==null)
138 {
139 _buffers=__buffers.get();
140 if (_buffers==null)
141 _buffers=new SslBuffers(_session.getPacketBufferSize()*2,_session.getApplicationBufferSize()*2);
142 _inbound=_buffers._in;
143 _outbound=_buffers._out;
144 _unwrapBuf=_buffers._unwrap;
145 __buffers.set(null);
146 }
147 }
148 }
149 }
150
151
152 private void releaseBuffers()
153 {
154 synchronized (this)
155 {
156 if (--_allocations==0)
157 {
158 if (_buffers!=null &&
159 _inbound.length()==0 &&
160 _outbound.length()==0 &&
161 _unwrapBuf.length()==0)
162 {
163 _inbound=null;
164 _outbound=null;
165 _unwrapBuf=null;
166 __buffers.set(_buffers);
167 _buffers=null;
168 }
169 }
170 }
171 }
172
173
174 public Connection handle() throws IOException
175 {
176 try
177 {
178 allocateBuffers();
179
180 boolean progress=true;
181
182 while (progress)
183 {
184 progress=false;
185
186
187 if (_engine.getHandshakeStatus()!=HandshakeStatus.NOT_HANDSHAKING)
188 progress=process(null,null);
189
190
191 AsyncConnection next = (AsyncConnection)_connection.handle();
192 if (next!=_connection && next!=null)
193 {
194 _connection=next;
195 progress=true;
196 }
197
198 _logger.debug("{} handle {} progress={}", _session, this, progress);
199 }
200 }
201 finally
202 {
203 releaseBuffers();
204
205 if (!_ishut && _sslEndPoint.isInputShutdown() && _sslEndPoint.isOpen())
206 {
207 _ishut=true;
208 try
209 {
210 _connection.onInputShutdown();
211 }
212 catch(Throwable x)
213 {
214 _logger.warn("onInputShutdown failed", x);
215 try{_sslEndPoint.close();}
216 catch(IOException e2){
217 _logger.ignore(e2);}
218 }
219 }
220 }
221
222 return this;
223 }
224
225
226 public boolean isIdle()
227 {
228 return false;
229 }
230
231
232 public boolean isSuspended()
233 {
234 return false;
235 }
236
237
238 public void onClose()
239 {
240 Connection connection = _sslEndPoint.getConnection();
241 if (connection != null && connection != this)
242 connection.onClose();
243 }
244
245
246 @Override
247 public void onIdleExpired(long idleForMs)
248 {
249 try
250 {
251 _logger.debug("onIdleExpired {}ms on {}",idleForMs,this);
252 if (_endp.isOutputShutdown())
253 _sslEndPoint.close();
254 else
255 _sslEndPoint.shutdownOutput();
256 }
257 catch (IOException e)
258 {
259 _logger.warn(e);
260 super.onIdleExpired(idleForMs);
261 }
262 }
263
264
265 public void onInputShutdown() throws IOException
266 {
267
268 }
269
270
271 private synchronized boolean process(Buffer toFill, Buffer toFlush) throws IOException
272 {
273 boolean some_progress=false;
274 try
275 {
276
277 allocateBuffers();
278
279
280 if (toFill==null)
281 {
282
283 _unwrapBuf.compact();
284 toFill=_unwrapBuf;
285 }
286
287 else if (toFill.capacity()<_session.getApplicationBufferSize())
288 {
289
290 boolean progress=process(null,toFlush);
291
292
293 if (_unwrapBuf!=null && _unwrapBuf.hasContent())
294 {
295
296 _unwrapBuf.skip(toFill.put(_unwrapBuf));
297 return true;
298 }
299 else
300
301 return progress;
302 }
303
304 else if (_unwrapBuf!=null && _unwrapBuf.hasContent())
305 {
306
307 _unwrapBuf.skip(toFill.put(_unwrapBuf));
308 return true;
309 }
310
311
312
313
314 if (toFlush==null)
315 toFlush=__ZERO_BUFFER;
316
317
318 boolean progress=true;
319 while (progress)
320 {
321 progress=false;
322
323
324 int filled=0,flushed=0;
325 try
326 {
327
328 if (_inbound.space()>0 && (filled=_endp.fill(_inbound))>0)
329 progress = true;
330
331
332 if (_outbound.hasContent() && (flushed=_endp.flush(_outbound))>0)
333 progress = true;
334 }
335 catch (IOException e)
336 {
337 _endp.close();
338 throw e;
339 }
340 finally
341 {
342 _logger.debug("{} {} {} filled={}/{} flushed={}/{}",_session,this,_engine.getHandshakeStatus(),filled,_inbound.length(),flushed,_outbound.length());
343 }
344
345
346 switch(_engine.getHandshakeStatus())
347 {
348 case FINISHED:
349 throw new IllegalStateException();
350
351 case NOT_HANDSHAKING:
352 {
353
354 if (toFill.space()>0 && _inbound.hasContent() && unwrap(toFill))
355 progress=true;
356
357
358 if (toFlush.hasContent() && _outbound.space()>0 && wrap(toFlush))
359 progress=true;
360 }
361 break;
362
363 case NEED_TASK:
364 {
365
366 Runnable task;
367 while ((task=_engine.getDelegatedTask())!=null)
368 {
369 progress=true;
370 task.run();
371 }
372
373 }
374 break;
375
376 case NEED_WRAP:
377 {
378
379 if (_handshook && !_allowRenegotiate)
380 _endp.close();
381 else if (wrap(toFlush))
382 progress=true;
383 }
384 break;
385
386 case NEED_UNWRAP:
387 {
388
389 if (_handshook && !_allowRenegotiate)
390 _endp.close();
391 else if (!_inbound.hasContent()&&filled==-1)
392 {
393
394 _endp.shutdownInput();
395 }
396 else if (unwrap(toFill))
397 progress=true;
398 }
399 break;
400 }
401
402
403 if (_endp.isOpen() && _endp.isInputShutdown() && !_inbound.hasContent())
404 _engine.closeInbound();
405
406 if (_endp.isOpen() && _engine.isOutboundDone() && !_outbound.hasContent())
407 _endp.shutdownOutput();
408
409
410 some_progress|=progress;
411 }
412
413
414 if (toFill==_unwrapBuf && _unwrapBuf.hasContent() && !_connection.isSuspended())
415 _aEndp.dispatch();
416 }
417 finally
418 {
419 releaseBuffers();
420 if (some_progress)
421 _progressed.set(true);
422 }
423 return some_progress;
424 }
425
426 private synchronized boolean wrap(final Buffer buffer) throws IOException
427 {
428 ByteBuffer bbuf=extractByteBuffer(buffer);
429 final SSLEngineResult result;
430
431 synchronized(bbuf)
432 {
433 _outbound.compact();
434 ByteBuffer out_buffer=_outbound.getByteBuffer();
435 synchronized(out_buffer)
436 {
437 try
438 {
439 bbuf.position(buffer.getIndex());
440 bbuf.limit(buffer.putIndex());
441 out_buffer.position(_outbound.putIndex());
442 out_buffer.limit(out_buffer.capacity());
443 result=_engine.wrap(bbuf,out_buffer);
444 if (_logger.isDebugEnabled())
445 _logger.debug("{} wrap {} {} consumed={} produced={}",
446 _session,
447 result.getStatus(),
448 result.getHandshakeStatus(),
449 result.bytesConsumed(),
450 result.bytesProduced());
451
452
453 buffer.skip(result.bytesConsumed());
454 _outbound.setPutIndex(_outbound.putIndex()+result.bytesProduced());
455 }
456 catch(SSLException e)
457 {
458 _logger.debug(String.valueOf(_endp), e);
459 _endp.close();
460 throw e;
461 }
462 finally
463 {
464 out_buffer.position(0);
465 out_buffer.limit(out_buffer.capacity());
466 bbuf.position(0);
467 bbuf.limit(bbuf.capacity());
468 }
469 }
470 }
471
472 switch(result.getStatus())
473 {
474 case BUFFER_UNDERFLOW:
475 throw new IllegalStateException();
476
477 case BUFFER_OVERFLOW:
478 break;
479
480 case OK:
481 if (result.getHandshakeStatus()==HandshakeStatus.FINISHED)
482 _handshook=true;
483 break;
484
485 case CLOSED:
486 _logger.debug("wrap CLOSE {} {}",this,result);
487 if (result.getHandshakeStatus()==HandshakeStatus.FINISHED)
488 _endp.close();
489 break;
490
491 default:
492 _logger.debug("{} wrap default {}",_session,result);
493 throw new IOException(result.toString());
494 }
495
496 return result.bytesConsumed()>0 || result.bytesProduced()>0;
497 }
498
499 private synchronized boolean unwrap(final Buffer buffer) throws IOException
500 {
501 if (!_inbound.hasContent())
502 return false;
503
504 ByteBuffer bbuf=extractByteBuffer(buffer);
505 final SSLEngineResult result;
506
507 synchronized(bbuf)
508 {
509 ByteBuffer in_buffer=_inbound.getByteBuffer();
510 synchronized(in_buffer)
511 {
512 try
513 {
514 bbuf.position(buffer.putIndex());
515 bbuf.limit(buffer.capacity());
516 in_buffer.position(_inbound.getIndex());
517 in_buffer.limit(_inbound.putIndex());
518
519 result=_engine.unwrap(in_buffer,bbuf);
520 if (_logger.isDebugEnabled())
521 _logger.debug("{} unwrap {} {} consumed={} produced={}",
522 _session,
523 result.getStatus(),
524 result.getHandshakeStatus(),
525 result.bytesConsumed(),
526 result.bytesProduced());
527
528 _inbound.skip(result.bytesConsumed());
529 _inbound.compact();
530 buffer.setPutIndex(buffer.putIndex()+result.bytesProduced());
531 }
532 catch(SSLException e)
533 {
534 _logger.debug(String.valueOf(_endp), e);
535 _endp.close();
536 throw e;
537 }
538 finally
539 {
540 in_buffer.position(0);
541 in_buffer.limit(in_buffer.capacity());
542 bbuf.position(0);
543 bbuf.limit(bbuf.capacity());
544 }
545 }
546 }
547
548 switch(result.getStatus())
549 {
550 case BUFFER_UNDERFLOW:
551 if (_endp.isInputShutdown())
552 _inbound.clear();
553 break;
554
555 case BUFFER_OVERFLOW:
556 if (_logger.isDebugEnabled()) _logger.debug("{} unwrap {} {}->{}",_session,result.getStatus(),_inbound.toDetailString(),buffer.toDetailString());
557 break;
558
559 case OK:
560 if (result.getHandshakeStatus()==HandshakeStatus.FINISHED)
561 _handshook=true;
562 break;
563
564 case CLOSED:
565 _logger.debug("unwrap CLOSE {} {}",this,result);
566 if (result.getHandshakeStatus()==HandshakeStatus.FINISHED)
567 _endp.close();
568 break;
569
570 default:
571 _logger.debug("{} wrap default {}",_session,result);
572 throw new IOException(result.toString());
573 }
574
575
576
577
578 return result.bytesConsumed()>0 || result.bytesProduced()>0;
579 }
580
581
582
583 private ByteBuffer extractByteBuffer(Buffer buffer)
584 {
585 if (buffer.buffer() instanceof NIOBuffer)
586 return ((NIOBuffer)buffer.buffer()).getByteBuffer();
587 return ByteBuffer.wrap(buffer.array());
588 }
589
590
591 public AsyncEndPoint getSslEndPoint()
592 {
593 return _sslEndPoint;
594 }
595
596
597 public String toString()
598 {
599 return String.format("%s %s", super.toString(), _sslEndPoint);
600 }
601
602
603
604 public class SslEndPoint implements AsyncEndPoint
605 {
606 public SSLEngine getSslEngine()
607 {
608 return _engine;
609 }
610
611 public AsyncEndPoint getEndpoint()
612 {
613 return _aEndp;
614 }
615
616 public void shutdownOutput() throws IOException
617 {
618 synchronized (SslConnection.this)
619 {
620 _logger.debug("{} ssl endp.oshut {}",_session,this);
621 _engine.closeOutbound();
622 _oshut=true;
623 }
624 flush();
625 }
626
627 public boolean isOutputShutdown()
628 {
629 synchronized (SslConnection.this)
630 {
631 return _oshut||!isOpen()||_engine.isOutboundDone();
632 }
633 }
634
635 public void shutdownInput() throws IOException
636 {
637 _logger.debug("{} ssl endp.ishut!",_session);
638
639
640 }
641
642 public boolean isInputShutdown()
643 {
644 synchronized (SslConnection.this)
645 {
646 return _endp.isInputShutdown() &&
647 !(_unwrapBuf!=null&&_unwrapBuf.hasContent()) &&
648 !(_inbound!=null&&_inbound.hasContent());
649 }
650 }
651
652 public void close() throws IOException
653 {
654 _logger.debug("{} ssl endp.close",_session);
655 _endp.close();
656 }
657
658 public int fill(Buffer buffer) throws IOException
659 {
660 int size=buffer.length();
661 process(buffer, null);
662
663 int filled=buffer.length()-size;
664
665 if (filled==0 && isInputShutdown())
666 return -1;
667 return filled;
668 }
669
670 public int flush(Buffer buffer) throws IOException
671 {
672 int size = buffer.length();
673 process(null, buffer);
674 return size-buffer.length();
675 }
676
677 public int flush(Buffer header, Buffer buffer, Buffer trailer) throws IOException
678 {
679 if (header!=null && header.hasContent())
680 return flush(header);
681 if (buffer!=null && buffer.hasContent())
682 return flush(buffer);
683 if (trailer!=null && trailer.hasContent())
684 return flush(trailer);
685 return 0;
686 }
687
688 public boolean blockReadable(long millisecs) throws IOException
689 {
690 long now = System.currentTimeMillis();
691 long end=millisecs>0?(now+millisecs):Long.MAX_VALUE;
692
693 while (now<end)
694 {
695 if (process(null,null))
696 break;
697 _endp.blockReadable(end-now);
698 now = System.currentTimeMillis();
699 }
700
701 return now<end;
702 }
703
704 public boolean blockWritable(long millisecs) throws IOException
705 {
706 return _endp.blockWritable(millisecs);
707 }
708
709 public boolean isOpen()
710 {
711 return _endp.isOpen();
712 }
713
714 public Object getTransport()
715 {
716 return _endp;
717 }
718
719 public void flush() throws IOException
720 {
721 process(null, null);
722 }
723
724 public void dispatch()
725 {
726 _aEndp.dispatch();
727 }
728
729 public void asyncDispatch()
730 {
731 _aEndp.asyncDispatch();
732 }
733
734 public void scheduleWrite()
735 {
736 _aEndp.scheduleWrite();
737 }
738
739 public void onIdleExpired(long idleForMs)
740 {
741 _aEndp.onIdleExpired(idleForMs);
742 }
743
744 public void setCheckForIdle(boolean check)
745 {
746 _aEndp.setCheckForIdle(check);
747 }
748
749 public boolean isCheckForIdle()
750 {
751 return _aEndp.isCheckForIdle();
752 }
753
754 public void scheduleTimeout(Task task, long timeoutMs)
755 {
756 _aEndp.scheduleTimeout(task,timeoutMs);
757 }
758
759 public void cancelTimeout(Task task)
760 {
761 _aEndp.cancelTimeout(task);
762 }
763
764 public boolean isWritable()
765 {
766 return _aEndp.isWritable();
767 }
768
769 public boolean hasProgressed()
770 {
771 return _progressed.getAndSet(false);
772 }
773
774 public String getLocalAddr()
775 {
776 return _aEndp.getLocalAddr();
777 }
778
779 public String getLocalHost()
780 {
781 return _aEndp.getLocalHost();
782 }
783
784 public int getLocalPort()
785 {
786 return _aEndp.getLocalPort();
787 }
788
789 public String getRemoteAddr()
790 {
791 return _aEndp.getRemoteAddr();
792 }
793
794 public String getRemoteHost()
795 {
796 return _aEndp.getRemoteHost();
797 }
798
799 public int getRemotePort()
800 {
801 return _aEndp.getRemotePort();
802 }
803
804 public boolean isBlocking()
805 {
806 return false;
807 }
808
809 public int getMaxIdleTime()
810 {
811 return _aEndp.getMaxIdleTime();
812 }
813
814 public void setMaxIdleTime(int timeMs) throws IOException
815 {
816 _aEndp.setMaxIdleTime(timeMs);
817 }
818
819 public Connection getConnection()
820 {
821 return _connection;
822 }
823
824 public void setConnection(Connection connection)
825 {
826 _connection=(AsyncConnection)connection;
827 }
828
829 public String toString()
830 {
831
832
833
834 Buffer inbound = _inbound;
835 Buffer outbound = _outbound;
836 Buffer unwrap = _unwrapBuf;
837 int i = inbound == null? -1 : inbound.length();
838 int o = outbound == null ? -1 : outbound.length();
839 int u = unwrap == null ? -1 : unwrap.length();
840 return String.format("SSL %s i/o/u=%d/%d/%d ishut=%b oshut=%b {%s}",
841 _engine.getHandshakeStatus(),
842 i, o, u,
843 _ishut, _oshut,
844 _connection);
845 }
846
847 }
848 }