This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
The following commit(s) were added to refs/heads/rocketmq-studio by this push:
new 238bab5e5 fix(auth): keep login lockouts effective at capacity (#3046)
238bab5e5 is described below
commit 238bab5e58e4c3fd43981200838b397cddbc879e
Author: xdz997 <[email protected]>
AuthorDate: Fri Sep 4 15:42:09 2026 +0800
fix(auth): keep login lockouts effective at capacity (#3046)
---
.../rocketmq/studio/auth/LoginRateLimiter.java | 242 +++++++++++++++++----
.../rocketmq/studio/auth/LoginRateLimiterTest.java | 69 ++++++
2 files changed, 265 insertions(+), 46 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/auth/LoginRateLimiter.java
b/server/src/main/java/org/apache/rocketmq/studio/auth/LoginRateLimiter.java
index 0dbe30ae1..e14a1f432 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/auth/LoginRateLimiter.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/auth/LoginRateLimiter.java
@@ -16,17 +16,19 @@ import org.springframework.stereotype.Component;
import java.time.Clock;
import java.time.Duration;
+import java.util.Arrays;
+import java.util.HashMap;
import java.util.Locale;
import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ThreadLocalRandom;
/**
* In-memory brute-force protection for the Studio login endpoint.
*
- * <p>Failed logins are counted per username. After {@link
#MAX_FAILED_ATTEMPTS} failures within
- * {@link #FAILURE_WINDOW}, further attempts for that username are rejected for
- * {@link #LOCK_DURATION}. A successful login resets the counter. The state is
deliberately kept
- * in memory only: it throttles online guessing attacks and resets on
restart.</p>
+ * <p>Failed logins are counted per normalized username. The exact tracker has
a strict size
+ * bound. When that bound is occupied, previously unseen usernames use a
second fixed-size set
+ * of hash buckets instead of failing open or growing memory. Collisions can
share a lock only
+ * while the exact tracker is saturated; they cannot disable rate limiting.</p>
*/
@Slf4j
@Component
@@ -36,10 +38,16 @@ public class LoginRateLimiter {
static final Duration FAILURE_WINDOW = Duration.ofMinutes(5);
static final Duration LOCK_DURATION = Duration.ofMinutes(5);
static final int MAX_TRACKED_USERNAMES = 10_000;
+ static final int OVERFLOW_BUCKET_COUNT = 1_024;
- private final Map<String, AttemptState> attempts = new
ConcurrentHashMap<>();
+ private final Map<String, AttemptState> attempts = new HashMap<>();
+ private final AttemptState[] overflowAttempts;
private final Clock clock;
private final int maxTrackedUsernames;
+ private final int overflowBucketMask;
+ private final int overflowHashSeed;
+
+ private long nextExactExpiryMillis = Long.MAX_VALUE;
public LoginRateLimiter() {
this(Clock.systemUTC());
@@ -50,78 +58,220 @@ public class LoginRateLimiter {
}
LoginRateLimiter(Clock clock, int maxTrackedUsernames) {
+ this(clock, maxTrackedUsernames, OVERFLOW_BUCKET_COUNT,
+ ThreadLocalRandom.current().nextInt());
+ }
+
+ LoginRateLimiter(Clock clock, int maxTrackedUsernames,
+ int overflowBucketCount, int overflowHashSeed) {
+ if (maxTrackedUsernames <= 0) {
+ throw new IllegalArgumentException("maxTrackedUsernames must be
positive");
+ }
+ if (overflowBucketCount <= 0
+ || (overflowBucketCount & (overflowBucketCount - 1)) != 0) {
+ throw new IllegalArgumentException("overflowBucketCount must be a
power of two");
+ }
this.clock = clock;
this.maxTrackedUsernames = maxTrackedUsernames;
+ this.overflowAttempts = new AttemptState[overflowBucketCount];
+ this.overflowBucketMask = overflowBucketCount - 1;
+ this.overflowHashSeed = overflowHashSeed;
}
/**
- * Rejects the attempt with HTTP 429 while the username is locked out.
+ * Rejects the attempt with HTTP 429 while its exact or overflow state is
locked.
*/
- public void checkAllowed(String username) {
+ public synchronized void checkAllowed(String username) {
String key = key(username);
- AttemptState state = attempts.get(key);
- if (state == null || state.lockedUntilMillis() == 0) {
+ long now = clock.millis();
+ AttemptState exact = activeExactState(key, now);
+ if (exact != null) {
+ rejectIfLocked(exact, now);
return;
}
- long now = clock.millis();
- if (state.lockedUntilMillis() > now) {
- long remainingSeconds = (state.lockedUntilMillis() - now + 999) /
1000;
- throw new BusinessException(429, "Too many failed login attempts;
try again in "
- + remainingSeconds + " seconds");
+
+ reclaimExpiredExactAttemptsIfDue(now);
+ if (attempts.size() < maxTrackedUsernames) {
+ return;
+ }
+
+ int bucket = overflowBucket(key);
+ AttemptState overflow = activeOverflowState(bucket, now);
+ if (overflow != null) {
+ rejectIfLocked(overflow, now);
}
- attempts.remove(key, state);
}
public synchronized void recordFailure(String username) {
String key = key(username);
long now = clock.millis();
- if (!attempts.containsKey(key) && attempts.size() >=
maxTrackedUsernames) {
- removeExpiredAttempts(now);
- if (attempts.size() >= maxTrackedUsernames) {
- log.debug("Login rate limiter is at capacity; ignoring a new
username");
- return;
- }
+ AttemptState exact = activeExactState(key, now);
+ if (exact != null) {
+ FailureUpdate update = incrementFailure(exact, now);
+ attempts.put(key, update.state());
+ noteExactExpiry(update.state());
+ logLock(username, update, false);
+ return;
}
- AttemptState state = attempts.compute(key, (ignored, current) -> {
- AttemptState base = current;
- if (base == null || base.lockedUntilMillis() != 0
- || now - base.windowStartMillis() >=
FAILURE_WINDOW.toMillis()) {
- base = new AttemptState(now, 0, 0);
- }
- int failures = base.failureCount() + 1;
- if (failures >= MAX_FAILED_ATTEMPTS) {
- return new AttemptState(now, 0, now +
LOCK_DURATION.toMillis());
+
+ reclaimExpiredExactAttemptsIfDue(now);
+ if (attempts.size() < maxTrackedUsernames) {
+ FailureUpdate update = incrementFailure(null, now);
+ attempts.put(key, update.state());
+ noteExactExpiry(update.state());
+ logLock(username, update, false);
+ return;
+ }
+
+ int bucket = overflowBucket(key);
+ AttemptState overflow = activeOverflowState(bucket, now);
+ FailureUpdate update = incrementFailure(overflow, now);
+ overflowAttempts[bucket] = update.state();
+ logLock(username, update, true);
+ }
+
+ public synchronized void recordSuccess(String username) {
+ String key = key(username);
+ if (attempts.remove(key) != null) {
+ // Once an exact slot is available, overflow state from the
saturated period is no
+ // longer needed and must not affect a later saturation episode.
+ clearOverflowAttempts();
+ return;
+ }
+ overflowAttempts[overflowBucket(key)] = null;
+ }
+
+ synchronized int trackedUsernameCount() {
+ return attempts.size();
+ }
+
+ synchronized int activeOverflowBucketCount() {
+ int active = 0;
+ long now = clock.millis();
+ for (int index = 0; index < overflowAttempts.length; index++) {
+ if (activeOverflowState(index, now) != null) {
+ active++;
}
- return new AttemptState(base.windowStartMillis(), failures, 0);
- });
- if (state != null && state.lockedUntilMillis() != 0) {
+ }
+ return active;
+ }
+
+ private AttemptState activeExactState(String key, long now) {
+ AttemptState state = attempts.get(key);
+ if (state == null || !state.expiredAt(now)) {
+ return state;
+ }
+ attempts.remove(key);
+ clearOverflowAttempts();
+ return null;
+ }
+
+ private AttemptState activeOverflowState(int bucket, long now) {
+ AttemptState state = overflowAttempts[bucket];
+ if (state == null || !state.expiredAt(now)) {
+ return state;
+ }
+ overflowAttempts[bucket] = null;
+ return null;
+ }
+
+ private void reclaimExpiredExactAttemptsIfDue(long now) {
+ if (now < nextExactExpiryMillis) {
+ return;
+ }
+ boolean removed = attempts.entrySet().removeIf(entry ->
entry.getValue().expiredAt(now));
+ nextExactExpiryMillis = attempts.values().stream()
+ .mapToLong(AttemptState::expiresAtMillis)
+ .min()
+ .orElse(Long.MAX_VALUE);
+ if (removed) {
+ clearOverflowAttempts();
+ }
+ }
+
+ private FailureUpdate incrementFailure(AttemptState current, long now) {
+ if (current != null && current.lockedAt(now)) {
+ // A request that passed checkAllowed before another request
established this lock
+ // may finish later. Its failure must not shorten or clear the
active lock.
+ return new FailureUpdate(current, false);
+ }
+ AttemptState base = current == null || current.expiredAt(now)
+ ? new AttemptState(now, 0, 0)
+ : current;
+ int failures = base.failureCount() + 1;
+ if (failures >= MAX_FAILED_ATTEMPTS) {
+ return new FailureUpdate(
+ new AttemptState(now, 0, now + LOCK_DURATION.toMillis()),
true);
+ }
+ return new FailureUpdate(
+ new AttemptState(base.windowStartMillis(), failures, 0),
false);
+ }
+
+ private void rejectIfLocked(AttemptState state, long now) {
+ if (!state.lockedAt(now)) {
+ return;
+ }
+ long remainingSeconds = (state.lockedUntilMillis() - now + 999) / 1000;
+ throw new BusinessException(429, "Too many failed login attempts; try
again in "
+ + remainingSeconds + " seconds");
+ }
+
+ private void logLock(String username, FailureUpdate update, boolean
overflow) {
+ if (!update.newlyLocked()) {
+ return;
+ }
+ if (overflow) {
+ log.warn("Locked a saturated login-rate bucket after {} failed
attempts within {} minutes",
+ MAX_FAILED_ATTEMPTS, FAILURE_WINDOW.toMinutes());
+ } else {
log.warn("Locked login for user {} after {} failed attempts within
{} minutes",
username, MAX_FAILED_ATTEMPTS, FAILURE_WINDOW.toMinutes());
}
}
- public void recordSuccess(String username) {
- attempts.remove(key(username));
+ private void noteExactExpiry(AttemptState state) {
+ nextExactExpiryMillis = Math.min(nextExactExpiryMillis,
state.expiresAtMillis());
}
- int trackedUsernameCount() {
- return attempts.size();
+ private void clearOverflowAttempts() {
+ Arrays.fill(overflowAttempts, null);
}
- private void removeExpiredAttempts(long now) {
- attempts.entrySet().removeIf(entry -> {
- AttemptState state = entry.getValue();
- if (state.lockedUntilMillis() != 0) {
- return state.lockedUntilMillis() <= now;
- }
- return now - state.windowStartMillis() >=
FAILURE_WINDOW.toMillis();
- });
+ private int overflowBucket(String key) {
+ long hash = 0xcbf29ce484222325L ^
Integer.toUnsignedLong(overflowHashSeed);
+ for (int index = 0; index < key.length(); index++) {
+ hash ^= key.charAt(index);
+ hash *= 0x100000001b3L;
+ }
+ hash ^= hash >>> 33;
+ hash *= 0xff51afd7ed558ccdL;
+ hash ^= hash >>> 33;
+ return ((int) hash) & overflowBucketMask;
}
private String key(String username) {
return username.trim().toLowerCase(Locale.ROOT);
}
+ private record FailureUpdate(AttemptState state, boolean newlyLocked) {
+ }
+
private record AttemptState(long windowStartMillis, int failureCount, long
lockedUntilMillis) {
+
+ boolean lockedAt(long now) {
+ return lockedUntilMillis > now;
+ }
+
+ boolean expiredAt(long now) {
+ return lockedUntilMillis != 0
+ ? lockedUntilMillis <= now
+ : now - windowStartMillis >= FAILURE_WINDOW.toMillis();
+ }
+
+ long expiresAtMillis() {
+ return lockedUntilMillis != 0
+ ? lockedUntilMillis
+ : windowStartMillis + FAILURE_WINDOW.toMillis();
+ }
}
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/auth/LoginRateLimiterTest.java
b/server/src/test/java/org/apache/rocketmq/studio/auth/LoginRateLimiterTest.java
index c18e08f21..617510167 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/auth/LoginRateLimiterTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/auth/LoginRateLimiterTest.java
@@ -150,6 +150,75 @@ class LoginRateLimiterTest {
.isInstanceOf(BusinessException.class);
}
+ @Test
+ void capacityShouldNotDisableRateLimitingForAnUntrackedUsernameTest() {
+ limiter = new LoginRateLimiter(clock, 2);
+ limiter.recordFailure("decoy-one");
+ limiter.recordFailure("decoy-two");
+
+ for (int attempt = 0; attempt < LoginRateLimiter.MAX_FAILED_ATTEMPTS;
attempt++) {
+ assertThatCode(() ->
limiter.checkAllowed("operator")).doesNotThrowAnyException();
+ limiter.recordFailure("operator");
+ }
+
+ assertThat(limiter.trackedUsernameCount()).isEqualTo(2);
+ assertThatThrownBy(() -> limiter.checkAllowed("operator"))
+ .isInstanceOf(BusinessException.class)
+ .satisfies(exception ->
+ assertThat(((BusinessException)
exception).getCode()).isEqualTo(429));
+ }
+
+ @Test
+ void trackerCapacityShouldRemainBoundedWhenAllSlotsAreLockedTest() {
+ limiter = new LoginRateLimiter(clock, 2);
+ for (int attempt = 0; attempt < LoginRateLimiter.MAX_FAILED_ATTEMPTS;
attempt++) {
+ limiter.recordFailure("operator");
+ limiter.recordFailure("second-user");
+ }
+
+ for (int attempt = 0; attempt < LoginRateLimiter.MAX_FAILED_ATTEMPTS;
attempt++) {
+ limiter.recordFailure("attacker");
+ }
+ assertThat(limiter.trackedUsernameCount()).isEqualTo(2);
+ assertThatThrownBy(() -> limiter.checkAllowed("operator"))
+ .isInstanceOf(BusinessException.class);
+ assertThatThrownBy(() -> limiter.checkAllowed("second-user"))
+ .isInstanceOf(BusinessException.class);
+ assertThatThrownBy(() -> limiter.checkAllowed("attacker"))
+ .isInstanceOf(BusinessException.class);
+ assertThat(limiter.activeOverflowBucketCount()).isEqualTo(1);
+ }
+
+ @Test
+ void failureAlreadyInFlightShouldNotClearAnActiveLockTest() {
+ for (int attempt = 0; attempt < LoginRateLimiter.MAX_FAILED_ATTEMPTS;
attempt++) {
+ limiter.recordFailure("operator");
+ }
+
+ // A login request can pass checkAllowed before another request
creates the lock,
+ // then finish password verification and record its failure after the
lock exists.
+ limiter.recordFailure("operator");
+
+ assertThatThrownBy(() -> limiter.checkAllowed("operator"))
+ .isInstanceOf(BusinessException.class)
+ .satisfies(exception ->
+ assertThat(((BusinessException)
exception).getCode()).isEqualTo(429));
+ }
+
+ @Test
+ void activeLocksShouldNotEvictEachOtherWhenNewFailuresArriveTest() {
+ limiter = new LoginRateLimiter(clock, 2);
+ for (int attempt = 0; attempt < LoginRateLimiter.MAX_FAILED_ATTEMPTS;
attempt++) {
+ limiter.recordFailure("operator");
+ limiter.recordFailure("second-user");
+ }
+ limiter.recordFailure("operator");
+
+ assertThat(limiter.trackedUsernameCount()).isEqualTo(2);
+ assertThatThrownBy(() -> limiter.checkAllowed("second-user"))
+ .isInstanceOf(BusinessException.class);
+ }
+
private static final class MutableClock extends Clock {
private Instant now;
