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 handleSshChanelOpened(final ChannelHandlerContext ctx) {
 107         LOG.trace("SSH subsystem channel opened successfully on channel: {}", ctx.channel());
 108
 109         if (negotiationFuture == null) {
 110             connectPromise.setSuccess();
 111         }
 112
 113         sshWriteAsyncHandler = new AsyncSshHandlerWriter(channel.getAsyncIn());
 114         ctx.fireChannelActive();
 115         channel.onClose(() -> disconnect(ctx, ctx.newPromise()));
 116     }
 117
 118     private synchronized void handleSshSetupFailure(final ChannelHandlerContext ctx, final Throwable error) {
 119         LOG.warn("Unable to setup SSH connection on channel: {}", ctx.channel(), error);
 120
 121         // If the promise is not yet done, we have failed with initial connect and set connectPromise to failure
 122         if (!connectPromise.isDone()) {
 123             connectPromise.setFailure(error);
 124         }
 125
 126         disconnect(ctx, ctx.newPromise());
 127     }
 128
 129     @Override
 130     public synchronized void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise promise) {
 131         sshWriteAsyncHandler.write(ctx, msg, promise);
 132     }
 133
 134     @Override
 135     public synchronized void connect(final ChannelHandlerContext ctx, final SocketAddress remoteAddress,
 136             final SocketAddress localAddress, final ChannelPromise promise) throws IOException {
 137         LOG.debug("SSH session connecting on channel {}. promise: {}", ctx.channel(), promise);
 138         connectPromise = requireNonNull(promise);
 139
 140         if (negotiationFuture != null) {
 141             negotiationFutureListener = future -> {
 142                 if (future.isSuccess()) {
 143                     promise.setSuccess();
 144                 }
 145             };
 146             //complete connection promise with netconf negotiation future
 147             negotiationFuture.addListener(negotiationFutureListener);
 148         }
 149
 150         LOG.debug("Starting SSH to {} on channel: {}", remoteAddress, ctx.channel());
 151         sshClient.connect(authenticationHandler.getUsername(), remoteAddress)
 152             // FIXME: this is a blocking call, we should handle this with a concurrently-scheduled timeout. We do not
 153             //        have a Timer ready, so perhaps we should be using the event loop?
 154             .verify(ctx.channel().config().getConnectTimeoutMillis(), TimeUnit.MILLISECONDS)
 155             .addListener(future -> onConnectComplete(future, ctx));
 156     }
 157
 158     private void onConnectComplete(final ConnectFuture future, final ChannelHandlerContext ctx) {
 159         final var cause = future.getException();
 160         if (cause != null) {
 161             handleSshSetupFailure(ctx, cause);
 162             return;
 163         }
 164
 165         final var clientSession = future.getSession();
 166         LOG.trace("SSH session {} created on channel: {}", clientSession, ctx.channel());
 167         verify(clientSession instanceof NettyAwareClientSession, "Unexpected session %s", clientSession);
 168         onConnectComplete((NettyAwareClientSession) clientSession, ctx);
 169     }
 170
 171     private synchronized void onConnectComplete(final NettyAwareClientSession clientSession,
 172             final ChannelHandlerContext ctx) {
 173         session = clientSession;
 174
 175         final AuthFuture authFuture;
 176         try {
 177             authFuture = authenticationHandler.authenticate(clientSession);
 178         } catch (final IOException e) {
 179             handleSshSetupFailure(ctx, e);
 180             return;
 181         }
 182
 183         authFuture.addListener(future -> onAuthComplete(future, clientSession, ctx));
 184     }
 185
 186     private void onAuthComplete(final AuthFuture future, final NettyAwareClientSession clientSession,
 187             final ChannelHandlerContext ctx) {
 188         final var cause = future.getException();
 189         if (cause != null) {
 190             handleSshSetupFailure(ctx, new AuthenticationFailedException("Authentication failed", cause));
 191             return;
 192         }
 193
 194         onAuthComplete(clientSession, ctx);
 195     }
 196
 197     private synchronized void onAuthComplete(final NettyAwareClientSession clientSession,
 198             final ChannelHandlerContext ctx) {
 199         LOG.debug("SSH session authenticated on channel: {}, server version: {}", ctx.channel(),
 200             clientSession.getServerVersion());
 201
 202         final OpenFuture openFuture;
 203         try {
 204             channel = clientSession.createSubsystemChannel(SUBSYSTEM, ctx);
 205             channel.setStreaming(ClientChannel.Streaming.Async);
 206             openFuture = channel.open();
 207         } catch (final IOException e) {
 208             handleSshSetupFailure(ctx, e);
 209             return;
 210         }
 211
 212         openFuture.addListener(future -> onOpenComplete(future, ctx));
 213     }
 214
 215     private void onOpenComplete(final OpenFuture future, final ChannelHandlerContext ctx) {
 216         final var cause = future.getException();
 217         if (cause != null) {
 218             handleSshSetupFailure(ctx, cause);
 219             return;
 220         }
 221
 222         handleSshChanelOpened(ctx);
 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 }