add int tests (#986)

Author: wylswz

src/itest/java/com/hierynomus/sshj/KeepAliveTest.java

@@ -0,0 +1,59 @@
+package com.hierynomus.sshj;
+
+import net.schmizz.keepalive.BoundedKeepAliveProvider;
+import net.schmizz.sshj.Config;
+import net.schmizz.sshj.DefaultConfig;
+import net.schmizz.sshj.SSHClient;
+import net.schmizz.sshj.common.LoggerFactory;
+import net.schmizz.sshj.transport.verification.PromiscuousVerifier;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.junit.jupiter.Container;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class KeepAliveTest {
+    @Container
+    SshdContainer sshd = new SshdContainer(SshdContainer.Builder
+            .defaultBuilder()
+            .withAllKeys()
+            .withPackages("iptables")
+            .withPrivileged(true));
+
+    @Test
+    void testKeepAlive() throws Exception {
+        sshd.start();
+
+        Config config = new DefaultConfig();
+        BoundedKeepAliveProvider p = new BoundedKeepAliveProvider(LoggerFactory.DEFAULT, 4);
+        p.setKeepAliveInterval(1);
+        p.setMaxKeepAliveCount(1);
+        config.setKeepAliveProvider(p);
+        List<SSHClient> clients = new ArrayList<>();
+        for (int i=0; i<10; i++) {
+            SSHClient c = new SSHClient(config);
+            c.addHostKeyVerifier(new PromiscuousVerifier());
+            c.connect("127.0.0.1", sshd.getFirstMappedPort());
+            c.authPassword("sshj", "ultrapassword");
+            var sess = c.startSession();
+            sess.allocateDefaultPTY();
+            clients.add(c);
+        }
+
+        for (SSHClient client : clients) {
+            Assertions.assertTrue(client.isConnected());
+        }
+
+        var res = sshd.execInContainer("iptables", "-A", "INPUT", "-p", "tcp", "--dport", "22", "-j", "DROP");
+        Assertions.assertEquals(0, res.getExitCode());
+        // wait for keepalive to take action
+        Thread.sleep(2000);
+
+        for (SSHClient client : clients) {
+            Assertions.assertFalse(client.isConnected());
+        }
+
+        p.shutdown();
+    }
+}

src/itest/java/com/hierynomus/sshj/SshdContainer.java

@@ -106,13 +106,24 @@ public static class Builder implements Consumer<DockerfileBuilder> {
         private List<String> hostKeys = new ArrayList<>();
         private List<String> certificates = new ArrayList<>();
         private @NotNull SshdConfigBuilder sshdConfig = SshdConfigBuilder.defaultBuilder();
+        private boolean privileged = false;
+        private List<String> packages = new ArrayList<>();
 
         public static Builder defaultBuilder() {
             Builder b = new Builder();
-
             return b;
         }
 
+        public @NotNull Builder withPrivileged(boolean privileged) {
+            this.privileged = privileged;
+            return this;
+        }
+
+        public @NotNull Builder withPackages(@NotNull String... packages) {
+            this.packages.addAll(List.of(packages));
+            return this;
+        }
+
 
         public @NotNull Builder withSshdConfig(@NotNull SshdConfigBuilder sshdConfig) {
             this.sshdConfig = sshdConfig;
@@ -153,6 +164,9 @@ public void accept(@NotNull DockerfileBuilder builder) {
             builder.expose(22);
             builder.copy("entrypoint.sh", "/entrypoint.sh");
 
+            if (!packages.isEmpty()) {
+                builder.run("apk add --no-cache " + String.join(" ", packages));
+            }
             builder.add("authorized_keys", "/home/sshj/.ssh/authorized_keys");
             builder.copy("test-container/trusted_ca_keys", "/etc/ssh/trusted_ca_keys");
 
@@ -201,6 +215,9 @@ public SshdContainer() {
 
     public SshdContainer(SshdContainer.Builder builder) {
         this(builder.buildInner());
+        if (builder.privileged) {
+            withPrivilegedMode(true);
+        }
     }
 
     public SshdContainer(@NotNull Future<String> future) {

src/main/java/net/schmizz/keepalive/BoundedKeepAliveProvider.java

@@ -1,6 +1,7 @@
 package net.schmizz.keepalive;
 
 import net.schmizz.sshj.Config;
+import net.schmizz.sshj.common.LoggerFactory;
 import net.schmizz.sshj.connection.ConnectionException;
 import net.schmizz.sshj.connection.ConnectionImpl;
 import net.schmizz.sshj.transport.TransportException;
@@ -29,8 +30,8 @@ public class BoundedKeepAliveProvider extends KeepAliveProvider {
     protected final KeepAliveMonitor monitor;
 
 
-    public BoundedKeepAliveProvider(Config config, int numberOfThreads) {
-        this.monitor = new KeepAliveMonitor(config, numberOfThreads);
+    public BoundedKeepAliveProvider(LoggerFactory loggerFactory, int numberOfThreads) {
+        this.monitor = new KeepAliveMonitor(loggerFactory, numberOfThreads);
     }
 
     public void setKeepAliveInterval(int interval) {
@@ -76,40 +77,38 @@ public void startKeepAlive() {
     }
 
     protected static class KeepAliveMonitor {
+        private final Logger logger;
 
-        private final int numberOfThreads;
-        private final PriorityBlockingQueue<Wrapper> Q =
+        private final PriorityBlockingQueue<Wrapper> q =
                 new PriorityBlockingQueue<>(32, Comparator.comparingLong(w -> w.nextTimeMillis));
-        private long idleSleepMillis = 100;
         private static final List<Thread> workerThreads = new ArrayList<>();
+
+        private volatile long idleSleepMillis = 100;
+        private final int numberOfThreads;
+
         volatile boolean started = false;
-        private final Logger logger;
 
         private final ReentrantLock lock = new ReentrantLock();
         private final Condition shutDown = lock.newCondition();
         private final AtomicInteger shutDownCnt = new AtomicInteger(0);
 
-        public KeepAliveMonitor(Config config, int numberOfThreads) {
+        public KeepAliveMonitor(LoggerFactory loggerFactory, int numberOfThreads) {
             this.numberOfThreads = numberOfThreads;
-            logger = config.getLoggerFactory().getLogger(KeepAliveMonitor.class);
+            logger = loggerFactory.getLogger(KeepAliveMonitor.class);
         }
 
         // made public for test
         public void register(KeepAlive keepAlive) {
             if (!started) {
                 start();
             }
-            Q.add(new Wrapper(keepAlive));
+            q.add(new Wrapper(keepAlive));
         }
 
         public void setIdleSleepMillis(long idleSleepMillis) {
             this.idleSleepMillis = idleSleepMillis;
         }
 
-        void unregister(KeepAlive keepAlive) {
-            Q.removeIf(w -> keepAlive == w.keepAlive);
-        }
-
         private void sleep() {
             sleep(idleSleepMillis);
         }
@@ -140,7 +139,7 @@ private void doStart() {
             while (!Thread.currentThread().isInterrupted()) {
                 Wrapper wrapper;
 
-                if (Q.isEmpty() || (wrapper = Q.poll()) == null) {
+                if (q.isEmpty() || (wrapper = q.poll()) == null) {
                     sleep();
                     continue;
                 }
@@ -154,7 +153,7 @@ private void doStart() {
 
                 try {
                     wrapper.keepAlive.doKeepAlive();
-                    Q.add(wrapper.reschedule());
+                    q.add(wrapper.reschedule());
                 } catch (Exception e) {
                     // If we weren't interrupted, kill the transport, then this exception was unexpected.
                     // Else we're in shutdown-mode already, so don't forcibly kill the transport.

src/test/java/com/hierynomus/sshj/keepalive/BoundedKeepAliveProviderTest.java

@@ -5,6 +5,7 @@
 import net.schmizz.keepalive.KeepAlive;
 import net.schmizz.sshj.DefaultConfig;
 import net.schmizz.sshj.SSHClient;
+import net.schmizz.sshj.common.LoggerFactory;
 import net.schmizz.sshj.connection.ConnectionException;
 import net.schmizz.sshj.connection.ConnectionImpl;
 import net.schmizz.sshj.transport.TransportException;
@@ -19,7 +20,7 @@
 
 class EventuallyFailKeepAlive extends KeepAlive {
     // they can survive first 2 checks, and fail at 3rd
-    int failAfter = 2;
+    int failAfter = 1;
     volatile int current = 0;
 
     protected EventuallyFailKeepAlive(ConnectionImpl conn, String name) {
@@ -45,7 +46,7 @@ public class BoundedKeepAliveProviderTest {
     @BeforeAll
     static void setUpBeforeClass() throws Exception {
 
-        kp = new BoundedKeepAliveProvider(defaultConfig, 2) {
+        kp = new BoundedKeepAliveProvider(LoggerFactory.DEFAULT, 2) {
             @Override
             public KeepAlive provide(ConnectionImpl connection) {
                 return new EventuallyFailKeepAlive(connection, "test") {
@@ -67,19 +68,19 @@ void testWithConnections(int numOfConnections) throws IOException, InterruptedEx
             fixture.connectClient(client);
         }
         // first two checks are ok
-        Thread.sleep(2000);
+        Thread.sleep(1000);
         Assertions.assertTrue(clients.stream().allMatch(SSHClient::isConnected));
 
-        // wait for 3rd check to take place, we wait additional 100ms for it to finish
-        Thread.sleep(1100);
+        // wait for 2nd check to take place, we wait additional 200ms for it to finish
+        Thread.sleep(1200);
         Assertions.assertTrue(clients.stream().noneMatch(SSHClient::isConnected));
         Assertions.assertEquals(0, fixture.getServer().getActiveSessions().size());
     }
 
     @Test
     void testBoundedKeepAlive() throws IOException, InterruptedException {
-        // 2 threads can handle 64 connections
-        testWithConnections(64);
+        // 2 threads can handle 32 connections
+        testWithConnections(32);
     }
 
     private List<SSHClient> setupClients(int numOfConnections) {