feat(new_checks): New AWS Organizations related checks (#2133)

Co-authored-by: Pepe Fagoaga <pepe@verica.io>
This commit is contained in:
Gabriel Soltz
2023-03-30 17:36:23 +02:00
committed by GitHub
parent e37d8fe45f
commit 608fd92861
18 changed files with 1086 additions and 1 deletions

View File

@@ -0,0 +1,102 @@
from re import search
from unittest import mock
from boto3 import client, session
from moto import mock_organizations
from prowler.providers.aws.lib.audit_info.audit_info import AWS_Audit_Info
from prowler.providers.aws.services.organizations.organizations_service import (
Organizations,
)
AWS_REGION = "us-east-1"
class Test_organizations_account_part_of_organizations:
# Mocked Audit Info
def set_mocked_audit_info(self):
audit_info = AWS_Audit_Info(
session_config=None,
original_session=None,
audit_session=session.Session(
profile_name=None,
botocore_session=None,
),
audited_account=None,
audited_user_id=None,
audited_partition="aws",
audited_identity_arn=None,
profile=None,
profile_region=None,
credentials=None,
assumed_role_info=None,
audited_regions=None,
organizations_metadata=None,
audit_resources=None,
)
return audit_info
@mock_organizations
def test_no_organization(self):
audit_info = self.set_mocked_audit_info()
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
):
with mock.patch(
"prowler.providers.aws.services.organizations.organizations_account_part_of_organizations.organizations_account_part_of_organizations.organizations_client",
new=Organizations(audit_info),
):
# Test Check
from prowler.providers.aws.services.organizations.organizations_account_part_of_organizations.organizations_account_part_of_organizations import (
organizations_account_part_of_organizations,
)
check = organizations_account_part_of_organizations()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"AWS Organizations is not in-use for this AWS Account",
result[0].status_extended,
)
assert result[0].resource_id == "AWS Organization"
assert result[0].resource_arn == ""
@mock_organizations
def test_organization(self):
audit_info = self.set_mocked_audit_info()
# Create Organization
conn = client("organizations")
response = conn.create_organization()
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
):
with mock.patch(
"prowler.providers.aws.services.organizations.organizations_account_part_of_organizations.organizations_account_part_of_organizations.organizations_client",
new=Organizations(audit_info),
):
# Test Check
from prowler.providers.aws.services.organizations.organizations_account_part_of_organizations.organizations_account_part_of_organizations import (
organizations_account_part_of_organizations,
)
check = organizations_account_part_of_organizations()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert search(
"Account is part of AWS Organization",
result[0].status_extended,
)
assert result[0].resource_id == response["Organization"]["Id"]
assert result[0].resource_arn == response["Organization"]["Arn"]

View File

@@ -0,0 +1,187 @@
from re import search
from unittest import mock
from boto3 import client, session
from moto import mock_organizations
from prowler.providers.aws.lib.audit_info.audit_info import AWS_Audit_Info
from prowler.providers.aws.services.organizations.organizations_service import (
Organizations,
)
AWS_REGION = "us-east-1"
class Test_organizations_delegated_administrators:
# Mocked Audit Info
def set_mocked_audit_info(self):
audit_info = AWS_Audit_Info(
session_config=None,
original_session=None,
audit_session=session.Session(
profile_name=None,
botocore_session=None,
),
audited_account=None,
audited_user_id=None,
audited_partition="aws",
audited_identity_arn=None,
profile=None,
profile_region=None,
credentials=None,
assumed_role_info=None,
audited_regions=None,
organizations_metadata=None,
audit_resources=None,
)
return audit_info
@mock_organizations
def test_no_organization(self):
audit_info = self.set_mocked_audit_info()
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
):
with mock.patch(
"prowler.providers.aws.services.organizations.organizations_delegated_administrators.organizations_delegated_administrators.organizations_client",
new=Organizations(audit_info),
):
# Test Check
from prowler.providers.aws.services.organizations.organizations_delegated_administrators.organizations_delegated_administrators import (
organizations_delegated_administrators,
)
check = organizations_delegated_administrators()
result = check.execute()
assert len(result) == 0
@mock_organizations
def test_organization_no_delegations(self):
audit_info = self.set_mocked_audit_info()
# Create Organization
conn = client("organizations", region_name=AWS_REGION)
response = conn.create_organization()
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
):
with mock.patch(
"prowler.providers.aws.services.organizations.organizations_delegated_administrators.organizations_delegated_administrators.organizations_client",
new=Organizations(audit_info),
):
# Test Check
from prowler.providers.aws.services.organizations.organizations_delegated_administrators.organizations_delegated_administrators import (
organizations_delegated_administrators,
)
check = organizations_delegated_administrators()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].resource_id == response["Organization"]["Id"]
assert result[0].resource_arn == response["Organization"]["Arn"]
assert search(
"No Delegated Administrators",
result[0].status_extended,
)
@mock_organizations
def test_organization_trusted_delegated(self):
audit_info = self.set_mocked_audit_info()
# Create Organization
conn = client("organizations", region_name=AWS_REGION)
response = conn.create_organization()
# Create Dummy Account
account = conn.create_account(
Email="test@test.com",
AccountName="test",
)
# Delegate Administrator
conn.register_delegated_administrator(
AccountId=account["CreateAccountStatus"]["AccountId"],
ServicePrincipal="config-multiaccountsetup.amazonaws.com",
)
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
):
with mock.patch(
"prowler.providers.aws.services.organizations.organizations_delegated_administrators.organizations_delegated_administrators.organizations_client",
new=Organizations(audit_info),
):
with mock.patch(
"prowler.providers.aws.services.organizations.organizations_delegated_administrators.organizations_delegated_administrators.get_config_var",
return_value=[account["CreateAccountStatus"]["AccountId"]],
):
# Test Check
from prowler.providers.aws.services.organizations.organizations_delegated_administrators.organizations_delegated_administrators import (
organizations_delegated_administrators,
)
check = organizations_delegated_administrators()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].resource_id == response["Organization"]["Id"]
assert result[0].resource_arn == response["Organization"]["Arn"]
assert search(
"Trusted Delegated Administrator",
result[0].status_extended,
)
@mock_organizations
def test_organization_untrusted_delegated(self):
audit_info = self.set_mocked_audit_info()
# Create Organization
conn = client("organizations", region_name=AWS_REGION)
response = conn.create_organization()
# Create Dummy Account
account = conn.create_account(
Email="test@test.com",
AccountName="test",
)
# Delegate Administrator
conn.register_delegated_administrator(
AccountId=account["CreateAccountStatus"]["AccountId"],
ServicePrincipal="config-multiaccountsetup.amazonaws.com",
)
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
):
with mock.patch(
"prowler.providers.aws.services.organizations.organizations_delegated_administrators.organizations_delegated_administrators.organizations_client",
new=Organizations(audit_info),
):
# Test Check
from prowler.providers.aws.services.organizations.organizations_delegated_administrators.organizations_delegated_administrators import (
organizations_delegated_administrators,
)
check = organizations_delegated_administrators()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert result[0].resource_id == response["Organization"]["Id"]
assert result[0].resource_arn == response["Organization"]["Arn"]
assert search(
"Untrusted Delegated Administrator",
result[0].status_extended,
)

View File

@@ -0,0 +1,196 @@
from re import search
from unittest import mock
from boto3 import client, session
from moto import mock_organizations
from prowler.providers.aws.lib.audit_info.audit_info import AWS_Audit_Info
from prowler.providers.aws.services.organizations.organizations_service import (
Organizations,
)
AWS_REGION = "us-east-1"
def scp_restrict_regions_with_deny():
return '{"Version":"2012-10-17","Statement":{"Effect":"Deny","NotAction":"s3:*","Resource":"*","Condition":{"StringNotEquals":{"aws:RequestedRegion":["eu-central-1"]}}}}'
class Test_organizations_scp_check_deny_regions:
# Mocked Audit Info
def set_mocked_audit_info(self):
audit_info = AWS_Audit_Info(
session_config=None,
original_session=None,
audit_session=session.Session(
profile_name=None,
botocore_session=None,
),
audited_account=None,
audited_user_id=None,
audited_partition="aws",
audited_identity_arn=None,
profile=None,
profile_region=None,
credentials=None,
assumed_role_info=None,
audited_regions=None,
organizations_metadata=None,
audit_resources=None,
)
return audit_info
@mock_organizations
def test_no_organization(self):
audit_info = self.set_mocked_audit_info()
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
):
with mock.patch(
"prowler.providers.aws.services.organizations.organizations_scp_check_deny_regions.organizations_scp_check_deny_regions.organizations_client",
new=Organizations(audit_info),
):
# Test Check
from prowler.providers.aws.services.organizations.organizations_scp_check_deny_regions.organizations_scp_check_deny_regions import (
organizations_scp_check_deny_regions,
)
check = organizations_scp_check_deny_regions()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"AWS Organizations is not in-use for this AWS Account",
result[0].status_extended,
)
assert result[0].resource_id == "AWS Organization"
assert result[0].resource_arn == ""
@mock_organizations
def test_organization_without_scp_deny_regions(self):
audit_info = self.set_mocked_audit_info()
# Create Organization
conn = client("organizations", region_name=AWS_REGION)
response = conn.create_organization()
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
):
with mock.patch(
"prowler.providers.aws.services.organizations.organizations_scp_check_deny_regions.organizations_scp_check_deny_regions.organizations_client",
new=Organizations(audit_info),
):
# Test Check
from prowler.providers.aws.services.organizations.organizations_scp_check_deny_regions.organizations_scp_check_deny_regions import (
organizations_scp_check_deny_regions,
)
check = organizations_scp_check_deny_regions()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert result[0].resource_id == response["Organization"]["Id"]
assert result[0].resource_arn == response["Organization"]["Arn"]
assert search(
"level but don't restrict AWS Regions",
result[0].status_extended,
)
@mock_organizations
def test_organization_with_scp_deny_regions_valid(self):
audit_info = self.set_mocked_audit_info()
# Create Organization
conn = client("organizations", region_name=AWS_REGION)
response = conn.create_organization()
# Create Policy
conn.create_policy(
Content=scp_restrict_regions_with_deny(),
Description="Test",
Name="Test",
Type="SERVICE_CONTROL_POLICY",
)
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
):
with mock.patch(
"prowler.providers.aws.services.organizations.organizations_scp_check_deny_regions.organizations_scp_check_deny_regions.organizations_client",
new=Organizations(audit_info),
):
with mock.patch(
"prowler.providers.aws.services.organizations.organizations_scp_check_deny_regions.organizations_scp_check_deny_regions.get_config_var",
return_value=["eu-central-1"],
):
# Test Check
from prowler.providers.aws.services.organizations.organizations_scp_check_deny_regions.organizations_scp_check_deny_regions import (
organizations_scp_check_deny_regions,
)
check = organizations_scp_check_deny_regions()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].resource_id == response["Organization"]["Id"]
assert result[0].resource_arn == response["Organization"]["Arn"]
assert search(
"restricting all configured regions found",
result[0].status_extended,
)
@mock_organizations
def test_organization_with_scp_deny_regions_not_valid(self):
audit_info = self.set_mocked_audit_info()
# Create Organization
conn = client("organizations", region_name=AWS_REGION)
response = conn.create_organization()
# Create Policy
conn.create_policy(
Content=scp_restrict_regions_with_deny(),
Description="Test",
Name="Test",
Type="SERVICE_CONTROL_POLICY",
)
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
):
with mock.patch(
"prowler.providers.aws.services.organizations.organizations_scp_check_deny_regions.organizations_scp_check_deny_regions.organizations_client",
new=Organizations(audit_info),
):
with mock.patch(
"prowler.providers.aws.services.organizations.organizations_scp_check_deny_regions.organizations_scp_check_deny_regions.get_config_var",
return_value=["us-east-1"],
):
# Test Check
from prowler.providers.aws.services.organizations.organizations_scp_check_deny_regions.organizations_scp_check_deny_regions import (
organizations_scp_check_deny_regions,
)
check = organizations_scp_check_deny_regions()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert result[0].resource_id == response["Organization"]["Id"]
assert result[0].resource_arn == response["Organization"]["Arn"]
assert search(
"restricting some AWS Regions, but not all the configured ones, please check config...",
result[0].status_extended,
)

View File

@@ -0,0 +1,91 @@
import json
from boto3 import client, session
from moto import mock_organizations
from moto.core import DEFAULT_ACCOUNT_ID
from prowler.providers.aws.lib.audit_info.audit_info import AWS_Audit_Info
from prowler.providers.aws.services.organizations.organizations_service import (
Organizations,
)
AWS_REGION = "eu-west-1"
def scp_restrict_regions_with_deny():
return '{"Version":"2012-10-17","Statement":{"Effect":"Deny","NotAction":"s3:*","Resource":"*","Condition":{"StringNotEquals":{"aws:RequestedRegion":["eu-central-1"]}}}}'
class Test_Organizations_Service:
# Mocked Audit Info
def set_mocked_audit_info(self):
audit_info = AWS_Audit_Info(
session_config=None,
original_session=None,
audit_session=session.Session(
profile_name=None,
botocore_session=None,
region_name=AWS_REGION,
),
audited_account=DEFAULT_ACCOUNT_ID,
audited_user_id=None,
audited_partition="aws",
audited_identity_arn=None,
profile=None,
profile_region=AWS_REGION,
credentials=None,
assumed_role_info=None,
audited_regions=None,
organizations_metadata=None,
audit_resources=None,
)
return audit_info
@mock_organizations
def test_service(self):
audit_info = self.set_mocked_audit_info()
organizations = Organizations(audit_info)
assert organizations.service == "organizations"
@mock_organizations
def test__describe_organization__(self):
# Create Organization
conn = client("organizations", region_name=AWS_REGION)
response = conn.create_organization()
# Mock
audit_info = self.set_mocked_audit_info()
organizations = Organizations(audit_info)
# Tests
assert len(organizations.organizations) == 1
assert organizations.organizations[0].arn == response["Organization"]["Arn"]
assert organizations.organizations[0].id == response["Organization"]["Id"]
assert (
organizations.organizations[0].master_id
== response["Organization"]["MasterAccountId"]
)
assert organizations.organizations[0].status == "ACTIVE"
assert organizations.organizations[0].delegated_administrators == []
@mock_organizations
def test__list_policies__(self):
# Create Policy
conn = client("organizations", region_name=AWS_REGION)
conn.create_organization()
response = conn.create_policy(
Content=scp_restrict_regions_with_deny(),
Description="Test",
Name="Test",
Type="SERVICE_CONTROL_POLICY",
)
# Mock
audit_info = self.set_mocked_audit_info()
organizations = Organizations(audit_info)
# Tests
assert len(organizations.policies) == 2
for policy in organizations.policies:
if policy.arn == response["Policy"]["PolicySummary"]["Arn"]:
assert policy.type == "SERVICE_CONTROL_POLICY"
assert policy.aws_managed is False
assert policy.content == json.loads(response["Policy"]["Content"])
assert policy.targets == []