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