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