feat(aws): add new check iam_root_credentials_management_enabled (#5801)

This commit is contained in:
Sergio Garcia
2024-11-18 10:59:35 -05:00
committed by GitHub
parent 8ddb9fbb84
commit c69571abcd
21 changed files with 541 additions and 308 deletions
@@ -5,35 +5,36 @@ from prowler.providers.aws.services.iam.iam_client import iam_client
class iam_no_root_access_key(Check):
def execute(self) -> Check_Report_AWS:
findings = []
response = iam_client.credential_report
for user in response:
if user["user"] == "<root_account>":
report = Check_Report_AWS(self.metadata())
report.region = iam_client.region
report.resource_id = user["user"]
report.resource_arn = user["arn"]
if (
user["access_key_1_active"] == "false"
and user["access_key_2_active"] == "false"
):
report.status = "PASS"
report.status_extended = (
f"User {user['user']} does not have access keys."
)
elif (
user["access_key_1_active"] == "true"
and user["access_key_2_active"] == "true"
):
report.status = "FAIL"
report.status_extended = (
f"User {user['user']} has two active access keys."
)
else:
report.status = "FAIL"
report.status_extended = (
f"User {user['user']} has one active access key."
)
findings.append(report)
# Check if the root credentials are managed by AWS Organizations
if "RootCredentialsManagement" not in iam_client.organization_features:
for user in iam_client.credential_report:
if user["user"] == "<root_account>":
report = Check_Report_AWS(self.metadata())
report.region = iam_client.region
report.resource_id = user["user"]
report.resource_arn = user["arn"]
if (
user["access_key_1_active"] == "false"
and user["access_key_2_active"] == "false"
):
report.status = "PASS"
report.status_extended = (
"Root account does not have access keys."
)
elif (
user["access_key_1_active"] == "true"
and user["access_key_2_active"] == "true"
):
report.status = "FAIL"
report.status_extended = (
"Root account has two active access keys."
)
else:
report.status = "FAIL"
report.status_extended = (
"Root account has one active access key."
)
findings.append(report)
break
return findings
@@ -0,0 +1,33 @@
{
"Provider": "aws",
"CheckID": "iam_root_credentials_management_enabled",
"CheckTitle": "Ensure centralized root credentials management is enabled",
"CheckType": [],
"ServiceName": "iam",
"SubServiceName": "",
"ResourceIdTemplate": "arn:partition:service:region:account-id:resource-id",
"Severity": "high",
"ResourceType": "Other",
"Description": "Checks if centralized management of root credentials for member accounts in AWS Organizations is enabled. This ensures that root credentials are managed centrally, reducing the risk of unauthorized access or mismanagement.",
"Risk": "Without centralized root credentials management, member accounts retain full control over their root user credentials, increasing the risk of credential misuse, mismanagement, or compromise.",
"RelatedUrl": "https://docs.aws.amazon.com/IAM/latest/UserGuide/id_root-user.html#id_root-user-access-management",
"Remediation": {
"Code": {
"CLI": "aws iam enable-organizations-root-credentials-management",
"NativeIaC": "",
"Other": "",
"Terraform": ""
},
"Recommendation": {
"Text": "Enable centralized management of root access for member accounts using the CLI or IAM console.",
"Url": "https://docs.aws.amazon.com/IAM/latest/UserGuide/id_root-enable-root-access.html"
}
},
"Categories": [],
"DependsOn": [],
"RelatedTo": [
"iam_root_hardware_mfa_enabled",
"iam_root_mfa_enabled"
],
"Notes": ""
}
@@ -0,0 +1,27 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.iam.iam_client import iam_client
from prowler.providers.aws.services.organizations.organizations_client import (
organizations_client,
)
class iam_root_credentials_management_enabled(Check):
def execute(self) -> Check_Report_AWS:
findings = []
if (
organizations_client.organization
and organizations_client.organization.status == "ACTIVE"
):
report = Check_Report_AWS(self.metadata())
report.region = iam_client.region
report.resource_arn = iam_client.audited_account_arn
report.resource_id = iam_client.audited_account
if "RootCredentialsManagement" in iam_client.organization_features:
report.status = "PASS"
report.status_extended = "Root credentials management is enabled."
else:
report.status = "FAIL"
report.status_extended = "Root credentials management is not enabled."
findings.append(report)
return findings
@@ -5,31 +5,36 @@ from prowler.providers.aws.services.iam.iam_client import iam_client
class iam_root_hardware_mfa_enabled(Check):
def execute(self) -> Check_Report_AWS:
findings = []
# This check is only avaible in Commercial Partition
# This check is only available in Commercial Partition
if iam_client.audited_partition == "aws":
if iam_client.account_summary:
virtual_mfa = False
report = Check_Report_AWS(self.metadata())
report.region = iam_client.region
report.resource_id = "<root_account>"
report.resource_arn = iam_client.mfa_arn_template
# Check if the root credentials are managed by AWS Organizations
if "RootCredentialsManagement" not in iam_client.organization_features:
if iam_client.account_summary:
virtual_mfa = False
report = Check_Report_AWS(self.metadata())
report.region = iam_client.region
report.resource_id = "<root_account>"
report.resource_arn = iam_client.mfa_arn_template
if iam_client.account_summary["SummaryMap"]["AccountMFAEnabled"] > 0:
for mfa in iam_client.virtual_mfa_devices:
# If the ARN of the associated IAM user of the Virtual MFA device is "arn:aws:iam::[aws-account-id]:root", your AWS root account is not using a hardware-based MFA device for MFA protection.
if "root" in mfa.get("User", {}).get("Arn", ""):
virtual_mfa = True
report.status = "FAIL"
report.status_extended = "Root account has a virtual MFA instead of a hardware MFA device enabled."
if not virtual_mfa:
report.status = "PASS"
report.status_extended = (
"Root account has a hardware MFA device enabled."
)
else:
report.status = "FAIL"
report.status_extended = "MFA is not enabled for root account."
if (
iam_client.account_summary["SummaryMap"]["AccountMFAEnabled"]
> 0
):
for mfa in iam_client.virtual_mfa_devices:
# If the ARN of the associated IAM user of the Virtual MFA device is "arn:aws:iam::[aws-account-id]:root", your AWS root account is not using a hardware-based MFA device for MFA protection.
if "root" in mfa.get("User", {}).get("Arn", ""):
virtual_mfa = True
report.status = "FAIL"
report.status_extended = "Root account has a virtual MFA instead of a hardware MFA device enabled."
if not virtual_mfa:
report.status = "PASS"
report.status_extended = (
"Root account has a hardware MFA device enabled."
)
else:
report.status = "FAIL"
report.status_extended = "MFA is not enabled for root account."
findings.append(report)
findings.append(report)
return findings
@@ -5,19 +5,23 @@ from prowler.providers.aws.services.iam.iam_client import iam_client
class iam_root_mfa_enabled(Check):
def execute(self) -> Check_Report_AWS:
findings = []
if iam_client.credential_report:
for user in iam_client.credential_report:
if user["user"] == "<root_account>":
report = Check_Report_AWS(self.metadata())
report.region = iam_client.region
report.resource_id = user["user"]
report.resource_arn = user["arn"]
if user["mfa_active"] == "false":
report.status = "FAIL"
report.status_extended = "MFA is not enabled for root account."
else:
report.status = "PASS"
report.status_extended = "MFA is enabled for root account."
findings.append(report)
# Check if the root credentials are managed by AWS Organizations
if "RootCredentialsManagement" not in iam_client.organization_features:
if iam_client.credential_report:
for user in iam_client.credential_report:
if user["user"] == "<root_account>":
report = Check_Report_AWS(self.metadata())
report.region = iam_client.region
report.resource_id = user["user"]
report.resource_arn = user["arn"]
if user["mfa_active"] == "false":
report.status = "FAIL"
report.status_extended = (
"MFA is not enabled for root account."
)
else:
report.status = "PASS"
report.status_extended = "MFA is enabled for root account."
findings.append(report)
return findings
@@ -102,6 +102,8 @@ class IAM(AWSService):
self._get_last_accessed_services()
self.user_temporary_credentials_usage = {}
self._get_user_temporary_credentials_usage()
self.organization_features = []
self._list_organizations_features()
# List missing tags
self.__threading_call__(self._list_tags, self.users)
self.__threading_call__(self._list_tags, self.roles)
@@ -964,6 +966,31 @@ class IAM(AWSService):
f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
def _list_organizations_features(self):
logger.info("IAM - List Organization Features...")
try:
organization_features = self.client.list_organizations_features()
self.organization_features = organization_features.get(
"EnabledFeatures", []
)
except ClientError as error:
if error.response["Error"]["Code"] == "OrganizationNotFoundException":
logger.warning(
f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
elif error.response["Error"]["Code"] == "ServiceAccessNotEnabledException":
logger.warning(
f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
else:
logger.error(
f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
except Exception as error:
logger.error(
f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
class MFADevice(BaseModel):
serial_number: str
@@ -7,21 +7,19 @@ from prowler.providers.aws.services.organizations.organizations_client import (
class organizations_account_part_of_organizations(Check):
def execute(self):
findings = []
for org in organizations_client.organizations:
if organizations_client.organization:
report = Check_Report_AWS(self.metadata())
if org.status == "ACTIVE":
if organizations_client.organization.status == "ACTIVE":
report.status = "PASS"
report.status_extended = (
f"AWS Organization {org.id} contains this AWS account."
)
report.status_extended = f"AWS Organization {organizations_client.organization.id} contains this AWS account."
else:
report.status = "FAIL"
report.status_extended = (
"AWS Organizations is not in-use for this AWS Account."
)
report.region = organizations_client.region
report.resource_id = org.id
report.resource_arn = org.arn
report.resource_id = organizations_client.organization.id
report.resource_arn = organizations_client.organization.arn
findings.append(report)
return findings
@@ -14,31 +14,33 @@ class organizations_delegated_administrators(Check):
)
)
for org in organizations_client.organizations:
if org.status == "ACTIVE":
report = Check_Report_AWS(self.metadata())
report.resource_id = org.id
report.resource_arn = org.arn
report.region = organizations_client.region
if org.delegated_administrators is None:
# Access Denied to list_policies
continue
if org.delegated_administrators:
for delegated_administrator in org.delegated_administrators:
if (
organizations_client.organization
and organizations_client.organization.status == "ACTIVE"
):
report = Check_Report_AWS(self.metadata())
report.resource_id = organizations_client.organization.id
report.resource_arn = organizations_client.organization.arn
report.region = organizations_client.region
if (
organizations_client.organization.delegated_administrators is not None
): # Check if Access Denied to list_delegated_administrators
if organizations_client.organization.delegated_administrators:
for (
delegated_administrator
) in organizations_client.organization.delegated_administrators:
if (
delegated_administrator.id
not in organizations_trusted_delegated_administrators
):
report.status = "FAIL"
report.status_extended = f"AWS Organization {org.id} has an untrusted Delegated Administrator: {delegated_administrator.id}."
report.status_extended = f"AWS Organization {organizations_client.organization.id} has an untrusted Delegated Administrator: {delegated_administrator.id}."
else:
report.status = "PASS"
report.status_extended = f"AWS Organization {org.id} has a trusted Delegated Administrator: {delegated_administrator.id}."
report.status_extended = f"AWS Organization {organizations_client.organization.id} has a trusted Delegated Administrator: {delegated_administrator.id}."
else:
report.status = "PASS"
report.status_extended = (
f"AWS Organization {org.id} has no Delegated Administrators."
)
report.status_extended = f"AWS Organization {organizations_client.organization.id} has no Delegated Administrators."
findings.append(report)
@@ -8,19 +8,23 @@ class organizations_opt_out_ai_services_policy(Check):
def execute(self):
findings = []
for org in organizations_client.organizations:
if org.policies is not None: # Access Denied to list_policies
if organizations_client.organization:
if (
organizations_client.organization.policies is not None
): # Access Denied to list_policies
report = Check_Report_AWS(self.metadata())
report.resource_id = org.id
report.resource_arn = org.arn
report.resource_id = organizations_client.organization.id
report.resource_arn = organizations_client.organization.arn
report.region = organizations_client.region
report.status = "FAIL"
report.status_extended = (
"AWS Organizations is not in-use for this AWS Account."
)
if org.status == "ACTIVE":
report.status_extended = f"AWS Organization {org.id} has not opted out of all AI services, granting consent for AWS to access its data."
for policy in org.policies.get("AISERVICES_OPT_OUT_POLICY", []):
if organizations_client.organization.status == "ACTIVE":
report.status_extended = f"AWS Organization {organizations_client.organization.id} has not opted out of all AI services, granting consent for AWS to access its data."
for policy in organizations_client.organization.policies.get(
"AISERVICES_OPT_OUT_POLICY", []
):
if (
policy.content.get("services", {})
.get("default", {})
@@ -29,7 +33,7 @@ class organizations_opt_out_ai_services_policy(Check):
== "optOut"
):
report.status = "PASS"
report.status_extended = f"AWS Organization {org.id} has opted out of all AI services, not granting consent for AWS to access its data."
report.status_extended = f"AWS Organization {organizations_client.organization.id} has opted out of all AI services, not granting consent for AWS to access its data."
break
findings.append(report)
@@ -11,25 +11,27 @@ class organizations_scp_check_deny_regions(Check):
"organizations_enabled_regions", []
)
for org in organizations_client.organizations:
if org.policies is not None: # Access denied to list policies
if organizations_client.organization:
if (
organizations_client.organization.policies is not None
): # Access denied to list policies
report = Check_Report_AWS(self.metadata())
report.resource_id = org.id
report.resource_arn = org.arn
report.resource_id = organizations_client.organization.id
report.resource_arn = organizations_client.organization.arn
report.region = organizations_client.region
report.status = "FAIL"
report.status_extended = (
"AWS Organizations is not in-use for this AWS Account."
)
if org.status == "ACTIVE":
report.status_extended = (
f"AWS Organizations {org.id} does not have SCP policies."
)
if organizations_client.organization.status == "ACTIVE":
report.status_extended = f"AWS Organizations {organizations_client.organization.id} does not have SCP policies."
# We use this flag if we find a statement that is restricting regions but not all the configured ones:
is_region_restricted_statement = False
for policy in org.policies.get("SERVICE_CONTROL_POLICY", []):
for policy in organizations_client.organization.policies.get(
"SERVICE_CONTROL_POLICY", []
):
# Statements are not always list
statements = policy.content.get("Statement")
@@ -54,14 +56,14 @@ class organizations_scp_check_deny_regions(Check):
):
# All defined regions are restricted, we exit here, no need to continue.
report.status = "PASS"
report.status_extended = f"AWS Organization {org.id} has SCP policy {policy.id} restricting all configured regions found."
report.status_extended = f"AWS Organization {organizations_client.organization.id} has SCP policy {policy.id} restricting all configured regions found."
findings.append(report)
return findings
else:
# Regions are restricted, but not the ones defined, we keep this finding, but we continue analyzing:
is_region_restricted_statement = True
report.status = "FAIL"
report.status_extended = f"AWS Organization {org.id} has SCP policies {policy.id} restricting some AWS Regions, but not all the configured ones, please check config."
report.status_extended = f"AWS Organization {organizations_client.organization.id} has SCP policies {policy.id} restricting some AWS Regions, but not all the configured ones, please check config."
# Allow if Condition = {"StringEquals": {"aws:RequestedRegion": [region1, region2]}}
if (
@@ -80,18 +82,18 @@ class organizations_scp_check_deny_regions(Check):
):
# All defined regions are restricted, we exit here, no need to continue.
report.status = "PASS"
report.status_extended = f"AWS Organization {org.id} has SCP policy {policy.id} restricting all configured regions found."
report.status_extended = f"AWS Organization {organizations_client.organization.id} has SCP policy {policy.id} restricting all configured regions found."
findings.append(report)
return findings
else:
# Regions are restricted, but not the ones defined, we keep this finding, but we continue analyzing:
is_region_restricted_statement = True
report.status = "FAIL"
report.status_extended = f"AWS Organization {org.id} has SCP policies {policy.id} restricting some AWS Regions, but not all the configured ones, please check config."
report.status_extended = f"AWS Organization {organizations_client.organization.id} has SCP policies {policy.id} restricting some AWS Regions, but not all the configured ones, please check config."
if not is_region_restricted_statement:
report.status = "FAIL"
report.status_extended = f"AWS Organization {org.id} has SCP policies but don't restrict AWS Regions."
report.status_extended = f"AWS Organization {organizations_client.organization.id} has SCP policies but don't restrict AWS Regions."
findings.append(report)
@@ -20,7 +20,7 @@ class Organizations(AWSService):
def __init__(self, provider):
# Call AWSService's __init__
super().__init__(__class__.__name__, provider)
self.organizations = []
self.organization = None
self.policies = {}
self.delegated_administrators = []
self._describe_organization()
@@ -43,13 +43,11 @@ class Organizations(AWSService):
error.response["Error"]["Code"]
== "AWSOrganizationsNotInUseException"
):
self.organizations.append(
Organization(
arn=self.audited_account_arn,
id="AWS Organization",
status="NOT_AVAILABLE",
master_id="",
)
self.organization = Organization(
arn=self.audited_account_arn,
id="AWS Organization",
status="NOT_AVAILABLE",
master_id="",
)
else:
logger.error(
@@ -59,25 +57,20 @@ class Organizations(AWSService):
if not self.audit_resources or (
is_resource_filtered(organization_arn, self.audit_resources)
):
self.organizations.append(
Organization(
arn=organization_arn,
id=organization_id,
status="ACTIVE",
master_id=organization_master_id,
policies=organization_policies,
delegated_administrators=organization_delegated_administrator,
)
self.organization = Organization(
arn=organization_arn,
id=organization_id,
status="ACTIVE",
master_id=organization_master_id,
policies=organization_policies,
delegated_administrators=organization_delegated_administrator,
)
else:
# is filtered
self.organizations.append(
Organization(
arn=self.audited_account_arn,
id="AWS Organization",
status="NOT_AVAILABLE",
master_id="",
)
self.organization = Organization(
arn=self.audited_account_arn,
id="AWS Organization",
status="NOT_AVAILABLE",
master_id="",
)
except Exception as error:
@@ -8,26 +8,28 @@ class organizations_tags_policies_enabled_and_attached(Check):
def execute(self):
findings = []
for org in organizations_client.organizations:
if org.policies is not None: # Access Denied to list_policies
if organizations_client.organization:
if (
organizations_client.organization.policies is not None
): # Access Denied to list_policies
report = Check_Report_AWS(self.metadata())
report.resource_id = org.id
report.resource_arn = org.arn
report.resource_id = organizations_client.organization.id
report.resource_arn = organizations_client.organization.arn
report.region = organizations_client.region
report.status = "FAIL"
report.status_extended = (
"AWS Organizations is not in-use for this AWS Account."
)
if org.status == "ACTIVE":
report.status_extended = (
f"AWS Organizations {org.id} does not have tag policies."
)
for policy in org.policies.get("TAG_POLICY", []):
report.status_extended = f"AWS Organization {org.id} has tag policies enabled but not attached."
if organizations_client.organization.status == "ACTIVE":
report.status_extended = f"AWS Organizations {organizations_client.organization.id} does not have tag policies."
for policy in organizations_client.organization.policies.get(
"TAG_POLICY", []
):
report.status_extended = f"AWS Organization {organizations_client.organization.id} has tag policies enabled but not attached."
if policy.targets:
report.status = "PASS"
report.status_extended = f"AWS Organization {org.id} has tag policies enabled and attached to an AWS account."
report.status_extended = f"AWS Organization {organizations_client.organization.id} has tag policies enabled and attached to an AWS account."
findings.append(report)
@@ -10,23 +10,25 @@ from prowler.providers.aws.services.servicecatalog.servicecatalog_client import
class servicecatalog_portfolio_shared_within_organization_only(Check):
def execute(self):
findings = []
for org in organizations_client.organizations:
if org.status == "ACTIVE":
for portfolio in servicecatalog_client.portfolios.values():
if portfolio.shares is not None:
report = Check_Report_AWS(self.metadata())
report.region = portfolio.region
report.resource_id = portfolio.id
report.resource_arn = portfolio.arn
report.resource_tags = portfolio.tags
report.status = "PASS"
report.status_extended = f"ServiceCatalog Portfolio {portfolio.name} is shared within your AWS Organization."
for portfolio_share in portfolio.shares:
if portfolio_share.type == "ACCOUNT":
report.status = "FAIL"
report.status_extended = f"ServiceCatalog Portfolio {portfolio.name} is shared with an account."
break
if (
organizations_client.organization
and organizations_client.organization.status == "ACTIVE"
):
for portfolio in servicecatalog_client.portfolios.values():
if portfolio.shares is not None:
report = Check_Report_AWS(self.metadata())
report.region = portfolio.region
report.resource_id = portfolio.id
report.resource_arn = portfolio.arn
report.resource_tags = portfolio.tags
report.status = "PASS"
report.status_extended = f"ServiceCatalog Portfolio {portfolio.name} is shared within your AWS Organization."
for portfolio_share in portfolio.shares:
if portfolio_share.type == "ACCOUNT":
report.status = "FAIL"
report.status_extended = f"ServiceCatalog Portfolio {portfolio.name} is shared with an account."
break
findings.append(report)
findings.append(report)
return findings
@@ -1,4 +1,3 @@
from re import search
from unittest import mock
from boto3 import client
@@ -39,11 +38,10 @@ class Test_iam_no_root_access_key_test:
check = iam_no_root_access_key()
result = check.execute()
# raise Exception
assert result[0].status == "PASS"
assert search(
"User <root_account> does not have access keys.",
result[0].status_extended,
assert (
result[0].status_extended
== "Root account does not have access keys."
)
assert result[0].resource_id == "<root_account>"
assert (
@@ -82,11 +80,10 @@ class Test_iam_no_root_access_key_test:
check = iam_no_root_access_key()
result = check.execute()
# raise Exception
assert result[0].status == "FAIL"
assert search(
"User <root_account> has one active access key.",
result[0].status_extended,
assert (
result[0].status_extended
== "Root account has one active access key."
)
assert result[0].resource_id == "<root_account>"
assert (
@@ -125,11 +122,10 @@ class Test_iam_no_root_access_key_test:
check = iam_no_root_access_key()
result = check.execute()
# raise Exception
assert result[0].status == "FAIL"
assert search(
"User <root_account> has one active access key.",
result[0].status_extended,
assert (
result[0].status_extended
== "Root account has one active access key."
)
assert result[0].resource_id == "<root_account>"
assert (
@@ -168,11 +164,10 @@ class Test_iam_no_root_access_key_test:
check = iam_no_root_access_key()
result = check.execute()
# raise Exception
assert result[0].status == "FAIL"
assert search(
"User <root_account> has two active access key.",
result[0].status_extended,
assert (
result[0].status_extended
== "Root account has two active access keys."
)
assert result[0].resource_id == "<root_account>"
assert (
@@ -0,0 +1,154 @@
from unittest import mock
import botocore
from boto3 import client
from moto import mock_aws
from tests.providers.aws.utils import (
AWS_ACCOUNT_ARN,
AWS_ACCOUNT_NUMBER,
AWS_REGION_US_EAST_1,
set_mocked_aws_provider,
)
# Original botocore _make_api_call function
orig = botocore.client.BaseClient._make_api_call
# Mocked botocore _make_api_call function
def mock_make_api_call_enabled(self, operation_name, kwarg):
if operation_name == "ListOrganizationsFeatures":
return {
"OrganizationId": "o-test",
"EnabledFeatures": ["RootSessions", "RootCredentialsManagement"],
}
# If we don't want to patch the API call
return orig(self, operation_name, kwarg)
def mock_make_api_call_disabled(self, operation_name, kwarg):
if operation_name == "ListOrganizationsFeatures":
return {"OrganizationId": "o-test", "EnabledFeatures": []}
# If we don't want to patch the API call
return orig(self, operation_name, kwarg)
class Test_iam_root_credentials_management_enabled_test:
@mock_aws
def test_no_organization(self):
from prowler.providers.aws.services.iam.iam_service import IAM
from prowler.providers.aws.services.organizations.organizations_service import (
Organizations,
)
aws_provider = set_mocked_aws_provider(
[AWS_REGION_US_EAST_1], create_default_organization=False
)
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.iam.iam_root_credentials_management_enabled.iam_root_credentials_management_enabled.iam_client",
new=IAM(aws_provider),
), mock.patch(
"prowler.providers.aws.services.iam.iam_root_credentials_management_enabled.iam_root_credentials_management_enabled.organizations_client",
new=Organizations(aws_provider),
):
from prowler.providers.aws.services.iam.iam_root_credentials_management_enabled.iam_root_credentials_management_enabled import (
iam_root_credentials_management_enabled,
)
check = iam_root_credentials_management_enabled()
result = check.execute()
assert len(result) == 0
@mock.patch(
"botocore.client.BaseClient._make_api_call", new=mock_make_api_call_enabled
)
@mock_aws
def test__root_credentials_management_enabled(self):
# Create Organization
conn = client("organizations")
conn.create_organization()
from prowler.providers.aws.services.iam.iam_service import IAM
from prowler.providers.aws.services.organizations.organizations_service import (
Organizations,
)
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.iam.iam_root_credentials_management_enabled.iam_root_credentials_management_enabled.iam_client",
new=IAM(aws_provider),
), mock.patch(
"prowler.providers.aws.services.iam.iam_root_credentials_management_enabled.iam_root_credentials_management_enabled.organizations_client",
new=Organizations(aws_provider),
):
from prowler.providers.aws.services.iam.iam_root_credentials_management_enabled.iam_root_credentials_management_enabled import (
iam_root_credentials_management_enabled,
)
check = iam_root_credentials_management_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "Root credentials management is enabled."
)
assert result[0].resource_id == AWS_ACCOUNT_NUMBER
assert result[0].resource_arn == AWS_ACCOUNT_ARN
assert result[0].region == AWS_REGION_US_EAST_1
@mock.patch(
"botocore.client.BaseClient._make_api_call", new=mock_make_api_call_disabled
)
@mock_aws
def test__root_credentials_management_disabled(self):
# Create Organization
conn = client("organizations")
conn.create_organization()
from prowler.providers.aws.services.iam.iam_service import IAM
from prowler.providers.aws.services.organizations.organizations_service import (
Organizations,
)
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.iam.iam_root_credentials_management_enabled.iam_root_credentials_management_enabled.iam_client",
new=IAM(aws_provider),
), mock.patch(
"prowler.providers.aws.services.iam.iam_root_credentials_management_enabled.iam_root_credentials_management_enabled.organizations_client",
new=Organizations(aws_provider),
):
from prowler.providers.aws.services.iam.iam_root_credentials_management_enabled.iam_root_credentials_management_enabled import (
iam_root_credentials_management_enabled,
)
check = iam_root_credentials_management_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "Root credentials management is not enabled."
)
assert result[0].resource_id == AWS_ACCOUNT_NUMBER
assert result[0].resource_arn == AWS_ACCOUNT_ARN
assert result[0].region == AWS_REGION_US_EAST_1
@@ -29,6 +29,7 @@ class Test_iam_root_hardware_mfa_enabled_test:
iam_client.audited_partition = "aws"
iam_client.region = AWS_REGION_US_EAST_1
iam_client.mfa_arn_template = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:mfa"
iam_client.organization_features = []
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
@@ -63,6 +64,7 @@ class Test_iam_root_hardware_mfa_enabled_test:
iam_client.audited_partition = "aws"
iam_client.region = AWS_REGION_US_EAST_1
iam_client.mfa_arn_template = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:mfa"
iam_client.organization_features = []
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
@@ -95,6 +97,7 @@ class Test_iam_root_hardware_mfa_enabled_test:
iam_client.audited_partition = "aws"
iam_client.region = AWS_REGION_US_EAST_1
iam_client.mfa_arn_template = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:mfa"
iam_client.organization_features = []
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
@@ -15,14 +15,12 @@ class Test_organizations_tags_policies_enabled_and_attached:
def test_organization_no_organization(self):
organizations_client = mock.MagicMock
organizations_client.region = AWS_REGION_EU_WEST_1
organizations_client.organizations = [
Organization(
arn=AWS_ACCOUNT_ARN,
id="AWS Organization",
status="NOT_AVAILABLE",
master_id="",
)
]
organizations_client.organization = Organization(
arn=AWS_ACCOUNT_ARN,
id="AWS Organization",
status="NOT_AVAILABLE",
master_id="",
)
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
@@ -55,16 +53,14 @@ class Test_organizations_tags_policies_enabled_and_attached:
def test_organization_with_AI_optout_no_policies(self):
organizations_client = mock.MagicMock
organizations_client.region = AWS_REGION_EU_WEST_1
organizations_client.organizations = [
Organization(
id="o-1234567890",
arn="arn:aws:organizations::1234567890:organization/o-1234567890",
status="ACTIVE",
master_id="1234567890",
policies={"AISERVICES_OPT_OUT_POLICY": []},
delegated_administrators=None,
)
]
organizations_client.organization = Organization(
id="o-1234567890",
arn="arn:aws:organizations::1234567890:organization/o-1234567890",
status="ACTIVE",
master_id="1234567890",
policies={"AISERVICES_OPT_OUT_POLICY": []},
delegated_administrators=None,
)
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
@@ -100,33 +96,29 @@ class Test_organizations_tags_policies_enabled_and_attached:
def test_organization_with_AI_optout_policy(self):
organizations_client = mock.MagicMock
organizations_client.region = AWS_REGION_EU_WEST_1
organizations_client.organizations = [
Organization(
id="o-1234567890",
arn="arn:aws:organizations::1234567890:organization/o-1234567890",
status="ACTIVE",
master_id="1234567890",
policies={
"AISERVICES_OPT_OUT_POLICY": [
Policy(
id="p-1234567890",
arn="arn:aws:organizations::1234567890:policy/o-1234567890/p-1234567890",
type="AISERVICES_OPT_OUT_POLICY",
aws_managed=False,
content={
"services": {
"default": {
"opt_out_policy": {"@@assign": "optOut"}
}
}
},
targets=[],
)
]
},
delegated_administrators=None,
)
]
organizations_client.organization = Organization(
id="o-1234567890",
arn="arn:aws:organizations::1234567890:organization/o-1234567890",
status="ACTIVE",
master_id="1234567890",
policies={
"AISERVICES_OPT_OUT_POLICY": [
Policy(
id="p-1234567890",
arn="arn:aws:organizations::1234567890:policy/o-1234567890/p-1234567890",
type="AISERVICES_OPT_OUT_POLICY",
aws_managed=False,
content={
"services": {
"default": {"opt_out_policy": {"@@assign": "optOut"}}
}
},
targets=[],
)
]
},
delegated_administrators=None,
)
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
@@ -162,27 +154,25 @@ class Test_organizations_tags_policies_enabled_and_attached:
def test_organization_with_AI_optout_policy_no_content(self):
organizations_client = mock.MagicMock
organizations_client.region = AWS_REGION_EU_WEST_1
organizations_client.organizations = [
Organization(
id="o-1234567890",
arn="arn:aws:organizations::1234567890:organization/o-1234567890",
status="ACTIVE",
master_id="1234567890",
policies={
"AISERVICES_OPT_OUT_POLICY": [
Policy(
id="p-1234567890",
arn="arn:aws:organizations::1234567890:policy/o-1234567890/p-1234567890",
type="AISERVICES_OPT_OUT_POLICY",
aws_managed=False,
content={},
targets=[],
)
]
},
delegated_administrators=None,
)
]
organizations_client.organization = Organization(
id="o-1234567890",
arn="arn:aws:organizations::1234567890:organization/o-1234567890",
status="ACTIVE",
master_id="1234567890",
policies={
"AISERVICES_OPT_OUT_POLICY": [
Policy(
id="p-1234567890",
arn="arn:aws:organizations::1234567890:policy/o-1234567890/p-1234567890",
type="AISERVICES_OPT_OUT_POLICY",
aws_managed=False,
content={},
targets=[],
)
]
},
delegated_administrators=None,
)
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
@@ -247,7 +247,7 @@ class Test_organizations_scp_check_deny_regions:
# Create Organization
conn = client("organizations", region_name=AWS_REGION_EU_WEST_1)
response = conn.create_organization()
org_arn = response["Organization"]["Arn"]
response["Organization"]["Arn"]
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -262,9 +262,7 @@ class Test_organizations_scp_check_deny_regions:
organizations_scp_check_deny_regions,
)
for org in organizations_client.organizations:
if org.arn == org_arn:
org.policies = None
organizations_client.organization.policies = None
check = organizations_scp_check_deny_regions()
result = check.execute()
@@ -28,15 +28,14 @@ class Test_Organizations_Service:
[AWS_REGION_EU_WEST_1], create_default_organization=False
)
organizations = Organizations(aws_provider)
assert len(organizations.organizations) == 1
assert organizations.organizations[0].arn == response["Organization"]["Arn"]
assert organizations.organizations[0].id == response["Organization"]["Id"]
assert organizations.organization.arn == response["Organization"]["Arn"]
assert organizations.organization.id == response["Organization"]["Id"]
assert (
organizations.organizations[0].master_id
organizations.organization.master_id
== response["Organization"]["MasterAccountId"]
)
assert organizations.organizations[0].status == "ACTIVE"
assert organizations.organizations[0].delegated_administrators == []
assert organizations.organization.status == "ACTIVE"
assert organizations.organization.delegated_administrators == []
@mock_aws
def test_list_policies(self):
@@ -15,14 +15,12 @@ class Test_organizations_tags_policies_enabled_and_attached:
def test_organization_no_organization(self):
organizations_client = mock.MagicMock
organizations_client.region = AWS_REGION_EU_WEST_1
organizations_client.organizations = [
Organization(
arn=AWS_ACCOUNT_ARN,
id="AWS Organization",
status="NOT_AVAILABLE",
master_id="",
)
]
organizations_client.organization = Organization(
arn=AWS_ACCOUNT_ARN,
id="AWS Organization",
status="NOT_AVAILABLE",
master_id="",
)
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
@@ -55,27 +53,25 @@ class Test_organizations_tags_policies_enabled_and_attached:
def test_organization_with_tag_policies_not_attached(self):
organizations_client = mock.MagicMock
organizations_client.region = AWS_REGION_EU_WEST_1
organizations_client.organizations = [
Organization(
id="o-1234567890",
arn="arn:aws:organizations::1234567890:organization/o-1234567890",
status="ACTIVE",
master_id="1234567890",
policies={
"TAG_POLICY": [
Policy(
id="p-1234567890",
arn="arn:aws:organizations::1234567890:policy/o-1234567890/p-1234567890",
type="TAG_POLICY",
aws_managed=False,
content={"tags": {"Owner": {}}},
targets=[],
)
]
},
delegated_administrators=None,
)
]
organizations_client.organization = Organization(
id="o-1234567890",
arn="arn:aws:organizations::1234567890:organization/o-1234567890",
status="ACTIVE",
master_id="1234567890",
policies={
"TAG_POLICY": [
Policy(
id="p-1234567890",
arn="arn:aws:organizations::1234567890:policy/o-1234567890/p-1234567890",
type="TAG_POLICY",
aws_managed=False,
content={"tags": {"Owner": {}}},
targets=[],
)
]
},
delegated_administrators=None,
)
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
@@ -111,27 +107,25 @@ class Test_organizations_tags_policies_enabled_and_attached:
def test_organization_with_tag_policies_attached(self):
organizations_client = mock.MagicMock
organizations_client.region = AWS_REGION_EU_WEST_1
organizations_client.organizations = [
Organization(
id="o-1234567890",
arn="arn:aws:organizations::1234567890:organization/o-1234567890",
status="ACTIVE",
master_id="1234567890",
policies={
"TAG_POLICY": [
Policy(
id="p-1234567890",
arn="arn:aws:organizations::1234567890:policy/o-1234567890/p-1234567890",
type="TAG_POLICY",
aws_managed=False,
content={"tags": {"Owner": {}}},
targets=["1234567890"],
)
]
},
delegated_administrators=None,
)
]
organizations_client.organization = Organization(
id="o-1234567890",
arn="arn:aws:organizations::1234567890:organization/o-1234567890",
status="ACTIVE",
master_id="1234567890",
policies={
"TAG_POLICY": [
Policy(
id="p-1234567890",
arn="arn:aws:organizations::1234567890:policy/o-1234567890/p-1234567890",
type="TAG_POLICY",
aws_managed=False,
content={"tags": {"Owner": {}}},
targets=["1234567890"],
)
]
},
delegated_administrators=None,
)
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])