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