netconf/netconf-netty-util/src/main/java/org/opendaylight/netconf/nettyutil/handler/ssh/client/AsyncSshHandler.java

   1 /*
   2  * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.netconf.nettyutil.handler.ssh.client;
   9
  10 import static com.google.common.base.Verify.verify;
  11 import static java.util.Objects.requireNonNull;
  12
  13 import io.netty.channel.ChannelHandlerContext;
  14 import io.netty.channel.ChannelOutboundHandlerAdapter;
  15 import io.netty.channel.ChannelPromise;
  16 import io.netty.util.concurrent.Future;
  17 import io.netty.util.concurrent.FutureListener;
  18 import java.io.IOException;
  19 import java.net.SocketAddress;
  20 import java.util.concurrent.TimeUnit;
  21 import java.util.concurrent.atomic.AtomicBoolean;
  22 import org.eclipse.jdt.annotation.Nullable;
  23 import org.opendaylight.netconf.nettyutil.handler.ssh.authentication.AuthenticationHandler;
  24 import org.opendaylight.netconf.shaded.sshd.client.channel.ClientChannel;
  25 import org.opendaylight.netconf.shaded.sshd.client.future.AuthFuture;
  26 import org.opendaylight.netconf.shaded.sshd.client.future.ConnectFuture;
  27 import org.opendaylight.netconf.shaded.sshd.client.future.OpenFuture;
  28 import org.opendaylight.netconf.shaded.sshd.client.session.ClientSession;
  29 import org.opendaylight.netconf.shaded.sshd.core.CoreModuleProperties;
  30 import org.slf4j.Logger;
  31 import org.slf4j.LoggerFactory;
  32
  33 /**
  34  * Netty SSH handler class. Acts as interface between Netty and SSH library.
  35  */
  36 public class AsyncSshHandler extends ChannelOutboundHandlerAdapter {
  37     private static final Logger LOG = LoggerFactory.getLogger(AsyncSshHandler.class);
  38
  39     public static final String SUBSYSTEM = "netconf";
  40
  41     public static final int SSH_DEFAULT_NIO_WORKERS = 8;
  42
  43     public static final NetconfSshClient DEFAULT_CLIENT;
  44
  45     static {
  46         final NetconfSshClient c = new NetconfClientBuilder().build();
  47         // Disable default timeouts from mina sshd
  48         c.getProperties().put(CoreModuleProperties.AUTH_TIMEOUT.getName(), "0");
  49         c.getProperties().put(CoreModuleProperties.IDLE_TIMEOUT.getName(), "0");
  50         c.getProperties().put(CoreModuleProperties.NIO2_READ_TIMEOUT.getName(), "0");
  51         c.getProperties().put(CoreModuleProperties.TCP_NODELAY.getName(), true);
  52
  53         // TODO make configurable, or somehow reuse netty threadpool
  54         c.setNioWorkers(SSH_DEFAULT_NIO_WORKERS);
  55         c.start();
  56         DEFAULT_CLIENT = c;
  57     }
  58
  59     private final AtomicBoolean isDisconnected = new AtomicBoolean();
  60     private final AuthenticationHandler authenticationHandler;
  61     private final Future<?> negotiationFuture;
  62     private final NetconfSshClient sshClient;
  63
  64     private AsyncSshHandlerWriter sshWriteAsyncHandler;
  65
  66     private NettyAwareChannelSubsystem channel;
  67     private ClientSession session;
  68     private ChannelPromise connectPromise;
  69     private FutureListener<Object> negotiationFutureListener;
  70
  71     public AsyncSshHandler(final AuthenticationHandler authenticationHandler, final NetconfSshClient sshClient,
  72             final Future<?> negotiationFuture) {
  73         this.authenticationHandler = requireNonNull(authenticationHandler);
  74         this.sshClient = requireNonNull(sshClient);
  75         this.negotiationFuture = negotiationFuture;
  76     }
  77
  78     /**
  79      * Constructor of {@code AsyncSshHandler}.
  80      *
  81      * @param authenticationHandler authentication handler
  82      * @param sshClient             started SshClient
  83      */
  84     public AsyncSshHandler(final AuthenticationHandler authenticationHandler, final NetconfSshClient sshClient) {
  85         this(authenticationHandler, sshClient, null);
  86     }
  87
  88     public static AsyncSshHandler createForNetconfSubsystem(final AuthenticationHandler authenticationHandler) {
  89         return new AsyncSshHandler(authenticationHandler, DEFAULT_CLIENT);
  90     }
  91
  92     /**
  93      * Create AsyncSshHandler for netconf subsystem. Negotiation future has to be set to success after successful
  94      * netconf negotiation.
  95      *
  96      * @param authenticationHandler authentication handler
  97      * @param negotiationFuture     negotiation future
  98      * @return                      {@code AsyncSshHandler}
  99      */
 100     public static AsyncSshHandler createForNetconfSubsystem(final AuthenticationHandler authenticationHandler,
 101             final Future<?> negotiationFuture, final @Nullable NetconfSshClient sshClient) {
 102         return new AsyncSshHandler(authenticationHandler, sshClient != null ? sshClient : DEFAULT_CLIENT,
 103                 negotiationFuture);
 104     }
 105
 106     private synchronized void handleSshSetupFailure(final ChannelHandlerContext ctx, final Throwable error) {
 107         LOG.warn("Unable to setup SSH connection on channel: {}", ctx.channel(), error);
 108
 109         // If the promise is not yet done, we have failed with initial connect and set connectPromise to failure
 110         if (!connectPromise.isDone()) {
 111             connectPromise.setFailure(error);
 112         }
 113
 114         disconnect(ctx, ctx.newPromise());
 115     }
 116
 117     @Override
 118     public synchronized void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise promise) {
 119         sshWriteAsyncHandler.write(ctx, msg, promise);
 120     }
 121
 122     @Override
 123     public synchronized void connect(final ChannelHandlerContext ctx, final SocketAddress remoteAddress,
 124             final SocketAddress localAddress, final ChannelPromise promise) throws IOException {
 125         LOG.debug("SSH session connecting on channel {}. promise: {}", ctx.channel(), promise);
 126         connectPromise = requireNonNull(promise);
 127
 128         if (negotiationFuture != null) {
 129             negotiationFutureListener = future -> {
 130                 if (future.isSuccess()) {
 131                     promise.setSuccess();
 132                 }
 133             };
 134             //complete connection promise with netconf negotiation future
 135             negotiationFuture.addListener(negotiationFutureListener);
 136         }
 137
 138         LOG.debug("Starting SSH to {} on channel: {}", remoteAddress, ctx.channel());
 139         sshClient.connect(authenticationHandler.getUsername(), remoteAddress)
 140             // FIXME: this is a blocking call, we should handle this with a concurrently-scheduled timeout. We do not
 141             //        have a Timer ready, so perhaps we should be using the event loop?
 142             .verify(ctx.channel().config().getConnectTimeoutMillis(), TimeUnit.MILLISECONDS)
 143             .addListener(future -> onConnectComplete(future, ctx));
 144     }
 145
 146     private void onConnectComplete(final ConnectFuture future, final ChannelHandlerContext ctx) {
 147         final var cause = future.getException();
 148         if (cause != null) {
 149             handleSshSetupFailure(ctx, cause);
 150             return;
 151         }
 152
 153         final var clientSession = future.getSession();
 154         LOG.trace("SSH session {} created on channel: {}", clientSession, ctx.channel());
 155         verify(clientSession instanceof NettyAwareClientSession, "Unexpected session %s", clientSession);
 156         onConnectComplete((NettyAwareClientSession) clientSession, ctx);
 157     }
 158
 159     private synchronized void onConnectComplete(final NettyAwareClientSession clientSession,
 160             final ChannelHandlerContext ctx) {
 161         session = clientSession;
 162
 163         final AuthFuture authFuture;
 164         try {
 165             authFuture = authenticationHandler.authenticate(clientSession);
 166         } catch (final IOException e) {
 167             handleSshSetupFailure(ctx, e);
 168             return;
 169         }
 170
 171         authFuture.addListener(future -> onAuthComplete(future, clientSession, ctx));
 172     }
 173
 174     private void onAuthComplete(final AuthFuture future, final NettyAwareClientSession clientSession,
 175             final ChannelHandlerContext ctx) {
 176         final var cause = future.getException();
 177         if (cause != null) {
 178             handleSshSetupFailure(ctx, new AuthenticationFailedException("Authentication failed", cause));
 179             return;
 180         }
 181
 182         onAuthComplete(clientSession, ctx);
 183     }
 184
 185     private synchronized void onAuthComplete(final NettyAwareClientSession clientSession,
 186             final ChannelHandlerContext ctx) {
 187         LOG.debug("SSH session authenticated on channel: {}, server version: {}", ctx.channel(),
 188             clientSession.getServerVersion());
 189
 190         final OpenFuture openFuture;
 191         try {
 192             channel = clientSession.createSubsystemChannel(SUBSYSTEM, ctx);
 193             channel.setStreaming(ClientChannel.Streaming.Async);
 194             openFuture = channel.open();
 195         } catch (final IOException e) {
 196             handleSshSetupFailure(ctx, e);
 197             return;
 198         }
 199
 200         openFuture.addListener(future -> onOpenComplete(future, ctx));
 201     }
 202
 203     private void onOpenComplete(final OpenFuture future, final ChannelHandlerContext ctx) {
 204         final var cause = future.getException();
 205         if (cause != null) {
 206             handleSshSetupFailure(ctx, cause);
 207             return;
 208         }
 209
 210         onOpenComplete(ctx);
 211     }
 212
 213     private synchronized void onOpenComplete(final ChannelHandlerContext ctx) {
 214         LOG.trace("SSH subsystem channel opened successfully on channel: {}", ctx.channel());
 215
 216         if (negotiationFuture == null) {
 217             connectPromise.setSuccess();
 218         }
 219
 220         sshWriteAsyncHandler = new AsyncSshHandlerWriter(channel.getAsyncIn());
 221         ctx.fireChannelActive();
 222         channel.onClose(() -> disconnect(ctx, ctx.newPromise()));
 223     }
 224
 225     @Override
 226     public void close(final ChannelHandlerContext ctx, final ChannelPromise promise) {
 227         disconnect(ctx, promise);
 228     }
 229
 230     @Override
 231     public void disconnect(final ChannelHandlerContext ctx, final ChannelPromise promise) {
 232         if (isDisconnected.compareAndSet(false, true)) {
 233             safelyDisconnect(ctx, promise);
 234         }
 235     }
 236
 237     @SuppressWarnings("checkstyle:IllegalCatch")
 238     private synchronized void safelyDisconnect(final ChannelHandlerContext ctx, final ChannelPromise promise) {
 239         LOG.trace("Closing SSH session on channel: {} with connect promise in state: {}",
 240                 ctx.channel(), connectPromise);
 241
 242         // If we have already succeeded and the session was dropped after,
 243         // we need to fire inactive to notify reconnect logic
 244         if (connectPromise.isSuccess()) {
 245             ctx.fireChannelInactive();
 246         }
 247
 248         if (sshWriteAsyncHandler != null) {
 249             sshWriteAsyncHandler.close();
 250         }
 251
 252         //If connection promise is not already set, it means negotiation failed
 253         //we must set connection promise to failure
 254         if (!connectPromise.isDone()) {
 255             connectPromise.setFailure(new IllegalStateException("Negotiation failed"));
 256         }
 257
 258         //Remove listener from negotiation future, we don't want notifications
 259         //from negotiation anymore
 260         if (negotiationFuture != null) {
 261             negotiationFuture.removeListener(negotiationFutureListener);
 262         }
 263
 264         if (session != null && !session.isClosed() && !session.isClosing()) {
 265             session.close(false).addListener(future -> {
 266                 synchronized (this) {
 267                     if (!future.isClosed()) {
 268                         session.close(true);
 269                     }
 270                     session = null;
 271                 }
 272             });
 273         }
 274
 275         // Super disconnect is necessary in this case since we are using NioSocketChannel and it needs
 276         // to cleanup its resources e.g. Socket that it tries to open in its constructor
 277         // (https://bugs.opendaylight.org/show_bug.cgi?id=2430)
 278         // TODO better solution would be to implement custom ChannelFactory + Channel
 279         // that will use mina SSH lib internally: port this to custom channel implementation
 280         try {
 281             // Disconnect has to be closed after inactive channel event was fired, because it interferes with it
 282             super.disconnect(ctx, ctx.newPromise());
 283         } catch (final Exception e) {
 284             LOG.warn("Unable to cleanup all resources for channel: {}. Ignoring.", ctx.channel(), e);
 285         }
 286
 287         if (channel != null) {
 288             //TODO: see if calling just close() is sufficient
 289             //channel.close(false);
 290             channel.close();
 291             channel = null;
 292         }
 293         promise.setSuccess();
 294         LOG.debug("SSH session closed on channel: {}", ctx.channel());
 295     }
 296 }