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