Api Rate Abuse in Phoenix with Bearer Tokens

Rate abuse in Phoenix applications that rely on Bearer Tokens occurs when an attacker can issue a high volume of authenticated requests using a single token, overwhelming the API or exhausting backend resources. Because Bearer Tokens are typically validated on each request rather than coupled with strong per-identity or per-client rate limits, an attacker who obtains or guesses a token can perform credential-based flooding, scraping, or brute-force actions at the scope allowed by the token’s permissions.

In a Phoenix/Elixir stack, this risk is amplified when rate limiting is implemented at a global or IP level without considering token identity. For example, if a Phoenix controller or pipeline applies Plug-based rate limits using PlugAttack or custom plugs keyed only by IP address, a single valid Bearer Token shared or leaked can be used to bypass those protections. Attack patterns include token replay from logs or client-side storage, token leakage via referrer headers, or token sharing across compromised clients, enabling a distributed attacker to pivot into a high-rate path once token access is obtained.

Consider an endpoint like /api/v1/reports/:id that requires a Bearer Token and returns large datasets. Without per-token or per-user rate limiting, an authenticated client can hammer the endpoint with paginated requests, leading to excessive database load, connection pool exhaustion, or denial of service for legitimate users. The vulnerability is not the token format itself, but the lack of identity-aware throttling and monitoring in the request pipeline. Because Phoenix pipelines often chain plugs for authentication and authorization, gaps at any stage—such as missing rate checks after token verification—create exploitable windows.

OpenAPI/Swagger analysis can surface missing rate limit definitions for authenticated paths. If the spec defines Bearer security schemes but omits operation-level rate-limit parameters or does not differentiate limits by scope, runtime scans may flag the endpoint as high risk for rate abuse. During black-box testing, a scanner can probe authenticated routes with repeated calls using the same token to observe whether throttling triggers, whether responses leak information under load, and whether token reuse is detectable through inconsistent enforcement.

Effective detection aligns with the 12 parallel security checks run by tools like middleBrick, particularly Rate Limiting, Authentication, and Property Authorization. These checks validate that authenticated endpoints enforce appropriate request ceilings per identity or token scope and that failures under load do not expose stack traces or sensitive data. Continuous monitoring can surface gradual increases in request volume per token, helping teams identify abuse patterns before they escalate into service-impacting incidents.

Remediation focuses on binding rate limits to the token identity or token scope rather than only IP or global thresholds. In Phoenix, this can be achieved by extracting the token subject or associated user ID during authentication and using it as a key for rate-limiting plugs. Below are concrete, syntactically correct examples that demonstrate how to implement token-aware rate limiting and how to structure requests that include Bearer Tokens correctly.

1. Token-aware rate limiting with hammer and Phoenix pipelines

Use the hammer library to scope rate limits to a token or user ID. In your authentication plug, resolve a rate-limit key from the token and pass it to Hammer checks before allowing the request to proceed.

defmodule MyAppWeb.Plugs.AuthenticateAndRateLimit do
  import Plug.Conn
  alias MyApp.Accounts

  def init(opts), do: opts

  def call(conn, _opts) do
    with [token] <- get_req_header(conn, "authorization") do
      case authenticate_token(token) do
        {:ok, %{subject: subject}} ->
          # subject can be user_id, tenant_id, or a token-specific identifier
          limit_key = {"token_rate", subject}
          case Hammer.check_rate(limit_key, 60_000, 100) do
            {:allow, _} -> conn
            {:deny, _} -> conn |> put_resp_content_type("application/json") |> send_resp(429, ~s({"error":"rate_limit_exceeded"}))
          end
        {:error, _} -> conn |> put_resp_content_type("application/json") |> send_resp(401, ~s({"error":"unauthorized"}))
      end
    else
      _ -> conn |> put_resp_content_type("application/json") |> send_resp(400, ~s({"error":"missing_authorization"}))
    end
  end

  defp authenticate_token(token) do
    # Validate Bearer token via your auth provider, return {:ok, %{subject: "user_id"}} or {:error, :invalid}
    # This is a placeholder for actual token validation logic
    if token == "valid_bearer_token_example" do
      {:ok, %{subject: "user_123"}}
    else
      {:error, :invalid}
    end
  end
end

2. Example HTTP request with Bearer Token

Clients must include the token in the Authorization header. Ensure TLS is enforced in production to prevent token interception.

curl -X GET "https://api.example.com/api/v1/reports/42" \
  -H "Authorization: Bearer valid_bearer_token_example" \
  -H "Accept: application/json"

3. Rejecting requests with malformed or missing tokens

Ensure your pipeline fails early when authorization headers are malformed or tokens are missing, to avoid ambiguous behavior and information leakage under load.

defmodule MyAppWeb.Plugs.EnsureBearerFormat do
  import Plug.Conn

  def init(opts), do: opts

  def call(conn, _opts) do
    case get_req_header(conn, "authorization") do
      ["Bearer " <> token] when byte_size(token) > 0 -> assign(conn, :token, token)
      _ -> conn |> put_resp_content_type("application/json") |> send_resp(400, ~s({"error":"malformed_authorization_header"}))
    end
  end
end

4. Scoping limits by operation and token claims

For endpoints with different sensitivity, derive rate-limit keys from token claims such as scopes or roles, and apply stricter ceilings for high-risk operations.

defmodule MyAppWeb.Plugs.ScopedRateLimit do
  import Plug.Conn

  def init(opts), do: opts

  def call(conn, _opts) do
    token_claims = conn.assigns.token_claims || %{}
    limit = Map.get(token_claims, "rate_limit", 100)
    scope_key = "scope_rate_#{token_claims["scope"]}"
    Hammer.check_rate({scope_key, conn.remote_ip}, 60_000, limit)
    # proceed or deny accordingly
  end
end

5. Monitoring and observability

Log rate-limit decisions and token usage patterns to detect abuse or misconfigured clients. Combine with Phoenix telemetry to correlate high request volumes with specific token identities or endpoints.