diff --git a/VotifierPlus/pom.xml b/VotifierPlus/pom.xml index 0eb8c46..1a6ed93 100644 --- a/VotifierPlus/pom.xml +++ b/VotifierPlus/pom.xml @@ -79,7 +79,7 @@ 3.5.4 - org.apache.maven.plugins + org.apache.maven.plugins maven-jar-plugin 3.5.0 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 f602927..47343d9 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java @@ -46,6 +46,7 @@ public Vote handle(Socket socket) { String remoteIp = "unknown"; String address = ""; String throttleKey = null; + String aggregateThrottleKey = null; boolean tunnelMode = false; boolean realIpKnown = false; @@ -57,9 +58,21 @@ public Vote handle(Socket socket) { address = accepted.getRemoteSocketAddress() == null ? "/" + remoteIp : accepted.getRemoteSocketAddress().toString(); + throttleKey = "tunnel:" + remoteIp; + tunnelMode = throttleService.isTunnelMode(remoteIp); + aggregateThrottleKey = throttleKey; receiver.debug("Accepted connection from: " + address); accepted.setSoTimeout(5000); + if (throttleService.isAggregateBlocked(aggregateThrottleKey)) { + long retry = throttleService.aggregateRetryAfterMs(aggregateThrottleKey); + String blockedKey = throttleService.aggregateBlockedKey(aggregateThrottleKey); + throttleService.logWarning(receiver, "throttle|" + blockedKey, + "Votifier throttling " + blockedKey + " (tunnel=" + tunnelMode + "), retry in " + + Math.max(0, retry / 1000) + "s"); + return null; + } + String challenge = receiver.getChallenge(); sendHandshakeIfNeeded(in, writer, challenge); @@ -73,12 +86,12 @@ public Vote handle(Socket socket) { realIp = proxyResult.getRealIp(); realIpKnown = realIp != null && !realIp.isEmpty(); - tunnelMode = throttleService.isTunnelMode(remoteIp); throttleKey = realIpKnown ? "ip:" + realIp : "tunnel:" + remoteIp; - if (throttleService.isBlocked(throttleKey)) { - long retry = throttleService.retryAfterMs(throttleKey); - throttleService.logWarning(receiver, "throttle|" + throttleKey, "Votifier throttling " + throttleKey + if (throttleService.isBlocked(throttleKey, aggregateThrottleKey)) { + long retry = throttleService.retryAfterMs(throttleKey, aggregateThrottleKey); + String blockedKey = throttleService.blockedKey(throttleKey, aggregateThrottleKey); + throttleService.logWarning(receiver, "throttle|" + blockedKey, "Votifier throttling " + blockedKey + " (tunnel=" + tunnelMode + "), retry in " + Math.max(0, retry / 1000) + "s"); return null; } @@ -97,7 +110,7 @@ public Vote handle(Socket socket) { } receiver.log("Received vote record -> " + vote); - throttleService.success(throttleKey); + throttleService.success(throttleKey, aggregateThrottleKey); if (!"TestVote".equalsIgnoreCase(vote.getTimeStamp())) { sendOkResponse(writer); @@ -109,7 +122,7 @@ public Vote handle(Socket socket) { throttleKey = "tunnel:" + remoteIp; } - throttleService.fail(throttleKey, tunnelMode, realIpKnown); + throttleService.fail(throttleKey, aggregateThrottleKey, tunnelMode, realIpKnown); throttleService.logWarning(receiver, "invalid|" + throttleKey, "Invalid vote format from " + remoteIp + ": " + ex.getMessage()); } catch (VoteAuthenticationException ex) { @@ -117,7 +130,7 @@ public Vote handle(Socket socket) { throttleKey = "tunnel:" + remoteIp; } - throttleService.fail(throttleKey, tunnelMode, realIpKnown); + throttleService.fail(throttleKey, aggregateThrottleKey, tunnelMode, realIpKnown); throttleService.logWarning(receiver, "auth|" + throttleKey, "Authentication failed from " + remoteIp + ": " + ex.getMessage()); } catch (MalformedJsonException ex) { @@ -125,7 +138,7 @@ public Vote handle(Socket socket) { throttleKey = "tunnel:" + remoteIp; } - throttleService.fail(throttleKey, tunnelMode, false); + throttleService.fail(throttleKey, aggregateThrottleKey, tunnelMode, false); throttleService.logWarning(receiver, "malformedjson|" + throttleKey, "Invalid vote format: Malformed JSON payload from " + remoteIp + " - " + ex.getMessage()); } catch (BadPaddingException ex) { @@ -133,7 +146,7 @@ public Vote handle(Socket socket) { throttleKey = "tunnel:" + remoteIp; } - throttleService.fail(throttleKey, tunnelMode, realIpKnown); + throttleService.fail(throttleKey, aggregateThrottleKey, tunnelMode, realIpKnown); throttleService.logWarning(receiver, "badpadding|" + throttleKey, "Decryption failed: Invalid V1 vote block / public key mismatch from " + remoteIp); } catch (SocketTimeoutException ex) { 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 61d8f41..526bf07 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteParser.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteParser.java @@ -538,7 +538,8 @@ private VoteRequest parseV2(byte[] data, VoteReceiver receiver, String address, String serviceName = requireString(votePayload, FIELD_SERVICE_NAME, "Inner JSON from " + address + ": "); String username = requireString(votePayload, FIELD_USERNAME, "Inner JSON from " + address + ": "); - String voteAddress = votePayload.get(FIELD_ADDRESS).getAsString(); + String voteAddress = requirePossiblyEmptyString(votePayload, FIELD_ADDRESS, + "Inner JSON from " + address + ": "); String timeStamp = requireString(votePayload, FIELD_TIMESTAMP, "Inner JSON from " + address + ": "); String receivedChallenge = requireString(votePayload, FIELD_CHALLENGE, "Inner JSON from " + address + ": ").trim(); @@ -591,6 +592,18 @@ private String requireString(JsonObject obj, String field, String errorPrefix) t return value; } + private String requirePossiblyEmptyString(JsonObject obj, String field, String errorPrefix) + throws InvalidVoteException { + if (!obj.has(field)) { + throw new InvalidVoteException(errorPrefix + "missing field '" + field + "'"); + } + try { + return obj.get(field).getAsString(); + } catch (Exception ex) { + throw new InvalidVoteException(errorPrefix + "invalid field '" + field + "'", ex); + } + } + private String readString(byte[] data, int offset) { StringBuilder builder = new StringBuilder(); for (int i = offset; i < data.length; i++) { diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java index 17e6a18..126e1c5 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java @@ -10,6 +10,8 @@ import java.util.concurrent.ConcurrentHashMap; public class VoteThrottleService { + private static final int MAX_TRACKED_KEYS = 4096; + private static final int AGGREGATE_OVERFLOW_BUCKETS = 64; private static final class LogState { private volatile long lastLogMs; @@ -21,14 +23,34 @@ private static final class ThrottleState { private volatile int failures; private volatile long throttledUntilMs; private volatile long bannedUntilMs; + /* Once a proxied identity contributes to this aggregate, direct successes + * must not treat the aggregate as private to the tunnel remote. */ + private volatile boolean sharedByProxiedIdentity; } private final ThrottleConfig config; private final ConcurrentHashMap logStates = new ConcurrentHashMap(); private final ConcurrentHashMap throttleStates = new ConcurrentHashMap(); + private final ConcurrentHashMap aggregateStates = + new ConcurrentHashMap(); + private final ThrottleState[] aggregateOverflowStates = new ThrottleState[AGGREGATE_OVERFLOW_BUCKETS]; + private final Object logStateLock = new Object(); + private final Object throttleStateLock = new Object(); + /* Guarded by logStateLock; avoid O(n) sweeps while all entries are active. */ + private long nextLogSweepMs; + /* Guarded by throttleStateLock; avoid O(n) sweeps while all entries are known active. */ + private long nextThrottleSweepMs; + private long nextAggregateSweepMs; + /* Configured tunnel remotes are finite, trusted aggregate identities. */ public VoteThrottleService(ThrottleConfig config) { this.config = config; + if (config != null) { + for (String remoteIp : config.tunnelRemoteIps) { + if (aggregateStates.size() >= MAX_TRACKED_KEYS) break; + aggregateStates.put("tunnel:" + remoteIp, new ThrottleState()); + } + } } public ThrottleConfig getConfig() { @@ -40,89 +62,226 @@ public boolean isTunnelMode(String remoteIp) { } public boolean isBlocked(String key) { + return isBlocked(key, null); + } + + /** + * Checks only the aggregate identity, without consulting a per-client state. + * This is used before proxy headers are read so an already-blocked remote can + * be rejected without making a primary tunnel state apply to every proxied + * client identity. + */ + public boolean isAggregateBlocked(String aggregateKey) { + if (config == null || !config.enabled || aggregateKey == null) { + return false; + } + + synchronized (throttleStateLock) { + return isBlocked(getAggregateState(aggregateKey)); + } + } + + public long aggregateRetryAfterMs(String aggregateKey) { + if (config == null || !config.enabled || aggregateKey == null) { + return 0L; + } + + synchronized (throttleStateLock) { + return retryAfterMs(getAggregateState(aggregateKey)); + } + } + + public String aggregateBlockedKey(String aggregateKey) { + if (aggregateKey == null) return null; + synchronized (throttleStateLock) { + if (aggregateStates.containsKey(aggregateKey)) return aggregateKey; + return "aggregate-overflow:" + + Math.floorMod(aggregateKey.hashCode(), AGGREGATE_OVERFLOW_BUCKETS); + } + } + + public boolean isBlocked(String key, String aggregateKey) { if (config == null || !config.enabled) { return false; } - ThrottleState state = throttleStates.get(key); + synchronized (throttleStateLock) { + ThrottleState state = throttleStates.get(key); + if (isBlocked(state)) return true; + if (aggregateKey != null) + return isBlocked(getAggregateState(aggregateKey)); + } + return false; + } + + public String blockedKey(String key, String aggregateKey) { + if (config == null || !config.enabled) { + return key; + } + synchronized (throttleStateLock) { + ThrottleState direct = throttleStates.get(key); + if (isBlocked(direct)) { + return key; + } + if (aggregateKey != null && isBlocked(getAggregateState(aggregateKey))) { + if (aggregateStates.containsKey(aggregateKey)) return aggregateKey; + return "aggregate-overflow:" + + Math.floorMod(aggregateKey.hashCode(), AGGREGATE_OVERFLOW_BUCKETS); + } + return key; + } + } + + private boolean isBlocked(ThrottleState state) { if (state == null) { return false; } - long now = System.currentTimeMillis(); return state.bannedUntilMs > now || state.throttledUntilMs > now; } public long retryAfterMs(String key) { + return retryAfterMs(key, null); + } + + public long retryAfterMs(String key, String aggregateKey) { ThrottleState state = throttleStates.get(key); + long retry = retryAfterMs(state); + synchronized (throttleStateLock) { + if (aggregateKey != null) + retry = Math.max(retry, retryAfterMs(getAggregateState(aggregateKey))); + } + return retry; + } + + private long retryAfterMs(ThrottleState state) { if (state == null) { return 0L; } - long now = System.currentTimeMillis(); return Math.max(state.bannedUntilMs, state.throttledUntilMs) - now; } public void fail(String key, boolean tunnelMode, boolean realIpKnown) { + fail(key, null, tunnelMode, realIpKnown); + } + + public void fail(String key, String aggregateKey, boolean tunnelMode, boolean realIpKnown) { if (config == null || !config.enabled) { return; } - long now = System.currentTimeMillis(); - ThrottleState state = getThrottleState(key); + synchronized (throttleStateLock) { + long now = System.currentTimeMillis(); + boolean aggregate = false; + ThrottleState state = throttleStates.get(key); + if (state == null && aggregateKey != null) { + ThrottleState existingAggregate = getAggregateState(aggregateKey); + if (hasActiveFailures(existingAggregate, now)) { + state = existingAggregate; + aggregate = true; + } + } + if (state == null) { + state = getThrottleState(key); + } + if (state == null && aggregateKey != null) { + state = getAggregateThrottleState(aggregateKey, now); + aggregate = state != null; + } + if (state == null) { + return; + } + if (aggregate && aggregateKey != null && !key.equals(aggregateKey)) { + state.sharedByProxiedIdentity = true; + } - if (now - state.windowStartMs > config.windowMs) { - state.windowStartMs = now; - state.failures = 0; - } + if (now - state.windowStartMs > config.windowMs) { + state.windowStartMs = now; + state.failures = 0; + } - state.failures++; + state.failures++; - if (config.perClientBanEnabled && realIpKnown && state.failures >= config.perClientBanFailures) { - state.bannedUntilMs = now + config.perClientBanForMs; - return; - } + if (!aggregate && config.perClientBanEnabled && realIpKnown + && state.failures >= config.perClientBanFailures) { + state.bannedUntilMs = now + config.perClientBanForMs; + return; + } - int threshold = tunnelMode ? config.tunnelFailures : config.failures; - long duration = tunnelMode ? config.tunnelThrottleForMs : config.throttleForMs; + int threshold = tunnelMode ? config.tunnelFailures : config.failures; + long duration = tunnelMode ? config.tunnelThrottleForMs : config.throttleForMs; - if (state.failures >= threshold) { - state.throttledUntilMs = now + duration; + if (state.failures >= threshold) { + state.throttledUntilMs = now + duration; + } } } + private boolean hasActiveFailures(ThrottleState state, long now) { + return state != null && (state.bannedUntilMs > now || state.throttledUntilMs > now + || state.failures > 0 && now - state.windowStartMs <= config.windowMs); + } + public void success(String key) { - ThrottleState state = throttleStates.get(key); - if (state != null) { - state.failures = 0; - state.windowStartMs = System.currentTimeMillis(); + success(key, null); + } + + public void success(String key, String aggregateKey) { + synchronized (throttleStateLock) { + long now = System.currentTimeMillis(); + ThrottleState state = throttleStates.get(key); + if (state != null) { + state.failures = 0; + state.windowStartMs = now; + if (state.bannedUntilMs <= now && state.throttledUntilMs <= now + && throttleStates.remove(key, state)) { + nextThrottleSweepMs = 0L; + } + } + /* A proxied success must not clear failures shared by other identities. */ + if (aggregateKey != null && key.equals(aggregateKey)) { + // Only a dedicated aggregate belongs to this identity. An overflow + // bucket is deliberately shared by many identities and must not be + // reset by one successful request. + state = aggregateStates.get(aggregateKey); + if (state != null && !state.sharedByProxiedIdentity) { + state.failures = 0; + state.windowStartMs = now; + if (!isConfiguredAggregateKey(aggregateKey) && state.bannedUntilMs <= now + && state.throttledUntilMs <= now && aggregateStates.remove(aggregateKey, state)) { + nextAggregateSweepMs = 0L; + } + } + } } } public String allowLog(String key, String msg) { - long now = System.currentTimeMillis(); - long windowMs = config != null ? Math.max(250L, config.logWindowMs) : 60_000L; - - LogState state = logStates.get(key); - if (state == null) { - LogState created = new LogState(); - LogState existing = logStates.putIfAbsent(key, created); - state = existing == null ? created : existing; - } + synchronized (logStateLock) { + long now = System.currentTimeMillis(); + long windowMs = config != null ? Math.max(250L, config.logWindowMs) : 60_000L; + LogState state = logStates.get(key); + if (state == null) { + trimLogStates(now); + state = new LogState(); + logStates.put(key, state); + } - if (now - state.lastLogMs >= windowMs) { - int suppressed = state.suppressed; - state.suppressed = 0; - state.lastLogMs = now; + if (now - state.lastLogMs >= windowMs) { + int suppressed = state.suppressed; + state.suppressed = 0; + state.lastLogMs = now; - if (suppressed > 0) { - return msg + " (suppressed " + suppressed + " similar in last " + windowMs + "ms)"; + if (suppressed > 0) { + return msg + " (suppressed " + suppressed + " similar in last " + windowMs + "ms)"; + } + return msg; } - return msg; - } - state.suppressed++; - return null; + state.suppressed++; + return null; + } } public void logWarning(VoteReceiver receiver, String key, String message) { @@ -141,11 +300,181 @@ public void logGenericError(String remoteIp, Exception ex) { private ThrottleState getThrottleState(String key) { ThrottleState state = throttleStates.get(key); if (state == null) { + if (!trimThrottleStates(System.currentTimeMillis())) { + return null; + } ThrottleState created = new ThrottleState(); created.windowStartMs = System.currentTimeMillis(); ThrottleState existing = throttleStates.putIfAbsent(key, created); state = existing == null ? created : existing; + if (existing == null && throttleStates.size() >= MAX_TRACKED_KEYS) { + nextThrottleSweepMs = nextThrottleSweepMs == 0L + ? earliestThrottleExpiry(throttleStates, false) + : earlierDeadline(nextThrottleSweepMs, stateExpiry(created)); + } + } + return state; + } + + private ThrottleState getAggregateThrottleState(String key, long now) { + ThrottleState state = aggregateStates.get(key); + if (state != null) { + return state; + } + if (!trimAggregateStates(now)) { + return getOverflowAggregateState(key, now); + } + + ThrottleState created = new ThrottleState(); + created.windowStartMs = now; + ThrottleState existing = aggregateStates.putIfAbsent(key, created); + if (existing == null && aggregateStates.size() >= MAX_TRACKED_KEYS) { + nextAggregateSweepMs = nextAggregateSweepMs == 0L + ? earliestThrottleExpiry(aggregateStates, true) + : earlierDeadline(nextAggregateSweepMs, stateExpiry(created)); + } + return existing == null ? created : existing; + } + + private long earliestThrottleExpiry(ConcurrentHashMap states, + boolean skipConfiguredAggregates) { + long earliest = Long.MAX_VALUE; + for (java.util.Map.Entry entry : states.entrySet()) { + if (!skipConfiguredAggregates || !isConfiguredAggregateKey(entry.getKey())) { + earliest = Math.min(earliest, stateExpiry(entry.getValue())); + } + } + return earliest; + } + + private static long earlierDeadline(long current, long candidate) { + return current == 0L ? candidate : Math.min(current, candidate); + } + + private ThrottleState getAggregateState(String key) { + ThrottleState state = aggregateStates.get(key); + if (state != null) { + return state; + } + long now = System.currentTimeMillis(); + if (aggregateStates.size() >= MAX_TRACKED_KEYS && trimAggregateStates(now)) { + state = aggregateStates.get(key); + if (state != null) { + return state; + } + } + return aggregateStates.size() >= MAX_TRACKED_KEYS + ? aggregateOverflowStates[Math.floorMod(key.hashCode(), AGGREGATE_OVERFLOW_BUCKETS)] : null; + } + + private ThrottleState getOverflowAggregateState(String key, long now) { + int index = Math.floorMod(key.hashCode(), AGGREGATE_OVERFLOW_BUCKETS); + ThrottleState state = aggregateOverflowStates[index]; + if (state == null) { + state = new ThrottleState(); + state.windowStartMs = now; + aggregateOverflowStates[index] = state; } return state; } -} \ No newline at end of file + + private void trimLogStates(long now) { + if (logStates.size() < MAX_TRACKED_KEYS) { + /* A miss immediately adds one entry. Preserve the cached deadline when + * that insertion will refill the map. */ + if (logStates.size() < MAX_TRACKED_KEYS - 1) nextLogSweepMs = 0L; + return; + } + long expiry = config != null ? Math.max(250L, config.logWindowMs) : 60_000L; + if (now < nextLogSweepMs) { + /* The map is still saturated, so make bounded progress without rescanning it. */ + removeOneIfFull(logStates); + return; + } + + long nextSweep = Long.MAX_VALUE; + for (java.util.Map.Entry entry : logStates.entrySet()) { + LogState state = entry.getValue(); + long stateExpiry = state.lastLogMs > Long.MAX_VALUE - expiry ? Long.MAX_VALUE + : state.lastLogMs + expiry; + if (stateExpiry <= now) { + logStates.remove(entry.getKey(), entry.getValue()); + } else { + nextSweep = Math.min(nextSweep, stateExpiry); + } + } + if (logStates.size() >= MAX_TRACKED_KEYS) { + removeOneIfFull(logStates); + } + /* Even when reclamation made one slot, allowLog immediately refills it. */ + nextLogSweepMs = nextSweep == Long.MAX_VALUE ? 0L : nextSweep; + } + + private boolean trimThrottleStates(long now) { + if (throttleStates.size() < MAX_TRACKED_KEYS) { + return true; + } + if (now < nextThrottleSweepMs) return false; + long nextSweep = Long.MAX_VALUE; + for (java.util.Map.Entry entry : throttleStates.entrySet()) { + ThrottleState state = entry.getValue(); + if (state.bannedUntilMs <= now && state.throttledUntilMs <= now + && now - state.windowStartMs > config.windowMs) { + throttleStates.remove(entry.getKey(), state); + } else nextSweep = Math.min(nextSweep, stateExpiry(state)); + } + if (throttleStates.size() < MAX_TRACKED_KEYS) { + /* Keep the earliest active expiry: the caller immediately refills the + * reclaimed slot, and the map can become saturated again before it. */ + nextThrottleSweepMs = nextSweep == Long.MAX_VALUE ? 0L : nextSweep; + return true; + } + nextThrottleSweepMs = nextSweep; + return false; + } + + private boolean trimAggregateStates(long now) { + if (aggregateStates.size() < MAX_TRACKED_KEYS) { + return true; + } + if (now < nextAggregateSweepMs) return false; + long nextSweep = Long.MAX_VALUE; + for (java.util.Map.Entry entry : aggregateStates.entrySet()) { + if (isConfiguredAggregateKey(entry.getKey())) { + continue; + } + ThrottleState state = entry.getValue(); + if (state.bannedUntilMs <= now && state.throttledUntilMs <= now + && now - state.windowStartMs > config.windowMs) { + aggregateStates.remove(entry.getKey(), state); + } else nextSweep = Math.min(nextSweep, stateExpiry(state)); + } + boolean available = aggregateStates.size() < MAX_TRACKED_KEYS; + /* As with primary states, preserve the deadline across the immediate + * insertion that consumes a reclaimed aggregate slot. */ + nextAggregateSweepMs = nextSweep == Long.MAX_VALUE + ? available ? 0L : Long.MAX_VALUE + : nextSweep; + return available; + } + + private long stateExpiry(ThrottleState state) { + long windowExpiry = state.windowStartMs > Long.MAX_VALUE - config.windowMs - 1L + ? Long.MAX_VALUE : state.windowStartMs + config.windowMs + 1L; + return Math.max(windowExpiry, Math.max(state.bannedUntilMs, state.throttledUntilMs)); + } + + private boolean isConfiguredAggregateKey(String key) { + return config != null && key.startsWith("tunnel:") + && config.tunnelRemoteIps.contains(key.substring("tunnel:".length())); + } + + private static void removeOneIfFull(ConcurrentHashMap states) { + if (states.size() >= MAX_TRACKED_KEYS) { + java.util.Iterator iterator = states.keySet().iterator(); + if (iterator.hasNext()) { + states.remove(iterator.next()); + } + } + } +} diff --git a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteConnectionHandlerTest.java b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteConnectionHandlerTest.java index 2232323..cef1eae 100644 --- a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteConnectionHandlerTest.java +++ b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteConnectionHandlerTest.java @@ -10,6 +10,7 @@ import java.io.OutputStream; 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; @@ -260,35 +261,53 @@ public Vote call() { } @Test - public void testHandleBlockedConnectionReturnsNull() throws Exception { + public void testProxyCanProvideClientIdentityBeforeThrottleDecision() throws Exception { receiver.setUseTokens(false); - ThrottleConfig config = new ThrottleConfig(true, Collections.emptySet(), "10s", 1, "30s", 1, "30s", - false, 999, "1s", "60s"); + ThrottleConfig config = new ThrottleConfig(true, Collections.emptySet(), "10s", 1, "30s", 1, + "30s", false, 999, "1s", "60s"); VoteThrottleService throttleService = new VoteThrottleService(config); VoteConnectionHandler handler = new VoteConnectionHandler(receiver, throttleService); - - throttleService.fail("tunnel:127.0.0.1", false, false); - assertTrue(throttleService.isBlocked("tunnel:127.0.0.1")); + throttleService.fail("tunnel:127.0.0.1", true, false); try (ServerSocket serverSocket = new ServerSocket(0); Socket client = new Socket("127.0.0.1", serverSocket.getLocalPort()); Socket accepted = serverSocket.accept()) { + Future future = executor.submit(() -> handler.handle(accepted)); + BufferedReader reader = new BufferedReader( + new InputStreamReader(client.getInputStream(), StandardCharsets.UTF_8)); + assertEquals("VOTIFIER 1", reader.readLine(), "Shared tunnels must reach proxy-header detection"); + client.close(); + assertNull(future.get()); + } + } - Future future = executor.submit(new Callable() { - @Override - public Vote call() { - return handler.handle(accepted); - } - }); + @Test + public void testAggregateBlockRejectsBeforeHandshakeAndPayloadWait() throws Exception { + receiver.setUseTokens(false); + ThrottleConfig config = new ThrottleConfig(true, Collections.singleton("127.0.0.1"), "5s", 1, "10s", 1, + "10s", false, 999, "1s", "60s"); + VoteThrottleService throttleService = new VoteThrottleService(config); + for (int i = 0; i < 4096; i++) { + throttleService.fail("ip:filler:" + i, false, true); + } + throttleService.fail("ip:blocked", "tunnel:127.0.0.1", false, true); + assertTrue(throttleService.isAggregateBlocked("tunnel:127.0.0.1")); + VoteConnectionHandler handler = new VoteConnectionHandler(receiver, throttleService); + try (ServerSocket serverSocket = new ServerSocket(0); + Socket client = new Socket("127.0.0.1", serverSocket.getLocalPort()); + Socket accepted = serverSocket.accept()) { + client.setSoTimeout(500); + Future future = executor.submit(() -> handler.handle(accepted)); BufferedReader clientReader = new BufferedReader( new InputStreamReader(client.getInputStream(), StandardCharsets.UTF_8)); - String handshake = clientReader.readLine(); - assertEquals("VOTIFIER 1", handshake); - - Vote vote = future.get(); - assertNull(vote); + try { + assertNull(clientReader.readLine(), "an aggregate-blocked remote must not receive a handshake"); + } catch (SocketTimeoutException ex) { + throw new AssertionError("aggregate rejection must happen before waiting for payload", ex); + } + assertNull(future.get(1, java.util.concurrent.TimeUnit.SECONDS)); } } @@ -454,6 +473,44 @@ public Vote call() { } } + @Test + public void testNonTunnelProxyOverflowUsesRemoteAggregate() throws Exception { + receiver.setUseTokens(false); + ThrottleConfig config = new ThrottleConfig(true, Collections.emptySet(), "5s", 1, "10s", 1, + "10s", false, 999, "1s", "60s"); + VoteThrottleService throttleService = new VoteThrottleService(config); + for (int i = 0; i < 4096; i++) { + throttleService.fail("ip:" + i, false, true); + } + VoteConnectionHandler handler = new VoteConnectionHandler(receiver, throttleService); + + try (ServerSocket serverSocket = new ServerSocket(0); + Socket client = new Socket("127.0.0.1", serverSocket.getLocalPort()); + Socket accepted = serverSocket.accept()) { + + Future future = executor.submit(new Callable() { + @Override + public Vote call() { + return handler.handle(accepted); + } + }); + + BufferedReader clientReader = new BufferedReader( + new InputStreamReader(client.getInputStream(), StandardCharsets.UTF_8)); + OutputStream clientOut = client.getOutputStream(); + assertEquals("VOTIFIER 1", clientReader.readLine()); + + clientOut.write("PROXY TCP4 203.0.113.10 127.0.0.1 1234 8192\r\n" + .getBytes(StandardCharsets.US_ASCII)); + clientOut.write(new byte[256]); + clientOut.flush(); + client.shutdownOutput(); + + assertNull(future.get()); + assertTrue(throttleService.isBlocked("ip:203.0.113.10", "tunnel:127.0.0.1")); + } + } + @Test public void testHandleTestVoteDoesNotSendOkResponse() throws Exception { receiver.setUseTokens(false); @@ -493,4 +550,4 @@ public Vote call() { assertTrue(!clientReader.ready(), "Did not expect an OK response for TestVote"); } } -} \ No newline at end of file +} 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 027979f..53fbf24 100644 --- a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverTest.java +++ b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverTest.java @@ -568,6 +568,30 @@ public void testV2VoteMissingUsernameField() throws Exception { assertTrue(exception.getMessage().contains("Missing required fields in inner JSON")); } + @Test + public void testV2VoteRejectsNullAddressAsInvalidVote() throws Exception { + JsonObject inner = validInnerVote(); + inner.add("address", null); + + InvalidVoteException exception = assertThrows(InvalidVoteException.class, + () -> parser.parse(v2Payload(inner), VoteProtocolVersion.V2, receiver, "test", + receiver.getChallenge())); + + assertTrue(exception.getMessage().contains("invalid field 'address'")); + } + + @Test + public void testV2VoteRejectsObjectAddressAsInvalidVote() throws Exception { + JsonObject inner = validInnerVote(); + inner.add("address", new JsonObject()); + + InvalidVoteException exception = assertThrows(InvalidVoteException.class, + () -> parser.parse(v2Payload(inner), VoteProtocolVersion.V2, receiver, "test", + receiver.getChallenge())); + + assertTrue(exception.getMessage().contains("invalid field 'address'")); + } + @Test public void testV2VoteInvalidChallenge() throws Exception { JsonObject inner = new JsonObject(); @@ -671,6 +695,24 @@ public void testBuildVoteFromParsedRequest() throws Exception { assertEquals("192.168.1.1", vote.getSourceAddress()); } + private JsonObject validInnerVote() { + JsonObject inner = new JsonObject(); + inner.addProperty("serviceName", "votifier.bencodez.com"); + inner.addProperty("username", "testUser"); + inner.addProperty("address", "127.0.0.1"); + inner.addProperty("timestamp", "TestTimestamp"); + inner.addProperty("challenge", receiver.getChallenge()); + return inner; + } + + private PushbackInputStream v2Payload(JsonObject inner) { + JsonObject outer = new JsonObject(); + outer.addProperty("payload", inner.toString()); + outer.addProperty("signature", "AA=="); + return new PushbackInputStream( + new ByteArrayInputStream(outer.toString().getBytes(StandardCharsets.UTF_8)), 512); + } + private byte[] createSignedV2Payload(String username) throws Exception { return createSignedV2Payload(username, "127.0.0.1", "testChallenge"); } diff --git a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java index ac14fee..249bc70 100644 --- a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java +++ b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java @@ -6,7 +6,14 @@ import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; +import java.lang.reflect.Field; import java.util.Collections; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import org.junit.jupiter.api.Test; @@ -137,6 +144,37 @@ public void testSuccessResetsFailureCounter() { assertTrue(service.isBlocked(key)); } + @Test + public void testSuccessImmediatelyReclaimsPrimaryCapacity() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 3, "10s", 3, "10s", false, 999, "1s")); + for (int index = 0; index < 4096; index++) service.fail("ip:active:" + index, false, true); + + service.fail("ip:overflow", false, true); + assertTrue(longField(service, "nextThrottleSweepMs") > System.currentTimeMillis()); + service.success("ip:active:0"); + + assertEquals(4095, mapSize(service, "throttleStates")); + assertEquals(0L, longField(service, "nextThrottleSweepMs")); + service.fail("ip:replacement", false, true); + assertTrue(stateMap(service, "throttleStates").containsKey("ip:replacement"), + "a successful identity must release its primary slot immediately"); + } + + @Test + public void testSuccessResetsOverflowAggregateForDirectPeer() { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 2, "10s", 2, "10s", false, 999, "1s")); + for (int i = 0; i < 4096; i++) { + service.fail("ip:" + i, false, false); + } + String key = "tunnel:direct"; + service.fail(key, key, false, false); + service.success(key, key); + service.fail(key, key, false, false); + assertFalse(service.isBlocked(key, key)); + } + @Test public void testWindowExpiryResetsFailureCounter() throws Exception { VoteThrottleService service = new VoteThrottleService( @@ -165,4 +203,607 @@ public void testTunnelModeDetection() { assertTrue(service.isTunnelMode("10.0.0.1")); assertFalse(service.isTunnelMode("10.0.0.2")); } -} \ No newline at end of file + + @Test + public void testAttackerControlledKeysAreBounded() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 3, "10s", 2, "10s", false, 999, "1s")); + for (int i = 0; i < 5000; i++) { + service.fail("ip:" + i, false, true); + service.allowLog("log:" + i, "message"); + } + assertTrue(mapSize(service, "throttleStates") <= 4096); + assertTrue(mapSize(service, "logStates") <= 4096); + } + + @Test + public void testConcurrentNewLogKeysStayWithinBound() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 3, "10s", 2, "10s", false, 999, "1s")); + int workers = 16; + int keysPerWorker = 512; + ExecutorService executor = Executors.newFixedThreadPool(workers); + CountDownLatch ready = new CountDownLatch(workers); + CountDownLatch start = new CountDownLatch(1); + CountDownLatch finished = new CountDownLatch(workers); + try { + for (int worker = 0; worker < workers; worker++) { + final int workerId = worker; + executor.execute(() -> { + ready.countDown(); + try { + start.await(); + for (int key = 0; key < keysPerWorker; key++) { + service.allowLog("concurrent-log:" + workerId + ':' + key, "message"); + } + } catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + } finally { + finished.countDown(); + } + }); + } + assertTrue(ready.await(5, TimeUnit.SECONDS)); + start.countDown(); + assertTrue(finished.await(10, TimeUnit.SECONDS)); + } finally { + executor.shutdownNow(); + } + assertEquals(4096, mapSize(service, "logStates")); + } + + @Test + public void testConcurrentFailuresUpdateStateAtomically() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("60s", 100000, "10s", 100000, "10s", false, 999, "1s")); + int workers = 16; + int failuresPerWorker = 256; + ExecutorService executor = Executors.newFixedThreadPool(workers); + CountDownLatch ready = new CountDownLatch(workers); + CountDownLatch start = new CountDownLatch(1); + CountDownLatch finished = new CountDownLatch(workers); + try { + for (int worker = 0; worker < workers; worker++) { + executor.execute(() -> { + ready.countDown(); + try { + start.await(); + for (int failure = 0; failure < failuresPerWorker; failure++) { + service.fail("concurrent-failure", false, false); + } + } catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + } finally { + finished.countDown(); + } + }); + } + assertTrue(ready.await(5, TimeUnit.SECONDS)); + start.countDown(); + assertTrue(finished.await(10, TimeUnit.SECONDS)); + } finally { + executor.shutdownNow(); + } + Field statesField = VoteThrottleService.class.getDeclaredField("throttleStates"); + statesField.setAccessible(true); + Map states = (Map) statesField.get(service); + Object state = states.get("concurrent-failure"); + Field failuresField = state.getClass().getDeclaredField("failures"); + failuresField.setAccessible(true); + assertEquals(workers * failuresPerWorker, failuresField.getInt(state)); + } + + @Test + public void testExistingKeyDoesNotEvictAnotherStateAtCapacity() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 2, "10s", 2, "10s", false, 999, "1s")); + for (int i = 0; i < 4096; i++) { + service.fail("ip:" + i, false, false); + } + Map states = stateMap(service, "throttleStates"); + Object existing = states.get("ip:100"); + service.fail("ip:100", false, false); + assertTrue(existing == states.get("ip:100")); + assertEquals(4096, states.size()); + } + + @Test + public void testExistingLogKeyDoesNotEvictAnotherStateAtCapacity() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 2, "10s", 2, "10s", false, 999, "1s")); + for (int i = 0; i < 4096; i++) { + service.allowLog("log:" + i, "message"); + } + Map states = stateMap(service, "logStates"); + Object existing = states.get("log:100"); + assertNull(service.allowLog("log:100", "message-again")); + assertTrue(existing == states.get("log:100")); + assertEquals(4096, states.size()); + } + + @Test + public void testSaturatedLogStateCachesExpiryAndEvictsWithoutRescanning() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 2, "10s", 2, "10s", false, 999, "1s")); + for (int i = 0; i < 4096; i++) { + service.allowLog("log:saturated:" + i, "message"); + } + + service.allowLog("log:saturated:first-miss", "message"); + long nextSweep = longField(service, "nextLogSweepMs"); + assertTrue(nextSweep > System.currentTimeMillis()); + service.allowLog("log:saturated:second-miss", "message"); + + assertEquals(nextSweep, longField(service, "nextLogSweepMs"), + "saturated log misses before expiry must reuse the cached sweep deadline"); + assertEquals(4096, mapSize(service, "logStates")); + } + + @Test + public void testReclaimedLogSlotRetainsNextExpiryAfterImmediateRefill() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 2, "10s", 2, "10s", false, 999, "1s")); + for (int i = 0; i < 4096; i++) { + service.allowLog("log:reclaim:" + i, "message"); + } + + Map logs = stateMap(service, "logStates"); + Object expired = logs.get("log:reclaim:0"); + Field lastLogField = expired.getClass().getDeclaredField("lastLogMs"); + lastLogField.setAccessible(true); + lastLogField.setLong(expired, System.currentTimeMillis() - 6000L); + Field nextSweepField = VoteThrottleService.class.getDeclaredField("nextLogSweepMs"); + nextSweepField.setAccessible(true); + nextSweepField.setLong(service, 0L); + + service.allowLog("log:reclaim:first-miss", "message"); + long nextSweep = longField(service, "nextLogSweepMs"); + assertTrue(nextSweep > System.currentTimeMillis()); + service.allowLog("log:reclaim:second-miss", "message"); + + assertEquals(nextSweep, longField(service, "nextLogSweepMs"), + "the refill after reclamation must retain the next active expiry"); + assertEquals(4096, logs.size()); + } + + @Test + public void testOverflowDoesNotEvictActiveBan() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 1, "10s", 1, "10s", true, 1, "60s")); + for (int i = 0; i < 4096; i++) { + service.fail("ip:" + i, false, true); + } + assertTrue(service.isBlocked("ip:100")); + service.fail("overflow", false, true); + assertTrue(service.isBlocked("ip:100")); + assertEquals(4096, mapSize(service, "throttleStates")); + } + + @Test + public void testFullThrottleMapAccountsNewIdentityAgainstAggregateTunnel() throws Exception { + VoteThrottleService service = new VoteThrottleService(tunnelCfg("proxy")); + for (int i = 0; i < 4096; i++) { + service.fail("ip:" + i, false, true); + } + + String aggregateKey = "tunnel:proxy"; + String newIdentity = "ip:new"; + service.fail(newIdentity, aggregateKey, false, true); + + assertTrue(service.isBlocked(newIdentity, aggregateKey)); + assertTrue(service.retryAfterMs(newIdentity, aggregateKey) > 0); + assertEquals(4096, mapSize(service, "throttleStates")); + } + + @Test + public void testAggregateFallbackRemainsBoundAfterPrimaryCapacityRecovers() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 3, "10s", 3, "10s", false, 999, "1s")); + String aggregateKey = "tunnel:proxy"; + for (int i = 0; i < 4096; i++) { + service.fail("ip:" + i, false, false); + } + + service.fail("ip:reused", aggregateKey, false, false); + service.fail("ip:reused", aggregateKey, false, false); + stateMap(service, "throttleStates").remove("ip:0"); + + service.fail("ip:reused", aggregateKey, false, false); + + assertTrue(service.isBlocked("ip:reused", aggregateKey), + "an identity that used aggregate fallback must not split its failure counter after capacity recovers"); + assertFalse(stateMap(service, "throttleStates").containsKey("ip:reused")); + } + + @Test + public void testAggregateBlockUsesAggregateLogKey() { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 1, "10s", 1, "10s", false, 999, "1s")); + for (int i = 0; i < 4096; i++) { + service.fail("ip:" + i, false, false); + } + String aggregateKey = "tunnel:proxy"; + service.fail("ip:rotated", aggregateKey, false, false); + assertEquals(aggregateKey, service.blockedKey("ip:another", aggregateKey)); + } + + @Test + public void testAggregateBlockAppliesToExistingPrimaryIdentity() { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 2, "10s", 2, "10s", false, 999, "1s")); + String primaryKey = "ip:existing"; + String aggregateKey = "tunnel:proxy"; + service.fail(primaryKey, false, false); + for (int i = 0; i < 4095; i++) { + service.fail("ip:filler:" + i, false, false); + } + service.fail("ip:proxied", aggregateKey, false, false); + service.fail("ip:proxied", aggregateKey, false, false); + + assertTrue(service.isBlocked(primaryKey, aggregateKey)); + assertEquals(aggregateKey, service.blockedKey(primaryKey, aggregateKey)); + assertTrue(service.retryAfterMs(primaryKey, aggregateKey) > 0); + } + + @Test + public void testProxiedSuccessDoesNotClearSharedAggregateFailures() { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 2, "10s", 2, "10s", false, 999, "1s")); + String aggregateKey = "tunnel:proxy"; + for (int i = 0; i < 4096; i++) { + service.fail("ip:filler:" + i, false, false); + } + service.fail("ip:proxied", aggregateKey, false, false); + service.success("ip:proxied", aggregateKey); + service.fail("ip:rotated", aggregateKey, false, false); + + assertTrue(service.isBlocked("ip:another", aggregateKey), + "a proxied success must not erase another identity's aggregate failure"); + } + + @Test + public void testDirectSuccessDoesNotClearAggregateSharedByProxiedFailures() { + VoteThrottleService service = new VoteThrottleService(tunnelCfg("proxy")); + for (int i = 0; i < 4096; i++) { + service.fail("ip:filler:" + i, false, true); + } + + String aggregateKey = "tunnel:proxy"; + service.fail("ip:proxied", aggregateKey, false, false); + assertTrue(service.isAggregateBlocked(aggregateKey)); + + service.success(aggregateKey, aggregateKey); + + assertTrue(service.isAggregateBlocked(aggregateKey), + "a direct success must not clear an aggregate that has been shared by proxied identities"); + } + + @Test + public void testFullMapPreservesInWindowCountersAndUsesAggregateTunnel() throws Exception { + ThrottleConfig config = new ThrottleConfig(true, Collections.singleton("proxy"), "5s", 2, "10s", 2, + "10s", false, 999, "1s", "60s"); + VoteThrottleService service = new VoteThrottleService(config); + for (int i = 0; i < 4096; i++) { + service.fail("ip:" + i, false, true); + } + + service.fail("ip:new", "tunnel:proxy", false, true); + assertFalse(service.isBlocked("ip:new", "tunnel:proxy")); + service.fail("ip:new", "tunnel:proxy", false, true); + + assertTrue(service.isBlocked("ip:new", "tunnel:proxy")); + assertEquals(4096, mapSize(service, "throttleStates")); + } + + @Test + public void testFullThrottleMapAccountsNewIdentityAgainstNonTunnelAggregate() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 1, "10s", 1, "10s", false, 999, "1s")); + for (int i = 0; i < 4096; i++) { + service.fail("ip:" + i, false, true); + } + + String aggregateKey = "tunnel:proxy"; + String newIdentity = "ip:new"; + service.fail(newIdentity, aggregateKey, false, true); + + assertTrue(service.isBlocked(newIdentity, aggregateKey)); + assertTrue(service.retryAfterMs(newIdentity, aggregateKey) > 0); + assertEquals(4096, mapSize(service, "throttleStates")); + assertEquals(1, mapSize(service, "aggregateStates")); + } + + @Test + public void testOverflowThrottleDoesNotBecomeGlobalAcrossTunnels() { + VoteThrottleService service = new VoteThrottleService(tunnelCfg("proxy-a", "proxy-b")); + for (int i = 0; i < 4096; i++) { + service.fail("ip:" + i, false, true); + } + + service.fail("ip:new-a", "tunnel:proxy-a", false, true); + + assertTrue(service.isBlocked("ip:new-a", "tunnel:proxy-a")); + assertFalse(service.isBlocked("ip:new-b", "tunnel:proxy-b")); + } + + @Test + public void testOverflowFallbackIgnoresPrimaryTunnelState() { + VoteThrottleService service = new VoteThrottleService(tunnelCfg("proxy")); + service.fail("tunnel:proxy", true, false); + for (int i = 0; i < 4095; i++) { + service.fail("ip:" + i, false, true); + } + + assertFalse(service.isBlocked("ip:new", "tunnel:proxy")); + assertEquals(0L, service.retryAfterMs("ip:new", "tunnel:proxy")); + service.fail("ip:new", "tunnel:proxy", false, true); + assertTrue(service.isBlocked("ip:new", "tunnel:proxy")); + } + + @Test + public void testAggregateCapacityUsesBoundedOverflowBuckets() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 2, "10s", 2, "10s", false, 999, "1s")); + for (int i = 0; i < 4096; i++) { + service.fail("ip:" + i, false, false); + } + for (int i = 0; i < 4096; i++) { + service.fail("ip:overflow:" + i, "remote:" + i, false, false); + } + String aggregateKey = "remote:overflow"; + service.fail("ip:new", aggregateKey, false, false); + service.fail("ip:new", aggregateKey, false, false); + assertTrue(service.isBlocked("ip:new", aggregateKey)); + assertEquals(4096, mapSize(service, "aggregateStates")); + } + + @Test + public void testOverflowBucketIsIgnoredAfterAggregateCapacityRecovers() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 2, "10s", 2, "10s", false, 999, "1s")); + for (int i = 0; i < 4096; i++) service.fail("ip:" + i, false, false); + for (int i = 0; i < 4096; i++) service.fail("ip:overflow:" + i, "remote:" + i, false, false); + assertEquals("Aa".hashCode(), "BB".hashCode()); + service.fail("ip:overflow-a", "Aa", false, false); + service.fail("ip:overflow-a", "Aa", false, false); + assertTrue(service.isBlocked("ip:overflow-b", "BB")); + assertEquals("aggregate-overflow:" + Math.floorMod("BB".hashCode(), 64), + service.blockedKey("ip:overflow-b", "BB"), + "all remotes rejected by one overflow bucket must share its log key"); + assertEquals("aggregate-overflow:" + Math.floorMod("BB".hashCode(), 64), + service.aggregateBlockedKey("BB"), + "pre-handshake aggregate rejection must use the same bounded log key"); + + Map aggregates = stateMap(service, "aggregateStates"); + aggregates.remove(aggregates.keySet().iterator().next()); + assertFalse(service.isBlocked("ip:overflow-b", "BB"), + "stale overflow state must not shadow a newly available aggregate slot"); + } + + @Test + public void testDirectSuccessDoesNotResetSharedAggregateOverflowBucket() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 2, "10s", 2, "10s", false, 999, "1s")); + for (int i = 0; i < 4096; i++) service.fail("ip:" + i, false, false); + for (int i = 0; i < 4096; i++) service.fail("ip:overflow:" + i, "remote:" + i, false, false); + assertEquals("Aa".hashCode(), "BB".hashCode()); + service.fail("ip:overflow-a", "Aa", false, false); + service.success("Aa", "Aa"); + service.fail("ip:overflow-b", "BB", false, false); + + assertTrue(service.isBlocked("ip:overflow-c", "BB"), + "a direct success must not erase failures belonging to a shared overflow bucket"); + } + + @Test + public void testBlockedCheckRechecksPrimaryAfterWaitingForFailureUpdate() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 2, "10s", 2, "10s", false, 999, "1s")); + String key = "ip:concurrent"; + service.fail(key, false, true); + AtomicBoolean blocked = new AtomicBoolean(); + Thread check = new Thread(() -> blocked.set(service.isBlocked(key, "tunnel:other"))); + Field lockField = VoteThrottleService.class.getDeclaredField("throttleStateLock"); + lockField.setAccessible(true); + Object lock = lockField.get(service); + + synchronized (lock) { + check.start(); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(2); + while (check.getState() != Thread.State.BLOCKED && System.nanoTime() < deadline) Thread.onSpinWait(); + assertEquals(Thread.State.BLOCKED, check.getState()); + service.fail(key, false, true); + } + check.join(TimeUnit.SECONDS.toMillis(2)); + + assertTrue(blocked.get(), "the primary throttle applied while waiting must not be bypassed"); + } + + @Test + public void testSaturatedPrimaryStateCachesItsNextPossibleSweep() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 2, "10s", 2, "10s", false, 999, "1s")); + for (int index = 0; index < 4096; index++) service.fail("ip:active:" + index, false, true); + + service.fail("ip:overflow:first", false, true); + long nextSweep = longField(service, "nextThrottleSweepMs"); + assertTrue(nextSweep > System.currentTimeMillis()); + service.fail("ip:overflow:second", false, true); + assertEquals(nextSweep, longField(service, "nextThrottleSweepMs"), + "a miss before the earliest expiry must reuse the saturated-state decision"); + } + + @Test + public void testInitialPrimarySaturationUsesAnOlderStateExpiry() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("60s", 999, "60s", 999, "60s", false, 999, "1s")); + for (int index = 0; index < 4095; index++) service.fail("ip:initial-primary:" + index, false, true); + + Map states = stateMap(service, "throttleStates"); + Object expired = states.get("ip:initial-primary:0"); + Field windowStartField = expired.getClass().getDeclaredField("windowStartMs"); + windowStartField.setAccessible(true); + long now = System.currentTimeMillis(); + for (Object state : states.values()) windowStartField.setLong(state, now + 60_000L); + windowStartField.setLong(expired, now - 61_000L); + + service.fail("ip:initial-primary:capacity", false, true); + assertTrue(longField(service, "nextThrottleSweepMs") <= System.currentTimeMillis(), + "the first saturated map must retain the earlier existing expiry"); + service.fail("ip:initial-primary:replacement", false, true); + + assertFalse(states.containsKey("ip:initial-primary:0")); + assertTrue(states.containsKey("ip:initial-primary:replacement")); + } + + @Test + public void testReclaimedPrimarySlotRetainsNextExpiryAfterImmediateRefill() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 999, "10s", 999, "10s", false, 999, "1s")); + for (int index = 0; index < 4096; index++) service.fail("ip:reclaim:" + index, false, true); + + Map states = stateMap(service, "throttleStates"); + Object expired = states.get("ip:reclaim:0"); + Field windowStartField = expired.getClass().getDeclaredField("windowStartMs"); + windowStartField.setAccessible(true); + long laterExistingExpiry = System.currentTimeMillis() + 60_000L; + for (Object state : states.values()) windowStartField.setLong(state, laterExistingExpiry); + windowStartField.setLong(expired, System.currentTimeMillis() - 6000L); + Field nextSweepField = VoteThrottleService.class.getDeclaredField("nextThrottleSweepMs"); + nextSweepField.setAccessible(true); + nextSweepField.setLong(service, 0L); + + service.fail("ip:reclaim:first-miss", false, true); + long nextSweep = longField(service, "nextThrottleSweepMs"); + assertTrue(nextSweep > System.currentTimeMillis()); + assertTrue(nextSweep < laterExistingExpiry, + "the newly inserted state's earlier expiry must bound the retained deadline"); + service.fail("ip:reclaim:second-miss", false, true); + + assertEquals(nextSweep, longField(service, "nextThrottleSweepMs"), + "the refill after reclamation must retain the next active expiry"); + assertEquals(4096, states.size()); + } + + @Test + public void testAggregateLookupReclaimsExpiredEntryBeforeUsingOverflowBucket() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 2, "10s", 2, "10s", false, 999, "1s")); + for (int i = 0; i < 4096; i++) { + service.fail("ip:primary:" + i, false, false); + } + for (int i = 0; i < 4096; i++) { + service.fail("ip:aggregate:" + i, "remote:" + i, false, false); + } + + Map aggregates = stateMap(service, "aggregateStates"); + Object expired = aggregates.get("remote:0"); + assertNotNull(expired); + Field windowStartField = expired.getClass().getDeclaredField("windowStartMs"); + windowStartField.setAccessible(true); + windowStartField.setLong(expired, System.currentTimeMillis() - 6000L); + Field nextSweepField = VoteThrottleService.class.getDeclaredField("nextAggregateSweepMs"); + nextSweepField.setAccessible(true); + nextSweepField.setLong(service, 0L); + + assertFalse(service.isAggregateBlocked("remote:not-yet-tracked")); + assertEquals(4095, aggregates.size(), + "an expired aggregate must be reclaimed before an overflow bucket is consulted"); + } + + @Test + public void testReclaimedAggregateSlotRetainsNextExpiryAfterImmediateRefill() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 999, "10s", 999, "10s", false, 999, "1s")); + for (int index = 0; index < 4096; index++) { + service.fail("ip:aggregate-primary-fill:" + index, false, true); + } + for (int index = 0; index < 4096; index++) { + service.fail("ip:aggregate-reclaim:" + index, "remote:aggregate-reclaim:" + index, false, false); + } + + Map aggregates = stateMap(service, "aggregateStates"); + Object expired = aggregates.get("remote:aggregate-reclaim:0"); + Field windowStartField = expired.getClass().getDeclaredField("windowStartMs"); + windowStartField.setAccessible(true); + long laterExistingExpiry = System.currentTimeMillis() + 60_000L; + for (Object state : aggregates.values()) windowStartField.setLong(state, laterExistingExpiry); + windowStartField.setLong(expired, System.currentTimeMillis() - 6000L); + Field nextSweepField = VoteThrottleService.class.getDeclaredField("nextAggregateSweepMs"); + nextSweepField.setAccessible(true); + nextSweepField.setLong(service, 0L); + + service.fail("ip:aggregate-reclaim:first-miss", "remote:aggregate-reclaim:first-miss", false, false); + long nextSweep = longField(service, "nextAggregateSweepMs"); + assertTrue(nextSweep > System.currentTimeMillis()); + assertTrue(nextSweep < laterExistingExpiry, + "the newly inserted aggregate's earlier expiry must bound the retained deadline"); + service.fail("ip:aggregate-reclaim:second-miss", "remote:aggregate-reclaim:second-miss", false, false); + + assertEquals(nextSweep, longField(service, "nextAggregateSweepMs"), + "the aggregate refill after reclamation must retain the next active expiry"); + assertEquals(4096, aggregates.size()); + } + + @Test + public void testInitialAggregateSaturationUsesAnOlderStateExpiry() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("60s", 999, "60s", 999, "60s", false, 999, "1s")); + for (int index = 0; index < 4096; index++) service.fail("ip:initial-aggregate-primary:" + index, false, true); + for (int index = 0; index < 4095; index++) { + service.fail("ip:initial-aggregate-client:" + index, "remote:initial-aggregate:" + index, false, false); + } + + Map aggregates = stateMap(service, "aggregateStates"); + Object expired = aggregates.get("remote:initial-aggregate:0"); + Field windowStartField = expired.getClass().getDeclaredField("windowStartMs"); + windowStartField.setAccessible(true); + long now = System.currentTimeMillis(); + for (Object state : aggregates.values()) windowStartField.setLong(state, now + 60_000L); + windowStartField.setLong(expired, now - 61_000L); + + service.fail("ip:initial-aggregate-client:capacity", "remote:initial-aggregate:capacity", false, false); + assertTrue(longField(service, "nextAggregateSweepMs") <= System.currentTimeMillis(), + "the first saturated aggregate map must retain the earlier existing expiry"); + service.fail("ip:initial-aggregate-client:replacement", "remote:initial-aggregate:replacement", false, false); + + assertFalse(aggregates.containsKey("remote:initial-aggregate:0")); + assertTrue(aggregates.containsKey("remote:initial-aggregate:replacement")); + } + + @Test + public void testConfiguredAggregateIdentitiesRemainBoundedAndCacheSaturation() throws Exception { + java.util.Set configuredRemotes = new java.util.HashSet(); + for (int index = 0; index < 5000; index++) configuredRemotes.add("192.0.2." + index); + VoteThrottleService service = new VoteThrottleService( + new ThrottleConfig(true, configuredRemotes, "5s", 999, "10s", 999, "10s", false, 999, "1s", "1s")); + + assertEquals(4096, mapSize(service, "aggregateStates")); + assertFalse(service.isAggregateBlocked("remote:unknown")); + assertEquals(Long.MAX_VALUE, longField(service, "nextAggregateSweepMs"), + "a map containing only pinned configured identities must not be rescanned on every miss"); + assertFalse(service.isAggregateBlocked("remote:another-unknown")); + assertEquals(4096, mapSize(service, "aggregateStates")); + } + + private static ThrottleConfig tunnelCfg(String... remoteIps) { + return new ThrottleConfig(true, new java.util.HashSet(java.util.Arrays.asList(remoteIps)), "5s", 1, + "10s", 1, "10s", true, 1, "60s", "60s"); + } + + private static int mapSize(VoteThrottleService service, String fieldName) throws Exception { + return stateMap(service, fieldName).size(); + } + + private static Map stateMap(VoteThrottleService service, String fieldName) throws Exception { + Field field = VoteThrottleService.class.getDeclaredField(fieldName); + field.setAccessible(true); + return (Map) field.get(service); + } + + private static long longField(VoteThrottleService service, String fieldName) throws Exception { + Field field = VoteThrottleService.class.getDeclaredField(fieldName); + field.setAccessible(true); + return field.getLong(service); + } +}