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