diff --git a/obp-api/src/main/scala/code/api/cache/Redis.scala b/obp-api/src/main/scala/code/api/cache/Redis.scala index cd0495b635..8cba91faa3 100644 --- a/obp-api/src/main/scala/code/api/cache/Redis.scala +++ b/obp-api/src/main/scala/code/api/cache/Redis.scala @@ -302,6 +302,94 @@ object Redis extends MdcLoggable { finally jedis.close() } + /** Outcome of [[quotaCheckAndCount]]. */ + sealed trait QuotaDecision + /** A period refused the call: `periodIndex` into the lists passed in (0-based), the counter's + * current count and its remaining TTL in seconds (negative when the key has no expiry or does + * not exist, as Redis TTL reports it). Nothing was counted. */ + final case class QuotaRefused(periodIndex: Int, count: Long, ttlSeconds: Long) extends QuotaDecision + /** Every period passed and the call was counted in all of them: (count after this call, + * remaining TTL in seconds) per key, in the order passed in. */ + final case class QuotaCounted(countsAndTtls: List[(Long, Long)]) extends QuotaDecision + + /** + * The consumer quota in one script: check every period's limit against its counter and, only + * when all pass, count the call in every period. + * + * KEYS[i] counter key of period i, shortest period first (precedence of the 429) + * ARGV[i] limit of period i: -1 unlimited (not checked, still counted), 0 blocked + * (refused without reading the counter's value), > 0 max calls in the period + * ARGV[n + i] length of period i in seconds, the TTL a new counter gets + * + * Returns {0, i, count, ttl} when period i refuses: count is what its counter holds now (0 when + * the key is absent or has no expiry) and ttl is the key's TTL as Redis reports it. Returns + * {1, c1, ttl1, ..., cn, ttln} after counting: the post-increment count and TTL of each key. + * A key found without an expiry is treated as expired on the check and recreated on the count, + * as incrementWithTtlScript does. Everything runs atomically server-side, so the check and + * the count of one call cannot interleave with another call's: at most `limit` calls pass. + */ + private val quotaCheckAndCountScript = + """local n = #KEYS + |for i = 1, n do + | local limit = tonumber(ARGV[i]) + | if limit >= 0 then + | local ttl = redis.call('TTL', KEYS[i]) + | local count = 0 + | if ttl > 0 then count = tonumber(redis.call('GET', KEYS[i]) or '0') end + | if limit == 0 or count + 1 > limit then + | return {0, i, count, ttl} + | end + | end + |end + |local out = {1} + |for i = 1, n do + | local c = redis.call('INCR', KEYS[i]) + | if c == 1 then + | redis.call('EXPIRE', KEYS[i], ARGV[n + i]) + | elseif redis.call('TTL', KEYS[i]) < 0 then + | redis.call('SET', KEYS[i], 1, 'EX', ARGV[n + i]) + | c = 1 + | end + | out[#out + 1] = c + | out[#out + 1] = redis.call('TTL', KEYS[i]) + |end + |return out""".stripMargin + + /** + * Check and count one call against several period counters in a single atomic round trip + * (see quotaCheckAndCountScript). `keys`, `limits` and `periodSeconds` are parallel lists, + * shortest period first. Throws when Redis is unreachable; the caller decides whether to fail + * open. + */ + def quotaCheckAndCount(keys: List[String], limits: List[Long], periodSeconds: List[Int]): QuotaDecision = { + require(keys.nonEmpty && keys.size == limits.size && keys.size == periodSeconds.size, + s"quotaCheckAndCount needs one limit and one period per key: ${keys.size} keys, ${limits.size} limits, ${periodSeconds.size} periods") + import scala.jdk.CollectionConverters._ + val startNanos = System.nanoTime() + var succeeded = false + try { + val reply: List[Long] = withJedis { jedis => + jedis.eval( + quotaCheckAndCountScript, + keys.asJava, + (limits.map(_.toString) ++ periodSeconds.map(_.toString)).asJava + ).asInstanceOf[java.util.List[java.lang.Long]].asScala.map(_.longValue()).toList + } + val decision = reply match { + case 0L :: index :: count :: ttl :: Nil => QuotaRefused(index.toInt - 1, count, ttl) + case 1L :: countsAndTtls if countsAndTtls.size == keys.size * 2 => + QuotaCounted(countsAndTtls.grouped(2).collect { case List(count, ttl) => (count, ttl) }.toList) + case other => throw new RuntimeException(s"quotaCheckAndCount: unexpected reply $other for ${keys.size} keys") + } + succeeded = true + decision + } finally { + code.telemetry.Telemetry.timer("obp.api.redis.commands", + "command" -> "EVAL_QUOTA_CHECK_AND_COUNT", "result" -> (if (succeeded) "success" else "error")) + .record(System.nanoTime() - startNanos, java.util.concurrent.TimeUnit.NANOSECONDS) + } + } + /** * Atomic `SET key value EX ttlSeconds NX` (Jedis 2.9.0 five-arg overload). Sets the key with a TTL * only if it does not already exist, in a single command. Returns true iff this call set the key. diff --git a/obp-api/src/main/scala/code/api/util/Glossary.scala b/obp-api/src/main/scala/code/api/util/Glossary.scala index 8c59b4f8a7..0cde9f60f7 100644 --- a/obp-api/src/main/scala/code/api/util/Glossary.scala +++ b/obp-api/src/main/scala/code/api/util/Glossary.scala @@ -606,8 +606,8 @@ object Glossary extends MdcLoggable { |1. **Rate Limit Records**: Stored in the `RateLimiting` table with date ranges (from_date, to_date) |2. **Multiple Records**: A consumer can have multiple active rate limit records that overlap |3. **Aggregation**: When multiple records are active, per period: `-1` values are ignored and the rest (`0` or positive) are summed; a sum of `0` blocks the period; nothing to sum (all `-1`) means unlimited - |4. **Enforcement**: On every API request, the system checks Redis counters against the aggregated limits - |5. **Counting**: Every served request is counted in the Redis counter of every period, whether or not that period has a limit, so the call-counter endpoints show a Consumer's activity even when nothing limits it. A blocked period (sum `0`) serves nothing, so nothing is counted under it. + |4. **Enforcement**: On every API request, one Redis script checks the counter of every period against the aggregated limits, shortest period first, and refuses with 429 at the first period that is full + |5. **Counting**: In the same script, once every period has passed, the request is counted in the Redis counter of every period, whether or not that period has a limit, so the call-counter endpoints show a Consumer's activity even when nothing limits it. A blocked period (sum `0`) serves nothing, so nothing is counted under it. Because the check and the count run atomically in one round trip, concurrent requests cannot pass a check together and overshoot a limit. | |### Time Periods | diff --git a/obp-api/src/main/scala/code/api/util/RateLimitingUtil.scala b/obp-api/src/main/scala/code/api/util/RateLimitingUtil.scala index 52b64c2afb..0c5c1e3dfb 100644 --- a/obp-api/src/main/scala/code/api/util/RateLimitingUtil.scala +++ b/obp-api/src/main/scala/code/api/util/RateLimitingUtil.scala @@ -228,40 +228,6 @@ object RateLimitingUtil extends MdcLoggable { RateLimitCounterState(calls, normalizedTtl, status) } private def createUniqueKey(consumerKey: String, period: LimitCallPeriod) = CALL_COUNTER_PREFIX + consumerKey + "_" + RateLimitingPeriod.toString(period) - private def underConsumerLimits(consumerKey: String, period: LimitCallPeriod, limit: Long): Boolean = { - - if (useConsumerLimits) { - (limit) match { - case l if l > 0 => // Limit is set, check against Redis counter - val state = getCounterState(consumerKey, period) - state.status match { - case "ACTIVE" => - // Counter is active, check if we're under limit - // +1 means we count the current call as well. We increment later i.e after successful call. - state.calls.getOrElse(0L) + 1 <= limit - case "NO_COUNTER" | "EXPIRED" => - // No counter or expired -> allow (first call or period expired) - true - case "REDIS_UNAVAILABLE" => - // Redis unavailable -> fail open (allow request) - logger.warn(s"Redis unavailable when checking rate limit for consumer $consumerKey, period $period - allowing request") - true - case _ => - // Unknown status -> fail open (allow request) - logger.warn(s"Unknown status '${state.status}' when checking rate limit for consumer $consumerKey, period $period - allowing request") - true - } - case 0 => - // A limit of 0 means blocked: refuse every call for this period, without touching Redis - false - case _ => - // A negative limit (-1) means unlimited for this period - true - } - } else { - true // Rate Limiting disabled implies successful result - } - } /** Pure Redis INCR with create-if-missing for a fully-formed key. * No gates, no key formatting — call sites pass the final key and supply their own enable flags. @@ -282,41 +248,65 @@ object RateLimitingUtil extends MdcLoggable { } } + /** The periods of the consumer quota, shortest first: the order the limits are checked in, so the + * 429 names the shortest exhausted period, and the order the X-Rate-Limit-* headers prefer. */ + private val consumerPeriods: List[LimitCallPeriod] = List(PER_SECOND, PER_MINUTE, PER_HOUR, PER_DAY, PER_WEEK, PER_MONTH) + + private def limitOf(c: CallLimit, period: LimitCallPeriod): Long = period match { + case PER_SECOND => c.per_second + case PER_MINUTE => c.per_minute + case PER_HOUR => c.per_hour + case PER_DAY => c.per_day + case PER_WEEK => c.per_week + case PER_MONTH => c.per_month + case PER_YEAR => -1 + } + + /** What one call did to the quota counters. */ + private[util] sealed trait QuotaOutcome + /** `period` refused the call (its limit is 0, or its counter is full). Nothing was counted. + * `resetSeconds` is the counter's remaining TTL, 0 when it has none. */ + private[util] final case class Refused(period: LimitCallPeriod, limit: Long, resetSeconds: Long) extends QuotaOutcome + /** Every period passed and the call was counted in all of them, unlimited ones included: + * (period, limit, (ttl, count after this call)) in [[consumerPeriods]] order. */ + private[util] final case class Counted(perPeriod: List[(LimitCallPeriod, Long, (Long, Long))]) extends QuotaOutcome + /** - * Count the call in one period's counter, after the request passed every limit check. - * - * Counted whatever the period's aggregated limit is, -1 (unlimited) included. Enforcement never - * reads the counter of an unlimited period - underConsumerLimits decides -1 and 0 without touching - * Redis and only compares the counter when the limit is positive - so a counter that is always - * kept changes nothing about what is refused, and gives visibility of the consumer's activity - * through the call-counter endpoints even when no limit applies. A period whose aggregated limit - * is 0 is refused before counting, so nothing is ever served or counted under it. + * Check the call against every period's limit and, when all pass, count it in every period, in + * ONE Redis round trip (Redis.quotaCheckAndCount, a Lua script). The script runs atomically, so + * two concurrent calls cannot both read a counter one below its limit and both pass: at most + * `limit` calls are served per period. * - * Only `use_consumer_limits=false` skips counting, so an instance without Redis is left alone. + * Counted whatever a period's limit is, -1 (unlimited) included: a limit of -1 is not checked, so + * a counter that is always kept changes nothing about what is refused, and gives visibility of + * the consumer's activity through the call-counter endpoints even when no limit applies. A limit + * of 0 refuses before counting, so nothing is ever served or counted under it. * * @param consumerKey The consumer ID (or the client IP for anonymous access) - * @param period The time period (PER_SECOND, PER_MINUTE, ...) - * @return (TTL in seconds, current counter value); (-1, -1) when not counted or Redis is unreachable + * @param periodsWithLimits the periods to check and count, shortest first, each with its limit + * @return None when nothing was checked or counted: `use_consumer_limits=false` (an instance + * without Redis is left alone) or Redis unreachable (fail open: the call is allowed and + * no X-Rate-Limit-* headers are set) */ - private def incrementConsumerCounters(consumerKey: String, period: LimitCallPeriod): (Long, Long) = - if (useConsumerLimits) incrementCounter(createUniqueKey(consumerKey, period), period) else (-1, -1) - - /** - * Get remaining TTL (time to live) for a rate limit counter. - * Used to populate X-Rate-Limit-Reset header when rate limit is exceeded. - * - * NOTE: This function could be further optimized by eliminating it entirely. - * We already call getCounterState() in underConsumerLimits(), so we could - * cache/reuse that TTL value instead of making another Redis call here. - * - * @param consumerKey The consumer ID or IP address - * @param period The time period - * @return Seconds until counter resets, or 0 if no counter exists - */ - private def ttl(consumerKey: String, period: LimitCallPeriod): Long = { - val state = getCounterState(consumerKey, period) - state.ttl.getOrElse(0L) - } + private def checkAndCount(consumerKey: String, periodsWithLimits: List[(LimitCallPeriod, Long)]): Option[QuotaOutcome] = + if (!useConsumerLimits) None + else try { + Redis.quotaCheckAndCount( + keys = periodsWithLimits.map { case (period, _) => createUniqueKey(consumerKey, period) }, + limits = periodsWithLimits.map(_._2), + periodSeconds = periodsWithLimits.map { case (period, _) => RateLimitingPeriod.toSeconds(period).toInt } + ) match { + case Redis.QuotaRefused(index, _, ttl) => + val (period, limit) = periodsWithLimits(index) + Some(Refused(period, limit, Math.max(0L, ttl))) // TTL -1/-2 (no expiry / no key) reads as 0, as getCounterState does + case Redis.QuotaCounted(countsAndTtls) => + Some(Counted(periodsWithLimits.zip(countsAndTtls).map { case ((period, limit), (count, ttl)) => (period, limit, (ttl, count)) })) + } + } catch { + case e: Throwable => + logger.error(s"Redis unavailable when checking the rate limit for $consumerKey - allowing request", e) + None + } @@ -359,7 +349,10 @@ object RateLimitingUtil extends MdcLoggable { * - Keys format: {consumer_id}_{PERIOD} (e.g., "consumer123_PER_MINUTE") * - Values: current call count within the time window * - TTL: automatically expires keys when time period ends - * - Atomic operations ensure thread-safe counter increments + * - One Lua script per request (Redis.quotaCheckAndCount) checks every period and, when all + * pass, counts the call in every period: one round trip, atomic, so concurrent calls cannot + * pass a check together and overshoot the limit. A refusal comes back with the counter's TTL, + * so the 429 headers need no further read. * * RATE LIMIT HEADERS: * - Sets X-Rate-Limit-Limit: maximum allowed requests for the period @@ -385,132 +378,58 @@ object RateLimitingUtil extends MdcLoggable { if (limit == 0) TooManyRequests + s" Anonymous access is blocked: the anonymous rate limit ${RateLimitingPeriod.humanReadable(period)} is 0." else TooManyRequests + s" We only allow $limit requests ${RateLimitingPeriod.humanReadable(period)} for anonymous access." - // Helper function to set rate limit headers in successful responses - def setXRateLimits(c: CallLimit, z: (Long, Long), period: LimitCallPeriod): Option[CallContext] = { - val limit = period match { - case PER_SECOND => c.per_second - case PER_MINUTE => c.per_minute - case PER_HOUR => c.per_hour - case PER_DAY => c.per_day - case PER_WEEK => c.per_week - case PER_MONTH => c.per_month - case PER_YEAR => -1 - } - userAndCallContext._2.map(_.copy(xRateLimitLimit = limit)) - .map(_.copy(xRateLimitReset = z._1)) - .map(_.copy(xRateLimitRemaining = Math.max(0, limit - z._2))) - } - // Helper function to set rate limit headers for anonymous access - def setXRateLimitsAnonymous(id: String, z: (Long, Long), period: LimitCallPeriod): Option[CallContext] = { - val limit = period match { - case PER_HOUR => perHourLimitAnonymous - case _ => -1 - } + // The X-Rate-Limit-* headers of a served call: the limit, the counter's TTL and what is left of the limit + def setXRateLimits(limit: Long, ttlAndCount: (Long, Long)): Option[CallContext] = userAndCallContext._2.map(_.copy(xRateLimitLimit = limit)) - .map(_.copy(xRateLimitReset = z._1)) - .map(_.copy(xRateLimitRemaining = Math.max(0, limit - z._2))) - } + .map(_.copy(xRateLimitReset = ttlAndCount._1)) + .map(_.copy(xRateLimitRemaining = Math.max(0, limit - ttlAndCount._2))) - // Helper function to create rate limit exceeded response with remaining TTL for authorized users - def exceededRateLimit(c: CallLimit, period: LimitCallPeriod): Option[CallContextLight] = { - val remain = ttl(c.consumer_id, period) - val limit = period match { - case PER_SECOND => c.per_second - case PER_MINUTE => c.per_minute - case PER_HOUR => c.per_hour - case PER_DAY => c.per_day - case PER_WEEK => c.per_week - case PER_MONTH => c.per_month - case PER_YEAR => -1 - } - userAndCallContext._2.map(_.copy(xRateLimitLimit = limit)) - .map(_.copy(xRateLimitReset = remain)) + // The X-Rate-Limit-* headers of a refused call: the exhausted limit, its reset and nothing remaining + def exceededRateLimit(refused: Refused): Option[CallContextLight] = + userAndCallContext._2.map(_.copy(xRateLimitLimit = refused.limit)) + .map(_.copy(xRateLimitReset = refused.resetSeconds)) .map(_.copy(xRateLimitRemaining = 0)).map(_.toLight) - } - // Helper function to create rate limit exceeded response for anonymous users - def exceededRateLimitAnonymous(id: String, period: LimitCallPeriod): Option[CallContextLight] = { - val remain = ttl(id, period) - val limit = period match { - case PER_HOUR => perHourLimitAnonymous - case _ => -1 - } - userAndCallContext._2.map(_.copy(xRateLimitLimit = limit)) - .map(_.copy(xRateLimitReset = remain)) - .map(_.copy(xRateLimitRemaining = 0)).map(_.toLight) - } + def refuse(message: String, refused: Refused): (Box[User], Option[CallContext]) = + (fullBoxOrException(Empty ~> APIFailureNewStyle(message, 429, exceededRateLimit(refused))), userAndCallContext._2) // Main logic: check if we have a CallContext and determine access type userAndCallContext._2 match { case Some(cc) => cc.rateLimiting match { case Some(rl) => // AUTHORIZED ACCESS - consumer has valid credentials and rate limits - // Create rate limiting key for Redis storage using consumer_id - val rateLimitingKey = rl.consumer_id - // Check if current request would exceed any of the 6 rate limits - val checkLimits = List( - underConsumerLimits(rateLimitingKey, PER_SECOND, rl.per_second), - underConsumerLimits(rateLimitingKey, PER_MINUTE, rl.per_minute), - underConsumerLimits(rateLimitingKey, PER_HOUR, rl.per_hour), - underConsumerLimits(rateLimitingKey, PER_DAY, rl.per_day), - underConsumerLimits(rateLimitingKey, PER_WEEK, rl.per_week), - underConsumerLimits(rateLimitingKey, PER_MONTH, rl.per_month) - ) - // Return 429 error for first exceeded limit (shorter periods take precedence) - checkLimits match { - case x1 :: x2 :: x3 :: x4 :: x5 :: x6 :: Nil if x1 == false => - (fullBoxOrException(Empty ~> APIFailureNewStyle(composeMsgAuthorizedAccess(PER_SECOND, rl.per_second, rl.consumer_id), 429, exceededRateLimit(rl, PER_SECOND))), userAndCallContext._2) - case x1 :: x2 :: x3 :: x4 :: x5 :: x6 :: Nil if x2 == false => - (fullBoxOrException(Empty ~> APIFailureNewStyle(composeMsgAuthorizedAccess(PER_MINUTE, rl.per_minute, rl.consumer_id), 429, exceededRateLimit(rl, PER_MINUTE))), userAndCallContext._2) - case x1 :: x2 :: x3 :: x4 :: x5 :: x6 :: Nil if x3 == false => - (fullBoxOrException(Empty ~> APIFailureNewStyle(composeMsgAuthorizedAccess(PER_HOUR, rl.per_hour, rl.consumer_id), 429, exceededRateLimit(rl, PER_HOUR))), userAndCallContext._2) - case x1 :: x2 :: x3 :: x4 :: x5 :: x6 :: Nil if x4 == false => - (fullBoxOrException(Empty ~> APIFailureNewStyle(composeMsgAuthorizedAccess(PER_DAY, rl.per_day, rl.consumer_id), 429, exceededRateLimit(rl, PER_DAY))), userAndCallContext._2) - case x1 :: x2 :: x3 :: x4 :: x5 :: x6 :: Nil if x5 == false => - (fullBoxOrException(Empty ~> APIFailureNewStyle(composeMsgAuthorizedAccess(PER_WEEK, rl.per_week, rl.consumer_id), 429, exceededRateLimit(rl, PER_WEEK))), userAndCallContext._2) - case x1 :: x2 :: x3 :: x4 :: x5 :: x6 :: Nil if x6 == false => - (fullBoxOrException(Empty ~> APIFailureNewStyle(composeMsgAuthorizedAccess(PER_MONTH, rl.per_month, rl.consumer_id), 429, exceededRateLimit(rl, PER_MONTH))), userAndCallContext._2) - case _ => - // All limits passed - count the call in every period, then set the X-Rate-Limit-* - // headers from the shortest period that HAS a limit. The counter alone cannot pick - // that period: every period is counted, unlimited ones included, so a live - // per-second counter says nothing about whether a per-second limit exists. With no - // limited period the CallContext keeps its defaults and the headers read -1. - val counted: List[(LimitCallPeriod, Long, (Long, Long))] = List( - (PER_SECOND, rl.per_second, incrementConsumerCounters(rateLimitingKey, PER_SECOND)), - (PER_MINUTE, rl.per_minute, incrementConsumerCounters(rateLimitingKey, PER_MINUTE)), - (PER_HOUR, rl.per_hour, incrementConsumerCounters(rateLimitingKey, PER_HOUR)), - (PER_DAY, rl.per_day, incrementConsumerCounters(rateLimitingKey, PER_DAY)), - (PER_WEEK, rl.per_week, incrementConsumerCounters(rateLimitingKey, PER_WEEK)), - (PER_MONTH, rl.per_month, incrementConsumerCounters(rateLimitingKey, PER_MONTH)) - ) - counted.collectFirst { case (period, limit, ttlAndCount) if limit > 0 && ttlAndCount._1 > 0 => (period, ttlAndCount) } match { - case Some((period, ttlAndCount)) => (userAndCallContext._1, setXRateLimits(rl, ttlAndCount, period)) - case None => (userAndCallContext._1, userAndCallContext._2) + // One round trip checks the 6 limits, shortest period first, and counts the call when all pass + checkAndCount(rl.consumer_id, consumerPeriods.map(period => (period, limitOf(rl, period)))) match { + case Some(refused: Refused) => + // 429 for the first exceeded limit (shorter periods take precedence) + refuse(composeMsgAuthorizedAccess(refused.period, refused.limit, rl.consumer_id), refused) + case Some(Counted(perPeriod)) => + // All limits passed and every period counted. Set the X-Rate-Limit-* headers from the + // shortest period that HAS a limit. The counter alone cannot pick that period: every + // period is counted, unlimited ones included, so a live per-second counter says nothing + // about whether a per-second limit exists. With no limited period the CallContext keeps + // its defaults and the headers read -1. + perPeriod.collectFirst { case (_, limit, ttlAndCount) if limit > 0 && ttlAndCount._1 > 0 => (limit, ttlAndCount) } match { + case Some((limit, ttlAndCount)) => (userAndCallContext._1, setXRateLimits(limit, ttlAndCount)) + case None => (userAndCallContext._1, userAndCallContext._2) } + case None => // limits off, or Redis unreachable: served, not counted, no headers + (userAndCallContext._1, userAndCallContext._2) } case None => // ANONYMOUS ACCESS - no consumer credentials, use IP-based limiting // Use client IP address as rate limiting key for anonymous access val consumerId = cc.ipAddress // Anonymous access only has per-hour limits to prevent abuse - val checkLimits = List( - underConsumerLimits(consumerId, PER_HOUR, perHourLimitAnonymous) - ) - checkLimits match { - case x1 :: Nil if !x1 => + val limit = perHourLimitAnonymous.toLong + checkAndCount(consumerId, List((PER_HOUR, limit))) match { + case Some(refused: Refused) => // Return 429 error if anonymous hourly limit exceeded - (fullBoxOrException(Empty ~> APIFailureNewStyle(composeMsgAnonymousAccess(PER_HOUR, perHourLimitAnonymous), 429, exceededRateLimitAnonymous(consumerId, PER_HOUR))), userAndCallContext._2) + refuse(composeMsgAnonymousAccess(PER_HOUR, limit), refused) + case Some(Counted((_, _, ttlAndCount) :: _)) if ttlAndCount._1 > 0 => + // Limit not exceeded - counted, set headers + (userAndCallContext._1, setXRateLimits(limit, ttlAndCount)) case _ => - // Limit not exceeded - increment counter and set headers - val incrementCounters = List ( - incrementConsumerCounters(consumerId, PER_HOUR) - ) - incrementCounters match { - case x1 :: Nil if x1._1 > 0 => - (userAndCallContext._1, setXRateLimitsAnonymous(consumerId, x1, PER_HOUR)) - case _ => - (userAndCallContext._1, userAndCallContext._2) - } + (userAndCallContext._1, userAndCallContext._2) } } case _ => // No CallContext available - pass through without rate limiting diff --git a/obp-api/src/test/scala/code/concurrency/CONCURRENCY_HAZARDS.md b/obp-api/src/test/scala/code/concurrency/CONCURRENCY_HAZARDS.md index 7e87bc2f38..dba94336bb 100644 --- a/obp-api/src/test/scala/code/concurrency/CONCURRENCY_HAZARDS.md +++ b/obp-api/src/test/scala/code/concurrency/CONCURRENCY_HAZARDS.md @@ -202,7 +202,7 @@ The remaining entries (Q, T, V, X, Y) are intentionally untested. | Q | `revokeAccess` vs `grant` check-then-act | — | Same `AccountAccess` check-then-act family as **R**; the window is narrow → non-deterministic barrier test would be flaky (false-green). The class is proven by **R**. | | T | `createTransactionRequestBulk` per-leg balance | — | Verdict: unconfirmed intra-request self-race. `saveTransaction` mutates the passed object's `accountBalance` field — sequential legs may see the updated value, not a stale one. Writing a possibly-false test was rejected. | | V | Berlin Group `usesSoFarTodayCounter` lost-increment | — | Same counter lost-update class as H/K; requires fully-signed recurring BG consent + TPP headers — disproportionate setup for a class already proven. | -| X | Consumer rate-limit `underConsumerLimits` TOCTOU | — | Real and high-impact (limit bypass), but active-limit lookup is cached ~1 hour → HTTP-layer timing unreliable → would be flaky. | +| X | Consumer rate-limit `underConsumerLimits` TOCTOU | `ConcurrentConsumerQuotaScriptTest` Q6 | **Fixed.** The GET-then-INCR gate is gone: `RateLimitingUtil.underCallLimits` now checks every period and counts the call in one Lua script (`Redis.quotaCheckAndCount`), so the check and the count cannot interleave. Q6 drives the script directly (20 callers, limit 5, exactly 5 pass), sidestepping the cached active-limit lookup that made an HTTP-layer test flaky. | | Y | `AuthRateLimiter` cold-start SET-vs-INCR collision | — | Same rate-limit class as X; runs in shadow mode by default. Same flakiness concern. | | Z | `MappedAgentProvider.updateAgentStatus` | — | Re-audited as **not a hazard**: sets both fields and `saveMe()`s the whole row — normal last-writer-wins PUT semantics, not field tearing. | diff --git a/obp-api/src/test/scala/code/concurrency/ConcurrentConsumerQuotaScriptTest.scala b/obp-api/src/test/scala/code/concurrency/ConcurrentConsumerQuotaScriptTest.scala new file mode 100644 index 0000000000..e2abe50ec7 --- /dev/null +++ b/obp-api/src/test/scala/code/concurrency/ConcurrentConsumerQuotaScriptTest.scala @@ -0,0 +1,173 @@ +/** +Open Bank Project - API +Copyright (C) 2011-2026, TESOBE GmbH. + +This program is free software: you can redistribute it and/or modify +it under the terms of the GNU Affero General Public License as published by +the Free Software Foundation, either version 3 of the License, or +(at your option) any later version. + +This program is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU Affero General Public License for more details. + +You should have received a copy of the GNU Affero General Public License +along with this program. If not, see . + +Email: contact@tesobe.com +TESOBE GmbH. +Osloer Strasse 16/17 +Berlin 13359, Germany + +This product includes software developed at +TESOBE (http://www.tesobe.com/) + */ +package code.concurrency + +import code.api.JedisMethod +import code.api.cache.Redis +import code.api.cache.Redis.{QuotaCounted, QuotaDecision, QuotaRefused} +import code.setup.RedisTestTarget + +import java.util.UUID +import java.util.concurrent.atomic.AtomicInteger + +/** + * Redis.quotaCheckAndCount: the one Lua script behind the consumer quota + * (RateLimitingUtil.underCallLimits). It checks every period's limit and, only when all pass, + * counts the call in every period, atomically. These scenarios pin the contract the Scala side + * relies on and close hazard X of CONCURRENCY_HAZARDS.md (the former GET-then-INCR gate). + * + * Needs a reachable Redis, like ConcurrentRateLimiterRaceTest. + */ +class ConcurrentConsumerQuotaScriptTest extends ConcurrentRaceSetup { + + private def redisUp: Boolean = Redis.isRedisReady + + private def freshKeys(n: Int): List[String] = { + val base = "__conc_quota_" + UUID.randomUUID.toString.take(8) + "_" + (1 to n).map(i => base + i).toList + } + + private def count(key: String): Long = Redis.use(JedisMethod.GET, key).map(_.toLong).getOrElse(0L) + private def ttl(key: String): Long = Redis.use(JedisMethod.TTL, key).map(_.toLong).getOrElse(-2L) + + private def call(keys: List[String], limits: List[Long], seconds: List[Int]): QuotaDecision = + Redis.quotaCheckAndCount(keys, limits, seconds) + + feature("Redis.quotaCheckAndCount checks every period and counts the call in one atomic script") { + + scenario("Q1: a call under every limit is counted in every period, unlimited periods included", ConcurrencyRace) { + RedisTestTarget.requireReachable(redisUp, "Q1") + Given("three periods: unlimited, 100 per 60s, 1000 per 3600s, counters absent") + val keys = freshKeys(3) + When("one call is checked and counted") + val decision = call(keys, List(-1L, 100L, 1000L), List(1, 60, 3600)) + Then("it is counted, and every key holds 1 with its period as TTL") + decision match { + case QuotaCounted(countsAndTtls) => + countsAndTtls.map(_._1) should equal(List(1L, 1L, 1L)) + countsAndTtls.map(_._2).zip(List(1, 60, 3600)).foreach { case (t, p) => t should (be > 0L and be <= p.toLong) } + case other => fail(s"expected QuotaCounted, got $other") + } + keys.map(count) should equal(List(1L, 1L, 1L)) + And("a second call makes them 2") + call(keys, List(-1L, 100L, 1000L), List(1, 60, 3600)) match { + case QuotaCounted(countsAndTtls) => countsAndTtls.map(_._1) should equal(List(2L, 2L, 2L)) + case other => fail(s"expected QuotaCounted, got $other") + } + } + + scenario("Q2: the shortest exhausted period refuses, with its count and TTL, and nothing is counted", ConcurrencyRace) { + RedisTestTarget.requireReachable(redisUp, "Q2") + Given("two periods limited to 2 per 60s and 2 per 3600s, each counter already at 2") + val keys = freshKeys(2) + call(keys, List(2L, 2L), List(60, 3600)) + call(keys, List(2L, 2L), List(60, 3600)) + When("a third call is checked") + val decision = call(keys, List(2L, 2L), List(60, 3600)) + Then("the first (shortest) period refuses, reporting a count of 2 and a live TTL") + decision match { + case QuotaRefused(index, c, t) => + index should equal(0) + c should equal(2L) + t should (be > 0L and be <= 60L) + case other => fail(s"expected QuotaRefused, got $other") + } + And("neither counter moved: a refused call costs nothing") + keys.map(count) should equal(List(2L, 2L)) + } + + scenario("Q3: a later period at its limit refuses even though earlier periods have room", ConcurrencyRace) { + RedisTestTarget.requireReachable(redisUp, "Q3") + Given("per-minute 100 and per-hour 1, with one call already counted") + val keys = freshKeys(2) + call(keys, List(100L, 1L), List(60, 3600)) + When("a second call is checked") + val decision = call(keys, List(100L, 1L), List(60, 3600)) + Then("the per-hour period (index 1) refuses and the per-minute counter stays at 1") + decision match { + case QuotaRefused(index, c, _) => + index should equal(1) + c should equal(1L) + case other => fail(s"expected QuotaRefused, got $other") + } + keys.map(count) should equal(List(1L, 1L)) + } + + scenario("Q4: a limit of 0 refuses before reading the value, and reports a TTL of -2 for an absent key", ConcurrencyRace) { + RedisTestTarget.requireReachable(redisUp, "Q4") + Given("an unlimited per-second period and a blocked per-minute period, no counters") + val keys = freshKeys(2) + When("a call is checked") + val decision = call(keys, List(-1L, 0L), List(1, 60)) + Then("the blocked period refuses with count 0 and the TTL Redis reports for a missing key") + decision should equal(QuotaRefused(1, 0L, -2L)) + And("the unlimited period was not counted either") + keys.map(count) should equal(List(0L, 0L)) + } + + scenario("Q5: a counter without an expiry is treated as expired on the check and recreated on the count", ConcurrencyRace) { + RedisTestTarget.requireReachable(redisUp, "Q5") + Given("a counter stuck at 50 with no TTL under a limit of 10") + val keys = freshKeys(1) + Redis.use(JedisMethod.SET, keys.head, value = Some("50")) + ttl(keys.head) should equal(-1L) + When("a call is checked and counted") + val decision = call(keys, List(10L), List(60)) + Then("it is served (the stuck key counts as 0) and the key is recreated at 1 with a TTL") + decision match { + case QuotaCounted((c, t) :: Nil) => + c should equal(1L) + t should (be > 0L and be <= 60L) + case other => fail(s"expected QuotaCounted, got $other") + } + count(keys.head) should equal(1L) + } + + scenario("Q6: at most `limit` of many concurrent callers pass the gate (hazard X closed)", ConcurrencyRace) { + RedisTestTarget.requireReachable(redisUp, "Q6") + Given("a per-minute limit of 5, an unlimited per-hour period, and 20 concurrent callers") + val keys = freshKeys(2) + val limit = 5L + val n = 20 + val passed = new AtomicInteger(0) + When(s"$n threads check-and-count through the script at the same time") + // Before this script the gate was GET (check) then INCR (count) as two round trips, so many + // callers read the same low count, all passed `count + 1 <= limit`, and all incremented. + val results = runConcurrentWithBarrier(n) { _ => + call(keys, List(limit, -1L), List(60, 3600)) match { + case QuotaCounted(_) => passed.incrementAndGet(); true + case QuotaRefused(_, _, _) => false + } + } + Then(s"exactly $limit callers were served and the counters hold exactly $limit") + results.count(_.isFailure) should equal(0) + withClue(s"passed=${passed.get()} limit=$limit: the check and the count run in one atomic script, so no interleaving can let more than `limit` through - ") { + passed.get().toLong should equal(limit) + } + keys.map(count) should equal(List(limit, limit)) + } + } +} diff --git a/obp-api/src/test/scala/code/concurrency/ConcurrentRateLimiterRaceTest.scala b/obp-api/src/test/scala/code/concurrency/ConcurrentRateLimiterRaceTest.scala index f6e58dcc3e..e19076f099 100644 --- a/obp-api/src/test/scala/code/concurrency/ConcurrentRateLimiterRaceTest.scala +++ b/obp-api/src/test/scala/code/concurrency/ConcurrentRateLimiterRaceTest.scala @@ -85,10 +85,11 @@ class ConcurrentRateLimiterRaceTest extends ConcurrentRaceSetup { val passedCount = passed.get() withClue( s"passedCount=$passedCount limit=$limit (results.size=${results.size}): " + - s"RateLimitingUtil checks the counter (GET) and increments it (INCR) as two separate Redis " + - s"round-trips. Under concurrency, many callers read the same low count, all pass `count+1 <= limit`, " + - s"then all increment — far more than `limit` requests slip through (rate-limit bypass). " + - s"Fix: a single atomic Lua `INCR + compare` so the gate and the increment cannot interleave — " + s"the gate and the increment must be one atomic Redis operation. RateLimitingUtil used to check " + + s"the counter (GET) and increment it (INCR) as two separate round-trips, so many callers read the " + + s"same low count, all passed `count+1 <= limit`, then all incremented (rate-limit bypass). " + + s"The consumer quota now runs check and count in one script (Redis.quotaCheckAndCount, see " + + s"ConcurrentConsumerQuotaScriptTest Q6); this primitive must stay atomic too — " ) { passedCount.toLong should be <= limit }