1 package org.eclipse.jetty.server.handler;
2
3 import java.io.IOException;
4 import java.net.InetSocketAddress;
5 import java.nio.channels.ClosedChannelException;
6 import java.nio.channels.SelectionKey;
7 import java.nio.channels.SocketChannel;
8 import java.util.Arrays;
9 import java.util.Collections;
10 import java.util.concurrent.ConcurrentHashMap;
11 import java.util.concurrent.ConcurrentMap;
12 import java.util.concurrent.CountDownLatch;
13 import java.util.concurrent.TimeUnit;
14 import javax.servlet.ServletException;
15 import javax.servlet.http.HttpServletRequest;
16 import javax.servlet.http.HttpServletResponse;
17
18 import org.eclipse.jetty.io.AbstractConnection;
19 import org.eclipse.jetty.http.HttpMethods;
20 import org.eclipse.jetty.http.HttpParser;
21 import org.eclipse.jetty.io.Buffer;
22 import org.eclipse.jetty.io.ConnectedEndPoint;
23 import org.eclipse.jetty.io.Connection;
24 import org.eclipse.jetty.io.EndPoint;
25 import org.eclipse.jetty.io.nio.IndirectNIOBuffer;
26 import org.eclipse.jetty.io.nio.SelectChannelEndPoint;
27 import org.eclipse.jetty.io.nio.SelectorManager;
28 import org.eclipse.jetty.server.Handler;
29 import org.eclipse.jetty.server.HttpConnection;
30 import org.eclipse.jetty.server.Request;
31 import org.eclipse.jetty.server.Server;
32 import org.eclipse.jetty.util.HostMap;
33 import org.eclipse.jetty.util.TypeUtil;
34 import org.eclipse.jetty.util.component.LifeCycle;
35 import org.eclipse.jetty.util.log.Log;
36 import org.eclipse.jetty.util.log.Logger;
37 import org.eclipse.jetty.util.thread.ThreadPool;
38
39
40
41
42
43
44 public class ConnectHandler extends HandlerWrapper
45 {
46 private final Logger _logger = Log.getLogger(getClass().getName());
47 private final SelectorManager _selectorManager = new Manager();
48 private volatile int _connectTimeout = 5000;
49 private volatile int _writeTimeout = 30000;
50 private volatile ThreadPool _threadPool;
51 private volatile boolean _privateThreadPool;
52 private HostMap<String> _white = new HostMap<String>();
53 private HostMap<String> _black = new HostMap<String>();
54
55 public ConnectHandler()
56 {
57 this(null);
58 }
59
60 public ConnectHandler(String[] white, String[] black)
61 {
62 this(null, white, black);
63 }
64
65 public ConnectHandler(Handler handler)
66 {
67 setHandler(handler);
68 }
69
70 public ConnectHandler(Handler handler, String[] white, String[] black)
71 {
72 setHandler(handler);
73 set(white, _white);
74 set(black, _black);
75 }
76
77
78
79
80 public int getConnectTimeout()
81 {
82 return _connectTimeout;
83 }
84
85
86
87
88 public void setConnectTimeout(int connectTimeout)
89 {
90 _connectTimeout = connectTimeout;
91 }
92
93
94
95
96 public int getWriteTimeout()
97 {
98 return _writeTimeout;
99 }
100
101
102
103
104 public void setWriteTimeout(int writeTimeout)
105 {
106 _writeTimeout = writeTimeout;
107 }
108
109 @Override
110 public void setServer(Server server)
111 {
112 super.setServer(server);
113
114 server.getContainer().update(this,null,_selectorManager,"selectManager");
115
116 if (_privateThreadPool)
117 server.getContainer().update(this,null,_privateThreadPool,"threadpool",true);
118 else
119 _threadPool=server.getThreadPool();
120 }
121
122
123
124
125 public ThreadPool getThreadPool()
126 {
127 return _threadPool;
128 }
129
130
131
132
133 public void setThreadPool(ThreadPool threadPool)
134 {
135 if (getServer()!=null)
136 getServer().getContainer().update(this,_privateThreadPool?_threadPool:null,threadPool,"threadpool",true);
137 _privateThreadPool=threadPool!=null;
138 _threadPool=threadPool;
139 }
140
141 @Override
142 protected void doStart() throws Exception
143 {
144 super.doStart();
145
146 if (_threadPool==null)
147 {
148 _threadPool=getServer().getThreadPool();
149 _privateThreadPool=false;
150 }
151 if (_threadPool instanceof LifeCycle && !((LifeCycle)_threadPool).isRunning())
152 ((LifeCycle)_threadPool).start();
153
154 _selectorManager.start();
155 _threadPool.dispatch(new Runnable()
156 {
157 public void run()
158 {
159 while (isRunning())
160 {
161 try
162 {
163 _selectorManager.doSelect(0);
164 }
165 catch (IOException x)
166 {
167 _logger.warn("Unexpected exception", x);
168 }
169 }
170 }
171 });
172 }
173
174 @Override
175 protected void doStop() throws Exception
176 {
177 _selectorManager.stop();
178
179 ThreadPool threadPool = _threadPool;
180 if (_privateThreadPool && _threadPool != null && threadPool instanceof LifeCycle)
181 ((LifeCycle)threadPool).stop();
182
183 super.doStop();
184 }
185
186 @Override
187 public void handle(String target, Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException
188 {
189 if (HttpMethods.CONNECT.equalsIgnoreCase(request.getMethod()))
190 {
191 _logger.debug("CONNECT request for {}", request.getRequestURI());
192 handleConnect(baseRequest, request, response, request.getRequestURI());
193 }
194 else
195 {
196 super.handle(target, baseRequest, request, response);
197 }
198 }
199
200
201
202
203
204
205
206
207
208
209
210
211
212 protected void handleConnect(Request baseRequest, HttpServletRequest request, HttpServletResponse response, String serverAddress) throws ServletException, IOException
213 {
214 boolean proceed = handleAuthentication(request, response, serverAddress);
215 if (!proceed)
216 return;
217
218 String host = serverAddress;
219 int port = 80;
220 int colon = serverAddress.indexOf(':');
221 if (colon > 0)
222 {
223 host = serverAddress.substring(0, colon);
224 port = Integer.parseInt(serverAddress.substring(colon + 1));
225 }
226
227 if (!validateDestination(host))
228 {
229 Log.info("ProxyHandler: Forbidden destination "+host);
230 response.setStatus(HttpServletResponse.SC_FORBIDDEN);
231 baseRequest.setHandled(true);
232 return;
233 }
234
235 SocketChannel channel = connectToServer(request, host, port);
236
237
238
239
240
241
242 HttpConnection httpConnection = HttpConnection.getCurrentConnection();
243 Buffer headerBuffer = ((HttpParser)httpConnection.getParser()).getHeaderBuffer();
244 Buffer bodyBuffer = ((HttpParser)httpConnection.getParser()).getBodyBuffer();
245 int length = headerBuffer == null ? 0 : headerBuffer.length();
246 length += bodyBuffer == null ? 0 : bodyBuffer.length();
247 IndirectNIOBuffer buffer = null;
248 if (length > 0)
249 {
250 buffer = new IndirectNIOBuffer(length);
251 if (headerBuffer != null)
252 {
253 buffer.put(headerBuffer);
254 headerBuffer.clear();
255 }
256 if (bodyBuffer != null)
257 {
258 buffer.put(bodyBuffer);
259 bodyBuffer.clear();
260 }
261 }
262
263 ConcurrentMap<String, Object> context = new ConcurrentHashMap<String, Object>();
264 prepareContext(request, context);
265
266 ClientToProxyConnection clientToProxy = prepareConnections(context, channel, buffer);
267
268
269 response.setStatus(HttpServletResponse.SC_OK);
270
271
272 baseRequest.getConnection().getGenerator().setPersistent(true);
273
274
275 response.getOutputStream().close();
276
277 upgradeConnection(request, response, clientToProxy);
278 }
279
280 private ClientToProxyConnection prepareConnections(ConcurrentMap<String, Object> context, SocketChannel channel, Buffer buffer)
281 {
282 HttpConnection httpConnection = HttpConnection.getCurrentConnection();
283 ClientToProxyConnection clientToProxy = newClientToProxyConnection(context, channel, httpConnection.getEndPoint(), httpConnection.getTimeStamp());
284 ProxyToServerConnection proxyToServer = newProxyToServerConnection(context, buffer,clientToProxy);
285 clientToProxy.setConnection(proxyToServer);
286 return clientToProxy;
287 }
288
289
290
291
292
293
294
295
296
297
298
299
300 protected boolean handleAuthentication(HttpServletRequest request, HttpServletResponse response, String address) throws ServletException, IOException
301 {
302 return true;
303 }
304
305 protected ClientToProxyConnection newClientToProxyConnection(ConcurrentMap<String, Object> context, SocketChannel channel, EndPoint endPoint, long timeStamp)
306 {
307 return new ClientToProxyConnection(context, channel, endPoint, timeStamp);
308 }
309
310 protected ProxyToServerConnection newProxyToServerConnection(ConcurrentMap<String, Object> context, Buffer buffer,ClientToProxyConnection toProxy)
311 {
312 return new ProxyToServerConnection(context, buffer,toProxy);
313 }
314
315 private SocketChannel connectToServer(HttpServletRequest request, String host, int port) throws IOException
316 {
317 SocketChannel channel = connect(request, host, port);
318 channel.configureBlocking(false);
319 return channel;
320 }
321
322
323
324
325
326
327
328
329
330 protected SocketChannel connect(HttpServletRequest request, String host, int port) throws IOException
331 {
332 _logger.debug("Establishing connection to {}:{}", host, port);
333
334 SocketChannel channel = SocketChannel.open();
335 channel.socket().setTcpNoDelay(true);
336 channel.socket().connect(new InetSocketAddress(host, port), getConnectTimeout());
337 _logger.debug("Established connection to {}:{}", host, port);
338 return channel;
339 }
340
341 protected void prepareContext(HttpServletRequest request, ConcurrentMap<String, Object> context)
342 {
343 }
344
345 private void upgradeConnection(HttpServletRequest request, HttpServletResponse response, Connection connection) throws IOException
346 {
347
348
349 request.setAttribute("org.eclipse.jetty.io.Connection", connection);
350 response.setStatus(HttpServletResponse.SC_SWITCHING_PROTOCOLS);
351 _logger.debug("Upgraded connection to {}", connection);
352 }
353
354 private void register(SocketChannel channel, ProxyToServerConnection proxyToServer) throws IOException
355 {
356 _selectorManager.register(channel, proxyToServer);
357 proxyToServer.waitReady(_connectTimeout);
358 }
359
360
361
362
363
364
365
366
367
368
369 protected int read(EndPoint endPoint, Buffer buffer, ConcurrentMap<String, Object> context) throws IOException
370 {
371 return endPoint.fill(buffer);
372 }
373
374
375
376
377
378
379
380
381
382
383 protected int write(EndPoint endPoint, Buffer buffer, ConcurrentMap<String, Object> context) throws IOException
384 {
385 if (buffer == null)
386 return 0;
387
388 int length = buffer.length();
389 StringBuilder builder = new StringBuilder();
390 int written = endPoint.flush(buffer);
391 builder.append(written);
392 buffer.compact();
393 if (!endPoint.isBlocking())
394 {
395 while (buffer.space() == 0)
396 {
397 boolean ready = endPoint.blockWritable(getWriteTimeout());
398 if (!ready)
399 throw new IOException("Write timeout");
400
401 written = endPoint.flush(buffer);
402 builder.append("+").append(written);
403 buffer.compact();
404 }
405 }
406 _logger.debug("Written {}/{} bytes {}", builder, length, endPoint);
407 return length;
408 }
409
410 private class Manager extends SelectorManager
411 {
412 @Override
413 protected SelectChannelEndPoint newEndPoint(SocketChannel channel, SelectSet selectSet, SelectionKey selectionKey) throws IOException
414 {
415 SelectChannelEndPoint endp = new SelectChannelEndPoint(channel, selectSet, selectionKey);
416 endp.setMaxIdleTime(_writeTimeout);
417 return endp;
418 }
419
420 @Override
421 protected Connection newConnection(SocketChannel channel, SelectChannelEndPoint endpoint)
422 {
423 ProxyToServerConnection proxyToServer = (ProxyToServerConnection)endpoint.getSelectionKey().attachment();
424 proxyToServer = new ProxyToServerConnection(endpoint,proxyToServer);
425 return proxyToServer;
426 }
427
428 @Override
429 protected void endPointOpened(SelectChannelEndPoint endpoint)
430 {
431 ProxyToServerConnection proxyToServer = (ProxyToServerConnection)endpoint.getSelectionKey().attachment();
432 proxyToServer.ready();
433 }
434
435 @Override
436 public boolean dispatch(Runnable task)
437 {
438 return _threadPool.dispatch(task);
439 }
440
441 @Override
442 protected void endPointClosed(SelectChannelEndPoint endpoint)
443 {
444 }
445
446 @Override
447 protected void endPointUpgraded(ConnectedEndPoint endpoint, Connection oldConnection)
448 {
449 }
450 }
451
452 public class ProxyToServerConnection extends AbstractConnection implements Connection
453 {
454 private final CountDownLatch _ready ;
455 private final Buffer _buffer;
456 private final ConcurrentMap<String, Object> _context;
457 private final ClientToProxyConnection _toClient;
458 private volatile Buffer _data;
459
460 public ProxyToServerConnection(ConcurrentMap<String, Object> context, Buffer data, ClientToProxyConnection toClient)
461 {
462 super(null);
463 _context = context;
464 _data = data;
465 _ready = new CountDownLatch(1);
466 _buffer = new IndirectNIOBuffer(1024);
467 _toClient=toClient;
468 }
469
470 public ProxyToServerConnection(EndPoint endp,ProxyToServerConnection connection)
471 {
472 super(endp);
473 _ready=connection._ready;
474 _buffer=connection._buffer;
475 _context=connection._context;
476 _data=connection._data;
477 _toClient=connection._toClient;
478 if (_toClient!=null)
479 _toClient.setConnection(this);
480
481 }
482
483 @Override
484 public String toString()
485 {
486 StringBuilder builder = new StringBuilder("ProxyToServer");
487 builder.append("(:").append(_endp.getLocalPort());
488 builder.append("<=>:").append(_endp.getRemotePort());
489 return builder.append(")").toString();
490 }
491
492 public Connection handle() throws IOException
493 {
494 _logger.debug("{}: begin reading from server", this);
495 try
496 {
497 if (_data != null)
498 {
499 int written = write(_endp, _data, _context);
500 _logger.debug("{}: written to server {} bytes", this, written);
501 _data = null;
502 }
503
504 while (true)
505 {
506 int read = read(_endp, _buffer, _context);
507
508 if (read == -1)
509 {
510 _logger.debug("{}: server closed connection {}", this, _endp);
511 close();
512 break;
513 }
514
515 if (read == 0)
516 break;
517
518 _logger.debug("{}: read from server {} bytes {}", this, read, _endp);
519 int written = write(_toClient.getEndPoint(), _buffer, _context);
520 _logger.debug("{}: written to {} {} bytes", this, _toClient, written);
521 }
522 return this;
523 }
524 catch (ClosedChannelException x)
525 {
526 _logger.debug(x);
527 throw x;
528 }
529 catch (IOException x)
530 {
531 _logger.warn(this + ": unexpected exception", x);
532 close();
533 throw x;
534 }
535 catch (RuntimeException x)
536 {
537 _logger.warn(this + ": unexpected exception", x);
538 close();
539 throw x;
540 }
541 finally
542 {
543 _logger.debug("{}: end reading from server", this);
544 }
545 }
546
547 public boolean isIdle()
548 {
549 return false;
550 }
551
552 public boolean isSuspended()
553 {
554 return false;
555 }
556
557 public void closed()
558 {
559 }
560
561 public void ready()
562 {
563 _ready.countDown();
564 }
565
566 public void waitReady(long timeout) throws IOException
567 {
568 try
569 {
570 _ready.await(timeout, TimeUnit.MILLISECONDS);
571 }
572 catch (final InterruptedException x)
573 {
574 throw new IOException(){{initCause(x);}};
575 }
576 }
577
578 public void closeClient() throws IOException
579 {
580 _toClient.closeClient();
581 }
582
583 public void closeServer() throws IOException
584 {
585 _endp.close();
586 }
587
588 public void close()
589 {
590 try
591 {
592 closeClient();
593 }
594 catch (IOException x)
595 {
596 _logger.debug(this + ": unexpected exception closing the client", x);
597 }
598
599 try
600 {
601 closeServer();
602 }
603 catch (IOException x)
604 {
605 _logger.debug(this + ": unexpected exception closing the server", x);
606 }
607 }
608 }
609
610 public class ClientToProxyConnection extends AbstractConnection implements Connection
611 {
612 private final Buffer _buffer = new IndirectNIOBuffer(1024);
613 private final ConcurrentMap<String, Object> _context;
614 private final SocketChannel _channel;
615 private volatile ProxyToServerConnection _toServer;
616 private boolean _firstTime = true;
617
618 public ClientToProxyConnection(ConcurrentMap<String, Object> context, SocketChannel channel, EndPoint endPoint, long timestamp)
619 {
620 super(endPoint,timestamp);
621 _context = context;
622 _channel = channel;
623 }
624
625 @Override
626 public String toString()
627 {
628 StringBuilder builder = new StringBuilder("ClientToProxy");
629 builder.append("(:").append(_endp.getLocalPort());
630 builder.append("<=>:").append(_endp.getRemotePort());
631 return builder.append(")").toString();
632 }
633
634 public Connection handle() throws IOException
635 {
636 _logger.debug("{}: begin reading from client", this);
637 try
638 {
639 if (_firstTime)
640 {
641 _firstTime = false;
642 register(_channel, _toServer);
643 _logger.debug("{}: registered channel {} with connection {}", this, _channel, _toServer);
644 }
645
646 while (true)
647 {
648 int read = read(_endp, _buffer, _context);
649
650 if (read == -1)
651 {
652 _logger.debug("{}: client closed connection {}", this, _endp);
653 close();
654 break;
655 }
656
657 if (read == 0)
658 break;
659
660 _logger.debug("{}: read from client {} bytes {}", this, read, _endp);
661 int written = write(_toServer.getEndPoint(), _buffer, _context);
662 _logger.debug("{}: written to {} {} bytes", this, _toServer, written);
663 }
664 return this;
665 }
666 catch (ClosedChannelException x)
667 {
668 _logger.debug(x);
669 closeServer();
670 throw x;
671 }
672 catch (IOException x)
673 {
674 _logger.warn(this + ": unexpected exception", x);
675 close();
676 throw x;
677 }
678 catch (RuntimeException x)
679 {
680 _logger.warn(this + ": unexpected exception", x);
681 close();
682 throw x;
683 }
684 finally
685 {
686 _logger.debug("{}: end reading from client", this);
687 }
688 }
689
690 public boolean isIdle()
691 {
692 return false;
693 }
694
695 public boolean isSuspended()
696 {
697 return false;
698 }
699
700 public void closed()
701 {
702 }
703
704 public void setConnection(ProxyToServerConnection connection)
705 {
706 _toServer = connection;
707 }
708
709 public void closeClient() throws IOException
710 {
711 _endp.close();
712 }
713
714 public void closeServer() throws IOException
715 {
716 _toServer.closeServer();
717 }
718
719 public void close()
720 {
721 try
722 {
723 closeClient();
724 }
725 catch (IOException x)
726 {
727 _logger.debug(this + ": unexpected exception closing the client", x);
728 }
729
730 try
731 {
732 closeServer();
733 }
734 catch (IOException x)
735 {
736 _logger.debug(this + ": unexpected exception closing the server", x);
737 }
738 }
739 }
740
741
742
743
744
745
746
747 public void addWhite(String entry)
748 {
749 add(entry, _white);
750 }
751
752
753
754
755
756
757
758 public void addBlack(String entry)
759 {
760 add(entry, _black);
761 }
762
763
764
765
766
767
768
769 public void setWhite(String[] entries)
770 {
771 set(entries, _white);
772 }
773
774
775
776
777
778
779
780 public void setBlack(String[] entries)
781 {
782 set(entries, _black);
783 }
784
785
786
787
788
789
790
791
792
793 protected void set(String[] entries, HostMap<String> hostMap)
794 {
795 hostMap.clear();
796
797 if (entries != null && entries.length > 0)
798 {
799 for (String addrPath:entries)
800 {
801 add(addrPath, hostMap);
802 }
803 }
804 }
805
806
807
808
809
810
811
812
813
814 private void add(String entry, HostMap<String> hostMap)
815 {
816 if (entry != null && entry.length() > 0)
817 {
818 entry = entry.trim();
819 if (hostMap.get(entry) == null)
820 {
821 hostMap.put(entry,entry);
822 }
823 }
824 }
825
826
827
828
829
830
831
832
833 public boolean validateDestination(String host)
834 {
835 if (_white.size()>0)
836 {
837 Object whiteObj = _white.getLazyMatches(host);
838 if (whiteObj == null)
839 {
840 return false;
841 }
842 }
843
844 if (_black.size() > 0)
845 {
846 Object blackObj = _black.getLazyMatches(host);
847 if (blackObj != null)
848 {
849 return false;
850 }
851 }
852
853 return true;
854 }
855
856 @Override
857 public void dump(Appendable out, String indent) throws IOException
858 {
859 dumpThis(out);
860 if (_privateThreadPool)
861 dump(out,indent,Arrays.asList(new Object[]{_threadPool,_selectorManager}),TypeUtil.asList(getHandlers()),getBeans());
862 else
863 dump(out,indent,Collections.singletonList(_selectorManager),TypeUtil.asList(getHandlers()),getBeans());
864 }
865
866
867 }