docs(sdk): add resource scan limits changelog

- Add AWS resource scan limit resolution by provider, service, and resource type
- Apply limits during discovery for high-volume AWS resources
- Preserve explicit resource ARN scans and unlimited default behavior
- Document configuration and add focused regression coverage

docs/user-guide/cli/tutorials/configuration_file.mdx

@@ -12,6 +12,40 @@ Additionally, you can input a custom configuration file using the `--config-file
 
 ## AWS
 
+### Resource Scan Limits
+
+AWS resource scan limits reduce expensive resource discovery and enrichment for selected services. They apply before findings are generated, so Prowler can avoid per-resource API calls when the configured limit is reached.
+
+<Warning>
+Resource scan limits can make compliance results incomplete because Prowler scans only part of the affected resource set. Explicit `--resource-arn` scans bypass resource scan limits. `--resource-tag` scans do not bypass these limits because tags are discovery filters.
+</Warning>
+
+Use `resource_scan_limits.default` to set a provider-wide default, `services.<service>.default` to override it for a service, and `services.<service>.resource_types.<resource_type>` to override it for one resource type. The precedence is resource type override, service default, provider default, and unlimited. Missing values, `null`, and `0` mean unlimited. Invalid values emit a runtime warning and are treated as unlimited.
+
+```yaml title="config.yaml"
+aws:
+  resource_scan_limits:
+    default: null
+    services:
+      ec2:
+        default: 100
+        resource_types:
+          snapshot: 25
+      ecs:
+        default: null
+        resource_types:
+          task_definition: 50
+```
+
+Initial AWS resource scan limit coverage includes:
+
+* **Lambda functions:** `awslambda.function`
+* **ECS task definitions:** `ecs.task_definition`
+* **EC2 EBS snapshots:** `ec2.snapshot`
+* **Backup recovery points:** `backup.recovery_point`
+* **CloudWatch log groups:** `cloudwatch.log_group`
+* **CodeArtifact packages:** `codeartifact.package`
+
 ### Configurable Checks
 The following list includes all the AWS checks with configurable variables that can be changed in the configuration yaml file:
 

prowler/CHANGELOG.md

@@ -16,6 +16,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
 - `user`, `systemlog` and `idp` service for Okta provider with `user_inactivity_automation_35d_enabled`, `systemlog_streaming_enabled` and `idp_smart_card_dod_approved_ca` checks [(#11496)](https://github.com/prowler-cloud/prowler/pull/11496)
 - AWS AI Security Framework support in the CLI dashboard [(#11475)](https://github.com/prowler-cloud/prowler/pull/11475)
 - `entra_service_principal_privileged_role_no_owners` check for M365 provider, failing when a service principal with a permanent Tier 0 directory role has owners on the service principal or its parent app registration [(#11070)](https://github.com/prowler-cloud/prowler/issues/11070)
+- Configurable AWS resource scan limits by provider, service, and resource type [(#11509)](https://github.com/prowler-cloud/prowler/pull/11509)
 
 ### 🐞 Fixed
 

prowler/config/config.yaml

@@ -3,6 +3,36 @@ aws:
   # AWS Global Configuration
   # aws.mute_non_default_regions --> Set to True to muted failed findings in non-default regions for AccessAnalyzer, GuardDuty, SecurityHub, DRS and Config
   mute_non_default_regions: False
+  # AWS resource scan limits for expensive resource discovery. null and 0 mean unlimited.
+  # Precedence: resource type override > service default > provider default > unlimited.
+  # Explicit --resource-arn scans bypass these limits; --resource-tag scans do not.
+  resource_scan_limits:
+    default: null
+    services:
+      awslambda:
+        default: null
+        resource_types:
+          function: null
+      backup:
+        default: null
+        resource_types:
+          recovery_point: null
+      cloudwatch:
+        default: null
+        resource_types:
+          log_group: null
+      codeartifact:
+        default: null
+        resource_types:
+          package: null
+      ec2:
+        default: null
+        resource_types:
+          snapshot: null
+      ecs:
+        default: null
+        resource_types:
+          task_definition: null
   # aws.disallowed_regions --> List of AWS regions to exclude from the scan.
   # Also settable via the PROWLER_AWS_DISALLOWED_REGIONS environment variable or
   # the --excluded-region CLI flag. Precedence: CLI > env var > config file.

prowler/providers/aws/aws_provider.py

@@ -94,6 +94,7 @@ class AwsProvider(Provider):
     _session: AWSSession
     _organizations_metadata: AWSOrganizationsInfo
     _audit_resources: list = []
+    _audit_resources_from_resource_arn: bool = False
     _audit_config: dict
     _scan_unused_services: bool = False
     _enabled_regions: set | None = None
@@ -417,12 +418,14 @@ class AwsProvider(Provider):
 
         # Parse Scan Tags after region exclusions are applied so tag discovery
         # also skips disallowed regions.
+        self._audit_resources_from_resource_arn = False
         if resource_tags:
             self._audit_resources = self.get_tagged_resources(resource_tags)
 
         # Parse Input Resource ARNs
         if resource_arn:
             self._audit_resources = resource_arn
+            self._audit_resources_from_resource_arn = True
 
         # Set ignore unused services
         self._scan_unused_services = scan_unused_services
@@ -468,6 +471,10 @@ class AwsProvider(Provider):
     def audit_resources(self):
         return self._audit_resources
 
+    @property
+    def audit_resources_from_resource_arn(self):
+        return self._audit_resources_from_resource_arn
+
     @property
     def scan_unused_services(self):
         return self._scan_unused_services

prowler/providers/aws/lib/resource_limit/resource_limit.py

@@ -0,0 +1,82 @@
+from threading import Lock
+
+from prowler.lib.logger import logger
+
+RESOURCE_SCAN_LIMITS_CONFIG_KEY = "resource_scan_limits"
+
+
+class ResourceScanLimiter:
+    def __init__(self, audit_config: dict, service: str, bypass_limits: bool = False):
+        self._config = audit_config.get(RESOURCE_SCAN_LIMITS_CONFIG_KEY, {}) or {}
+        self._service = service
+        self._bypass_limits = bypass_limits
+        self._counters = {}
+        self._lock = Lock()
+        if not self._bypass_limits and self.has_active_limits():
+            logger.warning(
+                "AWS resource scan limits are active for service "
+                f"{service}. Explicit --resource-arn scans bypass these limits. "
+                "Resource discovery may be truncated and compliance results may be incomplete."
+            )
+
+    def has_active_limits(self) -> bool:
+        service_config = self._service_config()
+        resource_types = service_config.get("resource_types", {}) or {}
+        if any(
+            self._parse_limit(limit) is not None
+            for limit in resource_types.values()
+            if limit is not None
+        ):
+            return True
+        if service_config.get("default") is not None:
+            return self._parse_limit(service_config.get("default")) is not None
+        if "default" in self._config:
+            return self._parse_limit(self._config.get("default")) is not None
+        return False
+
+    def limit_for(self, resource_type: str) -> int | None:
+        if self._bypass_limits:
+            return None
+
+        service_config = self._service_config()
+        resource_types = service_config.get("resource_types", {}) or {}
+        if resource_types.get(resource_type) is not None:
+            return self._parse_limit(resource_types.get(resource_type))
+        if service_config.get("default") is not None:
+            return self._parse_limit(service_config.get("default"))
+        if "default" in self._config:
+            return self._parse_limit(self._config.get("default"))
+        return None
+
+    def allow(self, resource_type: str) -> bool:
+        limit = self.limit_for(resource_type)
+        if limit is None:
+            return True
+
+        with self._lock:
+            count = self._counters.get(resource_type, 0)
+            if count >= limit:
+                return False
+            self._counters[resource_type] = count + 1
+            return True
+
+    def _service_config(self) -> dict:
+        services = self._config.get("services", {}) or {}
+        return services.get(self._service, {}) or {}
+
+    def _parse_limit(self, raw_limit) -> int | None:
+        if isinstance(raw_limit, bool):
+            logger.warning(
+                "Invalid AWS resource scan limit value "
+                f"{raw_limit!r} for service {self._service}; treating it as unlimited."
+            )
+            return None
+        if raw_limit is None or raw_limit == 0:
+            return None
+        if isinstance(raw_limit, int) and raw_limit > 0:
+            return raw_limit
+        logger.warning(
+            "Invalid AWS resource scan limit value "
+            f"{raw_limit!r} for service {self._service}; treating it as unlimited."
+        )
+        return None

prowler/providers/aws/lib/service/service.py

@@ -2,6 +2,7 @@ from concurrent.futures import ThreadPoolExecutor, as_completed
 
 from prowler.lib.logger import logger
 from prowler.providers.aws.aws_provider import AwsProvider
+from prowler.providers.aws.lib.resource_limit.resource_limit import ResourceScanLimiter
 
 # TODO: review the following code
 # from prowler.providers.aws.aws_provider import (
@@ -57,6 +58,14 @@ class AWSService:
         # We receive the service using __class__.__name__ or the service name in lowercase
         # e.g.: AccessAnalyzer --> we need a lowercase string, so service.lower()
         self.service = service.lower() if not service.islower() else service
+        resource_limit_service = (
+            "awslambda" if self.service == "lambda" else self.service
+        )
+        self.resource_scan_limiter = ResourceScanLimiter(
+            self.audit_config,
+            resource_limit_service,
+            bypass_limits=provider.audit_resources_from_resource_arn,
+        )
 
         # Generate Regional Clients
         if not global_service:

prowler/providers/aws/services/awslambda/awslambda_service.py

@@ -36,6 +36,8 @@ class Lambda(AWSService):
                             function["FunctionArn"], self.audit_resources
                         )
                     ):
+                        if not self.resource_scan_limiter.allow("function"):
+                            continue
                         lambda_name = function["FunctionName"]
                         lambda_arn = function["FunctionArn"]
                         vpc_config = function.get("VpcConfig", {})

prowler/providers/aws/services/backup/backup_service.py

@@ -195,6 +195,10 @@ class Backup(AWSService):
                         for recovery_point in page.get("RecoveryPoints", []):
                             arn = recovery_point.get("RecoveryPointArn")
                             if arn:
+                                if not self.resource_scan_limiter.allow(
+                                    "recovery_point"
+                                ):
+                                    continue
                                 self.recovery_points.append(
                                     RecoveryPoint(
                                         arn=arn,

prowler/providers/aws/services/cloudwatch/cloudwatch_service.py

@@ -160,6 +160,8 @@ class Logs(AWSService):
                     if not self.audit_resources or (
                         is_resource_filtered(log_group["arn"], self.audit_resources)
                     ):
+                        if not self.resource_scan_limiter.allow("log_group"):
+                            continue
                         never_expire = False
                         kms = log_group.get("kmsKeyId")
                         retention_days = log_group.get("retentionInDays")

prowler/providers/aws/services/codeartifact/codeartifact_service.py

@@ -69,6 +69,8 @@ class CodeArtifact(AWSService):
                         **list_packages_parameters
                     ):
                         for package in page["packages"]:
+                            if not self.resource_scan_limiter.allow("package"):
+                                continue
                             # Package information
                             package_format = package["format"]
                             package_namespace = package.get("namespace")

prowler/providers/aws/services/ec2/ec2_service.py

@@ -199,6 +199,8 @@ class EC2(AWSService):
                     if not self.audit_resources or (
                         is_resource_filtered(arn, self.audit_resources)
                     ):
+                        if not self.resource_scan_limiter.allow("snapshot"):
+                            continue
                         if snapshots_in_region is False:
                             snapshots_in_region = True
                         self.snapshots.append(

prowler/providers/aws/services/ecs/ecs_service.py

@@ -33,6 +33,8 @@ class ECS(AWSService):
                     if not self.audit_resources or (
                         is_resource_filtered(task_definition, self.audit_resources)
                     ):
+                        if not self.resource_scan_limiter.allow("task_definition"):
+                            continue
                         self.task_definitions[task_definition] = TaskDefinition(
                             # we want the family name without the revision
                             name=sub(":.*", "", task_definition.split("/")[-1]),

tests/config/config_test.py

@@ -80,6 +80,35 @@ old_config_aws = {
 }
 config_aws = {
     "mute_non_default_regions": False,
+    "resource_scan_limits": {
+        "default": None,
+        "services": {
+            "awslambda": {
+                "default": None,
+                "resource_types": {"function": None},
+            },
+            "backup": {
+                "default": None,
+                "resource_types": {"recovery_point": None},
+            },
+            "cloudwatch": {
+                "default": None,
+                "resource_types": {"log_group": None},
+            },
+            "codeartifact": {
+                "default": None,
+                "resource_types": {"package": None},
+            },
+            "ec2": {
+                "default": None,
+                "resource_types": {"snapshot": None},
+            },
+            "ecs": {
+                "default": None,
+                "resource_types": {"task_definition": None},
+            },
+        },
+    },
     "max_unused_access_keys_days": 45,
     "max_console_access_days": 45,
     "max_unused_sagemaker_access_days": 90,

tests/config/fixtures/config.yaml

@@ -3,6 +3,34 @@ aws:
   # AWS Global Configuration
   # aws.mute_non_default_regions --> Set to True to muted failed findings in non-default regions for AccessAnalyzer, GuardDuty, SecurityHub, DRS and Config
   mute_non_default_regions: False
+  # AWS resource scan limits for expensive resource discovery. null and 0 mean unlimited.
+  resource_scan_limits:
+    default: null
+    services:
+      awslambda:
+        default: null
+        resource_types:
+          function: null
+      backup:
+        default: null
+        resource_types:
+          recovery_point: null
+      cloudwatch:
+        default: null
+        resource_types:
+          log_group: null
+      codeartifact:
+        default: null
+        resource_types:
+          package: null
+      ec2:
+        default: null
+        resource_types:
+          snapshot: null
+      ecs:
+        default: null
+        resource_types:
+          task_definition: null
   # If you want to mute failed findings only in specific regions, create a file with the following syntax and run it with `prowler aws -w mutelist.yaml`:
   # Mutelist:
   #  Accounts:

tests/providers/aws/lib/resource_limit/resource_limit_test.py

@@ -0,0 +1,272 @@
+import logging
+
+import pytest
+
+from prowler.providers.aws.lib.resource_limit.resource_limit import ResourceScanLimiter
+
+
+def test_missing_config_is_unlimited():
+    limiter = ResourceScanLimiter({}, "ec2")
+
+    assert limiter.limit_for("snapshot") is None
+    assert limiter.allow("snapshot")
+    assert limiter.allow("snapshot")
+
+
+def test_null_and_zero_limits_are_unlimited():
+    config = {
+        "resource_scan_limits": {
+            "default": 0,
+            "services": {
+                "ec2": {
+                    "default": None,
+                    "resource_types": {"snapshot": 0},
+                }
+            },
+        }
+    }
+    limiter = ResourceScanLimiter(config, "ec2")
+
+    assert limiter.limit_for("snapshot") is None
+    assert limiter.allow("snapshot")
+    assert limiter.allow("snapshot")
+
+
+def test_provider_default_applies():
+    limiter = ResourceScanLimiter(
+        {"resource_scan_limits": {"default": 1}},
+        "ec2",
+    )
+
+    assert limiter.limit_for("snapshot") == 1
+    assert limiter.allow("snapshot")
+    assert not limiter.allow("snapshot")
+
+
+def test_service_default_overrides_provider_default():
+    limiter = ResourceScanLimiter(
+        {
+            "resource_scan_limits": {
+                "default": 1,
+                "services": {"ec2": {"default": 2}},
+            }
+        },
+        "ec2",
+    )
+
+    assert limiter.limit_for("snapshot") == 2
+    assert limiter.allow("snapshot")
+    assert limiter.allow("snapshot")
+    assert not limiter.allow("snapshot")
+
+
+def test_resource_type_override_overrides_service_default():
+    limiter = ResourceScanLimiter(
+        {
+            "resource_scan_limits": {
+                "default": 1,
+                "services": {
+                    "ec2": {
+                        "default": 2,
+                        "resource_types": {"snapshot": 3},
+                    }
+                },
+            }
+        },
+        "ec2",
+    )
+
+    assert limiter.limit_for("snapshot") == 3
+    assert limiter.allow("snapshot")
+    assert limiter.allow("snapshot")
+    assert limiter.allow("snapshot")
+    assert not limiter.allow("snapshot")
+
+
+def test_null_service_and_resource_placeholders_do_not_mask_provider_default():
+    limiter = ResourceScanLimiter(
+        {
+            "resource_scan_limits": {
+                "default": 2,
+                "services": {
+                    "ec2": {
+                        "default": None,
+                        "resource_types": {"snapshot": None},
+                    }
+                },
+            }
+        },
+        "ec2",
+    )
+
+    assert limiter.limit_for("snapshot") == 2
+    assert limiter.allow("snapshot")
+    assert limiter.allow("snapshot")
+    assert not limiter.allow("snapshot")
+
+
+def test_null_resource_type_placeholder_does_not_mask_service_default():
+    limiter = ResourceScanLimiter(
+        {
+            "resource_scan_limits": {
+                "default": 1,
+                "services": {
+                    "ec2": {
+                        "default": 2,
+                        "resource_types": {"snapshot": None},
+                    }
+                },
+            }
+        },
+        "ec2",
+    )
+
+    assert limiter.limit_for("snapshot") == 2
+    assert limiter.allow("snapshot")
+    assert limiter.allow("snapshot")
+    assert not limiter.allow("snapshot")
+
+
+def test_real_config_shape_null_placeholders_do_not_mask_provider_default():
+    limiter = ResourceScanLimiter(
+        {
+            "resource_scan_limits": {
+                "default": 2,
+                "services": {
+                    "awslambda": {
+                        "default": None,
+                        "resource_types": {"function": None},
+                    },
+                    "backup": {
+                        "default": None,
+                        "resource_types": {"recovery_point": None},
+                    },
+                    "cloudwatch": {
+                        "default": None,
+                        "resource_types": {"log_group": None},
+                    },
+                    "codeartifact": {
+                        "default": None,
+                        "resource_types": {"package": None},
+                    },
+                    "ec2": {
+                        "default": None,
+                        "resource_types": {"snapshot": None},
+                    },
+                    "ecs": {
+                        "default": None,
+                        "resource_types": {"task_definition": None},
+                    },
+                },
+            }
+        },
+        "ec2",
+    )
+
+    assert limiter.limit_for("snapshot") == 2
+    assert limiter.allow("snapshot")
+    assert limiter.allow("snapshot")
+    assert not limiter.allow("snapshot")
+
+
+def test_invalid_resource_type_override_disables_lower_precedence_limits(caplog):
+    limiter = ResourceScanLimiter(
+        {
+            "resource_scan_limits": {
+                "default": 1,
+                "services": {
+                    "ec2": {
+                        "default": 1,
+                        "resource_types": {"snapshot": "invalid"},
+                    }
+                },
+            }
+        },
+        "ec2",
+    )
+
+    with caplog.at_level(logging.WARNING):
+        assert limiter.limit_for("snapshot") is None
+
+    assert "Invalid AWS resource scan limit" in caplog.text
+    assert limiter.allow("snapshot")
+    assert limiter.allow("snapshot")
+
+
+def test_invalid_values_warn_and_resolve_to_unlimited(caplog):
+    limiter = ResourceScanLimiter(
+        {
+            "resource_scan_limits": {
+                "default": -1,
+                "services": {
+                    "ec2": {
+                        "default": "two",
+                        "resource_types": {"snapshot": "invalid"},
+                    }
+                },
+            }
+        },
+        "ec2",
+    )
+
+    with caplog.at_level(logging.WARNING):
+        assert limiter.limit_for("snapshot") is None
+
+    assert "Invalid AWS resource scan limit" in caplog.text
+    assert limiter.allow("snapshot")
+    assert limiter.allow("snapshot")
+
+
+@pytest.mark.parametrize("raw_limit", [True, False])
+def test_bool_values_warn_and_resolve_to_unlimited(caplog, raw_limit):
+    limiter = ResourceScanLimiter(
+        {
+            "resource_scan_limits": {
+                "services": {
+                    "ec2": {
+                        "resource_types": {"snapshot": raw_limit},
+                    }
+                }
+            }
+        },
+        "ec2",
+    )
+
+    caplog.clear()
+    with caplog.at_level(logging.WARNING):
+        assert limiter.limit_for("snapshot") is None
+
+    assert "Invalid AWS resource scan limit" in caplog.text
+    assert limiter.allow("snapshot")
+    assert limiter.allow("snapshot")
+
+
+def test_explicit_resource_arn_source_bypasses_limits():
+    limiter = ResourceScanLimiter(
+        {"resource_scan_limits": {"default": 1}},
+        "ec2",
+        bypass_limits=True,
+    )
+
+    assert limiter.limit_for("snapshot") is None
+    assert limiter.allow("snapshot")
+    assert limiter.allow("snapshot")
+
+
+def test_active_limits_emit_runtime_warning(caplog):
+    with caplog.at_level(logging.WARNING):
+        ResourceScanLimiter({"resource_scan_limits": {"default": 1}}, "ec2")
+
+    assert "AWS resource scan limits are active" in caplog.text
+    assert "compliance results may be incomplete" in caplog.text
+
+
+def test_explicit_resource_arn_bypass_does_not_emit_runtime_warning(caplog):
+    with caplog.at_level(logging.WARNING):
+        ResourceScanLimiter(
+            {"resource_scan_limits": {"default": 1}},
+            "ec2",
+            bypass_limits=True,
+        )
+
+    assert "AWS resource scan limits are active" not in caplog.text

tests/providers/aws/services/ec2/ec2_service_test.py

@@ -305,6 +305,38 @@ class Test_EC2_Service:
                 assert not snapshot.encrypted
                 assert not snapshot.public
 
+    @mock_aws
+    def test_describe_snapshots_honors_resource_scan_limit(self):
+        ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
+        ec2_resource = resource("ec2", region_name=AWS_REGION_US_EAST_1)
+        volume_id = ec2_resource.create_volume(
+            AvailabilityZone="us-east-1a",
+            Size=80,
+            VolumeType="gp2",
+        ).id
+        ec2_client.create_snapshot(VolumeId=volume_id)
+        ec2_client.create_snapshot(VolumeId=volume_id)
+        available_snapshots = ec2_client.describe_snapshots(OwnerIds=["self"])[
+            "Snapshots"
+        ]
+        aws_provider = set_mocked_aws_provider(
+            [AWS_REGION_US_EAST_1],
+            audit_config={
+                "resource_scan_limits": {
+                    "services": {
+                        "ec2": {
+                            "resource_types": {"snapshot": 1},
+                        }
+                    }
+                }
+            },
+        )
+
+        ec2 = EC2(aws_provider)
+
+        assert len(available_snapshots) > 1
+        assert len(ec2.snapshots) == 1
+
     # Test EC2 Get Snapshot Public
     @mock_aws
     def test__get_snapshot_public__(self):

tests/providers/aws/services/ecs/ecs_service_test.py

@@ -107,6 +107,17 @@ def mock_make_api_call(self, operation_name, kwarg):
     return make_api_call(self, operation_name, kwarg)
 
 
+def mock_make_api_call_two_task_definitions(self, operation_name, kwarg):
+    if operation_name == "ListTaskDefinitions":
+        return {
+            "taskDefinitionArns": [
+                "arn:aws:ecs:eu-west-1:123456789012:task-definition/test_cluster_1/test_ecs_task:1",
+                "arn:aws:ecs:eu-west-1:123456789012:task-definition/test_cluster_1/test_ecs_task_2:1",
+            ]
+        }
+    return mock_make_api_call(self, operation_name, kwarg)
+
+
 def mock_generate_regional_clients(provider, service):
     regional_client = provider._session.current_session.client(
         service, region_name=AWS_REGION_EU_WEST_1
@@ -153,6 +164,91 @@ class Test_ECS_Service:
         assert ecs.task_definitions[task_arn].revision == "1"
         assert ecs.task_definitions[task_arn].region == AWS_REGION_EU_WEST_1
 
+    @patch(
+        "botocore.client.BaseClient._make_api_call",
+        new=mock_make_api_call_two_task_definitions,
+    )
+    def test_list_task_definitions_honors_resource_scan_limit(self):
+        aws_provider = set_mocked_aws_provider(
+            [AWS_REGION_EU_WEST_1],
+            audit_config={
+                "resource_scan_limits": {
+                    "services": {
+                        "ecs": {
+                            "resource_types": {"task_definition": 1},
+                        }
+                    }
+                }
+            },
+        )
+
+        ecs = ECS(aws_provider)
+
+        assert len(ecs.task_definitions) == 1
+        assert (
+            "arn:aws:ecs:eu-west-1:123456789012:task-definition/test_cluster_1/test_ecs_task:1"
+            in ecs.task_definitions
+        )
+        assert (
+            "arn:aws:ecs:eu-west-1:123456789012:task-definition/test_cluster_1/test_ecs_task_2:1"
+            not in ecs.task_definitions
+        )
+
+    @patch(
+        "botocore.client.BaseClient._make_api_call",
+        new=mock_make_api_call_two_task_definitions,
+    )
+    def test_generic_audit_resources_do_not_bypass_resource_scan_limit(self):
+        task_arn = "arn:aws:ecs:eu-west-1:123456789012:task-definition/test_cluster_1/test_ecs_task:1"
+        task_arn_2 = "arn:aws:ecs:eu-west-1:123456789012:task-definition/test_cluster_1/test_ecs_task_2:1"
+        aws_provider = set_mocked_aws_provider(
+            [AWS_REGION_EU_WEST_1],
+            audit_config={
+                "resource_scan_limits": {
+                    "services": {
+                        "ecs": {
+                            "resource_types": {"task_definition": 1},
+                        }
+                    }
+                }
+            },
+        )
+        aws_provider._audit_resources = [task_arn, task_arn_2]
+
+        ecs = ECS(aws_provider)
+
+        assert len(ecs.task_definitions) == 1
+        assert task_arn in ecs.task_definitions
+        assert task_arn_2 not in ecs.task_definitions
+
+    @patch(
+        "botocore.client.BaseClient._make_api_call",
+        new=mock_make_api_call_two_task_definitions,
+    )
+    def test_explicit_resource_arn_origin_bypasses_resource_scan_limit(self):
+        task_arn = "arn:aws:ecs:eu-west-1:123456789012:task-definition/test_cluster_1/test_ecs_task:1"
+        task_arn_2 = "arn:aws:ecs:eu-west-1:123456789012:task-definition/test_cluster_1/test_ecs_task_2:1"
+        aws_provider = set_mocked_aws_provider(
+            [AWS_REGION_EU_WEST_1],
+            audit_config={
+                "resource_scan_limits": {
+                    "services": {
+                        "ecs": {
+                            "resource_types": {"task_definition": 1},
+                        }
+                    }
+                }
+            },
+        )
+        aws_provider._audit_resources = [task_arn, task_arn_2]
+        aws_provider._audit_resources_from_resource_arn = True
+
+        ecs = ECS(aws_provider)
+
+        assert len(ecs.task_definitions) == 2
+        assert task_arn in ecs.task_definitions
+        assert task_arn_2 in ecs.task_definitions
+
     @patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
     # Test describe ECS task definitions
     def test_describe_task_definitions(self):

tests/providers/aws/utils.py

@@ -156,6 +156,7 @@ def set_mocked_aws_provider(
     # TODO: we can create the organizations metadata here with moto
     provider._organizations_metadata = None
     provider._audit_resources = []
+    provider._audit_resources_from_resource_arn = False
     provider._audit_config = audit_config
     provider._fixer_config = fixer_config
     provider._mutelist = mutelist