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.rev190614.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, topologyId);
 220         // The facade we are going it present to NetconfDevice
 221         RemoteDeviceHandler salFacade;
 222         final KeepaliveSalFacade keepAliveFacade;
 223         if (keepaliveDelay > 0) {
 224             LOG.info("Adding keepalive facade, for device {}", nodeId);
 225             salFacade = keepAliveFacade = new KeepaliveSalFacade(deviceId, deviceSalFacade,
 226                 keepaliveExecutor.getExecutor(), keepaliveDelay, node.requireDefaultRequestTimeoutMillis().toJava());
 227         } else {
 228             salFacade = deviceSalFacade;
 229             keepAliveFacade = null;
 230         }
 231
 232         // Setup reconnection on empty context, if so configured
 233         if (nodeOptional != null && nodeOptional.getIgnoreMissingSchemaSources().getAllowed()) {
 234             LOG.warn("Ignoring missing schema sources is not currently implemented for {}", deviceId);
 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 SchemaResourcesDTO resources = schemaManager.getSchemaResources(node.getSchemaCacheDirectory(),
 244                 nodeId.getValue());
 245             device = new NetconfDeviceBuilder()
 246                 .setReconnectOnSchemasChange(node.requireReconnectOnChangedSchema())
 247                 .setSchemaResourcesDTO(resources)
 248                 .setGlobalProcessingExecutor(processingExecutor)
 249                 .setId(deviceId)
 250                 .setSalFacade(salFacade)
 251                 .setDeviceActionFactory(deviceActionFactory)
 252                 .setBaseSchemas(baseSchemas)
 253                 .build();
 254             yanglibRegistrations = registerDeviceSchemaSources(deviceId, node, resources);
 255         }
 256
 257         final int rpcMessageLimit = node.requireConcurrentRpcLimit().toJava();
 258         if (rpcMessageLimit < 1) {
 259             LOG.info("Concurrent rpc limit is smaller than 1, no limit will be enforced for device {}", deviceId);
 260         }
 261
 262         final var netconfDeviceCommunicator = new NetconfDeviceCommunicator(deviceId, device, rpcMessageLimit,
 263             NetconfNodeUtils.extractUserCapabilities(node));
 264
 265         if (keepAliveFacade != null) {
 266             keepAliveFacade.setListener(netconfDeviceCommunicator);
 267         }
 268
 269         return new NetconfConnectorDTO(netconfDeviceCommunicator, salFacade, yanglibRegistrations);
 270     }
 271
 272     private static List<SchemaSourceRegistration<?>> registerDeviceSchemaSources(final RemoteDeviceId remoteDeviceId,
 273             final NetconfNode node, final SchemaResourcesDTO resources) {
 274         final var yangLibrary = node.getYangLibrary();
 275         if (yangLibrary != null) {
 276             final Uri uri = yangLibrary.getYangLibraryUrl();
 277             if (uri != null) {
 278                 final List<SchemaSourceRegistration<?>> registrations = new ArrayList<>();
 279                 final String yangLibURL = uri.getValue();
 280                 final SchemaSourceRegistry schemaRegistry = resources.getSchemaRegistry();
 281
 282                 // pre register yang library sources as fallback schemas to schema registry
 283                 final LibraryModulesSchemas schemas;
 284                 final String yangLibUsername = yangLibrary.getUsername();
 285                 final String yangLigPassword = yangLibrary.getPassword();
 286                 if (yangLibUsername != null && yangLigPassword != null) {
 287                     schemas = LibraryModulesSchemas.create(yangLibURL, yangLibUsername, yangLigPassword);
 288                 } else {
 289                     schemas = LibraryModulesSchemas.create(yangLibURL);
 290                 }
 291
 292                 for (final Map.Entry<SourceIdentifier, URL> entry : schemas.getAvailableModels().entrySet()) {
 293                     registrations.add(schemaRegistry.registerSchemaSource(new YangLibrarySchemaYangSourceProvider(
 294                         remoteDeviceId, schemas.getAvailableModels()),
 295                         PotentialSchemaSource.create(entry.getKey(), YangTextSchemaSource.class,
 296                             PotentialSchemaSource.Costs.REMOTE_IO.getValue())));
 297                 }
 298                 return List.copyOf(registrations);
 299             }
 300         }
 301
 302         return List.of();
 303     }
 304
 305     /**
 306      * Sets the private key path from location specified in configuration file using blueprint.
 307      */
 308     public void setPrivateKeyPath(final String privateKeyPath) {
 309         this.privateKeyPath = privateKeyPath;
 310     }
 311
 312     /**
 313      * Sets the private key passphrase from location specified in configuration file using blueprint.
 314      */
 315     public void setPrivateKeyPassphrase(final String privateKeyPassphrase) {
 316         this.privateKeyPassphrase = privateKeyPassphrase;
 317     }
 318
 319     public NetconfReconnectingClientConfiguration getClientConfig(final NetconfClientSessionListener listener,
 320                                                                   final NetconfNode node, final NodeId nodeId) {
 321         final ReconnectStrategyFactory sf = new TimedReconnectStrategyFactory(eventExecutor,
 322                 node.requireMaxConnectionAttempts().toJava(), node.requireBetweenAttemptsTimeoutMillis().toJava(),
 323                 node.requireSleepFactor().decimalValue());
 324         final NetconfReconnectingClientConfigurationBuilder reconnectingClientConfigurationBuilder;
 325         final var protocol = node.getProtocol();
 326         if (node.requireTcpOnly()) {
 327             reconnectingClientConfigurationBuilder = NetconfReconnectingClientConfigurationBuilder.create()
 328                     .withProtocol(NetconfClientConfiguration.NetconfClientProtocol.TCP)
 329                     .withAuthHandler(getHandlerFromCredentials(node.getCredentials()));
 330         } else if (protocol == null || protocol.getName() == Name.SSH) {
 331             reconnectingClientConfigurationBuilder = NetconfReconnectingClientConfigurationBuilder.create()
 332                     .withProtocol(NetconfClientConfiguration.NetconfClientProtocol.SSH)
 333                     .withAuthHandler(getHandlerFromCredentials(node.getCredentials()));
 334         } else if (protocol.getName() == Name.TLS) {
 335             reconnectingClientConfigurationBuilder = NetconfReconnectingClientConfigurationBuilder.create()
 336                 .withSslHandlerFactory(new SslHandlerFactoryImpl(keystoreAdapter, protocol.getSpecification()))
 337                 .withProtocol(NetconfClientConfiguration.NetconfClientProtocol.TLS);
 338         } else {
 339             throw new IllegalStateException("Unsupported protocol type: " + protocol.getName());
 340         }
 341
 342         if (node.getOdlHelloMessageCapabilities() != null) {
 343             reconnectingClientConfigurationBuilder.withOdlHelloCapabilities(
 344                     Lists.newArrayList(node.getOdlHelloMessageCapabilities().getCapability()));
 345         }
 346
 347         return reconnectingClientConfigurationBuilder
 348                 .withName(nodeId.getValue())
 349                 .withAddress(NetconfNodeUtils.toInetSocketAddress(node))
 350                 .withConnectionTimeoutMillis(node.requireConnectionTimeoutMillis().toJava())
 351                 .withReconnectStrategy(sf.createReconnectStrategy())
 352                 .withConnectStrategyFactory(sf)
 353                 .withSessionListener(listener)
 354                 .build();
 355     }
 356
 357     private AuthenticationHandler getHandlerFromCredentials(final Credentials credentials) {
 358         if (credentials
 359                 instanceof org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.device.rev221225
 360                     .credentials.credentials.LoginPassword loginPassword) {
 361             return new LoginPasswordHandler(loginPassword.getUsername(), loginPassword.getPassword());
 362         }
 363         if (credentials instanceof LoginPwUnencrypted unencrypted) {
 364             final var loginPassword = unencrypted.getLoginPasswordUnencrypted();
 365             return new LoginPasswordHandler(loginPassword.getUsername(), loginPassword.getPassword());
 366         }
 367         if (credentials instanceof LoginPw loginPw) {
 368             final var loginPassword = loginPw.getLoginPassword();
 369             return new LoginPasswordHandler(loginPassword.getUsername(),
 370                     encryptionService.decrypt(loginPassword.getPassword()));
 371         }
 372         if (credentials instanceof KeyAuth keyAuth) {
 373             final var keyPair = keyAuth.getKeyBased();
 374             return new DatastoreBackedPublicKeyAuth(keyPair.getUsername(), keyPair.getKeyId(),
 375                     keystoreAdapter, encryptionService);
 376         }
 377         throw new IllegalStateException("Unsupported credential type: " + credentials.getClass());
 378     }
 379 }