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 ( MAX_STS_BOOTSTRAP_ATTEMPTS, AwsProvider, get_aws_region_for_sts, get_env_partition_bootstrap_region, get_env_partition_regions, get_partition_bootstrap_candidates, ) from prowler.providers.aws.config import ( AWS_STS_GLOBAL_ENDPOINT_REGION, BOTO3_CONNECT_TIMEOUT, BOTO3_READ_TIMEOUT, BOTO3_RETRIES_MAX_ATTEMPTS, BOTO3_USER_AGENT_EXTRA, ROLE_SESSION_NAME, get_boto3_retries_from_env, get_boto3_timeout_from_env, get_default_session_config, ) from prowler.providers.aws.exceptions.exceptions import ( AWSAccessKeyIDInvalidError, AWSArgumentTypeValidationError, AWSAssumeRoleError, AWSIAMRoleARNInvalidResourceTypeError, AWSInvalidBoto3RetriesError, AWSInvalidBoto3TimeoutError, 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.lib.session.aws_set_up_session import AwsSetUpSession 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_B_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_GOV_CLOUD_US_WEST_1, AWS_REGION_ISO_B_EAST_1, AWS_REGION_ISO_EAST_1, AWS_REGION_ISO_WEST_1, 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_organizations_uses_assumed_role_session_by_default(self): # Regression test for issue #10215. # When only `role_arn` is provided (no `organizations_role_arn`), # the FIRST attempt to fetch Organizations metadata must use the # assumed role session (current_session), not the pre-assume # credentials. This mirrors the CLI: `aws sts assume-role` followed # by `aws organizations describe-account` uses the assumed identity. role_arn = create_role(AWS_REGION_EU_WEST_1) captured_sessions = [] original_get_organizations_info = AwsProvider.get_organizations_info def capture(self, organizations_session, aws_account_id): captured_sessions.append(organizations_session) return original_get_organizations_info( self, organizations_session, aws_account_id ) with patch.object(AwsProvider, "get_organizations_info", capture): aws_provider = AwsProvider(role_arn=role_arn, session_duration=900) assert captured_sessions[0] is aws_provider.session.current_session assert captured_sessions[0] is not aws_provider.session.original_session @mock_aws def test_aws_provider_organizations_falls_back_to_original_session(self): # When `role_arn` is provided and the assumed role session cannot # retrieve Organizations metadata (e.g. management-account -> # member-account flow where the member account has no Organizations # permissions), retry with the original (pre-assume) session. role_arn = create_role(AWS_REGION_EU_WEST_1) captured_sessions = [] original_get_organizations_info = AwsProvider.get_organizations_info def capture(self, organizations_session, aws_account_id): captured_sessions.append(organizations_session) return original_get_organizations_info( self, organizations_session, aws_account_id ) with patch.object(AwsProvider, "get_organizations_info", capture): aws_provider = AwsProvider(role_arn=role_arn, session_duration=900) assert len(captured_sessions) == 2 assert captured_sessions[0] is aws_provider.session.current_session assert captured_sessions[1] is aws_provider.session.original_session @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_excluded_regions_removed_from_enabled_regions(self): aws_provider = AwsProvider(excluded_regions={AWS_REGION_EU_WEST_1}) assert AWS_REGION_EU_WEST_1 not in aws_provider._enabled_regions assert AWS_REGION_EU_WEST_1 not in aws_provider.generate_regional_clients("ec2") @mock_aws def test_excluded_regions_pruned_from_input_regions(self): aws_provider = AwsProvider( regions={AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1}, excluded_regions={AWS_REGION_EU_WEST_1}, ) assert AWS_REGION_EU_WEST_1 not in aws_provider._identity.audited_regions assert AWS_REGION_US_EAST_1 in aws_provider._identity.audited_regions @mock_aws def test_excluded_regions_from_config_file(self): with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as tmp: tmp.write(f"aws:\n disallowed_regions:\n - {AWS_REGION_EU_WEST_1}\n") config_path = tmp.name try: aws_provider = AwsProvider(config_path=config_path) assert AWS_REGION_EU_WEST_1 not in aws_provider._enabled_regions assert aws_provider._excluded_regions == {AWS_REGION_EU_WEST_1} finally: os.remove(config_path) @mock_aws def test_excluded_regions_from_env_on_direct_provider_init(self): with mock.patch.dict( os.environ, {"PROWLER_AWS_DISALLOWED_REGIONS": AWS_REGION_EU_WEST_1}, clear=False, ): aws_provider = AwsProvider() assert aws_provider._excluded_regions == {AWS_REGION_EU_WEST_1} assert AWS_REGION_EU_WEST_1 not in aws_provider._enabled_regions @mock_aws def test_excluded_regions_precedence_explicit_over_env_and_config(self): with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as tmp: tmp.write(f"aws:\n disallowed_regions:\n - {AWS_REGION_EU_WEST_1}\n") config_path = tmp.name try: with mock.patch.dict( os.environ, {"PROWLER_AWS_DISALLOWED_REGIONS": AWS_REGION_US_EAST_1}, clear=False, ): aws_provider = AwsProvider( config_path=config_path, excluded_regions={AWS_REGION_US_EAST_2}, ) assert aws_provider._excluded_regions == {AWS_REGION_US_EAST_2} assert AWS_REGION_US_EAST_2 not in aws_provider._enabled_regions assert AWS_REGION_EU_WEST_1 in aws_provider._enabled_regions assert AWS_REGION_US_EAST_1 in aws_provider._enabled_regions finally: os.remove(config_path) @mock_aws def test_excluded_regions_from_config_avoid_excluded_profile_region( self, monkeypatch ): monkeypatch.setenv("AWS_DEFAULT_REGION", AWS_REGION_EU_WEST_1) with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as tmp: tmp.write(f"aws:\n disallowed_regions:\n - {AWS_REGION_EU_WEST_1}\n") config_path = tmp.name try: aws_provider = AwsProvider(config_path=config_path) assert aws_provider.identity.profile_region == AWS_REGION_US_EAST_1 finally: os.remove(config_path) @mock_aws def test_aws_provider_raises_when_all_input_regions_are_excluded(self): with raises(AWSArgumentTypeValidationError): AwsProvider( regions={AWS_REGION_EU_WEST_1}, excluded_regions={AWS_REGION_EU_WEST_1}, ) def test_get_excluded_regions_from_env_parses_comma_list(self): with mock.patch.dict( os.environ, {"PROWLER_AWS_DISALLOWED_REGIONS": " me-south-1 , ap-east-1 ,, "}, ): assert Provider.get_excluded_regions_from_env() == { "me-south-1", "ap-east-1", } def test_get_excluded_regions_from_env_ignores_legacy_generic_name(self): with mock.patch.dict( os.environ, {"PROWLER_DISALLOWED_REGIONS": "me-south-1"}, clear=True, ): assert Provider.get_excluded_regions_from_env() == set() def test_get_excluded_regions_from_env_unset(self): with mock.patch.dict(os.environ, {}, clear=True): assert Provider.get_excluded_regions_from_env() == set() @mock_aws def test_print_credentials_shows_all_except_excluded_regions(self): aws_provider = AwsProvider( excluded_regions={AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1} ) with patch( "prowler.providers.aws.aws_provider.print_boxes" ) as mock_print_boxes: aws_provider.print_credentials() report_lines = mock_print_boxes.call_args.args[0] assert any( "AWS Regions:" in line and "all except eu-west-1, us-east-1" in line for line in report_lines ) @mock_aws def test_generate_regional_clients_all_enabled_regions(self): aws_provider = AwsProvider() response = aws_provider.generate_regional_clients("ec2") # Only commercial regions (not GovCloud/China) should have regional clients commercial_regions = { r for r in aws_provider._enabled_regions if not r.startswith("cn-") and not r.startswith("us-gov-") } assert set(response.keys()) == commercial_regions @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_get_default_region_global_service_ignores_profile_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("cloudfront", global_service=True) == AWS_REGION_US_EAST_1 ) @mock_aws def test_get_default_region_global_service_ignores_audited_regions(self): region = [AWS_REGION_EU_WEST_1] aws_provider = AwsProvider( regions=region, ) aws_provider._identity.profile_region = None assert ( aws_provider.get_default_region("route53", global_service=True) == AWS_REGION_US_EAST_1 ) @mock_aws def test_get_default_region_global_service_china_partition(self): aws_provider = AwsProvider() aws_provider._identity.partition = AWS_CHINA_PARTITION aws_provider._identity.profile_region = AWS_REGION_CN_NORTHWEST_1 assert ( aws_provider.get_default_region("cloudfront", global_service=True) == AWS_REGION_CN_NORTH_1 ) @mock_aws def test_get_default_region_global_service_gov_cloud_partition(self): aws_provider = AwsProvider() aws_provider._identity.partition = AWS_GOV_CLOUD_PARTITION aws_provider._identity.profile_region = "us-gov-west-1" assert ( aws_provider.get_default_region("shield", global_service=True) == AWS_REGION_GOV_CLOUD_US_EAST_1 ) @mock_aws def test_get_default_region_non_global_service_unaffected(self): """Ensure global_service=False (default) still follows profile region logic.""" 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", global_service=False) == AWS_REGION_EU_WEST_1 ) @mock_aws def test_aws_get_global_region(self): aws_provider = AwsProvider() aws_provider._identity.partition = AWS_COMMERCIAL_PARTITION assert aws_provider.get_global_region() == AWS_REGION_US_EAST_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_EAST_1 @mock_aws def test_aws_iso_b_get_global_region(self): aws_provider = AwsProvider() aws_provider._identity.partition = AWS_ISO_B_PARTITION assert aws_provider.get_global_region() == AWS_REGION_ISO_B_EAST_1 @mock_aws def test_get_global_region_for_an_unknown_partition(self): aws_provider = AwsProvider() aws_provider._identity.partition = "aws-unknown" assert aws_provider.get_global_region() == AWS_REGION_US_EAST_1 @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_available_aws_service_regions_commercial_and_gov_cloud(self): aws_provider = AwsProvider() assert AWS_REGION_US_EAST_1 in aws_provider.get_available_aws_service_regions( "ec2", AWS_COMMERCIAL_PARTITION ) assert ( AWS_REGION_GOV_CLOUD_US_EAST_1 in aws_provider.get_available_aws_service_regions( "ec2", AWS_GOV_CLOUD_PARTITION ) ) # A service recorded as unavailable in the partition yields an empty set assert ( aws_provider.get_available_aws_service_regions( "bedrock-agent", AWS_CHINA_PARTITION ) == set() ) @mock_aws def test_get_available_aws_service_regions_iso_partitions(self): aws_provider = AwsProvider() assert aws_provider.get_available_aws_service_regions( "ec2", AWS_ISO_PARTITION ) == { AWS_REGION_ISO_EAST_1, AWS_REGION_ISO_WEST_1, } assert aws_provider.get_available_aws_service_regions( "guardduty", AWS_ISO_B_PARTITION ) == {AWS_REGION_ISO_B_EAST_1} # Every service carries every ISO partition, empty when not available assert ( aws_provider.get_available_aws_service_regions( "bedrock", AWS_ISO_B_PARTITION ) == set() ) @mock_aws def test_get_available_aws_service_regions_unknown_partition(self): aws_provider = AwsProvider() assert ( aws_provider.get_available_aws_service_regions("ec2", "aws-unknown") == set() ) @mock_aws def test_get_available_aws_service_regions_unknown_service(self): aws_provider = AwsProvider() assert ( aws_provider.get_available_aws_service_regions( "unknown-service", AWS_COMMERCIAL_PARTITION ) == set() ) @mock_aws def test_generate_regional_clients_service_not_in_partition(self): aws_provider = AwsProvider() aws_provider._identity.partition = AWS_ISO_PARTITION response = aws_provider.generate_regional_clients("bedrock") assert response == {} @mock_aws def test_generate_regional_clients_returns_empty_dict_on_error(self): aws_provider = AwsProvider() with patch.object( AwsProvider, "get_available_aws_service_regions", side_effect=Exception("boom"), ): assert aws_provider.generate_regional_clients("ec2") == {} @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" def test_get_partition_bootstrap_candidates_adds_the_rest_of_the_partition( self, monkeypatch ): monkeypatch.setenv("PROWLER_AWS_PARTITION", AWS_GOV_CLOUD_PARTITION) assert get_partition_bootstrap_candidates( AWS_REGION_GOV_CLOUD_US_EAST_1, AWS_REGION_US_EAST_1 ) == [AWS_REGION_GOV_CLOUD_US_EAST_1, AWS_REGION_GOV_CLOUD_US_WEST_1] def test_get_partition_bootstrap_candidates_without_partition_offers_one_region( self, monkeypatch ): monkeypatch.delenv("PROWLER_AWS_PARTITION", raising=False) assert get_partition_bootstrap_candidates(AWS_REGION_EU_WEST_1) == [ AWS_REGION_EU_WEST_1 ] def test_get_partition_bootstrap_candidates_is_capped(self, monkeypatch): monkeypatch.setenv("PROWLER_AWS_PARTITION", AWS_COMMERCIAL_PARTITION) candidates = get_partition_bootstrap_candidates( AWS_REGION_EU_WEST_1, AWS_REGION_EU_WEST_1 ) assert len(candidates) == MAX_STS_BOOTSTRAP_ATTEMPTS assert candidates[0] == AWS_REGION_EU_WEST_1 def test_get_partition_bootstrap_candidates_tries_excluded_regions_last( self, monkeypatch ): monkeypatch.setenv("PROWLER_AWS_PARTITION", AWS_COMMERCIAL_PARTITION) partition_regions = get_env_partition_regions(AWS_REGION_EU_WEST_1) excluded_regions = set(partition_regions[1:3]) candidates = get_partition_bootstrap_candidates( AWS_REGION_EU_WEST_1, AWS_REGION_EU_WEST_1, excluded_regions ) assert candidates == [AWS_REGION_EU_WEST_1, *partition_regions[3:5]] def test_get_partition_bootstrap_candidates_keeps_excluded_regions_as_a_last_resort( self, monkeypatch ): monkeypatch.setenv("PROWLER_AWS_PARTITION", AWS_GOV_CLOUD_PARTITION) assert get_partition_bootstrap_candidates( AWS_REGION_GOV_CLOUD_US_EAST_1, AWS_REGION_US_EAST_1, {AWS_REGION_GOV_CLOUD_US_EAST_1, AWS_REGION_GOV_CLOUD_US_WEST_1}, ) == [AWS_REGION_GOV_CLOUD_US_EAST_1, AWS_REGION_GOV_CLOUD_US_WEST_1] def test_validate_credentials_falls_back_to_the_next_partition_region( self, monkeypatch ): monkeypatch.setenv("PROWLER_AWS_PARTITION", AWS_GOV_CLOUD_PARTITION) # A container may carry a region that belongs to no partition it scans current_session = session.Session(region_name=AWS_REGION_US_EAST_1) attempted_regions = [] def create_sts_session(session, aws_region): attempted_regions.append(aws_region) if aws_region == AWS_REGION_GOV_CLOUD_US_EAST_1: raise botocore.exceptions.EndpointConnectionError( endpoint_url=f"https://sts.{aws_region}.amazonaws.com" ) sts_client = mock.MagicMock() sts_client.get_caller_identity.return_value = { "UserId": "test-user-id", "Account": AWS_ACCOUNT_NUMBER, "Arn": AWS_GOV_CLOUD_ACCOUNT_ARN, } return sts_client with patch( "prowler.providers.aws.aws_provider.AwsProvider.create_sts_session", new=create_sts_session, ): caller_identity = AwsProvider.validate_credentials( session=current_session, aws_region=AWS_REGION_GOV_CLOUD_US_EAST_1 ) assert attempted_regions == [ AWS_REGION_GOV_CLOUD_US_EAST_1, AWS_REGION_GOV_CLOUD_US_WEST_1, ] assert caller_identity.region == AWS_REGION_GOV_CLOUD_US_WEST_1 def test_validate_credentials_falls_back_when_a_region_does_not_answer( self, monkeypatch ): monkeypatch.setenv("PROWLER_AWS_PARTITION", AWS_GOV_CLOUD_PARTITION) current_session = session.Session(region_name=AWS_REGION_US_EAST_1) attempted_regions = [] # The connection is accepted but nothing comes back before the read timeout def create_sts_session(session, aws_region): attempted_regions.append(aws_region) if aws_region == AWS_REGION_GOV_CLOUD_US_EAST_1: raise botocore.exceptions.ReadTimeoutError( endpoint_url=f"https://sts.{aws_region}.amazonaws.com" ) sts_client = mock.MagicMock() sts_client.get_caller_identity.return_value = { "UserId": "test-user-id", "Account": AWS_ACCOUNT_NUMBER, "Arn": AWS_GOV_CLOUD_ACCOUNT_ARN, } return sts_client with patch( "prowler.providers.aws.aws_provider.AwsProvider.create_sts_session", new=create_sts_session, ): caller_identity = AwsProvider.validate_credentials( session=current_session, aws_region=AWS_REGION_GOV_CLOUD_US_EAST_1 ) assert attempted_regions == [ AWS_REGION_GOV_CLOUD_US_EAST_1, AWS_REGION_GOV_CLOUD_US_WEST_1, ] assert caller_identity.region == AWS_REGION_GOV_CLOUD_US_WEST_1 def test_validate_credentials_raises_when_no_partition_region_answers( self, monkeypatch ): monkeypatch.setenv("PROWLER_AWS_PARTITION", AWS_GOV_CLOUD_PARTITION) current_session = session.Session(region_name=AWS_REGION_US_EAST_1) attempted_regions = [] def create_sts_session(session, aws_region): attempted_regions.append(aws_region) raise botocore.exceptions.EndpointConnectionError( endpoint_url=f"https://sts.{aws_region}.amazonaws.com" ) with patch( "prowler.providers.aws.aws_provider.AwsProvider.create_sts_session", new=create_sts_session, ): with raises(botocore.exceptions.EndpointConnectionError): AwsProvider.validate_credentials( session=current_session, aws_region=AWS_REGION_GOV_CLOUD_US_EAST_1 ) assert attempted_regions == [ AWS_REGION_GOV_CLOUD_US_EAST_1, AWS_REGION_GOV_CLOUD_US_WEST_1, ] def test_validate_credentials_avoids_an_excluded_region_when_failing_over( self, monkeypatch ): monkeypatch.setenv("PROWLER_AWS_PARTITION", AWS_COMMERCIAL_PARTITION) current_session = session.Session(region_name=AWS_REGION_EU_WEST_1) partition_regions = get_env_partition_regions(AWS_REGION_EU_WEST_1) excluded_region, answering_region = partition_regions[1:3] attempted_regions = [] def create_sts_session(session, aws_region): attempted_regions.append(aws_region) if aws_region == AWS_REGION_EU_WEST_1: raise botocore.exceptions.EndpointConnectionError( endpoint_url=f"https://sts.{aws_region}.amazonaws.com" ) sts_client = mock.MagicMock() sts_client.get_caller_identity.return_value = { "UserId": "test-user-id", "Account": AWS_ACCOUNT_NUMBER, "Arn": AWS_ACCOUNT_ARN, } return sts_client with patch( "prowler.providers.aws.aws_provider.AwsProvider.create_sts_session", new=create_sts_session, ): caller_identity = AwsProvider.validate_credentials( session=current_session, aws_region=AWS_REGION_EU_WEST_1, excluded_regions={excluded_region}, ) assert attempted_regions == [AWS_REGION_EU_WEST_1, answering_region] assert caller_identity.region == answering_region def test_validate_credentials_does_not_retry_a_credential_error(self, monkeypatch): monkeypatch.setenv("PROWLER_AWS_PARTITION", AWS_GOV_CLOUD_PARTITION) current_session = session.Session(region_name=AWS_REGION_US_EAST_1) attempted_regions = [] def create_sts_session(session, aws_region): attempted_regions.append(aws_region) sts_client = mock.MagicMock() sts_client.get_caller_identity.side_effect = ( botocore.exceptions.ClientError( {"Error": {"Code": "InvalidClientTokenId", "Message": "invalid"}}, "GetCallerIdentity", ) ) return sts_client with patch( "prowler.providers.aws.aws_provider.AwsProvider.create_sts_session", new=create_sts_session, ): with raises(AWSAccessKeyIDInvalidError): AwsProvider.validate_credentials( session=current_session, aws_region=AWS_REGION_GOV_CLOUD_US_EAST_1 ) assert attempted_regions == [AWS_REGION_GOV_CLOUD_US_EAST_1] def test_assume_role_falls_back_to_the_next_partition_region(self, monkeypatch): monkeypatch.setenv("PROWLER_AWS_PARTITION", AWS_GOV_CLOUD_PARTITION) current_session = session.Session(region_name=AWS_REGION_US_EAST_1) attempted_regions = [] def create_sts_session(session, aws_region): attempted_regions.append(aws_region) if aws_region == AWS_REGION_GOV_CLOUD_US_EAST_1: raise botocore.exceptions.EndpointConnectionError( endpoint_url=f"https://sts.{aws_region}.amazonaws.com" ) sts_client = mock.MagicMock() sts_client.assume_role.return_value = { "Credentials": { "AccessKeyId": "AKIAIOSFODNN7EXAMPLE", "SecretAccessKey": "secret", "SessionToken": "token", "Expiration": datetime.now() + timedelta(seconds=3600), } } return sts_client assumed_role_info = AWSAssumeRoleInfo( role_arn=ARN( arn=f"arn:{AWS_GOV_CLOUD_PARTITION}:iam::{AWS_ACCOUNT_NUMBER}:role/test-role" ), session_duration=3600, external_id=None, mfa_enabled=False, role_session_name=ROLE_SESSION_NAME, sts_region=AWS_REGION_GOV_CLOUD_US_EAST_1, ) with patch( "prowler.providers.aws.aws_provider.AwsProvider.create_sts_session", new=create_sts_session, ): credentials = AwsProvider.assume_role(current_session, assumed_role_info) assert attempted_regions == [ AWS_REGION_GOV_CLOUD_US_EAST_1, AWS_REGION_GOV_CLOUD_US_WEST_1, ] assert isinstance(credentials, AWSCredentials) assert credentials.aws_access_key_id == "AKIAIOSFODNN7EXAMPLE" # Refreshing the credentials later goes straight to the region that answered assert assumed_role_info.sts_region == AWS_REGION_GOV_CLOUD_US_WEST_1 def test_assume_role_does_not_retry_a_credential_error(self, monkeypatch): monkeypatch.setenv("PROWLER_AWS_PARTITION", AWS_GOV_CLOUD_PARTITION) current_session = session.Session(region_name=AWS_REGION_US_EAST_1) attempted_regions = [] def create_sts_session(session, aws_region): attempted_regions.append(aws_region) sts_client = mock.MagicMock() sts_client.assume_role.side_effect = botocore.exceptions.ClientError( {"Error": {"Code": "AccessDenied", "Message": "denied"}}, "AssumeRole", ) return sts_client assumed_role_info = AWSAssumeRoleInfo( role_arn=ARN( arn=f"arn:{AWS_GOV_CLOUD_PARTITION}:iam::{AWS_ACCOUNT_NUMBER}:role/test-role" ), session_duration=3600, external_id=None, mfa_enabled=False, role_session_name=ROLE_SESSION_NAME, sts_region=AWS_REGION_GOV_CLOUD_US_EAST_1, ) with patch( "prowler.providers.aws.aws_provider.AwsProvider.create_sts_session", new=create_sts_session, ): with raises(AWSAssumeRoleError): AwsProvider.assume_role(current_session, assumed_role_info) assert attempted_regions == [AWS_REGION_GOV_CLOUD_US_EAST_1] assert assumed_role_info.sts_region == AWS_REGION_GOV_CLOUD_US_EAST_1 def test_test_connection_role_validates_where_the_role_was_assumed( self, monkeypatch ): monkeypatch.setenv("PROWLER_AWS_PARTITION", AWS_GOV_CLOUD_PARTITION) monkeypatch.delenv("AWS_DEFAULT_REGION", raising=False) attempted_calls = [] def create_sts_session(session, aws_region): if aws_region == AWS_REGION_GOV_CLOUD_US_EAST_1: attempted_calls.append(aws_region) raise botocore.exceptions.ConnectTimeoutError( endpoint_url=f"https://sts.{aws_region}.amazonaws.com" ) sts_client = mock.MagicMock() def assume_role(**_): attempted_calls.append(("AssumeRole", aws_region)) return { "Credentials": { "AccessKeyId": "AKIAIOSFODNN7EXAMPLE", "SecretAccessKey": "secret", "SessionToken": "token", "Expiration": datetime.now() + timedelta(seconds=3600), } } def get_caller_identity(): attempted_calls.append(("GetCallerIdentity", aws_region)) return { "UserId": "test-user-id", "Account": AWS_ACCOUNT_NUMBER, "Arn": AWS_GOV_CLOUD_ACCOUNT_ARN, } sts_client.assume_role.side_effect = assume_role sts_client.get_caller_identity.side_effect = get_caller_identity return sts_client with patch( "prowler.providers.aws.aws_provider.AwsProvider.create_sts_session", new=create_sts_session, ): connection = AwsProvider.test_connection( role_arn=f"arn:{AWS_GOV_CLOUD_PARTITION}:iam::{AWS_ACCOUNT_NUMBER}:role/test-role", aws_access_key_id="test-access-key", aws_secret_access_key="test-secret-key", raise_on_exception=False, ) assert connection.is_connected # The unreachable region is paid for once, not again for the validation assert attempted_calls == [ AWS_REGION_GOV_CLOUD_US_EAST_1, ("AssumeRole", AWS_REGION_GOV_CLOUD_US_WEST_1), ("GetCallerIdentity", AWS_REGION_GOV_CLOUD_US_WEST_1), ] @mock_aws def test_aws_set_up_session_assumes_the_role_where_validation_got_an_answer( self, monkeypatch ): monkeypatch.setenv("PROWLER_AWS_PARTITION", AWS_GOV_CLOUD_PARTITION) monkeypatch.setenv("AWS_DEFAULT_REGION", AWS_REGION_US_EAST_1) answered = AWSCallerIdentity( user_id="test-user-id", account=AWS_ACCOUNT_NUMBER, arn=ARN(AWS_GOV_CLOUD_ACCOUNT_ARN), region=AWS_REGION_GOV_CLOUD_US_WEST_1, ) sts_regions = [] class RoleAssumed(Exception): pass def assume_role(session, assumed_role_info): sts_regions.append(assumed_role_info.sts_region) raise RoleAssumed with ( patch( "prowler.providers.aws.aws_provider.AwsProvider.validate_credentials", return_value=answered, ), patch( "prowler.providers.aws.aws_provider.AwsProvider.assume_role", side_effect=assume_role, ), ): with raises(RoleAssumed): AwsSetUpSession( role_arn=f"arn:{AWS_GOV_CLOUD_PARTITION}:iam::{AWS_ACCOUNT_NUMBER}:role/test-role", session_duration=900, external_id="test-external-id", role_session_name=ROLE_SESSION_NAME, aws_access_key_id="testing", aws_secret_access_key="testing", ) assert sts_regions == [AWS_REGION_GOV_CLOUD_US_WEST_1] def test_setup_session_mfa_falls_back_to_the_next_partition_region( self, monkeypatch ): monkeypatch.setenv("PROWLER_AWS_PARTITION", AWS_GOV_CLOUD_PARTITION) monkeypatch.setenv("AWS_DEFAULT_REGION", AWS_REGION_US_EAST_1) attempted_regions = [] def create_sts_session(session, aws_region): attempted_regions.append(aws_region) if aws_region == AWS_REGION_GOV_CLOUD_US_EAST_1: raise botocore.exceptions.EndpointConnectionError( endpoint_url=f"https://sts.{aws_region}.amazonaws.com" ) sts_client = mock.MagicMock() sts_client.get_session_token.return_value = { "Credentials": { "AccessKeyId": "AKIAIOSFODNN7EXAMPLE", "SecretAccessKey": "secret", "SessionToken": "token", } } return sts_client with ( patch( "prowler.providers.aws.aws_provider.AwsProvider.input_role_mfa_token_and_code", return_value=AWSMFAInfo( arn=f"arn:{AWS_GOV_CLOUD_PARTITION}:iam::{AWS_ACCOUNT_NUMBER}:mfa/test", totp="123456", ), ), patch( "prowler.providers.aws.aws_provider.AwsProvider.create_sts_session", new=create_sts_session, ), ): mfa_session = AwsProvider.setup_session( mfa=True, aws_access_key_id="test-access-key", aws_secret_access_key="test-secret-key", ) assert attempted_regions == [ AWS_REGION_GOV_CLOUD_US_EAST_1, AWS_REGION_GOV_CLOUD_US_WEST_1, ] assert mfa_session.get_credentials().access_key == "AKIAIOSFODNN7EXAMPLE" assert mfa_session.get_credentials().token == "token" @mock_aws def test_aws_provider_hands_excluded_regions_to_credential_validation( self, monkeypatch ): monkeypatch.setenv("PROWLER_AWS_PARTITION", AWS_GOV_CLOUD_PARTITION) monkeypatch.setenv("AWS_DEFAULT_REGION", AWS_REGION_US_EAST_1) handed = [] class Validated(Exception): pass # Stops at the validation: what it was handed is all this checks def validate_credentials(session, aws_region, excluded_regions=None): handed.append((aws_region, set(excluded_regions or ()))) raise Validated with patch( "prowler.providers.aws.aws_provider.AwsProvider.validate_credentials", side_effect=validate_credentials, ): with raises(Validated): AwsProvider(excluded_regions={AWS_REGION_GOV_CLOUD_US_EAST_1}) assert handed == [ (AWS_REGION_GOV_CLOUD_US_WEST_1, {AWS_REGION_GOV_CLOUD_US_EAST_1}) ] @mock_aws def test_aws_provider_assumes_the_role_where_validation_got_an_answer( self, monkeypatch ): monkeypatch.setenv("PROWLER_AWS_PARTITION", AWS_GOV_CLOUD_PARTITION) # Out of the partition, so the first candidate is botocore's, not this one monkeypatch.setenv("AWS_DEFAULT_REGION", AWS_REGION_US_EAST_1) role_arn = ( f"arn:{AWS_GOV_CLOUD_PARTITION}:iam::{AWS_ACCOUNT_NUMBER}:role/test-role" ) answered = AWSCallerIdentity( user_id="test-user-id", account=AWS_ACCOUNT_NUMBER, arn=ARN(AWS_GOV_CLOUD_ACCOUNT_ARN), region=AWS_REGION_GOV_CLOUD_US_WEST_1, ) with patch( "prowler.providers.aws.aws_provider.AwsProvider.validate_credentials", return_value=answered, ): aws_provider = AwsProvider(role_arn=role_arn, session_duration=900) assert ( aws_provider._assumed_role_configuration.info.sts_region == AWS_REGION_GOV_CLOUD_US_WEST_1 ) @mock_aws def test_aws_provider_assumes_the_organizations_role_where_validation_got_an_answer( self, monkeypatch ): monkeypatch.setenv("PROWLER_AWS_PARTITION", AWS_GOV_CLOUD_PARTITION) monkeypatch.setenv("AWS_DEFAULT_REGION", AWS_REGION_US_EAST_1) organizations_role_arn = f"arn:{AWS_GOV_CLOUD_PARTITION}:iam::{AWS_ACCOUNT_NUMBER}:role/organizations-role" answered = AWSCallerIdentity( user_id="test-user-id", account=AWS_ACCOUNT_NUMBER, arn=ARN(AWS_GOV_CLOUD_ACCOUNT_ARN), region=AWS_REGION_GOV_CLOUD_US_WEST_1, ) sts_regions = [] class RoleAssumed(Exception): pass # Stops at the assumption: the region it was handed is all this checks def assume_role(session, assumed_role_info): sts_regions.append(assumed_role_info.sts_region) raise RoleAssumed with ( patch( "prowler.providers.aws.aws_provider.AwsProvider.validate_credentials", return_value=answered, ), patch( "prowler.providers.aws.aws_provider.AwsProvider.assume_role", side_effect=assume_role, ), ): with raises(RoleAssumed): AwsProvider( organizations_role_arn=organizations_role_arn, session_duration=900, ) assert sts_regions == [AWS_REGION_GOV_CLOUD_US_WEST_1] @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_custom_endpoint_url(self): custom_endpoint = "http://localhost:4566" current_session = session.Session() aws_region = AWS_REGION_US_EAST_1 with mock.patch.dict(os.environ, {"AWS_ENDPOINT_URL": custom_endpoint}): 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 == custom_endpoint @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 def test_create_sts_session_empty_endpoint_url(self): current_session = session.Session() aws_region = AWS_REGION_US_EAST_1 with mock.patch.dict(os.environ, {"AWS_ENDPOINT_URL": ""}): 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_iso(self): current_session = session.Session() aws_region = "us-iso-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}.c2s.ic.gov" @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): # 34 aws + 2 aws-cn + 2 aws-us-gov + 1 aws-eusc + 7 ISO regions assert len(AwsProvider.get_regions(partition=None)) == 46 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_regions_iso_count(self): assert AwsProvider.get_regions(AWS_ISO_PARTITION) == { AWS_REGION_ISO_EAST_1, AWS_REGION_ISO_WEST_1, } 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 ) def test_get_aws_region_for_sts_avoids_excluded_session_region(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} ) == AWS_REGION_US_EAST_1 ) def test_get_aws_region_for_sts_env_partition_gov_cloud(self): with mock.patch.dict( os.environ, {"PROWLER_AWS_PARTITION": AWS_GOV_CLOUD_PARTITION}, clear=False, ): assert get_aws_region_for_sts(None, None) == AWS_REGION_GOV_CLOUD_US_EAST_1 def test_get_aws_region_for_sts_env_partition_china(self): with mock.patch.dict( os.environ, {"PROWLER_AWS_PARTITION": AWS_CHINA_PARTITION}, clear=False, ): assert get_aws_region_for_sts(None, None) == AWS_REGION_CN_NORTH_1 def test_get_aws_region_for_sts_env_partition_eusc(self): with mock.patch.dict( os.environ, {"PROWLER_AWS_PARTITION": AWS_EUSC_PARTITION}, clear=False, ): assert get_aws_region_for_sts(None, None) == AWS_REGION_EUSC_DE_EAST_1 def test_get_aws_region_for_sts_env_partition_iso(self): with mock.patch.dict( os.environ, {"PROWLER_AWS_PARTITION": AWS_ISO_PARTITION}, clear=False, ): assert get_aws_region_for_sts(None, None) == "us-iso-east-1" def test_get_aws_region_for_sts_env_partition_overrides_session_region(self): with mock.patch.dict( os.environ, {"PROWLER_AWS_PARTITION": AWS_GOV_CLOUD_PARTITION}, clear=False, ): assert ( get_aws_region_for_sts(AWS_REGION_EU_WEST_1, None) == AWS_REGION_GOV_CLOUD_US_EAST_1 ) def test_get_aws_region_for_sts_input_regions_take_precedence_over_env_partition( self, ): with mock.patch.dict( os.environ, {"PROWLER_AWS_PARTITION": AWS_GOV_CLOUD_PARTITION}, clear=False, ): assert ( get_aws_region_for_sts(None, {AWS_REGION_EU_WEST_1}) == AWS_REGION_EU_WEST_1 ) def test_get_aws_region_for_sts_env_partition_invalid_raises(self): with mock.patch.dict( os.environ, {"PROWLER_AWS_PARTITION": "aws-invalid"}, clear=False, ): with pytest.raises(AWSInvalidPartitionError): get_aws_region_for_sts(None, None) @mock_aws def test_test_connection_uses_env_partition_sts_region(self): with ( mock.patch.dict( os.environ, {"PROWLER_AWS_PARTITION": AWS_GOV_CLOUD_PARTITION}, clear=False, ), mock.patch.object( AwsProvider, "validate_credentials", return_value=AWSCallerIdentity( user_id="test-user-id", account=AWS_ACCOUNT_NUMBER, arn=ARN(AWS_GOV_CLOUD_ACCOUNT_ARN), region=AWS_REGION_GOV_CLOUD_US_EAST_1, ), ) as mock_validate_credentials, ): connection = AwsProvider.test_connection( aws_access_key_id="test-access-key", aws_secret_access_key="test-secret-key", raise_on_exception=False, ) assert connection.is_connected assert ( mock_validate_credentials.call_args.args[1] == AWS_REGION_GOV_CLOUD_US_EAST_1 ) @mock_aws def test_test_connection_role_uses_env_partition_sts_region(self): with ( mock.patch.dict( os.environ, {"PROWLER_AWS_PARTITION": AWS_GOV_CLOUD_PARTITION}, clear=False, ), mock.patch.object( AwsProvider, "assume_role", return_value=AWSCredentials( aws_access_key_id="assumed-access-key", aws_secret_access_key="assumed-secret-key", aws_session_token="assumed-session-token", expiration=datetime.now(), ), ) as mock_assume_role, mock.patch.object( AwsProvider, "validate_credentials", return_value=AWSCallerIdentity( user_id="test-user-id", account=AWS_ACCOUNT_NUMBER, arn=ARN(AWS_GOV_CLOUD_ACCOUNT_ARN), region=AWS_REGION_GOV_CLOUD_US_EAST_1, ), ), ): connection = AwsProvider.test_connection( role_arn=f"arn:{AWS_GOV_CLOUD_PARTITION}:iam::{AWS_ACCOUNT_NUMBER}:role/test-role", aws_access_key_id="test-access-key", aws_secret_access_key="test-secret-key", raise_on_exception=False, ) assert connection.is_connected assumed_role_info = mock_assume_role.call_args.args[1] assert assumed_role_info.sts_region == AWS_REGION_GOV_CLOUD_US_EAST_1 def test_get_aws_region_for_sts_env_partition_commercial(self): with mock.patch.dict( os.environ, {"PROWLER_AWS_PARTITION": AWS_COMMERCIAL_PARTITION}, clear=False, ): assert get_aws_region_for_sts(None, None) == AWS_REGION_US_EAST_1 def test_get_aws_region_for_sts_env_partition_excluded_region_stays_in_partition( self, ): with mock.patch.dict( os.environ, {"PROWLER_AWS_PARTITION": AWS_GOV_CLOUD_PARTITION}, clear=False, ): assert ( get_aws_region_for_sts(None, None, {AWS_REGION_GOV_CLOUD_US_EAST_1}) == "us-gov-west-1" ) def test_get_aws_region_for_sts_env_partition_all_regions_excluded_stays_in_partition( self, ): with mock.patch.dict( os.environ, {"PROWLER_AWS_PARTITION": AWS_GOV_CLOUD_PARTITION}, clear=False, ): assert ( get_aws_region_for_sts( None, None, {AWS_REGION_GOV_CLOUD_US_EAST_1, "us-gov-west-1"} ) == AWS_REGION_GOV_CLOUD_US_EAST_1 ) @mock_aws def test_setup_session_mfa_uses_env_partition_sts_region(self): with ( mock.patch.dict( os.environ, {"PROWLER_AWS_PARTITION": AWS_GOV_CLOUD_PARTITION}, clear=False, ), mock.patch.object( AwsProvider, "input_role_mfa_token_and_code", return_value=AWSMFAInfo( arn=f"arn:{AWS_GOV_CLOUD_PARTITION}:iam::{AWS_ACCOUNT_NUMBER}:mfa/test", totp="123456", ), ), mock.patch.object( AwsProvider, "create_sts_session", side_effect=AwsProvider.create_sts_session, ) as mock_create_sts_session, ): AwsProvider.setup_session( mfa=True, aws_access_key_id="test-access-key", aws_secret_access_key="test-secret-key", ) assert ( mock_create_sts_session.call_args.args[1] == AWS_REGION_GOV_CLOUD_US_EAST_1 ) def test_get_env_partition_regions_leads_with_session_region(self): with mock.patch.dict( os.environ, {"PROWLER_AWS_PARTITION": AWS_GOV_CLOUD_PARTITION}, clear=False, ): regions = get_env_partition_regions(AWS_REGION_GOV_CLOUD_US_WEST_1) assert regions[0] == AWS_REGION_GOV_CLOUD_US_WEST_1 assert set(regions) == set(get_env_partition_regions()) def test_get_env_partition_regions_ignores_session_region_outside_partition( self, ): with mock.patch.dict( os.environ, {"PROWLER_AWS_PARTITION": AWS_GOV_CLOUD_PARTITION}, clear=False, ): regions = get_env_partition_regions(AWS_REGION_EU_WEST_1) assert regions[0] == AWS_REGION_GOV_CLOUD_US_EAST_1 assert AWS_REGION_EU_WEST_1 not in regions def test_get_env_partition_bootstrap_region_prefers_session_region(self): with mock.patch.dict( os.environ, {"PROWLER_AWS_PARTITION": AWS_GOV_CLOUD_PARTITION}, clear=False, ): assert ( get_env_partition_bootstrap_region(AWS_REGION_GOV_CLOUD_US_WEST_1) == AWS_REGION_GOV_CLOUD_US_WEST_1 ) def test_get_env_partition_bootstrap_region_without_session_region(self): with mock.patch.dict( os.environ, {"PROWLER_AWS_PARTITION": AWS_GOV_CLOUD_PARTITION}, clear=False, ): assert ( get_env_partition_bootstrap_region() == AWS_REGION_GOV_CLOUD_US_EAST_1 ) def test_get_env_partition_bootstrap_region_without_partition(self): with mock.patch.dict(os.environ, {"PROWLER_AWS_PARTITION": ""}, clear=False): assert ( get_env_partition_bootstrap_region(AWS_REGION_GOV_CLOUD_US_WEST_1) is None ) def test_get_aws_region_for_sts_env_partition_prefers_session_region(self): with mock.patch.dict( os.environ, {"PROWLER_AWS_PARTITION": AWS_GOV_CLOUD_PARTITION}, clear=False, ): assert ( get_aws_region_for_sts(AWS_REGION_GOV_CLOUD_US_WEST_1, None) == AWS_REGION_GOV_CLOUD_US_WEST_1 ) def test_get_profile_region_env_partition_keeps_session_region_inside_partition( self, ): with mock.patch.dict( os.environ, {"PROWLER_AWS_PARTITION": AWS_GOV_CLOUD_PARTITION}, clear=False, ): aws_session = session.Session(region_name=AWS_REGION_GOV_CLOUD_US_WEST_1) assert ( AwsProvider.get_profile_region(aws_session) == AWS_REGION_GOV_CLOUD_US_WEST_1 ) def test_get_profile_region_env_partition_ignores_session_region_outside_partition( self, ): with mock.patch.dict( os.environ, {"PROWLER_AWS_PARTITION": AWS_GOV_CLOUD_PARTITION}, clear=False, ): aws_session = session.Session(region_name=AWS_REGION_US_EAST_1) assert ( AwsProvider.get_profile_region(aws_session) == AWS_REGION_GOV_CLOUD_US_EAST_1 ) def test_get_profile_region_env_partition_excluded_session_region_stays_in_partition( self, ): with mock.patch.dict( os.environ, {"PROWLER_AWS_PARTITION": AWS_GOV_CLOUD_PARTITION}, clear=False, ): aws_session = session.Session(region_name=AWS_REGION_GOV_CLOUD_US_WEST_1) assert ( AwsProvider.get_profile_region( aws_session, {AWS_REGION_GOV_CLOUD_US_WEST_1} ) == AWS_REGION_GOV_CLOUD_US_EAST_1 ) def test_get_profile_region_env_partition_all_regions_excluded_stays_in_partition( self, ): with mock.patch.dict( os.environ, {"PROWLER_AWS_PARTITION": AWS_GOV_CLOUD_PARTITION}, clear=False, ): aws_session = session.Session(region_name=AWS_REGION_GOV_CLOUD_US_WEST_1) gov_cloud_regions = set(get_env_partition_regions()) assert ( AwsProvider.get_profile_region(aws_session, gov_cloud_regions) == AWS_REGION_GOV_CLOUD_US_WEST_1 ) @mock_aws def test_test_connection_env_partition_prefers_session_region(self): with ( mock.patch.dict( os.environ, { "PROWLER_AWS_PARTITION": AWS_GOV_CLOUD_PARTITION, "AWS_DEFAULT_REGION": AWS_REGION_GOV_CLOUD_US_WEST_1, }, clear=False, ), mock.patch.object( AwsProvider, "validate_credentials", return_value=AWSCallerIdentity( user_id="test-user-id", account=AWS_ACCOUNT_NUMBER, arn=ARN(AWS_GOV_CLOUD_ACCOUNT_ARN), region=AWS_REGION_GOV_CLOUD_US_WEST_1, ), ) as mock_validate_credentials, ): connection = AwsProvider.test_connection( aws_access_key_id="test-access-key", aws_secret_access_key="test-secret-key", raise_on_exception=False, ) assert connection.is_connected assert ( mock_validate_credentials.call_args.args[1] == AWS_REGION_GOV_CLOUD_US_WEST_1 ) @mock_aws def test_test_connection_role_env_partition_prefers_session_region(self): with ( mock.patch.dict( os.environ, { "PROWLER_AWS_PARTITION": AWS_GOV_CLOUD_PARTITION, "AWS_DEFAULT_REGION": AWS_REGION_GOV_CLOUD_US_WEST_1, }, clear=False, ), mock.patch.object( AwsProvider, "assume_role", return_value=AWSCredentials( aws_access_key_id="assumed-access-key", aws_secret_access_key="assumed-secret-key", aws_session_token="assumed-session-token", expiration=datetime.now(), ), ) as mock_assume_role, mock.patch.object( AwsProvider, "validate_credentials", return_value=AWSCallerIdentity( user_id="test-user-id", account=AWS_ACCOUNT_NUMBER, arn=ARN(AWS_GOV_CLOUD_ACCOUNT_ARN), region=AWS_REGION_GOV_CLOUD_US_WEST_1, ), ), ): connection = AwsProvider.test_connection( role_arn=f"arn:{AWS_GOV_CLOUD_PARTITION}:iam::{AWS_ACCOUNT_NUMBER}:role/test-role", aws_access_key_id="test-access-key", aws_secret_access_key="test-secret-key", raise_on_exception=False, ) assert connection.is_connected assumed_role_info = mock_assume_role.call_args.args[1] assert assumed_role_info.sts_region == AWS_REGION_GOV_CLOUD_US_WEST_1 @mock_aws def test_setup_session_mfa_env_partition_prefers_session_region(self): with ( mock.patch.dict( os.environ, { "PROWLER_AWS_PARTITION": AWS_GOV_CLOUD_PARTITION, "AWS_DEFAULT_REGION": AWS_REGION_GOV_CLOUD_US_WEST_1, }, clear=False, ), mock.patch.object( AwsProvider, "input_role_mfa_token_and_code", return_value=AWSMFAInfo( arn=f"arn:{AWS_GOV_CLOUD_PARTITION}:iam::{AWS_ACCOUNT_NUMBER}:mfa/test", totp="123456", ), ), mock.patch.object( AwsProvider, "create_sts_session", side_effect=AwsProvider.create_sts_session, ) as mock_create_sts_session, ): AwsProvider.setup_session( mfa=True, aws_access_key_id="test-access-key", aws_secret_access_key="test-secret-key", ) assert ( mock_create_sts_session.call_args.args[1] == AWS_REGION_GOV_CLOUD_US_WEST_1 ) @mock_aws def test_test_connection_env_partition_mismatch(self): with ( mock.patch.dict( os.environ, {"PROWLER_AWS_PARTITION": AWS_GOV_CLOUD_PARTITION}, clear=False, ), mock.patch.object( AwsProvider, "validate_credentials", return_value=AWSCallerIdentity( user_id="test-user-id", account=AWS_ACCOUNT_NUMBER, arn=ARN(AWS_ACCOUNT_ARN), region=AWS_REGION_US_EAST_1, ), ), ): connection = AwsProvider.test_connection( aws_access_key_id="test-access-key", aws_secret_access_key="test-secret-key", raise_on_exception=False, ) assert not connection.is_connected assert isinstance(connection.error, AWSInvalidPartitionError) def test_get_profile_region_avoids_excluded_session_region(self): mocked_session = mock.Mock(region_name=AWS_REGION_EU_WEST_1) assert ( AwsProvider.get_profile_region(mocked_session, {AWS_REGION_EU_WEST_1}) == AWS_REGION_US_EAST_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"} assert session_config.connect_timeout == BOTO3_CONNECT_TIMEOUT assert session_config.read_timeout == BOTO3_READ_TIMEOUT @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"} assert session_config.connect_timeout == BOTO3_CONNECT_TIMEOUT assert session_config.read_timeout == BOTO3_READ_TIMEOUT def test_set_session_config_0_max_attempts_disables_retries(self): session_config = AwsProvider.set_session_config(0) assert session_config.retries == {"max_attempts": 0, "mode": "standard"} @mock_aws def test_aws_provider_0_max_attempts_reaches_clients(self): aws_provider = AwsProvider(retries_max_attempts=0) client = aws_provider.session.current_session.client( "ec2", region_name=AWS_REGION_US_EAST_1 ) # botocore rewrites max_attempts into total_max_attempts (retries + 1) assert client.meta.config.retries["total_max_attempts"] == 1 def test_set_session_config_timeouts(self): session_config = AwsProvider.set_session_config( None, connect_timeout=2, read_timeout=15 ) assert session_config.retries == {"max_attempts": 3, "mode": "standard"} assert session_config.connect_timeout == 2 assert session_config.read_timeout == 15 @mock_aws def test_aws_provider_timeouts_reach_session_config(self): aws_provider = AwsProvider(connect_timeout=2, read_timeout=15) assert aws_provider.session.session_config.connect_timeout == 2 assert aws_provider.session.session_config.read_timeout == 15 @mock_aws def test_aws_set_up_session_forwards_timeouts(self): aws_session = AwsSetUpSession( aws_access_key_id="testing", aws_secret_access_key="testing", connect_timeout=2, read_timeout=15, ) assert aws_session._session.session_config.connect_timeout == 2 assert aws_session._session.session_config.read_timeout == 15 def test_get_default_session_config(self): config = get_default_session_config() assert config.user_agent_extra == BOTO3_USER_AGENT_EXTRA assert config.retries == {"max_attempts": 3, "mode": "standard"} assert config.connect_timeout == BOTO3_CONNECT_TIMEOUT assert config.read_timeout == BOTO3_READ_TIMEOUT def test_get_default_session_config_timeouts_from_env(self): with mock.patch.dict( os.environ, { "PROWLER_AWS_BOTO3_CONNECT_TIMEOUT": "3", "PROWLER_AWS_BOTO3_READ_TIMEOUT": "20", }, ): config = get_default_session_config() assert config.connect_timeout == 3 assert config.read_timeout == 20 def test_set_session_config_argument_overrides_env_timeouts(self): with mock.patch.dict(os.environ, {"PROWLER_AWS_BOTO3_CONNECT_TIMEOUT": "3"}): config = AwsProvider.set_session_config(None, connect_timeout=7) assert config.connect_timeout == 7 @pytest.mark.parametrize("raw", ["0", "-5", "ten", "1.5"]) def test_get_boto3_timeout_from_env_rejects_non_positive_integers(self, raw): with mock.patch.dict(os.environ, {"PROWLER_AWS_BOTO3_CONNECT_TIMEOUT": raw}): with raises( AWSInvalidBoto3TimeoutError, match="PROWLER_AWS_BOTO3_CONNECT_TIMEOUT" ): get_boto3_timeout_from_env("PROWLER_AWS_BOTO3_CONNECT_TIMEOUT", 10) def test_get_default_session_config_retries_from_env(self): with mock.patch.dict( os.environ, {"PROWLER_AWS_BOTO3_RETRIES_MAX_ATTEMPTS": "1"} ): config = get_default_session_config() assert config.retries == {"max_attempts": 1, "mode": "standard"} def test_get_default_session_config_retries_from_env_0_disables_retries(self): with mock.patch.dict( os.environ, {"PROWLER_AWS_BOTO3_RETRIES_MAX_ATTEMPTS": "0"} ): config = get_default_session_config() assert config.retries == {"max_attempts": 0, "mode": "standard"} def test_set_session_config_argument_overrides_env_retries(self): with mock.patch.dict( os.environ, {"PROWLER_AWS_BOTO3_RETRIES_MAX_ATTEMPTS": "1"} ): config = AwsProvider.set_session_config(5) assert config.retries == {"max_attempts": 5, "mode": "standard"} @mock_aws def test_aws_provider_without_retries_argument_uses_env_retries(self): with mock.patch.dict( os.environ, {"PROWLER_AWS_BOTO3_RETRIES_MAX_ATTEMPTS": "0"} ): aws_provider = AwsProvider() client = aws_provider.session.current_session.client( "ec2", region_name=AWS_REGION_US_EAST_1 ) # botocore rewrites max_attempts into total_max_attempts (retries + 1) assert client.meta.config.retries["total_max_attempts"] == 1 @mock_aws def test_aws_provider_retries_argument_overrides_env_retries(self): with mock.patch.dict( os.environ, {"PROWLER_AWS_BOTO3_RETRIES_MAX_ATTEMPTS": "0"} ): aws_provider = AwsProvider(retries_max_attempts=7) client = aws_provider.session.current_session.client( "ec2", region_name=AWS_REGION_US_EAST_1 ) assert client.meta.config.retries["total_max_attempts"] == 8 @mock_aws def test_aws_set_up_session_without_retries_argument_uses_env_retries(self): with mock.patch.dict( os.environ, {"PROWLER_AWS_BOTO3_RETRIES_MAX_ATTEMPTS": "1"} ): aws_session = AwsSetUpSession( aws_access_key_id="testing", aws_secret_access_key="testing", ) client = aws_session._session.current_session.client( "ec2", region_name=AWS_REGION_US_EAST_1 ) assert client.meta.config.retries["total_max_attempts"] == 2 def test_test_connection_session_uses_env_retries(self): with ( mock.patch.dict( os.environ, {"PROWLER_AWS_BOTO3_RETRIES_MAX_ATTEMPTS": "0"} ), mock.patch.object( AwsProvider, "validate_credentials", return_value=AWSCallerIdentity( user_id="test-user-id", account=AWS_ACCOUNT_NUMBER, arn=ARN(AWS_ACCOUNT_ARN), region=AWS_REGION_US_EAST_1, ), ) as mock_validate_credentials, ): connection = AwsProvider.test_connection( aws_access_key_id="test-access-key", aws_secret_access_key="test-secret-key", raise_on_exception=False, ) assert connection.is_connected validated_session = mock_validate_credentials.call_args.args[0] assert validated_session._session.get_default_client_config().retries == { "max_attempts": 0, "mode": "standard", } @pytest.mark.parametrize("raw", ["-1", "three", "1.5"]) def test_get_boto3_retries_from_env_rejects_anything_but_non_negative_integers( self, raw ): with mock.patch.dict( os.environ, {"PROWLER_AWS_BOTO3_RETRIES_MAX_ATTEMPTS": raw} ): with raises( AWSInvalidBoto3RetriesError, match="PROWLER_AWS_BOTO3_RETRIES_MAX_ATTEMPTS", ): get_boto3_retries_from_env("PROWLER_AWS_BOTO3_RETRIES_MAX_ATTEMPTS", 3) def test_get_boto3_retries_from_env_blank_falls_back_to_default(self): with mock.patch.dict( os.environ, {"PROWLER_AWS_BOTO3_RETRIES_MAX_ATTEMPTS": " "} ): assert ( get_boto3_retries_from_env( "PROWLER_AWS_BOTO3_RETRIES_MAX_ATTEMPTS", BOTO3_RETRIES_MAX_ATTEMPTS ) == BOTO3_RETRIES_MAX_ATTEMPTS ) def test_get_boto3_timeout_from_env_blank_falls_back_to_default(self): with mock.patch.dict(os.environ, {"PROWLER_AWS_BOTO3_CONNECT_TIMEOUT": " "}): assert ( get_boto3_timeout_from_env("PROWLER_AWS_BOTO3_CONNECT_TIMEOUT", 10) == 10 ) @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)