mirror of
https://github.com/ghndrx/prowler.git
synced 2026-02-12 15:55:09 +00:00
fix(assume_role): Set the AWS STS endpoint region (#2587)
This commit is contained in:
@@ -24,7 +24,7 @@ class Test_AWS_Provider:
|
||||
role_name = "test-role"
|
||||
role_arn = f"arn:aws:iam::{ACCOUNT_ID}:role/{role_name}"
|
||||
session_duration_seconds = 900
|
||||
audited_regions = "eu-west-1"
|
||||
audited_regions = ["eu-west-1"]
|
||||
sessionName = "ProwlerAsessmentSession"
|
||||
# Boto 3 client to create our user
|
||||
iam_client = boto3.client("iam", region_name="us-east-1")
|
||||
@@ -105,7 +105,7 @@ class Test_AWS_Provider:
|
||||
role_name = "test-role"
|
||||
role_arn = f"arn:aws:iam::{ACCOUNT_ID}:role/{role_name}"
|
||||
session_duration_seconds = 900
|
||||
audited_regions = "eu-west-1"
|
||||
audited_regions = ["eu-west-1"]
|
||||
sessionName = "ProwlerAsessmentSession"
|
||||
# Boto 3 client to create our user
|
||||
iam_client = boto3.client("iam", region_name="us-east-1")
|
||||
@@ -184,6 +184,89 @@ class Test_AWS_Provider:
|
||||
"AssumedRoleId"
|
||||
].should.have.length_of(21 + 1 + len(sessionName))
|
||||
|
||||
@mock_iam
|
||||
@mock_sts
|
||||
def test_assume_role_with_sts_endpoint_region(self):
|
||||
# Variables
|
||||
role_name = "test-role"
|
||||
role_arn = f"arn:aws:iam::{ACCOUNT_ID}:role/{role_name}"
|
||||
session_duration_seconds = 900
|
||||
aws_region = "eu-west-1"
|
||||
sts_endpoint_region = aws_region
|
||||
audited_regions = [aws_region]
|
||||
sessionName = "ProwlerAsessmentSession"
|
||||
# Boto 3 client to create our user
|
||||
iam_client = boto3.client("iam", region_name=aws_region)
|
||||
# IAM user
|
||||
iam_user = iam_client.create_user(UserName="test-user")["User"]
|
||||
access_key = iam_client.create_access_key(UserName=iam_user["UserName"])[
|
||||
"AccessKey"
|
||||
]
|
||||
access_key_id = access_key["AccessKeyId"]
|
||||
secret_access_key = access_key["SecretAccessKey"]
|
||||
# New Boto3 session with the previously create user
|
||||
session = boto3.session.Session(
|
||||
aws_access_key_id=access_key_id,
|
||||
aws_secret_access_key=secret_access_key,
|
||||
region_name=aws_region,
|
||||
)
|
||||
|
||||
# Fulfil the input session object for Prowler
|
||||
audit_info = AWS_Audit_Info(
|
||||
session_config=None,
|
||||
original_session=session,
|
||||
audit_session=None,
|
||||
audited_account=None,
|
||||
audited_account_arn=None,
|
||||
audited_partition=None,
|
||||
audited_identity_arn=None,
|
||||
audited_user_id=None,
|
||||
profile=None,
|
||||
profile_region=None,
|
||||
credentials=None,
|
||||
assumed_role_info=AWS_Assume_Role(
|
||||
role_arn=role_arn,
|
||||
session_duration=session_duration_seconds,
|
||||
external_id=None,
|
||||
mfa_enabled=False,
|
||||
),
|
||||
audited_regions=audited_regions,
|
||||
organizations_metadata=None,
|
||||
audit_resources=None,
|
||||
mfa_enabled=False,
|
||||
)
|
||||
|
||||
# Call assume_role
|
||||
aws_provider = AWS_Provider(audit_info)
|
||||
assume_role_response = assume_role(
|
||||
aws_provider.aws_session, aws_provider.role_info, sts_endpoint_region
|
||||
)
|
||||
# Recover credentials for the assume role operation
|
||||
credentials = assume_role_response["Credentials"]
|
||||
# Test the response
|
||||
# SessionToken
|
||||
credentials["SessionToken"].should.have.length_of(356)
|
||||
credentials["SessionToken"].startswith("FQoGZXIvYXdzE")
|
||||
# AccessKeyId
|
||||
credentials["AccessKeyId"].should.have.length_of(20)
|
||||
credentials["AccessKeyId"].startswith("ASIA")
|
||||
# SecretAccessKey
|
||||
credentials["SecretAccessKey"].should.have.length_of(40)
|
||||
# Assumed Role
|
||||
assume_role_response["AssumedRoleUser"]["Arn"].should.equal(
|
||||
f"arn:aws:sts::{ACCOUNT_ID}:assumed-role/{role_name}/{sessionName}"
|
||||
)
|
||||
# AssumedRoleUser
|
||||
assert assume_role_response["AssumedRoleUser"]["AssumedRoleId"].startswith(
|
||||
"AROA"
|
||||
)
|
||||
assert assume_role_response["AssumedRoleUser"]["AssumedRoleId"].endswith(
|
||||
":" + sessionName
|
||||
)
|
||||
assume_role_response["AssumedRoleUser"]["AssumedRoleId"].should.have.length_of(
|
||||
21 + 1 + len(sessionName)
|
||||
)
|
||||
|
||||
def test_generate_regional_clients(self):
|
||||
# New Boto3 session with the previously create user
|
||||
session = boto3.session.Session(
|
||||
|
||||
@@ -188,6 +188,47 @@ class Test_AWS_Credentials:
|
||||
assert re.match("[0-9a-zA-Z]{20}", get_caller_identity["UserId"])
|
||||
assert get_caller_identity["Account"] == AWS_ACCOUNT_NUMBER
|
||||
|
||||
@mock_sts
|
||||
@mock_iam
|
||||
def test_validate_credentials_commercial_partition_with_regions_none_and_profile_region_but_sts_endpoint_region(
|
||||
self,
|
||||
):
|
||||
# AWS Region for AWS COMMERCIAL
|
||||
aws_region = "eu-west-1"
|
||||
sts_endpoint_region = aws_region
|
||||
aws_partition = "aws"
|
||||
# Create a mock IAM user
|
||||
iam_client = boto3.client("iam", region_name=aws_region)
|
||||
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
|
||||
session = boto3.session.Session(
|
||||
aws_access_key_id=access_key_id,
|
||||
aws_secret_access_key=secret_access_key,
|
||||
region_name=aws_region,
|
||||
)
|
||||
|
||||
get_caller_identity = validate_aws_credentials(
|
||||
session, None, sts_endpoint_region
|
||||
)
|
||||
|
||||
assert get_caller_identity["region"] == aws_region
|
||||
|
||||
caller_identity_arn = parse_iam_credentials_arn(get_caller_identity["Arn"])
|
||||
|
||||
assert caller_identity_arn.partition == aws_partition
|
||||
assert caller_identity_arn.region is None
|
||||
assert caller_identity_arn.resource == "test-user"
|
||||
assert caller_identity_arn.resource_type == "user"
|
||||
assert re.match("[0-9a-zA-Z]{20}", get_caller_identity["UserId"])
|
||||
assert get_caller_identity["Account"] == AWS_ACCOUNT_NUMBER
|
||||
|
||||
@mock_sts
|
||||
@mock_iam
|
||||
def test_validate_credentials_china_partition_without_regions_and_profile_region_so_us_east_1(
|
||||
@@ -268,6 +309,53 @@ class Test_AWS_Credentials:
|
||||
assert re.match("[0-9a-zA-Z]{20}", get_caller_identity["UserId"])
|
||||
assert get_caller_identity["Account"] == AWS_ACCOUNT_NUMBER
|
||||
|
||||
@mock_sts
|
||||
@mock_iam
|
||||
@patch(
|
||||
"botocore.client.BaseClient._make_api_call", new=mock_get_caller_identity_china
|
||||
)
|
||||
def test_validate_credentials_china_partition_without_regions_but_sts_endpoint_region(
|
||||
self,
|
||||
):
|
||||
# AWS Region for AWS CHINA
|
||||
aws_region = "cn-north-1"
|
||||
sts_endpoint_region = aws_region
|
||||
aws_partition = "aws-cn"
|
||||
# Create a mock IAM user
|
||||
iam_client = boto3.client("iam", region_name=aws_region)
|
||||
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
|
||||
session = boto3.session.Session(
|
||||
aws_access_key_id=access_key_id,
|
||||
aws_secret_access_key=secret_access_key,
|
||||
region_name=aws_region,
|
||||
)
|
||||
|
||||
get_caller_identity = validate_aws_credentials(
|
||||
session, None, sts_endpoint_region
|
||||
)
|
||||
|
||||
# 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
|
||||
assert get_caller_identity["region"] == aws_region
|
||||
|
||||
caller_identity_arn = parse_iam_credentials_arn(get_caller_identity["Arn"])
|
||||
|
||||
assert caller_identity_arn.partition == aws_partition
|
||||
assert caller_identity_arn.region is None
|
||||
assert caller_identity_arn.resource == "test-user"
|
||||
assert caller_identity_arn.resource_type == "user"
|
||||
assert re.match("[0-9a-zA-Z]{20}", get_caller_identity["UserId"])
|
||||
assert get_caller_identity["Account"] == AWS_ACCOUNT_NUMBER
|
||||
|
||||
@mock_sts
|
||||
@mock_iam
|
||||
@patch(
|
||||
@@ -310,3 +398,51 @@ class Test_AWS_Credentials:
|
||||
assert caller_identity_arn.resource_type == "user"
|
||||
assert re.match("[0-9a-zA-Z]{20}", get_caller_identity["UserId"])
|
||||
assert get_caller_identity["Account"] == AWS_ACCOUNT_NUMBER
|
||||
|
||||
@mock_sts
|
||||
@mock_iam
|
||||
@patch(
|
||||
"botocore.client.BaseClient._make_api_call",
|
||||
new=mock_get_caller_identity_gov_cloud,
|
||||
)
|
||||
def test_validate_credentials_gov_cloud_partition_without_regions_but_sts_endpoint_region(
|
||||
self,
|
||||
):
|
||||
# AWS Region for US GOV CLOUD
|
||||
aws_region = "us-gov-east-1"
|
||||
sts_endpoint_region = aws_region
|
||||
aws_partition = "aws-us-gov"
|
||||
# Create a mock IAM user
|
||||
iam_client = boto3.client("iam", region_name=aws_region)
|
||||
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
|
||||
session = boto3.session.Session(
|
||||
aws_access_key_id=access_key_id,
|
||||
aws_secret_access_key=secret_access_key,
|
||||
region_name=aws_region,
|
||||
)
|
||||
|
||||
get_caller_identity = validate_aws_credentials(
|
||||
session, None, sts_endpoint_region
|
||||
)
|
||||
|
||||
# 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
|
||||
assert get_caller_identity["region"] == aws_region
|
||||
|
||||
caller_identity_arn = parse_iam_credentials_arn(get_caller_identity["Arn"])
|
||||
|
||||
assert caller_identity_arn.partition == aws_partition
|
||||
assert caller_identity_arn.region is None
|
||||
assert caller_identity_arn.resource == "test-user"
|
||||
assert caller_identity_arn.resource_type == "user"
|
||||
assert re.match("[0-9a-zA-Z]{20}", get_caller_identity["UserId"])
|
||||
assert get_caller_identity["Account"] == AWS_ACCOUNT_NUMBER
|
||||
|
||||
Reference in New Issue
Block a user