Fix OPENDJ-422: Concurrent writes of large LDAP messages can become interle...

Matthew Swift

16.23.2012 4e84ab1700a2ed1771a41567796c3d9e9c658d88

Fix OPENDJ-422: Concurrent writes of large LDAP messages can become interleaved

* switch to Grizzly 2.1.9-SNAPSHOT
* use FilterChainBuilder to rebuild filter chain when SASL/TLS layers are added
* improve synchronization in LDAP client connection implementation.

 opendj3/opendj-ldap-sdk/pom.xml

@@ -47,7 +47,7 @@
    <dependency>
      <groupId>org.glassfish.grizzly</groupId>
      <artifactId>grizzly-framework</artifactId>
      <version>2.1.7</version>
      <version>2.1.9-SNAPSHOT</version>
      <scope>compile</scope>
    </dependency>
    <dependency>

 opendj3/opendj-ldap-sdk/src/main/java/com/forgerock/opendj/ldap/LDAPClientFilter.java

@@ -126,8 +126,8 @@
              if (!bindClient.evaluateResult(result))
              {
                // The server is expecting a multi stage bind response.
                final int msgID = ldapConnection
                    .addPendingRequest(pendingRequest);
                final int msgID =
                    ldapConnection.continuePendingBindRequest(future);

                final ASN1BufferWriter asn1Writer =
                    ASN1BufferWriter.getWriter();

 opendj3/opendj-ldap-sdk/src/main/java/com/forgerock/opendj/ldap/LDAPConnection.java

@@ -30,7 +30,6 @@



import static com.forgerock.opendj.ldap.SynchronizedConnection.synchronizeConnection;
import static org.forgerock.opendj.ldap.ErrorResultException.newErrorResult;

import java.io.IOException;
@@ -49,9 +48,9 @@
import org.forgerock.opendj.ldap.requests.*;
import org.forgerock.opendj.ldap.responses.*;
import org.glassfish.grizzly.CompletionHandler;
import org.glassfish.grizzly.filterchain.DefaultFilterChain;
import org.glassfish.grizzly.filterchain.Filter;
import org.glassfish.grizzly.filterchain.FilterChain;
import org.glassfish.grizzly.filterchain.FilterChainBuilder;
import org.glassfish.grizzly.ssl.SSLEngineConfigurator;
import org.glassfish.grizzly.ssl.SSLFilter;

@@ -69,9 +68,8 @@
final class LDAPConnection extends AbstractAsynchronousConnection implements
    Connection
{
  private final SynchronizedConnection<?> connection;
  private final org.glassfish.grizzly.Connection<?> connection;
  private Result connectionInvalidReason;
  private FilterChain customFilterChain;
  private boolean isClosed = false;
  private final List<ConnectionEventListener> listeners =
      new CopyOnWriteArrayList<ConnectionEventListener>();
@@ -97,8 +95,7 @@
  LDAPConnection(final org.glassfish.grizzly.Connection<?> connection,
      final LDAPOptions options)
  {
    // FIXME: remove synchronization when OPENDJ-422 is resolved.
    this.connection = synchronizeConnection(connection);
    this.connection = connection;
    this.options = options;
  }

@@ -255,9 +252,8 @@
    {
      context = request
          .createBindClient(connection.getPeerAddress() instanceof InetSocketAddress ?
              ((InetSocketAddress) connection
              .getPeerAddress()).getHostName() : connection.getPeerAddress()
              .toString());
              ((InetSocketAddress) connection.getPeerAddress()).getHostName()
              : connection.getPeerAddress().toString());
    }
    catch (final Exception e)
    {
@@ -567,7 +563,10 @@
   */
  public boolean isClosed()
  {
    return isClosed;
    synchronized (stateLock)
    {
      return isClosed;
    }
  }


@@ -577,7 +576,10 @@
   */
  public boolean isValid()
  {
    return connectionInvalidReason == null && !isClosed;
    synchronized (stateLock)
    {
      return connectionInvalidReason == null && !isClosed;
    }
  }


@@ -784,7 +786,7 @@



  int addPendingRequest(final AbstractLDAPFutureResultImpl<?> request)
  int continuePendingBindRequest(final LDAPBindFutureResultImpl future)
      throws ErrorResultException
  {
    final int newMsgID = nextMsgID.getAndIncrement();
@@ -794,7 +796,7 @@
      {
        throw newErrorResult(connectionInvalidReason);
      }
      pendingRequests.put(newMsgID, request);
      pendingRequests.put(newMsgID, future);
    }
    return newMsgID;
  }
@@ -969,9 +971,9 @@



  synchronized void handleUnsolicitedNotification(final ExtendedResult result)
  void handleUnsolicitedNotification(final ExtendedResult result)
  {
    if (isClosed)
    if (isClosed())
    {
      // Don't notify after connection is closed.
      return;
@@ -985,17 +987,41 @@



  /**
   * Installs a new Grizzly filter (e.g. SSL/SASL) beneath the top-level LDAP
   * filter.
   *
   * @param filter
   *          The filter to be installed.
   */
  void installFilter(final Filter filter)
  {
    if (customFilterChain == null)
    synchronized (stateLock)
    {
      customFilterChain = new DefaultFilterChain(
          (FilterChain) connection.getProcessor());
      connection.setProcessor(customFilterChain);
    }
      // Determine the index where the filter should be added.
      FilterChain oldFilterChain = (FilterChain) connection.getProcessor();
      int filterIndex = oldFilterChain.size() - 1;
      if (filter instanceof SSLFilter)
      {
        // Beneath any ConnectionSecurityLayerFilters if present, otherwise
        // beneath the LDAP filter.
        for (int i = oldFilterChain.size() - 2; i >= 0; i--)
        {
          if (!(oldFilterChain.get(i) instanceof ConnectionSecurityLayerFilter))
          {
            filterIndex = i + 1;
            break;
          }
        }
      }

    // Install the SSLFilter in the custom filter chain
    customFilterChain.add(customFilterChain.size() - 1, filter);
      // Create the new filter chain.
      FilterChain newFilterChain = FilterChainBuilder.stateless()
          .addAll(oldFilterChain)
          .add(filterIndex, filter)
          .build();
      connection.setProcessor(newFilterChain);
    }
  }


@@ -1008,9 +1034,19 @@
   */
  boolean isTLSEnabled()
  {
    final FilterChain currentFilterChain = (FilterChain) connection
        .getProcessor();
    return currentFilterChain.get(currentFilterChain.size() - 2) instanceof SSLFilter;
    synchronized (stateLock)
    {
      final FilterChain currentFilterChain = (FilterChain) connection
          .getProcessor();
      for (Filter filter : currentFilterChain)
      {
        if (filter instanceof SSLFilter)
        {
          return true;
        }
      }
      return false;
    }
  }


@@ -1029,28 +1065,28 @@



  synchronized void startTLS(final SSLContext sslContext,
  void startTLS(final SSLContext sslContext,
      final List<String> protocols, final List<String> cipherSuites,
      final CompletionHandler<SSLEngine> completionHandler) throws IOException
  {
    if (isTLSEnabled())
    synchronized (stateLock)
    {
      return;
      if (isTLSEnabled())
      {
        throw new IllegalStateException("TLS already enabled");
      }

      SSLEngineConfigurator sslEngineConfigurator = new SSLEngineConfigurator(
          sslContext, true, false, false);
      sslEngineConfigurator.setEnabledProtocols(protocols.isEmpty() ? null
          : protocols.toArray(new String[protocols.size()]));
      sslEngineConfigurator
          .setEnabledCipherSuites(cipherSuites.isEmpty() ? null : cipherSuites
              .toArray(new String[cipherSuites.size()]));
      SSLFilter sslFilter = new SSLFilter(null, sslEngineConfigurator);
      installFilter(sslFilter);
      sslFilter.handshake(connection, completionHandler);
    }

    SSLFilter sslFilter;
    SSLEngineConfigurator sslEngineConfigurator;

    sslEngineConfigurator = new SSLEngineConfigurator(sslContext, true, false,
        false);
    sslEngineConfigurator.setEnabledProtocols(protocols.isEmpty() ? null
        : protocols.toArray(new String[protocols.size()]));
    sslEngineConfigurator.setEnabledCipherSuites(cipherSuites.isEmpty() ? null
        : cipherSuites.toArray(new String[cipherSuites.size()]));
    sslFilter = new SSLFilter(null, sslEngineConfigurator);
    installFilter(sslFilter);
    sslFilter.handshake(connection.getUnsynchronizedConnection(),
        completionHandler);
  }



 opendj3/opendj-ldap-sdk/src/main/java/com/forgerock/opendj/ldap/LDAPConnectionFactoryImpl.java

@@ -45,8 +45,8 @@
import org.forgerock.opendj.ldap.responses.Result;
import org.glassfish.grizzly.CompletionHandler;
import org.glassfish.grizzly.EmptyCompletionHandler;
import org.glassfish.grizzly.filterchain.DefaultFilterChain;
import org.glassfish.grizzly.filterchain.FilterChain;
import org.glassfish.grizzly.filterchain.FilterChainBuilder;
import org.glassfish.grizzly.filterchain.TransportFilter;
import org.glassfish.grizzly.nio.transport.TCPNIOTransport;

@@ -261,9 +261,10 @@
    this.options = new LDAPOptions(options);
    this.clientFilter = new LDAPClientFilter(new LDAPReader(
        this.options.getDecodeOptions()), 0);
    this.defaultFilterChain = new DefaultFilterChain();
    this.defaultFilterChain.add(new TransportFilter());
    this.defaultFilterChain.add(clientFilter);
    this.defaultFilterChain = FilterChainBuilder.stateless()
        .add(new TransportFilter())
        .add(clientFilter)
        .build();
  }



 opendj3/opendj-ldap-sdk/src/main/java/com/forgerock/opendj/ldap/LDAPListenerImpl.java

@@ -39,8 +39,8 @@
import org.forgerock.opendj.ldap.LDAPClientContext;
import org.forgerock.opendj.ldap.LDAPListenerOptions;
import org.forgerock.opendj.ldap.ServerConnectionFactory;
import org.glassfish.grizzly.filterchain.DefaultFilterChain;
import org.glassfish.grizzly.filterchain.FilterChain;
import org.glassfish.grizzly.filterchain.FilterChainBuilder;
import org.glassfish.grizzly.filterchain.TransportFilter;
import org.glassfish.grizzly.nio.transport.TCPNIOServerConnection;
import org.glassfish.grizzly.nio.transport.TCPNIOTransport;
@@ -89,10 +89,14 @@
      this.transport = options.getTCPNIOTransport();
    }
    this.connectionFactory = factory;
    this.defaultFilterChain = new DefaultFilterChain();
    this.defaultFilterChain.add(new TransportFilter());
    this.defaultFilterChain.add(new LDAPServerFilter(this, new LDAPReader(
        new DecodeOptions(options.getDecodeOptions())), 0));

    final DecodeOptions decodeOptions = new DecodeOptions(options
        .getDecodeOptions());
    this.defaultFilterChain = FilterChainBuilder
        .stateless()
        .add(new TransportFilter())
        .add(new LDAPServerFilter(this, new LDAPReader(decodeOptions), 0))
        .build();

    this.serverConnection = transport.bind(address, options.getBacklog());
    this.serverConnection.setProcessor(defaultFilterChain);

 opendj3/opendj-ldap-sdk/src/main/java/com/forgerock/opendj/ldap/LDAPServerFilter.java

@@ -31,7 +31,6 @@


import static com.forgerock.opendj.ldap.LDAPConstants.OID_NOTICE_OF_DISCONNECTION;
import static com.forgerock.opendj.ldap.SynchronizedConnection.synchronizeConnection;

import java.io.IOException;
import java.net.InetSocketAddress;
@@ -215,8 +214,7 @@

    private ClientContextImpl(final Connection<?> connection)
    {
      // FIXME: remove synchronization when OPENDJ-422 is resolved.
      this.connection = synchronizeConnection(connection);
      this.connection = connection;
    }


@@ -320,27 +318,36 @@
    public void enableConnectionSecurityLayer(
        final ConnectionSecurityLayer layer)
    {
      installFilter(connection, new ConnectionSecurityLayerFilter(layer,
          connection.getTransport().getMemoryManager()));
      synchronized (this)
      {
        installFilter(new ConnectionSecurityLayerFilter(layer, connection
            .getTransport().getMemoryManager()));
      }
    }



    @Override
    public void enableTLS(final SSLContext sslContext, final String[] protocols,
        final String[] suites, final boolean wantClientAuth,
        final boolean needClientAuth)
    public void enableTLS(final SSLContext sslContext,
        final String[] protocols, final String[] suites,
        final boolean wantClientAuth, final boolean needClientAuth)
    {
      Validator.ensureNotNull(sslContext);
      SSLEngineConfigurator sslEngineConfigurator;
      synchronized (this)
      {
        if (isTLSEnabled())
        {
          throw new IllegalStateException("TLS already enabled");
        }

      sslEngineConfigurator = new SSLEngineConfigurator(sslContext, false,
          false, false);
      sslEngineConfigurator.setEnabledCipherSuites(suites);
      sslEngineConfigurator.setEnabledProtocols(protocols);
      sslEngineConfigurator.setWantClientAuth(wantClientAuth);
      sslEngineConfigurator.setNeedClientAuth(needClientAuth);
      installFilter(connection, new SSLFilter(sslEngineConfigurator, null));
        SSLEngineConfigurator sslEngineConfigurator =
            new SSLEngineConfigurator(sslContext, false, false, false);
        sslEngineConfigurator.setEnabledCipherSuites(suites);
        sslEngineConfigurator.setEnabledProtocols(protocols);
        sslEngineConfigurator.setWantClientAuth(wantClientAuth);
        sslEngineConfigurator.setNeedClientAuth(needClientAuth);
        installFilter(new SSLFilter(sslEngineConfigurator, null));
      }
    }


@@ -383,9 +390,61 @@



    private Connection<?> getConnection()
    /**
     * Installs a new Grizzly filter (e.g. SSL/SASL) beneath the top-level LDAP
     * filter.
     *
     * @param filter
     *          The filter to be installed.
     */
    private void installFilter(final Filter filter)
    {
      return connection;
      // Determine the index where the filter should be added.
      final FilterChain oldFilterChain = (FilterChain) connection.getProcessor();
      int filterIndex = oldFilterChain.size() - 1;
      if (filter instanceof SSLFilter)
      {
        // Beneath any ConnectionSecurityLayerFilters if present, otherwise
        // beneath the LDAP filter.
        for (int i = oldFilterChain.size() - 2; i >= 0; i--)
        {
          if (!(oldFilterChain.get(i) instanceof ConnectionSecurityLayerFilter))
          {
            filterIndex = i + 1;
            break;
          }
        }
      }

      // Create the new filter chain.
      final FilterChain newFilterChain = FilterChainBuilder.stateless()
          .addAll(oldFilterChain).add(filterIndex, filter).build();
      connection.setProcessor(newFilterChain);
    }



    /**
     * Indicates whether or not TLS is enabled this provided connection.
     *
     * @return {@code true} if TLS is enabled on this connection, otherwise
     *         {@code false}.
     */
    private boolean isTLSEnabled()
    {
      synchronized (this)
      {
        final FilterChain currentFilterChain = (FilterChain) connection
            .getProcessor();
        for (Filter filter : currentFilterChain)
        {
          if (filter instanceof SSLFilter)
          {
            return true;
          }
        }
        return false;
      }
    }

  }
@@ -880,7 +939,7 @@
        final ServerConnection<Integer> conn = clientContext
            .getServerConnection();
        final AddHandler handler = new AddHandler(messageID,
            clientContext.getConnection());
            ctx.getConnection());
        conn.handleAdd(messageID, request, handler, handler);
      }
    }
@@ -899,7 +958,7 @@
        final ServerConnection<Integer> conn = clientContext
            .getServerConnection();
        final BindHandler handler = new BindHandler(messageID,
            clientContext.getConnection());
            ctx.getConnection());
        conn.handleBind(messageID, version, bindContext, handler, handler);
      }
    }
@@ -918,7 +977,7 @@
        final ServerConnection<Integer> conn = clientContext
            .getServerConnection();
        final CompareHandler handler = new CompareHandler(messageID,
            clientContext.getConnection());
            ctx.getConnection());
        conn.handleCompare(messageID, request, handler, handler);
      }
    }
@@ -937,7 +996,7 @@
        final ServerConnection<Integer> conn = clientContext
            .getServerConnection();
        final DeleteHandler handler = new DeleteHandler(messageID,
            clientContext.getConnection());
            ctx.getConnection());
        conn.handleDelete(messageID, request, handler, handler);
      }
    }
@@ -956,7 +1015,7 @@
        final ServerConnection<Integer> conn = clientContext
            .getServerConnection();
        final ExtendedHandler<R> handler = new ExtendedHandler<R>(messageID,
            clientContext.getConnection());
            ctx.getConnection());
        conn.handleExtendedRequest(messageID, request, handler, handler);
      }
    }
@@ -975,7 +1034,7 @@
        final ServerConnection<Integer> conn = clientContext
            .getServerConnection();
        final ModifyDNHandler handler = new ModifyDNHandler(messageID,
            clientContext.getConnection());
            ctx.getConnection());
        conn.handleModifyDN(messageID, request, handler, handler);
      }
    }
@@ -994,7 +1053,7 @@
        final ServerConnection<Integer> conn = clientContext
            .getServerConnection();
        final ModifyHandler handler = new ModifyHandler(messageID,
            clientContext.getConnection());
            ctx.getConnection());
        conn.handleModify(messageID, request, handler, handler);
      }
    }
@@ -1013,7 +1072,7 @@
        final ServerConnection<Integer> conn = clientContext
            .getServerConnection();
        final SearchHandler handler = new SearchHandler(messageID,
            clientContext.getConnection());
            ctx.getConnection());
        conn.handleSearch(messageID, request, handler, handler);
      }
    }
@@ -1128,52 +1187,4 @@

    return ctx.getStopAction();
  }



  private synchronized void installFilter(final Connection<?> connection,
      final org.glassfish.grizzly.filterchain.Filter filter)
  {
    FilterChain filterChain = (FilterChain) connection.getProcessor();

    // Ensure that the SSL filter is not installed twice.
    if (filter instanceof SSLFilter)
    {
      for (Filter f : filterChain)
      {
        if (f instanceof SSLFilter)
        {
          // SSLFilter already installed.
          throw new IllegalStateException("SSL already installed on connection");
        }
      }
    }

    // Copy on update the default filter chain.
    if (listener.getDefaultFilterChain() == filterChain)
    {
      filterChain = new DefaultFilterChain(filterChain);
      connection.setProcessor(filterChain);
    }

    // Ensure that the SSL filter is beneath any connection security layer
    // filters.
    if (filter instanceof SSLFilter)
    {
      for (int i = filterChain.size() - 1; i >= 0; i--)
      {
        if (!(filterChain.get(i) instanceof ConnectionSecurityLayerFilter))
        {
          filterChain.add(i, filter);
          break;
        }
      }
    }
    else
    {
      // Add connection security layers to the end of the chain.
      filterChain.add(filterChain.size() - 1, filter);
    }

  }
}

 opendj3/opendj-ldap-sdk/src/main/java/com/forgerock/opendj/ldap/SynchronizedConnection.java

File was deleted

 opendj3/pom.xml

@@ -826,7 +826,6 @@
        <enabled>false</enabled>
      </releases>
    </repository>
    <!--
    <repository>
      <id>jvnet-nexus-snapshots</id>
      <url>https://maven.java.net/content/repositories/snapshots</url>
@@ -837,7 +836,6 @@
        <enabled>true</enabled>
      </snapshots>
    </repository>
    -->
  </repositories>
  <dependencyManagement>
    <dependencies>