opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/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
   9 package org.opendaylight.controller.netconf.nettyutil.handler.ssh.client;
  10
  11 import com.google.common.base.Preconditions;
  12 import io.netty.buffer.ByteBuf;
  13 import io.netty.buffer.Unpooled;
  14 import io.netty.channel.ChannelHandlerContext;
  15 import io.netty.channel.ChannelOutboundHandler;
  16 import io.netty.channel.ChannelOutboundHandlerAdapter;
  17 import io.netty.channel.ChannelPromise;
  18 import java.io.IOException;
  19 import java.net.SocketAddress;
  20 import org.apache.sshd.ClientChannel;
  21 import org.apache.sshd.ClientSession;
  22 import org.apache.sshd.SshClient;
  23 import org.apache.sshd.client.future.AuthFuture;
  24 import org.apache.sshd.client.future.ConnectFuture;
  25 import org.apache.sshd.client.future.OpenFuture;
  26 import org.apache.sshd.common.future.CloseFuture;
  27 import org.apache.sshd.common.future.SshFutureListener;
  28 import org.apache.sshd.common.io.IoInputStream;
  29 import org.apache.sshd.common.io.IoOutputStream;
  30 import org.apache.sshd.common.io.IoReadFuture;
  31 import org.apache.sshd.common.io.IoWriteFuture;
  32 import org.apache.sshd.common.io.WritePendingException;
  33 import org.apache.sshd.common.util.Buffer;
  34 import org.opendaylight.controller.netconf.nettyutil.handler.ssh.authentication.AuthenticationHandler;
  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 public class AsyncSshHandler extends ChannelOutboundHandlerAdapter {
  42
  43     private static final Logger logger = LoggerFactory.getLogger(AsyncSshHandler.class);
  44     public static final String SUBSYSTEM = "netconf";
  45
  46     public static final SshClient DEFAULT_CLIENT = SshClient.setUpDefaultClient();
  47
  48     public static final int SSH_DEFAULT_NIO_WORKERS = 8;
  49
  50     static {
  51         // TODO make configurable, or somehow reuse netty threadpool
  52         DEFAULT_CLIENT.setNioWorkers(SSH_DEFAULT_NIO_WORKERS);
  53         DEFAULT_CLIENT.start();
  54     }
  55
  56     private final AuthenticationHandler authenticationHandler;
  57     private final SshClient sshClient;
  58
  59     private SshReadAsyncListener sshReadAsyncListener;
  60     private SshWriteAsyncHandler sshWriteAsyncHandler;
  61
  62     private ClientChannel channel;
  63     private ClientSession session;
  64     private ChannelPromise connectPromise;
  65
  66
  67     public static AsyncSshHandler createForNetconfSubsystem(final AuthenticationHandler authenticationHandler) throws IOException {
  68         return new AsyncSshHandler(authenticationHandler, DEFAULT_CLIENT);
  69     }
  70
  71     /**
  72      *
  73      * @param authenticationHandler
  74      * @param sshClient started SshClient
  75      * @throws IOException
  76      */
  77     public AsyncSshHandler(final AuthenticationHandler authenticationHandler, final SshClient sshClient) throws IOException {
  78         this.authenticationHandler = Preconditions.checkNotNull(authenticationHandler);
  79         this.sshClient = Preconditions.checkNotNull(sshClient);
  80         // Start just in case
  81         sshClient.start();
  82     }
  83
  84     private void startSsh(final ChannelHandlerContext ctx, final SocketAddress address) {
  85         logger.debug("Starting SSH to {} on channel: {}", address, ctx.channel());
  86
  87         final ConnectFuture sshConnectionFuture = sshClient.connect(authenticationHandler.getUsername(), address);
  88         sshConnectionFuture.addListener(new SshFutureListener<ConnectFuture>() {
  89             @Override
  90             public void operationComplete(final ConnectFuture future) {
  91                 if (future.isConnected()) {
  92                     handleSshSessionCreated(future, ctx);
  93                 } else {
  94                     handleSshSetupFailure(ctx, future.getException());
  95                 }
  96             }
  97         });
  98     }
  99
 100     private synchronized void handleSshSessionCreated(final ConnectFuture future, final ChannelHandlerContext ctx) {
 101         try {
 102             logger.trace("SSH session created on channel: {}", ctx.channel());
 103
 104             session = future.getSession();
 105             final AuthFuture authenticateFuture = authenticationHandler.authenticate(session);
 106             authenticateFuture.addListener(new SshFutureListener<AuthFuture>() {
 107                 @Override
 108                 public void operationComplete(final AuthFuture future) {
 109                     if (future.isSuccess()) {
 110                         handleSshAuthenticated(session, ctx);
 111                     } else {
 112                         handleSshSetupFailure(ctx, future.getException());
 113                     }
 114                 }
 115             });
 116         } catch (final IOException e) {
 117             handleSshSetupFailure(ctx, e);
 118         }
 119     }
 120
 121     private synchronized void handleSshAuthenticated(final ClientSession session, final ChannelHandlerContext ctx) {
 122         try {
 123             logger.debug("SSH session authenticated on channel: {}, server version: {}", ctx.channel(), session.getServerVersion());
 124
 125             channel = session.createSubsystemChannel(SUBSYSTEM);
 126             channel.setStreaming(ClientChannel.Streaming.Async);
 127             channel.open().addListener(new SshFutureListener<OpenFuture>() {
 128                 @Override
 129                 public void operationComplete(final OpenFuture future) {
 130                     if(future.isOpened()) {
 131                         handleSshChanelOpened(ctx);
 132                     } else {
 133                         handleSshSetupFailure(ctx, future.getException());
 134                     }
 135                 }
 136             });
 137
 138
 139         } catch (final IOException e) {
 140             handleSshSetupFailure(ctx, e);
 141         }
 142     }
 143
 144     private synchronized void handleSshChanelOpened(final ChannelHandlerContext ctx) {
 145         logger.trace("SSH subsystem channel opened successfully on channel: {}", ctx.channel());
 146
 147         connectPromise.setSuccess();
 148         connectPromise = null;
 149
 150         sshReadAsyncListener = new SshReadAsyncListener(this, ctx, channel.getAsyncOut());
 151         // if readAsyncListener receives immediate close, it will close this handler and closing this handler sets channel variable to null
 152         if(channel != null) {
 153             sshWriteAsyncHandler = new SshWriteAsyncHandler(this, channel.getAsyncIn());
 154             ctx.fireChannelActive();
 155         }
 156     }
 157
 158     private synchronized void handleSshSetupFailure(final ChannelHandlerContext ctx, final Throwable e) {
 159         logger.warn("Unable to setup SSH connection on channel: {}", ctx.channel(), e);
 160         connectPromise.setFailure(e);
 161         connectPromise = null;
 162         throw new IllegalStateException("Unable to setup SSH connection on channel: " + ctx.channel(), e);
 163     }
 164
 165     @Override
 166     public synchronized void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise promise) {
 167         sshWriteAsyncHandler.write(ctx, msg, promise);
 168     }
 169
 170     @Override
 171     public synchronized void connect(final ChannelHandlerContext ctx, final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) throws Exception {
 172         this.connectPromise = promise;
 173         startSsh(ctx, remoteAddress);
 174     }
 175
 176     @Override
 177     public void close(final ChannelHandlerContext ctx, final ChannelPromise promise) throws Exception {
 178         disconnect(ctx, promise);
 179     }
 180
 181     @Override
 182     public synchronized void disconnect(final ChannelHandlerContext ctx, final ChannelPromise promise) {
 183         if(sshReadAsyncListener != null) {
 184             sshReadAsyncListener.close();
 185         }
 186
 187         if(sshWriteAsyncHandler != null) {
 188             sshWriteAsyncHandler.close();
 189         }
 190
 191         if(session!= null && !session.isClosed() && !session.isClosing()) {
 192             session.close(false).addListener(new SshFutureListener<CloseFuture>() {
 193                 @Override
 194                 public void operationComplete(final CloseFuture future) {
 195                     if (future.isClosed() == false) {
 196                         session.close(true);
 197                     }
 198                     session = null;
 199                 }
 200             });
 201         }
 202
 203         channel = null;
 204         promise.setSuccess();
 205
 206         logger.debug("SSH session closed on channel: {}", ctx.channel());
 207         ctx.fireChannelInactive();
 208     }
 209
 210     /**
 211      * Listener over async input stream from SSH session.
 212      * This listeners schedules reads in a loop until the session is closed or read fails.
 213      */
 214     private static class SshReadAsyncListener implements SshFutureListener<IoReadFuture>, AutoCloseable {
 215         private static final int BUFFER_SIZE = 8192;
 216
 217         private final ChannelOutboundHandler asyncSshHandler;
 218         private final ChannelHandlerContext ctx;
 219
 220         private IoInputStream asyncOut;
 221         private Buffer buf;
 222         private IoReadFuture currentReadFuture;
 223
 224         public SshReadAsyncListener(final ChannelOutboundHandler asyncSshHandler, final ChannelHandlerContext ctx, final IoInputStream asyncOut) {
 225             this.asyncSshHandler = asyncSshHandler;
 226             this.ctx = ctx;
 227             this.asyncOut = asyncOut;
 228             buf = new Buffer(BUFFER_SIZE);
 229             asyncOut.read(buf).addListener(this);
 230         }
 231
 232         @Override
 233         public synchronized void operationComplete(final IoReadFuture future) {
 234             if(future.getException() != null) {
 235                 if(asyncOut.isClosed() || asyncOut.isClosing()) {
 236                     // Ssh dropped
 237                     logger.debug("Ssh session dropped on channel: {}", ctx.channel(), future.getException());
 238                 } else {
 239                     logger.warn("Exception while reading from SSH remote on channel {}", ctx.channel(), future.getException());
 240                 }
 241                 invokeDisconnect();
 242                 return;
 243             }
 244
 245             if (future.getRead() > 0) {
 246                 ctx.fireChannelRead(Unpooled.wrappedBuffer(buf.array(), 0, future.getRead()));
 247
 248                 // Schedule next read
 249                 buf = new Buffer(BUFFER_SIZE);
 250                 currentReadFuture = asyncOut.read(buf);
 251                 currentReadFuture.addListener(this);
 252             }
 253         }
 254
 255         private void invokeDisconnect() {
 256             try {
 257                 asyncSshHandler.disconnect(ctx, ctx.newPromise());
 258             } catch (final Exception e) {
 259                 // This should not happen
 260                 throw new IllegalStateException(e);
 261             }
 262         }
 263
 264         @Override
 265         public synchronized void close() {
 266             // Remove self as listener on close to prevent reading from closed input
 267             if(currentReadFuture != null) {
 268                 currentReadFuture.removeListener(this);
 269             }
 270
 271             asyncOut = null;
 272         }
 273     }
 274
 275     private static final class SshWriteAsyncHandler implements AutoCloseable {
 276         public static final int MAX_PENDING_WRITES = 100;
 277
 278         private final ChannelOutboundHandler channelHandler;
 279         private IoOutputStream asyncIn;
 280
 281         // Counter that holds the amount of pending write messages
 282         // Pending write can occur in case remote window is full
 283         // In such case, we need to wait for the pending write to finish
 284         private int pendingWriteCounter;
 285         // Last write future, that can be pending
 286         private IoWriteFuture lastWriteFuture;
 287
 288         public SshWriteAsyncHandler(final ChannelOutboundHandler channelHandler, final IoOutputStream asyncIn) {
 289             this.channelHandler = channelHandler;
 290             this.asyncIn = asyncIn;
 291         }
 292
 293         int c = 0;
 294
 295         public synchronized void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise promise) {
 296             try {
 297                 if(asyncIn == null || asyncIn.isClosed() || asyncIn.isClosing()) {
 298                     // If we are closed/closing, set immediate fail
 299                     promise.setFailure(new IllegalStateException("Channel closed"));
 300                 } else {
 301                     lastWriteFuture = asyncIn.write(toBuffer(msg));
 302                     lastWriteFuture.addListener(new SshFutureListener<IoWriteFuture>() {
 303
 304                         @Override
 305                         public void operationComplete(final IoWriteFuture future) {
 306                             ((ByteBuf) msg).release();
 307
 308                             // Notify success or failure
 309                             if (future.isWritten()) {
 310                                 promise.setSuccess();
 311                             } else {
 312                                 promise.setFailure(future.getException());
 313                             }
 314
 315                             // Reset last pending future
 316                             synchronized (SshWriteAsyncHandler.this) {
 317                                 lastWriteFuture = null;
 318                             }
 319                         }
 320                     });
 321                 }
 322             } catch (final WritePendingException e) {
 323                 // Check limit for pending writes
 324                 pendingWriteCounter++;
 325                 if(pendingWriteCounter > MAX_PENDING_WRITES) {
 326                     promise.setFailure(e);
 327                     handlePendingFailed(ctx, new IllegalStateException("Too much pending writes(" + MAX_PENDING_WRITES + ") on channel: " + ctx.channel() +
 328                             ", remote window is not getting read or is too small"));
 329                 }
 330
 331                 // We need to reset buffer read index, since we've already read it when we tried to write it the first time
 332                 ((ByteBuf) msg).resetReaderIndex();
 333                 logger.debug("Write pending to SSH remote on channel: {}, current pending count: {}", ctx.channel(), pendingWriteCounter);
 334
 335                 // In case of pending, re-invoke write after pending is finished
 336                 Preconditions.checkNotNull(lastWriteFuture, "Write is pending, but there was no previous write attempt", e);
 337                 lastWriteFuture.addListener(new SshFutureListener<IoWriteFuture>() {
 338                     @Override
 339                     public void operationComplete(final IoWriteFuture future) {
 340                         // FIXME possible minor race condition, we cannot guarantee that this callback when pending is finished will be executed first
 341                         // External thread could trigger write on this instance while we are on this line
 342                         // Verify
 343                         if (future.isWritten()) {
 344                             synchronized (SshWriteAsyncHandler.this) {
 345                                 // Pending done, decrease counter
 346                                 pendingWriteCounter--;
 347                                 write(ctx, msg, promise);
 348                             }
 349                         } else {
 350                             // Cannot reschedule pending, fail
 351                             handlePendingFailed(ctx, e);
 352                         }
 353                     }
 354
 355                 });
 356             }
 357         }
 358
 359         private void handlePendingFailed(final ChannelHandlerContext ctx, final Exception e) {
 360             logger.warn("Exception while writing to SSH remote on channel {}", ctx.channel(), e);
 361             try {
 362                 channelHandler.disconnect(ctx, ctx.newPromise());
 363             } catch (final Exception ex) {
 364                 // This should not happen
 365                 throw new IllegalStateException(ex);
 366             }
 367         }
 368
 369         @Override
 370         public void close() {
 371             asyncIn = null;
 372         }
 373
 374         private Buffer toBuffer(final Object msg) {
 375             // TODO Buffer vs ByteBuf translate, Can we handle that better ?
 376             Preconditions.checkState(msg instanceof ByteBuf);
 377             final ByteBuf byteBuf = (ByteBuf) msg;
 378             final byte[] temp = new byte[byteBuf.readableBytes()];
 379             byteBuf.readBytes(temp, 0, byteBuf.readableBytes());
 380             return new Buffer(temp);
 381         }
 382
 383     }
 384 }