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