/* * 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 * * * Copyright 2006-2009 Sun Microsystems, Inc. * Portions Copyright 2011-2012 ForgeRock AS */ 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 static org.opends.server.util.StaticUtils.stackTraceToSingleLineString; import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; import java.net.Socket; import java.net.SocketException; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; import java.util.zip.DataFormatException; import javax.net.ssl.SSLSocket; import org.opends.server.loggers.debug.DebugTracer; /** * This class implements a protocol session using TLS. */ public final class TLSSocketSession implements ProtocolSession { /** * The tracer object for the debug logger. */ private static final DebugTracer TRACER = getTracer(); private final Socket plainSocket; private final SSLSocket secureSocket; private final InputStream plainInput; private final OutputStream plainOutput; private final byte[] rcvLengthBuf = new byte[8]; private final String readableRemoteAddress; private final String remoteAddress; /** * The time the last message published to this session. */ private volatile long lastPublishTime = 0; /** * The time the last message was received on this session. */ private volatile long lastReceiveTime = 0; // Close and error guarded by stateLock: use a different lock to publish since // publishing can block, and we don't want to block while closing failed // connections. private final Object stateLock = new Object(); private boolean closeInitiated = false; private Throwable sessionError = null; // Publish guarded by publishLock: use a full lock here so that we can // optionally publish StopMsg during close. private final Lock publishLock = new ReentrantLock(); // Does not need protecting: updated only during single threaded handshake. private short protocolVersion = ProtocolVersion.getCurrentVersion(); private InputStream input; private OutputStream output; /** * 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(final Socket socket, final SSLSocket secureSocket) throws IOException { if (debugEnabled()) { TRACER.debugInfo( "Creating TLSSocketSession from %s to %s in %s", socket.getLocalSocketAddress(), socket.getRemoteSocketAddress(), stackTraceToSingleLineString(new Exception())); } this.plainSocket = socket; this.secureSocket = secureSocket; this.plainInput = plainSocket.getInputStream(); this.plainOutput = plainSocket.getOutputStream(); this.input = secureSocket.getInputStream(); this.output = secureSocket.getOutputStream(); this.readableRemoteAddress = plainSocket.getRemoteSocketAddress() .toString(); this.remoteAddress = plainSocket.getInetAddress() .getHostAddress(); } /** * {@inheritDoc} */ @Override public void close() { Throwable localSessionError; synchronized (stateLock) { if (closeInitiated) { return; } localSessionError = sessionError; closeInitiated = true; } // Perform close outside of critical section. if (debugEnabled()) { if (localSessionError == null) { TRACER.debugInfo( "Closing TLSSocketSession from %s to %s in %s", plainSocket.getLocalSocketAddress(), plainSocket.getRemoteSocketAddress(), stackTraceToSingleLineString(new Exception())); } else { TRACER.debugInfo( "Aborting TLSSocketSession from %s to %s in %s due to the " + "following error: %s", plainSocket.getLocalSocketAddress(), plainSocket.getRemoteSocketAddress(), stackTraceToSingleLineString(new Exception()), stackTraceToSingleLineString(localSessionError)); } } // V4 protocol introduces a StopMsg to properly end communications. if (localSessionError == null) { if (protocolVersion >= ProtocolVersion.REPLICATION_PROTOCOL_V4) { if (publishLock.tryLock()) { try { publish(new StopMsg()); } catch (final IOException ignored) { // Ignore errors on close. } finally { publishLock.unlock(); } } } } try { plainSocket.close(); } catch (final IOException ignored) { // Ignore errors on close. } try { secureSocket.close(); } catch (final IOException ignored) { // Ignore errors on close. } } /** * {@inheritDoc} */ @Override public boolean closeInitiated() { synchronized (stateLock) { return closeInitiated; } } /** * {@inheritDoc} */ @Override public long getLastPublishTime() { return lastPublishTime; } /** * {@inheritDoc} */ @Override public long getLastReceiveTime() { if (lastReceiveTime == 0) { return System.currentTimeMillis(); } return lastReceiveTime; } /** * {@inheritDoc} */ @Override public String getReadableRemoteAddress() { return readableRemoteAddress; } /** * {@inheritDoc} */ @Override public String getRemoteAddress() { return remoteAddress; } /** * {@inheritDoc} */ @Override public boolean isEncrypted() { return input != plainInput; } /** * {@inheritDoc} */ @Override public void publish(final ReplicationMsg msg) throws IOException { publish(msg, ProtocolVersion.getCurrentVersion()); } /** * {@inheritDoc} */ @Override public void publish(final ReplicationMsg msg, final short reqProtocolVersion) throws IOException { final byte[] buffer = msg.getBytes(reqProtocolVersion); final String str = String.format("%08x", buffer.length); final byte[] sendLengthBuf = str.getBytes(); publishLock.lock(); try { output.write(sendLengthBuf); output.write(buffer); output.flush(); } catch (final IOException e) { setSessionError(e); throw e; } finally { publishLock.unlock(); } lastPublishTime = System.currentTimeMillis(); } /** * {@inheritDoc} */ @Override public ReplicationMsg receive() throws IOException, DataFormatException, NotSupportedOldVersionPDUException { try { // 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 operational. lastReceiveTime = System.currentTimeMillis(); while (length < 8) { final int read = input.read(rcvLengthBuf, length, 8 - length); if (read == -1) { lastReceiveTime = 0; throw new IOException("no more data"); } else { length += read; } } final int totalLength = Integer.parseInt(new String( rcvLengthBuf), 16); try { length = 0; final byte[] buffer = new byte[totalLength]; while (length < totalLength) { final int read = input.read(buffer, length, totalLength - length); if (read == -1) { lastReceiveTime = 0; throw new IOException("no more data"); } else { length += read; } } // We do not want the heartbeat to close the session when we are // processing a message even a time consuming one. lastReceiveTime = 0; return ReplicationMsg.generateMsg(buffer, protocolVersion); } catch (final OutOfMemoryError e) { throw new IOException("Packet too large, can't allocate " + totalLength + " bytes."); } } catch (final IOException e) { setSessionError(e); throw e; } catch (final DataFormatException e) { setSessionError(e); throw e; } catch (final NotSupportedOldVersionPDUException e) { setSessionError(e); throw e; } catch (final RuntimeException e) { setSessionError(e); throw e; } } /** * {@inheritDoc} */ @Override public void setProtocolVersion(final short version) { protocolVersion = version; } /** * {@inheritDoc} */ @Override public void setSoTimeout(final int timeout) throws SocketException { plainSocket.setSoTimeout(timeout); } /** * {@inheritDoc} */ @SuppressWarnings("unused") @Override public void stopEncryption() { // The secure socket has been configured not to auto close the underlying // plain socket. We should close it here and properly tear down the SSL // session, but this is not compatible with the existing protocol. if (false) { try { secureSocket.close(); } catch (IOException ignored) { // Ignore. } } input = plainInput; output = plainOutput; } private void setSessionError(final Exception e) { synchronized (stateLock) { if (sessionError == null) { sessionError = e; } } } }