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