diff --git a/docs/user-guide/cli/tutorials/configuration_file.mdx b/docs/user-guide/cli/tutorials/configuration_file.mdx
index caf8853a1b..fed428f628 100644
--- a/docs/user-guide/cli/tutorials/configuration_file.mdx
+++ b/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.
+
+
+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.
+
+
+Use `resource_scan_limits.default` to set a provider-wide default, `services..default` to override it for a service, and `services..resource_types.` 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:
diff --git a/prowler/config/config.yaml b/prowler/config/config.yaml
index 28f07c2051..b0b77c2c0b 100644
--- a/prowler/config/config.yaml
+++ b/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.
diff --git a/prowler/providers/aws/aws_provider.py b/prowler/providers/aws/aws_provider.py
index cf6cbdd73a..586e57efad 100644
--- a/prowler/providers/aws/aws_provider.py
+++ b/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
diff --git a/prowler/providers/aws/lib/resource_limit/__init__.py b/prowler/providers/aws/lib/resource_limit/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/prowler/providers/aws/lib/resource_limit/resource_limit.py b/prowler/providers/aws/lib/resource_limit/resource_limit.py
new file mode 100644
index 0000000000..493693701a
--- /dev/null
+++ b/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
diff --git a/prowler/providers/aws/lib/service/service.py b/prowler/providers/aws/lib/service/service.py
index ac241c64a2..c23d9c562e 100644
--- a/prowler/providers/aws/lib/service/service.py
+++ b/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:
diff --git a/prowler/providers/aws/services/awslambda/awslambda_service.py b/prowler/providers/aws/services/awslambda/awslambda_service.py
index 433a5ab588..e3e2930b86 100644
--- a/prowler/providers/aws/services/awslambda/awslambda_service.py
+++ b/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", {})
diff --git a/prowler/providers/aws/services/backup/backup_service.py b/prowler/providers/aws/services/backup/backup_service.py
index 4320d42c0b..0d963d0079 100644
--- a/prowler/providers/aws/services/backup/backup_service.py
+++ b/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,
diff --git a/prowler/providers/aws/services/cloudwatch/cloudwatch_service.py b/prowler/providers/aws/services/cloudwatch/cloudwatch_service.py
index 29ac103867..81bdf9ba3c 100644
--- a/prowler/providers/aws/services/cloudwatch/cloudwatch_service.py
+++ b/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")
diff --git a/prowler/providers/aws/services/codeartifact/codeartifact_service.py b/prowler/providers/aws/services/codeartifact/codeartifact_service.py
index 1465092063..4240895eb2 100644
--- a/prowler/providers/aws/services/codeartifact/codeartifact_service.py
+++ b/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")
diff --git a/prowler/providers/aws/services/ec2/ec2_service.py b/prowler/providers/aws/services/ec2/ec2_service.py
index ccdb9e58fe..c90ba74368 100644
--- a/prowler/providers/aws/services/ec2/ec2_service.py
+++ b/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(
diff --git a/prowler/providers/aws/services/ecs/ecs_service.py b/prowler/providers/aws/services/ecs/ecs_service.py
index 560125bf58..6e628098be 100644
--- a/prowler/providers/aws/services/ecs/ecs_service.py
+++ b/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]),
diff --git a/tests/config/config_test.py b/tests/config/config_test.py
index 2a7aecd330..78afa0c841 100644
--- a/tests/config/config_test.py
+++ b/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,
diff --git a/tests/config/fixtures/config.yaml b/tests/config/fixtures/config.yaml
index 39cba5f27d..975f0ac816 100644
--- a/tests/config/fixtures/config.yaml
+++ b/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:
diff --git a/tests/providers/aws/lib/resource_limit/resource_limit_test.py b/tests/providers/aws/lib/resource_limit/resource_limit_test.py
new file mode 100644
index 0000000000..89e3503f78
--- /dev/null
+++ b/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
diff --git a/tests/providers/aws/services/ec2/ec2_service_test.py b/tests/providers/aws/services/ec2/ec2_service_test.py
index aca200dd61..cc1985bd78 100644
--- a/tests/providers/aws/services/ec2/ec2_service_test.py
+++ b/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):
diff --git a/tests/providers/aws/services/ecs/ecs_service_test.py b/tests/providers/aws/services/ecs/ecs_service_test.py
index b472e94b88..cf8acfd94d 100644
--- a/tests/providers/aws/services/ecs/ecs_service_test.py
+++ b/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):
diff --git a/tests/providers/aws/utils.py b/tests/providers/aws/utils.py
index 0b11798e1f..6eb7406c6c 100644
--- a/tests/providers/aws/utils.py
+++ b/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