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