Require a subsystem for client connections

[netconf.git] / transport / transport-ssh / src / main / java / org / opendaylight / netconf / transport / ssh / SSHClient.java

diff --git a/transport/transport-ssh/src/main/java/org/opendaylight/netconf/transport/ssh/SSHClient.java b/transport/transport-ssh/src/main/java/org/opendaylight/netconf/transport/ssh/SSHClient.java
index 60152c57d4fb0d5066751a9087ef090cacade898..c8e465aef754eeaa20bc38506cab6b1d3bae19ef 100644 (file)
--- a/transport/transport-ssh/src/main/java/org/opendaylight/netconf/transport/ssh/SSHClient.java
+++ b/transport/transport-ssh/src/main/java/org/opendaylight/netconf/transport/ssh/SSHClient.java
@@ -13,7 +13,8 @@ import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.EventLoopGroup;
 import java.io.IOException;
 import org.eclipse.jdt.annotation.NonNull;
-import org.opendaylight.netconf.shaded.sshd.client.session.ClientSession;
+import org.opendaylight.netconf.shaded.sshd.client.future.AuthFuture;
+import org.opendaylight.netconf.shaded.sshd.client.future.OpenFuture;
 import org.opendaylight.netconf.shaded.sshd.common.session.Session;
 import org.opendaylight.netconf.shaded.sshd.netty.NettyIoServiceFactoryFactory;
 import org.opendaylight.netconf.transport.api.TransportChannelListener;
@@ -24,19 +25,31 @@ import org.opendaylight.netconf.transport.tcp.TCPServer;
 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.ssh.client.rev230417.SshClientGrouping;
 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.tcp.client.rev230417.TcpClientGrouping;
 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.tcp.server.rev230417.TcpServerGrouping;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * A {@link TransportStack} acting as an SSH client.
  */
 public final class SSHClient extends SSHTransportStack {
-    private SSHClient(final TransportChannelListener listener, final TransportSshClient sshClient) {
+    private static final Logger LOG = LoggerFactory.getLogger(SSHClient.class);
+
+    private final String subsystem;
+
+    private SSHClient(final String subsystem, final TransportChannelListener listener,
+            final TransportSshClient sshClient) {
         super(listener, sshClient, sshClient.getSessionFactory());
+        // Mirrors check in ChannelSubsystem's constructor
+        if (subsystem.isBlank()) {
+            throw new IllegalArgumentException("Blank subsystem");
+        }
+        this.subsystem = subsystem;
     }
 
     static SSHClient of(final NettyIoServiceFactoryFactory ioServiceFactory, final EventLoopGroup group,
-            final TransportChannelListener listener, final SshClientGrouping clientParams)
+            final String subsystem, final TransportChannelListener listener, final SshClientGrouping clientParams)
                 throws UnsupportedConfigurationException {
-        return new SSHClient(listener, new TransportSshClient.Builder(ioServiceFactory, group)
+        return new SSHClient(subsystem, listener, new TransportSshClient.Builder(ioServiceFactory, group)
             .transportParams(clientParams.getTransportParams())
             .keepAlives(clientParams.getKeepalives())
             .clientIdentity(clientParams.getClientIdentity())
@@ -56,15 +69,51 @@ public final class SSHClient extends SSHTransportStack {
 
     @Override
     void onKeyEstablished(final Session session) throws IOException {
-        if (!(session instanceof ClientSession clientSession)) {
-            throw new IOException("Unexpected session " + session);
+        // server key is accepted, trigger authentication flow
+        final var sessionId = sessionId(session);
+        LOG.debug("Authenticating session {}", sessionId);
+        cast(session).auth().addListener(future -> onAuthComplete(future, sessionId));
+    }
+
+    private void onAuthComplete(final AuthFuture future, final Long sessionId) {
+        if (!future.isSuccess()) {
+            LOG.info("Session {} authentication failed", sessionId);
+            deleteSession(sessionId);
+        } else {
+            LOG.debug("Session {} authenticated", sessionId);
         }
+    }
 
-        // server key is accepted, trigger authentication flow
-        clientSession.auth().addListener(future -> {
-            if (!future.isSuccess()) {
-                deleteSession(session);
-            }
-        });
+    @Override
+    void onAuthenticated(final Session session) throws IOException {
+        final var sessionId = sessionId(session);
+        LOG.debug("Opening \"{}\" subsystem on session {}", subsystem, sessionId);
+
+        final var underlay = underlayOf(sessionId);
+        if (underlay == null) {
+            throw new IOException("Cannot find underlay for " + session);
+        }
+
+        final var clientSession = cast(session);
+        final var channel = clientSession.createSubsystemChannel(subsystem);
+        channel.onClose(() -> clientSession.close(true));
+        channel.open(underlay).addListener(future -> onSubsystemOpenComplete(future, sessionId));
+    }
+
+    private void onSubsystemOpenComplete(final OpenFuture future, final Long sessionId) {
+        if (future.isOpened()) {
+            LOG.debug("Established transport on session {}", sessionId);
+            completeUnderlay(sessionId, underlay -> addTransportChannel(new SSHTransportChannel(underlay)));
+        } else {
+            LOG.error("Failed to establish transport on session {}", sessionId, future.getException());
+            deleteSession(sessionId);
+        }
+    }
+
+    private static TransportClientSession cast(final Session session) throws IOException {
+        if (session instanceof TransportClientSession clientSession) {
+            return clientSession;
+        }
+        throw new IOException("Unexpected session " + session);
     }
-}
\ No newline at end of file
+}