feat(okta): add configurable API request throttling and rate-limit retries (#11702)

docs/docs.json

@@ -359,7 +359,8 @@
                 "group": "Okta",
                 "pages": [
                   "user-guide/providers/okta/getting-started-okta",
-                  "user-guide/providers/okta/authentication"
+                  "user-guide/providers/okta/authentication",
+                  "user-guide/providers/okta/retry-configuration"
                 ]
               },
               {

docs/user-guide/providers/okta/retry-configuration.mdx

@@ -0,0 +1,123 @@
+---
+title: "Okta Rate Limit Configuration in Prowler"
+---
+
+import { VersionBadge } from "/snippets/version-badge.mdx"
+
+<VersionBadge version="5.32.0" />
+
+Prowler's Okta Provider manages API rate limits with two complementary controls:
+
+- **Request throttling (proactive):** Prowler paces outbound requests through a shared limiter so scans stay under Okta's rate limits and rarely trigger a rate-limit response in the first place.
+- **Retries (reactive):** When Okta still returns a rate-limit response (HTTP 429), the official Okta Python SDK reads the `X-Rate-Limit-Reset` header and waits until the window resets before retrying. This acts as a safety net for occasional bursts.
+
+Both controls are configurable through the configuration file or command line flags.
+
+## Request Throttling (Requests per Second)
+
+Throttling is the primary control for avoiding rate limits. Prowler limits the aggregate number of Okta API requests per second across every service in a scan.
+
+### Using the Command Line Flag
+
+```bash
+prowler okta --okta-requests-per-second 4
+```
+
+Set the value to `0` to disable throttling.
+
+### Using the Configuration File
+
+```yaml
+okta:
+  # Maximum aggregate Okta API requests per second. Default: 4. Set to 0 to disable.
+  okta_requests_per_second: 4
+```
+
+Okta enforces rate limits per endpoint, so this single global cap is a deliberately simple control. Lower the value if scans still hit limits on large organizations; raise it to scan faster when the organization has generous limits.
+
+## Retries
+
+Retries cover the cases throttling does not prevent, such as short bursts or per-endpoint limits lower than the global cap.
+
+### Using the Command Line Flag
+
+```bash
+prowler okta --okta-retries-max-attempts 8
+```
+
+### Using the Configuration File
+
+```yaml
+okta:
+  # Maximum retries on HTTP 429. Default: 5.
+  okta_max_retries: 8
+  # Per-request timeout in seconds. Default: 300.
+  okta_request_timeout: 300
+```
+
+The command line flags override the configuration file values.
+
+## How It Works
+
+- **Automatic detection:** The Okta SDK retries the retryable statuses 429, 503, and 504.
+- **Reset-aware backoff:** On a 429 response the SDK sleeps until the `X-Rate-Limit-Reset` window before each retry, rather than using a fixed delay.
+- **Bounded attempts:** `okta_max_retries` caps how many times a single request is retried. The Okta SDK default is 2, which is often too low for large organizations, so Prowler defaults to 5.
+
+## Request Timeout
+
+The `okta_request_timeout` setting plays a dual role in the Okta SDK:
+
+- It is the per-request socket timeout, bounding how long a single HTTP call can hang.
+- It is also the total wall-clock budget for the whole retry-and-backoff loop of one request.
+
+For this reason, the value defaults to 300 seconds rather than 0 (no timeout). A value of 0 leaves hung connections unbounded, while a value that is too low cuts the rate-limit waits short and reintroduces the errors. As a guideline, keep `okta_request_timeout` greater than or equal to `okta_max_retries` multiplied by 60 when raising the retry count, because Okta reset windows are typically up to one minute.
+
+## Error Example Handled
+
+```
+Okta HTTP 429: Too Many Requests. Hit rate limit. Retry request in 42 seconds.
+```
+
+## Validation
+
+### Debug Logging
+
+To confirm that throttling and retries are active, run a scan with debug logging:
+
+```bash
+prowler okta --okta-requests-per-second 4 --log-level DEBUG --log-file debuglogs.txt
+```
+
+### Check the Messages
+
+```bash
+grep -i "throttling\|rate limit\|retry" debuglogs.txt
+```
+
+### Expected Output
+
+When throttling is enabled, Prowler logs the configured rate at startup:
+
+```
+Okta request throttling enabled at 4 req/s
+```
+
+If a rate limit is still hit, the SDK logs the backoff:
+
+```
+Hit rate limit. Retry request in 42 seconds.
+```
+
+## Troubleshooting
+
+If scans continue to hit rate limits:
+
+1. Lower `--okta-requests-per-second` so requests are paced more conservatively.
+2. Raise `--okta-retries-max-attempts` (and keep `okta_request_timeout` proportionally large) so the safety net absorbs more bursts.
+3. Review the rate-limit allocation for the Okta organization and request an increase if needed.
+4. Verify throttling and retry behavior with debug logging.
+
+## Official References
+
+- [Okta Rate Limits](https://developer.okta.com/docs/reference/rate-limits/)
+- [Okta SDK for Python](https://github.com/okta/okta-sdk-python)

prowler/CHANGELOG.md

@@ -15,6 +15,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
 - `waf_regional_webacl_logging_enabled` check for AWS provider, verifying that each AWS WAF Classic Regional Web ACL has logging enabled to a Kinesis Data Firehose stream [(#11539)](https://github.com/prowler-cloud/prowler/pull/11539)
 - `sdk_only` provider property (default `true`) and `Provider.get_app_providers()`, so a provider (built-in or external) stays CLI/SDK-only and hidden from the app unless it declares `sdk_only = False` [(#11427)](https://github.com/prowler-cloud/prowler/pull/11427)
 - `Provider.get_scan_arguments()`, `Provider.get_connection_arguments()` and `Provider.get_credentials_schema()` contract methods, so a provider persisted as a stored uid plus a secret dict can be constructed and validated programmatically (to be consumed by the API in a later change) [(#11578)](https://github.com/prowler-cloud/prowler/pull/11578)
+- Okta API request throttling to proactively stay under rate limits, configurable via `okta_requests_per_second` in the config file and the `--okta-requests-per-second` CLI flag, plus configurable retries via `okta_max_retries` / `--okta-retries-max-attempts` as a safety net [(#11702)](https://github.com/prowler-cloud/prowler/pull/11702)
 - CIS Amazon Web Services Foundations Benchmark v7.0.0 compliance framework for the AWS provider, adding the new Organizations section (2.1.1-2.1.6), resource policy (2.21), web front-end access logging (4.10), and VPC Endpoints (6.8) recommendations [(#11707)](https://github.com/prowler-cloud/prowler/pull/11707)
 - CIS Microsoft Azure Foundations Benchmark v6.0.0 compliance framework for the Azure provider [(#11708)](https://github.com/prowler-cloud/prowler/pull/11708)
 - CIS Google Cloud Platform Foundation Benchmark v5.0.0 compliance framework for the GCP provider [(#11714)](https://github.com/prowler-cloud/prowler/pull/11714)

prowler/config/config.yaml

@@ -731,6 +731,27 @@ okta:
   # Defaults to 15 per DISA STIG V-273187 (OKTA-APP-000025); raise it only
   # with an explicit risk acceptance.
   okta_admin_console_idle_timeout_max_minutes: 15
+  # Okta API rate limiting
+  # Max retries on HTTP 429. The Okta SDK sleeps until the X-Rate-Limit-Reset
+  # window before each retry, so raising this lets scans ride out more rate-limit
+  # windows on busy orgs instead of failing with partial data. SDK default is 2.
+  okta_max_retries: 5
+  # Per-request timeout in seconds. In the Okta SDK this value plays a DUAL role:
+  # it is both the per-HTTP-call socket timeout AND the total wall-clock budget
+  # across the whole retry+backoff loop. It defaults to 300 (not 0) because it is
+  # the only effective hang guard, and 300s is the smallest value that still lets
+  # all okta_max_retries rate-limit waits (~60s Okta reset windows) complete
+  # without being cut short. Keep it roughly >= okta_max_retries * 60 if you
+  # raise okta_max_retries.
+  okta_request_timeout: 300
+  # Maximum aggregate Okta API requests per second. Prowler paces all requests
+  # through a shared limiter so scans stay under Okta's rate limits proactively,
+  # rather than relying on the 429 retry above as a safety net. Okta enforces
+  # limits per endpoint, so this is a deliberately simple global cap; lower it if
+  # scans still hit limits, raise it to scan faster. Set to 0 to disable. Valid
+  # range: 0 or 0.1..100 — non-zero rates below 0.1 are rejected because they
+  # would make a scan impractically slow.
+  okta_requests_per_second: 4
   # Okta Users
   # okta.user_inactivity_automation_35d_enabled
   # Maximum number of days a user can stay inactive before the

prowler/config/schema/okta.py

@@ -1,19 +1,45 @@
 """Okta provider config schema with safety bounds."""
 
-from typing import Optional
+from typing import Annotated, Optional
 
-from pydantic import Field
+from pydantic import AfterValidator, Field
 
 from prowler.config.schema.base import ProviderConfigBase
 
+# Lowest non-zero request rate we accept. Below this a scan is paced so slowly
+# it becomes impractical (e.g. 0.001 req/s is ~1000s per request, turning a
+# routine scan into days or years). 0 stays valid as the "disable throttling"
+# sentinel; anything between 0 and this floor is rejected so a typo can never
+# stall a scan.
+MIN_REQUESTS_PER_SECOND = 0.1
+
+
+def _validate_requests_per_second(value: Optional[float]) -> Optional[float]:
+    """Reject impractically slow non-zero request rates.
+
+    ``0`` (and ``None``) pass through unchanged — ``0`` is the documented
+    "disable throttling" sentinel. Any positive value below
+    ``MIN_REQUESTS_PER_SECOND`` is rejected; the ``ge``/``le`` bounds on the
+    field already handle negatives and the upper cap.
+    """
+    if value is None or value == 0:
+        return value
+    if value < MIN_REQUESTS_PER_SECOND:
+        raise ValueError(
+            f"must be 0 (disable throttling) or >= {MIN_REQUESTS_PER_SECOND}; "
+            "smaller rates make scans impractically slow"
+        )
+    return value
+
 
 class OktaProviderConfig(ProviderConfigBase):
     """Okta provider configuration schema.
 
     Bounds the session, idle-timeout and inactivity thresholds consumed by
-    the Okta checks. Every field is optional: when omitted (or dropped for
-    being out of range) the check falls back to its own DISA STIG-derived
-    default via ``audit_config.get(key, default)``.
+    the Okta checks, plus the provider's API rate-limit handling (proactive
+    request throttling and the SDK retry safety net). Every field is optional:
+    when omitted (or dropped for being out of range) the check falls back to
+    its own DISA STIG-derived default via ``audit_config.get(key, default)``.
     """
 
     okta_max_session_idle_minutes: Optional[int] = Field(
@@ -63,3 +89,36 @@ class OktaProviderConfig(ProviderConfigBase):
             "the built-in `OU=DoD` / `OU=ECA` patterns."
         ),
     )
+
+    # API rate limiting
+    okta_requests_per_second: Annotated[
+        Optional[float], AfterValidator(_validate_requests_per_second)
+    ] = Field(
+        default=None,
+        ge=0,
+        le=100,
+        description=(
+            "Maximum aggregate Okta API requests per second. Range: 0 or "
+            f"{MIN_REQUESTS_PER_SECOND}..100 (0 disables throttling). Non-zero "
+            f"values below {MIN_REQUESTS_PER_SECOND} are rejected to avoid "
+            "impractically slow scans."
+        ),
+    )
+    okta_max_retries: Optional[int] = Field(
+        default=None,
+        ge=0,
+        le=10,
+        description=(
+            "Max retries on Okta API rate limiting (HTTP 429). Range: 0..10 "
+            "(0 disables retries)."
+        ),
+    )
+    okta_request_timeout: Optional[int] = Field(
+        default=None,
+        ge=0,
+        le=3600,
+        description=(
+            "Per-request timeout in seconds; also the total budget for the SDK "
+            "retry loop. Range: 0..3600 (0 disables the timeout)."
+        ),
+    )

prowler/providers/common/provider.py

@@ -633,6 +633,12 @@ class Provider(ABC):
                             arguments, "okta_private_key_file", ""
                         ),
                         okta_scopes=getattr(arguments, "okta_scopes", None),
+                        okta_retries_max_attempts=getattr(
+                            arguments, "okta_retries_max_attempts", None
+                        ),
+                        okta_requests_per_second=getattr(
+                            arguments, "okta_requests_per_second", None
+                        ),
                         config_path=arguments.config_file,
                         mutelist_path=arguments.mutelist_file,
                         fixer_config=fixer_config,

prowler/providers/okta/lib/arguments/arguments.py

@@ -42,3 +42,25 @@ def init_parser(self):
         default=None,
         metavar="OKTA_SCOPES",
     )
+    okta_rate_limit_subparser = okta_parser.add_argument_group("Rate limiting")
+    okta_rate_limit_subparser.add_argument(
+        "--okta-retries-max-attempts",
+        type=int,
+        default=None,
+        help=(
+            "Maximum number of retries on Okta API rate limiting (HTTP 429). "
+            "Overrides the config.yaml value (okta_max_retries). Default: 5."
+        ),
+        metavar="OKTA_RETRIES_MAX_ATTEMPTS",
+    )
+    okta_rate_limit_subparser.add_argument(
+        "--okta-requests-per-second",
+        type=float,
+        default=None,
+        help=(
+            "Maximum aggregate Okta API requests per second. Throttles requests "
+            "to stay under Okta's rate limits. Overrides the config.yaml value "
+            "(okta_requests_per_second); set to 0 to disable. Default: 4."
+        ),
+        metavar="OKTA_REQUESTS_PER_SECOND",
+    )

prowler/providers/okta/lib/service/rate_limiter.py

@@ -0,0 +1,105 @@
+"""Client-side request throttling for the Okta provider.
+
+The Okta SDK already retries on HTTP 429 (see `service.py`), but retrying is
+reactive: it only helps *after* a rate limit has been hit, and each backoff
+waits out a full reset window. To avoid hitting Okta's limits in the first
+place, this module paces outbound requests with a shared token bucket.
+
+A single `OktaRateLimiter` instance lives on the provider and is shared by every
+service's SDK client, so the cap applies to the *aggregate* request rate rather
+than per client. The limiter is injected by wrapping the SDK's `HTTPClient`
+(via the `httpClient` config key) and awaiting `acquire()` before each call.
+
+Note: Okta enforces rate limits per endpoint, so a single requests-per-second
+cap is a deliberately simple, blunt control. It keeps bursty pagination from
+overrunning the limits without trying to model every per-endpoint budget.
+"""
+
+import asyncio
+import threading
+import time
+
+from okta.http_client import HTTPClient
+
+# Default aggregate request rate. Okta-managed orgs commonly throttle the
+# busiest endpoints around a handful of requests per second, so we pace below
+# that by default. Set `okta_requests_per_second` to 0 (or a negative value) to
+# disable throttling entirely.
+DEFAULT_REQUESTS_PER_SECOND = 4
+
+
+class OktaRateLimiter:
+    """Token-bucket limiter shared across a provider's Okta SDK clients.
+
+    The bucket refills at `requests_per_second` tokens per second up to a small
+    burst capacity. `acquire()` consumes one token, sleeping just long enough
+    when the bucket is empty. Token accounting is wall-clock based
+    (`time.monotonic`) so it stays correct across the separate event loops the
+    services spin up with `asyncio.run`.
+    """
+
+    def __init__(
+        self,
+        requests_per_second: float,
+        *,
+        clock=time.monotonic,
+        sleep=asyncio.sleep,
+    ):
+        if requests_per_second <= 0:
+            raise ValueError("requests_per_second must be greater than 0")
+        self._rate = float(requests_per_second)
+        # Allow up to one second of requests to burst, then settle to the rate.
+        self._capacity = max(1.0, self._rate)
+        self._tokens = self._capacity
+        self._clock = clock
+        self._sleep = sleep
+        self._last = clock()
+        # Guards the token math only; never held across an await.
+        self._lock = threading.Lock()
+
+    async def acquire(self) -> None:
+        """Block until a request token is available, then consume it."""
+        while True:
+            with self._lock:
+                now = self._clock()
+                self._tokens = min(
+                    self._capacity, self._tokens + (now - self._last) * self._rate
+                )
+                self._last = now
+                if self._tokens >= 1:
+                    self._tokens -= 1
+                    return
+                wait = (1 - self._tokens) / self._rate
+            await self._sleep(wait)
+
+
+def build_throttled_http_client(limiter: OktaRateLimiter) -> type[HTTPClient]:
+    """Return an `HTTPClient` subclass that paces requests through `limiter`.
+
+    The Okta SDK instantiates `config["httpClient"]` with its HTTP config, so we
+    return a class (not an instance) that closes over the shared limiter.
+
+    Args:
+        limiter: Shared token-bucket limiter that paces the aggregate request
+            rate across every service client of the provider.
+
+    Returns:
+        An `HTTPClient` subclass that awaits the limiter before each request.
+    """
+
+    class ThrottledHTTPClient(HTTPClient):
+        """`HTTPClient` that acquires a limiter token before each request."""
+
+        async def send_request(self, request):
+            """Acquire a rate-limit token, then delegate to the SDK client.
+
+            Args:
+                request: The request payload built by the Okta SDK.
+
+            Returns:
+                The result of the underlying `HTTPClient.send_request` call.
+            """
+            await limiter.acquire()
+            return await super().send_request(request)
+
+    return ThrottledHTTPClient

prowler/providers/okta/lib/service/service.py

@@ -3,11 +3,21 @@ from typing import TYPE_CHECKING
 
 from okta.client import Client as OktaSDKClient
 
+from prowler.providers.okta.lib.service.rate_limiter import build_throttled_http_client
 from prowler.providers.okta.models import OktaSession
 
 if TYPE_CHECKING:
     from prowler.providers.okta.okta_provider import OktaProvider
 
+# Okta API rate-limit handling. The okta-sdk-python `Client` already backs off
+# on HTTP 429 by sleeping until the `X-Rate-Limit-Reset` window before retrying,
+# but it only does so `maxRetries` times (SDK default 2). On busy orgs that is
+# too few and requests fail with partial data, so we raise it. See config.yaml
+# (`okta_max_retries` / `okta_request_timeout`) for the user-facing knobs and the
+# rationale behind the 300s timeout default.
+DEFAULT_MAX_RETRIES = 5
+DEFAULT_REQUEST_TIMEOUT = 300
+
 
 class OktaService:
     """Base class for Okta service implementations.
@@ -20,13 +30,34 @@ class OktaService:
     def __init__(self, service: str, provider: "OktaProvider"):
         self.provider = provider
         self.service = service
-        self.client = self.__set_client__(provider.session)
         self.audit_config = provider.audit_config
         self.fixer_config = provider.fixer_config
+        self.client = self.__set_client__(
+            provider.session, self.audit_config, provider.rate_limiter
+        )
 
     @staticmethod
-    def __set_client__(session: OktaSession) -> OktaSDKClient:
-        return OktaSDKClient(session.to_sdk_config())
+    def __set_client__(
+        session: OktaSession, audit_config: dict, rate_limiter=None
+    ) -> OktaSDKClient:
+        # Start from the shared SDK config and layer the rate-limit settings on
+        # top. `Client(config)` deep-merges these flat keys onto its defaults, so
+        # `rateLimit`/`requestTimeout` override the SDK's built-in values.
+        config = session.to_sdk_config()
+        audit_config = audit_config or {}
+        config["rateLimit"] = {
+            "maxRetries": audit_config.get("okta_max_retries", DEFAULT_MAX_RETRIES)
+        }
+        config["requestTimeout"] = audit_config.get(
+            "okta_request_timeout", DEFAULT_REQUEST_TIMEOUT
+        )
+        # Proactively pace outbound requests so scans stay under Okta's limits
+        # instead of relying on the 429 retry as a safety net. The limiter is
+        # shared across every service client of the provider, so the cap applies
+        # to the aggregate request rate.
+        if rate_limiter is not None:
+            config["httpClient"] = build_throttled_http_client(rate_limiter)
+        return OktaSDKClient(config)
 
     @staticmethod
     def _run(coro):

prowler/providers/okta/okta_provider.py

@@ -30,6 +30,10 @@ from prowler.providers.okta.exceptions.exceptions import (
     OktaSetUpSessionError,
 )
 from prowler.providers.okta.lib.mutelist.mutelist import OktaMutelist
+from prowler.providers.okta.lib.service.rate_limiter import (
+    DEFAULT_REQUESTS_PER_SECOND,
+    OktaRateLimiter,
+)
 from prowler.providers.okta.models import OktaIdentityInfo, OktaSession
 
 DEFAULT_SCOPES = [
@@ -85,6 +89,7 @@ class OktaProvider(Provider):
     _audit_config: dict
     _fixer_config: dict
     _mutelist: Mutelist
+    _rate_limiter: Optional[OktaRateLimiter]
     audit_metadata: Audit_Metadata
 
     def __init__(
@@ -94,6 +99,8 @@ class OktaProvider(Provider):
         okta_private_key: str = "",
         okta_private_key_file: str = "",
         okta_scopes: Optional[Union[str, list[str]]] = None,
+        okta_retries_max_attempts: int = None,
+        okta_requests_per_second: float = None,
         config_path: str = None,
         config_content: dict = None,
         fixer_config: dict = {},
@@ -125,6 +132,30 @@ class OktaProvider(Provider):
             if not config_path:
                 config_path = default_config_file_path
             self._audit_config = load_and_validate_config_file(self._type, config_path)
+
+        # CLI flags take precedence over config.yaml for the Okta API rate-limit
+        # settings. Services read these from audit_config when building the SDK
+        # client, so override the loaded values here.
+        if okta_retries_max_attempts is not None:
+            self._audit_config["okta_max_retries"] = okta_retries_max_attempts
+            logger.info(f"Okta max retries set to {okta_retries_max_attempts}")
+        if okta_requests_per_second is not None:
+            self._audit_config["okta_requests_per_second"] = okta_requests_per_second
+
+        # Build the shared request limiter once, here, so every service client
+        # paces against the same token bucket. A value of 0 (or below) disables
+        # throttling.
+        requests_per_second = self._audit_config.get(
+            "okta_requests_per_second", DEFAULT_REQUESTS_PER_SECOND
+        )
+        if requests_per_second and requests_per_second > 0:
+            self._rate_limiter = OktaRateLimiter(requests_per_second)
+            logger.info(
+                f"Okta request throttling enabled at {requests_per_second} req/s"
+            )
+        else:
+            self._rate_limiter = None
+
         self._fixer_config = fixer_config
 
         if mutelist_content:
@@ -164,6 +195,10 @@ class OktaProvider(Provider):
     def mutelist(self) -> OktaMutelist:
         return self._mutelist
 
+    @property
+    def rate_limiter(self) -> Optional[OktaRateLimiter]:
+        return self._rate_limiter
+
     @staticmethod
     def validate_arguments(
         okta_org_domain: str = "",

tests/config/schema/other_providers_schema_test.py

@@ -117,6 +117,88 @@ class Test_Cloudflare_Schema:
         assert _validate("cloudflare", {"max_retries": -1}) == {}
 
 
+class Test_Okta_Schema:
+    def test_valid_values_round_trip(self):
+        raw = {
+            "okta_max_session_idle_minutes": 15,
+            "okta_max_session_lifetime_minutes": 18 * 60,
+            "okta_admin_console_idle_timeout_max_minutes": 15,
+            "okta_user_inactivity_max_days": 35,
+            "okta_dod_approved_ca_issuer_patterns": [r"\bOU=DoD\b", r"\bOU=ECA\b"],
+        }
+        assert _validate("okta", raw) == raw
+
+    def test_zero_idle_minutes_dropped(self):
+        assert _validate("okta", {"okta_max_session_idle_minutes": 0}) == {}
+
+    def test_negative_inactivity_days_dropped(self):
+        assert _validate("okta", {"okta_user_inactivity_max_days": -1}) == {}
+
+    def test_full_rate_limit_config_round_trip(self):
+        raw = {
+            "okta_requests_per_second": 4.0,
+            "okta_max_retries": 5,
+            "okta_request_timeout": 300,
+        }
+        assert _validate("okta", raw) == raw
+
+    def test_requests_per_second_zero_allowed(self):
+        # 0 is documented as "disable throttling" in config.yaml.
+        assert _validate("okta", {"okta_requests_per_second": 0}) == {
+            "okta_requests_per_second": 0
+        }
+
+    def test_requests_per_second_sub_one_allowed(self):
+        assert _validate("okta", {"okta_requests_per_second": 0.5}) == {
+            "okta_requests_per_second": 0.5
+        }
+
+    def test_requests_per_second_floor_allowed(self):
+        # 0.1 is the lowest non-zero rate accepted.
+        assert _validate("okta", {"okta_requests_per_second": 0.1}) == {
+            "okta_requests_per_second": 0.1
+        }
+
+    def test_requests_per_second_below_floor_dropped(self):
+        # A tiny rate (e.g. 0.001 -> ~1000s/request) would make scans take
+        # days or years; it must be rejected, not silently honoured.
+        assert _validate("okta", {"okta_requests_per_second": 0.001}) == {}
+        assert _validate("okta", {"okta_requests_per_second": 0.05}) == {}
+
+    def test_requests_per_second_above_max_dropped(self):
+        assert _validate("okta", {"okta_requests_per_second": 101}) == {}
+
+    def test_requests_per_second_negative_dropped(self):
+        assert _validate("okta", {"okta_requests_per_second": -1}) == {}
+
+    def test_max_retries_zero_allowed(self):
+        assert _validate("okta", {"okta_max_retries": 0}) == {"okta_max_retries": 0}
+
+    def test_max_retries_out_of_range_dropped(self):
+        assert _validate("okta", {"okta_max_retries": -1}) == {}
+        assert _validate("okta", {"okta_max_retries": 11}) == {}
+
+    def test_request_timeout_zero_allowed(self):
+        # 0 is documented as "disable the timeout" in config.yaml.
+        assert _validate("okta", {"okta_request_timeout": 0}) == {
+            "okta_request_timeout": 0
+        }
+
+    def test_request_timeout_in_range_allowed(self):
+        assert _validate("okta", {"okta_request_timeout": 300}) == {
+            "okta_request_timeout": 300
+        }
+
+    def test_request_timeout_out_of_range_dropped(self):
+        assert _validate("okta", {"okta_request_timeout": -1}) == {}
+        assert _validate("okta", {"okta_request_timeout": 3601}) == {}
+
+    def test_non_numeric_value_dropped(self):
+        # A typo'd string must not flow through to the limiter (it would crash
+        # the `> 0` comparison during provider init).
+        assert _validate("okta", {"okta_requests_per_second": "fast"}) == {}
+
+
 class Test_Vercel_Schema:
     def test_owner_percentage_in_range(self):
         assert _validate("vercel", {"max_owner_percentage": 20}) == {
@@ -152,24 +234,6 @@ class Test_Vercel_Schema:
         assert _validate("vercel", raw) == raw
 
 
-class Test_Okta_Schema:
-    def test_valid_values_round_trip(self):
-        raw = {
-            "okta_max_session_idle_minutes": 15,
-            "okta_max_session_lifetime_minutes": 18 * 60,
-            "okta_admin_console_idle_timeout_max_minutes": 15,
-            "okta_user_inactivity_max_days": 35,
-            "okta_dod_approved_ca_issuer_patterns": [r"\bOU=DoD\b", r"\bOU=ECA\b"],
-        }
-        assert _validate("okta", raw) == raw
-
-    def test_zero_idle_minutes_dropped(self):
-        assert _validate("okta", {"okta_max_session_idle_minutes": 0}) == {}
-
-    def test_negative_inactivity_days_dropped(self):
-        assert _validate("okta", {"okta_user_inactivity_max_days": -1}) == {}
-
-
 class Test_AlibabaCloud_Schema:
     def test_valid_values_round_trip(self):
         raw = {

tests/providers/okta/lib/arguments/okta_arguments_test.py

@@ -40,6 +40,8 @@ class TestOktaArguments:
             "--okta-org-domain",
             "--okta-client-id",
             "--okta-scopes",
+            "--okta-retries-max-attempts",
+            "--okta-requests-per-second",
         }
 
     def test_secret_flags_not_registered(self):

tests/providers/okta/lib/service/okta_service_test.py

@@ -0,0 +1,79 @@
+from unittest import mock
+
+from okta.http_client import HTTPClient
+
+from prowler.providers.okta.lib.service.rate_limiter import OktaRateLimiter
+from prowler.providers.okta.lib.service.service import (
+    DEFAULT_MAX_RETRIES,
+    DEFAULT_REQUEST_TIMEOUT,
+    OktaService,
+)
+from tests.providers.okta.okta_fixtures import set_mocked_okta_provider
+
+
+def _build_service(audit_config: dict = None, rate_limiter=None):
+    """Instantiate OktaService with the SDK client patched, returning the
+    config dict that was handed to ``OktaSDKClient``."""
+    provider = set_mocked_okta_provider(
+        audit_config=audit_config, rate_limiter=rate_limiter
+    )
+    with mock.patch(
+        "prowler.providers.okta.lib.service.service.OktaSDKClient"
+    ) as mocked_client_cls:
+        OktaService("test", provider)
+    return mocked_client_cls.call_args.args[0]
+
+
+class Test_OktaService_set_client:
+    def test_defaults_applied_when_audit_config_empty(self):
+        config = _build_service(audit_config={})
+
+        assert config["rateLimit"] == {"maxRetries": DEFAULT_MAX_RETRIES}
+        assert config["requestTimeout"] == DEFAULT_REQUEST_TIMEOUT
+
+    def test_defaults_applied_when_audit_config_none(self):
+        # set_mocked_okta_provider coerces None to {}, but the helper also
+        # guards against a None audit_config defensively.
+        config = _build_service(audit_config=None)
+
+        assert config["rateLimit"] == {"maxRetries": DEFAULT_MAX_RETRIES}
+        assert config["requestTimeout"] == DEFAULT_REQUEST_TIMEOUT
+
+    def test_audit_config_values_override_defaults(self):
+        config = _build_service(
+            audit_config={"okta_max_retries": 9, "okta_request_timeout": 120}
+        )
+
+        assert config["rateLimit"] == {"maxRetries": 9}
+        assert config["requestTimeout"] == 120
+
+    def test_retries_disabled_with_zero(self):
+        config = _build_service(audit_config={"okta_max_retries": 0})
+
+        assert config["rateLimit"] == {"maxRetries": 0}
+
+    def test_preserves_session_sdk_config_keys(self):
+        config = _build_service(audit_config={})
+
+        # The rate-limit settings are layered on top of the shared session
+        # config, so the credential keys must remain intact.
+        assert config["orgUrl"] == "https://acme.okta.com"
+        assert config["authorizationMode"] == "PrivateKey"
+        assert config["clientId"]
+        assert config["privateKey"]
+        assert config["dpopEnabled"] is True
+
+    def test_no_http_client_injected_without_limiter(self):
+        config = _build_service(audit_config={})
+
+        assert "httpClient" not in config
+
+    def test_throttled_http_client_injected_with_limiter(self):
+        limiter = OktaRateLimiter(4)
+        config = _build_service(audit_config={}, rate_limiter=limiter)
+
+        # The SDK instantiates the class itself, so a throttled HTTPClient
+        # subclass must be injected (not an instance).
+        http_client_cls = config["httpClient"]
+        assert isinstance(http_client_cls, type)
+        assert issubclass(http_client_cls, HTTPClient)

tests/providers/okta/lib/service/rate_limiter_test.py

@@ -0,0 +1,99 @@
+import asyncio
+from unittest import mock
+
+import pytest
+
+from prowler.providers.okta.lib.service.rate_limiter import (
+    OktaRateLimiter,
+    build_throttled_http_client,
+)
+
+
+class FakeClock:
+    """Deterministic clock whose `sleep` advances time instead of waiting."""
+
+    def __init__(self):
+        self.now = 0.0
+        self.sleeps = []
+
+    def __call__(self):
+        return self.now
+
+    async def sleep(self, seconds):
+        self.sleeps.append(seconds)
+        self.now += seconds
+
+
+def _limiter(rate, clock):
+    return OktaRateLimiter(rate, clock=clock, sleep=clock.sleep)
+
+
+class Test_OktaRateLimiter:
+    def test_rejects_non_positive_rate(self):
+        with pytest.raises(ValueError):
+            OktaRateLimiter(0)
+        with pytest.raises(ValueError):
+            OktaRateLimiter(-1)
+
+    def test_initial_burst_does_not_sleep(self):
+        clock = FakeClock()
+        # capacity == rate == 2, so the first two tokens are free.
+        limiter = _limiter(2, clock)
+
+        asyncio.run(limiter.acquire())
+        asyncio.run(limiter.acquire())
+
+        assert clock.sleeps == []
+
+    def test_sleeps_to_maintain_rate_once_bucket_drained(self):
+        clock = FakeClock()
+        limiter = _limiter(2, clock)  # capacity 2, refill 2/s
+
+        # Drain the burst, then the third call must wait one refill interval.
+        asyncio.run(limiter.acquire())
+        asyncio.run(limiter.acquire())
+        asyncio.run(limiter.acquire())
+
+        assert clock.sleeps == [pytest.approx(0.5)]
+        assert clock.now == pytest.approx(0.5)
+
+    def test_elapsed_time_refills_tokens_without_sleeping(self):
+        clock = FakeClock()
+        limiter = _limiter(2, clock)
+
+        asyncio.run(limiter.acquire())
+        asyncio.run(limiter.acquire())
+        # Enough wall-clock passes to fully refill the bucket.
+        clock.now += 1.0
+        asyncio.run(limiter.acquire())
+
+        assert clock.sleeps == []
+
+    def test_rate_below_one_per_second(self):
+        clock = FakeClock()
+        limiter = _limiter(0.5, clock)  # capacity floored to 1.0
+
+        asyncio.run(limiter.acquire())  # free initial token
+        asyncio.run(limiter.acquire())  # must wait 1 / 0.5 = 2s
+
+        assert clock.sleeps == [pytest.approx(2.0)]
+
+
+class Test_build_throttled_http_client:
+    def test_acquires_before_delegating_to_super(self):
+        limiter = mock.MagicMock()
+        limiter.acquire = mock.AsyncMock()
+
+        throttled_cls = build_throttled_http_client(limiter)
+        client = throttled_cls({"headers": {}})
+
+        with mock.patch.object(
+            throttled_cls.__bases__[0],
+            "send_request",
+            new=mock.AsyncMock(return_value="response"),
+        ) as base_send:
+            result = asyncio.run(client.send_request({"method": "GET"}))
+
+        limiter.acquire.assert_awaited_once()
+        base_send.assert_awaited_once_with({"method": "GET"})
+        assert result == "response"

tests/providers/okta/okta_fixtures.py

@@ -11,6 +11,7 @@ def set_mocked_okta_provider(
     session: OktaSession = None,
     identity: OktaIdentityInfo = None,
     audit_config: dict = None,
+    rate_limiter=None,
 ):
     if session is None:
         session = OktaSession(
@@ -54,4 +55,7 @@ def set_mocked_okta_provider(
     provider.session = session
     provider.identity = identity
     provider.audit_config = audit_config or {}
+    # Default to no throttling so service tests build a plain SDK client; tests
+    # that exercise the limiter pass one explicitly.
+    provider.rate_limiter = rate_limiter
     return provider

tests/providers/okta/okta_provider_test.py

@@ -497,6 +497,85 @@ class Test_OktaProvider_init:
         assert provider.audit_config is not None
         assert provider.mutelist is not None
 
+    def test_default_max_retries_from_config_file(self, _clear_okta_env, tmp_path):
+        validate_p, session_p, identity_p = _mock_setup_paths()
+        with validate_p, session_p, identity_p:
+            provider = OktaProvider(
+                okta_org_domain=OKTA_ORG_DOMAIN,
+                okta_client_id=OKTA_CLIENT_ID,
+                okta_private_key_file="/tmp/key.pem",
+            )
+
+        # No CLI override: value comes from the bundled config.yaml default.
+        assert provider.audit_config["okta_max_retries"] == 5
+
+    def test_cli_retries_override_config_file(self, _clear_okta_env, tmp_path):
+        validate_p, session_p, identity_p = _mock_setup_paths()
+        with validate_p, session_p, identity_p:
+            provider = OktaProvider(
+                okta_org_domain=OKTA_ORG_DOMAIN,
+                okta_client_id=OKTA_CLIENT_ID,
+                okta_private_key_file="/tmp/key.pem",
+                okta_retries_max_attempts=9,
+            )
+
+        assert provider.audit_config["okta_max_retries"] == 9
+
+    def test_cli_retries_override_accepts_zero(self, _clear_okta_env, tmp_path):
+        validate_p, session_p, identity_p = _mock_setup_paths()
+        with validate_p, session_p, identity_p:
+            provider = OktaProvider(
+                okta_org_domain=OKTA_ORG_DOMAIN,
+                okta_client_id=OKTA_CLIENT_ID,
+                okta_private_key_file="/tmp/key.pem",
+                okta_retries_max_attempts=0,
+            )
+
+        # 0 disables retries and must not be treated as "unset".
+        assert provider.audit_config["okta_max_retries"] == 0
+
+    def test_rate_limiter_built_from_config_file_default(
+        self, _clear_okta_env, tmp_path
+    ):
+        validate_p, session_p, identity_p = _mock_setup_paths()
+        with validate_p, session_p, identity_p:
+            provider = OktaProvider(
+                okta_org_domain=OKTA_ORG_DOMAIN,
+                okta_client_id=OKTA_CLIENT_ID,
+                okta_private_key_file="/tmp/key.pem",
+            )
+
+        # Bundled config.yaml enables throttling at 4 req/s.
+        assert provider.rate_limiter is not None
+        assert provider.audit_config["okta_requests_per_second"] == 4
+
+    def test_cli_requests_per_second_override(self, _clear_okta_env, tmp_path):
+        validate_p, session_p, identity_p = _mock_setup_paths()
+        with validate_p, session_p, identity_p:
+            provider = OktaProvider(
+                okta_org_domain=OKTA_ORG_DOMAIN,
+                okta_client_id=OKTA_CLIENT_ID,
+                okta_private_key_file="/tmp/key.pem",
+                okta_requests_per_second=10,
+            )
+
+        assert provider.audit_config["okta_requests_per_second"] == 10
+        assert provider.rate_limiter is not None
+
+    def test_requests_per_second_zero_disables_throttling(
+        self, _clear_okta_env, tmp_path
+    ):
+        validate_p, session_p, identity_p = _mock_setup_paths()
+        with validate_p, session_p, identity_p:
+            provider = OktaProvider(
+                okta_org_domain=OKTA_ORG_DOMAIN,
+                okta_client_id=OKTA_CLIENT_ID,
+                okta_private_key_file="/tmp/key.pem",
+                okta_requests_per_second=0,
+            )
+
+        assert provider.rate_limiter is None
+
 
 class Test_OktaProvider_test_connection:
     def test_success(self, _clear_okta_env, tmp_path):