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