diff --git a/src/main/java/com/jcraft/jsch/ChannelForwardedTCPIP.java b/src/main/java/com/jcraft/jsch/ChannelForwardedTCPIP.java new file mode 100644 index 0000000..b2dccec --- /dev/null +++ b/src/main/java/com/jcraft/jsch/ChannelForwardedTCPIP.java @@ -0,0 +1,351 @@ +/* + * Copyright (c) 2002-2018 ymnk, JCraft,Inc. All rights reserved. + * + * Redistribution and use in source and binary forms, with or without modification, are permitted + * provided that the following conditions are met: + * + * 1. Redistributions of source code must retain the above copyright notice, this list of conditions + * and the following disclaimer. + * + * 2. Redistributions in binary form must reproduce the above copyright notice, this list of + * conditions and the following disclaimer in the documentation and/or other materials provided with + * the distribution. + * + * 3. The names of the authors may not be used to endorse or promote products derived from this + * software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE + * DISCLAIMED. IN NO EVENT SHALL JCRAFT, INC. OR ANY CONTRIBUTORS TO THIS SOFTWARE BE LIABLE FOR ANY + * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR + * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT + * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS + * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ + +/* + * VENDORED from com.github.mwiede:jsch 2.28.8 (same license as above), with ONE change: the local + * window of forwarded-tcpip channels is 2 MB instead of 128 KB, and packets 32 KB instead of 16 KB. + * + * Every connection through the tunnel server's port 2010 is one of these channels, and its local + * window is how much the server may send before this client acknowledges it: request bodies, + * uploads, everything a device sends towards the site under test. At 128 KB that capped each + * connection at 128 KB per round trip -- 3 MB/s at 20 ms to the tunnel server, 1.3 MB/s at 50 ms. + * With 2 MB, measured end to end against a real tunnel VM: 12.7-30 MB/s at 20 ms, 5-14 MB/s at + * 50 ms; downloads, latency and CPU unchanged. JSch has no setting for it: the constants are + * private, and Channel#setLocalWindowSizeMax (public since 2.28.8) only works before a channel is + * connected, while these are created and confirmed by Session when the server opens them. + * + * The project's classes come first on the classpath and in the shaded jar, so this replaces + * JSch's. JschForwardedWindowTest fails when the JSch version changes: copy that version's + * ChannelForwardedTCPIP over this file and re-apply the two constants (or drop this file once JSch + * makes the window configurable). + */ +package com.jcraft.jsch; + +import java.io.PipedOutputStream; +import java.net.Socket; +import java.util.Vector; + +public class ChannelForwardedTCPIP extends Channel { + + private static Vector pool = new Vector<>(); + + static final int LOCAL_WINDOW_SIZE_MAX = 0x200000; // TestingBot: was 0x20000 + static final int LOCAL_MAXIMUM_PACKET_SIZE = 0x8000; // TestingBot: was 0x4000 + + private static final int TIMEOUT = 10 * 1000; + + private Socket socket = null; + private ForwardedTCPIPDaemon daemon = null; + private Config config = null; + + ChannelForwardedTCPIP() { + super(); + lwsize_max = LOCAL_WINDOW_SIZE_MAX; + lwsize = LOCAL_WINDOW_SIZE_MAX; + lmpsize = LOCAL_MAXIMUM_PACKET_SIZE; + io = new IO(); + connected = true; + } + + @Override + public void run() { + try { + if (config instanceof ConfigDaemon) { + ConfigDaemon _config = (ConfigDaemon) config; + Class c = + Class.forName(_config.target).asSubclass(ForwardedTCPIPDaemon.class); + daemon = c.getDeclaredConstructor().newInstance(); + + PipedOutputStream out = new PipedOutputStream(); + io.setInputStream(new PassiveInputStream(out, 32 * 1024), false); + + daemon.setChannel(this, getInputStream(), out); + daemon.setArg(_config.arg); + getSession().getThreadFactory().newThread(daemon).start(); + } else { + ConfigLHost _config = (ConfigLHost) config; + socket = + (_config.factory == null) ? Util.createSocket(_config.target, _config.lport, TIMEOUT) + : _config.factory.createSocket(_config.target, _config.lport); + socket.setTcpNoDelay(true); + io.setInputStream(socket.getInputStream()); + io.setOutputStream(socket.getOutputStream()); + } + sendOpenConfirmation(); + } catch (Exception e) { + sendOpenFailure(SSH_OPEN_ADMINISTRATIVELY_PROHIBITED); + close = true; + disconnect(); + return; + } + + thread = Thread.currentThread(); + Buffer buf = new Buffer(rmpsize); + Packet packet = new Packet(buf); + int i = 0; + try { + Session _session = getSession(); + while (thread != null && io != null && io.in != null) { + i = io.in.read(buf.buffer, 14, buf.buffer.length - 14 - _session.getBufferMargin()); + if (i <= 0) { + eof(); + break; + } + packet.reset(); + buf.putByte((byte) Session.SSH_MSG_CHANNEL_DATA); + buf.putInt(recipient); + buf.putInt(i); + buf.skip(i); + synchronized (this) { + if (close) + break; + _session.write(packet, this, i); + } + } + } catch (Exception e) { + // System.err.println(e); + } + // thread=null; + // eof(); + disconnect(); + } + + @Override + void getData(Buffer buf) { + setRecipient(buf.getInt()); + setRemoteWindowSize(buf.getUInt()); + setRemotePacketSize(buf.getInt()); + byte[] addr = buf.getString(); + int port = buf.getInt(); + byte[] orgaddr = buf.getString(); + int orgport = buf.getInt(); + + /* + * System.err.println("addr: "+Util.byte2str(addr)); System.err.println("port: "+port); + * System.err.println("orgaddr: "+Util.byte2str(orgaddr)); + * System.err.println("orgport: "+orgport); + */ + + Session _session = null; + try { + _session = getSession(); + } catch (JSchException e) { + // session has been already down. + } + + this.config = getPort(_session, Util.byte2str(addr), port); + if (this.config == null) + this.config = getPort(_session, null, port); + + if (this.config == null) { + if (_session.getLogger().isEnabled(Logger.ERROR)) { + _session.getLogger().log(Logger.ERROR, + "ChannelForwardedTCPIP: " + Util.byte2str(addr) + ":" + port + " is not registered."); + } + } + } + + private static Config getPort(Session session, String address_to_bind, int rport) { + synchronized (pool) { + for (int i = 0; i < pool.size(); i++) { + Config bar = pool.elementAt(i); + if (bar.session != session) + continue; + if (bar.rport != rport) { + if (bar.rport != 0 || bar.allocated_rport != rport) + continue; + } + if (address_to_bind != null && !bar.address_to_bind.equals(address_to_bind)) + continue; + return bar; + } + return null; + } + } + + static String[] getPortForwarding(Session session) { + Vector foo = new Vector<>(); + synchronized (pool) { + for (int i = 0; i < pool.size(); i++) { + Config config = pool.elementAt(i); + if (config.session == session) { + if (config instanceof ConfigDaemon) + foo.addElement(config.allocated_rport + ":" + config.target + ":"); + else + foo.addElement( + config.allocated_rport + ":" + config.target + ":" + ((ConfigLHost) config).lport); + } + } + } + String[] bar = new String[foo.size()]; + for (int i = 0; i < foo.size(); i++) { + bar[i] = foo.elementAt(i); + } + return bar; + } + + static String normalize(String address) { + if (address == null) { + return "localhost"; + } else if (address.length() == 0 || address.equals("*")) { + return ""; + } else { + return address; + } + } + + static void addPort(Session session, String _address_to_bind, int port, int allocated_port, + String target, int lport, SocketFactory factory) throws JSchException { + String address_to_bind = normalize(_address_to_bind); + synchronized (pool) { + if (getPort(session, address_to_bind, port) != null) { + throw new JSchException("PortForwardingR: remote port " + port + " is already registered."); + } + ConfigLHost config = new ConfigLHost(); + config.session = session; + config.rport = port; + config.allocated_rport = allocated_port; + config.target = target; + config.lport = lport; + config.address_to_bind = address_to_bind; + config.factory = factory; + pool.addElement(config); + } + } + + static void addPort(Session session, String _address_to_bind, int port, int allocated_port, + String daemon, Object[] arg) throws JSchException { + String address_to_bind = normalize(_address_to_bind); + synchronized (pool) { + if (getPort(session, address_to_bind, port) != null) { + throw new JSchException("PortForwardingR: remote port " + port + " is already registered."); + } + ConfigDaemon config = new ConfigDaemon(); + config.session = session; + config.rport = port; + config.allocated_rport = port; + config.target = daemon; + config.arg = arg; + config.address_to_bind = address_to_bind; + pool.addElement(config); + } + } + + static void delPort(ChannelForwardedTCPIP c) { + Session _session = null; + try { + _session = c.getSession(); + } catch (JSchException e) { + // session has been already down. + } + if (_session != null && c.config != null) + delPort(_session, c.config.rport); + } + + static void delPort(Session session, int rport) { + delPort(session, null, rport); + } + + static void delPort(Session session, String address_to_bind, int rport) { + synchronized (pool) { + Config foo = getPort(session, normalize(address_to_bind), rport); + if (foo == null) + foo = getPort(session, null, rport); + if (foo == null) + return; + pool.removeElement(foo); + if (address_to_bind == null) { + address_to_bind = foo.address_to_bind; + } + if (address_to_bind == null) { + address_to_bind = "0.0.0.0"; + } + } + + Buffer buf = new Buffer(200); // ?? + Packet packet = new Packet(buf); + + try { + // byte SSH_MSG_GLOBAL_REQUEST 80 + // string "cancel-tcpip-forward" + // boolean want_reply + // string address_to_bind (e.g. "127.0.0.1") + // uint32 port number to bind + packet.reset(); + buf.putByte((byte) 80 /* SSH_MSG_GLOBAL_REQUEST */); + buf.putString(Util.str2byte("cancel-tcpip-forward")); + buf.putByte((byte) 0); + buf.putString(Util.str2byte(address_to_bind)); + buf.putInt(rport); + session.write(packet); + } catch (Exception e) { + // throw new JSchException(e.toString(), e); + } + } + + static void delPort(Session session) { + int[] rport = null; + int count = 0; + synchronized (pool) { + rport = new int[pool.size()]; + for (int i = 0; i < pool.size(); i++) { + Config config = pool.elementAt(i); + if (config.session == session) { + rport[count++] = config.rport; // ((Integer)bar[1]).intValue(); + } + } + } + for (int i = 0; i < count; i++) { + delPort(session, rport[i]); + } + } + + public int getRemotePort() { + return (config != null ? config.rport : 0); + } + + private void setSocketFactory(SocketFactory factory) { + if (config != null && (config instanceof ConfigLHost)) + ((ConfigLHost) config).factory = factory; + } + + abstract static class Config { + Session session; + int rport; + int allocated_rport; + String address_to_bind; + String target; + } + + static class ConfigDaemon extends Config { + Object[] arg; + } + + static class ConfigLHost extends Config { + int lport; + SocketFactory factory; + } +} diff --git a/src/main/java/ssh/SSHTunnel.java b/src/main/java/ssh/SSHTunnel.java index 01722a7..648c58e 100644 --- a/src/main/java/ssh/SSHTunnel.java +++ b/src/main/java/ssh/SSHTunnel.java @@ -21,6 +21,8 @@ import java.util.Timer; import java.util.TimerTask; import java.util.UUID; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.FutureTask; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.atomic.AtomicBoolean; import java.util.logging.Level; @@ -476,7 +478,7 @@ private int bindMuxPort() { for (int i = 0; i < MUX_PORT_COUNT; i++) { int port = muxFirstPort + (slot + i) % MUX_PORT_COUNT; try { - session.setPortForwardingR(port, REVERSE_FORWARD_HOST, app.getJettyPort()); + setPortForwardingR(port); return port; } catch (JSchException taken) { Logger.getLogger(SSHTunnel.class.getName()).log(Level.FINE, @@ -583,9 +585,53 @@ private void startExtra() { } } + /** + * Registers a reverse forward on a thread of its own. + * + *

JSch wakes the thread waiting for the server's reply to {@code tcpip-forward} by + * interrupting it, and sets the reply first. When the reply is quick -- always, on loopback -- + * the caller can see it, return, and receive the interrupt afterwards. The caller was left + * interrupted, so its next blocking call failed: the probe that follows a re-registration + * returned false at once, and {@code boot()} failed starting Jetty. A thread that ends here + * absorbs the interrupt instead. + * + *

The wait is uninterruptible, as JSch's own is, and bounded by JSch's ten-second reply + * timeout; an interrupt that arrives meanwhile is restored rather than failing the forward. + */ + private void setPortForwardingR(int port) throws JSchException { + Session current = session; + FutureTask request = new FutureTask<>(() -> { + current.setPortForwardingR(port, REVERSE_FORWARD_HOST, app.getJettyPort()); + return null; + }); + Thread requester = new Thread(request, "tcpip-forward-" + connectionId); + requester.setDaemon(true); + requester.start(); + boolean interrupted = false; + try { + while (true) { + try { + request.get(); + return; + } catch (InterruptedException again) { + interrupted = true; + } + } + } catch (ExecutionException failed) { + if (failed.getCause() instanceof JSchException jschFailure) { + throw jschFailure; + } + throw new JSchException(String.valueOf(failed.getCause()), failed.getCause()); + } finally { + if (interrupted) { + Thread.currentThread().interrupt(); + } + } + } + private boolean establishLegacyReverseForward() { try { - session.setPortForwardingR(remoteProxyPort, REVERSE_FORWARD_HOST, app.getJettyPort()); + setPortForwardingR(remoteProxyPort); boundRemotePort = remoteProxyPort; return true; } catch (JSchException ex) { diff --git a/src/test/java/com/jcraft/jsch/JschForwardedWindowTest.java b/src/test/java/com/jcraft/jsch/JschForwardedWindowTest.java new file mode 100644 index 0000000..e47d90e --- /dev/null +++ b/src/test/java/com/jcraft/jsch/JschForwardedWindowTest.java @@ -0,0 +1,30 @@ +package com.jcraft.jsch; + +import static org.assertj.core.api.Assertions.assertThat; + +import org.junit.jupiter.api.Test; + +/** + * Guards the vendored ChannelForwardedTCPIP (src/main/java/com/jcraft/jsch): it must be the class + * that is loaded, with the larger window, and it must come from the JSch version it was copied from. + */ +class JschForwardedWindowTest { + + /** The JSch release ChannelForwardedTCPIP.java was copied from. */ + private static final String VENDORED_FROM = "2.28.8"; + + @Test + void theVendoredClassIsTheOneLoaded() { + ChannelForwardedTCPIP channel = new ChannelForwardedTCPIP(); + + assertThat(channel.getLocalWindowSizeMax()).isEqualTo(0x200000); + assertThat(channel.lmpsize).isEqualTo(0x8000); + } + + @Test + void jschIsStillTheVersionItWasCopiedFrom() { + // A failure here is a JSch upgrade: copy the new version's ChannelForwardedTCPIP over the + // vendored one, re-apply the two constants, and update VENDORED_FROM. + assertThat(JSch.VERSION).isEqualTo(VENDORED_FROM); + } +}