T09 · Insecure Skill Coding Practices
- Location
scripts/token_bucket.py:44- Finding
Indefinite Blocking on Token Requests Exceeding Rate-Limiter Capacity
- Content
View full analysis
Vulnerability Details
File Location:
scripts/token_bucket.py, lines 44–46
Vulnerability Type: Unbounded wait / denial of service
Risk Level: MediumVulnerable Code
python def wait_for_slot(self, in_tokens: int, out_tokens: int): while not self.can_send(in_tokens, out_tokens): time.sleep(0.5)Technical Analysis
wait_for_slot()repeatedly callscan_send()until the request can fit within the configured per-minute limits. Ifin_tokensexceedsself.in_limitorout_tokensexceedsself.out_limit, the request can never be accepted, regardless of how long the method waits.The method does not validate whether a request can ever fit within the configured capacity. It also provides no timeout, cancellation mechanism, retry limit, or exception path. Consequently, a permanently impossible request causes the calling thread to sleep and retry indefinitely.
The public methods and constructor also do not reject negative values or invalid limit configurations. Although the primary exploitable condition is an oversized request, strict input validation is needed to prevent malformed accounting behavior.
Attack Path
- An application integrates
TokenBucket.wait_for_slot()before making API requests. - Token-count arguments are supplied directly by a user or calculated from attacker-controlled content.
- An attacker submits content that results in
in_tokensorout_tokensexceeding the corresponding per-minute capacity. can_send()returnsFalseon every attempt because the request can never fit.wait_for_slot()retries indefinitely at 0.5-second intervals.- Repeating this operation across concurrent requests can occupy available worker threads and exhaust service capacity.
Impact Assessment
The issue does not grant additional permissions, code execution, data access, or privilege escalation. Its impact is limited to availability.
A single oversized request can permanently block the wor ...[truncated 346 chars]
- An application integrates
- Remediation
View remediation
Remediation Suggestions
- Validate constructor limits as positive integers.
- Validate
in_tokensandout_tokensas non-negative integers. - Reject any individual request that exceeds bucket capacity immediately rather than waiting for an impossible slot.
- Add an optional timeout or cancellation mechanism for requests that are temporarily unable to proceed.
- Use a monotonic clock for elapsed-time calculations so system clock adjustments cannot disrupt rate-limit behavior.
- Add tests covering oversized requests, invalid values, zero or negative limits, timeout behavior, and concurrent callers.
Example hardening approach:
python def wait_for_slot( self, in_tokens: int, out_tokens: int, timeout: float | None = None, ): if not isinstance(in_tokens, int) or not isinstance(out_tokens, int): raise TypeError("Token counts must be integers") if in_tokens < 0 or out_tokens < 0: raise ValueError("Token counts must be non-negative") if in_tokens > self.in_limit or out_tokens > self.out_limit: raise ValueError("Request exceeds token bucket capacity") deadline = None if timeout is None else time.monotonic() + timeout while not self.can_send(in_tokens, out_tokens): if deadline is not None and time.monotonic() >= deadline: raise TimeoutError("Timed out waiting for rate-limit capacity") time.sleep(0.5)
