import json import os import re import tempfile from datetime import datetime, timedelta from json import dumps from re import search from unittest import mock import botocore import botocore.exceptions import pytest from boto3 import client, resource, session from mock import patch from moto import mock_aws from pytest import raises from tzlocal import get_localzone from prowler.providers.aws.aws_provider import AwsProvider, get_aws_region_for_sts from prowler.providers.aws.config import ( AWS_STS_GLOBAL_ENDPOINT_REGION, BOTO3_USER_AGENT_EXTRA, ROLE_SESSION_NAME, ) from prowler.providers.aws.exceptions.exceptions import ( AWSArgumentTypeValidationError, AWSIAMRoleARNInvalidResourceTypeError, AWSInvalidPartitionError, AWSInvalidProviderIdError, AWSNoCredentialsError, ) from prowler.providers.aws.lib.arn.models import ARN from prowler.providers.aws.lib.mutelist.mutelist import AWSMutelist from prowler.providers.aws.models import ( AWSAssumeRoleInfo, AWSCallerIdentity, AWSCredentials, AWSMFAInfo, AWSOrganizationsInfo, ) from prowler.providers.common.models import Connection from prowler.providers.common.provider import Provider from tests.providers.aws.utils import ( AWS_ACCOUNT_ARN, AWS_ACCOUNT_NUMBER, AWS_CHINA_PARTITION, AWS_COMMERCIAL_PARTITION, AWS_EUSC_PARTITION, AWS_GOV_CLOUD_ACCOUNT_ARN, AWS_GOV_CLOUD_PARTITION, AWS_ISO_PARTITION, AWS_REGION_CN_NORTH_1, AWS_REGION_CN_NORTHWEST_1, AWS_REGION_EU_CENTRAL_1, AWS_REGION_EU_WEST_1, AWS_REGION_EUSC_DE_EAST_1, AWS_REGION_GOV_CLOUD_US_EAST_1, AWS_REGION_ISO_GLOBAL, AWS_REGION_US_EAST_1, AWS_REGION_US_EAST_2, EXAMPLE_AMI_ID, create_role, set_mocked_aws_provider, ) make_api_call = botocore.client.BaseClient._make_api_call def mock_get_caller_identity_china(self, operation_name, kwarg): if operation_name == "GetCallerIdentity": return { "UserId": "XXXXXXXXXXXXXXXXXXXXX", "Account": AWS_ACCOUNT_NUMBER, "Arn": f"arn:{AWS_CHINA_PARTITION}:iam::{AWS_ACCOUNT_NUMBER}:user/test-user", } return make_api_call(self, operation_name, kwarg) def mock_get_caller_identity_gov_cloud(self, operation_name, kwarg): if operation_name == "GetCallerIdentity": return { "UserId": "XXXXXXXXXXXXXXXXXXXXX", "Account": AWS_ACCOUNT_NUMBER, "Arn": f"arn:{AWS_GOV_CLOUD_PARTITION}:iam::{AWS_ACCOUNT_NUMBER}:user/test-user", } return make_api_call(self, operation_name, kwarg) def mock_recover_checks_from_aws_provider(*_): return [ ( "accessanalyzer_enabled_without_findings", "/root_dir/fake_path/accessanalyzer/accessanalyzer_enabled_without_findings", ), ( "awslambda_function_url_cors_policy", "/root_dir/fake_path/awslambda/awslambda_function_url_cors_policy", ), ( "ec2_securitygroup_allow_ingress_from_internet_to_any_port", "/root_dir/fake_path/ec2/ec2_securitygroup_allow_ingress_from_internet_to_any_port", ), ] def mock_recover_checks_from_aws_provider_lambda_service(*_): return [ ( "awslambda_function_invoke_api_operations_cloudtrail_logging_enabled", "/root_dir/fake_path/awslambda/awslambda_function_invoke_api_operations_cloudtrail_logging_enabled", ), ( "awslambda_function_url_cors_policy", "/root_dir/fake_path/awslambda/awslambda_function_url_cors_policy", ), ( "awslambda_function_no_secrets_in_code", "/root_dir/fake_path/awslambda/awslambda_function_no_secrets_in_code", ), ] def mock_recover_checks_from_aws_provider_elb_service(*_): return [ ( "elb_insecure_ssl_ciphers", "/root_dir/fake_path/elb/elb_insecure_ssl_ciphers", ), ( "elb_internet_facing", "/root_dir/fake_path/elb/elb_internet_facing", ), ( "elb_logging_enabled", "/root_dir/fake_path/elb/elb_logging_enabled", ), ] def mock_recover_checks_from_aws_provider_efs_service(*_): return [ ( "efs_encryption_at_rest_enabled", "/root_dir/fake_path/efs/efs_encryption_at_rest_enabled", ), ( "efs_have_backup_enabled", "/root_dir/fake_path/efs/efs_have_backup_enabled", ), ( "efs_not_publicly_accessible", "/root_dir/fake_path/efs/efs_not_publicly_accessible", ), ] def mock_recover_checks_from_aws_provider_iam_service(*_): return [ ( "iam_customer_attached_policy_no_administrative_privileges", "/root_dir/fake_path/iam/iam_customer_attached_policy_no_administrative_privileges", ), ( "iam_check_saml_providers_sts", "/root_dir/fake_path/iam/iam_check_saml_providers_sts", ), ( "iam_password_policy_minimum_length_14", "/root_dir/fake_path/iam/iam_password_policy_minimum_length_14", ), ] def mock_recover_checks_from_aws_provider_s3_service(*_): return [ ( "s3_account_level_public_access_blocks", "/root_dir/fake_path/s3/s3_account_level_public_access_blocks", ), ( "s3_bucket_acl_prohibited", "/root_dir/fake_path/s3/s3_bucket_acl_prohibited", ), ( "s3_bucket_policy_public_write_access", "/root_dir/fake_path/s3/s3_bucket_policy_public_write_access", ), ] def mock_recover_checks_from_aws_provider_cloudwatch_service(*_): return [ ( "cloudwatch_changes_to_network_acls_alarm_configured", "/root_dir/fake_path/cloudwatch/cloudwatch_changes_to_network_acls_alarm_configured", ), ( "cloudwatch_changes_to_network_gateways_alarm_configured", "/root_dir/cloudwatch/cloudwatch_changes_to_network_gateways_alarm_configured", ), ( "cloudwatch_changes_to_network_route_tables_alarm_configured", "/root_dir/fake_path/cloudwatch/cloudwatch_changes_to_network_route_tables_alarm_configured", ), ] def mock_recover_checks_from_aws_provider_ec2_service(*_): return [ ( "ec2_securitygroup_allow_ingress_from_internet_to_any_port", "/root_dir/fake_path/ec2/ec2_securitygroup_allow_ingress_from_internet_to_any_port", ), ( "ec2_networkacl_allow_ingress_any_port", "/root_dir/fake_path/ec2/ec2_networkacl_allow_ingress_any_port", ), ( "ec2_ami_public", "/root_dir/fake_path/ec2/ec2_ami_public", ), ] def mock_recover_checks_from_aws_provider_rds_service(*_): return [ ( "rds_instance_backup_enabled", "/root_dir/fake_path/rds/rds_instance_backup_enabled", ), ( "rds_instance_deletion_protection", "/root_dir/fake_path/rds/rds_instance_deletion_protection", ), ( "rds_snapshots_public_access", "/root_dir/fake_path/rds/rds_snapshots_public_access", ), ] def mock_recover_checks_from_aws_provider_cognito_service(*_): return [] def mock_recover_checks_from_aws_provider_eks_service(*_): return [ ( "eks_cluster_not_publicly_accessible", "/root_dir/fake_path/eks/eks_cluster_not_publicly_accessible", ), ( "eks_cluster_uses_a_supported_version", "/root_dir/fake_path/eks/eks_cluster_uses_a_supported_version", ), ( "eks_cluster_network_policy_enabled", "/root_dir/fake_path/eks/eks_cluster_network_policy_enabled", ), ( "eks_control_plane_logging_all_types_enabled", "/root_dir/fake_path/eks/eks_control_plane_logging_all_types_enabled", ), ] class TestAWSProvider: @mock_aws def test_aws_provider_default(self): mfa = False scan_unused_services = True aws_provider = AwsProvider( mfa=mfa, scan_unused_services=scan_unused_services, ) assert aws_provider.type == "aws" assert aws_provider.scan_unused_services is True assert aws_provider.audit_config assert aws_provider.session.current_session.region_name == AWS_REGION_US_EAST_1 @mock_aws def test_aws_provider_with_static_credentials(self): # Create a mock IAM user iam_client = client("iam", region_name=AWS_REGION_EU_WEST_1) username = "test-user" iam_user = iam_client.create_user(UserName=username)["User"] # Create a mock IAM access keys access_key = iam_client.create_access_key(UserName=iam_user["UserName"])[ "AccessKey" ] credentials = { "aws_access_key_id": access_key["AccessKeyId"], "aws_secret_access_key": access_key["SecretAccessKey"], } aws_provider = AwsProvider(**credentials) assert aws_provider.type == "aws" # Session assert aws_provider.session.current_session.region_name == AWS_REGION_US_EAST_1 assert aws_provider.session.current_session.profile_name == "default" assert aws_provider.session.original_session.region_name == AWS_REGION_US_EAST_1 assert aws_provider.session.original_session.profile_name == "default" # Identity assert aws_provider.identity.account == AWS_ACCOUNT_NUMBER assert aws_provider.identity.account_arn == AWS_ACCOUNT_ARN assert ( aws_provider.identity.identity_arn == f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:user/{username}" ) assert aws_provider.identity.partition == AWS_COMMERCIAL_PARTITION assert aws_provider.identity.profile is None assert aws_provider.identity.profile_region == AWS_REGION_US_EAST_1 @mock_aws def test_aws_provider_with_session_credentials(self): sts_client = client("sts", region_name=AWS_REGION_EU_WEST_1) session_token = sts_client.get_session_token() session_credentials = { "aws_access_key_id": session_token["Credentials"]["AccessKeyId"], "aws_secret_access_key": session_token["Credentials"]["SecretAccessKey"], "aws_session_token": session_token["Credentials"]["SessionToken"], } aws_provider = AwsProvider(**session_credentials) assert aws_provider.type == "aws" # Session assert aws_provider.session.current_session.region_name == AWS_REGION_US_EAST_1 assert aws_provider.session.current_session.profile_name == "default" assert aws_provider.session.original_session.region_name == AWS_REGION_US_EAST_1 assert aws_provider.session.original_session.profile_name == "default" # Identity assert aws_provider.identity.account == AWS_ACCOUNT_NUMBER assert aws_provider.identity.account_arn == AWS_ACCOUNT_ARN # moto is the default user created by moto assert ( aws_provider.identity.identity_arn == f"arn:aws:sts::{AWS_ACCOUNT_NUMBER}:user/moto" ) assert aws_provider.identity.partition == AWS_COMMERCIAL_PARTITION assert aws_provider.identity.profile is None assert aws_provider.identity.profile_region == AWS_REGION_US_EAST_1 @mock_aws def test_aws_provider_organizations_delegated_administrator(self): organizations_client = client("organizations", region_name=AWS_REGION_EU_WEST_1) organization = organizations_client.create_organization()["Organization"] organizations_client.tag_resource( ResourceId=AWS_ACCOUNT_NUMBER, Tags=[ {"Key": "tagged", "Value": "true"}, ], ) aws_provider = AwsProvider() assert isinstance(aws_provider.organizations_metadata, AWSOrganizationsInfo) assert aws_provider.organizations_metadata.account_email == "master@example.com" assert aws_provider.organizations_metadata.account_name == "master" assert aws_provider.organizations_metadata.account_tags == {"tagged": "true"} assert ( aws_provider.organizations_metadata.organization_account_arn == f"arn:aws:organizations::{AWS_ACCOUNT_NUMBER}:account/{organization['Id']}/{AWS_ACCOUNT_NUMBER}" ) assert aws_provider.organizations_metadata.organization_id == organization["Id"] assert ( aws_provider.organizations_metadata.organization_arn == organization["Arn"] ) @mock_aws def test_aws_provider_organizations_none_organizations_metadata(self): aws_provider = AwsProvider() assert isinstance(aws_provider.organizations_metadata, AWSOrganizationsInfo) assert aws_provider.organizations_metadata.account_email == "" assert aws_provider.organizations_metadata.account_name == "" assert aws_provider.organizations_metadata.account_tags == [] assert aws_provider.organizations_metadata.organization_account_arn == "" assert aws_provider.organizations_metadata.organization_id == "" assert aws_provider.organizations_metadata.organization_arn == "" @mock_aws def test_aws_provider_organizations_with_role(self): iam_client = client("iam", region_name=AWS_REGION_EU_WEST_1) policy_name = "describe_organizations_policy" policy_document = { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "organizations:DescribeAccount", "organizations:ListTagsForResource", ], "Resource": "*", }, ], } policy = iam_client.create_policy( PolicyName=policy_name, PolicyDocument=dumps(policy_document), )["Policy"] assume_policy_document = { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": {"AWS": f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root"}, "Action": "sts:AssumeRole", } ], } role_name = "organizations_role" organizations_role = iam_client.create_role( RoleName=role_name, AssumeRolePolicyDocument=dumps(assume_policy_document) )["Role"] iam_client.attach_role_policy( RoleName=role_name, PolicyArn=policy["Arn"], ) organizations_client = client("organizations", region_name=AWS_REGION_EU_WEST_1) organization = organizations_client.create_organization()["Organization"] organizations_client.tag_resource( ResourceId=AWS_ACCOUNT_NUMBER, Tags=[ {"Key": "tagged", "Value": "true"}, ], ) organizations_role = organizations_role["Arn"] session_duration = 900 aws_provider = AwsProvider( organizations_role_arn=organizations_role, session_duration=session_duration, ) assert isinstance(aws_provider.organizations_metadata, AWSOrganizationsInfo) assert aws_provider.organizations_metadata.account_email == "master@example.com" assert aws_provider.organizations_metadata.account_name == "master" assert aws_provider.organizations_metadata.account_tags == {"tagged": "true"} assert ( aws_provider.organizations_metadata.organization_account_arn == f"arn:aws:organizations::{AWS_ACCOUNT_NUMBER}:account/{organization['Id']}/{AWS_ACCOUNT_NUMBER}" ) assert aws_provider.organizations_metadata.organization_id == organization["Id"] assert ( aws_provider.organizations_metadata.organization_arn == organization["Arn"] ) @mock_aws def test_aws_provider_session_with_mfa(self): mfa = True with patch( "prowler.providers.aws.aws_provider.AwsProvider.input_role_mfa_token_and_code", return_value=AWSMFAInfo( arn=f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:mfa/test-role-mfa", totp="111111", ), ): aws_provider = AwsProvider(mfa=mfa) assert aws_provider.type == "aws" assert aws_provider.scan_unused_services is False assert aws_provider.audit_config != {} assert ( aws_provider.session.current_session.region_name == AWS_REGION_US_EAST_1 ) assert ( aws_provider.session.current_session.region_name == AWS_REGION_US_EAST_1 ) @mock_aws def test_aws_provider_assume_role_with_mfa(self): # Variables mfa = True role_name = "test-role" role_arn = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/{role_name}" session_duration = 900 role_session_name = "ProwlerAssessmentSession" external_id = "test-external-id" with patch( "prowler.providers.aws.aws_provider.AwsProvider.input_role_mfa_token_and_code", return_value=AWSMFAInfo( arn=f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:mfa/test-role-mfa", totp="111111", ), ): aws_provider = AwsProvider( mfa=mfa, role_arn=role_arn, session_duration=session_duration, role_session_name=role_session_name, external_id=external_id, ) assert ( aws_provider.session.current_session.region_name == AWS_REGION_US_EAST_1 ) assert aws_provider.identity.account == AWS_ACCOUNT_NUMBER assert aws_provider.identity.account_arn == AWS_ACCOUNT_ARN assert aws_provider.identity.partition == AWS_COMMERCIAL_PARTITION assert isinstance( aws_provider._assumed_role_configuration.info, AWSAssumeRoleInfo ) assert aws_provider._assumed_role_configuration.info == AWSAssumeRoleInfo( role_arn=ARN(arn=role_arn), session_duration=session_duration, external_id=external_id, mfa_enabled=True, # <- MFA configuration role_session_name=role_session_name, sts_region=AWS_REGION_US_EAST_1, ) credentials = aws_provider._assumed_role_configuration.credentials assert isinstance(credentials, AWSCredentials) assert credentials.aws_access_key_id assert len(credentials.aws_access_key_id) == 20 assert search(r"^ASIA.*$", credentials.aws_access_key_id) assert credentials.aws_session_token assert len(credentials.aws_session_token) == 356 assert search(r"^FQoGZXIvYXdzE.*$", credentials.aws_session_token) assert credentials.aws_secret_access_key assert len(credentials.aws_secret_access_key) == 40 assert credentials.expiration # assert credentials.expiration == datetime.now(tzinfo=tzutc()) @mock_aws def test_aws_provider_assume_role_without_mfa(self): # Variables mfa = False role_name = "test-role" role_arn = ( f"arn:{AWS_COMMERCIAL_PARTITION}:iam::{AWS_ACCOUNT_NUMBER}:role/{role_name}" ) session_duration = 900 role_session_name = "ProwlerAssessmentSession" aws_provider = AwsProvider( mfa=mfa, role_arn=role_arn, session_duration=session_duration, role_session_name=role_session_name, ) assert aws_provider.session.current_session.region_name == AWS_REGION_US_EAST_1 assert aws_provider.identity.account == AWS_ACCOUNT_NUMBER assert aws_provider.identity.account_arn == AWS_ACCOUNT_ARN assert aws_provider.identity.partition == AWS_COMMERCIAL_PARTITION assert isinstance( aws_provider._assumed_role_configuration.info, AWSAssumeRoleInfo ) assert aws_provider._assumed_role_configuration.info == AWSAssumeRoleInfo( role_arn=ARN(arn=role_arn), session_duration=session_duration, external_id=None, mfa_enabled=False, # <- MFA configuration role_session_name=role_session_name, sts_region=AWS_REGION_US_EAST_1, ) credentials = aws_provider._assumed_role_configuration.credentials assert isinstance(credentials, AWSCredentials) assert credentials.aws_access_key_id assert len(credentials.aws_access_key_id) == 20 assert search(r"^ASIA.*$", credentials.aws_access_key_id) assert credentials.aws_session_token assert len(credentials.aws_session_token) == 356 assert search(r"^FQoGZXIvYXdzE.*$", credentials.aws_session_token) assert credentials.aws_secret_access_key assert len(credentials.aws_secret_access_key) == 40 assert credentials.expiration # assert credentials.expiration == datetime.now(tzinfo=tzutc()) @mock_aws def test_aws_provider_assume_role_without_mfa_gov_cloud(self, monkeypatch): # Set AWS_DEFAULT_REGION = 'us-gov-east-1' since is set by default to 'us-east-1 monkeypatch.setenv("AWS_DEFAULT_REGION", AWS_REGION_GOV_CLOUD_US_EAST_1) # Variables mfa = False role_name = "test-role" role_arn = ( f"arn:{AWS_GOV_CLOUD_PARTITION}:iam::{AWS_ACCOUNT_NUMBER}:role/{role_name}" ) session_duration = 900 role_session_name = "ProwlerAssessmentSession" aws_provider = AwsProvider( mfa=mfa, role_arn=role_arn, session_duration=session_duration, role_session_name=role_session_name, ) assert ( aws_provider.session.current_session.region_name == AWS_REGION_GOV_CLOUD_US_EAST_1 ) assert aws_provider.identity.account == AWS_ACCOUNT_NUMBER assert aws_provider.identity.account_arn == AWS_GOV_CLOUD_ACCOUNT_ARN assert aws_provider.identity.partition == AWS_GOV_CLOUD_PARTITION assert isinstance( aws_provider._assumed_role_configuration.info, AWSAssumeRoleInfo ) assert aws_provider._assumed_role_configuration.info == AWSAssumeRoleInfo( role_arn=ARN(arn=role_arn), session_duration=session_duration, external_id=None, mfa_enabled=False, # <- MFA configuration role_session_name=role_session_name, sts_region=AWS_REGION_GOV_CLOUD_US_EAST_1, ) credentials = aws_provider._assumed_role_configuration.credentials assert isinstance(credentials, AWSCredentials) assert credentials.aws_access_key_id assert len(credentials.aws_access_key_id) == 20 assert search(r"^ASIA.*$", credentials.aws_access_key_id) assert credentials.aws_session_token assert len(credentials.aws_session_token) == 356 assert search(r"^FQoGZXIvYXdzE.*$", credentials.aws_session_token) assert credentials.aws_secret_access_key assert len(credentials.aws_secret_access_key) == 40 assert credentials.expiration # assert credentials.expiration == datetime.now(tzinfo=tzutc()) @mock_aws def test_aws_provider_config(self): config = """ aws: test_key: value""" config_file_input = tempfile.NamedTemporaryFile(delete=False) config_file_input.write(bytes(config, encoding="raw_unicode_escape")) config_file_input.close() config_file_input = config_file_input.name aws_provider = AwsProvider( config_path=config_file_input, ) os.remove(config_file_input) assert aws_provider.audit_config == {"test_key": "value"} @mock_aws def test_aws_provider_mutelist(self): mutelist = { "Mutelist": { "Accounts": { AWS_ACCOUNT_NUMBER: { "Checks": { "test-check": { "Regions": [], "Resources": [], "Tags": [], "Exceptions": { "Accounts": [], "Regions": [], "Resources": [], "Tags": [], }, } } } } } } mutelist_file = tempfile.NamedTemporaryFile(delete=False) with open(mutelist_file.name, "w") as mutelist_file: mutelist_file.write(json.dumps(mutelist, indent=4)) aws_provider = AwsProvider(mutelist_path=mutelist_file.name) os.remove(mutelist_file.name) assert isinstance(aws_provider.mutelist, AWSMutelist) assert aws_provider.mutelist.mutelist == mutelist["Mutelist"] assert aws_provider.mutelist.mutelist_file_path == mutelist_file.name @mock_aws def test_aws_provider_mutelist_none(self): with patch( "prowler.providers.aws.aws_provider.get_default_mute_file_path", return_value=None, ): aws_provider = AwsProvider(mutelist_path=None) assert isinstance(aws_provider.mutelist, AWSMutelist) assert aws_provider.mutelist.mutelist == {} assert aws_provider.mutelist.mutelist_file_path is None @mock_aws def test_aws_provider_mutelist_s3(self): # Create mutelist temp file mutelist = { "Mutelist": { "Accounts": { AWS_ACCOUNT_NUMBER: { "Checks": { "test-check": { "Regions": [], "Resources": [], "Tags": [], "Exceptions": { "Accounts": [], "Regions": [], "Resources": [], "Tags": [], }, } } } } } } mutelist_file = tempfile.NamedTemporaryFile(delete=False) with open(mutelist_file.name, "w") as mutelist_file: mutelist_file.write(json.dumps(mutelist, indent=4)) # Create bucket and upload mutelist yaml s3_resource = resource("s3", region_name=AWS_REGION_US_EAST_1) bucket_name = "test-mutelist" mutelist_file_name = "mutelist.yaml" mutelist_bucket_object_uri = f"s3://{bucket_name}/{mutelist_file_name}" s3_resource.create_bucket(Bucket=bucket_name) s3_resource.Object(bucket_name, "mutelist.yaml").put( Body=open( mutelist_file.name, "rb", ) ) aws_provider = AwsProvider(mutelist_path=mutelist_bucket_object_uri) os.remove(mutelist_file.name) assert isinstance(aws_provider.mutelist, AWSMutelist) assert aws_provider.mutelist.mutelist == mutelist["Mutelist"] assert aws_provider.mutelist.mutelist_file_path == mutelist_bucket_object_uri @mock_aws def test_aws_provider_mutelist_lambda(self): # Create mutelist temp file mutelist = { "Mutelist": { "Accounts": { AWS_ACCOUNT_NUMBER: { "Checks": { "test-check": { "Regions": [], "Resources": [], "Tags": [], "Exceptions": { "Accounts": [], "Regions": [], "Resources": [], "Tags": [], }, } } } } } } lambda_mutelist_path = f"arn:aws:lambda:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:function:lambda-mutelist" aws_provider = AwsProvider() with patch( "prowler.providers.aws.lib.mutelist.mutelist.AWSMutelist.get_mutelist_file_from_lambda", return_value=mutelist["Mutelist"], ): aws_provider = AwsProvider(mutelist_path=lambda_mutelist_path) assert isinstance(aws_provider.mutelist, AWSMutelist) assert aws_provider.mutelist.mutelist == mutelist["Mutelist"] assert aws_provider.mutelist.mutelist_file_path == lambda_mutelist_path @mock_aws def test_aws_provider_mutelist_dynamodb(self): # Create mutelist temp file mutelist = { "Mutelist": { "Accounts": { AWS_ACCOUNT_NUMBER: { "Checks": { "test-check": { "Regions": [], "Resources": [], "Tags": [], "Exceptions": { "Accounts": [], "Regions": [], "Resources": [], "Tags": [], }, } } } } } } dynamodb_mutelist_path = f"arn:aws:dynamodb:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:table/mutelist-dynamo" aws_provider = AwsProvider() with patch( "prowler.providers.aws.lib.mutelist.mutelist.AWSMutelist.get_mutelist_file_from_dynamodb", return_value=mutelist["Mutelist"], ): aws_provider = AwsProvider(mutelist_path=dynamodb_mutelist_path) assert isinstance(aws_provider.mutelist, AWSMutelist) assert aws_provider.mutelist.mutelist == mutelist["Mutelist"] assert aws_provider.mutelist.mutelist_file_path == dynamodb_mutelist_path @mock_aws def test_empty_input_regions_in_arguments(self): aws_provider = AwsProvider(regions=None) assert isinstance(aws_provider, AwsProvider) @mock_aws def test_generate_regional_clients_all_enabled_regions(self): aws_provider = AwsProvider() response = aws_provider.generate_regional_clients("ec2") assert len(response.keys()) == 33 @mock_aws def test_generate_regional_clients_with_enabled_regions(self): aws_provider = AwsProvider() enabled_regions = [AWS_REGION_EU_WEST_1] aws_provider._enabled_regions = enabled_regions response = aws_provider.generate_regional_clients("ec2") assert list(response.keys()) == enabled_regions @mock_aws def test_generate_regional_clients_with_enabled_regions_and_input_regions(self): region = [AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1] aws_provider = AwsProvider( regions=region, ) enabled_regions = [AWS_REGION_EU_WEST_1] aws_provider._enabled_regions = enabled_regions response = aws_provider.generate_regional_clients("ec2") assert list(response.keys()) == enabled_regions @mock_aws def test_generate_regional_clients_cn_partition(self): region = [AWS_REGION_CN_NORTH_1, AWS_REGION_CN_NORTHWEST_1] aws_provider = AwsProvider( regions=region, ) response = aws_provider.generate_regional_clients("ec2") assert AWS_REGION_CN_NORTH_1 in response.keys() assert AWS_REGION_CN_NORTHWEST_1 in response.keys() @mock_aws def test_generate_regional_clients_cn_partition_not_present_service(self): region = ["cn-northwest-1", "cn-north-1"] aws_provider = AwsProvider( regions=region, ) response = aws_provider.generate_regional_clients("shield") assert response == {} @mock_aws def test_get_default_region(self): region = [AWS_REGION_EU_WEST_1] aws_provider = AwsProvider( regions=region, ) aws_provider._identity.profile_region = AWS_REGION_EU_WEST_1 assert aws_provider.get_default_region("ec2") == AWS_REGION_EU_WEST_1 @mock_aws def test_get_default_region_profile_region_not_audited(self): region = [AWS_REGION_EU_WEST_1] aws_provider = AwsProvider( regions=region, ) aws_provider._identity.profile_region = AWS_REGION_US_EAST_2 assert aws_provider.get_default_region("ec2") == AWS_REGION_EU_WEST_1 @mock_aws def test_get_default_region_non_profile_region(self): region = [AWS_REGION_EU_WEST_1] aws_provider = AwsProvider( regions=region, ) aws_provider._identity.profile_region = None assert aws_provider.get_default_region("ec2") == AWS_REGION_EU_WEST_1 @mock_aws def test_get_default_region_non_profile_or_audited_region(self): aws_provider = AwsProvider() aws_provider._identity.profile_region = None assert aws_provider.get_default_region("ec2") == AWS_REGION_US_EAST_1 @mock_aws def test_get_default_region_profile_region_not_present_in_service(self): region = [AWS_REGION_EU_WEST_1] aws_provider = AwsProvider( regions=region, ) aws_provider._identity.profile_region = "non-existent-region" assert aws_provider.get_default_region("ec2") == AWS_REGION_EU_WEST_1 @mock_aws def test_aws_gov_get_global_region(self): aws_provider = AwsProvider() aws_provider._identity.partition = AWS_GOV_CLOUD_PARTITION assert aws_provider.get_global_region() == AWS_REGION_GOV_CLOUD_US_EAST_1 @mock_aws def test_aws_cn_get_global_region(self): aws_provider = AwsProvider() aws_provider._identity.partition = AWS_CHINA_PARTITION assert aws_provider.get_global_region() == AWS_REGION_CN_NORTH_1 @mock_aws def test_aws_iso_get_global_region(self): aws_provider = AwsProvider() aws_provider._identity.partition = AWS_ISO_PARTITION assert aws_provider.get_global_region() == AWS_REGION_ISO_GLOBAL @mock_aws def test_aws_eusc_get_global_region(self): aws_provider = AwsProvider() aws_provider._identity.partition = AWS_EUSC_PARTITION assert aws_provider.get_global_region() == AWS_REGION_EUSC_DE_EAST_1 @mock_aws def test_get_available_aws_service_regions_with_us_east_1_audited(self): region = [AWS_REGION_US_EAST_1] aws_provider = AwsProvider( regions=region, ) with patch( "prowler.providers.aws.aws_provider.parse_json_file", return_value={ "services": { "ec2": { "regions": { "aws": [ "af-south-1", "ca-central-1", "eu-central-1", "eu-central-2", "eu-north-1", "eu-south-1", "eu-south-2", AWS_REGION_EU_WEST_1, "eu-west-2", "eu-west-3", "me-central-1", "me-south-1", "sa-east-1", AWS_REGION_US_EAST_1, "us-east-2", "us-west-1", "us-west-2", ], } } } }, ): assert aws_provider.get_available_aws_service_regions( "ec2", "aws", {AWS_REGION_US_EAST_1} ) == {AWS_REGION_US_EAST_1} @mock_aws def test_get_available_aws_service_regions_with_all_regions_audited(self): aws_provider = AwsProvider() with patch( "prowler.providers.aws.aws_provider.parse_json_file", return_value={ "services": { "ec2": { "regions": { "aws": [ "af-south-1", "ca-central-1", "eu-central-1", "eu-central-2", "eu-north-1", "eu-south-1", "eu-south-2", AWS_REGION_EU_WEST_1, "eu-west-2", "eu-west-3", "me-central-1", "me-south-1", "sa-east-1", AWS_REGION_US_EAST_1, "us-east-2", "us-west-1", "us-west-2", ], } } } }, ): assert ( len(aws_provider.get_available_aws_service_regions("ec2", "aws")) == 17 ) @mock_aws def test_get_tagged_resources(self): ec2_client = client("ec2", region_name=AWS_REGION_EU_CENTRAL_1) instances = ec2_client.run_instances( ImageId=EXAMPLE_AMI_ID, MinCount=1, MaxCount=1, InstanceType="t2.micro", TagSpecifications=[ { "ResourceType": "instance", "Tags": [ {"Key": "MY_TAG1", "Value": "MY_VALUE1"}, {"Key": "MY_TAG2", "Value": "MY_VALUE2"}, ], }, { "ResourceType": "instance", "Tags": [{"Key": "ami", "Value": "test"}], }, ], ) instance_id = instances["Instances"][0]["InstanceId"] instance_arn = f"arn:aws:ec2:{AWS_REGION_EU_CENTRAL_1}:{AWS_ACCOUNT_NUMBER}:ec2:instance/{instance_id}" image_id = ec2_client.create_image(Name="testami", InstanceId=instance_id)[ "ImageId" ] image_arn = f"arn:aws:ec2:{AWS_REGION_EU_CENTRAL_1}:{AWS_ACCOUNT_NUMBER}:ec2:image/{image_id}" ec2_client.create_tags( Resources=[image_id], Tags=[{"Key": "ami", "Value": "test"}] ) # Through the AWS provider region = [AWS_REGION_EU_CENTRAL_1] resource_tags = ["ami=test"] aws_provider = AwsProvider( regions=region, resource_tags=resource_tags, ) tagged_resources = aws_provider.audit_resources assert len(tagged_resources) == 2 assert image_arn in tagged_resources assert instance_arn in tagged_resources # Calling directly the function tagged_resources = aws_provider.get_tagged_resources(["MY_TAG1=MY_VALUE1"]) assert len(tagged_resources) == 1 assert instance_arn in tagged_resources @mock_aws def test_aws_provider_resource_tags(self): resource_arn = [AWS_ACCOUNT_ARN] aws_provider = AwsProvider( resource_arn=resource_arn, ) assert aws_provider.audit_resources == [AWS_ACCOUNT_ARN] @mock_aws def test_validate_credentials_commercial_partition_with_regions(self): # Create a mock IAM user iam_client = client("iam", region_name=AWS_REGION_EU_WEST_1) iam_user = iam_client.create_user(UserName="test-user")["User"] # Create a mock IAM access keys access_key = iam_client.create_access_key(UserName=iam_user["UserName"])[ "AccessKey" ] access_key_id = access_key["AccessKeyId"] secret_access_key = access_key["SecretAccessKey"] # Create AWS session to validate current_session = session.Session( aws_access_key_id=access_key_id, aws_secret_access_key=secret_access_key, region_name=AWS_REGION_EU_WEST_1, ) get_caller_identity = AwsProvider.validate_credentials( session=current_session, aws_region=AWS_REGION_EU_WEST_1 ) assert isinstance(get_caller_identity, AWSCallerIdentity) assert re.match("[0-9a-zA-Z]{20}", get_caller_identity.user_id) assert get_caller_identity.account == AWS_ACCOUNT_NUMBER assert get_caller_identity.region == AWS_REGION_EU_WEST_1 assert isinstance(get_caller_identity.arn, ARN) assert get_caller_identity.arn.partition == AWS_COMMERCIAL_PARTITION assert get_caller_identity.arn.region is None assert get_caller_identity.arn.resource == "test-user" assert get_caller_identity.arn.resource_type == "user" @mock_aws @patch( "botocore.client.BaseClient._make_api_call", new=mock_get_caller_identity_china ) def test_validate_credentials_china_partition(self): # Create a mock IAM user iam_client = client("iam", region_name=AWS_REGION_CN_NORTH_1) iam_user = iam_client.create_user(UserName="test-user")["User"] # Create a mock IAM access keys access_key = iam_client.create_access_key(UserName=iam_user["UserName"])[ "AccessKey" ] access_key_id = access_key["AccessKeyId"] secret_access_key = access_key["SecretAccessKey"] # Create AWS session to validate current_session = session.Session( aws_access_key_id=access_key_id, aws_secret_access_key=secret_access_key, region_name=AWS_REGION_CN_NORTH_1, ) # To use GovCloud or China it is either required: # - Set the AWS profile region with a valid partition region # - Use the -f/--region with a valid partition region get_caller_identity = AwsProvider.validate_credentials( session=current_session, aws_region=AWS_REGION_CN_NORTH_1 ) assert isinstance(get_caller_identity, AWSCallerIdentity) assert re.match("[0-9a-zA-Z]{20}", get_caller_identity.user_id) assert get_caller_identity.account == AWS_ACCOUNT_NUMBER assert get_caller_identity.region == AWS_REGION_CN_NORTH_1 assert isinstance(get_caller_identity.arn, ARN) assert get_caller_identity.arn.partition == AWS_CHINA_PARTITION assert get_caller_identity.arn.region is None assert get_caller_identity.arn.resource == "test-user" assert get_caller_identity.arn.resource_type == "user" @mock_aws @patch( "botocore.client.BaseClient._make_api_call", new=mock_get_caller_identity_gov_cloud, ) def test_validate_credentials_gov_cloud_partition(self): # Create a mock IAM user iam_client = client("iam", region_name=AWS_REGION_GOV_CLOUD_US_EAST_1) iam_user = iam_client.create_user(UserName="test-user")["User"] # Create a mock IAM access keys access_key = iam_client.create_access_key(UserName=iam_user["UserName"])[ "AccessKey" ] access_key_id = access_key["AccessKeyId"] secret_access_key = access_key["SecretAccessKey"] # Create AWS session to validate current_session = session.Session( aws_access_key_id=access_key_id, aws_secret_access_key=secret_access_key, region_name=AWS_REGION_GOV_CLOUD_US_EAST_1, ) # To use GovCloud or China it is either required: # - Set the AWS profile region with a valid partition region # - Use the -f/--region with a valid partition region get_caller_identity = AwsProvider.validate_credentials( session=current_session, aws_region=AWS_REGION_GOV_CLOUD_US_EAST_1 ) assert isinstance(get_caller_identity, AWSCallerIdentity) assert re.match("[0-9a-zA-Z]{20}", get_caller_identity.user_id) assert get_caller_identity.account == AWS_ACCOUNT_NUMBER assert get_caller_identity.region == AWS_REGION_GOV_CLOUD_US_EAST_1 assert isinstance(get_caller_identity.arn, ARN) assert get_caller_identity.arn.partition == AWS_GOV_CLOUD_PARTITION assert get_caller_identity.arn.region is None assert get_caller_identity.arn.resource == "test-user" assert get_caller_identity.arn.resource_type == "user" @mock_aws def test_test_connection_with_env_credentials(self, monkeypatch): # Create a mock IAM user iam_client = client("iam", region_name=AWS_REGION_US_EAST_1) iam_user = iam_client.create_user(UserName="test-user")["User"] # Create a mock IAM access keys access_key = iam_client.create_access_key(UserName=iam_user["UserName"])[ "AccessKey" ] monkeypatch.delenv("AWS_ACCESS_KEY_ID") monkeypatch.delenv("AWS_SECRET_ACCESS_KEY") monkeypatch.setenv("AWS_ACCESS_KEY_ID", access_key["AccessKeyId"]) monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", access_key["SecretAccessKey"]) connection = AwsProvider.test_connection() assert isinstance(connection, Connection) assert connection.is_connected assert connection.error is None def test_test_connection_without_credentials(self): with ( mock.patch("boto3.Session.get_credentials", return_value=None), mock.patch("botocore.session.Session.get_scoped_config", return_value={}), mock.patch("botocore.credentials.EnvProvider.load", return_value=None), mock.patch( "botocore.credentials.SharedCredentialProvider.load", return_value=None ), mock.patch( "botocore.credentials.InstanceMetadataProvider.load", return_value=None ), mock.patch.dict( "os.environ", { "AWS_ACCESS_KEY_ID": "", "AWS_SECRET_ACCESS_KEY": "", "AWS_SESSION_TOKEN": "", "AWS_PROFILE": "", }, clear=True, ), ): with raises(AWSNoCredentialsError) as exception: AwsProvider.test_connection( profile=None ) # No profile to avoid ProfileNotFound error assert exception.type == AWSNoCredentialsError assert "AWSNoCredentialsError[1002]: No AWS credentials found" in str( exception.value ) @mock_aws def test_test_connection_with_role_from_env(self, monkeypatch): # Create a mock IAM user iam_client = client("iam", region_name=AWS_REGION_US_EAST_1) iam_user = iam_client.create_user(UserName="test-user")["User"] # Create a mock IAM access keys access_key = iam_client.create_access_key(UserName=iam_user["UserName"])[ "AccessKey" ] monkeypatch.setenv("AWS_ACCESS_KEY_ID", access_key["AccessKeyId"]) monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", access_key["SecretAccessKey"]) role_name = "test-role" role_arn = ( f"arn:{AWS_COMMERCIAL_PARTITION}:iam::{AWS_ACCOUNT_NUMBER}:role/{role_name}" ) connection = AwsProvider.test_connection( role_arn=role_arn, role_session_name=ROLE_SESSION_NAME ) assert isinstance(connection, Connection) assert connection.is_connected assert connection.error is None @mock_aws def test_test_connection_with_role_from_env_invalid_session_duration(self): role_name = "test-role" role_arn = ( f"arn:{AWS_COMMERCIAL_PARTITION}:iam::{AWS_ACCOUNT_NUMBER}:role/{role_name}" ) with raises(AWSArgumentTypeValidationError) as exception: AwsProvider.test_connection(role_arn=role_arn, session_duration=899) assert exception.type == AWSArgumentTypeValidationError assert ( exception.value.args[0] == "[1003] Session Duration must be between 900 and 43200 seconds." ) @mock_aws def test_test_connection_with_role_from_env_invalid_session_duration_not_raise( self, ): role_name = "test-role" role_arn = ( f"arn:{AWS_COMMERCIAL_PARTITION}:iam::{AWS_ACCOUNT_NUMBER}:role/{role_name}" ) connection = AwsProvider.test_connection( role_arn=role_arn, session_duration=899, raise_on_exception=False ) assert isinstance(connection, Connection) assert not connection.is_connected assert isinstance(connection.error, AWSArgumentTypeValidationError) assert ( connection.error.args[0] == "[1003] Session Duration must be between 900 and 43200 seconds." ) @mock_aws def test_test_connection_with_role_from_env_invalid_session_name(self): role_name = "test-role" role_arn = ( f"arn:{AWS_COMMERCIAL_PARTITION}:iam::{AWS_ACCOUNT_NUMBER}:role/{role_name}" ) with raises(AWSArgumentTypeValidationError) as exception: AwsProvider.test_connection(role_arn=role_arn, role_session_name="???") assert exception.type == AWSArgumentTypeValidationError assert ( exception.value.args[0] == "[1003] Role Session Name must be between 2 and 64 characters and may contain alphanumeric characters, periods, hyphens, and underscores." ) @mock_aws def test_test_connection_with_role_from_env_invalid_role_arn(self): role_name = "test-role" role_arn = f"arn:{AWS_COMMERCIAL_PARTITION}:iam::{AWS_ACCOUNT_NUMBER}:not-role/{role_name}" with raises(AWSIAMRoleARNInvalidResourceTypeError) as exception: AwsProvider.test_connection(role_arn=role_arn) assert exception.type == AWSIAMRoleARNInvalidResourceTypeError assert ( exception.value.args[0] == "[1010] AWS IAM Role ARN resource type is invalid" ) @mock_aws def test_test_connection_with_static_credentials(self): # Create a mock IAM user iam_client = client("iam", region_name=AWS_REGION_EU_WEST_1) username = "test-user" iam_user = iam_client.create_user(UserName=username)["User"] # Create a mock IAM access keys access_key = iam_client.create_access_key(UserName=iam_user["UserName"])[ "AccessKey" ] credentials = { "aws_access_key_id": access_key["AccessKeyId"], "aws_secret_access_key": access_key["SecretAccessKey"], } connection = AwsProvider.test_connection(**credentials) assert isinstance(connection, Connection) assert connection.is_connected assert connection.error is None @mock_aws def test_test_connection_with_session_credentials(self): sts_client = client("sts", region_name=AWS_REGION_EU_WEST_1) session_token = sts_client.get_session_token() session_credentials = { "aws_access_key_id": session_token["Credentials"]["AccessKeyId"], "aws_secret_access_key": session_token["Credentials"]["SecretAccessKey"], "aws_session_token": session_token["Credentials"]["SessionToken"], } connection = AwsProvider.test_connection(**session_credentials) assert isinstance(connection, Connection) assert connection.is_connected assert connection.error is None @mock_aws def test_test_connection_with_own_account(self): sts_client = client("sts", region_name=AWS_REGION_EU_WEST_1) session_token = sts_client.get_session_token() session_credentials = { "aws_access_key_id": session_token["Credentials"]["AccessKeyId"], "aws_secret_access_key": session_token["Credentials"]["SecretAccessKey"], "aws_session_token": session_token["Credentials"]["SessionToken"], "provider_id": AWS_ACCOUNT_NUMBER, } connection = AwsProvider.test_connection(**session_credentials) assert isinstance(connection, Connection) assert connection.is_connected assert connection.error is None @mock_aws def test_test_connection_with_different_account(self): sts_client = client("sts", region_name=AWS_REGION_EU_WEST_1) session_token = sts_client.get_session_token() session_credentials = { "aws_access_key_id": session_token["Credentials"]["AccessKeyId"], "aws_secret_access_key": session_token["Credentials"]["SecretAccessKey"], "aws_session_token": session_token["Credentials"]["SessionToken"], "provider_id": "111122223333", } with raises(AWSInvalidProviderIdError) as exception: AwsProvider.test_connection(**session_credentials) assert exception.type == AWSInvalidProviderIdError assert ( exception.value.args[0] == "[1015] The provided AWS credentials belong to a different account" ) @mock_aws def test_test_connection_with_different_account_dont_raise(self): sts_client = client("sts", region_name=AWS_REGION_EU_WEST_1) session_token = sts_client.get_session_token() session_credentials = { "aws_access_key_id": session_token["Credentials"]["AccessKeyId"], "aws_secret_access_key": session_token["Credentials"]["SecretAccessKey"], "aws_session_token": session_token["Credentials"]["SessionToken"], "provider_id": "111122223333", } connection = AwsProvider.test_connection( **session_credentials, raise_on_exception=False ) assert isinstance(connection, Connection) assert not connection.is_connected assert isinstance(connection.error, AWSInvalidProviderIdError) assert ( connection.error.message == "The provided AWS credentials belong to a different account" ) assert connection.error.code == 1015 @mock_aws def test_test_connection_generic_exception(self): with patch( "prowler.providers.aws.aws_provider.AwsProvider.setup_session", side_effect=Exception(), ): connection = AwsProvider.test_connection(raise_on_exception=False) assert isinstance(connection, Connection) assert not connection.is_connected assert isinstance(connection.error, Exception) @mock_aws def test_create_sts_session(self): current_session = session.Session() aws_region = AWS_REGION_US_EAST_1 sts_session = AwsProvider.create_sts_session(current_session, aws_region) assert sts_session._service_model.service_name == "sts" assert sts_session._client_config.region_name == aws_region assert sts_session._endpoint._endpoint_prefix == "sts" assert sts_session._endpoint.host == f"https://sts.{aws_region}.amazonaws.com" @mock_aws def test_create_sts_session_gov_cloud(self): current_session = session.Session() aws_region = AWS_REGION_GOV_CLOUD_US_EAST_1 sts_session = AwsProvider.create_sts_session(current_session, aws_region) assert sts_session._service_model.service_name == "sts" assert sts_session._client_config.region_name == aws_region assert sts_session._endpoint._endpoint_prefix == "sts" assert sts_session._endpoint.host == f"https://sts.{aws_region}.amazonaws.com" @mock_aws def test_create_sts_session_china(self): current_session = session.Session() aws_region = AWS_REGION_CN_NORTH_1 sts_session = AwsProvider.create_sts_session(current_session, aws_region) assert sts_session._service_model.service_name == "sts" assert sts_session._client_config.region_name == aws_region assert sts_session._endpoint._endpoint_prefix == "sts" assert ( sts_session._endpoint.host == f"https://sts.{aws_region}.amazonaws.com.cn" ) @mock_aws def test_create_sts_session_eusc(self): current_session = session.Session() aws_region = AWS_REGION_EUSC_DE_EAST_1 sts_session = AwsProvider.create_sts_session(current_session, aws_region) assert sts_session._service_model.service_name == "sts" assert sts_session._client_config.region_name == aws_region assert sts_session._endpoint._endpoint_prefix == "sts" assert sts_session._endpoint.host == f"https://sts.{aws_region}.amazonaws.eu" @mock_aws @patch( "prowler.lib.check.utils.recover_checks_from_provider", new=mock_recover_checks_from_aws_provider_elb_service, ) def test_get_checks_from_input_arn_elb(self): expected_checks = [ "elb_insecure_ssl_ciphers", "elb_internet_facing", "elb_logging_enabled", ] aws_provider = AwsProvider() aws_provider._audit_resources = [ f"arn:aws:elasticloadbalancing:us-east-1:{AWS_ACCOUNT_NUMBER}:loadbalancer/test" ] recovered_checks = aws_provider.get_checks_from_input_arn() assert recovered_checks == expected_checks @mock_aws @patch( "prowler.lib.check.utils.recover_checks_from_provider", new=mock_recover_checks_from_aws_provider_efs_service, ) def test_get_checks_from_input_arn_efs(self): expected_checks = [ "efs_encryption_at_rest_enabled", "efs_have_backup_enabled", "efs_not_publicly_accessible", ] aws_provider = AwsProvider() aws_provider._audit_resources = [ f"arn:aws:elasticfilesystem:us-east-1:{AWS_ACCOUNT_NUMBER}:file-system/fs-01234567" ] recovered_checks = aws_provider.get_checks_from_input_arn() assert recovered_checks == expected_checks @mock_aws @patch( "prowler.lib.check.utils.recover_checks_from_provider", new=mock_recover_checks_from_aws_provider_lambda_service, ) def test_get_checks_from_input_arn_lambda(self): expected_checks = [ "awslambda_function_invoke_api_operations_cloudtrail_logging_enabled", "awslambda_function_no_secrets_in_code", "awslambda_function_url_cors_policy", ] aws_provider = AwsProvider() aws_provider._audit_resources = [ "arn:aws:lambda:us-east-1:123456789:function:test-lambda" ] recovered_checks = aws_provider.get_checks_from_input_arn() assert recovered_checks == expected_checks @mock_aws @patch( "prowler.lib.check.utils.recover_checks_from_provider", new=mock_recover_checks_from_aws_provider_iam_service, ) def test_get_checks_from_input_arn_iam(self): expected_checks = [ "iam_check_saml_providers_sts", "iam_customer_attached_policy_no_administrative_privileges", "iam_password_policy_minimum_length_14", ] aws_provider = AwsProvider() aws_provider._audit_resources = [ f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:user/user-name" ] recovered_checks = aws_provider.get_checks_from_input_arn() assert recovered_checks == expected_checks @mock_aws @mock_aws @patch( "prowler.lib.check.utils.recover_checks_from_provider", new=mock_recover_checks_from_aws_provider_s3_service, ) def test_get_checks_from_input_arn_s3(self): expected_checks = [ "s3_account_level_public_access_blocks", "s3_bucket_acl_prohibited", "s3_bucket_policy_public_write_access", ] aws_provider = AwsProvider() aws_provider._audit_resources = ["arn:aws:s3:::bucket-name"] recovered_checks = aws_provider.get_checks_from_input_arn() assert recovered_checks == expected_checks @mock_aws @patch( "prowler.lib.check.utils.recover_checks_from_provider", new=mock_recover_checks_from_aws_provider_cloudwatch_service, ) def test_get_checks_from_input_arn_cloudwatch(self): expected_checks = [ "cloudwatch_changes_to_network_acls_alarm_configured", "cloudwatch_changes_to_network_gateways_alarm_configured", "cloudwatch_changes_to_network_route_tables_alarm_configured", ] aws_provider = AwsProvider() aws_provider._audit_resources = [ f"arn:aws:logs:us-east-1:{AWS_ACCOUNT_NUMBER}:destination:testDestination" ] recovered_checks = aws_provider.get_checks_from_input_arn() assert recovered_checks == expected_checks @mock_aws @patch( "prowler.lib.check.utils.recover_checks_from_provider", new=mock_recover_checks_from_aws_provider_eks_service, ) def test_get_checks_from_input_arn_eks(self): expected_checks = [ "eks_cluster_not_publicly_accessible", "eks_cluster_uses_a_supported_version", "eks_cluster_network_policy_enabled", "eks_control_plane_logging_all_types_enabled", ] aws_provider = AwsProvider() aws_provider._audit_resources = [ f"arn:aws:eks:us-east-1:{AWS_ACCOUNT_NUMBER}:cluster/test-eks" ] recovered_checks = aws_provider.get_checks_from_input_arn() assert set(recovered_checks) == set(expected_checks) @mock_aws @patch( "prowler.lib.check.utils.recover_checks_from_provider", new=mock_recover_checks_from_aws_provider_cognito_service, ) def test_get_checks_from_input_arn_cognito(self): expected_checks = [] aws_provider = AwsProvider() aws_provider._audit_resources = [ f"arn:aws:cognito-idp:us-east-1:{AWS_ACCOUNT_NUMBER}:userpool/test" ] recovered_checks = aws_provider.get_checks_from_input_arn() assert recovered_checks == expected_checks @mock_aws @patch( "prowler.lib.check.utils.recover_checks_from_provider", new=mock_recover_checks_from_aws_provider_ec2_service, ) def test_get_checks_from_input_arn_ec2_security_group(self): expected_checks = ["ec2_securitygroup_allow_ingress_from_internet_to_any_port"] aws_provider = AwsProvider() aws_provider._audit_resources = [ f"arn:aws:ec2:us-east-1:{AWS_ACCOUNT_NUMBER}:security-group/sg-1111111111" ] recovered_checks = aws_provider.get_checks_from_input_arn() assert recovered_checks == expected_checks @mock_aws @patch( "prowler.lib.check.utils.recover_checks_from_provider", new=mock_recover_checks_from_aws_provider_ec2_service, ) def test_get_checks_from_input_arn_ec2_acl(self): expected_checks = ["ec2_networkacl_allow_ingress_any_port"] aws_provider = AwsProvider() aws_provider._audit_resources = [ f"arn:aws:ec2:us-west-2:{AWS_ACCOUNT_NUMBER}:network-acl/acl-1" ] recovered_checks = aws_provider.get_checks_from_input_arn() assert recovered_checks == expected_checks @mock_aws @patch( "prowler.lib.check.utils.recover_checks_from_provider", new=mock_recover_checks_from_aws_provider_rds_service, ) def test_get_checks_from_input_arn_rds_snapshots(self): expected_checks = ["rds_snapshots_public_access"] aws_provider = AwsProvider() aws_provider._audit_resources = [ f"arn:aws:rds:us-east-2:{AWS_ACCOUNT_NUMBER}:snapshot:rds:snapshot-1", ] recovered_checks = aws_provider.get_checks_from_input_arn() assert recovered_checks == expected_checks @mock_aws @patch( "prowler.lib.check.utils.recover_checks_from_provider", new=mock_recover_checks_from_aws_provider_ec2_service, ) def test_get_checks_from_input_arn_ec2_ami(self): expected_checks = ["ec2_ami_public"] aws_provider = AwsProvider() aws_provider._audit_resources = [ f"arn:aws:ec2:us-west-2:{AWS_ACCOUNT_NUMBER}:image/ami-1" ] recovered_checks = aws_provider.get_checks_from_input_arn() assert recovered_checks == expected_checks @mock_aws def test_get_regions_from_audit_resources_with_regions(self): audit_resources = [ f"arn:aws:lambda:us-east-1:{AWS_ACCOUNT_NUMBER}:function:test-lambda", f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:policy/test", f"arn:aws:ec2:eu-west-1:{AWS_ACCOUNT_NUMBER}:security-group/sg-test", "arn:aws:s3:::bucket-name", "arn:aws:apigateway:us-east-2::/restapis/api-id/stages/stage-name", ] expected_regions = {"us-east-1", "eu-west-1", "us-east-2"} aws_provider = AwsProvider() recovered_regions = aws_provider.get_regions_from_audit_resources( audit_resources ) assert recovered_regions == expected_regions @mock_aws def test_get_regions_from_audit_resources_without_regions(self): audit_resources = ["arn:aws:s3:::bucket-name"] aws_provider = AwsProvider() recovered_regions = aws_provider.get_regions_from_audit_resources( audit_resources ) assert not recovered_regions def test_get_regions_all_count(self): assert len(AwsProvider.get_regions(partition=None)) == 39 def test_get_regions_cn_count(self): assert len(AwsProvider.get_regions("aws-cn")) == 2 def test_get_regions_aws_count(self): assert len(AwsProvider.get_regions(partition="aws")) == 34 def test_get_all_regions(self): with patch( "prowler.providers.aws.aws_provider.read_aws_regions_file", return_value={ "services": { "acm": { "regions": { "aws": [ "af-south-1", ], "aws-cn": [ "cn-north-1", ], "aws-us-gov": [ "us-gov-west-1", ], } } } }, ): assert AwsProvider.get_regions(partition=None) == { "af-south-1", "cn-north-1", "us-gov-west-1", } def test_get_regions_with_us_gov_partition(self): with patch( "prowler.providers.aws.aws_provider.read_aws_regions_file", return_value={ "services": { "acm": { "regions": { "aws": [ "af-south-1", ], "aws-cn": [ "cn-north-1", ], "aws-us-gov": [ "us-gov-west-1", ], } } } }, ): assert AwsProvider.get_regions("aws-us-gov") == { "us-gov-west-1", } def test_get_regions_with_aws_partition(self): with patch( "prowler.providers.aws.aws_provider.read_aws_regions_file", return_value={ "services": { "acm": { "regions": { "aws": [ "af-south-1", ], "aws-cn": [ "cn-north-1", ], "aws-us-gov": [ "us-gov-west-1", ], } } } }, ): assert AwsProvider.get_regions("aws") == { "af-south-1", } def test_get_regions_with_cn_partition(self): with patch( "prowler.providers.aws.aws_provider.read_aws_regions_file", return_value={ "services": { "acm": { "regions": { "aws": [ "af-south-1", ], "aws-cn": [ "cn-north-1", ], "aws-us-gov": [ "us-gov-west-1", ], } } } }, ): assert AwsProvider.get_regions("aws-cn") == { "cn-north-1", } def test_get_regions_with_unknown_partition(self): with patch( "prowler.providers.aws.aws_provider.read_aws_regions_file", return_value={ "services": { "acm": { "regions": { "aws": [ "af-south-1", ], "aws-cn": [ "cn-north-1", ], "aws-us-gov": [ "us-gov-west-1", ], } } } }, ): partition = "unknown" with pytest.raises(AWSInvalidPartitionError) as exception: AwsProvider.get_regions(partition) assert exception.type == AWSInvalidPartitionError assert f"Invalid partition: {partition}" in exception.value.args[0] def test_get_aws_region_for_sts_input_regions_none_session_region_none(self): input_regions = None session_region = None assert ( get_aws_region_for_sts(session_region, input_regions) == AWS_STS_GLOBAL_ENDPOINT_REGION ) def test_get_aws_region_for_sts_input_regions_none_session_region_ireland(self): input_regions = None session_region = AWS_REGION_EU_WEST_1 assert ( get_aws_region_for_sts(session_region, input_regions) == AWS_REGION_EU_WEST_1 ) def test_get_aws_region_for_sts_input_regions_empty_session_region_none(self): input_regions = set() session_region = None assert ( get_aws_region_for_sts(session_region, input_regions) == AWS_STS_GLOBAL_ENDPOINT_REGION ) def test_get_aws_region_for_sts_input_regions_empty_session_region_ireland(self): input_regions = set() session_region = AWS_REGION_EU_WEST_1 assert ( get_aws_region_for_sts(session_region, input_regions) == AWS_REGION_EU_WEST_1 ) def test_get_aws_region_for_sts_input_regions_ireland_and_virgninia(self): input_regions = [AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1] session_region = None assert ( get_aws_region_for_sts(session_region, input_regions) == AWS_REGION_EU_WEST_1 ) @mock_aws def test_set_session_config_default(self): aws_provider = AwsProvider() session_config = aws_provider.set_session_config(None) assert session_config.user_agent_extra == BOTO3_USER_AGENT_EXTRA assert session_config.retries == {"max_attempts": 3, "mode": "standard"} @mock_aws def test_set_session_config_10_max_attempts(self): aws_provider = AwsProvider() session_config = aws_provider.set_session_config(10) assert session_config.user_agent_extra == BOTO3_USER_AGENT_EXTRA assert session_config.retries == {"max_attempts": 10, "mode": "standard"} @mock_aws @patch( "prowler.lib.check.utils.recover_checks_from_provider", new=mock_recover_checks_from_aws_provider_ec2_service, ) def test_get_checks_to_execute_by_audit_resources(self): aws_provider = AwsProvider() aws_provider._audit_resources = [ f"arn:aws:ec2:us-west-2:{AWS_ACCOUNT_NUMBER}:network-acl/acl-1" ] aws_provider.get_checks_to_execute_by_audit_resources() == { "ec2_networkacl_allow_ingress_any_port" } def test_update_provider_config_aws(self): aws_provider = set_mocked_aws_provider( audit_config={"shodan_api_key": "DEFAULT-KEY"} ) with patch( "prowler.providers.common.provider.Provider.get_global_provider", return_value=aws_provider, ): assert { "shodan_api_key": "TEST-API-KEY" } == Provider.update_provider_config( aws_provider.audit_config, "shodan_api_key", "TEST-API-KEY" ) def test_update_provider_config_aws_not_present(self): aws_provider = set_mocked_aws_provider( audit_config={"shodan_api_key": "DEFAULT-KEY"} ) with patch( "prowler.providers.common.provider.Provider.get_global_provider", return_value=aws_provider, ): assert {"shodan_api_key": "DEFAULT-KEY"} == Provider.update_provider_config( aws_provider.audit_config, "not_found", "not_value" ) @mock_aws def test_refresh_credentials_before_expiration(self): role_arn = create_role(AWS_REGION_EU_WEST_1) session_duration = 900 aws_provider = AwsProvider(role_arn=role_arn, session_duration=session_duration) current_credentials = ( aws_provider._assumed_role_configuration.credentials.__dict__ ) refreshed_credentials = { "access_key": current_credentials["aws_access_key_id"], "secret_key": current_credentials["aws_secret_access_key"], "token": current_credentials["aws_session_token"], "expiry_time": current_credentials.get( "expiration", current_credentials.get("expiry_time") ).isoformat(), } assert ( AwsProvider.refresh_credentials( aws_provider._assumed_role_configuration, aws_provider._session ) == refreshed_credentials ) @mock_aws def test_refresh_credentials_after_expiration(self): role_arn = create_role(AWS_REGION_EU_WEST_1) session_duration_in_seconds = 900 session_duration = session_duration_in_seconds aws_provider = AwsProvider(role_arn=role_arn, session_duration=session_duration) # Manually expire credentials aws_provider._assumed_role_configuration.credentials.expiration = datetime.now( get_localzone() ) - timedelta(seconds=session_duration_in_seconds) current_credentials = aws_provider._assumed_role_configuration.credentials # Refresh credentials refreshed_credentials = AwsProvider.refresh_credentials( aws_provider._assumed_role_configuration, aws_provider._session ) # Assert that the refreshed credentials are different access_key = refreshed_credentials.get("access_key") assert access_key != current_credentials.aws_access_key_id secret_key = refreshed_credentials.get("secret_key") assert secret_key != current_credentials.aws_secret_access_key session_token = refreshed_credentials.get("token") assert session_token != current_credentials.aws_session_token expiry_time = refreshed_credentials.get("expiry_time") expiry_time_formatted = datetime.fromisoformat(expiry_time) assert expiry_time != current_credentials.expiration assert datetime.now(get_localzone()) < expiry_time_formatted # Assert credentials format assert len(access_key) == 20 assert search(r"^ASIA.*$", access_key) assert len(secret_key) == 40 assert len(session_token) == 356 assert search(r"^FQoGZXIvYXdzE.*$", session_token)