1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.eclipse.jetty.proxy;
20
21 import java.io.Closeable;
22 import java.io.IOException;
23 import java.net.InetSocketAddress;
24 import java.nio.ByteBuffer;
25 import java.nio.channels.SelectionKey;
26 import java.nio.channels.SocketChannel;
27 import java.util.HashSet;
28 import java.util.Set;
29 import java.util.concurrent.ConcurrentHashMap;
30 import java.util.concurrent.ConcurrentMap;
31 import java.util.concurrent.Executor;
32
33 import javax.servlet.AsyncContext;
34 import javax.servlet.ServletException;
35 import javax.servlet.http.HttpServletRequest;
36 import javax.servlet.http.HttpServletResponse;
37
38 import org.eclipse.jetty.http.HttpHeader;
39 import org.eclipse.jetty.http.HttpHeaderValue;
40 import org.eclipse.jetty.http.HttpMethod;
41 import org.eclipse.jetty.io.ByteBufferPool;
42 import org.eclipse.jetty.io.Connection;
43 import org.eclipse.jetty.io.EndPoint;
44 import org.eclipse.jetty.io.MappedByteBufferPool;
45 import org.eclipse.jetty.io.SelectChannelEndPoint;
46 import org.eclipse.jetty.io.SelectorManager;
47 import org.eclipse.jetty.server.Handler;
48 import org.eclipse.jetty.server.HttpConnection;
49 import org.eclipse.jetty.server.HttpTransport;
50 import org.eclipse.jetty.server.Request;
51 import org.eclipse.jetty.server.handler.HandlerWrapper;
52 import org.eclipse.jetty.util.BufferUtil;
53 import org.eclipse.jetty.util.Callback;
54 import org.eclipse.jetty.util.Promise;
55 import org.eclipse.jetty.util.TypeUtil;
56 import org.eclipse.jetty.util.log.Log;
57 import org.eclipse.jetty.util.log.Logger;
58 import org.eclipse.jetty.util.thread.ScheduledExecutorScheduler;
59 import org.eclipse.jetty.util.thread.Scheduler;
60
61
62
63
64 public class ConnectHandler extends HandlerWrapper
65 {
66 protected static final Logger LOG = Log.getLogger(ConnectHandler.class);
67
68 private final Set<String> whiteList = new HashSet<>();
69 private final Set<String> blackList = new HashSet<>();
70 private Executor executor;
71 private Scheduler scheduler;
72 private ByteBufferPool bufferPool;
73 private SelectorManager selector;
74 private long connectTimeout = 15000;
75 private long idleTimeout = 30000;
76 private int bufferSize = 4096;
77
78 public ConnectHandler()
79 {
80 this(null);
81 }
82
83 public ConnectHandler(Handler handler)
84 {
85 setHandler(handler);
86 }
87
88 public Executor getExecutor()
89 {
90 return executor;
91 }
92
93 public void setExecutor(Executor executor)
94 {
95 this.executor = executor;
96 }
97
98 public Scheduler getScheduler()
99 {
100 return scheduler;
101 }
102
103 public void setScheduler(Scheduler scheduler)
104 {
105 this.scheduler = scheduler;
106 }
107
108 public ByteBufferPool getByteBufferPool()
109 {
110 return bufferPool;
111 }
112
113 public void setByteBufferPool(ByteBufferPool bufferPool)
114 {
115 this.bufferPool = bufferPool;
116 }
117
118
119
120
121 public long getConnectTimeout()
122 {
123 return connectTimeout;
124 }
125
126
127
128
129 public void setConnectTimeout(long connectTimeout)
130 {
131 this.connectTimeout = connectTimeout;
132 }
133
134
135
136
137 public long getIdleTimeout()
138 {
139 return idleTimeout;
140 }
141
142
143
144
145 public void setIdleTimeout(long idleTimeout)
146 {
147 this.idleTimeout = idleTimeout;
148 }
149
150 public int getBufferSize()
151 {
152 return bufferSize;
153 }
154
155 public void setBufferSize(int bufferSize)
156 {
157 this.bufferSize = bufferSize;
158 }
159
160 @Override
161 protected void doStart() throws Exception
162 {
163 if (executor == null)
164 executor = getServer().getThreadPool();
165
166 if (scheduler == null)
167 addBean(scheduler = new ScheduledExecutorScheduler());
168
169 if (bufferPool == null)
170 addBean(bufferPool = new MappedByteBufferPool());
171
172 addBean(selector = newSelectorManager());
173 selector.setConnectTimeout(getConnectTimeout());
174
175 super.doStart();
176 }
177
178 protected SelectorManager newSelectorManager()
179 {
180 return new ConnectManager(getExecutor(), getScheduler(), 1);
181 }
182
183 @Override
184 public void handle(String target, Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException
185 {
186 if (HttpMethod.CONNECT.is(request.getMethod()))
187 {
188 String serverAddress = request.getRequestURI();
189 if (LOG.isDebugEnabled())
190 LOG.debug("CONNECT request for {}", serverAddress);
191
192 handleConnect(baseRequest, request, response, serverAddress);
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 protected void handleConnect(Request baseRequest, final HttpServletRequest request, final HttpServletResponse response, String serverAddress)
211 {
212 baseRequest.setHandled(true);
213 try
214 {
215 boolean proceed = handleAuthentication(request, response, serverAddress);
216 if (!proceed)
217 {
218 if (LOG.isDebugEnabled())
219 LOG.debug("Missing proxy authentication");
220 sendConnectResponse(request, response, HttpServletResponse.SC_PROXY_AUTHENTICATION_REQUIRED);
221 return;
222 }
223
224 String host = serverAddress;
225 int port = 80;
226 int colon = serverAddress.indexOf(':');
227 if (colon > 0)
228 {
229 host = serverAddress.substring(0, colon);
230 port = Integer.parseInt(serverAddress.substring(colon + 1));
231 }
232
233 if (!validateDestination(host, port))
234 {
235 if (LOG.isDebugEnabled())
236 LOG.debug("Destination {}:{} forbidden", host, port);
237 sendConnectResponse(request, response, HttpServletResponse.SC_FORBIDDEN);
238 return;
239 }
240
241 final HttpTransport transport = baseRequest.getHttpChannel().getHttpTransport();
242
243 if (!(transport instanceof HttpConnection))
244 {
245 if (LOG.isDebugEnabled())
246 LOG.debug("CONNECT not supported for {}", transport);
247 sendConnectResponse(request, response, HttpServletResponse.SC_FORBIDDEN);
248 return;
249 }
250
251 final AsyncContext asyncContext = request.startAsync();
252 asyncContext.setTimeout(0);
253
254 if (LOG.isDebugEnabled())
255 LOG.debug("Connecting to {}:{}", host, port);
256
257 connectToServer(request, host, port, new Promise<SocketChannel>()
258 {
259 @Override
260 public void succeeded(SocketChannel channel)
261 {
262 ConnectContext connectContext = new ConnectContext(request, response, asyncContext, (HttpConnection)transport);
263 if (channel.isConnected())
264 selector.accept(channel, connectContext);
265 else
266 selector.connect(channel, connectContext);
267 }
268
269 @Override
270 public void failed(Throwable x)
271 {
272 onConnectFailure(request, response, asyncContext, x);
273 }
274 });
275 }
276 catch (Exception x)
277 {
278 onConnectFailure(request, response, null, x);
279 }
280 }
281
282 protected void connectToServer(HttpServletRequest request, String host, int port, Promise<SocketChannel> promise)
283 {
284 SocketChannel channel = null;
285 try
286 {
287 channel = SocketChannel.open();
288 channel.socket().setTcpNoDelay(true);
289 channel.configureBlocking(false);
290 InetSocketAddress address = newConnectAddress(host, port);
291 channel.connect(address);
292 promise.succeeded(channel);
293 }
294 catch (Throwable x)
295 {
296 close(channel);
297 promise.failed(x);
298 }
299 }
300
301 private void close(Closeable closeable)
302 {
303 try
304 {
305 if (closeable != null)
306 closeable.close();
307 }
308 catch (Throwable x)
309 {
310 LOG.ignore(x);
311 }
312 }
313
314
315
316
317
318
319
320
321 protected InetSocketAddress newConnectAddress(String host, int port)
322 {
323 return new InetSocketAddress(host, port);
324 }
325
326 protected void onConnectSuccess(ConnectContext connectContext, UpstreamConnection upstreamConnection)
327 {
328 ConcurrentMap<String, Object> context = connectContext.getContext();
329 HttpServletRequest request = connectContext.getRequest();
330 prepareContext(request, context);
331
332 HttpConnection httpConnection = connectContext.getHttpConnection();
333 EndPoint downstreamEndPoint = httpConnection.getEndPoint();
334 DownstreamConnection downstreamConnection = newDownstreamConnection(downstreamEndPoint, context, BufferUtil.EMPTY_BUFFER);
335 downstreamConnection.setInputBufferSize(getBufferSize());
336
337 upstreamConnection.setConnection(downstreamConnection);
338 downstreamConnection.setConnection(upstreamConnection);
339 if (LOG.isDebugEnabled())
340 LOG.debug("Connection setup completed: {}<->{}", downstreamConnection, upstreamConnection);
341
342 HttpServletResponse response = connectContext.getResponse();
343 sendConnectResponse(request, response, HttpServletResponse.SC_OK);
344
345 upgradeConnection(request, response, downstreamConnection);
346
347 connectContext.getAsyncContext().complete();
348 }
349
350 protected void onConnectFailure(HttpServletRequest request, HttpServletResponse response, AsyncContext asyncContext, Throwable failure)
351 {
352 if (LOG.isDebugEnabled())
353 LOG.debug("CONNECT failed", failure);
354 sendConnectResponse(request, response, HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
355 if (asyncContext != null)
356 asyncContext.complete();
357 }
358
359 private void sendConnectResponse(HttpServletRequest request, HttpServletResponse response, int statusCode)
360 {
361 try
362 {
363 response.setStatus(statusCode);
364 if (statusCode != HttpServletResponse.SC_OK)
365 response.setHeader(HttpHeader.CONNECTION.asString(), HttpHeaderValue.CLOSE.asString());
366 response.getOutputStream().close();
367 if (LOG.isDebugEnabled())
368 LOG.debug("CONNECT response sent {} {}", request.getProtocol(), response.getStatus());
369 }
370 catch (IOException x)
371 {
372 if (LOG.isDebugEnabled())
373 LOG.debug("Could not send CONNECT response", x);
374 }
375 }
376
377
378
379
380
381
382
383
384
385
386 protected boolean handleAuthentication(HttpServletRequest request, HttpServletResponse response, String address)
387 {
388 return true;
389 }
390
391
392
393
394 @Deprecated
395 protected DownstreamConnection newDownstreamConnection(EndPoint endPoint, ConcurrentMap<String, Object> context, ByteBuffer buffer)
396 {
397 return newDownstreamConnection(endPoint, context);
398 }
399
400 protected DownstreamConnection newDownstreamConnection(EndPoint endPoint, ConcurrentMap<String, Object> context)
401 {
402 return new DownstreamConnection(endPoint, getExecutor(), getByteBufferPool(), context);
403 }
404
405 protected UpstreamConnection newUpstreamConnection(EndPoint endPoint, ConnectContext connectContext)
406 {
407 return new UpstreamConnection(endPoint, getExecutor(), getByteBufferPool(), connectContext);
408 }
409
410 protected void prepareContext(HttpServletRequest request, ConcurrentMap<String, Object> context)
411 {
412 }
413
414 private void upgradeConnection(HttpServletRequest request, HttpServletResponse response, Connection connection)
415 {
416
417
418 request.setAttribute(HttpConnection.UPGRADE_CONNECTION_ATTRIBUTE, connection);
419 response.setStatus(HttpServletResponse.SC_SWITCHING_PROTOCOLS);
420 if (LOG.isDebugEnabled())
421 LOG.debug("Upgraded connection to {}", connection);
422 }
423
424
425
426
427
428
429
430
431
432
433
434 protected int read(EndPoint endPoint, ByteBuffer buffer, ConcurrentMap<String, Object> context) throws IOException
435 {
436 int read = read(endPoint, buffer);
437 if (LOG.isDebugEnabled())
438 LOG.debug("{} read {} bytes", this, read);
439 return read;
440 }
441
442
443
444
445 @Deprecated
446 protected int read(EndPoint endPoint, ByteBuffer buffer) throws IOException
447 {
448 return endPoint.fill(buffer);
449 }
450
451
452
453
454
455
456
457
458 protected void write(EndPoint endPoint, ByteBuffer buffer, Callback callback, ConcurrentMap<String, Object> context)
459 {
460 if (LOG.isDebugEnabled())
461 LOG.debug("{} writing {} bytes", this, buffer.remaining());
462 write(endPoint, buffer, callback);
463 }
464
465
466
467
468 @Deprecated
469 protected void write(EndPoint endPoint, ByteBuffer buffer, Callback callback)
470 {
471 endPoint.write(callback, buffer);
472 }
473
474 public Set<String> getWhiteListHosts()
475 {
476 return whiteList;
477 }
478
479 public Set<String> getBlackListHosts()
480 {
481 return blackList;
482 }
483
484
485
486
487
488
489
490
491 public boolean validateDestination(String host, int port)
492 {
493 String hostPort = host + ":" + port;
494 if (!whiteList.isEmpty())
495 {
496 if (!whiteList.contains(hostPort))
497 {
498 if (LOG.isDebugEnabled())
499 LOG.debug("Host {}:{} not whitelisted", host, port);
500 return false;
501 }
502 }
503 if (!blackList.isEmpty())
504 {
505 if (blackList.contains(hostPort))
506 {
507 if (LOG.isDebugEnabled())
508 LOG.debug("Host {}:{} blacklisted", host, port);
509 return false;
510 }
511 }
512 return true;
513 }
514
515 @Override
516 public void dump(Appendable out, String indent) throws IOException
517 {
518 dumpThis(out);
519 dump(out, indent, getBeans(), TypeUtil.asList(getHandlers()));
520 }
521
522 protected class ConnectManager extends SelectorManager
523 {
524 protected ConnectManager(Executor executor, Scheduler scheduler, int selectors)
525 {
526 super(executor, scheduler, selectors);
527 }
528
529 @Override
530 protected EndPoint newEndPoint(SocketChannel channel, ManagedSelector selector, SelectionKey selectionKey) throws IOException
531 {
532 return new SelectChannelEndPoint(channel, selector, selectionKey, getScheduler(), getIdleTimeout());
533 }
534
535 @Override
536 public Connection newConnection(SocketChannel channel, EndPoint endpoint, Object attachment) throws IOException
537 {
538 if (ConnectHandler.LOG.isDebugEnabled())
539 ConnectHandler.LOG.debug("Connected to {}", channel.getRemoteAddress());
540 ConnectContext connectContext = (ConnectContext)attachment;
541 UpstreamConnection connection = newUpstreamConnection(endpoint, connectContext);
542 connection.setInputBufferSize(getBufferSize());
543 return connection;
544 }
545
546 @Override
547 protected void connectionFailed(SocketChannel channel, final Throwable ex, final Object attachment)
548 {
549 close(channel);
550 getExecutor().execute(new Runnable()
551 {
552 public void run()
553 {
554 ConnectContext connectContext = (ConnectContext)attachment;
555 onConnectFailure(connectContext.request, connectContext.response, connectContext.asyncContext, ex);
556 }
557 });
558 }
559 }
560
561 protected static class ConnectContext
562 {
563 private final ConcurrentMap<String, Object> context = new ConcurrentHashMap<>();
564 private final HttpServletRequest request;
565 private final HttpServletResponse response;
566 private final AsyncContext asyncContext;
567 private final HttpConnection httpConnection;
568
569 public ConnectContext(HttpServletRequest request, HttpServletResponse response, AsyncContext asyncContext, HttpConnection httpConnection)
570 {
571 this.request = request;
572 this.response = response;
573 this.asyncContext = asyncContext;
574 this.httpConnection = httpConnection;
575 }
576
577 public ConcurrentMap<String, Object> getContext()
578 {
579 return context;
580 }
581
582 public HttpServletRequest getRequest()
583 {
584 return request;
585 }
586
587 public HttpServletResponse getResponse()
588 {
589 return response;
590 }
591
592 public AsyncContext getAsyncContext()
593 {
594 return asyncContext;
595 }
596
597 public HttpConnection getHttpConnection()
598 {
599 return httpConnection;
600 }
601 }
602
603 public class UpstreamConnection extends ProxyConnection
604 {
605 private ConnectContext connectContext;
606
607 public UpstreamConnection(EndPoint endPoint, Executor executor, ByteBufferPool bufferPool, ConnectContext connectContext)
608 {
609 super(endPoint, executor, bufferPool, connectContext.getContext());
610 this.connectContext = connectContext;
611 }
612
613 @Override
614 public void onOpen()
615 {
616 super.onOpen();
617 getExecutor().execute(new Runnable()
618 {
619 public void run()
620 {
621 onConnectSuccess(connectContext, UpstreamConnection.this);
622 fillInterested();
623 }
624 });
625 }
626
627 @Override
628 protected int read(EndPoint endPoint, ByteBuffer buffer) throws IOException
629 {
630 return ConnectHandler.this.read(endPoint, buffer, getContext());
631 }
632
633 @Override
634 protected void write(EndPoint endPoint, ByteBuffer buffer,Callback callback)
635 {
636 ConnectHandler.this.write(endPoint, buffer, callback, getContext());
637 }
638 }
639
640 public class DownstreamConnection extends ProxyConnection implements Connection.UpgradeTo
641 {
642 private ByteBuffer buffer;
643
644 public DownstreamConnection(EndPoint endPoint, Executor executor, ByteBufferPool bufferPool, ConcurrentMap<String, Object> context)
645 {
646 super(endPoint, executor, bufferPool, context);
647 }
648
649
650
651
652 @Deprecated
653 public DownstreamConnection(EndPoint endPoint, Executor executor, ByteBufferPool bufferPool, ConcurrentMap<String, Object> context, ByteBuffer buffer)
654 {
655 this(endPoint, executor, bufferPool, context);
656 }
657
658 @Override
659 public void onUpgradeTo(ByteBuffer buffer)
660 {
661 this.buffer = buffer == null ? BufferUtil.EMPTY_BUFFER : buffer;
662 }
663
664 @Override
665 public void onOpen()
666 {
667 super.onOpen();
668 final int remaining = buffer.remaining();
669 write(getConnection().getEndPoint(), buffer, new Callback()
670 {
671 @Override
672 public void succeeded()
673 {
674 if (LOG.isDebugEnabled())
675 LOG.debug("{} wrote initial {} bytes to server", DownstreamConnection.this, remaining);
676 fillInterested();
677 }
678
679 @Override
680 public void failed(Throwable x)
681 {
682 if (LOG.isDebugEnabled())
683 LOG.debug(this + " failed to write initial " + remaining + " bytes to server", x);
684 close();
685 getConnection().close();
686 }
687 });
688 }
689
690 @Override
691 protected int read(EndPoint endPoint, ByteBuffer buffer) throws IOException
692 {
693 return ConnectHandler.this.read(endPoint, buffer, getContext());
694 }
695
696 @Override
697 protected void write(EndPoint endPoint, ByteBuffer buffer, Callback callback)
698 {
699 ConnectHandler.this.write(endPoint, buffer, callback, getContext());
700 }
701 }
702 }