diff --git a/src/main/java/io/mapsmessaging/network/io/impl/ssl/SSLEndPoint.java b/src/main/java/io/mapsmessaging/network/io/impl/ssl/SSLEndPoint.java index f1c23e6e5f..107bd0ba38 100644 --- a/src/main/java/io/mapsmessaging/network/io/impl/ssl/SSLEndPoint.java +++ b/src/main/java/io/mapsmessaging/network/io/impl/ssl/SSLEndPoint.java @@ -245,13 +245,41 @@ protected int readBuffer(ByteBuffer applicationIn) throws IOException { if (result.getStatus() == Status.BUFFER_UNDERFLOW) { encryptedIn.compact(); + if (!encryptedIn.hasRemaining()) { + // The inbound buffer is full yet the engine still needs more bytes to + // decode a record: we can neither read more (no space) nor make + // progress. Left alone this spins the reader at 100% CPU and the + // socket never drains. Tear the endpoint down instead - a reconnect + // re-handshakes with a fresh buffer. + throw new IOException("SSL inbound buffer full on BUFFER_UNDERFLOW; record exceeds " + + encryptedIn.capacity() + " byte buffer, closing endpoint"); + } return response; } + if (result.getStatus() == Status.BUFFER_OVERFLOW) { + if (applicationIn.position() > 0) { + break; // hand back what has been decoded; the caller drains it and reads again + } + // Not even one record fits in an empty buffer, so it can never be decoded: nothing + // leaves encryptedIn, the socket stops draining and the peer stalls. Fail loudly. + throw new IOException("SSL read buffer of " + applicationIn.capacity() + + " bytes cannot hold a TLS record of up to " + + sslEngine.getSession().getApplicationBufferSize() + " bytes, closing endpoint"); + } + SSLEngineResult.HandshakeStatus hs = result.getHandshakeStatus(); - if (hs == SSLEngineResult.HandshakeStatus.NEED_TASK - || hs == SSLEngineResult.HandshakeStatus.NEED_WRAP) { - break; // let handshake manager drive next step + if (hs == SSLEngineResult.HandshakeStatus.NEED_TASK) { + // After the handshake the manager is SSLHandshakeManagerFinished, which drives + // nothing, so a task requested later (e.g. by a TLS 1.3 key update) would never + // run. Run it here and retry the unwrap. + if (!runDelegatedTasks()) { + break; // nothing to run, do not spin on an unchanged status + } + continue; + } + if (hs == SSLEngineResult.HandshakeStatus.NEED_WRAP) { + break; // needs an outbound record, driven by the write path } boolean noProgress = encryptedIn.position() == beforeIn && applicationIn.position() == beforeOut; @@ -269,6 +297,21 @@ protected int readBuffer(ByteBuffer applicationIn) throws IOException { return response; } + /** + * Runs any tasks the engine has delegated, as the handshake manager does during negotiation. + * + * @return true if at least one task was run, false if the engine had none pending + */ + private boolean runDelegatedTasks() { + Runnable runnable; + boolean ranTask = false; + while ((runnable = sslEngine.getDelegatedTask()) != null) { + runnable.run(); + ranTask = true; + } + return ranTask; + } + private SSLEngineResult handleSSLEngineResult(SSLEngineResult result) throws IOException { if (result.getStatus() == Status.CLOSED) { throw new IOException("Session closed"); diff --git a/src/main/java/io/mapsmessaging/network/protocol/impl/cot/CotProtocol.java b/src/main/java/io/mapsmessaging/network/protocol/impl/cot/CotProtocol.java index 458ee79c4e..3d100ca29f 100644 --- a/src/main/java/io/mapsmessaging/network/protocol/impl/cot/CotProtocol.java +++ b/src/main/java/io/mapsmessaging/network/protocol/impl/cot/CotProtocol.java @@ -88,6 +88,11 @@ public class CotProtocol extends Protocol implements Selectable { private static final String DESTINATION_NAME = "/tak/cot/inbound"; private static final int MAX_BUFFER_SIZE = 1_048_576; // 1MB safety cap against a malformed/never-terminated stream + // A TLS record is indivisible and decrypts to up to 16 KB, so every read must have room for a + // whole record. The packet handed over by protocol detection is only 1 KB and must not be reused + // for reads: over ssl:// any record carrying more than 1 KB of CoT would never decode. + private static final int MIN_READ_BUFFER_SIZE = 32 * 1024; + private static final String READER_THREAD_PREFIX = "CoT Protocol Reader::"; private static final String UPDATE_SOURCE = "cot-ingest"; private static final byte[] EVENT_START = " 0) { + if (initialPacket != null && initialPacket.available() > 0) { appendAndProcess(initialPacket); } endPoint.register(SelectionKey.OP_READ, this); } + /** The configured server read buffer size, never smaller than one TLS record and never above the 1 MB cap. */ + static int readBufferSize(EndPoint endPoint) { + long configured = 0; + if (endPoint.getConfig() != null && endPoint.getConfig().getEndPointConfig() != null) { + configured = endPoint.getConfig().getEndPointConfig().getServerReadBufferSize(); + } + return (int) Math.max(MIN_READ_BUFFER_SIZE, Math.min(configured, MAX_BUFFER_SIZE)); + } + private void createSession() throws IOException { try { SessionContextBuilder scb = new SessionContextBuilder(sessionId, new ProtocolClientConnection(this)); @@ -168,20 +182,41 @@ public void selected(Selectable selectable, Selector selector, int selection) { private final class ReadTask implements Runnable { @Override public void run() { - Thread.currentThread().setName("CoT Protocol Reader::" + Thread.currentThread().getName()); + // The pool reuses threads, so prefix for this run only - prefixing the current name every + // time grew it without bound over the life of the server. + Thread current = Thread.currentThread(); + String poolName = current.getName(); + current.setName(READER_THREAD_PREFIX + poolName); + try { + readAvailable(); + } finally { + current.setName(poolName); + } + } + + private void readAvailable() { try { int read; boolean first = true; do { packet.clear(); read = endPoint.readPacket(packet); - if (read > 0) { + int applicationBytes = packet.position(); + if (read > 0 && applicationBytes > 0) { EndPoint.totalReceived.increment(); packet.flip(); appendAndProcess(packet); - } else if (first) { + } else if (first && read <= 0) { + // Nothing readable on the first pass: the peer has gone away. closeQuietly(); return; + } else { + // read > 0 but no application data was produced - e.g. an SSL read + // consumed encrypted bytes for a partial record and decoded nothing + // yet (readPacket then reports the encrypted count, not application + // payload). Stop draining and wait for the selector to signal fresh + // data instead of busy-spinning on a count that carries no payload. + break; } first = false; } while (read > 0); diff --git a/src/test/java/io/mapsmessaging/network/io/impl/ssl/SSLEndPointReadOverflowTest.java b/src/test/java/io/mapsmessaging/network/io/impl/ssl/SSLEndPointReadOverflowTest.java new file mode 100644 index 0000000000..a80d0e13e0 --- /dev/null +++ b/src/test/java/io/mapsmessaging/network/io/impl/ssl/SSLEndPointReadOverflowTest.java @@ -0,0 +1,293 @@ +/* + * + * Copyright [ 2026 ] Ralf Himmelein and Claude + * + * Licensed under the Apache License, Version 2.0 with the Commons Clause + * (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 + * https://commonsclause.com/ + * + * 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 io.mapsmessaging.network.io.impl.ssl; + +import io.mapsmessaging.dto.rest.config.network.EndPointServerConfigDTO; +import io.mapsmessaging.dto.rest.config.network.impl.TcpConfigDTO; +import io.mapsmessaging.network.io.EndPointServerStatus; +import io.mapsmessaging.network.io.impl.Selector; +import org.junit.jupiter.api.Test; + +import javax.net.ssl.SSLEngine; +import javax.net.ssl.SSLEngineResult; +import javax.net.ssl.SSLSession; +import java.io.IOException; +import java.net.InetAddress; +import java.net.Socket; +import java.nio.ByteBuffer; +import java.nio.channels.SocketChannel; +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.Mockito.*; + +/** + * Inbound TLS records larger than the caller's read buffer. The engine mock answers exactly as the + * JDK's TLSv1.3 engine does (checked against a real engine): a record that does not fit in the + * destination returns BUFFER_OVERFLOW with nothing consumed or produced. + */ +class SSLEndPointReadOverflowTest { + + private static final int PACKET_BUFFER_SIZE = 16709; + private static final int APPLICATION_BUFFER_SIZE = 16704; + private static final int RECORD_ENCRYPTED_BYTES = 5038; + private static final int RECORD_PLAINTEXT_BYTES = 5000; + + @Test + void readBuffer_whenRecordCannotFitEmptyBuffer_closesInsteadOfHanging() throws Exception { + SSLEndPoint endPoint = new Fixture().createEndPoint(); + try { + ByteBuffer detectionSized = ByteBuffer.allocate(1024); + + IOException failure = assertThrows(IOException.class, () -> endPoint.readBuffer(detectionSized)); + + assertTrue(failure.getMessage().contains("cannot hold a TLS record")); + assertEquals(0, detectionSized.position()); + } finally { + endPoint.close(); + } + } + + @Test + void readBuffer_whenBufferHoldsTheRecord_decodesIt() throws Exception { + SSLEndPoint endPoint = new Fixture().createEndPoint(); + try { + ByteBuffer applicationIn = ByteBuffer.allocate(100_000); + + endPoint.readBuffer(applicationIn); + + assertEquals(RECORD_PLAINTEXT_BYTES, applicationIn.position()); + } finally { + endPoint.close(); + } + } + + @Test + void readBuffer_whenPartlyFilledBufferOverflows_returnsDecodedDataForTheCallerToDrain() throws Exception { + SSLEndPoint endPoint = new Fixture().createEndPoint(); + try { + ByteBuffer applicationIn = ByteBuffer.allocate(3000); + applicationIn.put(new byte[100]); + + assertDoesNotThrow(() -> endPoint.readBuffer(applicationIn)); + + assertEquals(100, applicationIn.position()); + } finally { + endPoint.close(); + } + } + + @Test + void overflowing_second_record_is_preserved_until_the_caller_drains_the_first() throws Exception { + Fixture fixture = new Fixture(); + AtomicBoolean delivered = new AtomicBoolean(); + when(fixture.channel.read(any(ByteBuffer.class))).thenAnswer(invocation -> { + if (delivered.getAndSet(true)) { + return 0; + } + ByteBuffer encrypted = invocation.getArgument(0); + encrypted.put(new byte[2 * RECORD_ENCRYPTED_BYTES]); + return 2 * RECORD_ENCRYPTED_BYTES; + }); + SSLEndPoint endPoint = fixture.createEndPoint(); + try { + ByteBuffer application = ByteBuffer.allocate(6000); + endPoint.readBuffer(application); + assertEquals(RECORD_PLAINTEXT_BYTES, application.position()); + assertTrue(endPoint.hasBufferedReadData()); + + application.clear(); + endPoint.readBuffer(application); + assertEquals(RECORD_PLAINTEXT_BYTES, application.position()); + assertFalse(endPoint.hasBufferedReadData()); + } finally { + endPoint.close(); + } + } + + @Test + void fragmented_record_is_retained_until_the_remaining_encrypted_bytes_arrive() throws Exception { + Fixture fixture = new Fixture(); + AtomicInteger reads = new AtomicInteger(); + when(fixture.channel.read(any(ByteBuffer.class))).thenAnswer(invocation -> { + int count = switch (reads.getAndIncrement()) { + case 0 -> 2000; + case 1 -> RECORD_ENCRYPTED_BYTES - 2000; + default -> 0; + }; + ByteBuffer encrypted = invocation.getArgument(0); + encrypted.put(new byte[count]); + return count; + }); + SSLEndPoint endPoint = fixture.createEndPoint(); + try { + ByteBuffer application = ByteBuffer.allocate(32768); + endPoint.readBuffer(application); + assertEquals(0, application.position()); + assertTrue(endPoint.hasBufferedReadData()); + + endPoint.readBuffer(application); + assertEquals(RECORD_PLAINTEXT_BYTES, application.position()); + assertFalse(endPoint.hasBufferedReadData()); + } finally { + endPoint.close(); + } + } + + @Test + void full_encrypted_buffer_underflow_fails_instead_of_spinning() throws Exception { + Fixture fixture = new Fixture(); + when(fixture.channel.read(any(ByteBuffer.class))).thenAnswer(invocation -> { + ByteBuffer encrypted = invocation.getArgument(0); + int count = encrypted.remaining(); + encrypted.put(new byte[count]); + return count; + }); + when(fixture.sslEngine.unwrap(any(ByteBuffer.class), any(ByteBuffer.class))).thenReturn( + new SSLEngineResult(SSLEngineResult.Status.BUFFER_UNDERFLOW, + SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING, 0, 0)); + SSLEndPoint endPoint = fixture.createEndPoint(); + try { + IOException failure = assertThrows(IOException.class, () -> endPoint.readBuffer(ByteBuffer.allocate(32768))); + assertTrue(failure.getMessage().contains("inbound buffer full")); + verify(fixture.sslEngine, times(1)).unwrap(any(ByteBuffer.class), any(ByteBuffer.class)); + } finally { + endPoint.close(); + } + } + + @Test + void delegated_task_runs_before_retrying_the_buffered_record() throws Exception { + Fixture fixture = new Fixture(); + AtomicBoolean taskRan = new AtomicBoolean(); + Runnable task = () -> taskRan.set(true); + when(fixture.sslEngine.getDelegatedTask()).thenReturn(task, (Runnable) null); + when(fixture.sslEngine.unwrap(any(ByteBuffer.class), any(ByteBuffer.class))).thenAnswer(invocation -> { + if (!taskRan.get()) { + return new SSLEngineResult(SSLEngineResult.Status.OK, SSLEngineResult.HandshakeStatus.NEED_TASK, 0, 0); + } + ByteBuffer encrypted = invocation.getArgument(0); + ByteBuffer application = invocation.getArgument(1); + encrypted.position(encrypted.position() + RECORD_ENCRYPTED_BYTES); + application.put(new byte[RECORD_PLAINTEXT_BYTES]); + return new SSLEngineResult(SSLEngineResult.Status.OK, SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING, + RECORD_ENCRYPTED_BYTES, RECORD_PLAINTEXT_BYTES); + }); + SSLEndPoint endPoint = fixture.createEndPoint(); + try { + ByteBuffer application = ByteBuffer.allocate(32768); + endPoint.readBuffer(application); + assertTrue(taskRan.get()); + assertEquals(RECORD_PLAINTEXT_BYTES, application.position()); + assertFalse(endPoint.hasBufferedReadData()); + } finally { + endPoint.close(); + } + } + + @Test + void need_task_without_a_delegated_task_returns_without_repeated_unwraps() throws Exception { + Fixture fixture = new Fixture(); + when(fixture.sslEngine.unwrap(any(ByteBuffer.class), any(ByteBuffer.class))).thenReturn( + new SSLEngineResult(SSLEngineResult.Status.OK, SSLEngineResult.HandshakeStatus.NEED_TASK, 0, 0)); + SSLEndPoint endPoint = fixture.createEndPoint(); + try { + ByteBuffer application = ByteBuffer.allocate(32768); + endPoint.readBuffer(application); + assertEquals(0, application.position()); + assertTrue(endPoint.hasBufferedReadData()); + verify(fixture.sslEngine, times(1)).unwrap(any(ByteBuffer.class), any(ByteBuffer.class)); + } finally { + endPoint.close(); + } + } + + private static final class Fixture { + + private final SocketChannel channel = mock(SocketChannel.class); + private final Socket socket = mock(Socket.class); + private final Selector selector = mock(Selector.class); + private final SSLEngine sslEngine = mock(SSLEngine.class); + private final SSLSession sslSession = mock(SSLSession.class); + private final EndPointServerStatus serverStatus = mock(EndPointServerStatus.class); + private final AtomicBoolean recordDelivered = new AtomicBoolean(); + + Fixture() throws Exception { + EndPointServerConfigDTO serverConfig = new EndPointServerConfigDTO(); + TcpConfigDTO tcpConfig = new TcpConfigDTO(); + tcpConfig.setTimeout(60_000); + serverConfig.setEndPointConfig(tcpConfig); + + when(serverStatus.getConfig()).thenReturn(serverConfig); + when(channel.socket()).thenReturn(socket); + when(channel.isConnected()).thenReturn(true); + when(socket.getLocalAddress()).thenReturn(InetAddress.getLoopbackAddress()); + when(sslEngine.getSession()).thenReturn(sslSession); + when(sslSession.getPacketBufferSize()).thenReturn(PACKET_BUFFER_SIZE); + when(sslSession.getApplicationBufferSize()).thenReturn(APPLICATION_BUFFER_SIZE); + when(sslEngine.getHandshakeStatus()).thenReturn(SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING); + + when(sslEngine.wrap(any(ByteBuffer.class), any(ByteBuffer.class))).thenReturn(new SSLEngineResult( + SSLEngineResult.Status.OK, SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING, 0, 0)); + when(channel.write(any(ByteBuffer.class))).thenAnswer(invocation -> { + ByteBuffer encrypted = invocation.getArgument(0); + int remaining = encrypted.remaining(); + encrypted.position(encrypted.limit()); + return remaining; + }); + + // The socket holds one encrypted record, delivered once. + when(channel.read(any(ByteBuffer.class))).thenAnswer(invocation -> { + if (recordDelivered.getAndSet(true)) { + return 0; + } + ByteBuffer encryptedIn = invocation.getArgument(0); + encryptedIn.put(new byte[RECORD_ENCRYPTED_BYTES]); + return RECORD_ENCRYPTED_BYTES; + }); + + when(sslEngine.unwrap(any(ByteBuffer.class), any(ByteBuffer.class))).thenAnswer(invocation -> { + ByteBuffer encryptedIn = invocation.getArgument(0); + ByteBuffer applicationIn = invocation.getArgument(1); + if (encryptedIn.remaining() < RECORD_ENCRYPTED_BYTES) { + return new SSLEngineResult(SSLEngineResult.Status.BUFFER_UNDERFLOW, + SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING, 0, 0); + } + if (applicationIn.remaining() < RECORD_PLAINTEXT_BYTES) { + return new SSLEngineResult(SSLEngineResult.Status.BUFFER_OVERFLOW, + SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING, 0, 0); + } + encryptedIn.position(encryptedIn.position() + RECORD_ENCRYPTED_BYTES); + applicationIn.put(new byte[RECORD_PLAINTEXT_BYTES]); + return new SSLEngineResult(SSLEngineResult.Status.OK, + SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING, RECORD_ENCRYPTED_BYTES, RECORD_PLAINTEXT_BYTES); + }); + + doAnswer(invocation -> null).when(selector).register(any(SocketChannel.class), anyInt(), any()); + } + + SSLEndPoint createEndPoint() throws Exception { + return new SSLEndPoint(1, sslEngine, channel, selector, endPoint -> { }, serverStatus, List.of()); + } + } +} diff --git a/src/test/java/io/mapsmessaging/network/io/impl/ssl/SSLEndPointTlsRecordTest.java b/src/test/java/io/mapsmessaging/network/io/impl/ssl/SSLEndPointTlsRecordTest.java new file mode 100644 index 0000000000..0eeb6067d0 --- /dev/null +++ b/src/test/java/io/mapsmessaging/network/io/impl/ssl/SSLEndPointTlsRecordTest.java @@ -0,0 +1,265 @@ +/* + * + * Copyright [ 2020 - 2024 ] Matthew Buckton + * Copyright [ 2024 - 2026 ] MapsMessaging B.V. + * + * Licensed under the Apache License, Version 2.0 with the Commons Clause + * (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 + * https://commonsclause.com/ + * + * 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 io.mapsmessaging.network.io.impl.ssl; + +import io.mapsmessaging.dto.rest.config.network.EndPointServerConfigDTO; +import io.mapsmessaging.dto.rest.config.network.impl.TcpConfigDTO; +import io.mapsmessaging.network.io.EndPointServerStatus; +import io.mapsmessaging.network.io.Packet; +import io.mapsmessaging.network.io.impl.Selector; +import java.io.InputStream; +import java.net.InetAddress; +import java.net.Socket; +import java.nio.ByteBuffer; +import java.nio.channels.SocketChannel; +import java.nio.file.Files; +import java.nio.file.Path; +import java.security.KeyStore; +import java.util.ArrayDeque; +import java.util.Arrays; +import java.util.List; +import java.util.Queue; +import javax.net.ssl.KeyManagerFactory; +import javax.net.ssl.SSLContext; +import javax.net.ssl.SSLEngine; +import javax.net.ssl.SSLEngineResult; +import javax.net.ssl.TrustManagerFactory; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; +import org.junit.jupiter.params.provider.ValueSource; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** Exercises the endpoint with real JDK records; only the non-blocking socket boundary is mocked. */ +@Timeout(10) +class SSLEndPointTlsRecordTest { + + @ParameterizedTest + @CsvSource({"TLSv1.2,5000", "TLSv1.3,5000", "TLSv1.2,16384", "TLSv1.3,16384"}) + void a_record_larger_than_detection_buffer_decodes_without_losing_payload(String protocol, int size) throws Exception { + try (Fixture fixture = new Fixture(protocol)) { + byte[] payload = payload(size, 11); + fixture.incoming.add(ByteBuffer.wrap(fixture.encrypt(payload))); + Packet packet = new Packet(32768, false); + + assertTrue(fixture.endPoint.readPacket(packet) > 0); + + assertArrayEquals(payload, decoded(packet)); + assertFalse(fixture.endPoint.hasBufferedReadData()); + } + } + + @ParameterizedTest + @ValueSource(strings = {"TLSv1.2", "TLSv1.3"}) + void a_second_buffered_record_survives_overflow_and_decodes_on_the_next_read(String protocol) throws Exception { + try (Fixture fixture = new Fixture(protocol)) { + byte[] first = payload(5000, 17); + byte[] second = payload(5000, 29); + byte[] firstRecord = fixture.encrypt(first); + byte[] secondRecord = fixture.encrypt(second); + ByteBuffer records = ByteBuffer.allocate(firstRecord.length + secondRecord.length); + records.put(firstRecord).put(secondRecord).flip(); + fixture.incoming.add(records); + Packet packet = new Packet(6000, false); + + fixture.endPoint.readPacket(packet); + assertArrayEquals(first, decoded(packet)); + assertTrue(fixture.incoming.isEmpty(), "Both records must already have left the socket"); + assertTrue(fixture.endPoint.hasBufferedReadData()); + + packet.clear(); + assertTrue(fixture.endPoint.readPacket(packet) > 0); + assertArrayEquals(second, decoded(packet)); + assertFalse(fixture.endPoint.hasBufferedReadData()); + } + } + + @ParameterizedTest + @ValueSource(strings = {"TLSv1.2", "TLSv1.3"}) + void a_fragmented_record_produces_payload_only_after_its_remaining_bytes_arrive(String protocol) throws Exception { + try (Fixture fixture = new Fixture(protocol)) { + byte[] payload = payload(5000, 37); + byte[] record = fixture.encrypt(payload); + fixture.incoming.add(ByteBuffer.wrap(Arrays.copyOfRange(record, 0, 7))); + Packet packet = new Packet(32768, false); + + assertTrue(fixture.endPoint.readPacket(packet) > 0); + assertEquals(0, packet.position()); + assertTrue(fixture.endPoint.hasBufferedReadData()); + + fixture.incoming.add(ByteBuffer.wrap(Arrays.copyOfRange(record, 7, record.length))); + assertTrue(fixture.endPoint.readPacket(packet) > 0); + assertArrayEquals(payload, decoded(packet)); + assertFalse(fixture.endPoint.hasBufferedReadData()); + } + } + + private static byte[] payload(int size, int seed) { + byte[] payload = new byte[size]; + for (int index = 0; index < size; index++) { + payload[index] = (byte) (index * 31 + seed); + } + return payload; + } + + private static byte[] decoded(Packet packet) { + packet.flip(); + byte[] payload = new byte[packet.available()]; + packet.get(payload); + return payload; + } + + private static SSLContext context() throws Exception { + char[] password = System.getProperty("javax.net.ssl.keyStorePassword", "password").toCharArray(); + KeyStore keys = KeyStore.getInstance("JKS"); + Path keyStore = Path.of(System.getProperty("javax.net.ssl.keyStore", "my-keystore.jks")); + try (InputStream input = Files.newInputStream(keyStore)) { + keys.load(input, password); + } + KeyManagerFactory keyManagers = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm()); + keyManagers.init(keys, password); + // The fixture trusts the same test identity at both ends; no external trust or network is needed. + TrustManagerFactory trustManagers = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm()); + trustManagers.init(keys); + SSLContext context = SSLContext.getInstance("TLS"); + context.init(keyManagers.getKeyManagers(), trustManagers.getTrustManagers(), null); + return context; + } + + private static final class Fixture implements AutoCloseable { + private final Queue incoming = new ArrayDeque<>(); + private final SSLEngine client; + private final SSLEndPoint endPoint; + + private Fixture(String protocol) throws Exception { + SSLContext context = context(); + client = context.createSSLEngine(); + client.setUseClientMode(true); + client.setEnabledProtocols(new String[]{protocol}); + SSLEngine server = context.createSSLEngine(); + server.setEnabledProtocols(new String[]{protocol}); + SocketChannel channel = mock(SocketChannel.class); + Socket socket = mock(Socket.class); + Selector selector = mock(Selector.class); + EndPointServerStatus status = mock(EndPointServerStatus.class); + EndPointServerConfigDTO config = new EndPointServerConfigDTO(); + config.setEndPointConfig(new TcpConfigDTO()); + when(status.getConfig()).thenReturn(config); + when(channel.socket()).thenReturn(socket); + when(channel.isConnected()).thenReturn(true); + when(socket.getLocalAddress()).thenReturn(InetAddress.getLoopbackAddress()); + when(channel.write(any(ByteBuffer.class))).thenAnswer(invocation -> { + ByteBuffer bytes = invocation.getArgument(0); + int count = bytes.remaining(); + bytes.position(bytes.limit()); + return count; + }); + when(channel.read(any(ByteBuffer.class))).thenAnswer(invocation -> { + if (incoming.isEmpty()) { + return 0; + } + ByteBuffer target = invocation.getArgument(0); + ByteBuffer source = incoming.element(); + int count = Math.min(target.remaining(), source.remaining()); + int limit = source.limit(); + source.limit(source.position() + count); + target.put(source); + source.limit(limit); + if (!source.hasRemaining()) { + incoming.remove(); + } + return count; + }); + endPoint = new SSLEndPoint(1, server, channel, selector, null, status, List.of()); + try { + handshake(server); + endPoint.handshakeManager = new SSLHandshakeManagerFinished(ByteBuffer.allocate(server.getSession().getApplicationBufferSize())); + } catch (Exception failure) { + endPoint.close(); + throw failure; + } + } + + private byte[] encrypt(byte[] payload) throws Exception { + ByteBuffer encrypted = ByteBuffer.allocate(65536); + ByteBuffer application = ByteBuffer.wrap(payload); + while (application.hasRemaining()) { + SSLEngineResult result = client.wrap(application, encrypted); + assertEquals(SSLEngineResult.Status.OK, result.getStatus()); + assertTrue(result.bytesConsumed() > 0, "Wrapping application data must make progress"); + } + encrypted.flip(); + byte[] record = new byte[encrypted.remaining()]; + encrypted.get(record); + return record; + } + + private void handshake(SSLEngine server) throws Exception { + client.beginHandshake(); + ByteBuffer clientToServer = ByteBuffer.allocate(131072); + ByteBuffer serverToClient = ByteBuffer.allocate(131072); + ByteBuffer sink = ByteBuffer.allocate(65536); + for (int attempt = 0; attempt < 100; attempt++) { + handshakeStep(client, server, clientToServer, sink); + handshakeStep(server, client, serverToClient, sink); + if (client.getHandshakeStatus() == SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING + && server.getHandshakeStatus() == SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING + && clientToServer.position() == 0 && serverToClient.position() == 0) { + return; + } + } + throw new AssertionError("TLS handshake did not complete"); + } + + private static void handshakeStep(SSLEngine sender, SSLEngine receiver, ByteBuffer wire, ByteBuffer sink) throws Exception { + runTasks(sender); + runTasks(receiver); + if (sender.getHandshakeStatus() == SSLEngineResult.HandshakeStatus.NEED_WRAP) { + sender.wrap(ByteBuffer.allocate(0), wire); + } + wire.flip(); + if (wire.hasRemaining()) { + sink.clear(); + receiver.unwrap(wire, sink); + } + wire.compact(); + runTasks(receiver); + } + + private static void runTasks(SSLEngine engine) { + Runnable task; + while ((task = engine.getDelegatedTask()) != null) { + task.run(); + } + } + + @Override + public void close() { + endPoint.close(); + } + } +} diff --git a/src/test/java/io/mapsmessaging/network/protocol/impl/cot/CotProtocolTest.java b/src/test/java/io/mapsmessaging/network/protocol/impl/cot/CotProtocolTest.java index d82cfa71a0..63b70dc5eb 100644 --- a/src/test/java/io/mapsmessaging/network/protocol/impl/cot/CotProtocolTest.java +++ b/src/test/java/io/mapsmessaging/network/protocol/impl/cot/CotProtocolTest.java @@ -1,8 +1,13 @@ package io.mapsmessaging.network.protocol.impl.cot; import io.mapsmessaging.api.MessageEvent; +import io.mapsmessaging.MessageDaemon; +import io.mapsmessaging.api.Destination; +import io.mapsmessaging.api.message.Message; +import io.mapsmessaging.api.features.DestinationType; import io.mapsmessaging.api.Session; import io.mapsmessaging.api.SessionManager; +import io.mapsmessaging.dto.rest.config.network.EndPointConfigDTO; import io.mapsmessaging.dto.rest.config.network.EndPointServerConfigDTO; import io.mapsmessaging.dto.rest.config.protocol.impl.CotProtocolConfigDTO; import io.mapsmessaging.engine.session.security.SecurityContext; @@ -10,13 +15,23 @@ import io.mapsmessaging.network.io.Packet; import io.mapsmessaging.utilities.admin.JMXManager; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; import org.mockito.MockedStatic; +import org.mockito.ArgumentCaptor; import javax.security.auth.Subject; +import java.lang.reflect.Constructor; +import java.nio.channels.SelectionKey; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; import static org.junit.jupiter.api.Assertions.*; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.*; class CotProtocolTest { @@ -62,6 +77,155 @@ void protocolIdentityInboundContractAndCloseUseInternalSession() throws Exceptio } } + @Test + @Timeout(10) + void readTask_whenReadReportsBytesButDecodesNoPayload_parksInsteadOfSpinning() throws Exception { + boolean original = JMXManager.isEnableJMX(); + JMXManager.setEnableJMX(false); + SessionManager manager = mock(SessionManager.class); + Session session = mock(Session.class); + SecurityContext security = mock(SecurityContext.class); + Subject subject = new Subject(); + when(session.getSecurityContext()).thenReturn(security); + when(security.getSubject()).thenReturn(subject); + when(manager.create(any(), any())).thenReturn(session); + + try (MockedStatic mocked = mockStatic(SessionManager.class)) { + mocked.when(SessionManager::getInstance).thenReturn(manager); + EndPoint endpoint = endpoint(); + // Simulate the wedged SSL case: readPacket reports a positive (encrypted) + // count but leaves the application packet empty - no decoded payload. With + // the old while(read > 0) loop this spun the reader at 100% CPU forever. + when(endpoint.readPacket(any())).thenReturn(16708); + + CotProtocol protocol = + new CotProtocol(endpoint, new Packet(0, false), new CotProtocolConfigDTO()); + // The constructor registers OP_READ once; clear that interaction first. + verify(endpoint).register(eq(SelectionKey.OP_READ), eq(protocol)); + clearInvocations(endpoint); + + protocol.selected(protocol, null, SelectionKey.OP_READ); + + // The reader must stop after a bounded number of reads and re-arm the + // selector, rather than looping on the empty positive count. + verify(endpoint, timeout(5000)).register(eq(SelectionKey.OP_READ), eq(protocol)); + verify(endpoint, atMost(5)).readPacket(any()); + verify(endpoint, never()).close(); + + protocol.close(); + } finally { + JMXManager.setEnableJMX(original); + } + } + + @Test + @Timeout(10) + void readTask_readsIntoBufferThatHoldsATlsRecord_notTheDetectionPacket() throws Exception { + withSession((endpoint, session) -> { + List readTargets = new CopyOnWriteArrayList<>(); + when(endpoint.readPacket(any())).thenAnswer(invocation -> { + readTargets.add(invocation.getArgument(0)); + return 16708; + }); + // Protocol detection hands over its own 1 KB buffer - over ssl:// a TLS record decrypts + // to up to 16 KB and can never be unwrapped into it. + Packet detectionPacket = new Packet(1024, true); + CotProtocol protocol = new CotProtocol(endpoint, detectionPacket, new CotProtocolConfigDTO()); + clearInvocations(endpoint); + + protocol.selected(protocol, null, SelectionKey.OP_READ); + verify(endpoint, timeout(5000)).register(eq(SelectionKey.OP_READ), eq(protocol)); + + assertFalse(readTargets.isEmpty()); + for (Packet target : readTargets) { + assertNotSame(detectionPacket, target); + assertTrue(target.getRawBuffer().capacity() >= 16704, + "read buffer of " + target.getRawBuffer().capacity() + " bytes cannot hold a TLS record"); + } + protocol.close(); + }); + } + + @Test + void readBufferSize_usesConfiguredSize_withinTlsRecordFloorAndSafetyCap() { + assertEquals(100_000, CotProtocol.readBufferSize(endpointWithReadBufferSize(100_000))); + assertEquals(32 * 1024, CotProtocol.readBufferSize(endpointWithReadBufferSize(10_240))); + assertEquals(32 * 1024, CotProtocol.readBufferSize(endpointWithReadBufferSize(1024))); + assertEquals(1_048_576, CotProtocol.readBufferSize(endpointWithReadBufferSize(4_000_000))); + assertEquals(32 * 1024, CotProtocol.readBufferSize(endpoint())); + } + + @Test + @Timeout(10) + void readTask_doesNotGrowThePooledThreadNameAcrossRuns() throws Exception { + withSession((endpoint, session) -> { + List readerNames = new CopyOnWriteArrayList<>(); + when(endpoint.readPacket(any())).thenAnswer(invocation -> { + readerNames.add(Thread.currentThread().getName()); + return 16708; + }); + CotProtocol protocol = new CotProtocol(endpoint, new Packet(0, false), new CotProtocolConfigDTO()); + + for (int run = 0; run < 3; run++) { + clearInvocations(endpoint); + protocol.selected(protocol, null, SelectionKey.OP_READ); + verify(endpoint, timeout(5000)).register(eq(SelectionKey.OP_READ), eq(protocol)); + } + + assertEquals(3, readerNames.size()); + for (String name : readerNames) { + assertTrue(name.startsWith("CoT Protocol Reader::"), name); + assertFalse(name.startsWith("CoT Protocol Reader::CoT Protocol Reader::"), name); + } + protocol.close(); + }); + } + + @Test + @Timeout(10) + void detection_prefix_and_large_following_read_publish_one_complete_event() throws Exception { + withSession((endpoint, session) -> { + Destination destination = mock(Destination.class); + when(session.findDestination("/tak/cot/inbound", DestinationType.TOPIC)) + .thenReturn(CompletableFuture.completedFuture(destination)); + byte[] event = ("" + "payload".repeat(800) + + "").getBytes(StandardCharsets.UTF_8); + int split = 128; + Packet detection = new Packet(1024, false); + detection.put(Arrays.copyOfRange(event, 0, split)); + detection.flip(); + AtomicBoolean delivered = new AtomicBoolean(); + when(endpoint.readPacket(any())).thenAnswer(invocation -> { + if (delivered.getAndSet(true)) { + return 0; + } + Packet target = invocation.getArgument(0); + target.put(Arrays.copyOfRange(event, split, event.length)); + return event.length - split; + }); + CotProtocol protocol = new CotProtocol(endpoint, detection, new CotProtocolConfigDTO()); + try (MockedStatic daemon = mockStatic(MessageDaemon.class)) { + daemon.when(MessageDaemon::getInstance).thenReturn(null); + verify(destination, never()).storeMessage(any()); + clearInvocations(endpoint); + // Run the actual reader on this thread so the singleton boundary remains isolated, + // independent of whether another protocol test has started the daemon. + Class readerType = Class.forName(CotProtocol.class.getName() + "$ReadTask"); + Constructor constructor = readerType.getDeclaredConstructor(CotProtocol.class); + constructor.setAccessible(true); + Runnable reader = (Runnable) constructor.newInstance(protocol); + reader.run(); + verify(endpoint).register(eq(SelectionKey.OP_READ), eq(protocol)); + ArgumentCaptor stored = ArgumentCaptor.forClass(Message.class); + verify(destination, times(1)).storeMessage(stored.capture()); + assertArrayEquals(event, stored.getValue().getOpaqueData()); + verify(endpoint, never()).close(); + } finally { + protocol.close(); + } + }); + } + @Test void staticFrameSearchHelpersHandleBoundsAndWhitespace() throws Exception { byte[] data = "xx 1tail".getBytes(); @@ -82,6 +246,36 @@ void staticFrameSearchHelpersHandleBoundsAndWhitespace() throws Exception { assertEquals(-1, index.invoke(null, data, "missing".getBytes(), 0)); } + @FunctionalInterface + private interface SessionBody { + void run(EndPoint endpoint, Session session) throws Exception; + } + + private static void withSession(SessionBody body) throws Exception { + boolean original = JMXManager.isEnableJMX(); + JMXManager.setEnableJMX(false); + SessionManager manager = mock(SessionManager.class); + Session session = mock(Session.class); + SecurityContext security = mock(SecurityContext.class); + when(session.getSecurityContext()).thenReturn(security); + when(security.getSubject()).thenReturn(new Subject()); + when(manager.create(any(), any())).thenReturn(session); + try (MockedStatic mocked = mockStatic(SessionManager.class)) { + mocked.when(SessionManager::getInstance).thenReturn(manager); + body.run(endpoint(), session); + } finally { + JMXManager.setEnableJMX(original); + } + } + + private static EndPoint endpointWithReadBufferSize(long size) { + EndPoint endpoint = endpoint(); + EndPointConfigDTO endPointConfig = mock(EndPointConfigDTO.class); + when(endPointConfig.getServerReadBufferSize()).thenReturn(size); + when(endpoint.getConfig().getEndPointConfig()).thenReturn(endPointConfig); + return endpoint; + } + private static EndPoint endpoint() { EndPoint endpoint = mock(EndPoint.class); EndPointServerConfigDTO config = mock(EndPointServerConfigDTO.class);