protocol/netconf-server/src/main/java/org/opendaylight/netconf/server/NetconfSubsystem.java

   1 /*
   2  * Copyright (c) 2023 PANTHEON.tech, s.r.o. 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 package org.opendaylight.netconf.server;
   9
  10 import static java.util.Objects.requireNonNull;
  11
  12 import io.netty.buffer.ByteBuf;
  13 import io.netty.buffer.Unpooled;
  14 import io.netty.channel.ChannelHandlerContext;
  15 import io.netty.channel.ChannelInboundHandlerAdapter;
  16 import io.netty.channel.ChannelOutboundHandlerAdapter;
  17 import io.netty.channel.ChannelPromise;
  18 import io.netty.channel.embedded.EmbeddedChannel;
  19 import io.netty.util.concurrent.DefaultPromise;
  20 import io.netty.util.concurrent.GlobalEventExecutor;
  21 import java.io.IOException;
  22 import java.net.InetSocketAddress;
  23 import java.nio.charset.StandardCharsets;
  24 import org.opendaylight.netconf.api.messages.NetconfHelloMessageAdditionalHeader;
  25 import org.opendaylight.netconf.shaded.sshd.common.io.IoInputStream;
  26 import org.opendaylight.netconf.shaded.sshd.common.io.IoOutputStream;
  27 import org.opendaylight.netconf.shaded.sshd.common.util.buffer.ByteArrayBuffer;
  28 import org.opendaylight.netconf.shaded.sshd.server.channel.ChannelDataReceiver;
  29 import org.opendaylight.netconf.shaded.sshd.server.channel.ChannelSession;
  30 import org.opendaylight.netconf.shaded.sshd.server.channel.ChannelSessionAware;
  31 import org.opendaylight.netconf.shaded.sshd.server.command.AbstractCommandSupport;
  32 import org.opendaylight.netconf.shaded.sshd.server.command.AsyncCommand;
  33 import org.slf4j.Logger;
  34 import org.slf4j.LoggerFactory;
  35
  36 final class NetconfSubsystem extends AbstractCommandSupport
  37         implements AsyncCommand, ChannelSessionAware, ChannelDataReceiver {
  38     private static final Logger LOG = LoggerFactory.getLogger(NetconfSubsystem.class);
  39
  40     private final EmbeddedChannel innerChannel = new EmbeddedChannel();
  41     private final ServerChannelInitializer channelInitializer;
  42
  43     private IoOutputStream ioOutputStream;
  44
  45     NetconfSubsystem(final String name, final ServerChannelInitializer channelInitializer) {
  46         super(name, null);
  47         this.channelInitializer = requireNonNull(channelInitializer);
  48     }
  49
  50     @Override
  51     public void run() {
  52         // not used
  53     }
  54
  55     @Override
  56     public void setIoInputStream(final IoInputStream in) {
  57         // not used
  58     }
  59
  60     @Override
  61     public void setIoErrorStream(final IoOutputStream err) {
  62         // not used
  63     }
  64
  65     @Override
  66     public void setIoOutputStream(final IoOutputStream out) {
  67         ioOutputStream = out;
  68
  69         /*
  70          * While NETCONF protocol handlers are designed to operate over Netty channel, the inner channel is used to
  71          * serve NETCONF over SSH.
  72          */
  73         // outbound packet handler, adding fist means it will be invoked last because of flow direction
  74         innerChannel.pipeline().addFirst(
  75             new ChannelOutboundHandlerAdapter() {
  76                 @Override
  77                 public void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise promise) {
  78                     if (msg instanceof ByteBuf byteBuf) {
  79                         // redirect channel outgoing packets to output stream linked to transport
  80                         final byte[] bytes = new byte[byteBuf.readableBytes()];
  81                         byteBuf.readBytes(bytes);
  82                         try {
  83                             ioOutputStream.writeBuffer(new ByteArrayBuffer(bytes))
  84                                 .addListener(future -> {
  85                                     if (future.isWritten()) {
  86                                         byteBuf.release(); // report outbound message being handled
  87                                         promise.setSuccess();
  88                                     } else if (future.getException() != null) {
  89                                         LOG.error("Error writing buffer", future.getException());
  90                                         promise.setFailure(future.getException());
  91                                     }
  92                                 });
  93                         } catch (IOException e) {
  94                             LOG.error("Error writing buffer", e);
  95                         }
  96                     }
  97                 }
  98             });
  99
 100         // inner channel termination handler
 101         innerChannel.pipeline().addLast(
 102             new ChannelInboundHandlerAdapter() {
 103                 @Override
 104                 public void channelInactive(final ChannelHandlerContext ctx) {
 105                     onExit(0);
 106                 }
 107             });
 108
 109         // NETCONF protocol handlers
 110         channelInitializer.initialize(innerChannel, new DefaultPromise<>(GlobalEventExecutor.INSTANCE));
 111         // trigger negotiation flow
 112         innerChannel.pipeline().fireChannelActive();
 113         // set additional info for upcoming netconf session
 114         innerChannel.writeInbound(Unpooled.wrappedBuffer(getHelloAdditionalMessageBytes()));
 115     }
 116
 117     @Override
 118     public void setChannelSession(final ChannelSession channelSession) {
 119         /*
 120          * Inbound packets handler
 121          * NOTE: The channel data receiver require to be set within current method, so it could be handled
 122          * with subsequent logic of ChannelSession#prepareChannelCommand() where this method is executed from.
 123          */
 124         channelSession.setDataReceiver(this);
 125     }
 126
 127     @Override
 128     public int data(final ChannelSession channel, final byte[] buf, final int start, final int len) {
 129         innerChannel.writeInbound(Unpooled.copiedBuffer(buf, start, len));
 130         return len;
 131     }
 132
 133     @Override
 134     public void close() {
 135         innerChannel.close();
 136     }
 137
 138     @Override
 139     protected void onExit(final int exitValue, final String exitMessage) {
 140         super.onExit(exitValue, exitMessage);
 141         innerChannel.close();
 142     }
 143
 144     private byte[] getHelloAdditionalMessageBytes() {
 145         final var session = getServerSession();
 146         final var address = (InetSocketAddress) session.getClientAddress();
 147         return new NetconfHelloMessageAdditionalHeader(session.getUsername(), address.getAddress().getHostAddress(),
 148             String.valueOf(address.getPort()), "ssh", "client")
 149             .toFormattedString().getBytes(StandardCharsets.UTF_8);
 150     }
 151 }