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         sshWriteAsyncHandler = new SshWriteAsyncHandler(this, channel.getAsyncIn());
 152
 153         ctx.fireChannelActive();
 154     }
 155
 156     private synchronized void handleSshSetupFailure(final ChannelHandlerContext ctx, final Throwable e) {
 157         logger.warn("Unable to setup SSH connection on channel: {}", ctx.channel(), e);
 158         connectPromise.setFailure(e);
 159         connectPromise = null;
 160         throw new IllegalStateException("Unable to setup SSH connection on channel: " + ctx.channel(), e);
 161     }
 162
 163     @Override
 164     public synchronized void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise promise) {
 165         sshWriteAsyncHandler.write(ctx, msg, promise);
 166     }
 167
 168     @Override
 169     public synchronized void connect(final ChannelHandlerContext ctx, final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) throws Exception {
 170         this.connectPromise = promise;
 171         startSsh(ctx, remoteAddress);
 172     }
 173
 174     @Override
 175     public void close(final ChannelHandlerContext ctx, final ChannelPromise promise) throws Exception {
 176         disconnect(ctx, promise);
 177     }
 178
 179     @Override
 180     public synchronized void disconnect(final ChannelHandlerContext ctx, final ChannelPromise promise) {
 181         if(sshReadAsyncListener != null) {
 182             sshReadAsyncListener.close();
 183         }
 184
 185         if(sshWriteAsyncHandler != null) {
 186             sshWriteAsyncHandler.close();
 187         }
 188
 189         if(session!= null && !session.isClosed() && !session.isClosing()) {
 190             session.close(false).addListener(new SshFutureListener<CloseFuture>() {
 191                 @Override
 192                 public void operationComplete(final CloseFuture future) {
 193                     if (future.isClosed() == false) {
 194                         session.close(true);
 195                     }
 196                     session = null;
 197                 }
 198             });
 199         }
 200
 201         channel = null;
 202         promise.setSuccess();
 203
 204         logger.debug("SSH session closed on channel: {}", ctx.channel());
 205         ctx.fireChannelInactive();
 206     }
 207
 208     /**
 209      * Listener over async input stream from SSH session.
 210      * This listeners schedules reads in a loop until the session is closed or read fails.
 211      */
 212     private static class SshReadAsyncListener implements SshFutureListener<IoReadFuture>, AutoCloseable {
 213         private static final int BUFFER_SIZE = 8192;
 214
 215         private final ChannelOutboundHandler asyncSshHandler;
 216         private final ChannelHandlerContext ctx;
 217
 218         private IoInputStream asyncOut;
 219         private Buffer buf;
 220         private IoReadFuture currentReadFuture;
 221
 222         public SshReadAsyncListener(final ChannelOutboundHandler asyncSshHandler, final ChannelHandlerContext ctx, final IoInputStream asyncOut) {
 223             this.asyncSshHandler = asyncSshHandler;
 224             this.ctx = ctx;
 225             this.asyncOut = asyncOut;
 226             buf = new Buffer(BUFFER_SIZE);
 227             asyncOut.read(buf).addListener(this);
 228         }
 229
 230         @Override
 231         public synchronized void operationComplete(final IoReadFuture future) {
 232             if(future.getException() != null) {
 233
 234                 if(asyncOut.isClosed() || asyncOut.isClosing()) {
 235
 236                     // Ssh dropped
 237                     logger.debug("Ssh session dropped on channel: {}", ctx.channel(), future.getException());
 238                     invokeDisconnect();
 239                     return;
 240                 } else {
 241                     logger.warn("Exception while reading from SSH remote on channel {}", ctx.channel(), future.getException());
 242                     invokeDisconnect();
 243                 }
 244             }
 245
 246             if (future.getRead() > 0) {
 247                 ctx.fireChannelRead(Unpooled.wrappedBuffer(buf.array(), 0, future.getRead()));
 248
 249                 // Schedule next read
 250                 buf = new Buffer(BUFFER_SIZE);
 251                 currentReadFuture = asyncOut.read(buf);
 252                 currentReadFuture.addListener(this);
 253             }
 254         }
 255
 256         private void invokeDisconnect() {
 257             try {
 258                 asyncSshHandler.disconnect(ctx, ctx.newPromise());
 259             } catch (final Exception e) {
 260                 // This should not happen
 261                 throw new IllegalStateException(e);
 262             }
 263         }
 264
 265         @Override
 266         public synchronized void close() {
 267             // Remove self as listener on close to prevent reading from closed input
 268             if(currentReadFuture != null) {
 269                 currentReadFuture.removeListener(this);
 270             }
 271
 272             asyncOut = null;
 273         }
 274     }
 275
 276     private static final class SshWriteAsyncHandler implements AutoCloseable {
 277         public static final int MAX_PENDING_WRITES = 100;
 278
 279         private final ChannelOutboundHandler channelHandler;
 280         private IoOutputStream asyncIn;
 281
 282         // Counter that holds the amount of pending write messages
 283         // Pending write can occur in case remote window is full
 284         // In such case, we need to wait for the pending write to finish
 285         private int pendingWriteCounter;
 286         // Last write future, that can be pending
 287         private IoWriteFuture lastWriteFuture;
 288
 289         public SshWriteAsyncHandler(final ChannelOutboundHandler channelHandler, final IoOutputStream asyncIn) {
 290             this.channelHandler = channelHandler;
 291             this.asyncIn = asyncIn;
 292         }
 293
 294         int c = 0;
 295
 296         public synchronized void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise promise) {
 297             try {
 298                 if(asyncIn == null || asyncIn.isClosed() || asyncIn.isClosing()) {
 299                     // If we are closed/closing, set immediate fail
 300                     promise.setFailure(new IllegalStateException("Channel closed"));
 301                 } else {
 302                     lastWriteFuture = asyncIn.write(toBuffer(msg));
 303                     lastWriteFuture.addListener(new SshFutureListener<IoWriteFuture>() {
 304
 305                         @Override
 306                         public void operationComplete(final IoWriteFuture future) {
 307                             ((ByteBuf) msg).release();
 308
 309                             // Notify success or failure
 310                             if (future.isWritten()) {
 311                                 promise.setSuccess();
 312                             } else {
 313                                 promise.setFailure(future.getException());
 314                             }
 315
 316                             // Reset last pending future
 317                             synchronized (SshWriteAsyncHandler.this) {
 318                                 lastWriteFuture = null;
 319                             }
 320                         }
 321                     });
 322                 }
 323             } catch (final WritePendingException e) {
 324                 // Check limit for pending writes
 325                 pendingWriteCounter++;
 326                 if(pendingWriteCounter > MAX_PENDING_WRITES) {
 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                 logger.debug("Write pending to SSH remote on channel: {}, current pending count: {}", ctx.channel(), pendingWriteCounter);
 332
 333                 // In case of pending, re-invoke write after pending is finished
 334                 lastWriteFuture.addListener(new SshFutureListener<IoWriteFuture>() {
 335                     @Override
 336                     public void operationComplete(final IoWriteFuture future) {
 337                         if (future.isWritten()) {
 338                             synchronized (SshWriteAsyncHandler.this) {
 339                                 // Pending done, decrease counter
 340                                 pendingWriteCounter--;
 341                             }
 342                             write(ctx, msg, promise);
 343                         } else {
 344                             // Cannot reschedule pending, fail
 345                             handlePendingFailed(ctx, e);
 346                         }
 347                     }
 348
 349                 });
 350             }
 351         }
 352
 353         private void handlePendingFailed(final ChannelHandlerContext ctx, final Exception e) {
 354             logger.warn("Exception while writing to SSH remote on channel {}", ctx.channel(), e);
 355             try {
 356                 channelHandler.disconnect(ctx, ctx.newPromise());
 357             } catch (final Exception ex) {
 358                 // This should not happen
 359                 throw new IllegalStateException(ex);
 360             }
 361         }
 362
 363         @Override
 364         public void close() {
 365             asyncIn = null;
 366         }
 367
 368         private Buffer toBuffer(final Object msg) {
 369             // TODO Buffer vs ByteBuf translate, Can we handle that better ?
 370             Preconditions.checkState(msg instanceof ByteBuf);
 371             final ByteBuf byteBuf = (ByteBuf) msg;
 372             final byte[] temp = new byte[byteBuf.readableBytes()];
 373             byteBuf.readBytes(temp, 0, byteBuf.readableBytes());
 374             return new Buffer(temp);
 375         }
 376
 377     }
 378 }