netconf/netconf-topology/src/main/java/org/opendaylight/netconf/topology/spi/AbstractNetconfTopology.java

   1 /*
   2  * Copyright (c) 2015 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 package org.opendaylight.netconf.topology.spi;
   9
  10 import static java.util.Objects.requireNonNull;
  11
  12 import com.google.common.annotations.VisibleForTesting;
  13 import com.google.common.collect.Lists;
  14 import com.google.common.util.concurrent.FutureCallback;
  15 import com.google.common.util.concurrent.Futures;
  16 import com.google.common.util.concurrent.ListenableFuture;
  17 import com.google.common.util.concurrent.ListeningExecutorService;
  18 import com.google.common.util.concurrent.MoreExecutors;
  19 import io.netty.util.concurrent.EventExecutor;
  20 import java.net.URL;
  21 import java.util.ArrayList;
  22 import java.util.HashMap;
  23 import java.util.List;
  24 import java.util.Map;
  25 import java.util.concurrent.ExecutionException;
  26 import org.opendaylight.aaa.encrypt.AAAEncryptionService;
  27 import org.opendaylight.controller.config.threadpool.ScheduledThreadPool;
  28 import org.opendaylight.controller.config.threadpool.ThreadPool;
  29 import org.opendaylight.mdsal.binding.api.DataBroker;
  30 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  31 import org.opendaylight.mdsal.dom.api.DOMMountPointService;
  32 import org.opendaylight.netconf.client.NetconfClientDispatcher;
  33 import org.opendaylight.netconf.client.NetconfClientSessionListener;
  34 import org.opendaylight.netconf.client.conf.NetconfClientConfiguration;
  35 import org.opendaylight.netconf.client.conf.NetconfReconnectingClientConfiguration;
  36 import org.opendaylight.netconf.client.conf.NetconfReconnectingClientConfigurationBuilder;
  37 import org.opendaylight.netconf.nettyutil.ReconnectStrategyFactory;
  38 import org.opendaylight.netconf.nettyutil.TimedReconnectStrategyFactory;
  39 import org.opendaylight.netconf.nettyutil.handler.ssh.authentication.AuthenticationHandler;
  40 import org.opendaylight.netconf.nettyutil.handler.ssh.authentication.LoginPasswordHandler;
  41 import org.opendaylight.netconf.sal.connect.api.DeviceActionFactory;
  42 import org.opendaylight.netconf.sal.connect.api.RemoteDevice;
  43 import org.opendaylight.netconf.sal.connect.api.RemoteDeviceHandler;
  44 import org.opendaylight.netconf.sal.connect.api.RemoteDeviceId;
  45 import org.opendaylight.netconf.sal.connect.api.SchemaResourceManager;
  46 import org.opendaylight.netconf.sal.connect.netconf.LibraryModulesSchemas;
  47 import org.opendaylight.netconf.sal.connect.netconf.NetconfDevice.SchemaResourcesDTO;
  48 import org.opendaylight.netconf.sal.connect.netconf.NetconfDeviceBuilder;
  49 import org.opendaylight.netconf.sal.connect.netconf.SchemalessNetconfDevice;
  50 import org.opendaylight.netconf.sal.connect.netconf.auth.DatastoreBackedPublicKeyAuth;
  51 import org.opendaylight.netconf.sal.connect.netconf.listener.NetconfDeviceCommunicator;
  52 import org.opendaylight.netconf.sal.connect.netconf.sal.KeepaliveSalFacade;
  53 import org.opendaylight.netconf.sal.connect.netconf.sal.NetconfKeystoreAdapter;
  54 import org.opendaylight.netconf.sal.connect.netconf.schema.YangLibrarySchemaYangSourceProvider;
  55 import org.opendaylight.netconf.sal.connect.netconf.schema.mapping.BaseNetconfSchemas;
  56 import org.opendaylight.netconf.sal.connect.util.SslHandlerFactoryImpl;
  57 import org.opendaylight.netconf.topology.api.NetconfTopology;
  58 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.Uri;
  59 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.device.rev221225.connection.parameters.Protocol.Name;
  60 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.device.rev221225.credentials.Credentials;
  61 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.device.rev221225.credentials.credentials.KeyAuth;
  62 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.device.rev221225.credentials.credentials.LoginPw;
  63 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.device.rev221225.credentials.credentials.LoginPwUnencrypted;
  64 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.optional.rev221225.NetconfNodeAugmentedOptional;
  65 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev221225.NetconfNode;
  66 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NetworkTopology;
  67 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NodeId;
  68 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.TopologyId;
  69 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
  70 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyBuilder;
  71 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyKey;
  72 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
  73 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  74 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
  75 import org.opendaylight.yangtools.yang.common.Empty;
  76 import org.opendaylight.yangtools.yang.model.repo.api.SourceIdentifier;
  77 import org.opendaylight.yangtools.yang.model.repo.api.YangTextSchemaSource;
  78 import org.opendaylight.yangtools.yang.model.repo.spi.PotentialSchemaSource;
  79 import org.opendaylight.yangtools.yang.model.repo.spi.SchemaSourceRegistration;
  80 import org.opendaylight.yangtools.yang.model.repo.spi.SchemaSourceRegistry;
  81 import org.slf4j.Logger;
  82 import org.slf4j.LoggerFactory;
  83
  84 public abstract class AbstractNetconfTopology implements NetconfTopology {
  85     private static final Logger LOG = LoggerFactory.getLogger(AbstractNetconfTopology.class);
  86
  87     // FIXME: extract this into caller and pass to constructor
  88     @Deprecated(forRemoval = true)
  89     public static final KeyedInstanceIdentifier<Topology, TopologyKey> DEFAULT_TOPOLOGY_IID =
  90         InstanceIdentifier.create(NetworkTopology.class)
  91         .child(Topology.class, new TopologyKey(new TopologyId(RemoteDeviceId.DEFAULT_TOPOLOGY_NAME)));
  92
  93     private final NetconfClientDispatcher clientDispatcher;
  94     private final EventExecutor eventExecutor;
  95     private final DeviceActionFactory deviceActionFactory;
  96     private final NetconfKeystoreAdapter keystoreAdapter;
  97     private final SchemaResourceManager schemaManager;
  98     private final BaseNetconfSchemas baseSchemas;
  99
 100     protected final ScheduledThreadPool keepaliveExecutor;
 101     protected final ListeningExecutorService processingExecutor;
 102     protected final DataBroker dataBroker;
 103     protected final DOMMountPointService mountPointService;
 104     protected final String topologyId;
 105     protected String privateKeyPath;
 106     protected String privateKeyPassphrase;
 107     protected final AAAEncryptionService encryptionService;
 108     protected final HashMap<NodeId, NetconfConnectorDTO> activeConnectors = new HashMap<>();
 109
 110     protected AbstractNetconfTopology(final String topologyId, final NetconfClientDispatcher clientDispatcher,
 111                                       final EventExecutor eventExecutor, final ScheduledThreadPool keepaliveExecutor,
 112                                       final ThreadPool processingExecutor, final SchemaResourceManager schemaManager,
 113                                       final DataBroker dataBroker, final DOMMountPointService mountPointService,
 114                                       final AAAEncryptionService encryptionService,
 115                                       final DeviceActionFactory deviceActionFactory,
 116                                       final BaseNetconfSchemas baseSchemas) {
 117         this.topologyId = requireNonNull(topologyId);
 118         this.clientDispatcher = clientDispatcher;
 119         this.eventExecutor = eventExecutor;
 120         this.keepaliveExecutor = keepaliveExecutor;
 121         this.processingExecutor = MoreExecutors.listeningDecorator(processingExecutor.getExecutor());
 122         this.schemaManager = requireNonNull(schemaManager);
 123         this.deviceActionFactory = deviceActionFactory;
 124         this.dataBroker = requireNonNull(dataBroker);
 125         this.mountPointService = mountPointService;
 126         this.encryptionService = encryptionService;
 127         this.baseSchemas = requireNonNull(baseSchemas);
 128
 129         keystoreAdapter = new NetconfKeystoreAdapter(dataBroker);
 130
 131         // FIXME: this should be a put(), as we are initializing and will be re-populating the datastore with all the
 132         //        devices. Whatever has been there before should be nuked to properly re-align lifecycle.
 133         final var wtx = dataBroker.newWriteOnlyTransaction();
 134         wtx.merge(LogicalDatastoreType.OPERATIONAL, InstanceIdentifier.builder(NetworkTopology.class)
 135             .child(Topology.class, new TopologyKey(new TopologyId(topologyId)))
 136             .build(), new TopologyBuilder().setTopologyId(new TopologyId(topologyId)).build());
 137         final var future = wtx.commit();
 138         try {
 139             future.get();
 140         } catch (InterruptedException | ExecutionException e) {
 141             LOG.error("Unable to initialize topology {}", topologyId, e);
 142             throw new IllegalStateException(e);
 143         }
 144
 145         LOG.debug("Topology {} initialized", topologyId);
 146     }
 147
 148     @Override
 149     public ListenableFuture<Empty> connectNode(final NodeId nodeId, final Node configNode) {
 150         LOG.info("Connecting RemoteDevice{{}} , with config {}", nodeId, hideCredentials(configNode));
 151         return setupConnection(nodeId, configNode);
 152     }
 153
 154     /**
 155      * Hiding of private credentials from node configuration (credentials data is replaced by asterisks).
 156      *
 157      * @param nodeConfiguration Node configuration container.
 158      * @return String representation of node configuration with credentials replaced by asterisks.
 159      */
 160     @VisibleForTesting
 161     public static String hideCredentials(final Node nodeConfiguration) {
 162         final NetconfNode netconfNodeAugmentation = nodeConfiguration.augmentation(NetconfNode.class);
 163         final String nodeCredentials = netconfNodeAugmentation.getCredentials().toString();
 164         final String nodeConfigurationString = nodeConfiguration.toString();
 165         return nodeConfigurationString.replace(nodeCredentials, "***");
 166     }
 167
 168     @Override
 169     public ListenableFuture<Empty> disconnectNode(final NodeId nodeId) {
 170         final var nodeName = nodeId.getValue();
 171         LOG.debug("Disconnecting RemoteDevice{{}}", nodeName);
 172
 173         final NetconfConnectorDTO connectorDTO = activeConnectors.remove(nodeId);
 174         if (connectorDTO == null) {
 175             return Futures.immediateFailedFuture(
 176                 new IllegalStateException("Cannot disconnect " + nodeName + " as it is not connected"));
 177         }
 178
 179         connectorDTO.close();
 180         return Futures.immediateFuture(Empty.value());
 181     }
 182
 183     protected ListenableFuture<Empty> setupConnection(final NodeId nodeId, final Node configNode) {
 184         final NetconfNode netconfNode = configNode.augmentation(NetconfNode.class);
 185         final NetconfNodeAugmentedOptional nodeOptional = configNode.augmentation(NetconfNodeAugmentedOptional.class);
 186
 187         requireNonNull(netconfNode.getHost());
 188         requireNonNull(netconfNode.getPort());
 189
 190         final NetconfConnectorDTO deviceCommunicatorDTO = createDeviceCommunicator(nodeId, netconfNode, nodeOptional);
 191         final NetconfDeviceCommunicator deviceCommunicator = deviceCommunicatorDTO.getCommunicator();
 192         final NetconfClientSessionListener netconfClientSessionListener = deviceCommunicatorDTO.getSessionListener();
 193         final NetconfReconnectingClientConfiguration clientConfig =
 194                 getClientConfig(netconfClientSessionListener, netconfNode, nodeId);
 195         final ListenableFuture<Empty> future =
 196                 deviceCommunicator.initializeRemoteConnection(clientDispatcher, clientConfig);
 197
 198         activeConnectors.put(nodeId, deviceCommunicatorDTO);
 199
 200         Futures.addCallback(future, new FutureCallback<>() {
 201             @Override
 202             public void onSuccess(final Empty result) {
 203                 LOG.debug("Connector for {} started succesfully", nodeId.getValue());
 204             }
 205
 206             @Override
 207             public void onFailure(final Throwable throwable) {
 208                 LOG.error("Connector for {} failed", nodeId.getValue(), throwable);
 209                 // remove this node from active connectors?
 210             }
 211         }, MoreExecutors.directExecutor());
 212
 213         return future;
 214     }
 215
 216     protected NetconfConnectorDTO createDeviceCommunicator(final NodeId nodeId, final NetconfNode node) {
 217         return createDeviceCommunicator(nodeId, node, null);
 218     }
 219
 220     protected NetconfConnectorDTO createDeviceCommunicator(final NodeId nodeId, final NetconfNode node,
 221             final NetconfNodeAugmentedOptional nodeOptional) {
 222         final var deviceId = NetconfNodeUtils.toRemoteDeviceId(nodeId, node);
 223         final long keepaliveDelay = node.requireKeepaliveDelay().toJava();
 224
 225         final var deviceSalFacade = new NetconfTopologyDeviceSalFacade(deviceId, mountPointService,
 226             node.requireLockDatastore(), dataBroker);
 227         // The facade we are going it present to NetconfDevice
 228         RemoteDeviceHandler salFacade;
 229         final KeepaliveSalFacade keepAliveFacade;
 230         if (keepaliveDelay > 0) {
 231             LOG.info("Adding keepalive facade, for device {}", nodeId);
 232             salFacade = keepAliveFacade = new KeepaliveSalFacade(deviceId, deviceSalFacade,
 233                 keepaliveExecutor.getExecutor(), keepaliveDelay, node.requireDefaultRequestTimeoutMillis().toJava());
 234         } else {
 235             salFacade = deviceSalFacade;
 236             keepAliveFacade = null;
 237         }
 238
 239         // Setup reconnection on empty context, if so configured
 240         if (nodeOptional != null && nodeOptional.getIgnoreMissingSchemaSources().getAllowed()) {
 241             LOG.warn("Ignoring missing schema sources is not currently implemented for {}", deviceId);
 242         }
 243
 244         final RemoteDevice<NetconfDeviceCommunicator> device;
 245         final List<SchemaSourceRegistration<?>> yanglibRegistrations;
 246         if (node.requireSchemaless()) {
 247             device = new SchemalessNetconfDevice(baseSchemas, deviceId, salFacade);
 248             yanglibRegistrations = List.of();
 249         } else {
 250             final SchemaResourcesDTO resources = schemaManager.getSchemaResources(node.getSchemaCacheDirectory(),
 251                 nodeId.getValue());
 252             device = new NetconfDeviceBuilder()
 253                 .setReconnectOnSchemasChange(node.requireReconnectOnChangedSchema())
 254                 .setSchemaResourcesDTO(resources)
 255                 .setGlobalProcessingExecutor(processingExecutor)
 256                 .setId(deviceId)
 257                 .setSalFacade(salFacade)
 258                 .setDeviceActionFactory(deviceActionFactory)
 259                 .setBaseSchemas(baseSchemas)
 260                 .build();
 261             yanglibRegistrations = registerDeviceSchemaSources(deviceId, node, resources);
 262         }
 263
 264         final int rpcMessageLimit = node.requireConcurrentRpcLimit().toJava();
 265         if (rpcMessageLimit < 1) {
 266             LOG.info("Concurrent rpc limit is smaller than 1, no limit will be enforced for device {}", deviceId);
 267         }
 268
 269         final var netconfDeviceCommunicator = new NetconfDeviceCommunicator(deviceId, device, rpcMessageLimit,
 270             NetconfNodeUtils.extractUserCapabilities(node));
 271
 272         if (keepAliveFacade != null) {
 273             keepAliveFacade.setListener(netconfDeviceCommunicator);
 274         }
 275
 276         return new NetconfConnectorDTO(netconfDeviceCommunicator, salFacade, yanglibRegistrations);
 277     }
 278
 279     private static List<SchemaSourceRegistration<?>> registerDeviceSchemaSources(final RemoteDeviceId remoteDeviceId,
 280             final NetconfNode node, final SchemaResourcesDTO resources) {
 281         final var yangLibrary = node.getYangLibrary();
 282         if (yangLibrary != null) {
 283             final Uri uri = yangLibrary.getYangLibraryUrl();
 284             if (uri != null) {
 285                 final List<SchemaSourceRegistration<?>> registrations = new ArrayList<>();
 286                 final String yangLibURL = uri.getValue();
 287                 final SchemaSourceRegistry schemaRegistry = resources.getSchemaRegistry();
 288
 289                 // pre register yang library sources as fallback schemas to schema registry
 290                 final LibraryModulesSchemas schemas;
 291                 final String yangLibUsername = yangLibrary.getUsername();
 292                 final String yangLigPassword = yangLibrary.getPassword();
 293                 if (yangLibUsername != null && yangLigPassword != null) {
 294                     schemas = LibraryModulesSchemas.create(yangLibURL, yangLibUsername, yangLigPassword);
 295                 } else {
 296                     schemas = LibraryModulesSchemas.create(yangLibURL);
 297                 }
 298
 299                 for (final Map.Entry<SourceIdentifier, URL> entry : schemas.getAvailableModels().entrySet()) {
 300                     registrations.add(schemaRegistry.registerSchemaSource(new YangLibrarySchemaYangSourceProvider(
 301                         remoteDeviceId, schemas.getAvailableModels()),
 302                         PotentialSchemaSource.create(entry.getKey(), YangTextSchemaSource.class,
 303                             PotentialSchemaSource.Costs.REMOTE_IO.getValue())));
 304                 }
 305                 return List.copyOf(registrations);
 306             }
 307         }
 308
 309         return List.of();
 310     }
 311
 312     /**
 313      * Sets the private key path from location specified in configuration file using blueprint.
 314      */
 315     public void setPrivateKeyPath(final String privateKeyPath) {
 316         this.privateKeyPath = privateKeyPath;
 317     }
 318
 319     /**
 320      * Sets the private key passphrase from location specified in configuration file using blueprint.
 321      */
 322     public void setPrivateKeyPassphrase(final String privateKeyPassphrase) {
 323         this.privateKeyPassphrase = privateKeyPassphrase;
 324     }
 325
 326     public NetconfReconnectingClientConfiguration getClientConfig(final NetconfClientSessionListener listener,
 327                                                                   final NetconfNode node, final NodeId nodeId) {
 328         final ReconnectStrategyFactory sf = new TimedReconnectStrategyFactory(eventExecutor,
 329                 node.requireMaxConnectionAttempts().toJava(), node.requireBetweenAttemptsTimeoutMillis().toJava(),
 330                 node.requireSleepFactor().decimalValue());
 331         final NetconfReconnectingClientConfigurationBuilder reconnectingClientConfigurationBuilder;
 332         final var protocol = node.getProtocol();
 333         if (node.requireTcpOnly()) {
 334             reconnectingClientConfigurationBuilder = NetconfReconnectingClientConfigurationBuilder.create()
 335                     .withProtocol(NetconfClientConfiguration.NetconfClientProtocol.TCP)
 336                     .withAuthHandler(getHandlerFromCredentials(node.getCredentials()));
 337         } else if (protocol == null || protocol.getName() == Name.SSH) {
 338             reconnectingClientConfigurationBuilder = NetconfReconnectingClientConfigurationBuilder.create()
 339                     .withProtocol(NetconfClientConfiguration.NetconfClientProtocol.SSH)
 340                     .withAuthHandler(getHandlerFromCredentials(node.getCredentials()));
 341         } else if (protocol.getName() == Name.TLS) {
 342             reconnectingClientConfigurationBuilder = NetconfReconnectingClientConfigurationBuilder.create()
 343                 .withSslHandlerFactory(new SslHandlerFactoryImpl(keystoreAdapter, protocol.getSpecification()))
 344                 .withProtocol(NetconfClientConfiguration.NetconfClientProtocol.TLS);
 345         } else {
 346             throw new IllegalStateException("Unsupported protocol type: " + protocol.getName());
 347         }
 348
 349         if (node.getOdlHelloMessageCapabilities() != null) {
 350             reconnectingClientConfigurationBuilder.withOdlHelloCapabilities(
 351                     Lists.newArrayList(node.getOdlHelloMessageCapabilities().getCapability()));
 352         }
 353
 354         return reconnectingClientConfigurationBuilder
 355                 .withName(nodeId.getValue())
 356                 .withAddress(NetconfNodeUtils.toInetSocketAddress(node))
 357                 .withConnectionTimeoutMillis(node.requireConnectionTimeoutMillis().toJava())
 358                 .withReconnectStrategy(sf.createReconnectStrategy())
 359                 .withConnectStrategyFactory(sf)
 360                 .withSessionListener(listener)
 361                 .build();
 362     }
 363
 364     private AuthenticationHandler getHandlerFromCredentials(final Credentials credentials) {
 365         if (credentials
 366                 instanceof org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.device.rev221225
 367                     .credentials.credentials.LoginPassword loginPassword) {
 368             return new LoginPasswordHandler(loginPassword.getUsername(), loginPassword.getPassword());
 369         }
 370         if (credentials instanceof LoginPwUnencrypted unencrypted) {
 371             final var loginPassword = unencrypted.getLoginPasswordUnencrypted();
 372             return new LoginPasswordHandler(loginPassword.getUsername(), loginPassword.getPassword());
 373         }
 374         if (credentials instanceof LoginPw loginPw) {
 375             final var loginPassword = loginPw.getLoginPassword();
 376             return new LoginPasswordHandler(loginPassword.getUsername(),
 377                     encryptionService.decrypt(loginPassword.getPassword()));
 378         }
 379         if (credentials instanceof KeyAuth keyAuth) {
 380             final var keyPair = keyAuth.getKeyBased();
 381             return new DatastoreBackedPublicKeyAuth(keyPair.getUsername(), keyPair.getKeyId(),
 382                     keystoreAdapter, encryptionService);
 383         }
 384         throw new IllegalStateException("Unsupported credential type: " + credentials.getClass());
 385     }
 386 }