diff --git a/src/main/java/io/mapsmessaging/network/protocol/impl/coap/CoapInterfaceManager.java b/src/main/java/io/mapsmessaging/network/protocol/impl/coap/CoapInterfaceManager.java index 57dcc8a59d..8724c71b06 100644 --- a/src/main/java/io/mapsmessaging/network/protocol/impl/coap/CoapInterfaceManager.java +++ b/src/main/java/io/mapsmessaging/network/protocol/impl/coap/CoapInterfaceManager.java @@ -32,6 +32,7 @@ import java.io.IOException; import java.net.SocketAddress; import java.nio.channels.SelectionKey; +import java.util.ArrayList; import java.util.HashMap; import java.util.LinkedHashMap; @@ -86,14 +87,14 @@ public boolean processPacket(Packet packet) throws IOException { @Override public void close() { - for (CoapProtocol protocol : currentSessions.values()) { + for (CoapProtocol protocol : new ArrayList<>(currentSessions.values())) { try { protocol.close(); } catch (IOException e) { // Ignore we are closing } - currentSessions.clear(); } + currentSessions.clear(); } @Override diff --git a/src/main/java/io/mapsmessaging/network/protocol/impl/coap/CoapProtocol.java b/src/main/java/io/mapsmessaging/network/protocol/impl/coap/CoapProtocol.java index 017134ceeb..e7cfd58e56 100644 --- a/src/main/java/io/mapsmessaging/network/protocol/impl/coap/CoapProtocol.java +++ b/src/main/java/io/mapsmessaging/network/protocol/impl/coap/CoapProtocol.java @@ -240,7 +240,7 @@ private void setOptions(@NotNull @NonNull MessageEvent messageEvent, Context con optionSet.add(eTag); } if (context.isObserve()) { - Observe option = new Observe((int) (context.getObserveId().incrementAndGet() & 0xffff)); + Observe option = new Observe((int) (context.getObserveId().incrementAndGet() & 0xffffff)); optionSet.putOption(option); } List schemas = SchemaManager.getInstance().getSchemaByContext(messageEvent.getDestinationName()); diff --git a/src/main/java/io/mapsmessaging/network/protocol/impl/coap/blockwise/BlockReceiveMonitor.java b/src/main/java/io/mapsmessaging/network/protocol/impl/coap/blockwise/BlockReceiveMonitor.java index 0672d79984..a0a2ebf36a 100644 --- a/src/main/java/io/mapsmessaging/network/protocol/impl/coap/blockwise/BlockReceiveMonitor.java +++ b/src/main/java/io/mapsmessaging/network/protocol/impl/coap/blockwise/BlockReceiveMonitor.java @@ -40,7 +40,7 @@ public synchronized void complete(String path) { blockBasedPackets.remove(path); } - public void scanForIdle() { + public synchronized void scanForIdle() { long expired = System.currentTimeMillis() - 30_000; List expiredList = new ArrayList<>(); blockBasedPackets.entrySet().stream() diff --git a/src/main/java/io/mapsmessaging/network/protocol/impl/coap/blockwise/ReceivePacket.java b/src/main/java/io/mapsmessaging/network/protocol/impl/coap/blockwise/ReceivePacket.java index c3f9851539..6001373b4f 100644 --- a/src/main/java/io/mapsmessaging/network/protocol/impl/coap/blockwise/ReceivePacket.java +++ b/src/main/java/io/mapsmessaging/network/protocol/impl/coap/blockwise/ReceivePacket.java @@ -22,6 +22,7 @@ import java.io.ByteArrayOutputStream; import java.io.IOException; import java.util.ArrayList; +import java.util.Arrays; import java.util.List; public class ReceivePacket { @@ -34,11 +35,18 @@ public ReceivePacket(int blockSize){ this.blockSize = 1 << ( blockSize +4); } - public void add(int index, byte[] block){ - while(index > blocks.size()){ - blocks.add(new byte[0]); + public boolean add(int index, byte[] block) { + if (index < 0 || block == null || block.length > blockSize) { + return false; } - blocks.add(index, block); + if (index < blocks.size()) { + return Arrays.equals(blocks.get(index), block); + } + if (index != blocks.size()) { + return false; + } + blocks.add(block); + return true; } public byte[] getFull(){ diff --git a/src/main/java/io/mapsmessaging/network/protocol/impl/coap/listeners/GetListener.java b/src/main/java/io/mapsmessaging/network/protocol/impl/coap/listeners/GetListener.java index 4b3615cfe2..a5ad752de0 100644 --- a/src/main/java/io/mapsmessaging/network/protocol/impl/coap/listeners/GetListener.java +++ b/src/main/java/io/mapsmessaging/network/protocol/impl/coap/listeners/GetListener.java @@ -122,9 +122,11 @@ private BasePacket buildSubscription(String path, BasePacket request, CoapProtoc Context subscriptionContext = protocol.getSubscriptionState().create(path, request); subscriptionContext.setSubscribedEventManager(session.addSubscription(context)); return null; - } catch (Exception exception) { + } catch (InterruptedException exception) { Thread.currentThread().interrupt(); return request.buildAckResponse(Code.INTERNAL_SERVER_ERROR); + } catch (Exception exception) { + return request.buildAckResponse(Code.INTERNAL_SERVER_ERROR); } } diff --git a/src/main/java/io/mapsmessaging/network/protocol/impl/coap/listeners/PublishListener.java b/src/main/java/io/mapsmessaging/network/protocol/impl/coap/listeners/PublishListener.java index 3666ac66b6..b42e47339f 100644 --- a/src/main/java/io/mapsmessaging/network/protocol/impl/coap/listeners/PublishListener.java +++ b/src/main/java/io/mapsmessaging/network/protocol/impl/coap/listeners/PublishListener.java @@ -98,21 +98,40 @@ protected Message build(BasePacket request, CoapProtocol protocol){ private boolean handleBlock(BasePacket request, String path, OptionSet optionSet, CoapProtocol protocol) { Block block = (Block) optionSet.getOption(BLOCK1); - BlockReceiveState blockReceiveState = protocol.getBlockReceiveMonitor().registerOrGet(block, path); - blockReceiveState.getReceivePacket().add(block.getNumber(), request.getPayload()); - if (request.getType().equals(TYPE.CON)) { - BasePacket response = request.buildAckResponse(Code.CONTINUE); - try { - protocol.sendResponse(response); - } catch (IOException e) { + byte[] payload = request.getPayload(); + int blockSize = 1 << (block.getSizeEx() + 4); + boolean valid = payload != null && payload.length <= blockSize && (!block.isMore() || payload.length == blockSize); + BlockReceiveState blockReceiveState = null; + if (valid) { + blockReceiveState = protocol.getBlockReceiveMonitor().registerOrGet(block, path); + valid = blockReceiveState.getReceivePacket().add(block.getNumber(), payload); + } + if (!valid) { + if (request.getType().equals(TYPE.CON)) { try { - protocol.close(); - } catch (IOException ex) { - // We are closing + protocol.sendResponse(request.buildAckResponse(Code.REQUEST_ENTITY_INCOMPLETE)); + } catch (IOException e) { + try { + protocol.close(); + } catch (IOException ex) { + // We are closing + } } } + return false; } if (block.isMore()) { + if (request.getType().equals(TYPE.CON)) { + try { + protocol.sendResponse(request.buildAckResponse(Code.CONTINUE)); + } catch (IOException e) { + try { + protocol.close(); + } catch (IOException ex) { + // We are closing + } + } + } return false; } protocol.getBlockReceiveMonitor().complete(path); diff --git a/src/main/java/io/mapsmessaging/network/protocol/impl/coap/packet/Code.java b/src/main/java/io/mapsmessaging/network/protocol/impl/coap/packet/Code.java index defa074097..ae3c1dcd39 100644 --- a/src/main/java/io/mapsmessaging/network/protocol/impl/coap/packet/Code.java +++ b/src/main/java/io/mapsmessaging/network/protocol/impl/coap/packet/Code.java @@ -68,6 +68,7 @@ public enum Code { NOT_FOUND(ERROR, 4, "Not Found"), METHOD_NOT_ALLOWED(ERROR, 5, "Method Not Allowed"), NOT_ACCEPTABLE(ERROR, 6, "Not Acceptable"), + REQUEST_ENTITY_INCOMPLETE(ERROR, 8, "Request Entity Incomplete"), PRECONDITION_FAILED(ERROR, 12, "Precondition Failed"), REQUEST_ENTITY_TOO_LARGE(ERROR, 13, "Request Entity Too Large"), UNSUPPORTED_CONTENT_FORMAT(ERROR, 15, "Unsupported Content-Format"), @@ -159,6 +160,8 @@ private static Code getError(int val) { case 6: return NOT_ACCEPTABLE; + case 8: + return REQUEST_ENTITY_INCOMPLETE; case 12: return PRECONDITION_FAILED; case 13: diff --git a/src/main/java/io/mapsmessaging/network/protocol/impl/coap/packet/PacketFactory.java b/src/main/java/io/mapsmessaging/network/protocol/impl/coap/packet/PacketFactory.java index 27879252ba..1d50da7076 100644 --- a/src/main/java/io/mapsmessaging/network/protocol/impl/coap/packet/PacketFactory.java +++ b/src/main/java/io/mapsmessaging/network/protocol/impl/coap/packet/PacketFactory.java @@ -50,6 +50,9 @@ public BasePacket parseFrame(Packet packet) throws IOException { } int rawCode = packet.get(packet.position() + 1) & 0xff; + if (rawCode == EMPTY && (tokenLength != 0 || packet.available() != 4)) { + throw new IOException("CoAP Empty message must not contain token, options or payload"); + } int codeClass = (rawCode >>> 5) & 0b111; int code = rawCode & 0b11111; BasePacket basePacket; diff --git a/src/main/java/io/mapsmessaging/network/protocol/impl/coap/packet/options/Observe.java b/src/main/java/io/mapsmessaging/network/protocol/impl/coap/packet/options/Observe.java index a1bd07aa4c..14dd6750d0 100644 --- a/src/main/java/io/mapsmessaging/network/protocol/impl/coap/packet/options/Observe.java +++ b/src/main/java/io/mapsmessaging/network/protocol/impl/coap/packet/options/Observe.java @@ -19,6 +19,8 @@ package io.mapsmessaging.network.protocol.impl.coap.packet.options; +import java.io.IOException; + public class Observe extends BinaryOption { public Observe() { @@ -30,6 +32,14 @@ public Observe(int val) { this.value = val; } + @Override + public void update(byte[] data) throws IOException { + if (data.length > 3) { + throw new IOException("Observe option exceeds three bytes"); + } + super.update(data); + } + public boolean register(){ return getValue() == 0; } diff --git a/src/test/java/io/mapsmessaging/network/protocol/impl/coap/BlockwiseSendTest.java b/src/test/java/io/mapsmessaging/network/protocol/impl/coap/BlockwiseSendTest.java index 6395e4a96d..015a41783b 100644 --- a/src/test/java/io/mapsmessaging/network/protocol/impl/coap/BlockwiseSendTest.java +++ b/src/test/java/io/mapsmessaging/network/protocol/impl/coap/BlockwiseSendTest.java @@ -22,6 +22,7 @@ import java.io.IOException; import org.eclipse.californium.core.CoapClient; import org.eclipse.californium.core.CoapResponse; +import org.eclipse.californium.core.coap.CoAP; import org.eclipse.californium.core.coap.Request; import org.eclipse.californium.elements.exception.ConnectorException; import org.junit.jupiter.api.Assertions; @@ -34,28 +35,38 @@ class BlockwiseSendTest extends BaseCoapTest { @ValueSource(ints = {0,1,2,3,4,5,6}) void basicSendTest(int size) throws ConnectorException, IOException { CoapClient client = new CoapClient(getUri()); - byte[] tmp = new byte[1024]; - for(int x=0;x response = ArgumentCaptor.forClass(BasePacket.class); + verify(protocol, times(1)).sendResponse(response.capture()); + assertEquals(TYPE.ACK, response.getValue().getType()); + assertEquals(Code.CONTINUE, response.getValue().getCode()); + } + + @Test + void missingBlockGetsRequestEntityIncompleteResponse() throws Exception { + CoapProtocol protocol = mock(CoapProtocol.class); + when(protocol.getBlockReceiveMonitor()).thenReturn(new BlockReceiveMonitor()); + BasePacket request = new BasePacket(3, TYPE.CON, Code.EMPTY, 1, 101, new byte[0]); + request.getOptions().add(new Block(Constants.BLOCK1, 2, false, 0)); + request.setPayload(new byte[]{1}); + + new PutListener().handle(request, protocol); + + ArgumentCaptor response = ArgumentCaptor.forClass(BasePacket.class); + verify(protocol, times(1)).sendResponse(response.capture()); + assertEquals(Code.REQUEST_ENTITY_INCOMPLETE, response.getValue().getCode()); + } + + @Test + void retransmittedBlockDoesNotDuplicateRepresentation() { + ReceivePacket upload = new ReceivePacket(0); + byte[] first = sequence(16); + byte[] last = new byte[]{42}; + + assertTrue(upload.add(0, first)); + assertTrue(upload.add(0, first)); + assertTrue(upload.add(1, last)); + byte[] expected = new byte[first.length + last.length]; + System.arraycopy(first, 0, expected, 0, first.length); + expected[first.length] = last[0]; + assertArrayEquals(expected, upload.getFull()); + } + + @Test + void incompleteAndConflictingBlocksAreRejectedWithoutCorruptingUpload() { + ReceivePacket upload = new ReceivePacket(0); + byte[] first = sequence(16); + + assertFalse(upload.add(1, first), "Missing block zero must not become a zero-length placeholder"); + assertTrue(upload.add(0, first)); + assertFalse(upload.add(0, new byte[]{42}), "Conflicting retransmission must not insert a duplicate"); + assertFalse(upload.add(2, new byte[]{42}), "Block gap must be rejected"); + assertArrayEquals(first, upload.getFull()); + } + @ParameterizedTest @MethodSource("sendPacketShapes") void blockPacketSplitsPayloadWithoutLossOrDuplication(int payloadSize, int blockSize) { diff --git a/src/test/java/io/mapsmessaging/network/protocol/impl/coap/UdpFaultProxyCoapIT.java b/src/test/java/io/mapsmessaging/network/protocol/impl/coap/UdpFaultProxyCoapIT.java new file mode 100644 index 0000000000..e73ed69168 --- /dev/null +++ b/src/test/java/io/mapsmessaging/network/protocol/impl/coap/UdpFaultProxyCoapIT.java @@ -0,0 +1,235 @@ +/* + * 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.protocol.impl.coap; + +import io.mapsmessaging.test.network.UdpFaultProxy; +import io.mapsmessaging.test.network.UdpFaultProxy.Direction; +import io.mapsmessaging.test.network.UdpFaultProxy.Stats; +import java.io.IOException; +import java.net.DatagramPacket; +import java.net.DatagramSocket; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.SocketException; +import java.time.Duration; +import org.eclipse.californium.core.CoapClient; +import org.eclipse.californium.core.CoapResponse; +import org.eclipse.californium.core.coap.CoAP; +import org.eclipse.californium.core.coap.Request; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** End-to-end CoAP exchanges across the protocol-neutral UDP fault proxy. */ +class UdpFaultProxyCoapIT extends BaseCoapTest { + + private static UdpFaultProxy proxy; + + @BeforeAll + static void startProxy() throws SocketException { + proxy = new UdpFaultProxy(0, new InetSocketAddress("127.0.0.1", 5683)); + } + + @BeforeEach + void resetRules() { + proxy.resetRules(); + } + + @AfterEach + void clearRules() { + proxy.resetRules(); + } + + @AfterAll + static void stopProxy() { + if (proxy != null) { + proxy.close(); + } + } + + @Test + void transparentProxyPreservesCoapPutAndGetPayload() throws Exception { + CoapClient client = createClient(); + try { + byte[] payload = payload(197); + CoapResponse putResponse = client.put(payload, 0); + assertSuccessfulPut(putResponse); + CoapResponse getResponse = client.get(); + assertNotNull(getResponse, "GET must receive a response through the proxy"); + assertArrayEquals(payload, getResponse.getPayload()); + assertTrue(proxy.stats().clientToServerForwarded() > 0); + assertTrue(proxy.stats().serverToClientForwarded() > 0); + assertEquals(0, proxy.stats().ioFailures()); + } finally { + client.shutdown(); + } + } + + @Test + void droppedFirstPutRequestIsRetransmittedAndPreservesPayload() throws Exception { + CoapClient client = createClient(); + try { + byte[] payload = payload(193); + Stats before = proxy.stats(); + proxy.dropNext(Direction.CLIENT_TO_SERVER, 1); + CoapResponse response = client.put(payload, 0); + assertSuccessfulPut(response); + Stats afterPut = proxy.stats(); + assertEquals(1L, afterPut.dropped() - before.dropped(), "The requested packet loss must actually occur"); + assertTrue(afterPut.clientToServerReceived() - before.clientToServerReceived() >= 2, + "The client must retransmit the missing confirmable PUT"); + assertEquals(0, afterPut.pendingRules()); + CoapResponse get = client.get(); + assertNotNull(get, "GET must succeed after PUT retransmission"); + assertArrayEquals(payload, get.getPayload()); + } finally { + client.shutdown(); + } + } + + @Test + void droppedBlock1ContinueIsRetransmittedWithoutPayloadCorruption() throws Exception { + CoapClient client = createClient(); + try { + byte[] payload = payload(1_024); + int blockSize = 256; + Stats before = proxy.stats(); + proxy.dropNext(Direction.SERVER_TO_CLIENT, 1); + for (int blockNumber = 0; blockNumber < payload.length / blockSize; blockNumber++) { + Request put = Request.newPut(); + byte[] blockPayload = new byte[blockSize]; + System.arraycopy(payload, blockNumber * blockSize, blockPayload, 0, blockSize); + put.setPayload(blockPayload); + boolean more = blockNumber != (payload.length / blockSize) - 1; + put.getOptions().setBlock1(4, more, blockNumber); + CoapResponse response = client.advanced(put); + assertNotNull(response, "Block1 exchange " + blockNumber + " must complete"); + if (more) { + assertEquals(CoAP.ResponseCode.CONTINUE, response.getCode(), "Block1 intermediate response"); + } else { + assertSuccessfulPut(response); + } + } + Stats afterPut = proxy.stats(); + assertEquals(1L, afterPut.dropped() - before.dropped(), "The first server response must be dropped"); + assertTrue(afterPut.clientToServerReceived() - before.clientToServerReceived() >= 5, + "Lost Continue must cause a retransmission in addition to the four Block1 requests"); + assertEquals(0, afterPut.pendingRules()); + CoapResponse get = client.get(); + assertNotNull(get, "GET must succeed after a retransmitted Block1 exchange"); + assertArrayEquals(payload, get.getPayload(), "Retransmission must not duplicate or reorder stored blocks"); + } finally { + client.shutdown(); + } + } + + @Test + void reorderedConfirmablePingsReceiveDistinctResetsInReversedOrder() throws Exception { + try (DatagramSocket client = rawClient()) { + Stats before = proxy.stats(); + proxy.reverseNext(Direction.CLIENT_TO_SERVER, 2, Duration.ofSeconds(2)); + sendPing(client, 0x2101); + sendPing(client, 0x2102); + assertEquals(0x2102, receiveResetMessageId(client)); + assertEquals(0x2101, receiveResetMessageId(client)); + Stats after = proxy.stats(); + assertEquals(2L, after.reversed() - before.reversed()); + assertEquals(0L, after.partialReverseFlushes() - before.partialReverseFlushes()); + assertEquals(0, after.pendingRules()); + } + } + + @Test + void incompleteReverseBatchFlushesSoStopAndWaitClientCanProgress() throws Exception { + try (DatagramSocket client = rawClient()) { + Stats before = proxy.stats(); + proxy.reverseNext(Direction.CLIENT_TO_SERVER, 2, Duration.ofMillis(150)); + sendPing(client, 0x2201); + assertEquals(0x2201, receiveResetMessageId(client)); + Stats after = proxy.stats(); + assertEquals(1L, after.partialReverseFlushes() - before.partialReverseFlushes()); + assertEquals(0L, after.reversed() - before.reversed()); + assertEquals(0, after.pendingRules()); + } + } + + @Test + void corruptedPingDoesNotPreventSubsequentValidCoapExchange() throws Exception { + try (DatagramSocket client = rawClient()) { + Stats before = proxy.stats(); + proxy.reverseRangeNext(Direction.CLIENT_TO_SERVER, 1, 0, 4); + sendPing(client, 0x2301); + sendPing(client, 0x2302); + assertEquals(0x2302, receiveResetMessageId(client)); + Stats after = proxy.stats(); + assertEquals(1L, after.modified() - before.modified()); + assertEquals(2L, after.clientToServerForwarded() - before.clientToServerForwarded()); + assertEquals(0, after.pendingRules()); + } + } + + private static CoapClient createClient() { + String original = getUri(); + String throughProxy = original.replace(":5683/", ":" + proxy.getListenPort() + "/"); + CoapClient client = new CoapClient(throughProxy); + client.setTimeout(15_000L); + return client; + } + + private static DatagramSocket rawClient() throws SocketException { + DatagramSocket client = new DatagramSocket(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0)); + client.setSoTimeout(5_000); + return client; + } + + private static void sendPing(DatagramSocket client, int messageId) throws IOException { + byte[] message = new byte[]{0x40, 0x00, (byte) (messageId >>> 8), (byte) messageId}; + client.send(new DatagramPacket(message, message.length, proxy.getListenAddress())); + } + + private static int receiveResetMessageId(DatagramSocket client) throws IOException { + byte[] buffer = new byte[1_024]; + DatagramPacket packet = new DatagramPacket(buffer, buffer.length); + client.receive(packet); + assertEquals(4, packet.getLength(), "CoAP empty Reset must have exactly four bytes"); + assertEquals(0x70, buffer[0] & 0xff, "Confirmable CoAP ping must receive Reset (RST)"); + assertEquals(0, buffer[1] & 0xff, "Reset must have empty code"); + return (buffer[2] & 0xff) << 8 | (buffer[3] & 0xff); + } + + private static void assertSuccessfulPut(CoapResponse response) { + assertNotNull(response, "PUT must receive a response"); + assertTrue(response.getCode() == CoAP.ResponseCode.CREATED || response.getCode() == CoAP.ResponseCode.CHANGED, + "Unexpected PUT response: " + response.getCode()); + } + + private static byte[] payload(int length) { + byte[] bytes = new byte[length]; + for (int index = 0; index < bytes.length; index++) { + bytes[index] = (byte) ((index * 37) & 0xff); + } + return bytes; + } +} diff --git a/src/test/java/io/mapsmessaging/network/protocol/impl/coap/UdpFaultProxyCoapReliabilityIT.java b/src/test/java/io/mapsmessaging/network/protocol/impl/coap/UdpFaultProxyCoapReliabilityIT.java new file mode 100644 index 0000000000..a23e796f04 --- /dev/null +++ b/src/test/java/io/mapsmessaging/network/protocol/impl/coap/UdpFaultProxyCoapReliabilityIT.java @@ -0,0 +1,222 @@ +/* + * Copyright [ 2024 - 2026 ] MapsMessaging B.V. + * Licensed under the Apache License, Version 2.0 with the Commons Clause. + */ +package io.mapsmessaging.network.protocol.impl.coap; + +import io.mapsmessaging.test.network.UdpFaultProxy; +import io.mapsmessaging.test.network.UdpFaultProxy.Direction; +import io.mapsmessaging.test.network.UdpFaultProxy.Stats; +import java.net.InetSocketAddress; +import java.time.Duration; +import java.util.Arrays; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.eclipse.californium.core.CoapClient; +import org.eclipse.californium.core.CoapHandler; +import org.eclipse.californium.core.CoapObserveRelation; +import org.eclipse.californium.core.CoapResponse; +import org.eclipse.californium.core.coap.CoAP; +import org.eclipse.californium.core.coap.Request; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import static org.junit.jupiter.api.Assertions.*; + +/** Additional real-CoAP network fault scenarios using the shared UDP proxy. */ +class UdpFaultProxyCoapReliabilityIT extends BaseCoapTest { + private static UdpFaultProxy proxy; + + @BeforeAll + static void startProxy() throws Exception { + proxy = new UdpFaultProxy(0, new InetSocketAddress("127.0.0.1", 5683)); + } + + @BeforeEach + void before() { + proxy.resetRules(); + } + + @AfterEach + void after() { + proxy.resetRules(); + } + + @AfterAll + static void stopProxy() { + if (proxy != null) { + proxy.close(); + } + } + + private static CoapClient client(String uri) { + CoapClient client = new CoapClient(uri.replace(":5683/", ":" + proxy.getListenPort() + "/")); + client.setTimeout(15_000L); + return client; + } + + private static final class ManagedClient implements AutoCloseable { + private final CoapClient client; + private ManagedClient(String uri) { + client = UdpFaultProxyCoapReliabilityIT.client(uri); + } + @Override + public void close() { + client.shutdown(); + } + } + + private static byte[] payload(int size, int seed) { + byte[] data = new byte[size]; + for (int index = 0; index < size; index++) { + data[index] = (byte) (seed + index * 31); + } + return data; + } + + private static void successfulPut(CoapResponse response) { + assertNotNull(response); + assertTrue(response.getCode() == CoAP.ResponseCode.CHANGED || response.getCode() == CoAP.ResponseCode.CREATED, + "Unexpected PUT response " + response.getCode()); + } + + @Test + void intermediateBlock1ResponseLossDoesNotCorruptReassembly() throws Exception { + try (ManagedClient managed = new ManagedClient(getUri())) { + CoapClient client = managed.client; + byte[] content = payload(1024, 17); + int blockSize = 256; + Stats before = proxy.stats(); + for (int block = 0; block < 4; block++) { + Request request = Request.newPut(); + request.setPayload(Arrays.copyOfRange(content, block * blockSize, (block + 1) * blockSize)); + request.getOptions().setBlock1(4, block < 3, block); + if (block == 1) { + proxy.dropNext(Direction.SERVER_TO_CLIENT, 1); + } + CoapResponse response = client.advanced(request); + assertNotNull(response, "Block " + block + " must complete despite lost Continue"); + if (block < 3) { + assertEquals(CoAP.ResponseCode.CONTINUE, response.getCode()); + } else { + successfulPut(response); + } + } + Stats after = proxy.stats(); + assertEquals(1, after.dropped() - before.dropped(), "The intermediate Continue must be lost"); + assertTrue(after.clientToServerReceived() - before.clientToServerReceived() >= 5, + "A block must be retransmitted"); + assertEquals(0, after.pendingRules()); + CoapResponse result = client.get(); + assertNotNull(result); + assertArrayEquals(content, result.getPayload()); + } + } + + @Test + void repeatedConfirmablePutIsIdempotentAtResourceState() throws Exception { + try (ManagedClient managed = new ManagedClient(getUri())) { + CoapClient client = managed.client; + byte[] value = payload(111, 29); + Stats before = proxy.stats(); + proxy.dropNext(Direction.SERVER_TO_CLIENT, 1); + successfulPut(client.put(value, 0)); + Stats after = proxy.stats(); + assertEquals(1, after.dropped() - before.dropped(), "Response loss must occur"); + assertTrue(after.clientToServerReceived() - before.clientToServerReceived() >= 2, + "Client must retransmit the confirmable request"); + CoapResponse get = client.get(); + assertNotNull(get); + assertArrayEquals(value, get.getPayload()); + assertEquals(0, proxy.stats().pendingRules()); + } + } + + @Test + void observeRecoversAfterLostNotificationAndCancels() throws Exception { + String uri = getUri(); + try (ManagedClient sub = new ManagedClient(uri); ManagedClient pub = new ManagedClient(uri)) { + CoapClient subscriber = sub.client; + CoapClient publisher = pub.client; + CountDownLatch initial = new CountDownLatch(1); + AtomicInteger notifications = new AtomicInteger(); + java.util.concurrent.atomic.AtomicBoolean observeError = new java.util.concurrent.atomic.AtomicBoolean(); + CoapObserveRelation relation = subscriber.observe(new CoapHandler() { + @Override + public void onLoad(CoapResponse response) { + notifications.incrementAndGet(); + initial.countDown(); + } + @Override + public void onError() { + observeError.set(true); + initial.countDown(); + } + }); + assertNotNull(relation); + successfulPut(publisher.put(payload(31, 0), 0)); + assertTrue(initial.await(10, TimeUnit.SECONDS), "Observe must receive the initial PUT notification before fault injection"); + assertFalse(observeError.get(), "Observe reported an error during establishment"); + assertTrue(notifications.get() > 0, "Observe must deliver a notification before fault injection"); + Stats before = proxy.stats(); + proxy.dropNext(Direction.SERVER_TO_CLIENT, 1); + successfulPut(publisher.put(payload(31, 1), 0)); + assertEquals(1, proxy.stats().dropped() - before.dropped(), "One server datagram must be lost"); + int observedBeforeRecovery = notifications.get(); + successfulPut(publisher.put(payload(31, 2), 0)); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); + while (notifications.get() <= observedBeforeRecovery && System.nanoTime() < deadline) { + Thread.sleep(20); + } + assertTrue(notifications.get() > observedBeforeRecovery, "Observe must receive a later update"); + relation.proactiveCancel(); + assertEquals(0, proxy.stats().pendingRules()); + } + } + + @Test + void concurrentClientsKeepPayloadsSeparateDuringReordering() throws Exception { + String firstUri = getUri(); + String secondUri = getUri(); + try (ManagedClient one = new ManagedClient(firstUri); ManagedClient two = new ManagedClient(secondUri)) { + CoapClient first = one.client; + CoapClient second = two.client; + byte[] firstPayload = payload(127, 3); + byte[] secondPayload = payload(129, 79); + Stats before = proxy.stats(); + proxy.reverseNext(Direction.CLIENT_TO_SERVER, 2, Duration.ofSeconds(2)); + CountDownLatch start = new CountDownLatch(1); + java.util.concurrent.atomic.AtomicReference failure = new java.util.concurrent.atomic.AtomicReference<>(); + Thread firstThread = new Thread(() -> putConcurrently(first, firstPayload, start, failure)); + Thread secondThread = new Thread(() -> putConcurrently(second, secondPayload, start, failure)); + firstThread.start(); + secondThread.start(); + start.countDown(); + firstThread.join(20_000); + secondThread.join(20_000); + assertFalse(firstThread.isAlive() || secondThread.isAlive(), "Concurrent PUTs must finish"); + if (failure.get() != null) { + fail("Concurrent PUT failed", failure.get()); + } + Stats after = proxy.stats(); + assertEquals(2, after.reversed() - before.reversed(), "Both datagrams must be reordered"); + assertEquals(0, after.pendingRules()); + assertArrayEquals(firstPayload, first.get().getPayload()); + assertArrayEquals(secondPayload, second.get().getPayload()); + assertEquals(0, proxy.stats().ioFailures()); + } + } + + private static void putConcurrently(CoapClient client, byte[] value, CountDownLatch start, + java.util.concurrent.atomic.AtomicReference failure) { + try { + start.await(); + successfulPut(client.put(value, 0)); + } catch (Throwable exception) { + failure.compareAndSet(null, exception); + } + } +} diff --git a/src/test/java/io/mapsmessaging/network/protocol/impl/coap/packet/CoapProtocolConformanceTest.java b/src/test/java/io/mapsmessaging/network/protocol/impl/coap/packet/CoapProtocolConformanceTest.java index 5cfecae712..d1c87e2737 100644 --- a/src/test/java/io/mapsmessaging/network/protocol/impl/coap/packet/CoapProtocolConformanceTest.java +++ b/src/test/java/io/mapsmessaging/network/protocol/impl/coap/packet/CoapProtocolConformanceTest.java @@ -7,7 +7,6 @@ import io.mapsmessaging.network.protocol.impl.coap.packet.options.Block; import java.io.IOException; import java.nio.ByteBuffer; -import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Tag; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; @@ -102,7 +101,6 @@ void contentResponseIsNotDispatchedAsFetchRequest() throws Exception { } @Test - @Disabled("Known CoAP empty-message validation gap: MSG-369") @ProtocolRequirement( specification = RFC7252, value = "RFC 7252 §4.1: an Empty message has Code 0.00, TKL 0, and no payload", @@ -112,7 +110,6 @@ void emptyMessageWithTokenIsRejected() { } @Test - @Disabled("Known CoAP Observe option-length validation gap: MSG-369") @ProtocolRequirement( specification = RFC7641, value = "RFC 7641 §2: Observe Option format is uint with length 0..3 bytes", @@ -124,6 +121,15 @@ void fourByteObserveOptionIsRejected() { })); } + @Test + @ProtocolRequirement( + specification = RFC7959, + value = "RFC 7959 §2.9: code 4.08 identifies an incomplete block-wise upload", + source = ProtocolRequirement.COAP_RFC7959_SOURCE) + void incompleteBlockResponseCodeRoundTrips() { + assertEquals(Code.REQUEST_ENTITY_INCOMPLETE, Code.valueOf((byte) 0x88)); + } + @ParameterizedTest @ValueSource(ints = {0,1,2,3,4,5,6}) @ProtocolRequirement( diff --git a/src/test/java/io/mapsmessaging/network/protocol/impl/mavlink/MavlinkUdpFaultIT.java b/src/test/java/io/mapsmessaging/network/protocol/impl/mavlink/MavlinkUdpFaultIT.java new file mode 100644 index 0000000000..ce014e424f --- /dev/null +++ b/src/test/java/io/mapsmessaging/network/protocol/impl/mavlink/MavlinkUdpFaultIT.java @@ -0,0 +1,114 @@ +/* + * Copyright [ 2024 - 2026 ] MapsMessaging B.V. + * Licensed under the Apache License, Version 2.0 with the Commons Clause. + */ +package io.mapsmessaging.network.protocol.impl.mavlink; + +import com.google.gson.JsonObject; +import com.google.gson.JsonParser; +import io.mapsmessaging.api.MessageListener; +import io.mapsmessaging.api.Session; +import io.mapsmessaging.api.SubscriptionContextBuilder; +import io.mapsmessaging.api.features.ClientAcknowledgement; +import io.mapsmessaging.api.features.QualityOfService; +import io.mapsmessaging.api.message.Message; +import io.mapsmessaging.engine.destination.subscription.SubscriptionContext; +import io.mapsmessaging.test.BaseTestConfig; +import io.mapsmessaging.test.network.UdpFaultProxy; +import io.mapsmessaging.test.network.UdpFaultProxy.Direction; +import java.net.DatagramPacket; +import java.net.DatagramSocket; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.Test; +import static org.junit.jupiter.api.Assertions.*; + +/** Validate real MAVLink UDP delivery and isolation under deliberate faults. */ +class MavlinkUdpFaultIT extends BaseTestConfig { + private static final InetSocketAddress MAPS = new InetSocketAddress("127.0.0.1", 14550); + + @Test + void droppedHeartbeatDoesNotCorruptNextValidHeartbeat() throws Exception { + verifyRecovery(false); + } + + @Test + void corruptedHeartbeatDoesNotCorruptNextValidHeartbeat() throws Exception { + verifyRecovery(true); + } + + private void verifyRecovery(boolean corrupt) throws Exception { + List messages = new CopyOnWriteArrayList<>(); + CountDownLatch delivered = new CountDownLatch(1); + MessageListener listener = event -> { + messages.add(event.getMessage()); + delivered.countDown(); + event.getCompletionTask().run(); + }; + Session session = createSession("mavlink-proxy-" + System.nanoTime(), 60, 60, false, listener); + assertNotNull(session); + SubscriptionContext subscription = new SubscriptionContextBuilder("/mavlink/#", ClientAcknowledgement.AUTO) + .setQos(QualityOfService.AT_MOST_ONCE).setReceiveMaximum(100).build(); + try (UdpFaultProxy proxy = new UdpFaultProxy(0, MAPS); + DatagramSocket sender = new DatagramSocket()) { + session.addSubscription(subscription); + if (corrupt) { + proxy.reverseRangeNext(Direction.CLIENT_TO_SERVER, 1, 0, 4); + } else { + proxy.dropNext(Direction.CLIENT_TO_SERVER, 1); + } + send(sender, proxy, heartbeat(31, 41, 1)); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(2); + while (proxy.stats().clientToServerReceived() < 1 && System.nanoTime() < deadline) { + Thread.sleep(5); + } + assertEquals(1, corrupt ? proxy.stats().modified() : proxy.stats().dropped(), "Fault must actually fire"); + send(sender, proxy, heartbeat(32, 41, 1)); + assertTrue(delivered.await(5, TimeUnit.SECONDS), "Valid heartbeat following network fault must be delivered"); + assertEquals(1, messages.size(), "Malformed or dropped datagram must not create a second heartbeat"); + JsonObject root = JsonParser.parseString(new String(messages.getFirst().getOpaqueData(), StandardCharsets.UTF_8)).getAsJsonObject(); + JsonObject mavlink = root.getAsJsonObject("mavlink"); + assertEquals(32, mavlink.get("sequence").getAsInt()); + assertEquals(41, mavlink.get("systemId").getAsInt()); + assertEquals(0, mavlink.get("messageId").getAsInt()); + assertEquals(0, proxy.stats().ioFailures()); + session.removeSubscription(subscription.getKey()); + } finally { + close(session); + } + } + + private static void send(DatagramSocket socket, UdpFaultProxy proxy, byte[] data) throws Exception { + socket.send(new DatagramPacket(data, data.length, proxy.getListenAddress())); + } + + private static byte[] heartbeat(int sequence, int systemId, int componentId) { + byte[] frame = new byte[17]; + frame[0] = (byte) 0xfe; + frame[1] = 9; + frame[2] = (byte) sequence; + frame[3] = (byte) systemId; + frame[4] = (byte) componentId; + frame[5] = 0; + frame[14] = 3; + int crc = 0xffff; + for (int index = 1; index < 15; index++) { + crc = accumulate(frame[index], crc); + } + crc = accumulate((byte) 50, crc); + frame[15] = (byte) crc; + frame[16] = (byte) (crc >>> 8); + return frame; + } + + private static int accumulate(byte input, int crc) { + int value = (input ^ (crc & 0xff)) & 0xff; + value = (value ^ ((value << 4) & 0xff)) & 0xff; + return ((crc >>> 8) ^ (value << 8) ^ (value << 3) ^ (value >>> 4)) & 0xffff; + } +} diff --git a/src/test/java/io/mapsmessaging/network/protocol/impl/mqtt_sn/MqttSn12UdpFaultIT.java b/src/test/java/io/mapsmessaging/network/protocol/impl/mqtt_sn/MqttSn12UdpFaultIT.java new file mode 100644 index 0000000000..8e6b3f83f1 --- /dev/null +++ b/src/test/java/io/mapsmessaging/network/protocol/impl/mqtt_sn/MqttSn12UdpFaultIT.java @@ -0,0 +1,79 @@ +/* + * Copyright [ 2024 - 2026 ] MapsMessaging B.V. + * Licensed under the Apache License, Version 2.0 with the Commons Clause. + */ +package io.mapsmessaging.network.protocol.impl.mqtt_sn; + +import io.mapsmessaging.test.network.UdpFaultProxy; +import io.mapsmessaging.test.network.UdpFaultProxy.Direction; +import java.net.DatagramPacket; +import java.net.DatagramSocket; +import java.net.InetSocketAddress; +import java.net.SocketTimeoutException; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import org.junit.jupiter.api.Test; +import static org.junit.jupiter.api.Assertions.*; + +/** Real MQTT-SN 1.2 wire exchanges under deterministic UDP loss and corruption. */ +class MqttSn12UdpFaultIT extends BaseMqttSnConfig { + + @Test + void lostConnectIsRetriedWithoutCreatingAnInvalidSession() throws Exception { + try (UdpFaultProxy proxy = new UdpFaultProxy(0, new InetSocketAddress("127.0.0.1", 1884)); + DatagramSocket socket = new DatagramSocket()) { + socket.setSoTimeout(2_000); + byte[] connect = connectPacket("sn12-loss-" + System.nanoTime()); + proxy.dropNext(Direction.CLIENT_TO_SERVER, 1); + send(socket, proxy, connect); + assertThrows(SocketTimeoutException.class, () -> receive(socket)); + assertEquals(1L, proxy.stats().dropped(), "The first CONNECT must actually be lost"); + send(socket, proxy, connect); + byte[] reply = receive(socket); + assertEquals(0x05, reply[1] & 0xff, "Recovered CONNECT must receive CONNACK"); + assertEquals(0x00, reply[2] & 0xff, "CONNACK must report accepted"); + assertEquals(0, proxy.stats().pendingRules()); + } + } + + @Test + void malformedConnectDoesNotPreventNextValidConnect() throws Exception { + try (UdpFaultProxy proxy = new UdpFaultProxy(0, new InetSocketAddress("127.0.0.1", 1884)); + DatagramSocket socket = new DatagramSocket()) { + socket.setSoTimeout(600); + byte[] connect = connectPacket("sn12-corrupt-" + System.nanoTime()); + proxy.reverseRangeNext(Direction.CLIENT_TO_SERVER, 1, 0, 4); + send(socket, proxy, connect); + assertThrows(SocketTimeoutException.class, () -> receive(socket), + "Malformed CONNECT must not be acknowledged"); + assertEquals(1L, proxy.stats().modified(), "The corrupted datagram must pass through the proxy"); + send(socket, proxy, connect); + assertEquals(0x05, receive(socket)[1] & 0xff, "Valid CONNECT must still work"); + assertEquals(0, proxy.stats().pendingRules()); + } + } + + private static byte[] connectPacket(String name) { + byte[] identity = name.getBytes(StandardCharsets.UTF_8); + byte[] result = new byte[6 + identity.length]; + result[0] = (byte) result.length; + result[1] = 0x04; + result[2] = 0x04; + result[3] = 0x01; + result[4] = 0x00; + result[5] = 0x3c; + System.arraycopy(identity, 0, result, 6, identity.length); + return result; + } + + private static void send(DatagramSocket socket, UdpFaultProxy proxy, byte[] payload) throws Exception { + socket.send(new DatagramPacket(payload, payload.length, proxy.getListenAddress())); + } + + private static byte[] receive(DatagramSocket socket) throws Exception { + byte[] bytes = new byte[1_024]; + DatagramPacket packet = new DatagramPacket(bytes, bytes.length); + socket.receive(packet); + return Arrays.copyOf(bytes, packet.getLength()); + } +} diff --git a/src/test/java/io/mapsmessaging/network/protocol/impl/mqtt_sn/v2_0/MqttSn2UdpFaultIT.java b/src/test/java/io/mapsmessaging/network/protocol/impl/mqtt_sn/v2_0/MqttSn2UdpFaultIT.java new file mode 100644 index 0000000000..66798bdb4f --- /dev/null +++ b/src/test/java/io/mapsmessaging/network/protocol/impl/mqtt_sn/v2_0/MqttSn2UdpFaultIT.java @@ -0,0 +1,82 @@ +/* + * Copyright [ 2024 - 2026 ] MapsMessaging B.V. + * Licensed under the Apache License, Version 2.0 with the Commons Clause. + */ +package io.mapsmessaging.network.protocol.impl.mqtt_sn.v2_0; + +import io.mapsmessaging.mqttsn.ConnectOptions; +import io.mapsmessaging.mqttsn.DecodedPacket; +import io.mapsmessaging.mqttsn.MqttSnCodec; +import io.mapsmessaging.mqttsn.PacketType; +import io.mapsmessaging.network.protocol.impl.mqtt_sn.BaseMqttSnConfig; +import io.mapsmessaging.test.network.UdpFaultProxy; +import io.mapsmessaging.test.network.UdpFaultProxy.Direction; +import java.net.InetSocketAddress; +import java.net.DatagramPacket; +import java.net.DatagramSocket; +import java.nio.ByteBuffer; +import java.util.Arrays; +import java.time.Duration; +import java.util.UUID; +import org.junit.jupiter.api.Test; +import static org.junit.jupiter.api.Assertions.*; + +/** Real v2 reference-client UDP exchanges through the same fault proxy as CoAP. */ +class MqttSn2UdpFaultIT extends BaseMqttSnConfig { + @Test + void droppedConnectCanBeRetriedByIndependentClient() throws Exception { + try (UdpFaultProxy proxy = new UdpFaultProxy(0, new InetSocketAddress("127.0.0.1", 1884)); + DatagramSocket client = new DatagramSocket()) { + byte[] connect = MqttSnCodec.encodeConnect(new ConnectOptions( + true, false, false, 1, 60, 0, "v2-loss-" + UUID.randomUUID())); + proxy.dropNext(Direction.CLIENT_TO_SERVER, 1); + send(client, proxy, connect); + // The reference client's read timeout is the observable proof that the first request did not arrive. + awaitDrop(proxy); + assertEquals(1L, proxy.stats().dropped(), "First CONNECT must be dropped"); + send(client, proxy, connect); + DecodedPacket reply = receive(client); + assertEquals(PacketType.CONNACK, reply.type()); + assertEquals(0, MqttSnCodec.decodeConnAck(reply).reasonCode()); + assertEquals(0, proxy.stats().pendingRules()); + } + } + + @Test + void lostConnackCanBeRecoveredByResendingSameConnect() throws Exception { + try (UdpFaultProxy proxy = new UdpFaultProxy(0, new InetSocketAddress("127.0.0.1", 1884)); + DatagramSocket client = new DatagramSocket()) { + byte[] connect = MqttSnCodec.encodeConnect(new ConnectOptions( + true, false, false, 1, 60, 0, "v2-ack-loss-" + UUID.randomUUID())); + proxy.dropNext(Direction.SERVER_TO_CLIENT, 1); + send(client, proxy, connect); + awaitDrop(proxy); + assertEquals(1L, proxy.stats().dropped(), "First CONNACK must be dropped"); + send(client, proxy, connect); + DecodedPacket reply = receive(client); + assertEquals(PacketType.CONNACK, reply.type()); + assertEquals(0, MqttSnCodec.decodeConnAck(reply).reasonCode()); + assertEquals(0, proxy.stats().pendingRules()); + } + } + + private static void awaitDrop(UdpFaultProxy proxy) throws InterruptedException { + long deadline = System.nanoTime() + Duration.ofSeconds(2).toNanos(); + while (proxy.stats().dropped() == 0 && System.nanoTime() < deadline) { + Thread.sleep(5); + } + assertEquals(1L, proxy.stats().dropped(), "Fault must actually be injected before retry"); + } + + private static void send(DatagramSocket client, UdpFaultProxy proxy, byte[] packet) throws Exception { + client.send(new DatagramPacket(packet, packet.length, proxy.getListenAddress())); + } + + private static DecodedPacket receive(DatagramSocket client) throws Exception { + client.setSoTimeout(5_000); + byte[] bytes = new byte[2_048]; + DatagramPacket packet = new DatagramPacket(bytes, bytes.length); + client.receive(packet); + return MqttSnCodec.decode(ByteBuffer.wrap(Arrays.copyOf(bytes, packet.getLength()))); + } +} diff --git a/src/test/java/io/mapsmessaging/test/network/UdpFaultProxy.java b/src/test/java/io/mapsmessaging/test/network/UdpFaultProxy.java new file mode 100644 index 0000000000..8d3e25f679 --- /dev/null +++ b/src/test/java/io/mapsmessaging/test/network/UdpFaultProxy.java @@ -0,0 +1,477 @@ +/* + * 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.test.network; + +import java.io.IOException; +import java.net.DatagramPacket; +import java.net.DatagramSocket; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.net.SocketException; +import java.net.SocketTimeoutException; +import java.time.Duration; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.EnumMap; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.TimeUnit; + +/** + * Test-only, protocol-independent UDP forwarder with deterministic packet faults. + * + *

Rules are FIFO within each direction and shared by all clients of this proxy. + * Each datagram consumes at most one rule. Reverse batches retain the original + * destination and socket for each packet, so reordering never mixes responses + * between clients. A partial reverse batch is forwarded in arrival order after + * its deadline; the rule is then retired. The deadline starts on the first match. + * + *

All sockets and rules are owned by this instance. A JUnit class can create + * the proxy in {@code @BeforeAll} and close it in {@code @AfterAll}, clearing + * fault rules between tests using {@link #resetRules()}. The proxy listens only + * on the loopback interface; no global test-suite startup is required. + */ +public final class UdpFaultProxy implements AutoCloseable { + + public enum Direction { + CLIENT_TO_SERVER, + SERVER_TO_CLIENT + } + + public record Stats( + long clientToServerReceived, + long serverToClientReceived, + long clientToServerForwarded, + long serverToClientForwarded, + long dropped, + long modified, + long reversed, + long partialReverseFlushes, + long invalidMutations, + long ioFailures, + int activePeers, + int pendingRules) { + } + + private enum FaultType { + DROP, + REVERSE_PACKETS, + INSERT_BYTES, + REVERSE_BYTES + } + + private static final int MAX_DATAGRAM_SIZE = 65_507; + private static final int MAX_BATCH_PACKETS = 1_024; + private static final int MAX_CLIENT_PEERS = 64; + private static final int MAX_PENDING_RULES_PER_DIRECTION = 32; + private static final int SOCKET_POLL_MILLIS = 200; + + private final Object lock = new Object(); + private final InetSocketAddress target; + private final DatagramSocket listener; + private final Thread listenerThread; + private final ScheduledExecutorService scheduler; + private final Map peers = new HashMap<>(); + private final EnumMap> rules = new EnumMap<>(Direction.class); + private boolean closed; + private long c2sReceived; + private long s2cReceived; + private long c2sForwarded; + private long s2cForwarded; + private long dropped; + private long modified; + private long reversed; + private long partialReverseFlushes; + private long invalidMutations; + private long ioFailures; + + /** Create a proxy on a known port; port 0 requests an ephemeral test port. */ + public UdpFaultProxy(int listenPort, InetSocketAddress target) throws SocketException { + if (listenPort < 0 || listenPort > 65_535) { + throw new IllegalArgumentException("Invalid listen port: " + listenPort); + } + this.target = Objects.requireNonNull(target, "target"); + if (target.isUnresolved() || target.getPort() == 0 || !target.getAddress().isLoopbackAddress()) { + throw new IllegalArgumentException("Target must be a resolved, nonzero loopback UDP endpoint"); + } + listener = new DatagramSocket(new InetSocketAddress(InetAddress.getLoopbackAddress(), listenPort)); + try { + listener.setSoTimeout(SOCKET_POLL_MILLIS); + } catch (SocketException exception) { + listener.close(); + throw exception; + } + for (Direction direction : Direction.values()) { + rules.put(direction, new ArrayDeque<>()); + } + ThreadFactory schedulerThreads = runnable -> { + Thread thread = new Thread(runnable, "udp-fault-proxy-timer-" + listener.getLocalPort()); + thread.setDaemon(true); + return thread; + }; + scheduler = Executors.newSingleThreadScheduledExecutor(schedulerThreads); + listenerThread = new Thread(this::readClients, "udp-fault-proxy-listener-" + listener.getLocalPort()); + listenerThread.setDaemon(true); + listenerThread.start(); + } + + public int getListenPort() { + return listener.getLocalPort(); + } + + public InetSocketAddress getListenAddress() { + return new InetSocketAddress(listener.getLocalAddress(), listener.getLocalPort()); + } + + public void dropNext(Direction direction, int count) { + addRule(direction, new Rule(FaultType.DROP, count, 0, 0, null, null)); + } + + /** Reverse arrival order for a batch, or flush a partial batch unchanged at timeout. */ + public void reverseNext(Direction direction, int count, Duration timeout) { + Objects.requireNonNull(timeout, "timeout"); + if (timeout.isNegative() || timeout.isZero() || timeout.toMillis() <= 0) { + throw new IllegalArgumentException("Reverse timeout must be at least 1 ms"); + } + addRule(direction, new Rule(FaultType.REVERSE_PACKETS, count, 0, 0, null, timeout)); + } + + /** Insert bytes before the byte at offset (offset == datagram length appends). */ + public void injectNext(Direction direction, int count, int offset, byte[] bytes) { + if (offset < 0) { + throw new IllegalArgumentException("Offset cannot be negative"); + } + byte[] copy = Objects.requireNonNull(bytes, "bytes").clone(); + if (copy.length == 0) { + throw new IllegalArgumentException("Inserted bytes must not be empty"); + } + addRule(direction, new Rule(FaultType.INSERT_BYTES, count, offset, 0, copy, null)); + } + + /** Reverse a zero-based, start-inclusive, end-exclusive payload byte range. */ + public void reverseRangeNext(Direction direction, int count, int startInclusive, int endExclusive) { + if (startInclusive < 0 || endExclusive <= startInclusive) { + throw new IllegalArgumentException("Invalid byte range"); + } + addRule(direction, new Rule(FaultType.REVERSE_BYTES, count, startInclusive, endExclusive, null, null)); + } + + /** Stop applying queued rules. Buffered reverse packets are forwarded unchanged. */ + public void resetRules() { + synchronized (lock) { + ensureOpen(); + for (ArrayDeque queue : rules.values()) { + while (!queue.isEmpty()) { + Rule rule = queue.removeFirst(); + cancelTimer(rule); + forwardBuffered(rule, false); + } + } + } + } + + public Stats stats() { + synchronized (lock) { + int ruleCount = 0; + for (ArrayDeque queue : rules.values()) { + ruleCount += queue.size(); + } + return new Stats(c2sReceived, s2cReceived, c2sForwarded, s2cForwarded, + dropped, modified, reversed, partialReverseFlushes, invalidMutations, + ioFailures, peers.size(), ruleCount); + } + } + + private void addRule(Direction direction, Rule rule) { + Objects.requireNonNull(direction, "direction"); + if (rule.remaining <= 0 || rule.remaining > MAX_BATCH_PACKETS) { + throw new IllegalArgumentException("Rule count must be 1.." + MAX_BATCH_PACKETS); + } + synchronized (lock) { + ensureOpen(); + ArrayDeque queue = rules.get(direction); + if (queue.size() >= MAX_PENDING_RULES_PER_DIRECTION) { + throw new IllegalStateException("Too many pending UDP fault rules"); + } + queue.addLast(rule); + } + } + + private void ensureOpen() { + if (closed) { + throw new IllegalStateException("UDP proxy has been closed"); + } + } + + private void readClients() { + receiveLoop(listener, null); + } + + private void receiveLoop(DatagramSocket socket, Peer fixedPeer) { + while (!socket.isClosed()) { + byte[] buffer = new byte[MAX_DATAGRAM_SIZE]; + DatagramPacket datagram = new DatagramPacket(buffer, buffer.length); + try { + socket.receive(datagram); + byte[] data = Arrays.copyOfRange(datagram.getData(), datagram.getOffset(), datagram.getOffset() + datagram.getLength()); + synchronized (lock) { + if (closed) { + return; + } + if (fixedPeer == null) { + c2sReceived++; + Peer peer = peers.get(datagram.getSocketAddress()); + if (peer == null) { + if (peers.size() >= MAX_CLIENT_PEERS) { + dropped++; + continue; + } + peer = createPeer(datagram.getSocketAddress()); + peers.put(datagram.getSocketAddress(), peer); + peer.thread.start(); + } + process(new Envelope(data, peer.socket, target, Direction.CLIENT_TO_SERVER)); + } else { + s2cReceived++; + process(new Envelope(data, listener, fixedPeer.clientAddress, Direction.SERVER_TO_CLIENT)); + } + } + } catch (SocketTimeoutException ignored) { + // Periodic poll permits deterministic shutdown. + } catch (SocketException exception) { + if (!socket.isClosed()) { + countIoFailure(); + } + return; + } catch (IOException exception) { + countIoFailure(); + } + } + } + + private Peer createPeer(SocketAddress client) throws SocketException { + DatagramSocket upstream = new DatagramSocket(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0)); + try { + upstream.connect(target); + upstream.setSoTimeout(SOCKET_POLL_MILLIS); + } catch (SocketException exception) { + upstream.close(); + throw exception; + } + Peer peer = new Peer(client, upstream); + peer.thread = new Thread(() -> receiveLoop(upstream, peer), "udp-fault-proxy-peer-" + upstream.getLocalPort()); + peer.thread.setDaemon(true); + return peer; + } + + private void process(Envelope packet) { + ArrayDeque queue = rules.get(packet.direction); + Rule rule = queue.peekFirst(); + if (rule == null) { + forward(packet); + return; + } + + rule.remaining--; + switch (rule.type) { + case DROP: + dropped++; + break; + case INSERT_BYTES: + if (rule.start > packet.data.length || packet.data.length + rule.bytes.length > MAX_DATAGRAM_SIZE) { + invalidMutations++; + dropped++; + } else { + byte[] result = new byte[packet.data.length + rule.bytes.length]; + System.arraycopy(packet.data, 0, result, 0, rule.start); + System.arraycopy(rule.bytes, 0, result, rule.start, rule.bytes.length); + System.arraycopy(packet.data, rule.start, result, rule.start + rule.bytes.length, packet.data.length - rule.start); + modified++; + forward(packet.withData(result)); + } + break; + case REVERSE_BYTES: + if (rule.end > packet.data.length) { + invalidMutations++; + dropped++; + } else { + byte[] result = packet.data.clone(); + for (int first = rule.start, last = rule.end - 1; first < last; first++, last--) { + byte value = result[first]; + result[first] = result[last]; + result[last] = value; + } + modified++; + forward(packet.withData(result)); + } + break; + case REVERSE_PACKETS: + rule.buffer.add(packet); + if (rule.timer == null) { + rule.timer = scheduler.schedule(() -> expireReverse(packet.direction, rule), rule.timeout.toMillis(), TimeUnit.MILLISECONDS); + } + if (rule.remaining == 0) { + cancelTimer(rule); + forwardBuffered(rule, true); + } + break; + } + + if (rule.remaining == 0) { + queue.removeFirst(); + } + } + + private void expireReverse(Direction direction, Rule rule) { + synchronized (lock) { + if (closed || rules.get(direction).peekFirst() != rule) { + return; + } + rules.get(direction).removeFirst(); + partialReverseFlushes++; + forwardBuffered(rule, false); + } + } + + private void forwardBuffered(Rule rule, boolean reverse) { + if (reverse && rule.buffer.size() > 1) { + reversed += rule.buffer.size(); + } + if (reverse) { + for (int index = rule.buffer.size() - 1; index >= 0; index--) { + forward(rule.buffer.get(index)); + } + } else { + for (Envelope packet : rule.buffer) { + forward(packet); + } + } + rule.buffer.clear(); + } + + private void forward(Envelope packet) { + try { + DatagramPacket datagram = new DatagramPacket(packet.data, packet.data.length, packet.destination); + packet.socket.send(datagram); + if (packet.direction == Direction.CLIENT_TO_SERVER) { + c2sForwarded++; + } else { + s2cForwarded++; + } + } catch (IOException exception) { + ioFailures++; + } + } + + private void countIoFailure() { + synchronized (lock) { + if (!closed) { + ioFailures++; + } + } + } + + private void cancelTimer(Rule rule) { + if (rule.timer != null) { + rule.timer.cancel(false); + rule.timer = null; + } + } + + @Override + public void close() { + List threads = new ArrayList<>(); + synchronized (lock) { + if (closed) { + return; + } + closed = true; + listener.close(); + threads.add(listenerThread); + for (Peer peer : peers.values()) { + peer.socket.close(); + threads.add(peer.thread); + } + for (ArrayDeque queue : rules.values()) { + for (Rule rule : queue) { + cancelTimer(rule); + dropped += rule.buffer.size(); + rule.buffer.clear(); + } + queue.clear(); + } + } + scheduler.shutdownNow(); + for (Thread thread : threads) { + if (Thread.currentThread() == thread) { + continue; + } + try { + thread.join(1_000); + } catch (InterruptedException exception) { + Thread.currentThread().interrupt(); + break; + } + } + } + + private record Envelope(byte[] data, DatagramSocket socket, SocketAddress destination, Direction direction) { + private Envelope withData(byte[] replacement) { + return new Envelope(replacement, socket, destination, direction); + } + } + + private static final class Peer { + private final SocketAddress clientAddress; + private final DatagramSocket socket; + private Thread thread; + + private Peer(SocketAddress clientAddress, DatagramSocket socket) { + this.clientAddress = clientAddress; + this.socket = socket; + } + } + + private static final class Rule { + private final FaultType type; + private final int start; + private final int end; + private final byte[] bytes; + private final Duration timeout; + private final List buffer = new ArrayList<>(); + private ScheduledFuture timer; + private int remaining; + + private Rule(FaultType type, int remaining, int start, int end, byte[] bytes, Duration timeout) { + this.type = type; + this.remaining = remaining; + this.start = start; + this.end = end; + this.bytes = bytes; + this.timeout = timeout; + } + } +} diff --git a/src/test/java/io/mapsmessaging/test/network/UdpFaultProxyTest.java b/src/test/java/io/mapsmessaging/test/network/UdpFaultProxyTest.java new file mode 100644 index 0000000000..a8d533e874 --- /dev/null +++ b/src/test/java/io/mapsmessaging/test/network/UdpFaultProxyTest.java @@ -0,0 +1,341 @@ +/* + * 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.test.network; + +import java.io.IOException; +import java.net.DatagramPacket; +import java.net.DatagramSocket; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.SocketException; +import java.net.SocketTimeoutException; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.Arrays; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.jupiter.api.Test; + +import static io.mapsmessaging.test.network.UdpFaultProxy.Direction.CLIENT_TO_SERVER; +import static io.mapsmessaging.test.network.UdpFaultProxy.Direction.SERVER_TO_CLIENT; +import static org.junit.jupiter.api.Assertions.*; + +class UdpFaultProxyTest { + + @Test + void forwardsDatagramsAndRoutesResponsesToTheirOriginalClients() throws Exception { + try (EchoServer server = new EchoServer(); + UdpFaultProxy proxy = new UdpFaultProxy(0, server.address()); + DatagramSocket first = client(); + DatagramSocket second = client()) { + send(first, proxy, "client-one"); + send(second, proxy, "client-two"); + assertEquals("client-one", receive(first)); + assertEquals("client-two", receive(second)); + assertEquals(2, proxy.stats().clientToServerForwarded()); + assertEquals(2, proxy.stats().serverToClientForwarded()); + assertEquals(2, proxy.stats().activePeers()); + assertEquals(0, proxy.stats().ioFailures()); + } + } + + @Test + void dropsExactlyTheNextTwoClientDatagrams() throws Exception { + try (EchoServer server = new EchoServer(); + UdpFaultProxy proxy = new UdpFaultProxy(0, server.address()); + DatagramSocket client = client()) { + proxy.dropNext(CLIENT_TO_SERVER, 2); + send(client, proxy, "lost-one"); + send(client, proxy, "lost-two"); + send(client, proxy, "arrived"); + assertEquals("arrived", receive(client)); + assertEquals(3, proxy.stats().clientToServerReceived()); + assertEquals(1, proxy.stats().clientToServerForwarded()); + assertEquals(2, proxy.stats().dropped()); + assertEquals(0, proxy.stats().pendingRules()); + } + } + + @Test + void dropsServerRepliesWithoutDroppingClientRequests() throws Exception { + try (EchoServer server = new EchoServer(); + UdpFaultProxy proxy = new UdpFaultProxy(0, server.address()); + DatagramSocket client = client()) { + proxy.dropNext(SERVER_TO_CLIENT, 1); + send(client, proxy, "reply-dropped"); + send(client, proxy, "reply-delivered"); + assertEquals("reply-delivered", receive(client)); + assertEquals(2, proxy.stats().clientToServerForwarded()); + assertEquals(1, proxy.stats().serverToClientForwarded()); + assertEquals(1, proxy.stats().dropped()); + } + } + + @Test + void reversesTheNextThreeDatagramsInArrivalOrder() throws Exception { + try (EchoServer server = new EchoServer(); + UdpFaultProxy proxy = new UdpFaultProxy(0, server.address()); + DatagramSocket client = client()) { + proxy.reverseNext(CLIENT_TO_SERVER, 3, Duration.ofSeconds(2)); + send(client, proxy, "one"); + send(client, proxy, "two"); + send(client, proxy, "three"); + assertEquals("three", receive(client)); + assertEquals("two", receive(client)); + assertEquals("one", receive(client)); + assertEquals(3, proxy.stats().reversed()); + assertEquals(0, proxy.stats().partialReverseFlushes()); + } + } + + @Test + void canReverseRepliesWithoutAffectingClientToServerArrivalOrder() throws Exception { + try (EchoServer server = new EchoServer(); + UdpFaultProxy proxy = new UdpFaultProxy(0, server.address()); + DatagramSocket client = client()) { + proxy.reverseNext(SERVER_TO_CLIENT, 3, Duration.ofSeconds(2)); + send(client, proxy, "reply-one"); + send(client, proxy, "reply-two"); + send(client, proxy, "reply-three"); + assertEquals("reply-three", receive(client)); + assertEquals("reply-two", receive(client)); + assertEquals("reply-one", receive(client)); + assertEquals(3, proxy.stats().reversed()); + assertEquals(3, proxy.stats().clientToServerForwarded()); + } + } + + @Test + void independentProxiesCanControlDifferentUdpServerPorts() throws Exception { + try (EchoServer coapServer = new EchoServer(); + EchoServer mavlinkServer = new EchoServer(); + UdpFaultProxy coapProxy = new UdpFaultProxy(0, coapServer.address()); + UdpFaultProxy mavlinkProxy = new UdpFaultProxy(0, mavlinkServer.address()); + DatagramSocket coapClient = client(); + DatagramSocket mavlinkClient = client()) { + coapProxy.dropNext(CLIENT_TO_SERVER, 1); + send(coapClient, coapProxy, "coap-dropped"); + send(mavlinkClient, mavlinkProxy, "mavlink-unaffected"); + send(coapClient, coapProxy, "coap-forwarded"); + assertEquals("mavlink-unaffected", receive(mavlinkClient)); + assertEquals("coap-forwarded", receive(coapClient)); + assertEquals(1, coapProxy.stats().dropped()); + assertEquals(0, mavlinkProxy.stats().dropped()); + } + } + + @Test + void partialReverseBatchFlushesInOriginalOrderAfterDeadline() throws Exception { + try (EchoServer server = new EchoServer(); + UdpFaultProxy proxy = new UdpFaultProxy(0, server.address()); + DatagramSocket client = client()) { + proxy.reverseNext(CLIENT_TO_SERVER, 3, Duration.ofMillis(100)); + send(client, proxy, "first"); + send(client, proxy, "second"); + assertEquals("first", receive(client)); + assertEquals("second", receive(client)); + assertEquals(1, proxy.stats().partialReverseFlushes()); + assertEquals(0, proxy.stats().reversed()); + assertEquals(0, proxy.stats().pendingRules()); + } + } + + @Test + void insertsBytesAtSpecifiedPositionForExactlyTwoDatagrams() throws Exception { + try (EchoServer server = new EchoServer(); + UdpFaultProxy proxy = new UdpFaultProxy(0, server.address()); + DatagramSocket client = client()) { + proxy.injectNext(CLIENT_TO_SERVER, 2, 2, bytes("XX")); + send(client, proxy, "abcd"); + send(client, proxy, "1234"); + send(client, proxy, "normal"); + assertEquals("abXXcd", receive(client)); + assertEquals("12XX34", receive(client)); + assertEquals("normal", receive(client)); + assertEquals(2, proxy.stats().modified()); + } + } + + @Test + void reversesOnlyTheRequestedByteRange() throws Exception { + try (EchoServer server = new EchoServer(); + UdpFaultProxy proxy = new UdpFaultProxy(0, server.address()); + DatagramSocket client = client()) { + proxy.reverseRangeNext(CLIENT_TO_SERVER, 1, 1, 4); + send(client, proxy, "abcdef"); + send(client, proxy, "untouched"); + assertEquals("adcbef", receive(client)); + assertEquals("untouched", receive(client)); + assertEquals(1, proxy.stats().modified()); + } + } + + @Test + void supportsFaultsOnServerToClientPackets() throws Exception { + try (EchoServer server = new EchoServer(); + UdpFaultProxy proxy = new UdpFaultProxy(0, server.address()); + DatagramSocket client = client()) { + proxy.injectNext(SERVER_TO_CLIENT, 1, 0, bytes("prefix-")); + proxy.reverseRangeNext(SERVER_TO_CLIENT, 1, 0, 4); + send(client, proxy, "hello"); + send(client, proxy, "abcdE"); + assertEquals("prefix-hello", receive(client)); + assertEquals("dcbaE", receive(client)); + assertEquals(2, proxy.stats().modified()); + } + } + + @Test + void queuesRulesAndResetFlushesBufferedPackets() throws Exception { + try (EchoServer server = new EchoServer(); + UdpFaultProxy proxy = new UdpFaultProxy(0, server.address()); + DatagramSocket client = client()) { + proxy.dropNext(CLIENT_TO_SERVER, 1); + proxy.reverseNext(CLIENT_TO_SERVER, 3, Duration.ofSeconds(2)); + send(client, proxy, "dropped"); + send(client, proxy, "buffered"); + // The ingress loop is asynchronous. Wait until the buffered datagram has been accepted. + waitFor(() -> proxy.stats().clientToServerReceived() == 2); + assertEquals(1, proxy.stats().pendingRules()); + proxy.resetRules(); + assertEquals("buffered", receive(client)); + assertEquals(0, proxy.stats().pendingRules()); + assertEquals(1, proxy.stats().dropped()); + } + } + + @Test + void invalidMutationsDropTheDatagramAndIncrementErrorCounters() throws Exception { + try (EchoServer server = new EchoServer(); + UdpFaultProxy proxy = new UdpFaultProxy(0, server.address()); + DatagramSocket client = client()) { + proxy.reverseRangeNext(CLIENT_TO_SERVER, 1, 0, 20); + send(client, proxy, "short"); + send(client, proxy, "valid"); + assertEquals("valid", receive(client)); + assertEquals(1, proxy.stats().invalidMutations()); + assertEquals(1, proxy.stats().dropped()); + } + } + + @Test + void rejectsInvalidRuleConfigurationWithoutMutatingProxy() throws Exception { + try (EchoServer server = new EchoServer(); + UdpFaultProxy proxy = new UdpFaultProxy(0, server.address())) { + assertThrows(IllegalArgumentException.class, () -> proxy.dropNext(CLIENT_TO_SERVER, 0)); + assertThrows(IllegalArgumentException.class, () -> proxy.dropNext(CLIENT_TO_SERVER, 1_025)); + assertThrows(IllegalArgumentException.class, () -> proxy.injectNext(CLIENT_TO_SERVER, 1, -1, bytes("x"))); + assertThrows(IllegalArgumentException.class, () -> proxy.injectNext(CLIENT_TO_SERVER, 1, 0, new byte[0])); + assertThrows(IllegalArgumentException.class, () -> proxy.reverseRangeNext(CLIENT_TO_SERVER, 1, 4, 2)); + assertThrows(IllegalArgumentException.class, () -> proxy.reverseNext(CLIENT_TO_SERVER, 1, Duration.ZERO)); + assertEquals(0, proxy.stats().pendingRules()); + } + } + + @Test + void closeReleasesKnownListeningPortAndRejectsNewRules() throws Exception { + try (EchoServer server = new EchoServer()) { + UdpFaultProxy proxy = new UdpFaultProxy(0, server.address()); + int port = proxy.getListenPort(); + proxy.close(); + proxy.close(); + assertThrows(IllegalStateException.class, () -> proxy.dropNext(CLIENT_TO_SERVER, 1)); + try (DatagramSocket rebound = new DatagramSocket(new InetSocketAddress(InetAddress.getLoopbackAddress(), port))) { + assertEquals(port, rebound.getLocalPort()); + } + } + } + + private static DatagramSocket client() throws SocketException { + DatagramSocket socket = new DatagramSocket(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0)); + socket.setSoTimeout(2_000); + return socket; + } + + private static void send(DatagramSocket client, UdpFaultProxy proxy, String value) throws IOException { + byte[] data = bytes(value); + client.send(new DatagramPacket(data, data.length, proxy.getListenAddress())); + } + + private static String receive(DatagramSocket socket) throws IOException { + byte[] data = new byte[2_048]; + DatagramPacket packet = new DatagramPacket(data, data.length); + socket.receive(packet); + return new String(packet.getData(), packet.getOffset(), packet.getLength(), StandardCharsets.UTF_8); + } + + private static byte[] bytes(String value) { + return value.getBytes(StandardCharsets.UTF_8); + } + + private static void waitFor(Check check) throws InterruptedException { + long deadline = System.nanoTime() + Duration.ofSeconds(2).toNanos(); + while (!check.matches() && System.nanoTime() < deadline) { + Thread.sleep(5); + } + assertTrue(check.matches(), "Expected asynchronous packet processing to complete"); + } + + private interface Check { + boolean matches(); + } + + private static final class EchoServer implements AutoCloseable { + private final DatagramSocket socket; + private final Thread thread; + private final AtomicReference failure = new AtomicReference<>(); + + private EchoServer() throws SocketException { + socket = new DatagramSocket(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0)); + socket.setSoTimeout(200); + thread = new Thread(() -> { + while (!socket.isClosed()) { + byte[] bytes = new byte[65_507]; + DatagramPacket packet = new DatagramPacket(bytes, bytes.length); + try { + socket.receive(packet); + socket.send(new DatagramPacket( + Arrays.copyOf(packet.getData(), packet.getLength()), packet.getLength(), packet.getSocketAddress())); + } catch (SocketTimeoutException ignored) { + // Allow orderly shutdown. + } catch (SocketException exception) { + if (!socket.isClosed()) { + failure.compareAndSet(null, exception); + } + return; + } catch (IOException exception) { + failure.compareAndSet(null, exception); + return; + } + } + }, "udp-fault-proxy-test-echo"); + thread.setDaemon(true); + thread.start(); + } + + private InetSocketAddress address() { + return new InetSocketAddress(socket.getLocalAddress(), socket.getLocalPort()); + } + + @Override + public void close() throws InterruptedException { + socket.close(); + thread.join(1_000); + assertNull(failure.get(), "Echo fixture failed"); + } + } +}