| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF licenses this file to You under the Apache License, Version 2.0 |
| * (the "License"); you may not use this file except in compliance with |
| * the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| package org.apache.nifi.remote.io.socket.ssl; |
| |
| import org.apache.nifi.remote.exception.TransmissionDisabledException; |
| import org.apache.nifi.remote.io.socket.BufferStateManager; |
| import org.apache.nifi.remote.io.socket.BufferStateManager.Direction; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import javax.net.ssl.SSLContext; |
| import javax.net.ssl.SSLEngine; |
| import javax.net.ssl.SSLEngineResult; |
| import javax.net.ssl.SSLEngineResult.Status; |
| import javax.net.ssl.SSLException; |
| import javax.net.ssl.SSLHandshakeException; |
| import javax.net.ssl.SSLSession; |
| import java.io.Closeable; |
| import java.io.IOException; |
| import java.net.InetAddress; |
| import java.net.InetSocketAddress; |
| import java.net.Socket; |
| import java.net.SocketAddress; |
| import java.net.SocketTimeoutException; |
| import java.nio.ByteBuffer; |
| import java.nio.channels.ClosedByInterruptException; |
| import java.nio.channels.SocketChannel; |
| import java.util.concurrent.TimeUnit; |
| |
| /** |
| * SSLSocketChannel supports reading and writing bytes using TLS and NIO SocketChannels with configurable timeouts |
| */ |
| public class SSLSocketChannel implements Closeable { |
| private static final Logger LOGGER = LoggerFactory.getLogger(SSLSocketChannel.class); |
| |
| private static final int DISCARD_BUFFER_LENGTH = 8192; |
| private static final int END_OF_STREAM = -1; |
| private static final byte[] EMPTY_MESSAGE = new byte[0]; |
| private static final long BUFFER_FULL_EMPTY_WAIT_NANOS = TimeUnit.NANOSECONDS.convert(1, TimeUnit.MILLISECONDS); |
| private static final long FINISH_CONNECT_SLEEP = 50; |
| private static final long INITIAL_INCREMENTAL_SLEEP = 1; |
| private static final boolean CLIENT_AUTHENTICATION_REQUIRED = true; |
| |
| private final String remoteAddress; |
| private final int port; |
| private final SSLEngine engine; |
| private final SocketAddress socketAddress; |
| private final BufferStateManager streamInManager; |
| private final BufferStateManager streamOutManager; |
| private final BufferStateManager appDataManager; |
| private final SocketChannel channel; |
| private int timeoutMillis = 30000; |
| |
| private volatile boolean interrupted = false; |
| private volatile ChannelStatus channelStatus = ChannelStatus.DISCONNECTED; |
| |
| /** |
| * SSLSocketChannel constructor with SSLContext and remote address parameters |
| * |
| * @param sslContext SSLContext used to create SSLEngine with specified client mode |
| * @param remoteAddress Remote Address used for connection |
| * @param port Remote Port used for connection |
| * @param bindAddress Local address used for binding server channel when provided |
| * @param useClientMode Use Client Mode |
| * @throws IOException Thrown on failures creating Socket Channel |
| */ |
| public SSLSocketChannel(final SSLContext sslContext, final String remoteAddress, final int port, final InetAddress bindAddress, final boolean useClientMode) throws IOException { |
| this.engine = createEngine(sslContext, useClientMode); |
| this.channel = createSocketChannel(bindAddress); |
| this.socketAddress = new InetSocketAddress(remoteAddress, port); |
| this.remoteAddress = remoteAddress; |
| this.port = port; |
| |
| streamInManager = new BufferStateManager(ByteBuffer.allocate(engine.getSession().getPacketBufferSize())); |
| streamOutManager = new BufferStateManager(ByteBuffer.allocate(engine.getSession().getPacketBufferSize())); |
| appDataManager = new BufferStateManager(ByteBuffer.allocate(engine.getSession().getApplicationBufferSize())); |
| } |
| |
| /** |
| * SSLSocketChannel constructor with SSLContext and connected SocketChannel |
| * |
| * @param sslContext SSLContext used to create SSLEngine with specified client mode |
| * @param socketChannel Connected SocketChannel |
| * @param useClientMode Use Client Mode |
| * @throws IOException Thrown on SocketChannel.getRemoteAddress() |
| */ |
| public SSLSocketChannel(final SSLContext sslContext, final SocketChannel socketChannel, final boolean useClientMode) throws IOException { |
| this(createEngine(sslContext, useClientMode), socketChannel); |
| } |
| |
| /** |
| * SSLSocketChannel constructor with configured SSLEngine and connected SocketChannel |
| * |
| * @param sslEngine SSLEngine configured with mode and client authentication |
| * @param socketChannel Connected SocketChannel |
| * @throws IOException Thrown on SocketChannel.getRemoteAddress() |
| */ |
| public SSLSocketChannel(final SSLEngine sslEngine, final SocketChannel socketChannel) throws IOException { |
| if (!socketChannel.isConnected()) { |
| throw new IllegalArgumentException("Connected SocketChannel required"); |
| } |
| |
| socketChannel.configureBlocking(false); |
| this.channel = socketChannel; |
| this.socketAddress = socketChannel.getRemoteAddress(); |
| final Socket socket = socketChannel.socket(); |
| this.remoteAddress = socket.getInetAddress().toString(); |
| this.port = socket.getPort(); |
| this.engine = sslEngine; |
| |
| streamInManager = new BufferStateManager(ByteBuffer.allocate(engine.getSession().getPacketBufferSize())); |
| streamOutManager = new BufferStateManager(ByteBuffer.allocate(engine.getSession().getPacketBufferSize())); |
| appDataManager = new BufferStateManager(ByteBuffer.allocate(engine.getSession().getApplicationBufferSize())); |
| } |
| |
| public void setTimeout(final int timeoutMillis) { |
| this.timeoutMillis = timeoutMillis; |
| } |
| |
| public int getTimeout() { |
| return timeoutMillis; |
| } |
| |
| /** |
| * Connect Channel when not connected and perform handshake process |
| * |
| * @throws IOException Thrown on connection failures |
| */ |
| public void connect() throws IOException { |
| channelStatus = ChannelStatus.CONNECTING; |
| |
| try { |
| if (!channel.isConnected()) { |
| logOperation("Connection Started"); |
| final long started = System.currentTimeMillis(); |
| |
| if (!channel.connect(socketAddress)) { |
| while (!channel.finishConnect()) { |
| checkInterrupted(); |
| checkTimeoutExceeded(started); |
| |
| try { |
| TimeUnit.MILLISECONDS.sleep(FINISH_CONNECT_SLEEP); |
| } catch (final InterruptedException e) { |
| logOperation("Connection Interrupted"); |
| } |
| } |
| } |
| } |
| channelStatus = ChannelStatus.CONNECTED; |
| } catch (final Exception e) { |
| close(); |
| throw new SSLException(String.format("[%s:%d] Connection Failed", remoteAddress, port), e); |
| } |
| |
| try { |
| performHandshake(); |
| } catch (final IOException e) { |
| close(); |
| throw new SSLException(String.format("[%s:%d] Handshake Failed", remoteAddress, port), e); |
| } |
| } |
| |
| /** |
| * Shutdown Socket Channel input and read available bytes |
| * |
| * @throws IOException Thrown on Socket Channel failures |
| */ |
| public void consume() throws IOException { |
| channel.shutdownInput(); |
| |
| final byte[] byteBuffer = new byte[DISCARD_BUFFER_LENGTH]; |
| final ByteBuffer buffer = ByteBuffer.wrap(byteBuffer); |
| int readCount; |
| do { |
| readCount = channel.read(buffer); |
| buffer.flip(); |
| } while (readCount > 0); |
| } |
| |
| /** |
| * Is Channel Closed |
| * |
| * @return Channel Closed Status |
| */ |
| public boolean isClosed() { |
| if (ChannelStatus.CLOSED == channelStatus) { |
| return true; |
| } |
| |
| // Read Channel to determine closed status |
| final ByteBuffer inputBuffer = streamInManager.prepareForWrite(engine.getSession().getPacketBufferSize()); |
| int bytesRead; |
| try { |
| bytesRead = channel.read(inputBuffer); |
| } catch (final IOException e) { |
| LOGGER.warn("[{}:{}] Closed Status Read Failed", remoteAddress, port, e); |
| bytesRead = END_OF_STREAM; |
| } |
| logOperationBytes("Closed Status Read", bytesRead); |
| |
| if (bytesRead == 0) { |
| return false; |
| } else if (bytesRead > 0) { |
| try { |
| final SSLEngineResult unwrapResult = unwrap(); |
| if (Status.CLOSED == unwrapResult.getStatus()) { |
| readChannelDiscard(); |
| engine.closeInbound(); |
| } else { |
| streamInManager.compact(); |
| return false; |
| } |
| } catch (final IOException e) { |
| LOGGER.warn("[{}:{}] Closed Status Unwrap Failed", remoteAddress, port, e); |
| } |
| } |
| |
| // Close Channel when encountering end of stream or closed status |
| try { |
| close(); |
| } catch (final IOException e) { |
| LOGGER.warn("[{}:{}] Close Failed", remoteAddress, port, e); |
| } |
| return true; |
| } |
| |
| /** |
| * Close Channel and process notifications |
| * |
| * @throws IOException Thrown on SSLEngine.wrap() failures |
| */ |
| @Override |
| public void close() throws IOException { |
| logOperation("Close Requested"); |
| if (channelStatus == ChannelStatus.CLOSED) { |
| return; |
| } |
| |
| try { |
| engine.closeOutbound(); |
| |
| streamOutManager.clear(); |
| final ByteBuffer inputBuffer = ByteBuffer.wrap(EMPTY_MESSAGE); |
| final ByteBuffer outputBuffer = streamOutManager.prepareForWrite(engine.getSession().getApplicationBufferSize()); |
| SSLEngineResult wrapResult = wrap(inputBuffer, outputBuffer); |
| Status status = wrapResult.getStatus(); |
| if (Status.OK == status) { |
| logOperation("Clearing Outbound Buffer"); |
| outputBuffer.clear(); |
| wrapResult = wrap(inputBuffer, outputBuffer); |
| status = wrapResult.getStatus(); |
| } |
| if (Status.CLOSED == status) { |
| final ByteBuffer streamOutputBuffer = streamOutManager.prepareForRead(1); |
| try { |
| writeChannel(streamOutputBuffer); |
| } catch (final IOException e) { |
| logOperation(String.format("Write Close Notification Failed: %s", e.getMessage())); |
| } |
| } else { |
| throw new SSLException(String.format("[%s:%d] Invalid Wrap Result Status [%s]", remoteAddress, port, status)); |
| } |
| } finally { |
| channelStatus = ChannelStatus.CLOSED; |
| readChannelDiscard(); |
| closeQuietly(channel.socket()); |
| closeQuietly(channel); |
| logOperation("Close Completed"); |
| } |
| } |
| |
| /** |
| * Get application bytes available for reading |
| * |
| * @return Number of application bytes available for reading |
| * @throws IOException Thrown on failures checking for available bytes |
| */ |
| public int available() throws IOException { |
| ByteBuffer appDataBuffer = appDataManager.prepareForRead(1); |
| ByteBuffer streamDataBuffer = streamInManager.prepareForRead(1); |
| final int buffered = appDataBuffer.remaining() + streamDataBuffer.remaining(); |
| if (buffered > 0) { |
| return buffered; |
| } |
| |
| if (!isDataAvailable()) { |
| return 0; |
| } |
| |
| appDataBuffer = appDataManager.prepareForRead(1); |
| streamDataBuffer = streamInManager.prepareForRead(1); |
| return appDataBuffer.remaining() + streamDataBuffer.remaining(); |
| } |
| |
| /** |
| * Is data available for reading |
| * |
| * @return Data available status |
| * @throws IOException Thrown on SocketChannel.read() failures |
| */ |
| public boolean isDataAvailable() throws IOException { |
| final ByteBuffer appDataBuffer = appDataManager.prepareForRead(1); |
| final ByteBuffer streamDataBuffer = streamInManager.prepareForRead(1); |
| |
| if (appDataBuffer.remaining() > 0 || streamDataBuffer.remaining() > 0) { |
| return true; |
| } |
| |
| final ByteBuffer writableBuffer = streamInManager.prepareForWrite(engine.getSession().getPacketBufferSize()); |
| final int bytesRead = channel.read(writableBuffer); |
| return (bytesRead > 0); |
| } |
| |
| /** |
| * Read and return one byte |
| * |
| * @return Byte read or -1 when end of stream reached |
| * @throws IOException Thrown on read failures |
| */ |
| public int read() throws IOException { |
| final byte[] buffer = new byte[1]; |
| |
| final int bytesRead = read(buffer); |
| if (bytesRead == END_OF_STREAM) { |
| return END_OF_STREAM; |
| } |
| |
| return Byte.toUnsignedInt(buffer[0]); |
| } |
| |
| /** |
| * Read available bytes into buffer |
| * |
| * @param buffer Byte array buffer |
| * @return Number of bytes read |
| * @throws IOException Thrown on read failures |
| */ |
| public int read(final byte[] buffer) throws IOException { |
| return read(buffer, 0, buffer.length); |
| } |
| |
| /** |
| * Read available bytes into buffer based on offset and length requested |
| * |
| * @param buffer Byte array buffer |
| * @param offset Buffer offset |
| * @param len Length of bytes to read |
| * @return Number of bytes read |
| * @throws IOException Thrown on read failures |
| */ |
| public int read(final byte[] buffer, final int offset, final int len) throws IOException { |
| logOperationBytes("Read Requested", len); |
| checkChannelStatus(); |
| |
| int applicationBytesRead = readApplicationBuffer(buffer, offset, len); |
| if (applicationBytesRead > 0) { |
| return applicationBytesRead; |
| } |
| appDataManager.clear(); |
| |
| while (true) { |
| final SSLEngineResult unwrapResult = unwrap(); |
| |
| if (SSLEngineResult.HandshakeStatus.FINISHED == unwrapResult.getHandshakeStatus()) { |
| // RFC 8446 Section 4.6 describes Post-Handshake Messages for TLS 1.3 |
| logOperation("Processing Post-Handshake Messages"); |
| continue; |
| } |
| |
| final Status status = unwrapResult.getStatus(); |
| switch (status) { |
| case BUFFER_OVERFLOW: |
| throw new IllegalStateException(String.format("SSLEngineResult Status [%s] not allowed from unwrap", status)); |
| case BUFFER_UNDERFLOW: |
| final ByteBuffer streamBuffer = streamInManager.prepareForWrite(engine.getSession().getPacketBufferSize()); |
| final int channelBytesRead = readChannel(streamBuffer); |
| logOperationBytes("Channel Read Completed", channelBytesRead); |
| if (channelBytesRead == END_OF_STREAM) { |
| return END_OF_STREAM; |
| } |
| break; |
| case CLOSED: |
| applicationBytesRead = readApplicationBuffer(buffer, offset, len); |
| if (applicationBytesRead == 0) { |
| return END_OF_STREAM; |
| } |
| streamInManager.compact(); |
| return applicationBytesRead; |
| case OK: |
| applicationBytesRead = readApplicationBuffer(buffer, offset, len); |
| if (applicationBytesRead == 0) { |
| throw new IOException("Read Application Buffer Failed"); |
| } |
| streamInManager.compact(); |
| return applicationBytesRead; |
| } |
| } |
| } |
| |
| /** |
| * Write one byte to channel |
| * |
| * @param data Byte to be written |
| * @throws IOException Thrown on write failures |
| */ |
| public void write(final int data) throws IOException { |
| write(new byte[]{(byte) data}, 0, 1); |
| } |
| |
| /** |
| * Write bytes to channel |
| * |
| * @param data Byte array to be written |
| * @throws IOException Thrown on write failures |
| */ |
| public void write(final byte[] data) throws IOException { |
| write(data, 0, data.length); |
| } |
| |
| /** |
| * Write data to channel performs multiple iterations based on data length |
| * |
| * @param data Byte array to be written |
| * @param offset Byte array offset |
| * @param len Length of bytes for writing |
| * @throws IOException Thrown on write failures |
| */ |
| public void write(final byte[] data, final int offset, final int len) throws IOException { |
| logOperationBytes("Write Started", len); |
| checkChannelStatus(); |
| |
| final int applicationBufferSize = engine.getSession().getApplicationBufferSize(); |
| logOperationBytes("Write Application Buffer Size", applicationBufferSize); |
| int iterations = len / applicationBufferSize; |
| if (len % applicationBufferSize > 0) { |
| iterations++; |
| } |
| |
| for (int i = 0; i < iterations; i++) { |
| streamOutManager.clear(); |
| final int itrOffset = offset + i * applicationBufferSize; |
| final int itrLen = Math.min(len - itrOffset, applicationBufferSize); |
| final ByteBuffer byteBuffer = ByteBuffer.wrap(data, itrOffset, itrLen); |
| |
| final BufferStateManager bufferStateManager = new BufferStateManager(byteBuffer, Direction.READ); |
| final Status status = wrapWriteChannel(bufferStateManager); |
| switch (status) { |
| case BUFFER_OVERFLOW: |
| streamOutManager.ensureSize(engine.getSession().getPacketBufferSize()); |
| appDataManager.ensureSize(engine.getSession().getApplicationBufferSize()); |
| continue; |
| case OK: |
| continue; |
| case CLOSED: |
| throw new IOException("Channel is closed"); |
| case BUFFER_UNDERFLOW: |
| throw new AssertionError("Got Buffer Underflow but should not have..."); |
| } |
| } |
| } |
| |
| /** |
| * Interrupt processing and disable transmission |
| */ |
| public void interrupt() { |
| this.interrupted = true; |
| } |
| |
| private void performHandshake() throws IOException { |
| logOperation("Handshake Started"); |
| channelStatus = ChannelStatus.HANDSHAKING; |
| engine.beginHandshake(); |
| |
| SSLEngineResult.HandshakeStatus handshakeStatus = engine.getHandshakeStatus(); |
| while (true) { |
| logHandshakeStatus(handshakeStatus); |
| |
| switch (handshakeStatus) { |
| case FINISHED: |
| case NOT_HANDSHAKING: |
| channelStatus = ChannelStatus.ESTABLISHED; |
| final SSLSession session = engine.getSession(); |
| LOGGER.debug("[{}:{}] [{}] Negotiated Protocol [{}] Cipher Suite [{}]", |
| remoteAddress, |
| port, |
| channelStatus, |
| session.getProtocol(), |
| session.getCipherSuite() |
| ); |
| return; |
| case NEED_TASK: |
| runDelegatedTasks(); |
| handshakeStatus = engine.getHandshakeStatus(); |
| break; |
| case NEED_UNWRAP: |
| final SSLEngineResult unwrapResult = unwrap(); |
| handshakeStatus = unwrapResult.getHandshakeStatus(); |
| Status unwrapResultStatus = unwrapResult.getStatus(); |
| |
| if (unwrapResultStatus == Status.BUFFER_UNDERFLOW) { |
| final ByteBuffer writableDataIn = streamInManager.prepareForWrite(engine.getSession().getPacketBufferSize()); |
| final int bytesRead = readChannel(writableDataIn); |
| logOperationBytes("Handshake Channel Read", bytesRead); |
| |
| if (bytesRead == END_OF_STREAM) { |
| throw getHandshakeException(handshakeStatus, "End of Stream Found"); |
| } |
| } else if (unwrapResultStatus == Status.CLOSED) { |
| throw getHandshakeException(handshakeStatus, "Channel Closed"); |
| } else { |
| streamInManager.compact(); |
| appDataManager.clear(); |
| } |
| break; |
| case NEED_WRAP: |
| final ByteBuffer outboundBuffer = streamOutManager.prepareForWrite(engine.getSession().getApplicationBufferSize()); |
| final SSLEngineResult wrapResult = wrap(ByteBuffer.wrap(EMPTY_MESSAGE), outboundBuffer); |
| handshakeStatus = wrapResult.getHandshakeStatus(); |
| final Status wrapResultStatus = wrapResult.getStatus(); |
| |
| if (wrapResultStatus == Status.BUFFER_OVERFLOW) { |
| streamOutManager.prepareForWrite(engine.getSession().getApplicationBufferSize()); |
| } else if (wrapResultStatus == Status.OK) { |
| final ByteBuffer streamBuffer = streamOutManager.prepareForRead(1); |
| final int bytesRemaining = streamBuffer.remaining(); |
| writeChannel(streamBuffer); |
| logOperationBytes("Handshake Channel Write Completed", bytesRemaining); |
| streamOutManager.clear(); |
| } else { |
| throw getHandshakeException(handshakeStatus, String.format("Wrap Failed [%s]", wrapResult.getStatus())); |
| } |
| break; |
| } |
| } |
| } |
| |
| private int readChannel(final ByteBuffer outputBuffer) throws IOException { |
| logOperation("Channel Read Started"); |
| |
| final long started = System.currentTimeMillis(); |
| long sleepNanoseconds = INITIAL_INCREMENTAL_SLEEP; |
| while (true) { |
| checkInterrupted(); |
| |
| if (outputBuffer.remaining() == 0) { |
| return 0; |
| } |
| |
| final int channelBytesRead = channel.read(outputBuffer); |
| if (channelBytesRead == 0) { |
| checkTimeoutExceeded(started); |
| sleepNanoseconds = incrementalSleep(sleepNanoseconds); |
| continue; |
| } |
| |
| return channelBytesRead; |
| } |
| } |
| |
| private void writeChannel(final ByteBuffer inputBuffer) throws IOException { |
| long lastWriteCompleted = System.currentTimeMillis(); |
| |
| int totalBytes = 0; |
| long sleepNanoseconds = INITIAL_INCREMENTAL_SLEEP; |
| while (inputBuffer.hasRemaining()) { |
| checkInterrupted(); |
| |
| final int written = channel.write(inputBuffer); |
| totalBytes += written; |
| |
| if (written > 0) { |
| lastWriteCompleted = System.currentTimeMillis(); |
| } else { |
| checkTimeoutExceeded(lastWriteCompleted); |
| sleepNanoseconds = incrementalSleep(sleepNanoseconds); |
| } |
| } |
| |
| logOperationBytes("Channel Write Completed", totalBytes); |
| } |
| |
| private long incrementalSleep(final long nanoseconds) throws IOException { |
| try { |
| TimeUnit.NANOSECONDS.sleep(nanoseconds); |
| } catch (final InterruptedException e) { |
| close(); |
| Thread.currentThread().interrupt(); |
| throw new ClosedByInterruptException(); |
| } |
| return Math.min(nanoseconds * 2, BUFFER_FULL_EMPTY_WAIT_NANOS); |
| } |
| |
| private void readChannelDiscard() { |
| try { |
| final ByteBuffer readBuffer = ByteBuffer.allocate(DISCARD_BUFFER_LENGTH); |
| int bytesRead = channel.read(readBuffer); |
| while (bytesRead > 0) { |
| readBuffer.clear(); |
| bytesRead = channel.read(readBuffer); |
| } |
| } catch (final IOException e) { |
| LOGGER.debug("[{}:{}] Read Channel Discard Failed", remoteAddress, port, e); |
| } |
| } |
| |
| private int readApplicationBuffer(final byte[] buffer, final int offset, final int len) { |
| logOperationBytes("Application Buffer Read Requested", len); |
| final ByteBuffer appDataBuffer = appDataManager.prepareForRead(len); |
| |
| final int appDataRemaining = appDataBuffer.remaining(); |
| logOperationBytes("Application Buffer Remaining", appDataRemaining); |
| if (appDataRemaining > 0) { |
| final int bytesToCopy = Math.min(len, appDataBuffer.remaining()); |
| appDataBuffer.get(buffer, offset, bytesToCopy); |
| |
| final int bytesCopied = appDataRemaining - appDataBuffer.remaining(); |
| logOperationBytes("Application Buffer Copied", bytesCopied); |
| return bytesCopied; |
| } |
| return 0; |
| } |
| |
| private Status wrapWriteChannel(final BufferStateManager inputManager) throws IOException { |
| final ByteBuffer inputBuffer = inputManager.prepareForRead(0); |
| final ByteBuffer outputBuffer = streamOutManager.prepareForWrite(engine.getSession().getApplicationBufferSize()); |
| |
| logOperationBytes("Wrap Started", inputBuffer.remaining()); |
| Status status = Status.OK; |
| while (inputBuffer.remaining() > 0) { |
| final SSLEngineResult result = wrap(inputBuffer, outputBuffer); |
| status = result.getStatus(); |
| if (status == Status.OK) { |
| final ByteBuffer readableOutBuff = streamOutManager.prepareForRead(0); |
| writeChannel(readableOutBuff); |
| streamOutManager.clear(); |
| } else { |
| break; |
| } |
| } |
| |
| return status; |
| } |
| |
| private SSLEngineResult wrap(final ByteBuffer inputBuffer, final ByteBuffer outputBuffer) throws SSLException { |
| final SSLEngineResult result = engine.wrap(inputBuffer, outputBuffer); |
| logEngineResult(result, "WRAP Completed"); |
| return result; |
| } |
| |
| private SSLEngineResult unwrap() throws IOException { |
| final ByteBuffer streamBuffer = streamInManager.prepareForRead(engine.getSession().getPacketBufferSize()); |
| final ByteBuffer applicationBuffer = appDataManager.prepareForWrite(engine.getSession().getApplicationBufferSize()); |
| final SSLEngineResult result = engine.unwrap(streamBuffer, applicationBuffer); |
| logEngineResult(result, "UNWRAP Completed"); |
| return result; |
| } |
| |
| private void runDelegatedTasks() { |
| Runnable delegatedTask; |
| while ((delegatedTask = engine.getDelegatedTask()) != null) { |
| logOperation("Running Delegated Task"); |
| delegatedTask.run(); |
| } |
| } |
| |
| private void closeQuietly(final Closeable closeable) { |
| try { |
| closeable.close(); |
| } catch (final Exception e) { |
| logOperation(String.format("Close failed: %s", e.getMessage())); |
| } |
| } |
| |
| private SSLHandshakeException getHandshakeException(final SSLEngineResult.HandshakeStatus handshakeStatus, final String message) { |
| final String formatted = String.format("[%s:%d] Handshake Status [%s] %s", remoteAddress, port, handshakeStatus, message); |
| return new SSLHandshakeException(formatted); |
| } |
| |
| private void checkChannelStatus() throws IOException { |
| if (ChannelStatus.ESTABLISHED != channelStatus) { |
| connect(); |
| } |
| } |
| |
| private void checkInterrupted() { |
| if (interrupted) { |
| throw new TransmissionDisabledException(); |
| } |
| } |
| |
| private void checkTimeoutExceeded(final long started) throws SocketTimeoutException { |
| if (System.currentTimeMillis() > started + timeoutMillis) { |
| throw new SocketTimeoutException(String.format("Timeout Exceeded [%d ms] for [%s:%d]", timeoutMillis, remoteAddress, port)); |
| } |
| } |
| |
| private void logOperation(final String operation) { |
| LOGGER.trace("[{}:{}] [{}] {}", remoteAddress, port, channelStatus, operation); |
| } |
| |
| private void logOperationBytes(final String operation, final int bytes) { |
| LOGGER.trace("[{}:{}] [{}] {} Bytes [{}]", remoteAddress, port, channelStatus, operation, bytes); |
| } |
| |
| private void logHandshakeStatus(final SSLEngineResult.HandshakeStatus handshakeStatus) { |
| LOGGER.trace("[{}:{}] [{}] Handshake Status [{}]", remoteAddress, port, channelStatus, handshakeStatus); |
| } |
| |
| private void logEngineResult(final SSLEngineResult result, final String method) { |
| LOGGER.trace("[{}:{}] [{}] {} Status [{}] Handshake Status [{}] Produced [{}] Consumed [{}]", |
| remoteAddress, |
| port, |
| channelStatus, |
| method, |
| result.getStatus(), |
| result.getHandshakeStatus(), |
| result.bytesProduced(), |
| result.bytesConsumed() |
| ); |
| } |
| |
| private static SocketChannel createSocketChannel(final InetAddress bindAddress) throws IOException { |
| final SocketChannel socketChannel = SocketChannel.open(); |
| if (bindAddress != null) { |
| final SocketAddress socketAddress = new InetSocketAddress(bindAddress, 0); |
| socketChannel.bind(socketAddress); |
| } |
| socketChannel.configureBlocking(false); |
| return socketChannel; |
| } |
| |
| private static SSLEngine createEngine(final SSLContext sslContext, final boolean useClientMode) { |
| final SSLEngine sslEngine = sslContext.createSSLEngine(); |
| sslEngine.setUseClientMode(useClientMode); |
| sslEngine.setNeedClientAuth(CLIENT_AUTHENTICATION_REQUIRED); |
| return sslEngine; |
| } |
| |
| private enum ChannelStatus { |
| DISCONNECTED, |
| |
| CONNECTING, |
| |
| CONNECTED, |
| |
| HANDSHAKING, |
| |
| ESTABLISHED, |
| |
| CLOSED |
| } |
| } |
