netconf/netconf-topology/src/main/java/org/opendaylight/netconf/topology/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
   9 package org.opendaylight.netconf.topology;
  10
  11 import com.google.common.base.Optional;
  12 import com.google.common.base.Preconditions;
  13 import com.google.common.base.Strings;
  14 import com.google.common.collect.Lists;
  15 import com.google.common.collect.Sets;
  16 import com.google.common.util.concurrent.FutureCallback;
  17 import com.google.common.util.concurrent.Futures;
  18 import com.google.common.util.concurrent.ListenableFuture;
  19 import com.google.common.util.concurrent.MoreExecutors;
  20 import com.google.common.util.concurrent.Uninterruptibles;
  21 import io.netty.handler.ssl.SslHandler;
  22 import io.netty.util.concurrent.EventExecutor;
  23 import java.io.File;
  24 import java.io.IOException;
  25 import java.math.BigDecimal;
  26 import java.net.InetSocketAddress;
  27 import java.net.URL;
  28 import java.security.GeneralSecurityException;
  29 import java.security.KeyStore;
  30 import java.util.ArrayList;
  31 import java.util.HashMap;
  32 import java.util.List;
  33 import java.util.Map;
  34 import java.util.Set;
  35 import java.util.concurrent.TimeUnit;
  36 import javax.net.ssl.KeyManagerFactory;
  37 import javax.net.ssl.SSLContext;
  38 import javax.net.ssl.SSLEngine;
  39 import javax.net.ssl.TrustManagerFactory;
  40 import org.opendaylight.aaa.encrypt.AAAEncryptionService;
  41 import org.opendaylight.controller.config.threadpool.ScheduledThreadPool;
  42 import org.opendaylight.controller.config.threadpool.ThreadPool;
  43 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  44 import org.opendaylight.controller.md.sal.dom.api.DOMMountPointService;
  45 import org.opendaylight.netconf.api.NetconfMessage;
  46 import org.opendaylight.netconf.client.NetconfClientDispatcher;
  47 import org.opendaylight.netconf.client.NetconfClientSessionListener;
  48 import org.opendaylight.netconf.client.SslHandlerFactory;
  49 import org.opendaylight.netconf.client.conf.NetconfClientConfiguration;
  50 import org.opendaylight.netconf.client.conf.NetconfReconnectingClientConfiguration;
  51 import org.opendaylight.netconf.client.conf.NetconfReconnectingClientConfigurationBuilder;
  52 import org.opendaylight.netconf.nettyutil.handler.ssh.authentication.AuthenticationHandler;
  53 import org.opendaylight.netconf.nettyutil.handler.ssh.authentication.LoginPasswordHandler;
  54 import org.opendaylight.netconf.sal.connect.api.RemoteDevice;
  55 import org.opendaylight.netconf.sal.connect.api.RemoteDeviceHandler;
  56 import org.opendaylight.netconf.sal.connect.netconf.LibraryModulesSchemas;
  57 import org.opendaylight.netconf.sal.connect.netconf.NetconfDevice;
  58 import org.opendaylight.netconf.sal.connect.netconf.NetconfDeviceBuilder;
  59 import org.opendaylight.netconf.sal.connect.netconf.NetconfStateSchemasResolverImpl;
  60 import org.opendaylight.netconf.sal.connect.netconf.SchemalessNetconfDevice;
  61 import org.opendaylight.netconf.sal.connect.netconf.auth.DatastoreBackedPublicKeyAuth;
  62 import org.opendaylight.netconf.sal.connect.netconf.listener.NetconfDeviceCapabilities;
  63 import org.opendaylight.netconf.sal.connect.netconf.listener.NetconfDeviceCommunicator;
  64 import org.opendaylight.netconf.sal.connect.netconf.listener.NetconfSessionPreferences;
  65 import org.opendaylight.netconf.sal.connect.netconf.listener.UserPreferences;
  66 import org.opendaylight.netconf.sal.connect.netconf.sal.KeepaliveSalFacade;
  67 import org.opendaylight.netconf.sal.connect.netconf.sal.NetconfKeystoreAdapter;
  68 import org.opendaylight.netconf.sal.connect.netconf.schema.YangLibrarySchemaYangSourceProvider;
  69 import org.opendaylight.netconf.sal.connect.util.RemoteDeviceId;
  70 import org.opendaylight.netconf.topology.api.NetconfTopology;
  71 import org.opendaylight.netconf.topology.api.SchemaRepositoryProvider;
  72 import org.opendaylight.protocol.framework.ReconnectStrategy;
  73 import org.opendaylight.protocol.framework.ReconnectStrategyFactory;
  74 import org.opendaylight.protocol.framework.TimedReconnectStrategy;
  75 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.Host;
  76 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.IpAddress;
  77 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.NetconfNode;
  78 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.connection.parameters.Protocol.Name;
  79 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.connection.parameters.protocol.Specification;
  80 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.connection.parameters.protocol.specification.TlsCase;
  81 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.connection.status.available.capabilities.AvailableCapability.CapabilityOrigin;
  82 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.credentials.Credentials;
  83 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.credentials.credentials.KeyAuth;
  84 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.credentials.credentials.LoginPw;
  85 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.credentials.credentials.LoginPwUnencrypted;
  86 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.credentials.credentials.key.auth.KeyBased;
  87 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.credentials.credentials.login.pw.LoginPassword;
  88 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.credentials.credentials.login.pw.unencrypted.LoginPasswordUnencrypted;
  89 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NodeId;
  90 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
  91 import org.opendaylight.yangtools.yang.model.repo.api.SchemaContextFactory;
  92 import org.opendaylight.yangtools.yang.model.repo.api.SchemaRepository;
  93 import org.opendaylight.yangtools.yang.model.repo.api.SchemaSourceFilter;
  94 import org.opendaylight.yangtools.yang.model.repo.api.SourceIdentifier;
  95 import org.opendaylight.yangtools.yang.model.repo.api.YangTextSchemaSource;
  96 import org.opendaylight.yangtools.yang.model.repo.spi.PotentialSchemaSource;
  97 import org.opendaylight.yangtools.yang.model.repo.spi.SchemaSourceRegistration;
  98 import org.opendaylight.yangtools.yang.model.repo.spi.SchemaSourceRegistry;
  99 import org.opendaylight.yangtools.yang.model.repo.util.FilesystemSchemaSourceCache;
 100 import org.opendaylight.yangtools.yang.model.repo.util.InMemorySchemaSourceCache;
 101 import org.opendaylight.yangtools.yang.parser.repo.SharedSchemaRepository;
 102 import org.opendaylight.yangtools.yang.parser.rfc7950.repo.ASTSchemaSource;
 103 import org.opendaylight.yangtools.yang.parser.rfc7950.repo.TextToASTTransformer;
 104 import org.slf4j.Logger;
 105 import org.slf4j.LoggerFactory;
 106
 107 public abstract class AbstractNetconfTopology implements NetconfTopology {
 108
 109     private static final Logger LOG = LoggerFactory.getLogger(AbstractNetconfTopology.class);
 110
 111     protected static final long DEFAULT_REQUEST_TIMEOUT_MILLIS = 60000L;
 112     protected static final int DEFAULT_KEEPALIVE_DELAY = 0;
 113     protected static final boolean DEFAULT_RECONNECT_ON_CHANGED_SCHEMA = false;
 114     protected static final int DEFAULT_CONCURRENT_RPC_LIMIT = 0;
 115     private static final int DEFAULT_MAX_CONNECTION_ATTEMPTS = 0;
 116     private static final int DEFAULT_BETWEEN_ATTEMPTS_TIMEOUT_MILLIS = 2000;
 117     private static final long DEFAULT_CONNECTION_TIMEOUT_MILLIS = 20000L;
 118     private static final BigDecimal DEFAULT_SLEEP_FACTOR = new BigDecimal(1.5);
 119
 120     // constants related to Schema Cache(s)
 121     /**
 122      * Filesystem based caches are stored relative to the cache directory.
 123      */
 124     private static final String CACHE_DIRECTORY = "cache";
 125
 126     /**
 127      * The default cache directory relative to <code>CACHE_DIRECTORY</code>.
 128      */
 129     private static final String DEFAULT_CACHE_DIRECTORY = "schema";
 130
 131     /**
 132      * The qualified schema cache directory <code>cache/schema</code>.
 133      */
 134     private static final String QUALIFIED_DEFAULT_CACHE_DIRECTORY =
 135             CACHE_DIRECTORY + File.separator + DEFAULT_CACHE_DIRECTORY;
 136
 137     /**
 138      * The name for the default schema repository.
 139      */
 140     private static final String DEFAULT_SCHEMA_REPOSITORY_NAME = "sal-netconf-connector";
 141
 142     /**
 143      * The default schema repository in the case that one is not specified.
 144      */
 145     private static final SharedSchemaRepository DEFAULT_SCHEMA_REPOSITORY =
 146             new SharedSchemaRepository(DEFAULT_SCHEMA_REPOSITORY_NAME);
 147
 148     public static final InMemorySchemaSourceCache<ASTSchemaSource> DEFAULT_AST_CACHE =
 149             InMemorySchemaSourceCache.createSoftCache(DEFAULT_SCHEMA_REPOSITORY, ASTSchemaSource.class);
 150
 151     /**
 152      * The default factory for creating <code>SchemaContext</code> instances.
 153      */
 154     private static final SchemaContextFactory DEFAULT_SCHEMA_CONTEXT_FACTORY =
 155             DEFAULT_SCHEMA_REPOSITORY.createSchemaContextFactory(SchemaSourceFilter.ALWAYS_ACCEPT);
 156
 157     /**
 158      * Keeps track of initialized Schema resources.  A Map is maintained in which the key represents the name
 159      * of the schema cache directory, and the value is a corresponding <code>SchemaResourcesDTO</code>.  The
 160      * <code>SchemaResourcesDTO</code> is essentially a container that allows for the extraction of the
 161      * <code>SchemaRegistry</code> and <code>SchemaContextFactory</code> which should be used for a particular
 162      * Netconf mount.  Access to <code>SCHEMA_RESOURCES_DTO_MAP</code> should be surrounded by appropriate
 163      * synchronization locks.
 164      */
 165     private static final Map<String, NetconfDevice.SchemaResourcesDTO> SCHEMA_RESOURCES_DTO_MAP = new HashMap<>();
 166
 167     // Initializes default constant instances for the case when the default schema repository
 168     // directory cache/schema is used.
 169     static {
 170         SCHEMA_RESOURCES_DTO_MAP.put(DEFAULT_CACHE_DIRECTORY,
 171                 new NetconfDevice.SchemaResourcesDTO(DEFAULT_SCHEMA_REPOSITORY, DEFAULT_SCHEMA_REPOSITORY,
 172                         DEFAULT_SCHEMA_CONTEXT_FACTORY,
 173                         new NetconfStateSchemasResolverImpl()));
 174         DEFAULT_SCHEMA_REPOSITORY.registerSchemaSourceListener(DEFAULT_AST_CACHE);
 175         DEFAULT_SCHEMA_REPOSITORY.registerSchemaSourceListener(
 176                 TextToASTTransformer.create(DEFAULT_SCHEMA_REPOSITORY, DEFAULT_SCHEMA_REPOSITORY));
 177
 178         /*
 179          * Create the default <code>FilesystemSchemaSourceCache</code>, which stores cached files
 180          * in <code>cache/schema</code>. Try up to 3 times - we've seen intermittent failures on jenkins where
 181          * FilesystemSchemaSourceCache throws an IAE due to mkdirs failure. The theory is that there's a race
 182          * creating the dir and it already exists when mkdirs is called (mkdirs returns false in this case). In this
 183          * scenario, a retry should succeed.
 184          */
 185         int tries = 1;
 186         while (true) {
 187             try {
 188                 FilesystemSchemaSourceCache<YangTextSchemaSource> defaultCache =
 189                         new FilesystemSchemaSourceCache<>(DEFAULT_SCHEMA_REPOSITORY, YangTextSchemaSource.class,
 190                                 new File(QUALIFIED_DEFAULT_CACHE_DIRECTORY));
 191                 DEFAULT_SCHEMA_REPOSITORY.registerSchemaSourceListener(defaultCache);
 192                 break;
 193             } catch (IllegalArgumentException e) {
 194                 if (tries++ >= 3) {
 195                     LOG.error("Error creating default schema cache", e);
 196                     break;
 197                 }
 198                 Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
 199             }
 200         }
 201     }
 202
 203     protected final String topologyId;
 204     private final NetconfClientDispatcher clientDispatcher;
 205     private final EventExecutor eventExecutor;
 206     protected final ScheduledThreadPool keepaliveExecutor;
 207     protected final ThreadPool processingExecutor;
 208     protected final SharedSchemaRepository sharedSchemaRepository;
 209     protected final DataBroker dataBroker;
 210     protected final DOMMountPointService mountPointService;
 211     private final NetconfKeystoreAdapter keystoreAdapter;
 212     protected SchemaSourceRegistry schemaRegistry = DEFAULT_SCHEMA_REPOSITORY;
 213     protected SchemaRepository schemaRepository = DEFAULT_SCHEMA_REPOSITORY;
 214     protected SchemaContextFactory schemaContextFactory = DEFAULT_SCHEMA_CONTEXT_FACTORY;
 215     protected String privateKeyPath;
 216     protected String privateKeyPassphrase;
 217     protected final AAAEncryptionService encryptionService;
 218     protected final HashMap<NodeId, NetconfConnectorDTO> activeConnectors = new HashMap<>();
 219
 220     protected AbstractNetconfTopology(final String topologyId, final NetconfClientDispatcher clientDispatcher,
 221                                       final EventExecutor eventExecutor, final ScheduledThreadPool keepaliveExecutor,
 222                                       final ThreadPool processingExecutor,
 223                                       final SchemaRepositoryProvider schemaRepositoryProvider,
 224                                       final DataBroker dataBroker, final DOMMountPointService mountPointService,
 225                                       final AAAEncryptionService encryptionService) {
 226         this.topologyId = topologyId;
 227         this.clientDispatcher = clientDispatcher;
 228         this.eventExecutor = eventExecutor;
 229         this.keepaliveExecutor = keepaliveExecutor;
 230         this.processingExecutor = processingExecutor;
 231         this.sharedSchemaRepository = schemaRepositoryProvider.getSharedSchemaRepository();
 232         this.dataBroker = dataBroker;
 233         this.mountPointService = mountPointService;
 234         this.encryptionService = encryptionService;
 235
 236         this.keystoreAdapter = new NetconfKeystoreAdapter(dataBroker);
 237     }
 238
 239     public void setSchemaRegistry(final SchemaSourceRegistry schemaRegistry) {
 240         this.schemaRegistry = schemaRegistry;
 241     }
 242
 243     public void setSchemaContextFactory(final SchemaContextFactory schemaContextFactory) {
 244         this.schemaContextFactory = schemaContextFactory;
 245     }
 246
 247     @Override
 248     public ListenableFuture<NetconfDeviceCapabilities> connectNode(final NodeId nodeId, final Node configNode) {
 249         LOG.info("Connecting RemoteDevice{{}} , with config {}", nodeId, configNode);
 250         return setupConnection(nodeId, configNode);
 251     }
 252
 253     @Override
 254     public ListenableFuture<Void> disconnectNode(final NodeId nodeId) {
 255         LOG.debug("Disconnecting RemoteDevice{{}}", nodeId.getValue());
 256         if (!activeConnectors.containsKey(nodeId)) {
 257             return Futures.immediateFailedFuture(
 258                     new IllegalStateException("Unable to disconnect device that is not connected"));
 259         }
 260
 261         // retrieve connection, and disconnect it
 262         final NetconfConnectorDTO connectorDTO = activeConnectors.remove(nodeId);
 263         connectorDTO.getCommunicator().close();
 264         connectorDTO.getFacade().close();
 265         return Futures.immediateFuture(null);
 266     }
 267
 268     protected ListenableFuture<NetconfDeviceCapabilities> setupConnection(final NodeId nodeId,
 269                                                                           final Node configNode) {
 270         final NetconfNode netconfNode = configNode.augmentation(NetconfNode.class);
 271
 272         Preconditions.checkNotNull(netconfNode.getHost());
 273         Preconditions.checkNotNull(netconfNode.getPort());
 274         Preconditions.checkNotNull(netconfNode.isTcpOnly());
 275
 276         final NetconfConnectorDTO deviceCommunicatorDTO = createDeviceCommunicator(nodeId, netconfNode);
 277         final NetconfDeviceCommunicator deviceCommunicator = deviceCommunicatorDTO.getCommunicator();
 278         final NetconfClientSessionListener netconfClientSessionListener = deviceCommunicatorDTO.getSessionListener();
 279         final NetconfReconnectingClientConfiguration clientConfig =
 280                 getClientConfig(netconfClientSessionListener, netconfNode);
 281         final ListenableFuture<NetconfDeviceCapabilities> future =
 282                 deviceCommunicator.initializeRemoteConnection(clientDispatcher, clientConfig);
 283
 284         activeConnectors.put(nodeId, deviceCommunicatorDTO);
 285
 286         Futures.addCallback(future, new FutureCallback<NetconfDeviceCapabilities>() {
 287             @Override
 288             public void onSuccess(final NetconfDeviceCapabilities result) {
 289                 LOG.debug("Connector for : " + nodeId.getValue() + " started succesfully");
 290             }
 291
 292             @Override
 293             public void onFailure(final Throwable throwable) {
 294                 LOG.error("Connector for : " + nodeId.getValue() + " failed");
 295                 // remove this node from active connectors?
 296             }
 297         }, MoreExecutors.directExecutor());
 298
 299         return future;
 300     }
 301
 302     protected NetconfConnectorDTO createDeviceCommunicator(final NodeId nodeId,
 303                                                            final NetconfNode node) {
 304         //setup default values since default value is not supported in mdsal
 305         final long defaultRequestTimeoutMillis = node.getDefaultRequestTimeoutMillis() == null
 306                 ? DEFAULT_REQUEST_TIMEOUT_MILLIS : node.getDefaultRequestTimeoutMillis();
 307         final long keepaliveDelay = node.getKeepaliveDelay() == null
 308                 ? DEFAULT_KEEPALIVE_DELAY : node.getKeepaliveDelay();
 309         final boolean reconnectOnChangedSchema = node.isReconnectOnChangedSchema() == null
 310                 ? DEFAULT_RECONNECT_ON_CHANGED_SCHEMA : node.isReconnectOnChangedSchema();
 311
 312         final IpAddress ipAddress = node.getHost().getIpAddress();
 313         final InetSocketAddress address = new InetSocketAddress(ipAddress.getIpv4Address() != null
 314                 ? ipAddress.getIpv4Address().getValue() : ipAddress.getIpv6Address().getValue(),
 315                 node.getPort().getValue());
 316         final RemoteDeviceId remoteDeviceId = new RemoteDeviceId(nodeId.getValue(), address);
 317
 318         RemoteDeviceHandler<NetconfSessionPreferences> salFacade =
 319                 createSalFacade(remoteDeviceId);
 320
 321         if (keepaliveDelay > 0) {
 322             LOG.warn("Adding keepalive facade, for device {}", nodeId);
 323             salFacade = new KeepaliveSalFacade(remoteDeviceId, salFacade, keepaliveExecutor.getExecutor(),
 324                     keepaliveDelay, defaultRequestTimeoutMillis);
 325         }
 326
 327         // pre register yang library sources as fallback schemas to schema registry
 328         final List<SchemaSourceRegistration<YangTextSchemaSource>> registeredYangLibSources = Lists.newArrayList();
 329         if (node.getYangLibrary() != null) {
 330             final String yangLibURL = node.getYangLibrary().getYangLibraryUrl().getValue();
 331             final String yangLibUsername = node.getYangLibrary().getUsername();
 332             final String yangLigPassword = node.getYangLibrary().getPassword();
 333
 334             final LibraryModulesSchemas libraryModulesSchemas;
 335             if (yangLibURL != null) {
 336                 if (yangLibUsername != null && yangLigPassword != null) {
 337                     libraryModulesSchemas = LibraryModulesSchemas.create(yangLibURL, yangLibUsername, yangLigPassword);
 338                 } else {
 339                     libraryModulesSchemas = LibraryModulesSchemas.create(yangLibURL);
 340                 }
 341
 342                 for (final Map.Entry<SourceIdentifier, URL> sourceIdentifierURLEntry
 343                         : libraryModulesSchemas.getAvailableModels().entrySet()) {
 344                     registeredYangLibSources
 345                         .add(schemaRegistry.registerSchemaSource(
 346                                 new YangLibrarySchemaYangSourceProvider(remoteDeviceId,
 347                                         libraryModulesSchemas.getAvailableModels()),
 348                                 PotentialSchemaSource.create(sourceIdentifierURLEntry.getKey(),
 349                                         YangTextSchemaSource.class, PotentialSchemaSource.Costs.REMOTE_IO.getValue())));
 350                 }
 351             }
 352         }
 353
 354         final NetconfDevice.SchemaResourcesDTO schemaResourcesDTO = setupSchemaCacheDTO(nodeId, node);
 355         final RemoteDevice<NetconfSessionPreferences, NetconfMessage, NetconfDeviceCommunicator> device;
 356         if (node.isSchemaless()) {
 357             device = new SchemalessNetconfDevice(remoteDeviceId, salFacade);
 358         } else {
 359             device = new NetconfDeviceBuilder()
 360                     .setReconnectOnSchemasChange(reconnectOnChangedSchema)
 361                     .setSchemaResourcesDTO(schemaResourcesDTO)
 362                     .setGlobalProcessingExecutor(processingExecutor.getExecutor())
 363                     .setId(remoteDeviceId)
 364                     .setSalFacade(salFacade)
 365                     .build();
 366         }
 367
 368         final Optional<UserPreferences> userCapabilities = getUserCapabilities(node);
 369         final int rpcMessageLimit =
 370                 node.getConcurrentRpcLimit() == null ? DEFAULT_CONCURRENT_RPC_LIMIT : node.getConcurrentRpcLimit();
 371
 372         if (rpcMessageLimit < 1) {
 373             LOG.info("Concurrent rpc limit is smaller than 1, no limit will be enforced for device {}", remoteDeviceId);
 374         }
 375
 376         NetconfDeviceCommunicator netconfDeviceCommunicator =
 377              userCapabilities.isPresent() ? new NetconfDeviceCommunicator(remoteDeviceId, device,
 378                      userCapabilities.get(), rpcMessageLimit)
 379             : new NetconfDeviceCommunicator(remoteDeviceId, device, rpcMessageLimit);
 380
 381         if (salFacade instanceof KeepaliveSalFacade) {
 382             ((KeepaliveSalFacade)salFacade).setListener(netconfDeviceCommunicator);
 383         }
 384         return new NetconfConnectorDTO(netconfDeviceCommunicator, salFacade);
 385     }
 386
 387     protected NetconfDevice.SchemaResourcesDTO setupSchemaCacheDTO(final NodeId nodeId, final NetconfNode node) {
 388         // Setup information related to the SchemaRegistry, SchemaResourceFactory, etc.
 389         NetconfDevice.SchemaResourcesDTO schemaResourcesDTO = null;
 390         final String moduleSchemaCacheDirectory = node.getSchemaCacheDirectory();
 391         // Only checks to ensure the String is not empty or null; further checks related to directory
 392         // accessibility and file permissionsare handled during the FilesystemSchemaSourceCache initialization.
 393         if (!Strings.isNullOrEmpty(moduleSchemaCacheDirectory)) {
 394             // If a custom schema cache directory is specified, create the backing DTO; otherwise,
 395             // the SchemaRegistry and SchemaContextFactory remain the default values.
 396             if (!moduleSchemaCacheDirectory.equals(DEFAULT_CACHE_DIRECTORY)) {
 397                 // Multiple modules may be created at once;
 398                 // synchronize to avoid issues with data consistency among threads.
 399                 synchronized (SCHEMA_RESOURCES_DTO_MAP) {
 400                     // Look for the cached DTO to reuse SchemaRegistry and SchemaContextFactory variables
 401                     // if they already exist
 402                     schemaResourcesDTO = SCHEMA_RESOURCES_DTO_MAP.get(moduleSchemaCacheDirectory);
 403                     if (schemaResourcesDTO == null) {
 404                         schemaResourcesDTO = createSchemaResourcesDTO(moduleSchemaCacheDirectory);
 405                         schemaResourcesDTO.getSchemaRegistry().registerSchemaSourceListener(
 406                                 TextToASTTransformer.create((SchemaRepository) schemaResourcesDTO.getSchemaRegistry(),
 407                                         schemaResourcesDTO.getSchemaRegistry())
 408                         );
 409                         SCHEMA_RESOURCES_DTO_MAP.put(moduleSchemaCacheDirectory, schemaResourcesDTO);
 410                     }
 411                 }
 412                 LOG.info("Netconf connector for device {} will use schema cache directory {} instead of {}",
 413                         nodeId.getValue(), moduleSchemaCacheDirectory, DEFAULT_CACHE_DIRECTORY);
 414             }
 415         } else {
 416             LOG.warn("schema-cache-directory for {} is null or empty;  using the default {}",
 417                     nodeId.getValue(), QUALIFIED_DEFAULT_CACHE_DIRECTORY);
 418         }
 419
 420         if (schemaResourcesDTO == null) {
 421             schemaResourcesDTO = new NetconfDevice.SchemaResourcesDTO(schemaRegistry, schemaRepository,
 422                     schemaContextFactory, new NetconfStateSchemasResolverImpl());
 423         }
 424
 425         return schemaResourcesDTO;
 426     }
 427
 428     /**
 429      * Creates the backing Schema classes for a particular directory.
 430      *
 431      * @param moduleSchemaCacheDirectory The string directory relative to "cache"
 432      * @return A DTO containing the Schema classes for the Netconf mount.
 433      */
 434     private NetconfDevice.SchemaResourcesDTO createSchemaResourcesDTO(final String moduleSchemaCacheDirectory) {
 435         final SharedSchemaRepository repository = new SharedSchemaRepository(moduleSchemaCacheDirectory);
 436         final SchemaContextFactory contextFactory
 437                 = repository.createSchemaContextFactory(SchemaSourceFilter.ALWAYS_ACCEPT);
 438         setSchemaRegistry(repository);
 439         setSchemaContextFactory(contextFactory);
 440         final FilesystemSchemaSourceCache<YangTextSchemaSource> deviceCache =
 441                 createDeviceFilesystemCache(moduleSchemaCacheDirectory);
 442         repository.registerSchemaSourceListener(deviceCache);
 443         repository.registerSchemaSourceListener(
 444             InMemorySchemaSourceCache.createSoftCache(repository, ASTSchemaSource.class));
 445         return new NetconfDevice.SchemaResourcesDTO(repository, repository, contextFactory,
 446                 new NetconfStateSchemasResolverImpl());
 447     }
 448
 449     /**
 450      * Creates a <code>FilesystemSchemaSourceCache</code> for the custom schema cache directory.
 451      *
 452      * @param schemaCacheDirectory The custom cache directory relative to "cache"
 453      * @return A <code>FilesystemSchemaSourceCache</code> for the custom schema cache directory
 454      */
 455     private FilesystemSchemaSourceCache<YangTextSchemaSource> createDeviceFilesystemCache(
 456             final String schemaCacheDirectory) {
 457         final String relativeSchemaCacheDirectory = CACHE_DIRECTORY + File.separator + schemaCacheDirectory;
 458         return new FilesystemSchemaSourceCache<>(schemaRegistry, YangTextSchemaSource.class,
 459                 new File(relativeSchemaCacheDirectory));
 460     }
 461
 462     /**
 463      * Sets the private key path from location specified in configuration file using blueprint.
 464      */
 465     public void setPrivateKeyPath(final String privateKeyPath) {
 466         this.privateKeyPath = privateKeyPath;
 467     }
 468
 469     /**
 470      * Sets the private key passphrase from location specified in configuration file using blueprint.
 471      */
 472     public void setPrivateKeyPassphrase(final String privateKeyPassphrase) {
 473         this.privateKeyPassphrase = privateKeyPassphrase;
 474     }
 475
 476     public NetconfReconnectingClientConfiguration getClientConfig(final NetconfClientSessionListener listener,
 477                                                                   final NetconfNode node) {
 478
 479         //setup default values since default value is not supported in mdsal
 480         final long clientConnectionTimeoutMillis = node.getConnectionTimeoutMillis() == null
 481                 ? DEFAULT_CONNECTION_TIMEOUT_MILLIS : node.getConnectionTimeoutMillis();
 482         final long maxConnectionAttempts = node.getMaxConnectionAttempts() == null
 483                 ? DEFAULT_MAX_CONNECTION_ATTEMPTS : node.getMaxConnectionAttempts();
 484         final int betweenAttemptsTimeoutMillis = node.getBetweenAttemptsTimeoutMillis() == null
 485                 ? DEFAULT_BETWEEN_ATTEMPTS_TIMEOUT_MILLIS : node.getBetweenAttemptsTimeoutMillis();
 486         final BigDecimal sleepFactor = node.getSleepFactor() == null ? DEFAULT_SLEEP_FACTOR : node.getSleepFactor();
 487
 488         final InetSocketAddress socketAddress = getSocketAddress(node.getHost(), node.getPort().getValue());
 489
 490         final ReconnectStrategyFactory sf = new TimedReconnectStrategyFactory(eventExecutor,
 491                 maxConnectionAttempts, betweenAttemptsTimeoutMillis, sleepFactor);
 492         final ReconnectStrategy strategy = sf.createReconnectStrategy();
 493
 494         final NetconfReconnectingClientConfigurationBuilder reconnectingClientConfigurationBuilder =
 495                 NetconfReconnectingClientConfigurationBuilder.create();
 496
 497         if (node.isTcpOnly() || node.getProtocol() == null || node.getProtocol().getName() == Name.SSH) {
 498             final AuthenticationHandler authHandler = getHandlerFromCredentials(node.getCredentials());
 499             reconnectingClientConfigurationBuilder
 500                 .withAuthHandler(authHandler)
 501                 .withProtocol(node.isTcpOnly() ? NetconfClientConfiguration.NetconfClientProtocol.TCP :
 502                     NetconfClientConfiguration.NetconfClientProtocol.SSH);
 503         } else if (node.getProtocol().getName() == Name.TLS) {
 504             final SslHandlerFactory sslHandlerFactory = new SslHandlerFactoryImpl(keystoreAdapter,
 505                     node.getProtocol().getSpecification());
 506             reconnectingClientConfigurationBuilder
 507                 .withSslHandlerFactory(sslHandlerFactory)
 508                 .withProtocol(NetconfClientConfiguration.NetconfClientProtocol.TLS);
 509         } else {
 510             throw new IllegalStateException("Unsupported protocol type: " + node.getProtocol().getName().getClass());
 511         }
 512
 513         return reconnectingClientConfigurationBuilder
 514                 .withAddress(socketAddress)
 515                 .withConnectionTimeoutMillis(clientConnectionTimeoutMillis)
 516                 .withReconnectStrategy(strategy)
 517                 .withConnectStrategyFactory(sf)
 518                 .withSessionListener(listener)
 519                 .build();
 520     }
 521
 522     private AuthenticationHandler getHandlerFromCredentials(final Credentials credentials) {
 523         if (credentials instanceof org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology
 524                 .rev150114.netconf.node.credentials.credentials.LoginPassword) {
 525             final org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology
 526                     .rev150114.netconf.node.credentials.credentials.LoginPassword loginPassword
 527                     = (org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology
 528                     .rev150114.netconf.node.credentials.credentials.LoginPassword) credentials;
 529             return new LoginPasswordHandler(loginPassword.getUsername(), loginPassword.getPassword());
 530         }
 531         if (credentials instanceof LoginPwUnencrypted) {
 532             final LoginPasswordUnencrypted loginPassword =
 533                     ((LoginPwUnencrypted) credentials).getLoginPasswordUnencrypted();
 534             return new LoginPasswordHandler(loginPassword.getUsername(), loginPassword.getPassword());
 535         }
 536         if (credentials instanceof LoginPw) {
 537             final LoginPassword loginPassword = ((LoginPw) credentials).getLoginPassword();
 538             return new LoginPasswordHandler(loginPassword.getUsername(),
 539                     encryptionService.decrypt(loginPassword.getPassword()));
 540         }
 541         if (credentials instanceof KeyAuth) {
 542             final KeyBased keyPair = ((KeyAuth) credentials).getKeyBased();
 543             return new DatastoreBackedPublicKeyAuth(keyPair.getUsername(), keyPair.getKeyId(),
 544                     keystoreAdapter, encryptionService);
 545         }
 546         throw new IllegalStateException("Unsupported credential type: " + credentials.getClass());
 547     }
 548
 549     protected abstract RemoteDeviceHandler<NetconfSessionPreferences> createSalFacade(RemoteDeviceId id);
 550
 551     private static InetSocketAddress getSocketAddress(final Host host, final int port) {
 552         if (host.getDomainName() != null) {
 553             return new InetSocketAddress(host.getDomainName().getValue(), port);
 554         }
 555
 556         final IpAddress ipAddress = host.getIpAddress();
 557         final String ip = ipAddress.getIpv4Address() != null ? ipAddress.getIpv4Address().getValue()
 558                 : ipAddress.getIpv6Address().getValue();
 559         return new InetSocketAddress(ip, port);
 560     }
 561
 562     private static Optional<UserPreferences> getUserCapabilities(final NetconfNode node) {
 563         // if none of yang-module-capabilities or non-module-capabilities is specified
 564         // just return absent
 565         if (node.getYangModuleCapabilities() == null && node.getNonModuleCapabilities() == null) {
 566             return Optional.absent();
 567         }
 568
 569         final List<String> capabilities = new ArrayList<>();
 570
 571         boolean overrideYangModuleCaps = false;
 572         if (node.getYangModuleCapabilities() != null) {
 573             capabilities.addAll(node.getYangModuleCapabilities().getCapability());
 574             overrideYangModuleCaps = node.getYangModuleCapabilities().isOverride();
 575         }
 576
 577         //non-module capabilities should not exist in yang module capabilities
 578         final NetconfSessionPreferences netconfSessionPreferences = NetconfSessionPreferences.fromStrings(capabilities);
 579         Preconditions.checkState(netconfSessionPreferences.getNonModuleCaps().isEmpty(),
 580                 "List yang-module-capabilities/capability should contain only module based capabilities. "
 581                         + "Non-module capabilities used: " + netconfSessionPreferences.getNonModuleCaps());
 582
 583         boolean overrideNonModuleCaps = false;
 584         if (node.getNonModuleCapabilities() != null) {
 585             capabilities.addAll(node.getNonModuleCapabilities().getCapability());
 586             overrideNonModuleCaps = node.getNonModuleCapabilities().isOverride();
 587         }
 588
 589         return Optional.of(new UserPreferences(NetconfSessionPreferences
 590             .fromStrings(capabilities, CapabilityOrigin.UserDefined), overrideYangModuleCaps, overrideNonModuleCaps));
 591     }
 592
 593     private static final class TimedReconnectStrategyFactory implements ReconnectStrategyFactory {
 594         private final Long connectionAttempts;
 595         private final EventExecutor executor;
 596         private final double sleepFactor;
 597         private final int minSleep;
 598
 599         TimedReconnectStrategyFactory(final EventExecutor executor, final Long maxConnectionAttempts,
 600                                       final int minSleep, final BigDecimal sleepFactor) {
 601             if (maxConnectionAttempts != null && maxConnectionAttempts > 0) {
 602                 connectionAttempts = maxConnectionAttempts;
 603             } else {
 604                 connectionAttempts = null;
 605             }
 606
 607             this.sleepFactor = sleepFactor.doubleValue();
 608             this.executor = executor;
 609             this.minSleep = minSleep;
 610         }
 611
 612         @Override
 613         public ReconnectStrategy createReconnectStrategy() {
 614             return new TimedReconnectStrategy(executor, minSleep,
 615                     minSleep, sleepFactor, null /*maxSleep*/, connectionAttempts, null /*deadline*/);
 616         }
 617     }
 618
 619     protected static class NetconfConnectorDTO implements AutoCloseable {
 620
 621         private final NetconfDeviceCommunicator communicator;
 622         private final RemoteDeviceHandler<NetconfSessionPreferences> facade;
 623
 624         public NetconfConnectorDTO(final NetconfDeviceCommunicator communicator,
 625                                    final RemoteDeviceHandler<NetconfSessionPreferences> facade) {
 626             this.communicator = communicator;
 627             this.facade = facade;
 628         }
 629
 630         public NetconfDeviceCommunicator getCommunicator() {
 631             return communicator;
 632         }
 633
 634         public RemoteDeviceHandler<NetconfSessionPreferences> getFacade() {
 635             return facade;
 636         }
 637
 638         public NetconfClientSessionListener getSessionListener() {
 639             return communicator;
 640         }
 641
 642         @Override
 643         public void close() {
 644             communicator.close();
 645             facade.close();
 646         }
 647     }
 648
 649     private static final class SslHandlerFactoryImpl implements SslHandlerFactory {
 650         private final NetconfKeystoreAdapter keystoreAdapter;
 651         private final Optional<Specification> specOptional;
 652
 653         SslHandlerFactoryImpl(final NetconfKeystoreAdapter keystoreAdapter, final Specification specification) {
 654             this.keystoreAdapter = keystoreAdapter;
 655             this.specOptional = Optional.fromNullable(specification);
 656         }
 657
 658         @Override
 659         public SslHandler createSslHandler() {
 660             try {
 661                 final KeyStore keyStore = keystoreAdapter.getJavaKeyStore();
 662
 663                 final KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
 664                 kmf.init(keyStore, "".toCharArray());
 665
 666                 final TrustManagerFactory tmf =
 667                         TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
 668                 tmf.init(keyStore);
 669
 670                 final SSLContext sslCtx = SSLContext.getInstance("TLS");
 671                 sslCtx.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null);
 672                 final SSLEngine engine = sslCtx.createSSLEngine();
 673                 engine.setUseClientMode(true);
 674
 675                 final Set<String> protocols = Sets.newHashSet(engine.getSupportedProtocols());
 676                 if (specOptional.isPresent()) {
 677                     final Specification specification = specOptional.get();
 678                     if (!(specification instanceof TlsCase)) {
 679                         throw new IllegalArgumentException("Cannot get TLS specification from: " + specification);
 680                     }
 681                     protocols.removeAll(((TlsCase)specification).getTls().getExcludedVersions());
 682                 }
 683
 684                 engine.setEnabledProtocols(protocols.toArray(new String[0]));
 685                 engine.setEnabledCipherSuites(engine.getSupportedCipherSuites());
 686                 engine.setEnableSessionCreation(true);
 687
 688                 return new SslHandler(engine);
 689             } catch (GeneralSecurityException | IOException exc) {
 690                 throw new IllegalStateException(exc);
 691             }
 692         }
 693     }
 694 }