From 83d8cfa82994302404fc7f98ae89a70095bb9091 Mon Sep 17 00:00:00 2001 From: hackertwinten Date: Mon, 24 Aug 2026 16:18:56 +0200 Subject: [PATCH] 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 --- .../ec2-securitygroup-not-used-batch.fixed.md | 1 + .../aws/services/batch/batch_service.py | 64 +++- .../ec2_securitygroup_not_used.metadata.json | 2 +- .../ec2_securitygroup_not_used.py | 22 +- .../aws/services/batch/batch_service_test.py | 149 ++++++++- .../ec2_securitygroup_not_used_test.py | 301 ++++++++++++++++++ 6 files changed, 530 insertions(+), 9 deletions(-) create mode 100644 prowler/changelog.d/ec2-securitygroup-not-used-batch.fixed.md diff --git a/prowler/changelog.d/ec2-securitygroup-not-used-batch.fixed.md b/prowler/changelog.d/ec2-securitygroup-not-used-batch.fixed.md new file mode 100644 index 0000000000..a979e6f27f --- /dev/null +++ b/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 diff --git a/prowler/providers/aws/services/batch/batch_service.py b/prowler/providers/aws/services/batch/batch_service.py index d0110ffbaf..9902fa6175 100644 --- a/prowler/providers/aws/services/batch/batch_service.py +++ b/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 = [ diff --git a/prowler/providers/aws/services/ec2/ec2_securitygroup_not_used/ec2_securitygroup_not_used.metadata.json b/prowler/providers/aws/services/ec2/ec2_securitygroup_not_used/ec2_securitygroup_not_used.metadata.json index e417b1d2d2..f85dcf2610 100644 --- a/prowler/providers/aws/services/ec2/ec2_securitygroup_not_used/ec2_securitygroup_not_used.metadata.json +++ b/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": [ diff --git a/prowler/providers/aws/services/ec2/ec2_securitygroup_not_used/ec2_securitygroup_not_used.py b/prowler/providers/aws/services/ec2/ec2_securitygroup_not_used/ec2_securitygroup_not_used.py index c45693c498..54a0ff1f90 100644 --- a/prowler/providers/aws/services/ec2/ec2_securitygroup_not_used/ec2_securitygroup_not_used.py +++ b/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) diff --git a/tests/providers/aws/services/batch/batch_service_test.py b/tests/providers/aws/services/batch/batch_service_test.py index 7af2a64bb2..de47efd1ba 100644 --- a/tests/providers/aws/services/batch/batch_service_test.py +++ b/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 diff --git a/tests/providers/aws/services/ec2/ec2_securitygroup_not_used/ec2_securitygroup_not_used_test.py b/tests/providers/aws/services/ec2/ec2_securitygroup_not_used/ec2_securitygroup_not_used_test.py index b959f4b257..f3dac7fc3e 100644 --- a/tests/providers/aws/services/ec2/ec2_securitygroup_not_used/ec2_securitygroup_not_used_test.py +++ b/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 == []