diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java index 7da1835..f602927 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java @@ -83,14 +83,7 @@ public Vote handle(Socket socket) { return null; } - VoteProtocolVersion version = voteParser.detectVersion(in); - receiver.debug("Detected vote protocol version: " + version); - - if (receiver.isDisableV1() && version == VoteProtocolVersion.V1) { - throw new VoteAuthenticationException("Votifier V1 votes are disabled by configuration"); - } - - VoteRequest request = voteParser.parse(in, version, receiver, address, challenge, accepted); + VoteRequest request = voteParser.parse(in, receiver, address, challenge, accepted); Vote vote = new Vote(); vote.setServiceName(request.getServiceName()); diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteParser.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteParser.java index 5aced45..0f849d1 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteParser.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteParser.java @@ -10,6 +10,7 @@ import java.io.ByteArrayOutputStream; import java.io.PushbackInputStream; import java.net.Socket; +import java.net.SocketException; import java.net.SocketTimeoutException; import java.nio.charset.StandardCharsets; import java.security.Key; @@ -36,6 +37,7 @@ public class VoteParser { private static final int PROTOCOL_VERSION_PREFIX_BYTES = 2; private static final int V1_BLOCK_BYTES = 256; private static final int MAX_V2_PACKET_BYTES = 4 + 0xFFFF; + private static final int V2_PACKET_READ_TIMEOUT_MS = 5000; private static final int V1_COLLISION_GRACE_TIMEOUT_MS = 250; private static final String FIELD_PAYLOAD = "payload"; @@ -55,9 +57,15 @@ public class VoteParser { * @throws Exception if there is not enough data to determine the protocol */ public VoteProtocolVersion detectVersion(PushbackInputStream in) throws Exception { + return detectVersion(in, null, 0); + } + + private VoteProtocolVersion detectVersion(PushbackInputStream in, Socket socket, long deadlineNanos) + throws Exception { byte[] header = new byte[PROTOCOL_VERSION_PREFIX_BYTES]; int bytesRead = 0; while (bytesRead < header.length) { + setRemainingTimeout(socket, deadlineNanos); int read = in.read(header, bytesRead, header.length - bytesRead); if (read == -1) { break; @@ -83,6 +91,35 @@ public VoteProtocolVersion detectVersion(PushbackInputStream in) throws Exceptio return VoteProtocolVersion.V1; } + /** + * Detects and parses a network vote under one packet-read deadline. + * + * @param in the input stream + * @param receiver the vote receiver + * @param address remote address string for logging/errors + * @param challenge expected challenge for V2 + * @param socket accepted connection socket + * @return parsed vote request data + * @throws Exception on protocol detection, parsing, validation, or authentication errors + */ + public VoteRequest parse(PushbackInputStream in, VoteReceiver receiver, String address, String challenge, + Socket socket) throws Exception { + int previousTimeout = socket.getSoTimeout(); + long deadlineNanos = System.nanoTime() + V2_PACKET_READ_TIMEOUT_MS * 1_000_000L; + try { + VoteProtocolVersion version = detectVersion(in, socket, deadlineNanos); + receiver.debug("Detected vote protocol version: " + version); + + if (receiver.isDisableV1() && version == VoteProtocolVersion.V1) { + throw new VoteAuthenticationException("Votifier V1 votes are disabled by configuration"); + } + + return parseWithDeadline(in, version, receiver, address, challenge, socket, deadlineNanos); + } finally { + socket.setSoTimeout(previousTimeout); + } + } + /** * Parses the vote payload based on the detected protocol version. * @@ -100,8 +137,9 @@ public VoteRequest parse(PushbackInputStream in, VoteProtocolVersion version, Vo } /** - * Parses a vote payload with access to the connection socket so an ambiguous - * V1 collision can receive a short, bounded TCP-fragment grace period. + * Parses a vote payload with access to the connection socket so V2 reads have + * an absolute deadline and an ambiguous V1 collision can receive a short, + * bounded TCP-fragment grace period. * * @param in the input stream * @param version the detected protocol version @@ -118,26 +156,44 @@ public VoteRequest parse(PushbackInputStream in, VoteProtocolVersion version, Vo return parseV1(in, receiver, address); } + int previousTimeout = socket == null ? 0 : socket.getSoTimeout(); + long deadlineNanos = socket == null ? 0 + : System.nanoTime() + V2_PACKET_READ_TIMEOUT_MS * 1_000_000L; + try { + return parseWithDeadline(in, version, receiver, address, challenge, socket, deadlineNanos); + } finally { + if (socket != null) { + socket.setSoTimeout(previousTimeout); + } + } + } + + private VoteRequest parseWithDeadline(PushbackInputStream in, VoteProtocolVersion version, VoteReceiver receiver, + String address, String challenge, Socket socket, long deadlineNanos) throws Exception { + if (version == VoteProtocolVersion.V1) { + return parseV1(in, receiver, address); + } + ByteArrayOutputStream voteData = new ByteArrayOutputStream(); - if (!readToSize(in, voteData, PROTOCOL_VERSION_PREFIX_BYTES)) { + if (!readToSize(in, voteData, PROTOCOL_VERSION_PREFIX_BYTES, socket, deadlineNanos)) { throw new InvalidVoteException("Incomplete V2 protocol prefix from " + address); } byte[] prefix = voteData.toByteArray(); short magic = (short) (((prefix[0] & 0xFF) << 8) | (prefix[1] & 0xFF)); if (magic == PROTOCOL_2_MAGIC) { - return parseFramedV2(in, voteData, receiver, address, challenge, socket); + return parseFramedV2(in, voteData, receiver, address, challenge, socket, deadlineNanos); } if ((char) prefix[0] == '{') { - return parseUnframedV2(in, voteData, receiver, address, challenge, socket); + return parseUnframedV2(in, voteData, receiver, address, challenge, socket, deadlineNanos); } throw new InvalidVoteException("Invalid V2 protocol prefix from " + address); } private VoteRequest parseFramedV2(PushbackInputStream in, ByteArrayOutputStream voteData, VoteReceiver receiver, - String address, String challenge, Socket socket) throws Exception { - if (!readToSize(in, voteData, 4)) { + String address, String challenge, Socket socket, long deadlineNanos) throws Exception { + if (!readToSize(in, voteData, 4, socket, deadlineNanos)) { throw new InvalidVoteException("Incomplete V2 frame header from " + address); } @@ -149,7 +205,7 @@ private VoteRequest parseFramedV2(PushbackInputStream in, ByteArrayOutputStream // A randomized V1 block can claim a framed length greater than 256. Test // the complete V1-sized prefix before blocking for the rest of that frame. if (frameBytes > V1_BLOCK_BYTES && !receiver.isDisableV1()) { - if (!readToSize(in, voteData, V1_BLOCK_BYTES)) { + if (!readToSize(in, voteData, V1_BLOCK_BYTES, socket, deadlineNanos)) { throw new InvalidVoteException("Incomplete V2 frame from " + address + " (expected " + frameBytes + " bytes, got " + voteData.size() + ")"); } @@ -160,7 +216,7 @@ private VoteRequest parseFramedV2(PushbackInputStream in, ByteArrayOutputStream } } - if (!readToSize(in, voteData, frameBytes)) { + if (!readToSize(in, voteData, frameBytes, socket, deadlineNanos)) { throw new InvalidVoteException("Incomplete V2 frame from " + address + " (expected " + frameBytes + " bytes, got " + voteData.size() + ")"); } @@ -171,12 +227,12 @@ private VoteRequest parseFramedV2(PushbackInputStream in, ByteArrayOutputStream if (v1Failure != null) { v2Failure.addSuppressed(v1Failure); } - return fallbackToBufferedV1OrThrow(in, voteData, receiver, address, v2Failure, socket); + return fallbackToBufferedV1OrThrow(in, voteData, receiver, address, v2Failure, socket, deadlineNanos); } } private VoteRequest parseUnframedV2(PushbackInputStream in, ByteArrayOutputStream voteData, VoteReceiver receiver, - String address, String challenge, Socket socket) throws Exception { + String address, String challenge, Socket socket, long deadlineNanos) throws Exception { JsonObjectBoundaryScanner scanner = new JsonObjectBoundaryScanner(); byte[] prefix = voteData.toByteArray(); int jsonBoundaryBytes = scanner.scan(prefix, 0, prefix.length); @@ -206,6 +262,7 @@ private VoteRequest parseUnframedV2(PushbackInputStream in, ByteArrayOutputStrea int nextBoundary = voteData.size() < V1_BLOCK_BYTES ? V1_BLOCK_BYTES : MAX_V2_PACKET_BYTES; int maxRead = Math.min(buffer.length, nextBoundary - voteData.size()); + setRemainingTimeout(socket, deadlineNanos); int read = in.read(buffer, 0, maxRead); if (read == -1) { InvalidVoteException failure = new InvalidVoteException("Incomplete V2 JSON payload from " + address); @@ -234,7 +291,7 @@ private VoteRequest parseUnframedV2(PushbackInputStream in, ByteArrayOutputStrea if (v1Failure != null) { v2Failure.addSuppressed(v1Failure); } - return fallbackToBufferedV1OrThrow(in, voteData, receiver, address, v2Failure, socket); + return fallbackToBufferedV1OrThrow(in, voteData, receiver, address, v2Failure, socket, deadlineNanos); } } @@ -293,8 +350,14 @@ private VoteRequest parseV1(PushbackInputStream in, VoteReceiver receiver, Strin } private boolean readToSize(PushbackInputStream in, ByteArrayOutputStream data, int targetBytes) throws Exception { + return readToSize(in, data, targetBytes, null, 0); + } + + private boolean readToSize(PushbackInputStream in, ByteArrayOutputStream data, int targetBytes, Socket socket, + long deadlineNanos) throws Exception { byte[] buffer = new byte[Math.min(4096, Math.max(1, targetBytes - data.size()))]; while (data.size() < targetBytes) { + setRemainingTimeout(socket, deadlineNanos); int read = in.read(buffer, 0, Math.min(buffer.length, targetBytes - data.size())); if (read == -1) { return false; @@ -304,8 +367,23 @@ private boolean readToSize(PushbackInputStream in, ByteArrayOutputStream data, i return true; } + private void setRemainingTimeout(Socket socket, long deadlineNanos) throws SocketException, SocketTimeoutException { + if (socket == null) { + return; + } + + long remainingNanos = deadlineNanos - System.nanoTime(); + if (remainingNanos <= 0) { + throw new SocketTimeoutException("V2 packet read deadline exceeded"); + } + + int remainingMillis = (int) Math.max(1, (remainingNanos + 999_999L) / 1_000_000L); + int currentTimeout = socket.getSoTimeout(); + socket.setSoTimeout(currentTimeout <= 0 ? remainingMillis : Math.min(currentTimeout, remainingMillis)); + } + private VoteRequest fallbackToBufferedV1OrThrow(PushbackInputStream in, ByteArrayOutputStream voteData, - VoteReceiver receiver, String address, Exception v2Failure, Socket socket) throws Exception { + VoteReceiver receiver, String address, Exception v2Failure, Socket socket, long deadlineNanos) throws Exception { if (receiver.isDisableV1() || voteData.size() > V1_BLOCK_BYTES) { throw v2Failure; } @@ -316,7 +394,7 @@ private VoteRequest fallbackToBufferedV1OrThrow(PushbackInputStream in, ByteArra if (in.available() < remaining || !readToSize(in, voteData, V1_BLOCK_BYTES)) { throw v2Failure; } - } else if (!readV1CollisionRemainder(in, voteData, socket, v2Failure)) { + } else if (!readV1CollisionRemainder(in, voteData, socket, v2Failure, deadlineNanos)) { throw v2Failure; } } @@ -330,9 +408,12 @@ private VoteRequest fallbackToBufferedV1OrThrow(PushbackInputStream in, ByteArra } private boolean readV1CollisionRemainder(PushbackInputStream in, ByteArrayOutputStream voteData, Socket socket, - Exception v2Failure) throws Exception { + Exception v2Failure, long v2DeadlineNanos) throws Exception { int previousTimeout = socket.getSoTimeout(); long deadlineNanos = System.nanoTime() + V1_COLLISION_GRACE_TIMEOUT_MS * 1_000_000L; + if (v2DeadlineNanos > 0) { + deadlineNanos = Math.min(deadlineNanos, v2DeadlineNanos); + } byte[] buffer = new byte[V1_BLOCK_BYTES - voteData.size()]; try { while (voteData.size() < V1_BLOCK_BYTES) { diff --git a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverTest.java b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverTest.java index 2d68e4e..fa11c82 100644 --- a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverTest.java +++ b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverTest.java @@ -12,6 +12,7 @@ import java.io.PushbackInputStream; import java.net.ServerSocket; import java.net.Socket; +import java.net.SocketTimeoutException; import java.nio.charset.StandardCharsets; import java.security.Key; import java.security.KeyPair; @@ -273,6 +274,16 @@ public void testInvalidFramedV2CollisionGraceIsBounded() throws Exception { } } + @Test + public void testIncompleteUnframedV2PrefixAndBodyShareTotalDeadline() throws Exception { + assertV2TrickleHasTotalDeadline(new byte[] { '{' }, '{', 3000); + } + + @Test + public void testIncompleteFramedV2TrickleHasTotalDeadline() throws Exception { + assertV2TrickleHasTotalDeadline(new byte[] { 0x73, 0x3A, 0, (byte) 128 }, '{', 750); + } + @Test public void testV1MagicPrefixCollisionAttemptsV1Fallback() throws Exception { byte[] v1Block = new byte[256]; @@ -660,6 +671,50 @@ private byte[] createSignedV2Payload(String username) throws Exception { return outer.toString().getBytes(StandardCharsets.UTF_8); } + private void assertV2TrickleHasTotalDeadline(byte[] initialPayload, int trickleByte, long firstWriteDelayMillis) + throws Exception { + try (ServerSocket server = new ServerSocket(0); + Socket client = new Socket("127.0.0.1", server.getLocalPort()); + Socket accepted = server.accept(); + PushbackInputStream in = new PushbackInputStream(accepted.getInputStream(), 512)) { + accepted.setSoTimeout(5000); + client.getOutputStream().write(initialPayload); + client.getOutputStream().flush(); + + Thread trickleWriter = new Thread(() -> { + try { + long writeDelayMillis = firstWriteDelayMillis; + while (!Thread.currentThread().isInterrupted()) { + Thread.sleep(writeDelayMillis); + client.getOutputStream().write(trickleByte); + client.getOutputStream().flush(); + writeDelayMillis = 750; + } + } catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + } catch (Exception ex) { + throw new RuntimeException(ex); + } + }, "V2-Trickle-Writer"); + trickleWriter.setDaemon(true); + trickleWriter.start(); + + try { + long startedAt = System.nanoTime(); + assertThrows(SocketTimeoutException.class, + () -> parser.parse(in, receiver, "test-address", receiver.getChallenge(), accepted)); + long elapsedMillis = (System.nanoTime() - startedAt) / 1_000_000; + + assertTrue(elapsedMillis >= 4000, "V2 packet deadline fired prematurely: " + elapsedMillis + "ms"); + assertTrue(elapsedMillis < 7000, "V2 packet trickle reset the total deadline: " + elapsedMillis + "ms"); + assertEquals(5000, accepted.getSoTimeout()); + } finally { + trickleWriter.interrupt(); + trickleWriter.join(1000); + } + } + } + private byte[] frameV2Payload(byte[] jsonPayload) throws Exception { ByteArrayOutputStream framedPayload = new ByteArrayOutputStream(); framedPayload.write(0x73);