diff --git a/src/main/java/com/testingbot/tunnel/App.java b/src/main/java/com/testingbot/tunnel/App.java index d482062..34930cd 100644 --- a/src/main/java/com/testingbot/tunnel/App.java +++ b/src/main/java/com/testingbot/tunnel/App.java @@ -93,6 +93,10 @@ public class App { com.testingbot.tunnel.proxy.NegotiateHosts.none(); /** Null means "keep the built-in default", so an absent flag changes nothing. */ private Integer httpDialTimeoutSeconds = null; + /** SSH connections to the tunnel server; see {@code --ssh-connections}. */ + private int sshConnections = DEFAULT_SSH_CONNECTIONS; + static final int DEFAULT_SSH_CONNECTIONS = 4; + static final int MAX_SSH_CONNECTIONS = 8; private Integer httpIdleTimeoutSeconds = null; private String controlProxy = null; private String controlProxyAuth = null; @@ -426,6 +430,16 @@ static Options buildOptions() { krb5Hosts.setArgName("HOST,HOST"); options.addOption(krb5Hosts); + Option sshConnections = new Option(null, "ssh-connections", true, + "How many SSH connections to open to the tunnel server, 1-" + MAX_SSH_CONNECTIONS + + " (default " + DEFAULT_SSH_CONNECTIONS + "). Browser connections are spread over " + + "them, so on a lossy or distant link one lost packet no longer stalls every request " + + "and the tunnel is not held to one TCP connection's share of the link; when one " + + "connection drops the others keep carrying traffic. Tunnel servers that do not " + + "support this get a single connection, as do tunnels with 1."); + sshConnections.setArgName("n"); + options.addOption(sshConnections); + Option dialTimeout = new Option(null, "http-dial-timeout", true, "Seconds to wait for a TCP connection to a target or upstream proxy before giving " + "up (default 5 for the API, 15 for tunnelled connections). Lower it on a network " @@ -845,6 +859,19 @@ static void applyOptions(App app, CommandLine commandLine) throws Exception { } app.httpDialTimeoutSeconds = positiveSeconds(commandLine, "http-dial-timeout"); + if (commandLine.hasOption("ssh-connections")) { + String value = commandLine.getOptionValue("ssh-connections").trim(); + try { + int n = Integer.parseInt(value); + if (n < 1 || n > MAX_SSH_CONNECTIONS) { + throw new NumberFormatException(value); + } + app.sshConnections = n; + } catch (NumberFormatException invalid) { + throw new ParseException("Invalid --ssh-connections '" + value + + "': expected a whole number from 1 to " + MAX_SSH_CONNECTIONS); + } + } app.httpIdleTimeoutSeconds = positiveSeconds(commandLine, "http-idle-timeout"); if (commandLine.hasOption("dns")) { @@ -1537,6 +1564,11 @@ boolean startProxies() { } } + if (tunnel != null) { + // Before the self-test: on a server with tunnel-mux, port 2010 is only served once + // the server has reached this proxy through one of our forwards. + tunnel.settleForwarding(); + } return selfTest(); } @@ -1590,7 +1622,7 @@ boolean selfTest() { /** Package-private so a test can drive the retry without a live SSH session. */ boolean tunnelIsForwarding() { - return tunnel != null && tunnel.isForwardingEstablished(); + return tunnel != null && tunnel.isAnyConnectionForwarding(); } /** @@ -2403,6 +2435,18 @@ static Integer positiveSeconds(CommandLine commandLine, String option) throws Pa * up quickly, a tunnelled connection should not -- and one number for both would be worse * than either. */ + public int getSshConnections() { + return sshConnections; + } + + public void setSshConnections(int sshConnections) { + this.sshConnections = sshConnections; + } + + public boolean isNoProxy() { + return noProxy; + } + public Integer getHttpDialTimeoutSeconds() { return httpDialTimeoutSeconds; } diff --git a/src/main/java/ssh/CustomConnectionMonitor.java b/src/main/java/ssh/CustomConnectionMonitor.java index 6d38ded..f839b76 100644 --- a/src/main/java/ssh/CustomConnectionMonitor.java +++ b/src/main/java/ssh/CustomConnectionMonitor.java @@ -24,14 +24,19 @@ public class CustomConnectionMonitor { private static final Logger LOG = Logger.getLogger(CustomConnectionMonitor.class.getName()); - static final long CURRENT_RETRY_DELAY = 5000; - static final int MAX_RETRIES = 30; + /** + * Between reconnect attempts while the network is down: the first one that can get through + * should not wait long once it is back (5 s -> 2 s took 2010's return after a 20-second outage + * from 25.1 s to 23.6 s). MAX_RETRIES keeps the ~150 s before the tunnel is rebuilt. + */ + static final long CURRENT_RETRY_DELAY = 2000; + static final int MAX_RETRIES = 75; /** * The first reconnect goes out after this long, not after the full retry delay: the session * has already been gone for the keepalive's ~15 seconds, and every second more is a second * of failing tests. */ - static final long FIRST_RETRY_DELAY = 1000; + static final long FIRST_RETRY_DELAY = 250; /** Between re-requests of the reverse forward on a session that is up. */ static final long FORWARD_RETRY_DELAY = 2000; /** About two minutes of re-requests before the tunnel is rebuilt. */ diff --git a/src/main/java/ssh/SSHTunnel.java b/src/main/java/ssh/SSHTunnel.java index 8d9e86c..01722a7 100644 --- a/src/main/java/ssh/SSHTunnel.java +++ b/src/main/java/ssh/SSHTunnel.java @@ -16,10 +16,12 @@ import java.io.InputStream; import java.io.OutputStream; import java.nio.charset.StandardCharsets; +import java.util.List; import java.util.Locale; import java.util.Timer; import java.util.TimerTask; import java.util.UUID; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.atomic.AtomicBoolean; import java.util.logging.Level; import java.util.logging.Logger; @@ -44,6 +46,24 @@ public class SSHTunnel implements ReconnectableTunnel { private volatile boolean probeSupported; private int probeFailures; + /** The first connection carries the hub forward and owns readiness and the extra connections. */ + private final boolean primary; + /** Which of the multiplexed ports this connection tries first. */ + private final int slot; + /** The server's tunnel-mux took 2010 for our forwards; sticky for this server. */ + private volatile boolean muxed; + /** The server has no tunnel-mux (or it is not wanted): this connection binds 2010 itself. */ + private volatile boolean muxRuledOut; + /** A multiplexed port bound but not yet confirmed: waiting for the local proxy to listen. */ + private volatile boolean muxPending; + /** The remote port this connection's reverse forward holds, or -1. */ + private volatile int boundRemotePort = -1; + private final List extras = new CopyOnWriteArrayList<>(); + private volatile Thread extrasStarter; + private final SSHTunnel owner; + private volatile boolean proxyKeptForOthers; + private volatile int forwardRestarts; + /** Long enough for a slow network and a proxy hop, short enough to fail rather than hang. */ static final int DEFAULT_CONNECT_TIMEOUT_MS = 30_000; @@ -65,12 +85,18 @@ public class SSHTunnel implements ReconnectableTunnel { * so after a NAT change or a laptop switching networks the session looked connected until the * operating system gave up retransmitting -- about 15 minutes on Linux -- and all that time * the tunnel server kept port 2010 for a client that was gone. + * + *

2 s x 3: a session that hears nothing at all from the server for ~6 s is dead. That is + * the floor of how fast port 2010 can come back after a network change, so it is kept short; + * the server answers every keepalive at once, so a healthy connection - busy or idle - is + * never that quiet. Measured on a NAT change: 2010 back after ~8 s (5 s x 3: ~18 s); a + * 4-second outage passes without a reconnect, and nothing reconnected under load. */ - static final int SERVER_ALIVE_INTERVAL_MS = 5_000; + static final int SERVER_ALIVE_INTERVAL_MS = 2_000; static final int SERVER_ALIVE_COUNT_MAX = 3; /** How often the connection and forwarding monitors look. */ - static final long CONNECTION_CHECK_INTERVAL_MS = 2_000; + static final long CONNECTION_CHECK_INTERVAL_MS = 500; static final long FORWARDING_CHECK_INTERVAL_MS = 5_000; /** How long one end-to-end probe of the reverse forward may take. */ @@ -78,9 +104,27 @@ public class SSHTunnel implements ReconnectableTunnel { /** Consecutive failed probes before the reverse forward is re-registered. */ static final int REVERSE_PROBE_FAILURES_BEFORE_REPAIR = 2; + /** + * Ports for the reverse forwards of a multi-connection tunnel (--ssh-connections). The + * server's tunnel-mux notices them, checks each one with the same probe request as {@link + * #probeReverseForward}, and then serves 2010 itself, spreading connections over them. + * Sixteen, so a reconnect finds a free one while a dead connection's session still holds its + * port on the server. + */ + static final int MUX_FIRST_PORT = 2020; + static final int MUX_PORT_COUNT = 16; + /** + * How long a server gets to serve 2010 through a new forward before it is taken as one + * without tunnel-mux. tunnel-mux sees a forward within 0.25 s and serves 2010 once its probe + * came back through it, one round trip over the customer's link; this is also what a tunnel on + * an image without it waits once at startup (5 s made it 20 s to ready instead of 15). + */ + static final int MUX_CONFIRM_MS = 3_000; + private final int sshPort; private final String hubHost; private final int remoteProxyPort; + private int muxFirstPort = MUX_FIRST_PORT; public SSHTunnel(App app, String server) throws Exception { this(app, server, DEFAULT_SSH_PORT, DEFAULT_HUB_HOST, DEFAULT_REMOTE_PROXY_PORT); @@ -103,18 +147,38 @@ public SSHTunnel(App app, String server) throws Exception { */ SSHTunnel(App app, String server, int sshPort, String hubHost, int remoteProxyPort) throws Exception { - /* Create a connection instance */ + this(app, server, sshPort, hubHost, remoteProxyPort, MUX_FIRST_PORT); + } + + /** As above, with the multiplexed ports moved, for the same reason as remoteProxyPort. */ + SSHTunnel(App app, String server, int sshPort, String hubHost, int remoteProxyPort, + int muxFirstPort) throws Exception { + this(app, server, sshPort, hubHost, remoteProxyPort, muxFirstPort, null, 0); + this.connect(); + } + + /** An extra connection of {@code owner}'s tunnel; not connected yet, see {@link #startExtra}. */ + private SSHTunnel(App app, String server, int sshPort, String hubHost, int remoteProxyPort, + int muxFirstPort, SSHTunnel owner, int slot) { this.app = app; this.server = server; this.sshPort = sshPort; this.hubHost = hubHost; this.remoteProxyPort = remoteProxyPort; + this.muxFirstPort = muxFirstPort; this.connectionId = UUID.randomUUID().toString().substring(0, 8); + this.owner = owner; + this.primary = owner == null; + this.slot = slot; + // an extra connection exists only once the server's tunnel-mux has been seen + this.muxed = !primary; this.jsch = new JSch(); this.session = null; - this.connectionMonitor = new CustomConnectionMonitor(this, this.app); - this.connect(); + ReconnectHost host = ReconnectHost.of(app); + this.connectionMonitor = new CustomConnectionMonitor(this, + primary ? new PrimaryHost(host) : new ExtraHost(), Scheduler.timerBased(), + CustomConnectionMonitor.CURRENT_RETRY_DELAY); } /** @@ -266,6 +330,14 @@ public void stop(boolean quitting) { if (quitting) { this.shuttingDown.set(true); connectionMonitor.cancel(); + Thread starter = extrasStarter; + if (starter != null) { + starter.interrupt(); + } + for (SSHTunnel extra : extras) { + extra.stop(true); + } + extras.clear(); } this.stop(); } @@ -273,6 +345,8 @@ public void stop(boolean quitting) { public void stop() { synchronized (lifecycleLock) { portForwardingEstablished = false; + boundRemotePort = -1; + muxPending = false; Logger.getLogger(SSHTunnel.class.getName()).log(Level.INFO, String.format("[%s] Stopping secure tunnel", connectionId)); if (keepAliveTimer != null) { keepAliveTimer.cancel(); @@ -295,6 +369,55 @@ public boolean isForwardingEstablished() { return portForwardingEstablished && isAuthenticated() && !shuttingDown.get(); } + /** True when this connection or any of its extra connections carries traffic. */ + public boolean isAnyConnectionForwarding() { + if (isForwardingEstablished() && !muxPending) { + return true; + } + for (SSHTunnel extra : extras) { + if (extra.isForwardingEstablished()) { + return true; + } + } + return false; + } + + /** Whether a connection other than this one still carries traffic. */ + private boolean othersForwarding() { + SSHTunnel first = primary ? this : owner; + if (first != this && first.isForwardingEstablished()) { + return true; + } + for (SSHTunnel extra : first.extras) { + if (extra != this && extra.isForwardingEstablished()) { + return true; + } + } + return false; + } + + /** Connections currently forwarding, this one included; for the log. */ + int forwardingConnections() { + int n = isForwardingEstablished() ? 1 : 0; + for (SSHTunnel extra : extras) { + n += extra.isForwardingEstablished() ? 1 : 0; + } + return n; + } + + /** The port this connection's reverse forward holds on the server; -1 when none. */ + int getBoundRemotePort() { + return boundRemotePort; + } + + boolean isMuxed() { + return muxed; + } + + List getExtras() { + return extras; + } + // Delivery target for the reverse forward. JSch connects to this host for every // forwarded-tcpip channel; "0.0.0.0" is not connectable on macOS (TB-253), and // "localhost" could resolve to ::1 while Jetty listens on IPv4 only. @@ -306,19 +429,164 @@ public void createPortForwarding() { // exactly what happened on every restart, because the reverse forward was still // registered on the server and re-registering it throws. boolean reverseOk = establishReverseForward(); - boolean localOk = establishLocalForward(); + // The hub forward is a local port, so only one connection can hold it: the first. + boolean localOk = !primary || establishLocalForward(); portForwardingEstablished = reverseOk && localOk; if (portForwardingEstablished) { - Logger.getLogger(SSHTunnel.class.getName()).log(Level.INFO, - String.format("[%s] Port forwarding established: %s:%d -> %s:%d, localhost:%d -> %s:%d", - connectionId, server, remoteProxyPort, REVERSE_FORWARD_HOST, app.getJettyPort(), app.getSSHPort(), hubHost, app.getHubPort())); + Logger.getLogger(SSHTunnel.class.getName()).log(Level.INFO, primary + ? String.format("[%s] Port forwarding established: %s:%d -> %s:%d, localhost:%d -> %s:%d", + connectionId, server, boundRemotePort, REVERSE_FORWARD_HOST, app.getJettyPort(), app.getSSHPort(), hubHost, app.getHubPort()) + : String.format("[%s] Extra connection forwarding: %s:%d -> %s:%d", + connectionId, server, boundRemotePort, REVERSE_FORWARD_HOST, app.getJettyPort())); + } + if (muxPending && verifyReverseForwardDelivery()) { + // a reconnect: the local proxy already listens, so this can be decided now + settleForwarding(); } } + /** More than one connection asked for, and something on this side answers the probe. */ + private boolean wantsMux() { + return app.getSshConnections() > 1 && !app.isNoProxy() && !muxRuledOut; + } + private boolean establishReverseForward() { + if (wantsMux()) { + int port = bindMuxPort(); + if (port > 0) { + boundRemotePort = port; + // Confirmed later, once the local proxy listens: the server only serves 2010 + // after its probe through this forward came back from that proxy. + muxPending = !muxed; + return true; + } + if (muxed) { + // every multiplexed port refused: still held by dead sessions; asked again later + return false; + } + // the server permits none of them: an image without tunnel-mux + muxRuledOut = true; + } + return establishLegacyReverseForward(); + } + + /** Binds a free port among the multiplexed ones, this connection's own slot first. */ + 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()); + return port; + } catch (JSchException taken) { + Logger.getLogger(SSHTunnel.class.getName()).log(Level.FINE, + String.format("[%s] Port %d on the tunnel server refused: %s", connectionId, port, taken.getMessage())); + } + } + return -1; + } + + /** + * Decides a multiplexed forward bound before the local proxy listened: when port 2010 now + * reaches this client the server's tunnel-mux took it over, and the extra connections are + * opened; otherwise the server has none, and this connection binds 2010 itself as every + * client before did. Called by App once the proxy is up, and by a reconnect. + */ + public synchronized void settleForwarding() { + if (!muxPending) { + return; + } + long deadline = System.currentTimeMillis() + MUX_CONFIRM_MS; + boolean served = false; + while (!served && System.currentTimeMillis() < deadline && isAuthenticated()) { + served = probeReverseForward(Math.max(500, (int) (deadline - System.currentTimeMillis()))); + if (!served) { + try { + Thread.sleep(200); + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + break; + } + } + } + muxPending = false; + if (served) { + muxed = true; + Logger.getLogger(SSHTunnel.class.getName()).log(Level.INFO, String.format( + "[%s] The tunnel server spreads port %d over this tunnel's SSH connections; opening %d more", + connectionId, remoteProxyPort, app.getSshConnections() - 1 - extras.size())); + startExtras(); + return; + } + Logger.getLogger(SSHTunnel.class.getName()).log(Level.INFO, String.format( + "[%s] The tunnel server does not support several SSH connections; using one, on port %d", + connectionId, remoteProxyPort)); + muxRuledOut = true; + try { + session.delPortForwardingR(boundRemotePort); + } catch (Exception notRegistered) { + // gone with the session + } + boundRemotePort = -1; + portForwardingEstablished = establishLegacyReverseForward() && establishLocalForwardPresent(); + } + + private boolean establishLocalForwardPresent() { + if (!primary) { + return true; + } + try { + if (localForwardingActive(session.getPortForwardingL(), app.getSSHPort())) { + return true; + } + } catch (JSchException gone) { + return false; + } + return establishLocalForward(); + } + + /** Opens the extra connections in the background; the tunnel is ready without them. */ + private void startExtras() { + if (!primary || shuttingDown.get() || extrasStarter != null) { + return; + } + Thread starter = new Thread(() -> { + for (int i = 1; i < app.getSshConnections() && !shuttingDown.get(); i++) { + SSHTunnel extra = new SSHTunnel(app, server, sshPort, hubHost, remoteProxyPort, + muxFirstPort, this, i); + extras.add(extra); + extra.startExtra(); + } + Logger.getLogger(SSHTunnel.class.getName()).log(Level.INFO, String.format( + "[%s] %d of %d SSH connections forwarding", connectionId, + forwardingConnections(), app.getSshConnections())); + }, "ExtraConnections-" + connectionId); + starter.setDaemon(true); + extrasStarter = starter; + starter.start(); + } + + /** Connects an extra connection; a failure is retried by its own connection monitor. */ + private void startExtra() { + try { + connect(); + createPortForwarding(); + if (!isForwardingEstablished()) { + connectionMonitor.connectionLost(new Exception("extra connection could not forward")); + } + } catch (Exception failed) { + Logger.getLogger(SSHTunnel.class.getName()).log(Level.WARNING, String.format( + "[%s] Extra connection failed: %s; retrying", connectionId, failed.getMessage())); + if (!shuttingDown.get()) { + connectionMonitor.connectionLost(failed); + } + } + } + + private boolean establishLegacyReverseForward() { try { session.setPortForwardingR(remoteProxyPort, REVERSE_FORWARD_HOST, app.getJettyPort()); + boundRemotePort = remoteProxyPort; return true; } catch (JSchException ex) { Logger.getLogger(SSHTunnel.class.getName()).log(Level.SEVERE, @@ -342,11 +610,15 @@ private boolean establishLocalForward() { /** Drops both forwards, ignoring ones that are not registered. */ private void removePortForwarding() { try { - session.delPortForwardingR(remoteProxyPort); + session.delPortForwardingR(boundRemotePort > 0 ? boundRemotePort : remoteProxyPort); } catch (Exception notRegistered) { Logger.getLogger(SSHTunnel.class.getName()).log(Level.FINE, String.format("[%s] Reverse forward was not registered", connectionId)); } + boundRemotePort = -1; + if (!primary) { + return; + } try { session.delPortForwardingL(app.getSSHPort()); } catch (Exception notRegistered) { @@ -591,9 +863,11 @@ ReverseHealthTracker reverseHealth() { public void run() { try { if (session != null && session.isConnected() && !shuttingDown.get()) { - // Check if port forwarding is still active by testing the forwarded ports - boolean localForwardingActive = - localForwardingActive(session.getPortForwardingL(), app.getSSHPort()); + // Check if port forwarding is still active by testing the forwarded ports. + // Only the first connection holds the hub forward: an extra one reported it + // lost every 5 s and re-registered its reverse forward, often on a new port. + boolean localForwardingActive = !primary + || localForwardingActive(session.getPortForwardingL(), app.getSSHPort()); if (!localForwardingActive && portForwardingEstablished) { Logger.getLogger(SSHTunnel.class.getName()).log(Level.WARNING, @@ -645,6 +919,10 @@ private void checkRemoteForward() { } return; } + if (!primary || muxPending) { + // the server's tunnel-mux checks each forward itself; 2010 is everyone's + return; + } if (probeReverseForward(REVERSE_PROBE_TIMEOUT_MS)) { probeSupported = true; probeFailures = 0; @@ -668,8 +946,89 @@ private void checkRemoteForward() { } } + /** + * The first connection's view of the App while it reconnects. With other connections still + * forwarding, the local proxy keeps running and the tunnel stays ready: their traffic goes + * through that proxy, and stopping it made one dropped connection an outage for all of them. + */ + private final class PrimaryHost implements ReconnectHost { + private final ReconnectHost app; + + PrimaryHost(ReconnectHost app) { + this.app = app; + } + + @Override + public void stopLocalProxy() { + if (othersForwarding()) { + proxyKeptForOthers = true; + Logger.getLogger(SSHTunnel.class.getName()).log(Level.INFO, String.format( + "[%s] The other SSH connections keep carrying traffic while this one reconnects", connectionId)); + return; + } + app.stopLocalProxy(); + } + + @Override + public void startLocalProxy() { + if (proxyKeptForOthers) { + proxyKeptForOthers = false; + return; + } + app.startLocalProxy(); + } + + @Override + public void rebuildTunnel() throws Exception { + app.rebuildTunnel(); + } + + @Override + public void setReady(boolean ready) { + if (!ready && othersForwarding()) { + return; + } + app.setReady(ready); + } + } + + /** + * An extra connection's view: it never touches the local proxy, which belongs to the first, + * and never rebuilds the tunnel; it keeps reconnecting until the tunnel stops. It reports + * not-ready only when it was the last connection carrying traffic. + */ + private final class ExtraHost implements ReconnectHost { + @Override + public void stopLocalProxy() { + } + + @Override + public void startLocalProxy() { + } + + @Override + public void rebuildTunnel() { + if (!shuttingDown.get()) { + connectionMonitor.connectionLost(new Exception("still unreachable; continuing to retry")); + } + } + + @Override + public void setReady(boolean ready) { + if (!ready && !othersForwarding()) { + ReconnectHost.of(app).setReady(false); + } + } + } + + /** Times the forwards were dropped and requested again on this session; for tests. */ + int getForwardRestarts() { + return forwardRestarts; + } + @Override public synchronized void restartPortForwarding() { + forwardRestarts += 1; try { Logger.getLogger(SSHTunnel.class.getName()).log(Level.INFO, String.format("[%s] Restarting port forwarding", connectionId)); diff --git a/src/test/java/com/testingbot/tunnel/SshConnectionsOptionTest.java b/src/test/java/com/testingbot/tunnel/SshConnectionsOptionTest.java new file mode 100644 index 0000000..1e6d811 --- /dev/null +++ b/src/test/java/com/testingbot/tunnel/SshConnectionsOptionTest.java @@ -0,0 +1,44 @@ +package com.testingbot.tunnel; + +import org.apache.commons.cli.CommandLine; +import org.apache.commons.cli.ParseException; +import org.apache.commons.cli.PosixParser; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** {@code --ssh-connections}: how many SSH connections the tunnel opens. */ +class SshConnectionsOptionTest { + + private static CommandLine parse(String... args) throws ParseException { + return new PosixParser().parse(App.buildOptions(), args); + } + + private static App configured(String... args) throws Exception { + App app = new App(); + App.applyOptions(app, parse(args)); + return app; + } + + @Test + void defaultsToFour() throws Exception { + assertThat(configured().getSshConnections()).isEqualTo(App.DEFAULT_SSH_CONNECTIONS).isEqualTo(4); + } + + @Test + void acceptsOneToEight() throws Exception { + assertThat(configured("--ssh-connections", "1").getSshConnections()).isEqualTo(1); + assertThat(configured("--ssh-connections", "8").getSshConnections()).isEqualTo(8); + } + + @Test + void rejectsAnythingElse() { + for (String bad : new String[]{"0", "9", "-1", "two", "2.5"}) { + assertThatThrownBy(() -> configured("--ssh-connections", bad)) + .as(bad) + .isInstanceOf(ParseException.class) + .hasMessageContaining("--ssh-connections"); + } + } +} diff --git a/src/test/java/ssh/SSHTunnelEmbeddedServerTest.java b/src/test/java/ssh/SSHTunnelEmbeddedServerTest.java index 943e56b..f699b07 100644 --- a/src/test/java/ssh/SSHTunnelEmbeddedServerTest.java +++ b/src/test/java/ssh/SSHTunnelEmbeddedServerTest.java @@ -110,6 +110,8 @@ private App app() throws Exception { app.setClientSecret(PASSWORD); app.setJettyPort(findFreePort()); app.setSeleniumPort(findFreePort()); + // One connection: what these tests are about. Several are SSHTunnelMultiConnectionTest's. + app.setSshConnections(1); return app; } diff --git a/src/test/java/ssh/SSHTunnelMultiConnectionTest.java b/src/test/java/ssh/SSHTunnelMultiConnectionTest.java new file mode 100644 index 0000000..5775ca9 --- /dev/null +++ b/src/test/java/ssh/SSHTunnelMultiConnectionTest.java @@ -0,0 +1,479 @@ +package ssh; + +import com.testingbot.tunnel.App; +import com.testingbot.tunnel.Await; +import com.testingbot.tunnel.ForwardProbeHandler; +import com.testingbot.tunnel.TestPorts; +import org.apache.sshd.server.SshServer; +import org.apache.sshd.server.forward.AcceptAllForwardingFilter; +import org.apache.sshd.server.keyprovider.SimpleGeneratorHostKeyProvider; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.BufferedReader; +import java.io.Closeable; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.io.OutputStream; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.nio.charset.StandardCharsets; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Locale; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * --ssh-connections against a real SSH server in this JVM, with a stand-in for the tunnel + * server's tunnel-mux (ansible tunnel02/scripts/tunnel-mux.py): it serves the tunnel's port only + * once a forward on the multiplexed ports answers the probe, and spreads connections over the + * forwards that do. + * + *

The two directions of compatibility are what matter: a server without tunnel-mux (every + * production image so far) must get exactly the single connection on the tunnel port that every + * client before used, and one connection asked for must never touch the multiplexed ports. + */ +class SSHTunnelMultiConnectionTest { + + private static final String USER = "tunnel-user"; + private static final String PASSWORD = "tunnel-secret"; + + private SshServer sshd; + private ExecutorService pool; + private ServerSocket localProxy; + private final AtomicInteger proxied = new AtomicInteger(); + private SSHTunnel tunnel; + private StandInMux mux; + private int tunnelPort; + private int muxFirstPort; + + @BeforeEach + void setUp(@TempDir Path tmp) throws Exception { + pool = Executors.newCachedThreadPool(); + sshd = SshServer.setUpDefaultServer(); + sshd.setHost("127.0.0.1"); + sshd.setPort(0); + sshd.setKeyPairProvider(new SimpleGeneratorHostKeyProvider(tmp.resolve("hostkey.ser"))); + sshd.setPasswordAuthenticator((username, password, session) -> + USER.equals(username) && PASSWORD.equals(password)); + sshd.setForwardingFilter(AcceptAllForwardingFilter.INSTANCE); + sshd.start(); + tunnelPort = TestPorts.free(); + muxFirstPort = freeRange(SSHTunnel.MUX_PORT_COUNT); + } + + @AfterEach + void tearDown() throws Exception { + if (tunnel != null) { + tunnel.stop(true); + } + if (mux != null) { + mux.close(); + } + if (localProxy != null) { + localProxy.close(); + } + sshd.stop(true); + pool.shutdownNow(); + Thread.interrupted(); + pool.awaitTermination(5, TimeUnit.SECONDS); + } + + @Test + void opensEveryConnectionWhenTheServerSpreadsThePort() throws Exception { + mux = new StandInMux(tunnelPort, muxFirstPort); + App app = app(3); + tunnel = connect(app); + + tunnel.createPortForwarding(); + tunnel.settleForwarding(); + + assertThat(tunnel.isMuxed()).isTrue(); + assertThat(tunnel.isForwardingEstablished()).isTrue(); + Await.until("all three connections forwarding", () -> tunnel.forwardingConnections() == 3); + Set ports = new HashSet<>(); + ports.add(tunnel.getBoundRemotePort()); + for (SSHTunnel extra : tunnel.getExtras()) { + ports.add(extra.getBoundRemotePort()); + } + assertThat(ports).hasSize(3) + .allSatisfy(p -> assertThat(p).isBetween(muxFirstPort, muxFirstPort + SSHTunnel.MUX_PORT_COUNT - 1)); + // the hub forward is a local port: only the first connection holds it + for (SSHTunnel extra : tunnel.getExtras()) { + assertThat(SSHTunnel.localForwardingActive(extra.getSession().getPortForwardingL(), app.getSSHPort())).isFalse(); + } + + Await.until("the stand-in sees every forward", () -> mux.healthy() == 3); + for (int i = 0; i < 30; i++) { + assertThat(get(tunnelPort)).isEqualTo("proxied"); + } + assertThat(mux.usedBackends()).hasSize(3); + } + + @Test + void theForwardsStayPutAcrossForwardingChecks() throws Exception { + mux = new StandInMux(tunnelPort, muxFirstPort); + App app = app(3); + tunnel = connect(app); + tunnel.createPortForwarding(); + tunnel.settleForwarding(); + Await.until("all three connections forwarding", () -> tunnel.forwardingConnections() == 3); + List before = boundPorts(); + + // two runs of each connection's forwarding monitor + Thread.sleep(2 * SSHTunnel.FORWARDING_CHECK_INTERVAL_MS + 1_000); + + assertThat(boundPorts()).isEqualTo(before); + assertThat(tunnel.forwardingConnections()).isEqualTo(3); + // nothing was re-registered: on a real server a re-registered forward often lands on + // another port while the old one is still being released (seen in the lab) + assertThat(tunnel.getForwardRestarts()).isZero(); + assertThat(tunnel.getExtras()).allSatisfy(extra -> assertThat(extra.getForwardRestarts()).isZero()); + } + + private List boundPorts() { + List ports = new ArrayList<>(); + ports.add(tunnel.getBoundRemotePort()); + for (SSHTunnel extra : tunnel.getExtras()) { + ports.add(extra.getBoundRemotePort()); + } + return ports; + } + + @Test + void aServerWithoutTunnelMuxGetsOneConnectionOnTheTunnelPort() throws Exception { + App app = app(4); + tunnel = connect(app); + + tunnel.createPortForwarding(); + long start = System.currentTimeMillis(); + tunnel.settleForwarding(); + + assertThat(System.currentTimeMillis() - start).isLessThan(SSHTunnel.MUX_CONFIRM_MS + 2_000L); + assertThat(tunnel.isMuxed()).isFalse(); + assertThat(tunnel.getBoundRemotePort()).isEqualTo(tunnelPort); + assertThat(tunnel.isForwardingEstablished()).isTrue(); + assertThat(tunnel.getExtras()).isEmpty(); + assertThat(tunnel.getSession().getPortForwardingR()) + .noneMatch(f -> f.startsWith(muxFirstPort + ":") || f.contains(":" + muxFirstPort + ":")); + assertThat(get(tunnelPort)).isEqualTo("proxied"); + + // a reconnect on the same server does not try again + tunnel.restartPortForwarding(); + assertThat(tunnel.getBoundRemotePort()).isEqualTo(tunnelPort); + } + + @Test + void oneConnectionAskedForNeverTouchesTheMultiplexedPorts() throws Exception { + mux = new StandInMux(tunnelPort, muxFirstPort); + App app = app(1); + tunnel = connect(app); + + tunnel.createPortForwarding(); + tunnel.settleForwarding(); + + assertThat(tunnel.getBoundRemotePort()).isEqualTo(tunnelPort); + assertThat(tunnel.getExtras()).isEmpty(); + assertThat(mux.healthy()).isZero(); + assertThat(get(tunnelPort)).isEqualTo("proxied"); + } + + @Test + void theOtherConnectionsCarryTrafficWhileTheFirstReconnects() throws Exception { + mux = new StandInMux(tunnelPort, muxFirstPort); + App app = app(3); + tunnel = connect(app); + tunnel.createPortForwarding(); + tunnel.settleForwarding(); + Await.until("all three connections forwarding", () -> tunnel.forwardingConnections() == 3); + Await.until("the stand-in sees every forward", () -> mux.healthy() == 3); + int firstPort = tunnel.getBoundRemotePort(); + + tunnel.getSession().disconnect(); + Await.until("the stand-in drops the first connection's forward", () -> !mux.isHealthy(firstPort)); + + assertThat(tunnel.isAnyConnectionForwarding()).isTrue(); + for (int i = 0; i < 10; i++) { + assertThat(get(tunnelPort)).isEqualTo("proxied"); + } + Await.until("the first connection is back", 15_000, () -> tunnel.isForwardingEstablished()); + assertThat(tunnel.getBoundRemotePort()) + .isBetween(muxFirstPort, muxFirstPort + SSHTunnel.MUX_PORT_COUNT - 1); + } + + @Test + void anExtraConnectionThatDropsReconnectsOnItsOwn() throws Exception { + mux = new StandInMux(tunnelPort, muxFirstPort); + App app = app(2); + tunnel = connect(app); + tunnel.createPortForwarding(); + tunnel.settleForwarding(); + Await.until("both connections forwarding", () -> tunnel.forwardingConnections() == 2); + SSHTunnel extra = tunnel.getExtras().get(0); + + extra.getSession().disconnect(); + + Await.until("the extra connection is back", 15_000, + () -> extra.isForwardingEstablished() && extra.getSession().isConnected()); + assertThat(tunnel.isForwardingEstablished()).isTrue(); + assertThat(tunnel.getExtras()).hasSize(1); + } + + /* ---------------------------------------------------------------- helpers */ + + private App app(int connections) throws Exception { + App app = new App(); + app.setClientKey(USER); + app.setClientSecret(PASSWORD); + app.setJettyPort(TestPorts.free()); + app.setSeleniumPort(TestPorts.free()); + app.setSshConnections(connections); + localProxy = proxyAnswering(app.getJettyPort()); + return app; + } + + private SSHTunnel connect(App app) throws Exception { + ServerSocket hub = new ServerSocket(0, 50, InetAddress.getLoopbackAddress()); + pool.submit(() -> { + while (!hub.isClosed()) { + try (Socket ignored = hub.accept()) { + // never used here + } catch (IOException closed) { + return; + } + } + }); + app.setHubPort(hub.getLocalPort()); + return new SSHTunnel(app, "127.0.0.1", sshd.getPort(), "127.0.0.1", tunnelPort, muxFirstPort); + } + + /** The client's local proxy: answers the forward probe like ForwardProbeHandler, anything else "proxied". */ + private ServerSocket proxyAnswering(int port) throws IOException { + ServerSocket server = new ServerSocket(); + server.setReuseAddress(true); + server.bind(new InetSocketAddress(InetAddress.getLoopbackAddress(), port)); + pool.submit(() -> { + while (!server.isClosed()) { + Socket socket; + try { + socket = server.accept(); + } catch (IOException closed) { + return; + } + pool.submit(() -> answer(socket)); + } + }); + return server; + } + + private void answer(Socket socket) { + try (socket) { + BufferedReader in = new BufferedReader(new InputStreamReader(socket.getInputStream(), StandardCharsets.ISO_8859_1)); + String nonce = null; + String line = in.readLine(); + while ((line = in.readLine()) != null && !line.isEmpty()) { + String lower = line.toLowerCase(Locale.ROOT); + if (lower.startsWith(ForwardProbeHandler.PROBE_HEADER.toLowerCase(Locale.ROOT) + ":")) { + nonce = line.substring(line.indexOf(':') + 1).trim(); + } + } + OutputStream out = socket.getOutputStream(); + if (nonce != null) { + out.write(("HTTP/1.1 204 No Content\r\n" + ForwardProbeHandler.PROBE_HEADER + ": " + nonce + + "\r\nConnection: close\r\n\r\n").getBytes(StandardCharsets.ISO_8859_1)); + } else { + proxied.incrementAndGet(); + out.write("HTTP/1.1 200 OK\r\nContent-Length: 7\r\nConnection: close\r\n\r\nproxied" + .getBytes(StandardCharsets.ISO_8859_1)); + } + out.flush(); + } catch (IOException ignored) { + // the peer went away + } + } + + /** One plain request into {@code port}; the body, or the failure. */ + private static String get(int port) { + try (Socket s = new Socket()) { + s.connect(new InetSocketAddress("127.0.0.1", port), 3_000); + s.setSoTimeout(5_000); + s.getOutputStream().write("GET http://app.local/ HTTP/1.1\r\nHost: app.local\r\nConnection: close\r\n\r\n" + .getBytes(StandardCharsets.ISO_8859_1)); + String answer = new String(s.getInputStream().readAllBytes(), StandardCharsets.ISO_8859_1); + return answer.substring(answer.indexOf("\r\n\r\n") + 4); + } catch (IOException failed) { + return "failed: " + failed.getMessage(); + } + } + + private static int freeRange(int count) throws IOException { + java.util.Random random = new java.util.Random(); + for (int attempt = 0; attempt < 200; attempt++) { + int base = 33_000 + random.nextInt(20_000); + boolean free = true; + for (int p = base; p < base + count && free; p++) { + try (ServerSocket probe = new ServerSocket()) { + probe.bind(new InetSocketAddress(InetAddress.getLoopbackAddress(), p)); + } catch (IOException taken) { + free = false; + } + } + if (free) { + return base; + } + } + throw new IOException("no free range of " + count + " ports"); + } + + /** + * What tunnel-mux does, minimally: probe each multiplexed port, serve the tunnel port once one + * answers, and hand each connection to the next healthy forward. + */ + private final class StandInMux implements Closeable { + private final int listenPort; + private final int firstPort; + private final Set healthy = ConcurrentHashMap.newKeySet(); + private final Set used = ConcurrentHashMap.newKeySet(); + private volatile ServerSocket server; + private volatile boolean closed; + private final AtomicInteger next = new AtomicInteger(); + + StandInMux(int listenPort, int firstPort) { + this.listenPort = listenPort; + this.firstPort = firstPort; + pool.submit(this::probeForever); + } + + int healthy() { + return healthy.size(); + } + + boolean isHealthy(int port) { + return healthy.contains(port); + } + + Set usedBackends() { + return used; + } + + private void probeForever() { + while (!closed) { + for (int p = firstPort; p < firstPort + SSHTunnel.MUX_PORT_COUNT; p++) { + if (probe(p)) { + healthy.add(p); + } else { + healthy.remove(p); + } + } + if (!healthy.isEmpty() && server == null) { + bind(); + } + try { + Thread.sleep(100); + } catch (InterruptedException stop) { + return; + } + } + } + + private boolean probe(int port) { + String nonce = UUID.randomUUID().toString(); + try (Socket s = new Socket()) { + s.connect(new InetSocketAddress("127.0.0.1", port), 500); + s.setSoTimeout(1_000); + s.getOutputStream().write(("GET http://" + ForwardProbeHandler.PROBE_HOST + "/ HTTP/1.1\r\nHost: " + + ForwardProbeHandler.PROBE_HOST + "\r\n" + ForwardProbeHandler.PROBE_HEADER + ": " + nonce + + "\r\nConnection: close\r\n\r\n").getBytes(StandardCharsets.ISO_8859_1)); + String answer = new String(s.getInputStream().readAllBytes(), StandardCharsets.ISO_8859_1); + return answer.contains(nonce); + } catch (IOException failed) { + return false; + } + } + + private void bind() { + try { + ServerSocket socket = new ServerSocket(); + socket.setReuseAddress(true); + socket.bind(new InetSocketAddress(InetAddress.getLoopbackAddress(), listenPort)); + server = socket; + pool.submit(this::acceptForever); + } catch (IOException taken) { + // the client bound the tunnel port itself; nothing to take over + } + } + + private void acceptForever() { + while (!closed) { + Socket client; + try { + client = server.accept(); + } catch (IOException stop) { + return; + } + pool.submit(() -> relay(client)); + } + } + + private void relay(Socket client) { + List live = new ArrayList<>(healthy); + if (live.isEmpty()) { + closeQuietly(client); + return; + } + live.sort(null); + int port = live.get(Math.floorMod(next.getAndIncrement(), live.size())); + try { + Socket backend = new Socket("127.0.0.1", port); + used.add(port); + pool.submit(() -> pipe(client, backend)); + pool.submit(() -> pipe(backend, client)); + } catch (IOException failed) { + closeQuietly(client); + } + } + + private void pipe(Socket from, Socket to) { + try { + InputStream in = from.getInputStream(); + OutputStream out = to.getOutputStream(); + in.transferTo(out); + to.shutdownOutput(); + } catch (IOException ignored) { + closeQuietly(from); + closeQuietly(to); + } + } + + @Override + public void close() throws IOException { + closed = true; + if (server != null) { + server.close(); + } + } + } + + private static void closeQuietly(Socket socket) { + try { + socket.close(); + } catch (IOException ignored) { + // nothing to do + } + } +}