Changes for replication security issues 511, 512, 608.

coulbeck

28.54.2007 f7036e50348484f4daf39f9e8457de602ab83939

Changes for replication security issues 511, 512, 608.

There is a new backend representing the certificate trust store, which allows the setup code to query and populate the trust store over protocol. However, we are using blind trust until that piece of the setup code is ready.

The encryption settings are currently global to the server instance in a new crypto manager config entry. Authentication will always be performed so the main setting is whether to use encryption or not. In the future (post 1.0) we will need to allow encryption to be configured on or off depending on which replication server we are connecting to but we need some discussion on the best way to specify that in configuration.

 opends/resource/config/config.ldif

@@ -76,6 +76,13 @@
ds-cfg-acl-handler-class: org.opends.server.authorization.dseecompat.AciHandler
ds-cfg-acl-handler-enabled: true

dn: cn=Crypto Manager,cn=config
objectClass: top
objectClass: ds-cfg-crypto-manager
cn: Crypto Manager
ds-cfg-ssl-cert-nickname: ads-certificate
ds-cfg-ssl-encryption: false

dn: cn=Account Status Notification Handlers,cn=config
objectClass: top
objectClass: ds-cfg-branch
@@ -304,6 +311,19 @@
ds-cfg-backend-writability-mode: enabled
ds-cfg-backend-base-dn: cn=config

dn: ds-cfg-backend-id=ads-truststore,cn=Backends,cn=config
objectClass: top
objectClass: ds-cfg-backend
objectClass: ds-cfg-trust-store-backend
ds-cfg-backend-id: ads-truststore
ds-cfg-backend-enabled: true
ds-cfg-backend-class: org.opends.server.backends.TrustStoreBackend
ds-cfg-backend-writability-mode: enabled
ds-cfg-backend-base-dn: cn=trust-store
ds-cfg-trust-store-type: JKS
ds-cfg-trust-store-file: config/ads-truststore
ds-cfg-trust-store-pin-file: config/ads-truststore.pin

dn: ds-cfg-backend-id=monitor,cn=Backends,cn=config
objectClass: top
objectClass: ds-cfg-backend

 opends/resource/schema/02-config.ldif

@@ -1576,6 +1576,31 @@
  NAME 'ds-cfg-backend-compact-encoding'
  SYNTAX 1.3.6.1.4.1.1466.115.121.1.7 SINGLE-VALUE
  X-ORIGIN 'OpenDS Directory Server' )
<<<<<<< .mine
attributeTypes: ( ds-cfg-ssl-encryption-oid NAME 'ds-cfg-ssl-encryption'
  SYNTAX 1.3.6.1.4.1.1466.115.121.1.7 SINGLE-VALUE
  X-ORIGIN 'OpenDS Directory Server' )
attributeTypes: ( ds-cfg-public-key-certificate-OID
  NAME 'ds-cfg-public-key-certificate'
  DESC 'cryptographic public-key certificate'
  SYNTAX 1.3.6.1.4.1.1466.115.121.1.8
  SINGLE-VALUE
  X-ORIGIN 'OpenDS Directory Server' )
attributeTypes: ( ds-cfg-key-id-OID
  NAME 'ds-cfg-key-id'
  DESC 'cryptographic cipher-key unique identifier'
  SYNTAX 1.3.6.1.4.1.1466.115.121.1.15
  SINGLE-VALUE
  X-ORIGIN 'OpenDS Directory Server' )
attributeTypes: ( ds-cfg-key-compromised-time-OID
  NAME 'ds-cfg-key-compromised-time'
  DESC 'The time a cryptographic cipher key was suspected to be compromised'
  EQUALITY generalizedTimeMatch
  ORDERING generalizedTimeOrderingMatch
  SYNTAX 1.3.6.1.4.1.1466.115.121.1.24
  SINGLE-VALUE
  X-ORIGIN 'OpenDS Directory Server' )
=======
attributeTypes: ( 1.3.6.1.4.1.26027.1.1.476
  NAME 'ds-cfg-email-address-attribute-type'
  SYNTAX 1.3.6.1.4.1.1466.115.121.1.15 X-ORIGIN 'OpenDS Directory Server' )
@@ -1586,6 +1611,7 @@
attributeTypes: ( 1.3.6.1.4.1.26027.1.1.478
  NAME 'ds-cfg-message-template-file' SYNTAX 1.3.6.1.4.1.1466.115.121.1.15
  X-ORIGIN 'OpenDS Directory Server' )
>>>>>>> .r2828
objectClasses: ( 1.3.6.1.4.1.26027.1.2.1
  NAME 'ds-cfg-access-control-handler' SUP top STRUCTURAL
  MUST ( cn $ ds-cfg-acl-handler-class $ ds-cfg-acl-handler-enabled )
@@ -2235,4 +2261,31 @@
  ds-cfg-message-template-file ) MAY ( ds-cfg-email-address-attribute-type $
  ds-cfg-recipient-address $ ds-cfg-message-subject )
  X-ORIGIN 'OpenDS Directory Server' )
objectClasses: ( ds-cfg-trust-store-backend-oid
  NAME 'ds-cfg-trust-store-backend'
  SUP ds-cfg-backend STRUCTURAL
  MAY ( ds-cfg-trust-store-type $ ds-cfg-trust-store-file $
  ds-cfg-trust-store-pin $ ds-cfg-trust-store-pin-property $
  ds-cfg-trust-store-pin-environment-variable $ ds-cfg-trust-store-pin-file )
  X-ORIGIN 'OpenDS Directory Server' )
objectClasses: ( ds-cfg-crypto-manager-oid
  NAME 'ds-cfg-crypto-manager'
  SUP top
  STRUCTURAL
  MAY ( cn $ ds-cfg-ssl-protocol $ ds-cfg-ssl-cipher-suite $
  ds-cfg-ssl-cert-nickname $ ds-cfg-ssl-encryption )
  X-ORIGIN 'OpenDS Directory Server' )
objectClasses: ( ds-cfg-instance-key-OID
  NAME 'ds-cfg-instance-key'
  SUP top
  STRUCTURAL
  MUST ( ds-cfg-key-id $ ds-cfg-public-key-certificate )
  MAY ds-cfg-key-compromised-time
  X-ORIGIN 'OpenDS Directory Server' )
objectClasses: ( ds-cfg-self-signed-cert-request-OID
  NAME 'ds-cfg-self-signed-cert-request'
  SUP top
  STRUCTURAL
  MUST ( ds-cfg-key-id )
  X-ORIGIN 'OpenDS Directory Server' )


 opends/src/admin/defn/org/opends/server/admin/std/MultimasterDomainConfiguration.xml

@@ -201,7 +201,7 @@
  <adm:property name="window-size" mandatory="false"
    multi-valued="false">
    <adm:synopsis>
      Specifies the window size that will this Domain must use when
      Specifies the window size that this domain will use when
      communicating with replication servers.
    </adm:synopsis>
    <adm:default-behavior>

 opends/src/admin/defn/org/opends/server/admin/std/RootConfiguration.xml

@@ -81,6 +81,19 @@
      </cli:relation>
    </adm:profile>
  </adm:relation>
  <adm:relation name="crypto-manager">
    <adm:one-to-one />
    <adm:profile name="ldap">
      <ldap:rdn-sequence>
        cn=Crypto Manager,cn=config
      </ldap:rdn-sequence>
    </adm:profile>
    <adm:profile name="cli">
      <cli:relation>
        <cli:default-property name="enabled" />
      </cli:relation>
    </adm:profile>
  </adm:relation>
  <adm:relation name="group-implementation">
    <adm:one-to-many />
    <adm:profile name="ldap">

 opends/src/ads/org/opends/admin/ads/ServerDescriptor.java

@@ -871,7 +871,8 @@
    "schema".equalsIgnoreCase(id) ||
    "config".equalsIgnoreCase(id) ||
    "monitor".equalsIgnoreCase(id) ||
    "backup".equalsIgnoreCase(id);
    "backup".equalsIgnoreCase(id) ||
    "ads-truststore".equalsIgnoreCase(id);
  }

  /**

 opends/src/messages/messages/backend.properties

@@ -870,3 +870,76 @@
INFO_TASK_COMPLETION_BODY_299=Task ID:  %s\r\nTask State:  %s\r\nScheduled \
 Start Time:  %s\r\nActual Start Time:  %s\r\nCompletion Time:  %s\r\n\r\nLog \
 Messages:\r\n
SEVERE_ERR_TRUSTSTORE_GET_ENTRY_NULL_300=Unable to retrieve an entry from the \
 trust store backend because the requested entry was null
SEVERE_ERR_TRUSTSTORE_INVALID_BASE_301=Requested entry %s does not exist in \
 the trust store backend
SEVERE_ERR_TRUSTSTORE_DN_DOES_NOT_SPECIFY_CERTIFICATE_302=Unable to process \
 entry %s in the trust store backend because the requested DN is one level \
 below the base DN but does not specify a certificate name
SEVERE_ERR_TRUSTSTORE_INVALID_CERTIFICATE_303=Unable to retrieve entry %s \
 from the trust store backend because the requested certificate is invalid: \
 %s
SEVERE_ERR_TRUSTSTORE_MODIFY_NOT_SUPPORTED_304=Modify operations are not \
 supported in the trust store backend
SEVERE_ERR_TRUSTSTORE_MODIFY_DN_NOT_SUPPORTED_305=Modify DN operations are not \
 supported in the trust store backend
SEVERE_ERR_TRUSTSTORE_CANNOT_DECODE_TRUSTSTORE_ROOT_DN_306=Unable to \
 initialize the trust store backend because an error occurred while \
 attempting to decode the base DN for the backend:  %s
SEVERE_ERR_TRUSTSTORE_IMPORT_AND_EXPORT_NOT_SUPPORTED_307=LDIF import and \
 export operations are not supported in the trust store backend
SEVERE_ERR_TRUSTSTORE_BACKUP_AND_RESTORE_NOT_SUPPORTED_308=Backup and \
 restore operations are not supported in the trust store backend
SEVERE_ERR_TRUSTSTORE_NO_SUCH_FILE_309=The trust store file %s \
 specified in attribute ds-cfg-trust-store-file of configuration entry %s does \
 not exist
SEVERE_ERR_TRUSTSTORE_INVALID_TYPE_310=The trust store type %s \
 specified in attribute ds-cfg-trust-store-type of configuration entry %s is \
 not valid:  %s
SEVERE_ERR_TRUSTSTORE_PIN_NO_SUCH_FILE_311=File %s specified in \
 attribute ds-cfg-trust-store-pin-file of configuration entry %s should \
 contain the PIN needed to access the trust store, but this file \
 does not exist
SEVERE_ERR_TRUSTSTORE_PIN_FILE_CANNOT_READ_312=An error occurred while \
 trying to read the trust store PIN from file %s specified in configuration \
 attribute ds-cfg-trust-store-pin-file of configuration entry %s:  %s
SEVERE_ERR_TRUSTSTORE_PIN_FILE_EMPTY_313=File %s specified in \
 attribute ds-cfg-trust-store-pin-file of configuration entry %s should \
 contain the PIN needed to access the trust store, but this file \
 is empty
SEVERE_ERR_TRUSTSTORE_PIN_ENVAR_NOT_SET_314=Environment variable %s \
 which is specified in attribute ds-cfg-trust-store-pin-environment-variable \
 of configuration entry %s should contain the PIN needed to access the \
 trust store, but this property is not set
SEVERE_ERR_TRUSTSTORE_PIN_PROPERTY_NOT_SET_315=Java property %s which \
 is specified in attribute ds-cfg-trust-store-pin-property of configuration \
 entry %s should contain the PIN needed to access the file-based trust \
 manager, but this property is not set
SEVERE_ERR_TRUSTSTORE_CANNOT_DETERMINE_FILE_316=An unexpected error \
 occurred while trying to determine the value of configuration attribute \
 ds-cfg-trust-store-file in configuration entry %s:  %s
SEVERE_ERR_TRUSTSTORE_CANNOT_LOAD_317=An error occurred while trying \
 to load the trust store contents from file %s:  %s
SEVERE_ERR_TRUSTSTORE_CANNOT_CREATE_FACTORY_318=An error occurred \
 while trying to create a trust manager factory to access the contents of \
 trust store file %s:  %s
SEVERE_ERR_TRUSTSTORE_ALIAS_IN_USE_319=The certificate entry %s already exists
SEVERE_ERR_TRUSTSTORE_CANNOT_GENERATE_CERT_320=Error while attempting to \
 generate a self-signed certificate %s in the trust store file %s: %s
SEVERE_ERR_TRUSTSTORE_CANNOT_ADD_CERT_321=Error while trying to add \
 certificate %s to the trust store file %s: %s
SEVERE_ERR_TRUSTSTORE_DELETE_NOT_SUPPORTED_322=Delete operations are not \
 supported in the trust store backend
SEVERE_ERR_TRUSTSTORE_ENTRY_MISSING_CERT_ATTR_323=The entry %s could not be \
 added because it does not contain a certificate attribute %s
SEVERE_ERR_TRUSTSTORE_ENTRY_HAS_MULTIPLE_CERT_ATTRS_324=The entry %s could \
 not be added because it contains multiple certificate attributes %s
SEVERE_ERR_TRUSTSTORE_ENTRY_MISSING_CERT_VALUE_325=The entry %s could not be \
 added because it does not contain a value of certificate attribute %s
SEVERE_ERR_TRUSTSTORE_ENTRY_HAS_MULTIPLE_CERT_VALUES_326=The entry %s could \
 not be added because it contains multiple values of certificate attribute %s
SEVERE_ERR_TRUSTSTORE_CANNOT_WRITE_CERT_327=Error while writing certificate %s \
 to a file: %s
SEVERE_WARN_TRUSTSTORE_SET_PERMISSIONS_FAILED_328=Failed to set permissions \
 on trust store file %s

 opends/src/messages/messages/core.properties

@@ -1645,3 +1645,10 @@
FATAL_ERR_CANNOT_SET_ENVIRONMENT_CONFIG_WHILE_RUNNING_648=The Directory \
 Server is currently running.  The environment configuration may not be \
 altered while the server is online
SEVERE_ERR_CRYPTOMGR_SSL_CONTEXT_CANNOT_INITIALIZE_649=An error occurred \
 while attempting to initialize a SSL context for server to server \
 communication:  %s
SEVERE_ERR_CRYPTOMGR_ADS_TRUST_STORE_BACKEND_NOT_ENABLED_650=The ADS trust \
 store backend %s is not enabled
SEVERE_ERR_CRYPTOMGR_ADS_TRUST_STORE_BACKEND_WRONG_CLASS_651=The backend %s \
 is not a trust store backend

 opends/src/messages/messages/tools.properties

@@ -2296,3 +2296,5 @@
INFO_LABEL_DBTEST_VLV_INDEX_LAST_SORT_KEYS_1371=Last Sort Keys
SEVERE_ERR_DBTEST_CANNOT_DECODE_SIZE_1372=An error occured while parsing the \
  min/max data size %s as a integer: %s
SEVERE_ERR_CONFIGDS_CANNOT_ENABLE_ADS_TRUST_STORE_1373=An error occurred while \
 attempting to enable the ADS trust store: %s

 opends/src/quicksetup/org/opends/quicksetup/installer/Installer.java

@@ -3577,7 +3577,7 @@
    switch (type)
    {
    case 0:
      // Will return a figure
      // Will return a digit
      d = next % 10;
      if (d < 0)
      {

 opends/src/server/org/opends/server/api/KeyManagerProvider.java

@@ -99,7 +99,7 @@
   *          for this key manager provider, or {@code false} if not.
   */
  public boolean isConfigurationAcceptable(
                      KeyManagerCfg configuration,
                      T configuration,
                      List<Message> unacceptableReasons)
  {
    // This default implementation does not perform any special

 opends/src/server/org/opends/server/config/ConfigConstants.java

@@ -2842,7 +2842,7 @@


  /**
   * The DN of the entry that be the base of the configuration information for
   * The DN of the entry that is the base of the configuration information for
   * the Directory Server trust manager providers.
   */
  public static final String DN_TRUSTMANAGER_PROVIDER_CONFIG_BASE =
@@ -2851,6 +2851,70 @@


  /**
   * The ADS trust store backend id.
   */
  public static final String ID_ADS_TRUST_STORE_BACKEND = "ads-truststore";



  /**
   * The DN of the trust store backend configuration entry.
   */
  public static final String DN_TRUST_STORE_BACKEND =
       ATTR_BACKEND_ID + "=" + ID_ADS_TRUST_STORE_BACKEND +
            "," + DN_BACKEND_BASE;



  /**
   * Alias of the local instance certificate in the ADS keystore.
   */
  public static final String ADS_CERTIFICATE_ALIAS = "ads-certificate";



  /**
   * The DN of the entry that will serve as the base for local ADS trust store
   * information.
   */
  public static final String DN_TRUST_STORE_ROOT = "cn=ads-truststore";



  /**
   * The name of the attribute that holds a server certificate alias.
   */
  public static final String ATTR_CERT_ALIAS = NAME_PREFIX_CFG + "key-id";



  /**
   * The name of the objectclass that will be used for a server
   * certificate entry.
   */
  public static final String OC_INSTANCE_KEY =
       NAME_PREFIX_CFG + "instance-key";



  /**
   * The name of the objectclass that will be used for a self-signed
   * certificate request.
   */
  public static final String OC_SELF_SIGNED_CERT_REQUEST =
       NAME_PREFIX_CFG + "self-signed-cert-request";



  /**
   * The name of the attribute that is used to specify a server
   * instance key.
   */
  public static final String ATTR_ADS_CERTIFICATE =
       NAME_PREFIX_CFG + "public-key-certificate";


  /**
   * The DN of the entry that will serve as the base for all Directory Server
   * loggers.
   */

 opends/src/server/org/opends/server/core/DirectoryServer.java

@@ -30,15 +30,7 @@

import org.opends.server.admin.ClassLoaderProvider;
import org.opends.server.admin.server.ServerManagementContext;
import org.opends.server.admin.std.server.AlertHandlerCfg;
import org.opends.server.admin.std.server.AttributeSyntaxCfg;
import org.opends.server.admin.std.server.ConnectionHandlerCfg;
import org.opends.server.admin.std.server.DirectoryStringAttributeSyntaxCfg;
import org.opends.server.admin.std.server.MonitorProviderCfg;
import org.opends.server.admin.std.server.PasswordValidatorCfg;
import org.opends.server.admin.std.server.RootCfg;
import org.opends.server.admin.std.server.RootDSEBackendCfg;
import org.opends.server.admin.std.server.SynchronizationProviderCfg;
import org.opends.server.admin.std.server.*;
import org.opends.server.api.AccountStatusNotificationHandler;
import org.opends.server.api.AlertGenerator;
import org.opends.server.api.AlertHandler;
@@ -2189,7 +2181,10 @@
  public void initializeCryptoManager()
         throws ConfigException, InitializationException
  {
    cryptoManager = new CryptoManager();
    RootCfg root =
         ServerManagementContext.getInstance().getRootConfiguration();
    CryptoManagerCfg cryptoManagerCfg = root.getCryptoManager();
    cryptoManager = new CryptoManager(cryptoManagerCfg);
  }



 opends/src/server/org/opends/server/extensions/FileBasedKeyManagerProvider.java

@@ -43,7 +43,6 @@

import org.opends.server.admin.server.ConfigurationChangeListener;
import org.opends.server.admin.std.server.FileBasedKeyManagerCfg;
import org.opends.server.admin.std.server.KeyManagerCfg;
import org.opends.server.api.KeyManagerProvider;
import org.opends.server.config.ConfigException;
import org.opends.server.core.DirectoryServer;
@@ -315,11 +314,10 @@
   * {@inheritDoc}
   */
  @Override()
  public boolean isConfigurationAcceptable(KeyManagerCfg configuration,
  public boolean isConfigurationAcceptable(FileBasedKeyManagerCfg configuration,
                                           List<Message> unacceptableReasons)
  {
    FileBasedKeyManagerCfg config = (FileBasedKeyManagerCfg) configuration;
    return isConfigurationChangeAcceptable(config, unacceptableReasons);
    return isConfigurationChangeAcceptable(configuration, unacceptableReasons);
  }



 opends/src/server/org/opends/server/extensions/PKCS11KeyManagerProvider.java

@@ -40,7 +40,6 @@
import javax.net.ssl.KeyManagerFactory;

import org.opends.server.admin.server.ConfigurationChangeListener;
import org.opends.server.admin.std.server.KeyManagerCfg;
import org.opends.server.admin.std.server.PKCS11KeyManagerCfg;
import org.opends.server.api.KeyManagerProvider;
import org.opends.server.config.ConfigException;
@@ -273,11 +272,10 @@
   * {@inheritDoc}
   */
  @Override()
  public boolean isConfigurationAcceptable(KeyManagerCfg configuration,
  public boolean isConfigurationAcceptable(PKCS11KeyManagerCfg configuration,
                                           List<Message> unacceptableReasons)
  {
    PKCS11KeyManagerCfg config = (PKCS11KeyManagerCfg) configuration;
    return isConfigurationChangeAcceptable(config, unacceptableReasons);
    return isConfigurationChangeAcceptable(configuration, unacceptableReasons);
  }



 opends/src/server/org/opends/server/replication/plugin/ReplicationBroker.java

@@ -54,15 +54,7 @@
import org.opends.server.protocols.ldap.LDAPFilter;
import org.opends.server.replication.common.ChangeNumber;
import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.protocol.ProtocolSession;
import org.opends.server.replication.protocol.ProtocolVersion;
import org.opends.server.replication.protocol.ReplServerStartMessage;
import org.opends.server.replication.protocol.ReplicationMessage;
import org.opends.server.replication.protocol.ServerStartMessage;
import org.opends.server.replication.protocol.SocketSession;
import org.opends.server.replication.protocol.UpdateMessage;
import org.opends.server.replication.protocol.WindowMessage;
import org.opends.server.replication.protocol.WindowProbe;
import org.opends.server.replication.protocol.*;
import org.opends.server.types.DN;
import org.opends.server.types.DereferencePolicy;
import org.opends.server.types.ResultCode;
@@ -101,6 +93,7 @@
  private int maxRcvWindow;
  private int timeout = 0;
  private short protocolVersion;
  private ReplSessionSecurity replSessionSecurity;

  /**
   * The time in milliseconds between heartbeats from the replication
@@ -150,10 +143,12 @@
   * @param window The size of the send and receive window to use.
   * @param heartbeatInterval The interval between heartbeats requested of the
   * replicationServer, or zero if no heartbeats are requested.
   * @param replSessionSecurity The session security configuration.
   */
  public ReplicationBroker(ServerState state, DN baseDn, short serverID,
      int maxReceiveQueue, int maxReceiveDelay, int maxSendQueue,
      int maxSendDelay, int window, long heartbeatInterval)
      int maxSendDelay, int window, long heartbeatInterval,
      ReplSessionSecurity replSessionSecurity)
  {
    this.baseDn = baseDn;
    this.serverID = serverID;
@@ -169,6 +164,7 @@
    this.halfRcvWindow = window/2;
    this.heartbeatInterval = heartbeatInterval;
    this.protocolVersion = ProtocolVersion.currentVersion();
    this.replSessionSecurity = replSessionSecurity;
  }

  /**
@@ -199,7 +195,6 @@
   * Connect to a ReplicationServer.
   *
   * @throws NumberFormatException address was invalid
   * @throws IOException error during connection phase
   */
  private void connect()
  {
@@ -236,15 +231,16 @@
            socket.setReceiveBufferSize(1000000);
            socket.setTcpNoDelay(true);
            socket.connect(ServerAddr, 500);
            session = new SocketSession(socket);

            session = replSessionSecurity.createClientSession(server, socket);
            boolean isSslEncryption =
                 replSessionSecurity.isSslEncryption(server);
            /*
             * Send our ServerStartMessage.
             */
            ServerStartMessage msg = new ServerStartMessage(serverID, baseDn,
                maxReceiveDelay, maxReceiveQueue, maxSendDelay, maxSendQueue,
                halfRcvWindow*2, heartbeatInterval, state,
                protocolVersion);
                protocolVersion, isSslEncryption);
            session.publish(msg);


@@ -264,6 +260,11 @@
                startMsg.getVersion());
            session.setSoTimeout(timeout);

            if (!isSslEncryption)
            {
              session.stopEncryption();
            }

            /*
             * We must not publish changes to a replicationServer that has not
             * seen all our previous changes because this could cause some
@@ -854,4 +855,18 @@
  {
    return !connectionError;
  }

  /**
   * Determine whether the connection to the replication server is encrypted.
   * @return true if the connection is encrypted, false otherwise.
   */
  public boolean isSessionEncrypted()
  {
    boolean isEncrypted = false;
    if (session != null)
    {
      return session.isEncrypted();
    }
    return isEncrypted;
  }
}

 opends/src/server/org/opends/server/replication/plugin/ReplicationDomain.java

@@ -80,22 +80,7 @@
import org.opends.server.replication.common.ChangeNumber;
import org.opends.server.replication.common.ChangeNumberGenerator;
import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.protocol.AckMessage;
import org.opends.server.replication.protocol.AddContext;
import org.opends.server.replication.protocol.AddMsg;
import org.opends.server.replication.protocol.DeleteContext;
import org.opends.server.replication.protocol.DoneMessage;
import org.opends.server.replication.protocol.EntryMessage;
import org.opends.server.replication.protocol.ErrorMessage;
import org.opends.server.replication.protocol.InitializeRequestMessage;
import org.opends.server.replication.protocol.InitializeTargetMessage;
import org.opends.server.replication.protocol.ModifyContext;
import org.opends.server.replication.protocol.ModifyDNMsg;
import org.opends.server.replication.protocol.ModifyDnContext;
import org.opends.server.replication.protocol.OperationContext;
import org.opends.server.replication.protocol.RoutableMessage;
import org.opends.server.replication.protocol.ReplicationMessage;
import org.opends.server.replication.protocol.UpdateMessage;
import org.opends.server.replication.protocol.*;
import org.opends.server.tasks.InitializeTargetTask;
import org.opends.server.tasks.InitializeTask;
import org.opends.server.tasks.TaskUtils;
@@ -390,7 +375,7 @@
     */
    broker = new ReplicationBroker(state, baseDN, serverId, maxReceiveQueue,
        maxReceiveDelay, maxSendQueue, maxSendDelay, window,
        heartbeatInterval);
        heartbeatInterval, new ReplSessionSecurity(configuration));

    broker.start(replicationServers);

@@ -2997,4 +2982,13 @@
  {
    return configDn;
  }

  /**
   * Determine whether the connection to the replication server is encrypted.
   * @return true if the connection is encrypted, false otherwise.
   */
  public boolean isSessionEncrypted()
  {
    return broker.isSessionEncrypted();
  }
}

 opends/src/server/org/opends/server/replication/plugin/ReplicationMonitor.java

@@ -150,6 +150,9 @@
    attr = new Attribute(type, ATTR_SERVER_STATE, values);
    attributes.add(attr);

    attributes.add(new Attribute("ssl-encryption",
                                 String.valueOf(domain.isSessionEncrypted())));

    return attributes;

  }

 opends/src/server/org/opends/server/replication/protocol/ProtocolSession.java

@@ -83,6 +83,17 @@
                         DataFormatException;

  /**
   * Stop using the security layer, if there is any.
   */
  public abstract void stopEncryption();

  /**
   * Determine whether the session is using a security layer.
   * @return true if the connection is encrypted, false otherwise.
   */
  public abstract boolean isEncrypted();

  /**
   * Retrieve the IP address of the remote server.
   *
   * @return The IP address of the remote server.

 opends/src/server/org/opends/server/replication/protocol/ReplServerStartMessage.java

@@ -43,12 +43,18 @@
{
  private static final long serialVersionUID = -5871385537169856856L;

  private String baseDn = null;
  private short serverId;
  private String serverURL;
  private String baseDn = null;
  private int windowSize;
  private ServerState serverState;

  private int windowSize;
  /**
   * Whether to continue using SSL to encrypt messages after the start
   * messages have been exchanged.
   */
  private boolean sslEncryption;


  /**
   * Create a ReplServerStartMessage.
@@ -59,11 +65,14 @@
   * @param windowSize The window size.
   * @param serverState our ServerState for this baseDn.
   * @param protocolVersion The replication protocol version of the creator.
   * @param sslEncryption Whether to continue using SSL to encrypt messages
   *                      after the start messages have been exchanged.
   */
  public ReplServerStartMessage(short serverId, String serverURL, DN baseDn,
                               int windowSize,
                               ServerState serverState,
                               short protocolVersion)
                               short protocolVersion,
                               boolean sslEncryption)
  {
    super(protocolVersion);
    this.serverId = serverId;
@@ -74,6 +83,7 @@
      this.baseDn = null;
    this.windowSize = windowSize;
    this.serverState = serverState;
    this.sslEncryption = sslEncryption;
  }

  /**
@@ -125,6 +135,13 @@
      pos += length +1;

      /*
       * read the sslEncryption setting
       */
      length = getNextLength(in, pos);
      sslEncryption = Boolean.valueOf(new String(in, pos, length, "UTF-8"));
      pos += length +1;

      /*
      * read the ServerState
      */
      serverState = new ServerState(in, pos, in.length-1);
@@ -194,9 +211,12 @@
      byte[] byteServerUrl = serverURL.getBytes("UTF-8");
      byte[] byteServerState = serverState.getBytes();
      byte[] byteWindowSize = String.valueOf(windowSize).getBytes("UTF-8");
      byte[] byteSSLEncryption =
                     String.valueOf(sslEncryption).getBytes("UTF-8");

      int length = byteDn.length + 1 + byteServerId.length + 1 +
                   byteServerUrl.length + 1 + byteWindowSize.length + 1 +
                   byteSSLEncryption.length + 1 +
                   byteServerState.length + 1;

      /* encode the header in a byte[] large enough to also contain the mods */
@@ -215,6 +235,9 @@
      /* put the window size */
      pos = addByteArray(byteWindowSize, resultByteArray, pos);

      /* put the SSL Encryption setting */
      pos = addByteArray(byteSSLEncryption, resultByteArray, pos);

      /* put the ServerState */
      pos = addByteArray(byteServerState, resultByteArray, pos);

@@ -235,4 +258,16 @@
  {
    return windowSize;
  }

  /**
   * Get the SSL encryption value for the server that created the
   * message.
   *
   * @return The SSL encryption value for the server that created the
   *         message.
   */
  public boolean getSSLEncryption()
  {
    return sslEncryption;
  }
}

 opends/src/server/org/opends/server/replication/protocol/ReplSessionSecurity.java

New file
@@ -0,0 +1,283 @@
/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License, Version 1.0 only
 * (the "License").  You may not use this file except in compliance
 * with the License.
 *
 * You can obtain a copy of the license at
 * trunk/opends/resource/legal-notices/OpenDS.LICENSE
 * or https://OpenDS.dev.java.net/OpenDS.LICENSE.
 * See the License for the specific language governing permissions
 * and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL HEADER in each
 * file and include the License file at
 * trunk/opends/resource/legal-notices/OpenDS.LICENSE.  If applicable,
 * add the following below this CDDL HEADER, with the fields enclosed
 * by brackets "[]" replaced with your own identifying information:
 *      Portions Copyright [yyyy] [name of copyright owner]
 *
 * CDDL HEADER END
 *
 *
 *      Portions Copyright 2007 Sun Microsystems, Inc.
 */

package org.opends.server.replication.protocol;

import org.opends.server.admin.std.server.ReplicationServerCfg;
import org.opends.server.admin.std.server.MultimasterDomainCfg;
import org.opends.server.types.DirectoryConfig;
import org.opends.server.types.CryptoManager;
import org.opends.server.config.ConfigException;

import javax.net.ssl.SSLSocket;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSocketFactory;
import java.util.SortedSet;
import java.net.Socket;
import java.io.IOException;

/**
 * This class represents the security configuration for replication protocol
 * sessions. It contains all the configuration required to use SSL, and it
 * determines whether encryption should be enabled for a session to a given
 * replication server.
 *
 */
public class ReplSessionSecurity
{
  /**
   * Whether the replication server should listen on a secure port.
   * Set false for test purposes only.
   */
  private static boolean useSSL = true;

  /**
   * Whether replication sessions use SSL encryption.
   */
  private boolean sslEncryption;

  /**
   * The name of the local certificate to use, or null if none is specified.
   */
  private String sslCertNickname;

  /**
   * The set of enabled SSL protocols, or null for the default set.
   */
  private String sslProtocols[];

  /**
   * The set of enabled SSL cipher suites, or null for the default set.
   */
  private String sslCipherSuites[];


  /**
   * Create a ReplSessionSecurity instance from the supplied configuration
   * values.
   *
   * @param sslCertNickname The name of the local certificate to use, or null
   *                        if none is specified.
   * @param sslProtocols    The protocols that should be enabled, or null if
   *                        the default protocols should be used.
   * @param sslCipherSuites The cipher suites that should be enabled, or null
   *                        if the default cipher suites should be used.
   * @param sslEncryption   Whether replication sessions use SSL encryption.
   *
   * @throws ConfigException    If the supplied configuration was not valid.
   */
  public ReplSessionSecurity(String sslCertNickname,
                             SortedSet<String> sslProtocols,
                             SortedSet<String> sslCipherSuites,
                             boolean sslEncryption)
       throws ConfigException
  {
    if (sslProtocols == null || sslProtocols.size() == 0)
    {
      this.sslProtocols = null;
    }
    else
    {
      this.sslProtocols = new String[sslProtocols.size()];
      sslProtocols.toArray(this.sslProtocols);
    }

    if (sslCipherSuites == null || sslCipherSuites.size() == 0)
    {
      this.sslCipherSuites = null;
    }
    else
    {
      this.sslCipherSuites = new String[sslProtocols.size()];
      sslProtocols.toArray(this.sslCipherSuites);
    }

    this.sslEncryption = sslEncryption;
    this.sslCertNickname = sslCertNickname;
  }

  /**
   * Create a ReplSessionSecurity instance from a provided replication server
   * configuration.
   *
   * @param replServerCfg The replication server configuration.
   *
   * @throws ConfigException If the supplied configuration was not valid.
   */
  public ReplSessionSecurity(ReplicationServerCfg replServerCfg)
       throws ConfigException
  {
    // Currently use global settings from the crypto manager.
    this(DirectoryConfig.getCryptoManager().getSslCertNickname(),
         DirectoryConfig.getCryptoManager().getSslProtocols(),
         DirectoryConfig.getCryptoManager().getSslCipherSuites(),
         DirectoryConfig.getCryptoManager().isSslEncryption());
  }

  /**
   * Create a ReplSessionSecurity instance from a provided multimaster domain
   * configuration.
   *
   * @param multimasterDomainCfg The multimaster domain configuration.
   *
   * @throws ConfigException If the supplied configuration was not valid.
   */
  public ReplSessionSecurity(MultimasterDomainCfg multimasterDomainCfg)
       throws ConfigException
  {
    // Currently use global settings from the crypto manager.
    this(DirectoryConfig.getCryptoManager().getSslCertNickname(),
         DirectoryConfig.getCryptoManager().getSslProtocols(),
         DirectoryConfig.getCryptoManager().getSslCipherSuites(),
         DirectoryConfig.getCryptoManager().isSslEncryption());
  }

  /**
   * Determine whether a given replication server is listening on a secure
   * port.
   * @param serverURL The replication server URL.
   * @return true if the given replication server is listening on a secure
   *         port, or false if it is listening on a non-secure port.
   */
  private boolean isSecurePort(String serverURL)
  {
    // Always true unless changed for test purposes.
    return useSSL;
  }

  /**
   * Determine whether sessions to a given replication server should be
   * encrypted.
   * @param serverURL The replication server URL.
   * @return true if sessions to the given replication server should be
   *         encrypted, or false if they should not be encrypted.
   */
  public boolean isSslEncryption(String serverURL)
  {
    // Currently use global settings from the crypto manager.
    return sslEncryption;
  }

  /**
   * Create a new protocol session in the client role on the provided socket.
   * @param serverURL The remote replication server to which the socket is
   *                  connected.
   * @param socket The connected socket.
   * @return The new protocol session.
   * @throws ConfigException If the protocol session could not be established
   *                         due to a configuration problem.
   * @throws IOException     If the protocol session could not be established
   *                         for some other reason.
   */
  public ProtocolSession createClientSession(String serverURL, Socket socket)
       throws ConfigException, IOException
  {
    boolean useSSL = isSecurePort(serverURL);
    if (useSSL)
    {
      // Create a new SSL context every time to make sure we pick up the
      // latest contents of the trust store.
      CryptoManager cryptoManager = DirectoryConfig.getCryptoManager();
      SSLContext sslContext = cryptoManager.getSslContext(sslCertNickname);
      SSLSocketFactory sslSocketFactory = sslContext.getSocketFactory();

      SSLSocket secureSocket = (SSLSocket)
           sslSocketFactory.createSocket(socket,
                                         socket.getInetAddress().getHostName(),
                                         socket.getPort(), false);
      secureSocket.setUseClientMode(true);

      if (sslProtocols != null)
      {
        secureSocket.setEnabledProtocols(sslProtocols);
      }

      if (sslCipherSuites != null)
      {
        secureSocket.setEnabledCipherSuites(sslCipherSuites);
      }

      // Force TLS negotiation now.
      secureSocket.startHandshake();

      return new TLSSocketSession(socket, secureSocket);
    }
    else
    {
      return new SocketSession(socket);
    }
  }

  /**
   * Create a new protocol session in the server role on the provided socket.
   * @param socket The connected socket.
   * @return The new protocol session.
   * @throws ConfigException If the protocol session could not be established
   *                         due to a configuration problem.
   * @throws IOException     If the protocol session could not be established
   *                         for some other reason.
   */
  public ProtocolSession createServerSession(Socket socket)
       throws ConfigException, IOException
  {
    if (useSSL)
    {
      // Create a new SSL context every time to make sure we pick up the
      // latest contents of the trust store.
      CryptoManager cryptoManager = DirectoryConfig.getCryptoManager();
      SSLContext sslContext = cryptoManager.getSslContext(sslCertNickname);
      SSLSocketFactory sslSocketFactory = sslContext.getSocketFactory();

      SSLSocket secureSocket = (SSLSocket)
           sslSocketFactory.createSocket(socket,
                                         socket.getInetAddress().getHostName(),
                                         socket.getPort(), false);
      secureSocket.setUseClientMode(false);
      secureSocket.setNeedClientAuth(true);

      if (sslProtocols != null)
      {
        secureSocket.setEnabledProtocols(sslProtocols);
      }

      if (sslCipherSuites != null)
      {
        secureSocket.setEnabledCipherSuites(sslCipherSuites);
      }

      // Force TLS negotiation now.
      secureSocket.startHandshake();

      return new TLSSocketSession(socket, secureSocket);
    }
    else
    {
      return new SocketSession(socket);
    }
  }

}

 opends/src/server/org/opends/server/replication/protocol/ServerStartMessage.java

@@ -63,6 +63,12 @@
  private long heartbeatInterval = 0;

  /**
   * Whether to continue using SSL to encrypt messages after the start
   * messages have been exchanged.
   */
  private boolean sslEncryption;

  /**
   * Create a new ServerStartMessage.
   *
   * @param serverId The serverId of the server for which the ServerStartMessage
@@ -76,13 +82,16 @@
   * @param heartbeatInterval The requested heartbeat interval.
   * @param serverState  The state of this server.
   * @param protocolVersion The replication protocol version of the creator.
   * @param sslEncryption Whether to continue using SSL to encrypt messages
   *                      after the start messages have been exchanged.
   */
  public ServerStartMessage(short serverId, DN baseDn, int maxReceiveDelay,
                            int maxReceiveQueue, int maxSendDelay,
                            int maxSendQueue, int windowSize,
                            long heartbeatInterval,
                            ServerState serverState,
                            short protocolVersion)
                            short protocolVersion,
                            boolean sslEncryption)
  {
    super(protocolVersion);

@@ -94,7 +103,7 @@
    this.maxSendQueue = maxSendQueue;
    this.windowSize = windowSize;
    this.heartbeatInterval = heartbeatInterval;

    this.sslEncryption = sslEncryption;
    this.serverState = serverState;

    try
@@ -194,6 +203,13 @@
      pos += length +1;

      /*
       * read the sslEncryption setting
       */
      length = getNextLength(in, pos);
      sslEncryption = Boolean.valueOf(new String(in, pos, length, "UTF-8"));
      pos += length +1;

      /*
      * read the ServerState
      */
      serverState = new ServerState(in, pos, in.length-1);
@@ -308,6 +324,8 @@
                     String.valueOf(windowSize).getBytes("UTF-8");
      byte[] byteHeartbeatInterval =
                     String.valueOf(heartbeatInterval).getBytes("UTF-8");
      byte[] byteSSLEncryption =
                     String.valueOf(sslEncryption).getBytes("UTF-8");
      byte[] byteServerState = serverState.getBytes();

      int length = byteDn.length + 1 + byteServerId.length + 1 +
@@ -318,6 +336,7 @@
                   byteMaxSendQueue.length + 1 +
                   byteWindowSize.length + 1 +
                   byteHeartbeatInterval.length + 1 +
                   byteSSLEncryption.length + 1 +
                   byteServerState.length + 1;

      /* encode the header in a byte[] large enough to also contain the mods */
@@ -342,6 +361,8 @@

      pos = addByteArray(byteHeartbeatInterval, resultByteArray, pos);

      pos = addByteArray(byteSSLEncryption, resultByteArray, pos);

      pos = addByteArray(byteServerState, resultByteArray, pos);

      return resultByteArray;
@@ -373,4 +394,16 @@
  {
    return heartbeatInterval;
  }

  /**
   * Get the SSL encryption value for the ldap server that created the
   * message.
   *
   * @return The SSL encryption value for the ldap server that created the
   *         message.
   */
  public boolean getSSLEncryption()
  {
    return sslEncryption;
  }
}

 opends/src/server/org/opends/server/replication/protocol/SocketSession.java

@@ -165,6 +165,22 @@
  /**
   * {@inheritDoc}
   */
  public void stopEncryption()
  {
    // There is no security layer.
  }

  /**
   * {@inheritDoc}
   */
  public boolean isEncrypted()
  {
    return false;
  }

  /**
   * {@inheritDoc}
   */
  public long getLastPublishTime()
  {
    return lastPublishTime;

 opends/src/server/org/opends/server/replication/protocol/TLSSocketSession.java

New file
@@ -0,0 +1,231 @@
/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License, Version 1.0 only
 * (the "License").  You may not use this file except in compliance
 * with the License.
 *
 * You can obtain a copy of the license at
 * trunk/opends/resource/legal-notices/OpenDS.LICENSE
 * or https://OpenDS.dev.java.net/OpenDS.LICENSE.
 * See the License for the specific language governing permissions
 * and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL HEADER in each
 * file and include the License file at
 * trunk/opends/resource/legal-notices/OpenDS.LICENSE.  If applicable,
 * add the following below this CDDL HEADER, with the fields enclosed
 * by brackets "[]" replaced with your own identifying information:
 *      Portions Copyright [yyyy] [name of copyright owner]
 *
 * CDDL HEADER END
 *
 *
 *      Portions Copyright 2006-2007 Sun Microsystems, Inc.
 */
package org.opends.server.replication.protocol;

import static org.opends.server.loggers.debug.DebugLogger.debugEnabled;
import static org.opends.server.loggers.debug.DebugLogger.getTracer;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.net.SocketException;
import java.util.zip.DataFormatException;

import org.opends.server.loggers.debug.DebugTracer;

import javax.net.ssl.SSLSocket;

/**
 * This class implements a protocol session using TLS.
 */
public class TLSSocketSession implements ProtocolSession
{
  /**
   * The tracer object for the debug logger.
   */
  private static final DebugTracer TRACER = getTracer();

  private Socket plainSocket;
  private SSLSocket secureSocket;
  private InputStream input;
  private OutputStream output;
  private InputStream plainInput;
  private OutputStream plainOutput;
  byte[] rcvLengthBuf = new byte[8];

  /**
   * The time the last message published to this session.
   */
  private long lastPublishTime = 0;


  /**
   * The time the last message was received on this session.
   */
  private long lastReceiveTime = 0;


  /**
   * Creates a new TLSSocketSession.
   *
   * @param socket       The regular Socket on which the SocketSession will be
   *                     based.
   * @param secureSocket The secure Socket on which the SocketSession will be
   *                     based.
   * @throws IOException When an IException happens on the socket.
   */
  public TLSSocketSession(Socket socket, SSLSocket secureSocket)
       throws IOException
  {
    plainSocket = socket;
    this.secureSocket = secureSocket;
    plainInput = plainSocket.getInputStream();
    plainOutput = plainSocket.getOutputStream();
    input = secureSocket.getInputStream();
    output = secureSocket.getOutputStream();
  }


  /**
   * {@inheritDoc}
   */
  public void close() throws IOException
  {
    if (debugEnabled())
    {
      TRACER.debugVerbose("Closing SocketSession.");
    }
    if (plainSocket != null && !plainSocket.isClosed())
    {
      plainSocket.close();
    }
    if (secureSocket != null && !secureSocket.isClosed())
    {
      secureSocket.close();
    }
  }

  /**
   * {@inheritDoc}
   */
  public synchronized void publish(ReplicationMessage msg)
         throws IOException
  {
    byte[] buffer = msg.getBytes();
    String str = String.format("%08x", buffer.length);
    byte[] sendLengthBuf = str.getBytes();

    output.write(sendLengthBuf);
    output.write(buffer);
    output.flush();

    lastPublishTime = System.currentTimeMillis();
  }

  /**
   * {@inheritDoc}
   */
  public ReplicationMessage receive() throws IOException,
      ClassNotFoundException, DataFormatException
  {
    /* Read the first 8 bytes containing the packet length */
    int length = 0;

    /* Let's start the stop-watch before waiting on read */
    /* for the heartbeat check to be operationnal        */
    lastReceiveTime = System.currentTimeMillis();

    while (length<8)
    {
      int read = input.read(rcvLengthBuf, length, 8-length);
      if (read == -1)
      {
        lastReceiveTime=0;
        throw new IOException("no more data");
      }
      else
      {
        length += read;
      }
    }

    int totalLength = Integer.parseInt(new String(rcvLengthBuf), 16);

    try
    {
      length = 0;
      byte[] buffer = new byte[totalLength];
      while (length < totalLength)
      {
        length += input.read(buffer, length, totalLength - length);
      }
      /* We do not want the heartbeat to close the session when */
      /* we are processing a message even a time consuming one. */
      lastReceiveTime=0;
      return ReplicationMessage.generateMsg(buffer);
    }
    catch (OutOfMemoryError e)
    {
      throw new IOException("Packet too large, can't allocate "
                            + totalLength + " bytes.");
    }
  }

  /**
   * {@inheritDoc}
   */
  public void stopEncryption()
  {
    input = plainInput;
    output = plainOutput;
  }

  /**
   * {@inheritDoc}
   */
  public boolean isEncrypted()
  {
    return !(input == plainInput);
  }

  /**
   * {@inheritDoc}
   */
  public long getLastPublishTime()
  {
    return lastPublishTime;
  }

  /**
   * {@inheritDoc}
   */
  public long getLastReceiveTime()
  {
    if (lastReceiveTime==0)
    {
      return System.currentTimeMillis();
    }
    return lastReceiveTime;
  }

  /**
   * {@inheritDoc}
   */
  public String getRemoteAddress()
  {
    return plainSocket.getInetAddress().getHostAddress();
  }

  /**
   * {@inheritDoc}
   */
  public void setSoTimeout(int timeout) throws SocketException
  {
    plainSocket.setSoTimeout(timeout);
  }
}

 opends/src/server/org/opends/server/replication/server/ReplicationServer.java

@@ -53,7 +53,8 @@
import org.opends.server.api.MonitorProvider;
import org.opends.server.config.ConfigException;
import org.opends.server.core.DirectoryServer;
import org.opends.server.replication.protocol.SocketSession;
import org.opends.server.replication.protocol.ReplSessionSecurity;
import org.opends.server.replication.protocol.ProtocolSession;
import org.opends.server.types.Attribute;
import org.opends.server.types.AttributeType;
import org.opends.server.types.AttributeValue;
@@ -103,6 +104,7 @@
  private int replicationPort;
                        // de deleted from the persistent storage.
  private boolean stopListen = false;
  private ReplSessionSecurity replSessionSecurity;

  /**
   * Creates a new Replication server using the provided configuration entry.
@@ -129,7 +131,7 @@
    {
      dbDirname = "changelogDb";
    }
    // Chech that this path exists or create it.
    // Check that this path exists or create it.
    File f = getFileForPath(dbDirname);
    try
    {
@@ -149,6 +151,7 @@
      throw new ConfigException(msg, e);
    }

    replSessionSecurity = new ReplSessionSecurity(configuration);
    initialize(replicationServerId, replicationPort);
    configuration.addChangeListener(this);
    DirectoryServer.registerMonitorProvider(this);
@@ -164,7 +167,7 @@

  void runListen()
  {
    Socket newSocket = null;
    Socket newSocket;
    while ((shutdown == false) && (stopListen  == false))
    {
      // Wait on the replicationServer port.
@@ -177,10 +180,13 @@
        newSocket.setReceiveBufferSize(1000000);
        newSocket.setTcpNoDelay(true);
        newSocket.setKeepAlive(true);
        ServerHandler handler = new ServerHandler(
                                     new SocketSession(newSocket), queueSize);
        handler.start(null, serverId, serverURL, rcvWindow, this);
      } catch (IOException e)
        ProtocolSession session =
             replSessionSecurity.createServerSession(newSocket);
        ServerHandler handler = new ServerHandler(session, queueSize);
        handler.start(null, serverId, serverURL, rcvWindow,
                      false, this);
      }
      catch (Exception e)
      {
        // The socket has probably been closed as part of the
        // shutdown or changing the port number process.
@@ -264,6 +270,7 @@
    int separator = serverURL.lastIndexOf(':');
    String port = serverURL.substring(separator + 1);
    String hostname = serverURL.substring(0, separator);
    boolean sslEncryption = replSessionSecurity.isSslEncryption(serverURL);

    try
    {
@@ -275,10 +282,12 @@
      socket.connect(ServerAddr, 500);

      ServerHandler handler = new ServerHandler(
                                      new SocketSession(socket), queueSize);
     handler.start(baseDn, serverId, this.serverURL, rcvWindow, this);
           replSessionSecurity.createClientSession(serverURL, socket),
           queueSize);
      handler.start(baseDn, serverId, this.serverURL, rcvWindow,
                    sslEncryption, this);
    }
    catch (IOException e)
    catch (Exception e)
    {
      // ignore
    }
@@ -523,7 +532,7 @@
    }

    if ((configuration.getReplicationDbDirectory() != null) &&
        (dbDirname != configuration.getReplicationDbDirectory()))
        (!dbDirname.equals(configuration.getReplicationDbDirectory())))
    {
      return new ConfigChangeResult(ResultCode.SUCCESS, true);
    }

 opends/src/server/org/opends/server/replication/server/ServerHandler.java

@@ -51,18 +51,7 @@
import org.opends.server.core.DirectoryServer;
import org.opends.server.replication.common.ChangeNumber;
import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.protocol.ProtocolVersion;
import org.opends.server.replication.protocol.AckMessage;
import org.opends.server.replication.protocol.ReplServerStartMessage;
import org.opends.server.replication.protocol.HeartbeatThread;
import org.opends.server.replication.protocol.ProtocolSession;
import org.opends.server.replication.protocol.RoutableMessage;
import org.opends.server.replication.protocol.ServerStartMessage;
import org.opends.server.replication.protocol.ReplicationMessage;
import org.opends.server.replication.protocol.UpdateMessage;
import org.opends.server.replication.protocol.WindowMessage;
import org.opends.server.replication.protocol.WindowProbe;
import org.opends.server.replication.protocol.ReplServerInfoMessage;
import org.opends.server.replication.protocol.*;
import org.opends.server.types.Attribute;
import org.opends.server.types.AttributeType;
import org.opends.server.types.AttributeValue;
@@ -180,12 +169,16 @@
   * @param replicationServerURL The URL of the replicationServer that creates
   *                             this server handler.
   * @param windowSize the window size that this server handler must use.
   * @param sslEncryption For outgoing connections indicates whether encryption
   *                      should be used after the exchange of start messages.
   *                      Ignored for incoming connections.
   * @param replicationServer the ReplicationServer that created this server
   *                          handler.
   */
  public void start(DN baseDn, short replicationServerId,
                    String replicationServerURL,
                    int windowSize, ReplicationServer replicationServer)
                    int windowSize, boolean sslEncryption,
                    ReplicationServer replicationServer)
  {
    this.replicationServerId = replicationServerId;
    rcvWindowSizeHalf = windowSize/2;
@@ -202,8 +195,7 @@
        ReplServerStartMessage msg =
          new ReplServerStartMessage(replicationServerId, replicationServerURL,
                                    baseDn, windowSize, localServerState,
                                    protocolVersion);

                                    protocolVersion, sslEncryption);
        session.publish(msg);
      }

@@ -227,6 +219,9 @@
        maxSendQueue = receivedMsg.getMaxSendQueue();
        heartbeatInterval = receivedMsg.getHeartbeatInterval();

        // The session initiator decides whether to use SSL.
        sslEncryption = receivedMsg.getSSLEncryption();

        if (maxReceiveQueue > 0)
          restartReceiveQueue = (maxReceiveQueue > 1000 ?
                                  maxReceiveQueue - 200 :
@@ -266,7 +261,7 @@
        ReplServerStartMessage myStartMsg =
          new ReplServerStartMessage(replicationServerId, replicationServerURL,
                                    this.baseDn, windowSize, localServerState,
                                    protocolVersion);
                                    protocolVersion, sslEncryption);
        session.publish(myStartMsg);
        sendWindowSize = receivedMsg.getWindowSize();
      }
@@ -288,12 +283,15 @@
          replicationCache = replicationServer.getReplicationCache(this.baseDn);
          ServerState serverState = replicationCache.getDbServerState();

          // The session initiator decides whether to use SSL.
          sslEncryption = receivedMsg.getSSLEncryption();

          // Publish our start message
          ReplServerStartMessage outMsg =
            new ReplServerStartMessage(replicationServerId,
                                       replicationServerURL,
                                       this.baseDn, windowSize, serverState,
                                       protocolVersion);
                                       protocolVersion, sslEncryption);
          session.publish(outMsg);
        }
        else
@@ -309,6 +307,11 @@
        return;   // we did not recognize the message, ignore it
      }

      if (!sslEncryption)
      {
        session.stopEncryption();
      }

      replicationCache = replicationServer.getReplicationCache(this.baseDn);

      boolean started;
@@ -1230,6 +1233,10 @@
    }
    Attribute attr = new Attribute(type, ATTR_SERVER_STATE, values);
    attributes.add(attr);

    attributes.add(new Attribute("ssl-encryption",
                                 String.valueOf(session.isEncrypted())));

    return attributes;
  }


 opends/src/server/org/opends/server/types/CryptoManager.java

@@ -26,7 +26,7 @@
 */
package org.opends.server.types;
import org.opends.messages.Message;

import static org.opends.messages.CoreMessages.*;


import java.io.InputStream;
@@ -40,19 +40,29 @@
import java.util.zip.DataFormatException;
import java.util.zip.Deflater;
import java.util.zip.Inflater;
import java.util.SortedSet;
import javax.crypto.Cipher;
import javax.crypto.Mac;
import javax.crypto.NoSuchPaddingException;
import javax.crypto.SecretKey;
import javax.crypto.spec.IvParameterSpec;
import javax.crypto.spec.SecretKeySpec;
import javax.net.ssl.KeyManager;
import javax.net.ssl.TrustManager;
import javax.net.ssl.SSLContext;
import javax.net.ssl.X509ExtendedKeyManager;

import org.opends.server.config.ConfigException;
import org.opends.server.config.ConfigConstants;

import static org.opends.server.loggers.debug.DebugLogger.*;
import org.opends.server.loggers.debug.DebugTracer;
import static org.opends.server.util.StaticUtils.*;

import org.opends.server.util.SelectableCertificateKeyManager;
import org.opends.server.api.Backend;
import org.opends.server.core.DirectoryServer;
import org.opends.server.backends.TrustStoreBackend;
import org.opends.server.admin.std.server.CryptoManagerCfg;

/**
 * This class provides the interface to the Directory Server
@@ -92,10 +102,23 @@
  // The preferred MAC algorithm for the Directory Server.
  private String preferredMACAlgorithm;

  // The name of the local certificate to use for SSL.
  private String sslCertNickname;

  // Whether replication sessions use SSL encryption.
  private boolean sslEncryption;

  // The set of SSL protocols enabled or null for the default set.
  private SortedSet<String> sslProtocols;

  // The set of SSL cipher suites enabled or null for the default set.
  private SortedSet<String> sslCipherSuites;

  /**
   * Creates a new instance of this crypto manager object.
   * Creates a new instance of this crypto manager object from a given
   * configuration.
   *
   * @param   cfg  The configuration of this crypto manager.
   *
   * @throws  ConfigException  If a problem occurs while creating this
   *                           crypto manager that is a result of a
@@ -106,7 +129,7 @@
   *                                   that is not the result of a
   *                                   problem in the configuration.
   */
  public CryptoManager()
  public CryptoManager(CryptoManagerCfg cfg)
         throws ConfigException, InitializationException
  {
    // FIXME -- Get the defaults from the configuration rather than
@@ -175,6 +198,11 @@
                     Message.raw("Can't get preferred cipher:  " +
                     getExceptionMessage(e).toString()), e);
    }

    sslCertNickname = cfg.getSSLCertNickname();
    sslEncryption   = cfg.isSSLEncryption();
    sslProtocols    = cfg.getSSLProtocols();
    sslCipherSuites = cfg.getSSLCipherSuites();
  }


@@ -865,5 +893,122 @@
      inflater.end();
    }
  }


  /**
   * Retrieve the ADS trust store backend.
   * @return The ADS trust store backend.
   * @throws ConfigException If the ADS trust store backend is
   *                         not configured.
   */
  private TrustStoreBackend getTrustStoreBackend()
       throws ConfigException
  {
    Backend b = DirectoryServer.getBackend(
         ConfigConstants.ID_ADS_TRUST_STORE_BACKEND);
    if (b == null)
    {
      Message msg =
           ERR_CRYPTOMGR_ADS_TRUST_STORE_BACKEND_NOT_ENABLED.get(
                ConfigConstants.ID_ADS_TRUST_STORE_BACKEND);
      throw new ConfigException(msg);
    }
    if (!(b instanceof TrustStoreBackend))
    {
      Message msg =
           ERR_CRYPTOMGR_ADS_TRUST_STORE_BACKEND_WRONG_CLASS.get(
                ConfigConstants.ID_ADS_TRUST_STORE_BACKEND);
      throw new ConfigException(msg);
    }
    return (TrustStoreBackend)b;
  }

  /**
   * Create an SSL context that may be used for communication to
   * another ADS component.
   *
   * @param sslCertNickname The name of the local certificate to use,
   *                        or null if none is specified.
   * @return A new SSL Context.
   * @throws ConfigException If the context could not be created.
   */
  public SSLContext getSslContext(String sslCertNickname)
       throws ConfigException
  {
    SSLContext sslContext;
    try
    {
      TrustStoreBackend trustStoreBackend = getTrustStoreBackend();
      KeyManager[] keyManagers = trustStoreBackend.getKeyManagers();
      TrustManager[] trustManagers =
           trustStoreBackend.getTrustManagers();

      sslContext = SSLContext.getInstance("TLS");

      if (sslCertNickname == null)
      {
        sslContext.init(keyManagers, trustManagers, null);
      }
      else
      {
        X509ExtendedKeyManager[] extendedKeyManagers =
             SelectableCertificateKeyManager.wrap(
                  keyManagers,
                  sslCertNickname);
        sslContext.init(extendedKeyManagers, trustManagers, null);
      }
    }
    catch (Exception e)
    {
      if (debugEnabled())
      {
        TRACER.debugCaught(DebugLogLevel.ERROR, e);
      }

      Message message =
           ERR_CRYPTOMGR_SSL_CONTEXT_CANNOT_INITIALIZE.get(
                getExceptionMessage(e));
      throw new ConfigException(message, e);
    }

    return sslContext;
  }


  /**
   * Get the name of the local certificate to use for SSL.
   * @return The name of the local certificate to use for SSL.
   */
  public String getSslCertNickname()
  {
    return sslCertNickname;
  }

  /**
   * Determine whether SSL encryption is enabled.
   * @return true if SSL encryption is enabled.
   */
  public boolean isSslEncryption()
  {
    return sslEncryption;
  }

  /**
   * Get the set of enabled SSL protocols.
   * @return The set of enabled SSL protocols.
   */
  public SortedSet<String> getSslProtocols()
  {
    return sslProtocols;
  }

  /**
   * Get the set of enabled SSL cipher suites.
   * @return The set of enabled SSL cipher suites.
   */
  public SortedSet<String> getSslCipherSuites()
  {
    return sslCipherSuites;
  }
}


 opends/tests/unit-tests-testng/resource/config-changes.ldif

@@ -335,6 +335,25 @@
-
replace: ds-cfg-key-store-pin-file

dn: cn=Crypto Manager,cn=config
changetype: modify
replace: ds-cfg-ssl-cert-nickname
ds-cfg-ssl-cert-nickname: server-cert
-

dn: ds-cfg-backend-id=ads-truststore,cn=Backends,cn=config
changetype: modify
replace: ds-cfg-backend-enabled
ds-cfg-backend-enabled: true
-
replace: ds-cfg-trust-store-file
ds-cfg-trust-store-file: config/server.keystore
-
replace: ds-cfg-trust-store-pin
ds-cfg-trust-store-pin: password
-
replace: ds-cfg-trust-store-pin-file

dn: cn=PKCS12,cn=Key Manager Providers,cn=config
changetype: modify
replace: ds-cfg-key-manager-provider-enabled

 opends/tests/unit-tests-testng/src/server/org/opends/server/TestCaseUtils.java

@@ -53,12 +53,9 @@
import org.opends.server.core.DirectoryServer;
import org.opends.server.core.LockFileManager;
import org.opends.server.extensions.ConfigFileHandler;
import org.opends.server.loggers.AccessLogger;
import org.opends.server.loggers.TextErrorLogPublisher;
import org.opends.server.loggers.ErrorLogger;
import org.opends.server.loggers.TextAccessLogPublisher;
import org.opends.server.loggers.debug.TextDebugLogPublisher;
import org.opends.server.loggers.debug.DebugLogger;
import org.opends.server.plugins.InvocationCounterPlugin;
import org.opends.server.protocols.internal.InternalClientConnection;
import org.opends.server.protocols.asn1.ASN1Reader;

 opends/tests/unit-tests-testng/src/server/org/opends/server/replication/ProtocolWindowTest.java

@@ -351,7 +351,8 @@
    ReplicationBroker broker = new ReplicationBroker(
        new ServerState(), 
        baseDn, 
        (short) 13, 0, 0, 0, 0, 1000, 0);
        (short) 13, 0, 0, 0, 0, 1000, 0,
        getReplSessionSecurity());
 

    // Check broker hard-coded version

 opends/tests/unit-tests-testng/src/server/org/opends/server/replication/ReplicationTestCase.java

@@ -28,13 +28,13 @@

import static org.opends.server.config.ConfigConstants.ATTR_TASK_COMPLETION_TIME;
import static org.opends.server.config.ConfigConstants.ATTR_TASK_STATE;
import org.opends.server.config.ConfigException;
import static org.opends.server.loggers.ErrorLogger.logError;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;

import java.net.SocketException;
import java.util.ArrayList;
import java.util.LinkedHashSet;
import java.util.LinkedList;
@@ -58,6 +58,7 @@
import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.plugin.PersistentServerState;
import org.opends.server.replication.plugin.ReplicationBroker;
import org.opends.server.replication.protocol.ReplSessionSecurity;
import org.opends.server.schema.DirectoryStringSyntax;
import org.opends.server.schema.IntegerSyntax;
import org.opends.server.types.Attribute;
@@ -141,7 +142,7 @@
  protected ReplicationBroker openReplicationSession(
      final DN baseDn, short serverId, int window_size,
      int port, int timeout, boolean emptyOldChanges)
          throws Exception, SocketException
          throws Exception
  {
    ServerState state;
    if (emptyOldChanges)
@@ -150,7 +151,8 @@
       state = new ServerState();

    ReplicationBroker broker = new ReplicationBroker(
        state, baseDn, serverId, 0, 0, 0, 0, window_size, 0);
        state, baseDn, serverId, 0, 0, 0, 0, window_size, 0,
        getReplSessionSecurity());
    ArrayList<String> servers = new ArrayList<String>(1);
    servers.add("localhost:" + port);
    broker.start(servers);
@@ -188,10 +190,11 @@
  protected ReplicationBroker openReplicationSession(
      final DN baseDn, short serverId, int window_size,
      int port, int timeout, ServerState state)
          throws Exception, SocketException
          throws Exception
  {
    ReplicationBroker broker = new ReplicationBroker(
        state, baseDn, serverId, 0, 0, 0, 0, window_size, 0);
        state, baseDn, serverId, 0, 0, 0, 0, window_size, 0,
        getReplSessionSecurity());
    ArrayList<String> servers = new ArrayList<String>(1);
    servers.add("localhost:" + port);
    broker.start(servers);
@@ -210,7 +213,7 @@
      final DN baseDn, short serverId, int window_size,
      int port, int timeout, int maxSendQueue, int maxRcvQueue,
      boolean emptyOldChanges)
          throws Exception, SocketException
          throws Exception
  {
    ServerState state;
    if (emptyOldChanges)
@@ -220,7 +223,8 @@

    ReplicationBroker broker = new ReplicationBroker(
        state, baseDn, serverId, maxRcvQueue, 0,
        maxSendQueue, 0, window_size, 0);
        maxSendQueue, 0, window_size, 0,
        getReplSessionSecurity());
    ArrayList<String> servers = new ArrayList<String>(1);
    servers.add("localhost:" + port);
    broker.start(servers);
@@ -606,4 +610,17 @@
                 "The task completed in an unexpected state");
  }

  /**
   * Create a new replication session security object that can be used in
   * unit tests.
   *
   * @return A new replication session security object.
   * @throws ConfigException If an error occurs.
   */
  protected static ReplSessionSecurity getReplSessionSecurity()
       throws ConfigException
  {
    return new ReplSessionSecurity(null, null, null, true);
  }

}

 opends/tests/unit-tests-testng/src/server/org/opends/server/replication/plugin/HistoricalTest.java

@@ -29,8 +29,6 @@

import org.opends.server.replication.ReplicationTestCase;
import org.opends.server.replication.common.ChangeNumber;
import org.opends.server.replication.plugin.ReplicationBroker;
import org.opends.server.replication.plugin.Historical;
import org.opends.server.replication.protocol.ModifyMsg;
import org.opends.server.TestCaseUtils;
import org.opends.server.protocols.internal.InternalClientConnection;

 opends/tests/unit-tests-testng/src/server/org/opends/server/replication/protocol/SynchronizationMsgTest.java

@@ -493,7 +493,7 @@
  {
    state.update(new ChangeNumber((long)1, 1,(short)1));
    ServerStartMessage msg = new ServerStartMessage(serverId, baseDN,
        window, window, window, window, window, window, state, (short)1);
        window, window, window, window, window, window, state, (short)1, true);
    ServerStartMessage newMsg = new ServerStartMessage(msg.getBytes());
    assertEquals(msg.getServerId(), newMsg.getServerId());
    assertEquals(msg.getBaseDn(), newMsg.getBaseDn());
@@ -503,6 +503,7 @@
    assertEquals(msg.getMaxSendQueue(), newMsg.getMaxSendQueue());
    assertEquals(msg.getWindowSize(), newMsg.getWindowSize());
    assertEquals(msg.getHeartbeatInterval(), newMsg.getHeartbeatInterval());
    assertEquals(msg.getSSLEncryption(), newMsg.getSSLEncryption());
    assertEquals(msg.getServerState().getMaxChangeNumber((short)1),
        newMsg.getServerState().getMaxChangeNumber((short)1));
    assertEquals(msg.getVersion(), newMsg.getVersion());
@@ -526,15 +527,16 @@
  {
    state.update(new ChangeNumber((long)1, 1,(short)1));
    ReplServerStartMessage msg = new ReplServerStartMessage(serverId,
        url, baseDN, window, state, (short)1);
        url, baseDN, window, state, (short)1, true);
    ReplServerStartMessage newMsg = new ReplServerStartMessage(msg.getBytes());
    assertEquals(msg.getServerId(), newMsg.getServerId());
    assertEquals(msg.getServerURL(), newMsg.getServerURL());
    assertEquals(msg.getBaseDn(), newMsg.getBaseDn());
    assertEquals(msg.getWindowSize(), newMsg.getWindowSize());
    assertEquals(msg.getWindowSize(), newMsg.getWindowSize());
    assertEquals(msg.getServerState().getMaxChangeNumber((short)1),
        newMsg.getServerState().getMaxChangeNumber((short)1));
    assertEquals(msg.getVersion(), newMsg.getVersion());
    assertEquals(msg.getSSLEncryption(), newMsg.getSSLEncryption());
  }

  /**

 opends/tests/unit-tests-testng/src/server/org/opends/server/replication/server/DbHandlerTest.java

@@ -34,9 +34,6 @@
import org.opends.server.replication.common.ChangeNumber;
import org.opends.server.replication.common.ChangeNumberGenerator;
import org.opends.server.replication.protocol.DeleteMsg;
import org.opends.server.replication.server.ReplicationServer;
import org.opends.server.replication.server.ReplicationDbEnv;
import org.opends.server.replication.server.DbHandler;
import org.opends.server.types.DN;
import org.testng.annotations.Test;
import static org.testng.Assert.*;

 opends/tests/unit-tests-testng/src/server/org/opends/server/replication/server/ReplicationServerTest.java

@@ -48,25 +48,14 @@
import org.opends.server.replication.common.ChangeNumberGenerator;
import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.plugin.ReplicationBroker;
import org.opends.server.replication.protocol.AddMsg;
import org.opends.server.replication.protocol.DeleteMsg;
import org.opends.server.replication.protocol.ModifyDNMsg;
import org.opends.server.replication.protocol.ModifyDnContext;
import org.opends.server.replication.protocol.ModifyMsg;
import org.opends.server.replication.protocol.ProtocolVersion;
import org.opends.server.replication.protocol.ReplServerStartMessage;
import org.opends.server.replication.protocol.ReplicationMessage;
import org.opends.server.replication.protocol.ServerStartMessage;
import org.opends.server.replication.protocol.SocketSession;
import org.opends.server.replication.protocol.WindowMessage;
import org.opends.server.replication.protocol.WindowProbe;
import org.opends.server.replication.server.ReplicationServer;
import org.opends.server.replication.protocol.*;
import org.opends.server.types.Attribute;
import org.opends.server.types.DN;
import org.opends.server.types.Entry;
import org.opends.server.types.Modification;
import org.opends.server.types.ModificationType;
import org.opends.server.types.RDN;
import org.opends.server.types.DirectoryConfig;
import org.opends.server.util.TimeThread;
import org.opends.server.workflowelement.localbackend.LocalBackendModifyDNOperation;
import org.testng.annotations.AfterClass;
@@ -782,14 +771,20 @@
     * This should guarantee that old changes are not perturbing this test.
     */

    // open the first session to the
    // open the first session to the replication server
    String serverURL = "localhost:" + replicationServerPort;
    InetSocketAddress ServerAddr = new InetSocketAddress(
        InetAddress.getByName("localhost"), replicationServerPort);
    Socket socket = new Socket();
    socket.setReceiveBufferSize(1000000);
    socket.setTcpNoDelay(true);
    socket.connect(ServerAddr, 500);
    SocketSession session = new SocketSession(socket);
    ReplSessionSecurity replSessionSecurity = getReplSessionSecurity();
    ProtocolSession session =
         replSessionSecurity.createClientSession(serverURL, socket);

    boolean sslEncryption =
         DirectoryConfig.getCryptoManager().isSslEncryption();

    try
    {
@@ -797,7 +792,7 @@
      ServerStartMessage msg =
        new ServerStartMessage((short) 1723, DN.decode("dc=example,dc=com"),
            0, 0, 0, 0, WINDOW, (long) 5000, new ServerState(),
            ProtocolVersion.currentVersion());
            ProtocolVersion.currentVersion(), sslEncryption);
      session.publish(msg);

      // Read the Replication Server state from the ReplServerStartMessage that
@@ -807,6 +802,11 @@
      int serverwindow = replStartMsg.getWindowSize();
      ServerState replServerState = replStartMsg.getServerState();

      if (!sslEncryption)
      {
        session.stopEncryption();
      }

      // close the session
      session.close();

@@ -815,19 +815,24 @@
      socket.setReceiveBufferSize(1000000);
      socket.setTcpNoDelay(true);
      socket.connect(ServerAddr, 500);
      session = new SocketSession(socket);
      session = replSessionSecurity.createClientSession(serverURL, socket);

      // send a ServerStartMessage containing the ServerState that was just
      // received.
      msg = new ServerStartMessage(
          (short) 1724, DN.decode("dc=example,dc=com"),
          0, 0, 0, 0, WINDOW, (long) 5000, replServerState,
          ProtocolVersion.currentVersion());
          ProtocolVersion.currentVersion(), sslEncryption);
      session.publish(msg);

      // Read the ReplServerStartMessage that come back.
      session.receive();

      if (!sslEncryption)
      {
        session.stopEncryption();
      }

      // Now comes the real test : check that the Replication Server
      // answers correctly to a WindowProbe Message.
      session.publish(new WindowProbe());