Bound total V2 packet read time

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());

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) {

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);