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