2 * Copyright (c) 2016 Brocade Communication Systems and others. All rights reserved.
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
8 package org.opendaylight.netconf.callhome.protocol;
10 import static com.google.common.base.Preconditions.checkArgument;
11 import static java.util.Objects.requireNonNull;
13 import com.google.common.annotations.VisibleForTesting;
14 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
15 import io.netty.channel.ChannelHandlerContext;
16 import io.netty.channel.ChannelOutboundHandlerAdapter;
17 import io.netty.channel.ChannelPromise;
18 import io.netty.channel.EventLoopGroup;
19 import io.netty.util.concurrent.GlobalEventExecutor;
20 import io.netty.util.concurrent.Promise;
21 import java.io.IOException;
22 import java.net.InetSocketAddress;
23 import java.security.PublicKey;
24 import java.util.concurrent.ConcurrentHashMap;
25 import java.util.concurrent.ConcurrentMap;
26 import org.eclipse.jdt.annotation.Nullable;
27 import org.opendaylight.netconf.client.NetconfClientSession;
28 import org.opendaylight.netconf.client.NetconfClientSessionListener;
29 import org.opendaylight.netconf.client.NetconfClientSessionNegotiatorFactory;
30 import org.opendaylight.netconf.nettyutil.handler.ssh.client.AsyncSshHandlerWriter;
31 import org.opendaylight.netconf.nettyutil.handler.ssh.client.NetconfClientSessionImpl;
32 import org.opendaylight.netconf.shaded.sshd.client.channel.ChannelSubsystem;
33 import org.opendaylight.netconf.shaded.sshd.client.channel.ClientChannel;
34 import org.opendaylight.netconf.shaded.sshd.client.future.AuthFuture;
35 import org.opendaylight.netconf.shaded.sshd.client.future.OpenFuture;
36 import org.opendaylight.netconf.shaded.sshd.client.session.ClientSession;
37 import org.opendaylight.netconf.shaded.sshd.common.future.SshFutureListener;
38 import org.opendaylight.netconf.shaded.sshd.common.session.Session;
39 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.device.rev230430.connection.parameters.Protocol.Name;
40 import org.slf4j.Logger;
41 import org.slf4j.LoggerFactory;
43 // Non-final for testing
44 class CallHomeSessionContext implements CallHomeProtocolSessionContext {
46 private static final Logger LOG = LoggerFactory.getLogger(CallHomeSessionContext.class);
47 private static final String NETCONF = "netconf";
50 static final Session.AttributeKey<CallHomeSessionContext> SESSION_KEY = new Session.AttributeKey<>();
52 private final ClientSession sshSession;
53 private final CallHomeAuthorization authorization;
54 private final Factory factory;
56 private volatile boolean activated;
58 private final InetSocketAddress remoteAddress;
59 private final PublicKey serverKey;
61 @SuppressFBWarnings(value = "MC_OVERRIDABLE_METHOD_CALL_IN_CONSTRUCTOR", justification = "Passing 'this' around")
62 CallHomeSessionContext(final ClientSession sshSession, final CallHomeAuthorization authorization,
63 final Factory factory) {
64 this.authorization = requireNonNull(authorization, "authorization");
65 checkArgument(this.authorization.isServerAllowed(), "Server was not allowed.");
66 this.factory = requireNonNull(factory);
67 this.sshSession = requireNonNull(sshSession);
68 remoteAddress = (InetSocketAddress) this.sshSession.getIoSession().getRemoteAddress();
69 serverKey = this.sshSession.getServerKey();
72 final void associate() {
73 sshSession.setAttribute(SESSION_KEY, this);
76 static CallHomeSessionContext getFrom(final ClientSession sshSession) {
77 return sshSession.getAttribute(SESSION_KEY);
80 AuthFuture authorize() throws IOException {
81 authorization.applyTo(sshSession);
82 return sshSession.auth();
85 void openNetconfChannel() {
86 LOG.debug("Opening NETCONF Subsystem on {}", sshSession);
88 final MinaSshNettyChannel nettyChannel = newMinaSshNettyChannel();
89 final ClientChannel netconfChannel =
90 ((NetconfClientSessionImpl) sshSession).createSubsystemChannel(NETCONF, nettyChannel.pipeline());
91 netconfChannel.setStreaming(ClientChannel.Streaming.Async);
92 netconfChannel.open().addListener(newSshFutureListener(netconfChannel, nettyChannel));
93 } catch (IOException e) {
94 throw new IllegalStateException(e);
98 SshFutureListener<OpenFuture> newSshFutureListener(final ClientChannel netconfChannel,
99 final MinaSshNettyChannel nettyChannel) {
101 if (future.isOpened()) {
102 factory.getChannelOpenListener().onNetconfSubsystemOpened(this,
103 listener -> doActivate(netconfChannel, listener, nettyChannel));
105 channelOpenFailed(future.getException());
111 public void terminate() {
112 sshSession.close(false);
117 public Name getTransportType() {
121 private void channelOpenFailed(final Throwable throwable) {
122 LOG.error("Unable to open netconf subsystem, disconnecting.", throwable);
123 sshSession.close(false);
126 private synchronized Promise<NetconfClientSession> doActivate(final ClientChannel netconfChannel,
127 final NetconfClientSessionListener listener, final MinaSshNettyChannel nettyChannel) {
129 return newSessionPromise().setFailure(new IllegalStateException("Session already activated."));
132 nettyChannel.pipeline().addFirst(new SshWriteAsyncHandlerAdapter(netconfChannel));
133 LOG.info("Activating Netconf channel for {} with {}", getRemoteAddress(), listener);
134 final Promise<NetconfClientSession> activationPromise = newSessionPromise();
135 factory.getChannelInitializer(listener).initialize(nettyChannel, activationPromise);
136 ((ChannelSubsystem) netconfChannel).onClose(nettyChannel::doNettyDisconnect);
137 factory.getNettyGroup().register(nettyChannel).awaitUninterruptibly(500);
138 return activationPromise;
141 @Deprecated(since = "7.0.0", forRemoval = true)
142 protected MinaSshNettyChannel newMinaSshNettyChannel() {
143 return new MinaSshNettyChannel(this, sshSession);
146 private static Promise<NetconfClientSession> newSessionPromise() {
147 return GlobalEventExecutor.INSTANCE.newPromise();
151 public PublicKey getRemoteServerKey() {
156 public InetSocketAddress getRemoteAddress() {
157 return remoteAddress;
161 public String getSessionId() {
162 return authorization.getSessionName();
166 factory.remove(this);
169 static class Factory {
170 private final ConcurrentMap<String, CallHomeSessionContext> sessions = new ConcurrentHashMap<>();
171 private final EventLoopGroup nettyGroup;
172 private final NetconfClientSessionNegotiatorFactory negotiatorFactory;
173 private final CallHomeNetconfSubsystemListener subsystemListener;
175 Factory(final EventLoopGroup nettyGroup, final NetconfClientSessionNegotiatorFactory negotiatorFactory,
176 final CallHomeNetconfSubsystemListener subsystemListener) {
177 this.nettyGroup = requireNonNull(nettyGroup);
178 this.negotiatorFactory = requireNonNull(negotiatorFactory);
179 this.subsystemListener = requireNonNull(subsystemListener);
182 ReverseSshChannelInitializer getChannelInitializer(final NetconfClientSessionListener listener) {
183 return ReverseSshChannelInitializer.create(negotiatorFactory, listener);
186 CallHomeNetconfSubsystemListener getChannelOpenListener() {
187 return subsystemListener;
190 EventLoopGroup getNettyGroup() {
194 @Nullable CallHomeSessionContext createIfNotExists(final ClientSession sshSession,
195 final CallHomeAuthorization authorization) {
196 final var newSession = new CallHomeSessionContext(sshSession, authorization, this);
197 final var existing = sessions.putIfAbsent(newSession.getSessionId(), newSession);
198 if (existing == null) {
199 // There was no mapping, but now there is. Associate the context with the session.
200 newSession.associate();
204 // We already have a mapping, do not create a new one. But also check if the current session matches
205 // the one stored in the session. This can happen during re-keying.
206 return existing == CallHomeSessionContext.getFrom(sshSession) ? existing : null;
209 void remove(final CallHomeSessionContext session) {
210 sessions.remove(session.getSessionId(), session);
214 static class SshWriteAsyncHandlerAdapter extends ChannelOutboundHandlerAdapter {
215 private final AsyncSshHandlerWriter sshWriteAsyncHandler;
216 private final ClientChannel sshChannel;
218 SshWriteAsyncHandlerAdapter(final ClientChannel sshChannel) {
219 this.sshChannel = sshChannel;
220 sshWriteAsyncHandler = new AsyncSshHandlerWriter(sshChannel.getAsyncIn());
224 public void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise promise) {
225 sshWriteAsyncHandler.write(ctx, msg, promise);
228 public ClientChannel getSshChannel() {