import javax.net.ssl.SSLContext;
import org.opendaylight.controller.netconf.client.NetconfClientDispatcher;
+import org.opendaylight.controller.netconf.client.NetconfSshClientDispatcher;
+import org.opendaylight.controller.netconf.util.handler.ssh.authentication.AuthenticationHandler;
+import org.opendaylight.controller.netconf.util.handler.ssh.authentication.LoginPassword;
import org.opendaylight.controller.sal.connect.netconf.NetconfDevice;
import org.osgi.framework.BundleContext;
EventLoopGroup bossGroup = getBossThreadGroupDependency();
EventLoopGroup workerGroup = getWorkerThreadGroupDependency();
Optional<SSLContext> maybeContext = Optional.absent();
- NetconfClientDispatcher dispatcher = new NetconfClientDispatcher(maybeContext , bossGroup, workerGroup);
-
+ NetconfClientDispatcher dispatcher = null;
+ if(getTcpOnly()) {
+ dispatcher = new NetconfClientDispatcher(maybeContext , bossGroup, workerGroup);
+ } else {
+ AuthenticationHandler authHandler = new LoginPassword(getUsername(),getPassword());
+ dispatcher = new NetconfSshClientDispatcher(authHandler , bossGroup, workerGroup);
+ }
getDomRegistryDependency().registerProvider(device, bundleContext);
device.start(dispatcher);
leaf port {
type uint32;
}
+
+ leaf tcp-only {
+ type boolean;
+ }
+ leaf username {
+ type string;
+ }
+
+ leaf password {
+ type string;
+ }
container dom-registry {
uses config:service-ref {
refine type {
javax.xml.xpath,
org.opendaylight.controller.netconf.api,
org.opendaylight.controller.netconf.util,
- org.opendaylight.controller.netconf.util.xml,
+ org.opendaylight.controller.netconf.util.*,
org.opendaylight.protocol.framework,
org.slf4j,
org.w3c.dom,
- org.xml.sax
+ org.xml.sax,
+ io.netty.handler.codec
</Import-Package>
</instructions>
</configuration>
package org.opendaylight.controller.netconf.client;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+
+import javax.net.ssl.SSLContext;
+
+import org.opendaylight.controller.netconf.api.NetconfMessage;
+import org.opendaylight.controller.netconf.api.NetconfSession;
+import org.opendaylight.controller.netconf.api.NetconfTerminationReason;
+import org.opendaylight.controller.netconf.util.AbstractChannelInitializer;
+import org.opendaylight.controller.netconf.util.handler.FramingMechanismHandlerFactory;
+import org.opendaylight.controller.netconf.util.handler.NetconfMessageAggregator;
+import org.opendaylight.controller.netconf.util.handler.ssh.SshHandler;
+import org.opendaylight.controller.netconf.util.handler.ssh.authentication.AuthenticationHandler;
+import org.opendaylight.controller.netconf.util.handler.ssh.client.Invoker;
+import org.opendaylight.controller.netconf.util.messages.FramingMechanism;
+import org.opendaylight.controller.netconf.util.messages.NetconfMessageFactory;
+import org.opendaylight.protocol.framework.ProtocolHandlerFactory;
+import org.opendaylight.protocol.framework.ProtocolMessageDecoder;
+import org.opendaylight.protocol.framework.ProtocolMessageEncoder;
+import org.opendaylight.protocol.framework.ReconnectStrategy;
+import org.opendaylight.protocol.framework.SessionListener;
+import org.opendaylight.protocol.framework.SessionListenerFactory;
+
+import com.google.common.base.Optional;
+
+import io.netty.channel.ChannelHandler;
import io.netty.channel.EventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.util.HashedWheelTimer;
+import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.Promise;
public class NetconfSshClientDispatcher extends NetconfClientDispatcher {
- public NetconfSshClientDispatcher(EventLoopGroup bossGroup, EventLoopGroup workerGroup) {
- super(null, bossGroup, workerGroup);
+ private AuthenticationHandler authHandler;
+ private HashedWheelTimer timer;
+ private NetconfClientSessionNegotiatorFactory negotatorFactory;
+
+ public NetconfSshClientDispatcher(AuthenticationHandler authHandler, EventLoopGroup bossGroup,
+ EventLoopGroup workerGroup) {
+ super(Optional.<SSLContext> absent(), bossGroup, workerGroup);
+ this.authHandler = authHandler;
+ this.timer = new HashedWheelTimer();
+ this.negotatorFactory = new NetconfClientSessionNegotiatorFactory(timer);
+ }
+
+ @Override
+ public Future<NetconfClientSession> createClient(InetSocketAddress address,
+ final NetconfClientSessionListener sessionListener, ReconnectStrategy strat) {
+ return super.createClient(address, strat, new PipelineInitializer<NetconfClientSession>() {
+
+ @Override
+ public void initializeChannel(SocketChannel arg0, Promise<NetconfClientSession> arg1) {
+ new NetconfSshClientInitializer(authHandler, negotatorFactory, sessionListener).initialize(arg0, arg1);
+ }
+
+ });
+ }
+
+ private static final class NetconfSshClientInitializer extends AbstractChannelInitializer {
+
+ private final NetconfHandlerFactory handlerFactory;
+ private final AuthenticationHandler authenticationHandler;
+ private final NetconfClientSessionNegotiatorFactory negotiatorFactory;
+ private final NetconfClientSessionListener sessionListener;
+
+ public NetconfSshClientInitializer(AuthenticationHandler authHandler,
+ NetconfClientSessionNegotiatorFactory negotiatorFactory,
+ final NetconfClientSessionListener sessionListener) {
+ this.handlerFactory = new NetconfHandlerFactory(new NetconfMessageFactory());
+ this.authenticationHandler = authHandler;
+ this.negotiatorFactory = negotiatorFactory;
+ this.sessionListener = sessionListener;
+ }
+
+ @Override
+ public void initialize(SocketChannel ch, Promise<? extends NetconfSession> promise) {
+ try {
+ Invoker invoker = Invoker.subsystem("netconf");
+ ch.pipeline().addFirst(new SshHandler(authenticationHandler, invoker));
+ ch.pipeline().addLast("aggregator", new NetconfMessageAggregator(FramingMechanism.EOM));
+ ch.pipeline().addLast(handlerFactory.getDecoders());
+ initializeAfterDecoder(ch, promise);
+ ch.pipeline().addLast("frameEncoder",
+ FramingMechanismHandlerFactory.createHandler(FramingMechanism.EOM));
+ ch.pipeline().addLast(handlerFactory.getEncoders());
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ protected void initializeAfterDecoder(SocketChannel ch, Promise<? extends NetconfSession> promise) {
+ ch.pipeline().addLast("negotiator", negotiatorFactory.getSessionNegotiator(new SessionListenerFactory() {
+ @Override
+ public SessionListener<NetconfMessage, NetconfClientSession, NetconfTerminationReason> getSessionListener() {
+ return sessionListener;
+ }
+ }, ch, promise));
+
+ }
+ }
+
+ private static final class NetconfHandlerFactory extends ProtocolHandlerFactory<NetconfMessage> {
+
+ public NetconfHandlerFactory(final NetconfMessageFactory msgFactory) {
+ super(msgFactory);
+ }
+
+ @Override
+ public ChannelHandler[] getEncoders() {
+ return new ChannelHandler[] { new ProtocolMessageEncoder(this.msgFactory) };
+ }
+
+ @Override
+ public ChannelHandler[] getDecoders() {
+ return new ChannelHandler[] { new ProtocolMessageDecoder(this.msgFactory) };
+ }
}
}
org.opendaylight.controller.netconf.util.mapping,
org.opendaylight.controller.netconf.util.messages,
org.opendaylight.controller.netconf.util.handler,
+ org.opendaylight.controller.netconf.util.handler.*,
</Export-Package>
<Import-Package>
com.google.common.base,
package org.opendaylight.controller.netconf.util.handler.ssh;
+import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelOutboundHandlerAdapter;
import io.netty.channel.ChannelPromise;
+
import java.io.IOException;
import java.net.SocketAddress;
+
import org.opendaylight.controller.netconf.util.handler.ssh.authentication.AuthenticationHandler;
import org.opendaylight.controller.netconf.util.handler.ssh.client.Invoker;
import org.opendaylight.controller.netconf.util.handler.ssh.client.SshClient;
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
- this.sshClientAdapter.write((String) msg);
+ this.sshClientAdapter.write((ByteBuf) msg);
}
@Override
import java.io.IOException;
/**
- * Class Providing username/password authentication option to {@link org.opendaylight.controller.netconf.util.handler.ssh.SshHandler}
+ * Class Providing username/password authentication option to
+ * {@link org.opendaylight.controller.netconf.util.handler.ssh.SshHandler}
*/
public class LoginPassword extends AuthenticationHandler {
private final String username;
public void authenticate(Connection connection) throws IOException {
boolean isAuthenticated = connection.authenticateWithPassword(username, password);
- if (isAuthenticated == false) throw new IOException("Authentication failed.");
+ if (isAuthenticated == false)
+ throw new IOException("Authentication failed.");
}
}
import java.util.HashMap;
import java.util.Map;
-
/**
* Wrapper class around GANYMED SSH java library.
*/
private final AuthenticationHandler authenticationHandler;
private Connection connection;
- public SshClient(VirtualSocket socket,
- AuthenticationHandler authenticationHandler) throws IOException {
+ public SshClient(VirtualSocket socket, AuthenticationHandler authenticationHandler) throws IOException {
this.socket = socket;
this.authenticationHandler = authenticationHandler;
}
public SshSession openSession() throws IOException {
- if(connection == null) connect();
+ if (connection == null)
+ connect();
- Session session = connection.openSession();
+ Session session = connection.openSession();
SshSession sshSession = new SshSession(session);
openSessions.put(openSessions.size(), sshSession);
private void connect() throws IOException {
connection = new Connection(socket);
+
connection.connect();
authenticationHandler.authenticate(connection);
}
public void closeSession(SshSession session) {
- if( session.getState() == Channel.STATE_OPEN
- || session.getState() == Channel.STATE_OPENING) {
+ if (session.getState() == Channel.STATE_OPEN || session.getState() == Channel.STATE_OPENING) {
session.session.close();
}
}
public void close() {
- for(SshSession session : openSessions.values()) closeSession(session);
+ for (SshSession session : openSessions.values())
+ closeSession(session);
openSessions.clear();
- if(connection != null) connection.close();
+ if (connection != null)
+ connection.close();
}
}
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
+
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
+import java.util.LinkedList;
+import java.util.Queue;
import java.util.concurrent.atomic.AtomicBoolean;
import org.opendaylight.controller.netconf.util.handler.ssh.virtualsocket.VirtualSocketException;
+
/**
- * Worker thread class. Handles all downstream and upstream events in SSH Netty pipeline.
+ * Worker thread class. Handles all downstream and upstream events in SSH Netty
+ * pipeline.
*/
public class SshClientAdapter implements Runnable {
private final SshClient sshClient;
private InputStream stdErr;
private OutputStream stdIn;
+ private Queue<ByteBuf> postponned = new LinkedList<>();
+
+
private ChannelHandlerContext ctx;
private ChannelPromise disconnectPromise;
private final Object lock = new Object();
- public SshClientAdapter(SshClient sshClient,
- Invoker invoker) {
+ public SshClientAdapter(SshClient sshClient, Invoker invoker) {
this.sshClient = sshClient;
this.invoker = invoker;
}
try {
session = sshClient.openSession();
invoker.invoke(session);
-
stdOut = session.getStdout();
stdErr = session.getStderr();
- synchronized(lock) {
+ synchronized (lock) {
+
stdIn = session.getStdin();
+ ByteBuf message = null;
+ while ((message = postponned.poll()) != null) {
+ writeImpl(message);
+ }
}
while (stopRequested.get() == false) {
byte[] readBuff = new byte[1024];
int c = stdOut.read(readBuff);
-
+ if (c == -1) {
+ continue;
+ }
byte[] tranBuff = new byte[c];
System.arraycopy(readBuff, 0, tranBuff, 0, c);
sshClient.close();
synchronized (lock) {
- if(disconnectPromise != null) ctx.disconnect(disconnectPromise);
+ if (disconnectPromise != null)
+ ctx.disconnect(disconnectPromise);
}
}
}
// TODO: needs rework to match netconf framer API.
- public void write(String message) throws IOException {
+ public void write(ByteBuf message) throws IOException {
synchronized (lock) {
- if (stdIn == null) throw new IllegalStateException("StdIn not available");
+ if (stdIn == null) {
+ postponned.add(message);
+ return;
+ }
+ writeImpl(message);
}
- stdIn.write(message.getBytes());
+ }
+
+ private void writeImpl(ByteBuf message) throws IOException {
+ message.getBytes(0, stdIn, message.readableBytes());
stdIn.flush();
}
}
public void start(ChannelHandlerContext ctx) {
- if(this.ctx != null) return; // context is already associated.
-
+ if (this.ctx != null)
+ return; // context is already associated.
this.ctx = ctx;
new Thread(this).start();
}
Code Review / controller.git / commitdiff