fix(aws): treat security groups on Batch compute environments as used (#12458)

Co-authored-by: hackertwinten <193916571+hackertwinten@users.noreply.github.com>
Co-authored-by: Daniel Barranquero <danielbo2001@gmail.com>

prowler/changelog.d/ec2-securitygroup-not-used-batch.fixed.md

@@ -0,0 +1 @@
+`ec2_securitygroup_not_used` no longer reports a false positive for security groups attached only to an AWS Batch compute environment, which holds them in configuration without creating a network interface while scaled down to zero instances

prowler/providers/aws/services/batch/batch_service.py

@@ -1,6 +1,7 @@
 from itertools import zip_longest
 from typing import Optional
 
+from botocore.exceptions import ClientError
 from pydantic.v1 import BaseModel
 
 from prowler.lib.logger import logger
@@ -34,18 +35,38 @@ class BatchJobDefinition(BaseModel):
     container_properties: BatchContainerProperties
 
 
+class BatchComputeEnvironment(BaseModel):
+    """An AWS Batch compute environment with its networking configuration."""
+
+    name: str
+    arn: str
+    region: str
+    security_groups: list[str] = []
+    subnets: list[str] = []
+
+
 class Batch(AWSService):
-    """AWS Batch service client for listing job definitions."""
+    """AWS Batch service client for listing job definitions and compute environments."""
 
     def __init__(self, provider):
         super().__init__(__class__.__name__, provider)
         self.job_definitions = {}
         self._job_definitions_by_region = {}
+        self.compute_environments = {}
+        # Security groups referenced by compute environments. A compute
+        # environment holds them in its configuration even while it is scaled
+        # down to zero instances, so no ENI exists to reveal the association.
+        self.security_groups_in_use = set()
+        # Regions whose compute environments could not be listed. Their
+        # security group associations are unknown rather than absent, so
+        # consumers must not read an empty result as "nothing is attached".
+        self.compute_environment_lookup_failed_regions = set()
         self.job_definition_limit = get_resource_scan_limit(
             self.audit_config, "max_batch_job_definitions"
         )
         self.__threading_call__(self._list_job_definitions)
         self._select_job_definitions_for_analysis()
+        self.__threading_call__(self._describe_compute_environments)
 
     def _list_job_definitions(self, regional_client):
         """List ACTIVE job definitions for a regional client."""
@@ -89,6 +110,47 @@ class Batch(AWSService):
                 f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
             )
 
+    def _describe_compute_environments(self, regional_client):
+        """Describe the compute environments for a regional client."""
+        logger.info("Batch - Describing Compute Environments...")
+        try:
+            paginator = regional_client.get_paginator("describe_compute_environments")
+            for page in paginator.paginate():
+                for compute_environment in page.get("computeEnvironments", []):
+                    arn = compute_environment["computeEnvironmentArn"]
+                    if self.audit_resources and not is_resource_filtered(
+                        arn, self.audit_resources
+                    ):
+                        continue
+                    compute_resources = compute_environment.get("computeResources", {})
+                    security_groups = compute_resources.get("securityGroupIds", [])
+                    self.security_groups_in_use.update(security_groups)
+                    self.compute_environments[arn] = BatchComputeEnvironment(
+                        name=compute_environment["computeEnvironmentName"],
+                        arn=arn,
+                        region=regional_client.region,
+                        security_groups=security_groups,
+                        subnets=compute_resources.get("subnets", []),
+                    )
+        except ClientError as error:
+            self.compute_environment_lookup_failed_regions.add(regional_client.region)
+            if error.response["Error"]["Code"] in (
+                "AccessDeniedException",
+                "UnrecognizedClientException",
+            ):
+                logger.warning(
+                    f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                )
+            else:
+                logger.error(
+                    f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                )
+        except Exception as error:
+            self.compute_environment_lookup_failed_regions.add(regional_client.region)
+            logger.error(
+                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
     def _select_job_definitions_for_analysis(self):
         """Apply the global resource limit, interleaving regions fairly."""
         interleaved = [

prowler/providers/aws/services/ec2/ec2_securitygroup_not_used/ec2_securitygroup_not_used.metadata.json

@@ -11,7 +11,7 @@
   "Severity": "low",
   "ResourceType": "AwsEc2SecurityGroup",
   "ResourceGroup": "network",
-  "Description": "EC2 security groups, except `default`, are assessed for **unused** status: zero attached network interfaces, no AWS Lambda associations, and no references from other security groups.",
+  "Description": "EC2 security groups, except `default`, are assessed for **unused** status: zero attached network interfaces, no AWS Lambda associations, no AWS Batch compute environment associations, and no references from other security groups.",
   "Risk": "Orphaned security groups may later be attached with **overly permissive rules** without review, enabling unintended inbound or lateral access that compromises **confidentiality** and **integrity**. They also create **configuration drift**, increasing the chance of misapplied access controls.",
   "RelatedUrl": "",
   "AdditionalURLs": [

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

@@ -1,5 +1,6 @@
 from prowler.lib.check.models import Check, Check_Report_AWS
 from prowler.providers.aws.services.awslambda.awslambda_client import awslambda_client
+from prowler.providers.aws.services.batch.batch_client import batch_client
 from prowler.providers.aws.services.ec2.ec2_client import ec2_client
 
 
@@ -18,6 +19,9 @@ class ec2_securitygroup_not_used(Check):
                 sg_in_lambda = (
                     security_group.id in awslambda_client.security_groups_in_use
                 )
+                # A Batch compute environment scaled down to zero instances
+                # keeps its security groups in configuration without any ENI
+                sg_in_batch = security_group.id in batch_client.security_groups_in_use
                 sg_associated = False
                 for sg in ec2_client.security_groups.values():
                     if security_group.id in sg.associated_sgs:
@@ -25,10 +29,24 @@ class ec2_securitygroup_not_used(Check):
                 if (
                     len(security_group.network_interfaces) == 0
                     and not sg_in_lambda
+                    and not sg_in_batch
                     and not sg_associated
                 ):
-                    report.status = "FAIL"
-                    report.status_extended = f"Security group {security_group.name} ({security_group.id}) it is not being used."
+                    # Compute environments failing to list leaves their security
+                    # group associations unknown, not absent, so reporting the
+                    # group as unused would be a guess. Not being able to read
+                    # the compute environments is a lack of visibility, not a
+                    # misconfiguration, so report MANUAL rather than asserting a
+                    # status either way.
+                    if (
+                        security_group.region
+                        in batch_client.compute_environment_lookup_failed_regions
+                    ):
+                        report.status = "MANUAL"
+                        report.status_extended = f"Security group {security_group.name} ({security_group.id}) usage could not be verified because AWS Batch compute environments could not be listed in region {security_group.region}; grant batch:DescribeComputeEnvironments and run the check again."
+                    else:
+                        report.status = "FAIL"
+                        report.status_extended = f"Security group {security_group.name} ({security_group.id}) it is not being used."
 
                 findings.append(report)
 

tests/providers/aws/services/batch/batch_service_test.py

@@ -1,6 +1,7 @@
 from unittest.mock import patch
 
 import botocore
+from botocore.exceptions import ClientError
 
 from prowler.providers.aws.services.batch.batch_service import Batch
 from tests.providers.aws.utils import (
@@ -11,6 +12,16 @@ from tests.providers.aws.utils import (
 
 make_api_call = botocore.client.BaseClient._make_api_call
 
+COMPUTE_ENVIRONMENT_NAME = "test-compute-environment"
+COMPUTE_ENVIRONMENT_ARN = f"arn:aws:batch:eu-west-1:123456789012:compute-environment/{COMPUTE_ENVIRONMENT_NAME}"
+
+
+def default_api_call(self, operation_name, kwarg):
+    """Fall back to the real call, stubbing the APIs a test does not exercise."""
+    if operation_name == "DescribeComputeEnvironments":
+        return {"computeEnvironments": []}
+    return make_api_call(self, operation_name, kwarg)
+
 
 def mock_make_api_call(self, operation_name, kwarg):
     if operation_name == "DescribeJobDefinitions":
@@ -31,7 +42,7 @@ def mock_make_api_call(self, operation_name, kwarg):
                 }
             ]
         }
-    return make_api_call(self, operation_name, kwarg)
+    return default_api_call(self, operation_name, kwarg)
 
 
 def mock_generate_regional_clients(provider, service):
@@ -114,7 +125,7 @@ class Test_Batch_Service:
         def mock_make_api_call_empty(self, operation_name, kwarg):
             if operation_name == "DescribeJobDefinitions":
                 return {"jobDefinitions": []}
-            return make_api_call(self, operation_name, kwarg)
+            return default_api_call(self, operation_name, kwarg)
 
         with patch(
             "botocore.client.BaseClient._make_api_call",
@@ -144,7 +155,7 @@ class Test_Batch_Service:
                         for i in (3, 2, 1)
                     ]
                 }
-            return make_api_call(self, operation_name, kwarg)
+            return default_api_call(self, operation_name, kwarg)
 
         with patch(
             "botocore.client.BaseClient._make_api_call", new=counting_make_api_call
@@ -176,7 +187,7 @@ class Test_Batch_Service:
                         for i in (3, 2, 1)
                     ]
                 }
-            return make_api_call(self, operation_name, kwarg)
+            return default_api_call(self, operation_name, kwarg)
 
         with patch(
             "botocore.client.BaseClient._make_api_call", new=counting_make_api_call
@@ -190,6 +201,134 @@ class Test_Batch_Service:
             assert [jd.revision for jd in batch.job_definitions.values()] == [3, 2]
             assert len(describe_calls) == 1
 
+    def test_describe_compute_environments(self):
+        def compute_environments_api_call(self, operation_name, kwarg):
+            if operation_name == "DescribeComputeEnvironments":
+                return {
+                    "computeEnvironments": [
+                        {
+                            "computeEnvironmentName": COMPUTE_ENVIRONMENT_NAME,
+                            "computeEnvironmentArn": COMPUTE_ENVIRONMENT_ARN,
+                            "computeResources": {
+                                "securityGroupIds": ["sg-1", "sg-2"],
+                                "subnets": ["subnet-1"],
+                            },
+                        }
+                    ]
+                }
+            return mock_make_api_call(self, operation_name, kwarg)
+
+        with patch(
+            "botocore.client.BaseClient._make_api_call",
+            new=compute_environments_api_call,
+        ):
+            aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+            batch = Batch(aws_provider)
+
+            assert len(batch.compute_environments) == 1
+            compute_environment = batch.compute_environments[COMPUTE_ENVIRONMENT_ARN]
+            assert compute_environment.name == COMPUTE_ENVIRONMENT_NAME
+            assert compute_environment.arn == COMPUTE_ENVIRONMENT_ARN
+            assert compute_environment.region == AWS_REGION_EU_WEST_1
+            assert compute_environment.security_groups == ["sg-1", "sg-2"]
+            assert compute_environment.subnets == ["subnet-1"]
+            assert batch.security_groups_in_use == {"sg-1", "sg-2"}
+
+    @patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
+    def test_no_compute_environments(self):
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+        batch = Batch(aws_provider)
+
+        assert len(batch.compute_environments) == 0
+        assert batch.security_groups_in_use == set()
+        assert batch.compute_environment_lookup_failed_regions == set()
+
+    def test_compute_environments_access_denied(self):
+        def access_denied_api_call(self, operation_name, kwarg):
+            if operation_name == "DescribeComputeEnvironments":
+                raise ClientError(
+                    {
+                        "Error": {
+                            "Code": "AccessDeniedException",
+                            "Message": "User is not authorized to perform: batch:DescribeComputeEnvironments",
+                        }
+                    },
+                    operation_name,
+                )
+            return mock_make_api_call(self, operation_name, kwarg)
+
+        with patch(
+            "botocore.client.BaseClient._make_api_call", new=access_denied_api_call
+        ):
+            aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+            batch = Batch(aws_provider)
+
+            # The associations are unknown, not absent
+            assert batch.compute_environments == {}
+            assert batch.security_groups_in_use == set()
+            assert batch.compute_environment_lookup_failed_regions == {
+                AWS_REGION_EU_WEST_1
+            }
+            # Unrelated discovery is unaffected
+            assert len(batch.job_definitions) == 1
+
+    def test_compute_environment_without_compute_resources(self):
+        def unmanaged_api_call(self, operation_name, kwarg):
+            # UNMANAGED compute environments carry no `computeResources` block
+            if operation_name == "DescribeComputeEnvironments":
+                return {
+                    "computeEnvironments": [
+                        {
+                            "computeEnvironmentName": COMPUTE_ENVIRONMENT_NAME,
+                            "computeEnvironmentArn": COMPUTE_ENVIRONMENT_ARN,
+                        }
+                    ]
+                }
+            return mock_make_api_call(self, operation_name, kwarg)
+
+        with patch("botocore.client.BaseClient._make_api_call", new=unmanaged_api_call):
+            aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+            batch = Batch(aws_provider)
+
+            assert len(batch.compute_environments) == 1
+            compute_environment = batch.compute_environments[COMPUTE_ENVIRONMENT_ARN]
+            assert compute_environment.security_groups == []
+            assert compute_environment.subnets == []
+            assert batch.security_groups_in_use == set()
+
+    def test_audit_resources_filters_compute_environments(self):
+        def compute_environments_api_call(self, operation_name, kwarg):
+            if operation_name == "DescribeComputeEnvironments":
+                return {
+                    "computeEnvironments": [
+                        {
+                            "computeEnvironmentName": f"compute-environment-{i}",
+                            "computeEnvironmentArn": f"arn:aws:batch:eu-west-1:123456789012:compute-environment/compute-environment-{i}",
+                            "computeResources": {
+                                "securityGroupIds": [f"sg-{i}"],
+                                "subnets": ["subnet-1"],
+                            },
+                        }
+                        for i in (1, 2)
+                    ]
+                }
+            return mock_make_api_call(self, operation_name, kwarg)
+
+        with patch(
+            "botocore.client.BaseClient._make_api_call",
+            new=compute_environments_api_call,
+        ):
+            aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+            aws_provider._audit_resources = [
+                "arn:aws:batch:eu-west-1:123456789012:compute-environment/compute-environment-1"
+            ]
+            batch = Batch(aws_provider)
+
+            assert list(batch.compute_environments) == [
+                "arn:aws:batch:eu-west-1:123456789012:compute-environment/compute-environment-1"
+            ]
+            assert batch.security_groups_in_use == {"sg-1"}
+
     def test_audit_resources_filters_job_definitions(self):
         def counting_make_api_call(self, operation_name, kwarg):
             if operation_name == "DescribeJobDefinitions":
@@ -207,7 +346,7 @@ class Test_Batch_Service:
                         for i in (1, 2)
                     ]
                 }
-            return make_api_call(self, operation_name, kwarg)
+            return default_api_call(self, operation_name, kwarg)
 
         with patch(
             "botocore.client.BaseClient._make_api_call", new=counting_make_api_call

tests/providers/aws/services/ec2/ec2_securitygroup_not_used/ec2_securitygroup_not_used_test.py

@@ -36,6 +36,10 @@ class Test_ec2_securitygroup_not_used:
         awslambda_client = mock.MagicMock()
         awslambda_client.functions = {}
         awslambda_client.security_groups_in_use = {sg_id}
+        batch_client = mock.MagicMock()
+        batch_client.compute_environments = {}
+        batch_client.security_groups_in_use = set()
+        batch_client.compute_environment_lookup_failed_regions = set()
         aws_provider = set_mocked_aws_provider()
 
         with (
@@ -51,6 +55,10 @@ class Test_ec2_securitygroup_not_used:
                 "prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used.awslambda_client",
                 new=awslambda_client,
             ),
+            mock.patch(
+                "prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used.batch_client",
+                new=batch_client,
+            ),
         ):
             from prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used import (
                 ec2_securitygroup_not_used,
@@ -72,6 +80,7 @@ class Test_ec2_securitygroup_not_used:
         ec2_client.create_vpc(CidrBlock="10.0.0.0/16")
 
         from prowler.providers.aws.services.awslambda.awslambda_service import Lambda
+        from prowler.providers.aws.services.batch.batch_service import Batch
         from prowler.providers.aws.services.ec2.ec2_service import EC2
 
         aws_provider = set_mocked_aws_provider(
@@ -91,6 +100,10 @@ class Test_ec2_securitygroup_not_used:
                 "prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used.awslambda_client",
                 new=Lambda(aws_provider),
             ),
+            mock.patch(
+                "prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used.batch_client",
+                new=Batch(aws_provider),
+            ),
         ):
             # Test Check
             from prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used import (
@@ -115,6 +128,7 @@ class Test_ec2_securitygroup_not_used:
         )
 
         from prowler.providers.aws.services.awslambda.awslambda_service import Lambda
+        from prowler.providers.aws.services.batch.batch_service import Batch
         from prowler.providers.aws.services.ec2.ec2_service import EC2
 
         aws_provider = set_mocked_aws_provider(
@@ -134,6 +148,10 @@ class Test_ec2_securitygroup_not_used:
                 "prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used.awslambda_client",
                 new=Lambda(aws_provider),
             ),
+            mock.patch(
+                "prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used.batch_client",
+                new=Batch(aws_provider),
+            ),
         ):
             # Test Check
             from prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used import (
@@ -173,6 +191,7 @@ class Test_ec2_securitygroup_not_used:
         subnet.create_network_interface(Groups=[sg.id])
 
         from prowler.providers.aws.services.awslambda.awslambda_service import Lambda
+        from prowler.providers.aws.services.batch.batch_service import Batch
         from prowler.providers.aws.services.ec2.ec2_service import EC2
 
         aws_provider = set_mocked_aws_provider(
@@ -192,6 +211,10 @@ class Test_ec2_securitygroup_not_used:
                 "prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used.awslambda_client",
                 new=Lambda(aws_provider),
             ),
+            mock.patch(
+                "prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used.batch_client",
+                new=Batch(aws_provider),
+            ),
         ):
             # Test Check
             from prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used import (
@@ -259,6 +282,7 @@ class Test_ec2_securitygroup_not_used:
         )
 
         from prowler.providers.aws.services.awslambda.awslambda_service import Lambda
+        from prowler.providers.aws.services.batch.batch_service import Batch
         from prowler.providers.aws.services.ec2.ec2_service import EC2
 
         aws_provider = set_mocked_aws_provider(
@@ -278,6 +302,10 @@ class Test_ec2_securitygroup_not_used:
                 "prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used.awslambda_client",
                 new=Lambda(aws_provider),
             ),
+            mock.patch(
+                "prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used.batch_client",
+                new=Batch(aws_provider),
+            ),
         ):
             # Test Check
             from prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used import (
@@ -338,6 +366,7 @@ class Test_ec2_securitygroup_not_used:
         )
 
         from prowler.providers.aws.services.awslambda.awslambda_service import Lambda
+        from prowler.providers.aws.services.batch.batch_service import Batch
         from prowler.providers.aws.services.ec2.ec2_service import EC2
 
         aws_provider = set_mocked_aws_provider(
@@ -357,6 +386,10 @@ class Test_ec2_securitygroup_not_used:
                 "prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used.awslambda_client",
                 new=Lambda(aws_provider),
             ),
+            mock.patch(
+                "prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used.batch_client",
+                new=Batch(aws_provider),
+            ),
         ):
             # Test Check
             from prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used import (
@@ -394,3 +427,271 @@ class Test_ec2_securitygroup_not_used:
             assert result[1].resource_id == sg1.id
             assert result[1].resource_details == sg_name1
             assert result[1].resource_tags == []
+
+    def test_ec2_sg_used_by_batch_compute_environment_without_enis(self):
+        from prowler.providers.aws.services.ec2.ec2_service import SecurityGroup
+
+        sg_id = "sg-batch"
+        sg_name = "batch-sg"
+        security_group = SecurityGroup(
+            name=sg_name,
+            region=AWS_REGION_US_EAST_1,
+            arn=f"arn:aws:ec2:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:security-group/{sg_id}",
+            id=sg_id,
+            vpc_id="vpc-test",
+            associated_sgs=[],
+            network_interfaces=[],
+            ingress_rules=[],
+            egress_rules=[],
+            tags=[],
+        )
+        ec2_client = mock.MagicMock()
+        ec2_client.security_groups = {security_group.arn: security_group}
+        awslambda_client = mock.MagicMock()
+        awslambda_client.functions = {}
+        awslambda_client.security_groups_in_use = set()
+        batch_client = mock.MagicMock()
+        batch_client.security_groups_in_use = {sg_id}
+        batch_client.compute_environment_lookup_failed_regions = set()
+        aws_provider = set_mocked_aws_provider()
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=aws_provider,
+            ),
+            mock.patch(
+                "prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used.ec2_client",
+                new=ec2_client,
+            ),
+            mock.patch(
+                "prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used.awslambda_client",
+                new=awslambda_client,
+            ),
+            mock.patch(
+                "prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used.batch_client",
+                new=batch_client,
+            ),
+        ):
+            from prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used import (
+                ec2_securitygroup_not_used,
+            )
+
+            result = ec2_securitygroup_not_used().execute()
+
+        assert len(result) == 1
+        assert result[0].status == "PASS"
+        assert (
+            result[0].status_extended
+            == f"Security group {sg_name} ({sg_id}) it is being used."
+        )
+
+    def test_ec2_sg_not_failed_when_batch_lookup_failed(self):
+        from prowler.providers.aws.services.ec2.ec2_service import SecurityGroup
+
+        sg_id = "sg-unknown"
+        sg_name = "unknown-sg"
+        security_group = SecurityGroup(
+            name=sg_name,
+            region=AWS_REGION_US_EAST_1,
+            arn=f"arn:aws:ec2:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:security-group/{sg_id}",
+            id=sg_id,
+            vpc_id="vpc-test",
+            associated_sgs=[],
+            network_interfaces=[],
+            ingress_rules=[],
+            egress_rules=[],
+            tags=[],
+        )
+        ec2_client = mock.MagicMock()
+        ec2_client.security_groups = {security_group.arn: security_group}
+        awslambda_client = mock.MagicMock()
+        awslambda_client.functions = {}
+        awslambda_client.security_groups_in_use = set()
+        # Compute environments could not be listed in this region, e.g. because
+        # batch:DescribeComputeEnvironments was denied
+        batch_client = mock.MagicMock()
+        batch_client.security_groups_in_use = set()
+        batch_client.compute_environment_lookup_failed_regions = {AWS_REGION_US_EAST_1}
+        aws_provider = set_mocked_aws_provider()
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=aws_provider,
+            ),
+            mock.patch(
+                "prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used.ec2_client",
+                new=ec2_client,
+            ),
+            mock.patch(
+                "prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used.awslambda_client",
+                new=awslambda_client,
+            ),
+            mock.patch(
+                "prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used.batch_client",
+                new=batch_client,
+            ),
+        ):
+            from prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used import (
+                ec2_securitygroup_not_used,
+            )
+
+            result = ec2_securitygroup_not_used().execute()
+
+        # Unknown associations must not be reported as unused; a lack of
+        # visibility is reported as MANUAL rather than asserted either way
+        assert len(result) == 1
+        assert result[0].status == "MANUAL"
+        assert result[0].region == AWS_REGION_US_EAST_1
+        assert result[0].resource_id == sg_id
+        assert (
+            result[0].status_extended
+            == f"Security group {sg_name} ({sg_id}) usage could not be verified because AWS Batch compute environments could not be listed in region {AWS_REGION_US_EAST_1}; grant batch:DescribeComputeEnvironments and run the check again."
+        )
+
+    def test_ec2_sg_still_reported_used_when_batch_lookup_failed(self):
+        from prowler.providers.aws.services.ec2.ec2_service import SecurityGroup
+
+        sg_id = "sg-lambda"
+        sg_name = "lambda-sg"
+        security_group = SecurityGroup(
+            name=sg_name,
+            region=AWS_REGION_US_EAST_1,
+            arn=f"arn:aws:ec2:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:security-group/{sg_id}",
+            id=sg_id,
+            vpc_id="vpc-test",
+            associated_sgs=[],
+            network_interfaces=[],
+            ingress_rules=[],
+            egress_rules=[],
+            tags=[],
+        )
+        ec2_client = mock.MagicMock()
+        ec2_client.security_groups = {security_group.arn: security_group}
+        awslambda_client = mock.MagicMock()
+        awslambda_client.functions = {}
+        awslambda_client.security_groups_in_use = {sg_id}
+        batch_client = mock.MagicMock()
+        batch_client.security_groups_in_use = set()
+        batch_client.compute_environment_lookup_failed_regions = {AWS_REGION_US_EAST_1}
+        aws_provider = set_mocked_aws_provider()
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=aws_provider,
+            ),
+            mock.patch(
+                "prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used.ec2_client",
+                new=ec2_client,
+            ),
+            mock.patch(
+                "prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used.awslambda_client",
+                new=awslambda_client,
+            ),
+            mock.patch(
+                "prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used.batch_client",
+                new=batch_client,
+            ),
+        ):
+            from prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used import (
+                ec2_securitygroup_not_used,
+            )
+
+            result = ec2_securitygroup_not_used().execute()
+
+        # A known association is still reported, the failed lookup is irrelevant
+        assert len(result) == 1
+        assert result[0].status == "PASS"
+        assert (
+            result[0].status_extended
+            == f"Security group {sg_name} ({sg_id}) it is being used."
+        )
+
+    @mock_aws
+    def test_ec2_sg_used_by_batch_compute_environment(self):
+        # Create EC2 Mocked Resources
+        ec2 = resource("ec2", AWS_REGION_US_EAST_1)
+        ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
+        vpc_id = ec2_client.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]["VpcId"]
+        sg_name = "test-sg"
+        sg = ec2.create_security_group(
+            GroupName=sg_name, Description="test", VpcId=vpc_id
+        )
+        subnet = ec2.create_subnet(VpcId=vpc_id, CidrBlock="10.0.0.0/18")
+        iam_client = client("iam", region_name=AWS_REGION_US_EAST_1)
+        service_role = iam_client.create_role(
+            RoleName="batch-service-role",
+            AssumeRolePolicyDocument="some policy",
+            Path="/my-path/",
+        )["Role"]["Arn"]
+        instance_profile = iam_client.create_instance_profile(
+            InstanceProfileName="batch-instance-profile"
+        )["InstanceProfile"]["Arn"]
+        batch = client("batch", region_name=AWS_REGION_US_EAST_1)
+        batch.create_compute_environment(
+            computeEnvironmentName="test-compute-environment",
+            type="MANAGED",
+            state="ENABLED",
+            computeResources={
+                "type": "EC2",
+                "minvCpus": 0,
+                "maxvCpus": 4,
+                "instanceTypes": ["optimal"],
+                "subnets": [subnet.id],
+                "securityGroupIds": [sg.id],
+                "instanceRole": instance_profile,
+            },
+            serviceRole=service_role,
+        )
+
+        from prowler.providers.aws.services.awslambda.awslambda_service import Lambda
+        from prowler.providers.aws.services.batch.batch_service import Batch
+        from prowler.providers.aws.services.ec2.ec2_service import EC2
+
+        aws_provider = set_mocked_aws_provider(
+            audited_regions=["us-east-1", "eu-west-1"]
+        )
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=aws_provider,
+            ),
+            mock.patch(
+                "prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used.ec2_client",
+                new=EC2(aws_provider),
+            ),
+            mock.patch(
+                "prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used.awslambda_client",
+                new=Lambda(aws_provider),
+            ),
+            mock.patch(
+                "prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used.batch_client",
+                new=Batch(aws_provider),
+            ),
+        ):
+            # Test Check
+            from prowler.providers.aws.services.ec2.ec2_securitygroup_not_used.ec2_securitygroup_not_used import (
+                ec2_securitygroup_not_used,
+            )
+
+            check = ec2_securitygroup_not_used()
+            result = check.execute()
+
+            # One custom sg, attached to a Batch compute environment with no ENIs
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert result[0].region == AWS_REGION_US_EAST_1
+            assert (
+                result[0].status_extended
+                == f"Security group {sg_name} ({sg.id}) it is being used."
+            )
+            assert (
+                result[0].resource_arn
+                == f"arn:{aws_provider.identity.partition}:ec2:{AWS_REGION_US_EAST_1}:{aws_provider.identity.account}:security-group/{sg.id}"
+            )
+            assert result[0].resource_id == sg.id
+            assert result[0].resource_details == sg_name
+            assert result[0].resource_tags == []