apps/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.client.mdsal.DatastoreBackedPublicKeyAuth;
  38 import org.opendaylight.netconf.client.mdsal.LibraryModulesSchemas;
  39 import org.opendaylight.netconf.client.mdsal.LibrarySchemaSourceProvider;
  40 import org.opendaylight.netconf.client.mdsal.NetconfDevice.SchemaResourcesDTO;
  41 import org.opendaylight.netconf.client.mdsal.NetconfDeviceBuilder;
  42 import org.opendaylight.netconf.client.mdsal.NetconfDeviceCommunicator;
  43 import org.opendaylight.netconf.client.mdsal.SchemalessNetconfDevice;
  44 import org.opendaylight.netconf.client.mdsal.api.BaseNetconfSchemas;
  45 import org.opendaylight.netconf.client.mdsal.api.CredentialProvider;
  46 import org.opendaylight.netconf.client.mdsal.api.DeviceActionFactory;
  47 import org.opendaylight.netconf.client.mdsal.api.RemoteDevice;
  48 import org.opendaylight.netconf.client.mdsal.api.RemoteDeviceHandler;
  49 import org.opendaylight.netconf.client.mdsal.api.RemoteDeviceId;
  50 import org.opendaylight.netconf.client.mdsal.api.SchemaResourceManager;
  51 import org.opendaylight.netconf.client.mdsal.api.SslHandlerFactoryProvider;
  52 import org.opendaylight.netconf.nettyutil.ReconnectStrategyFactory;
  53 import org.opendaylight.netconf.nettyutil.TimedReconnectStrategyFactory;
  54 import org.opendaylight.netconf.nettyutil.handler.ssh.authentication.AuthenticationHandler;
  55 import org.opendaylight.netconf.nettyutil.handler.ssh.authentication.LoginPasswordHandler;
  56 import org.opendaylight.netconf.sal.connect.netconf.sal.KeepaliveSalFacade;
  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.rev230430.connection.parameters.Protocol.Name;
  60 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.device.rev230430.credentials.Credentials;
  61 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.device.rev230430.credentials.credentials.KeyAuth;
  62 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.device.rev230430.credentials.credentials.LoginPw;
  63 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.device.rev230430.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.common.Empty;
  75 import org.opendaylight.yangtools.yang.model.repo.api.SourceIdentifier;
  76 import org.opendaylight.yangtools.yang.model.repo.api.YangTextSchemaSource;
  77 import org.opendaylight.yangtools.yang.model.repo.spi.PotentialSchemaSource;
  78 import org.opendaylight.yangtools.yang.model.repo.spi.SchemaSourceRegistration;
  79 import org.opendaylight.yangtools.yang.model.repo.spi.SchemaSourceRegistry;
  80 import org.slf4j.Logger;
  81 import org.slf4j.LoggerFactory;
  82
  83 public abstract class AbstractNetconfTopology implements NetconfTopology {
  84     private static final Logger LOG = LoggerFactory.getLogger(AbstractNetconfTopology.class);
  85
  86     private final NetconfClientDispatcher clientDispatcher;
  87     private final EventExecutor eventExecutor;
  88     private final DeviceActionFactory deviceActionFactory;
  89     private final CredentialProvider credentialProvider;
  90     private final SslHandlerFactoryProvider sslHandlerFactoryProvider;
  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 final AAAEncryptionService encryptionService;
 100     protected final HashMap<NodeId, NetconfConnectorDTO> activeConnectors = new HashMap<>();
 101
 102     protected AbstractNetconfTopology(final String topologyId, final NetconfClientDispatcher clientDispatcher,
 103                                       final EventExecutor eventExecutor, final ScheduledThreadPool keepaliveExecutor,
 104                                       final ThreadPool processingExecutor, final SchemaResourceManager schemaManager,
 105                                       final DataBroker dataBroker, final DOMMountPointService mountPointService,
 106                                       final AAAEncryptionService encryptionService,
 107                                       final DeviceActionFactory deviceActionFactory,
 108                                       final BaseNetconfSchemas baseSchemas,
 109                                       final CredentialProvider credentialProvider,
 110                                       final SslHandlerFactoryProvider sslHandlerFactoryProvider) {
 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         this.credentialProvider = requireNonNull(credentialProvider);
 123         this.sslHandlerFactoryProvider = requireNonNull(sslHandlerFactoryProvider);
 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             final NetconfNodeAugmentedOptional nodeOptional) {
 212         final var deviceId = NetconfNodeUtils.toRemoteDeviceId(nodeId, node);
 213         final long keepaliveDelay = node.requireKeepaliveDelay().toJava();
 214
 215         final var deviceSalFacade = new NetconfTopologyDeviceSalFacade(deviceId, mountPointService,
 216             node.requireLockDatastore(), dataBroker);
 217         // The facade we are going it present to NetconfDevice
 218         RemoteDeviceHandler salFacade;
 219         final KeepaliveSalFacade keepAliveFacade;
 220         if (keepaliveDelay > 0) {
 221             LOG.info("Adding keepalive facade, for device {}", nodeId);
 222             salFacade = keepAliveFacade = new KeepaliveSalFacade(deviceId, deviceSalFacade,
 223                 keepaliveExecutor.getExecutor(), keepaliveDelay, node.requireDefaultRequestTimeoutMillis().toJava());
 224         } else {
 225             salFacade = deviceSalFacade;
 226             keepAliveFacade = null;
 227         }
 228
 229         // Setup reconnection on empty context, if so configured
 230         if (nodeOptional != null && nodeOptional.getIgnoreMissingSchemaSources().getAllowed()) {
 231             LOG.warn("Ignoring missing schema sources is not currently implemented for {}", deviceId);
 232         }
 233
 234         final RemoteDevice<NetconfDeviceCommunicator> device;
 235         final List<SchemaSourceRegistration<?>> yanglibRegistrations;
 236         if (node.requireSchemaless()) {
 237             device = new SchemalessNetconfDevice(baseSchemas, deviceId, salFacade);
 238             yanglibRegistrations = List.of();
 239         } else {
 240             final SchemaResourcesDTO resources = schemaManager.getSchemaResources(node.getSchemaCacheDirectory(),
 241                 nodeId.getValue());
 242             device = new NetconfDeviceBuilder()
 243                 .setReconnectOnSchemasChange(node.requireReconnectOnChangedSchema())
 244                 .setSchemaResourcesDTO(resources)
 245                 .setGlobalProcessingExecutor(processingExecutor)
 246                 .setId(deviceId)
 247                 .setSalFacade(salFacade)
 248                 .setDeviceActionFactory(deviceActionFactory)
 249                 .setBaseSchemas(baseSchemas)
 250                 .build();
 251             yanglibRegistrations = registerDeviceSchemaSources(deviceId, node, resources);
 252         }
 253
 254         final int rpcMessageLimit = node.requireConcurrentRpcLimit().toJava();
 255         if (rpcMessageLimit < 1) {
 256             LOG.info("Concurrent rpc limit is smaller than 1, no limit will be enforced for device {}", deviceId);
 257         }
 258
 259         final var netconfDeviceCommunicator = new NetconfDeviceCommunicator(deviceId, device, rpcMessageLimit,
 260             NetconfNodeUtils.extractUserCapabilities(node));
 261
 262         if (keepAliveFacade != null) {
 263             keepAliveFacade.setListener(netconfDeviceCommunicator);
 264         }
 265
 266         return new NetconfConnectorDTO(netconfDeviceCommunicator, salFacade, yanglibRegistrations);
 267     }
 268
 269     private static List<SchemaSourceRegistration<?>> registerDeviceSchemaSources(final RemoteDeviceId remoteDeviceId,
 270             final NetconfNode node, final SchemaResourcesDTO resources) {
 271         final var yangLibrary = node.getYangLibrary();
 272         if (yangLibrary != null) {
 273             final Uri uri = yangLibrary.getYangLibraryUrl();
 274             if (uri != null) {
 275                 final List<SchemaSourceRegistration<?>> registrations = new ArrayList<>();
 276                 final String yangLibURL = uri.getValue();
 277                 final SchemaSourceRegistry schemaRegistry = resources.getSchemaRegistry();
 278
 279                 // pre register yang library sources as fallback schemas to schema registry
 280                 final LibraryModulesSchemas schemas;
 281                 final String yangLibUsername = yangLibrary.getUsername();
 282                 final String yangLigPassword = yangLibrary.getPassword();
 283                 if (yangLibUsername != null && yangLigPassword != null) {
 284                     schemas = LibraryModulesSchemas.create(yangLibURL, yangLibUsername, yangLigPassword);
 285                 } else {
 286                     schemas = LibraryModulesSchemas.create(yangLibURL);
 287                 }
 288
 289                 for (final Map.Entry<SourceIdentifier, URL> entry : schemas.getAvailableModels().entrySet()) {
 290                     registrations.add(schemaRegistry.registerSchemaSource(new LibrarySchemaSourceProvider(
 291                         remoteDeviceId, schemas.getAvailableModels()),
 292                         PotentialSchemaSource.create(entry.getKey(), YangTextSchemaSource.class,
 293                             PotentialSchemaSource.Costs.REMOTE_IO.getValue())));
 294                 }
 295                 return List.copyOf(registrations);
 296             }
 297         }
 298
 299         return List.of();
 300     }
 301
 302     public NetconfReconnectingClientConfiguration getClientConfig(final NetconfClientSessionListener listener,
 303                                                                   final NetconfNode node, final NodeId nodeId) {
 304         final ReconnectStrategyFactory sf = new TimedReconnectStrategyFactory(eventExecutor,
 305                 node.requireMaxConnectionAttempts().toJava(), node.requireBetweenAttemptsTimeoutMillis().toJava(),
 306                 node.requireSleepFactor().decimalValue());
 307         final NetconfReconnectingClientConfigurationBuilder reconnectingClientConfigurationBuilder;
 308         final var protocol = node.getProtocol();
 309         if (node.requireTcpOnly()) {
 310             reconnectingClientConfigurationBuilder = NetconfReconnectingClientConfigurationBuilder.create()
 311                     .withProtocol(NetconfClientConfiguration.NetconfClientProtocol.TCP)
 312                     .withAuthHandler(getHandlerFromCredentials(node.getCredentials()));
 313         } else if (protocol == null || protocol.getName() == Name.SSH) {
 314             reconnectingClientConfigurationBuilder = NetconfReconnectingClientConfigurationBuilder.create()
 315                     .withProtocol(NetconfClientConfiguration.NetconfClientProtocol.SSH)
 316                     .withAuthHandler(getHandlerFromCredentials(node.getCredentials()));
 317         } else if (protocol.getName() == Name.TLS) {
 318             reconnectingClientConfigurationBuilder = NetconfReconnectingClientConfigurationBuilder.create()
 319                 .withSslHandlerFactory(sslHandlerFactoryProvider.getSslHandlerFactory(protocol.getSpecification()))
 320                 .withProtocol(NetconfClientConfiguration.NetconfClientProtocol.TLS);
 321         } else {
 322             throw new IllegalStateException("Unsupported protocol type: " + protocol.getName());
 323         }
 324
 325         if (node.getOdlHelloMessageCapabilities() != null) {
 326             reconnectingClientConfigurationBuilder.withOdlHelloCapabilities(
 327                     Lists.newArrayList(node.getOdlHelloMessageCapabilities().getCapability()));
 328         }
 329
 330         return reconnectingClientConfigurationBuilder
 331                 .withName(nodeId.getValue())
 332                 .withAddress(NetconfNodeUtils.toInetSocketAddress(node))
 333                 .withConnectionTimeoutMillis(node.requireConnectionTimeoutMillis().toJava())
 334                 .withReconnectStrategy(sf.createReconnectStrategy())
 335                 .withConnectStrategyFactory(sf)
 336                 .withSessionListener(listener)
 337                 .build();
 338     }
 339
 340     private AuthenticationHandler getHandlerFromCredentials(final Credentials credentials) {
 341         if (credentials
 342                 instanceof org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.device.rev230430
 343                     .credentials.credentials.LoginPassword loginPassword) {
 344             return new LoginPasswordHandler(loginPassword.getUsername(), loginPassword.getPassword());
 345         }
 346         if (credentials instanceof LoginPwUnencrypted unencrypted) {
 347             final var loginPassword = unencrypted.getLoginPasswordUnencrypted();
 348             return new LoginPasswordHandler(loginPassword.getUsername(), loginPassword.getPassword());
 349         }
 350         if (credentials instanceof LoginPw loginPw) {
 351             final var loginPassword = loginPw.getLoginPassword();
 352             return new LoginPasswordHandler(loginPassword.getUsername(),
 353                     encryptionService.decrypt(loginPassword.getPassword()));
 354         }
 355         if (credentials instanceof KeyAuth keyAuth) {
 356             final var keyPair = keyAuth.getKeyBased();
 357             return new DatastoreBackedPublicKeyAuth(keyPair.getUsername(), keyPair.getKeyId(), credentialProvider,
 358                 encryptionService);
 359         }
 360         throw new IllegalStateException("Unsupported credential type: " + credentials.getClass());
 361     }
 362 }