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