test(audit_info): refactor athena (#3101)

This commit is contained in:
Pepe Fagoaga
2023-12-05 10:03:11 +01:00
committed by GitHub
parent 9236f5d058
commit 1b85af95c0
3 changed files with 37 additions and 131 deletions

View File

@@ -1,15 +1,12 @@
from boto3 import session
from botocore.client import BaseClient
from mock import patch
from moto import mock_athena
from prowler.providers.aws.lib.audit_info.models import AWS_Audit_Info
from prowler.providers.aws.services.athena.athena_service import Athena
from prowler.providers.common.models import Audit_Metadata
AWS_ACCOUNT_NUMBER = "123456789012"
AWS_REGION = "eu-west-1"
from tests.providers.aws.audit_info_utils import (
AWS_REGION_EU_WEST_1,
set_mocked_aws_audit_info,
)
# Mocking Access Analyzer Calls
make_api_call = BaseClient._make_api_call
@@ -43,9 +40,11 @@ def mock_make_api_call(self, operation_name, kwarg):
# Mock generate_regional_clients()
def mock_generate_regional_clients(service, audit_info, _):
regional_client = audit_info.audit_session.client(service, region_name=AWS_REGION)
regional_client.region = AWS_REGION
return {AWS_REGION: regional_client}
regional_client = audit_info.audit_session.client(
service, region_name=AWS_REGION_EU_WEST_1
)
regional_client.region = AWS_REGION_EU_WEST_1
return {AWS_REGION_EU_WEST_1: regional_client}
# Patch every AWS call using Boto3 and generate_regional_clients to have 1 client
@@ -54,49 +53,18 @@ def mock_generate_regional_clients(service, audit_info, _):
new=mock_generate_regional_clients,
)
class Test_Athena_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,
),
audited_account=AWS_ACCOUNT_NUMBER,
audited_account_arn=f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root",
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,
mfa_enabled=False,
audit_metadata=Audit_Metadata(
services_scanned=0,
expected_checks=[],
completed_checks=0,
audit_progress=0,
),
)
return audit_info
# Test Athena Get Workgrups
@mock_athena
def test__get_workgroups__not_encrypted(self):
default_workgroup_name = "primary"
audit_info = self.set_mocked_audit_info()
workgroup_arn = f"arn:{audit_info.audited_partition}:athena:{AWS_REGION}:{audit_info.audited_account}:workgroup/{default_workgroup_name}"
audit_info = set_mocked_aws_audit_info([AWS_REGION_EU_WEST_1])
workgroup_arn = f"arn:{audit_info.audited_partition}:athena:{AWS_REGION_EU_WEST_1}:{audit_info.audited_account}:workgroup/{default_workgroup_name}"
athena = Athena(audit_info)
assert len(athena.workgroups) == 1
assert athena.workgroups[workgroup_arn]
assert athena.workgroups[workgroup_arn].arn == workgroup_arn
assert athena.workgroups[workgroup_arn].name == default_workgroup_name
assert athena.workgroups[workgroup_arn].region == AWS_REGION
assert athena.workgroups[workgroup_arn].region == AWS_REGION_EU_WEST_1
assert athena.workgroups[workgroup_arn].tags == []
assert (
athena.workgroups[workgroup_arn].encryption_configuration.encrypted is False
@@ -113,7 +81,7 @@ class Test_Athena_Service:
@mock_athena
def test__get_workgroups__encrypted(self):
default_workgroup_name = "primary"
audit_info = self.set_mocked_audit_info()
audit_info = set_mocked_aws_audit_info([AWS_REGION_EU_WEST_1])
# Athena client
# This API call is not implemented by Moto
@@ -129,13 +97,13 @@ class Test_Athena_Service:
# },
# )
workgroup_arn = f"arn:{audit_info.audited_partition}:athena:{AWS_REGION}:{audit_info.audited_account}:workgroup/{default_workgroup_name}"
workgroup_arn = f"arn:{audit_info.audited_partition}:athena:{AWS_REGION_EU_WEST_1}:{audit_info.audited_account}:workgroup/{default_workgroup_name}"
athena = Athena(audit_info)
assert len(athena.workgroups) == 1
assert athena.workgroups[workgroup_arn]
assert athena.workgroups[workgroup_arn].arn == workgroup_arn
assert athena.workgroups[workgroup_arn].name == default_workgroup_name
assert athena.workgroups[workgroup_arn].region == AWS_REGION
assert athena.workgroups[workgroup_arn].region == AWS_REGION_EU_WEST_1
assert athena.workgroups[workgroup_arn].tags == []
assert (
athena.workgroups[workgroup_arn].encryption_configuration.encrypted is True

View File

@@ -1,56 +1,25 @@
from unittest import mock
from boto3 import session
from mock import patch
from moto import mock_athena
from prowler.providers.aws.lib.audit_info.models import AWS_Audit_Info
from prowler.providers.common.models import Audit_Metadata
from tests.providers.aws.audit_info_utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_EU_WEST_1,
set_mocked_aws_audit_info,
)
from tests.providers.aws.services.athena.athena_service_test import mock_make_api_call
AWS_REGION = "eu-west-1"
AWS_ACCOUNT_NUMBER = "123456789012"
ATHENA_PRIMARY_WORKGROUP = "primary"
ATHENA_PRIMARY_WORKGROUP_ARN = f"arn:aws:athena:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:workgroup/{ATHENA_PRIMARY_WORKGROUP}"
ATHENA_PRIMARY_WORKGROUP_ARN = f"arn:aws:athena:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:workgroup/{ATHENA_PRIMARY_WORKGROUP}"
class Test_athena_workgroup_encryption:
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=AWS_ACCOUNT_NUMBER,
audited_account_arn=f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root",
audited_user_id=None,
audited_partition="aws",
audited_identity_arn=None,
profile=None,
profile_region=None,
credentials=None,
assumed_role_info=None,
audited_regions=[AWS_REGION],
organizations_metadata=None,
audit_resources=None,
mfa_enabled=False,
audit_metadata=Audit_Metadata(
services_scanned=0,
expected_checks=[],
completed_checks=0,
audit_progress=0,
),
)
return audit_info
@mock_athena
def test_primary_workgroup_not_encrypted(self):
from prowler.providers.aws.services.athena.athena_service import Athena
current_audit_info = self.set_mocked_audit_info()
current_audit_info = set_mocked_aws_audit_info([AWS_REGION_EU_WEST_1])
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
@@ -74,14 +43,14 @@ class Test_athena_workgroup_encryption:
)
assert result[0].resource_id == ATHENA_PRIMARY_WORKGROUP
assert result[0].resource_arn == ATHENA_PRIMARY_WORKGROUP_ARN
assert result[0].region == AWS_REGION
assert result[0].region == AWS_REGION_EU_WEST_1
assert result[0].resource_tags == []
@mock_athena
def test_primary_workgroup_not_encrypted_ignoring(self):
from prowler.providers.aws.services.athena.athena_service import Athena
current_audit_info = self.set_mocked_audit_info()
current_audit_info = set_mocked_aws_audit_info([AWS_REGION_EU_WEST_1])
current_audit_info.ignore_unused_services = True
with mock.patch(
@@ -106,7 +75,7 @@ class Test_athena_workgroup_encryption:
def test_primary_workgroup_encrypted(self):
from prowler.providers.aws.services.athena.athena_service import Athena
current_audit_info = self.set_mocked_audit_info()
current_audit_info = set_mocked_aws_audit_info([AWS_REGION_EU_WEST_1])
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
@@ -130,5 +99,5 @@ class Test_athena_workgroup_encryption:
)
assert result[0].resource_id == ATHENA_PRIMARY_WORKGROUP
assert result[0].resource_arn == ATHENA_PRIMARY_WORKGROUP_ARN
assert result[0].region == AWS_REGION
assert result[0].region == AWS_REGION_EU_WEST_1
assert result[0].resource_tags == []

View File

@@ -1,56 +1,25 @@
from unittest import mock
from boto3 import session
from mock import patch
from moto import mock_athena
from prowler.providers.aws.lib.audit_info.models import AWS_Audit_Info
from prowler.providers.common.models import Audit_Metadata
from tests.providers.aws.audit_info_utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_EU_WEST_1,
set_mocked_aws_audit_info,
)
from tests.providers.aws.services.athena.athena_service_test import mock_make_api_call
AWS_REGION = "eu-west-1"
AWS_ACCOUNT_NUMBER = "123456789012"
ATHENA_PRIMARY_WORKGROUP = "primary"
ATHENA_PRIMARY_WORKGROUP_ARN = f"arn:aws:athena:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:workgroup/{ATHENA_PRIMARY_WORKGROUP}"
ATHENA_PRIMARY_WORKGROUP_ARN = f"arn:aws:athena:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:workgroup/{ATHENA_PRIMARY_WORKGROUP}"
class Test_athena_workgroup_enforce_configuration:
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=AWS_ACCOUNT_NUMBER,
audited_account_arn=f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root",
audited_user_id=None,
audited_partition="aws",
audited_identity_arn=None,
profile=None,
profile_region=None,
credentials=None,
assumed_role_info=None,
audited_regions=[AWS_REGION],
organizations_metadata=None,
audit_resources=None,
mfa_enabled=False,
audit_metadata=Audit_Metadata(
services_scanned=0,
expected_checks=[],
completed_checks=0,
audit_progress=0,
),
)
return audit_info
@mock_athena
def test_primary_workgroup_configuration_not_enforced(self):
from prowler.providers.aws.services.athena.athena_service import Athena
current_audit_info = self.set_mocked_audit_info()
current_audit_info = set_mocked_aws_audit_info([AWS_REGION_EU_WEST_1])
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
@@ -74,14 +43,14 @@ class Test_athena_workgroup_enforce_configuration:
)
assert result[0].resource_id == ATHENA_PRIMARY_WORKGROUP
assert result[0].resource_arn == ATHENA_PRIMARY_WORKGROUP_ARN
assert result[0].region == AWS_REGION
assert result[0].region == AWS_REGION_EU_WEST_1
assert result[0].resource_tags == []
@mock_athena
def test_primary_workgroup_configuration_not_enforced_ignoring(self):
from prowler.providers.aws.services.athena.athena_service import Athena
current_audit_info = self.set_mocked_audit_info()
current_audit_info = set_mocked_aws_audit_info([AWS_REGION_EU_WEST_1])
current_audit_info.ignore_unused_services = True
with mock.patch(
@@ -106,7 +75,7 @@ class Test_athena_workgroup_enforce_configuration:
def test_primary_workgroup_configuration_enforced(self):
from prowler.providers.aws.services.athena.athena_service import Athena
current_audit_info = self.set_mocked_audit_info()
current_audit_info = set_mocked_aws_audit_info([AWS_REGION_EU_WEST_1])
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
@@ -130,5 +99,5 @@ class Test_athena_workgroup_enforce_configuration:
)
assert result[0].resource_id == ATHENA_PRIMARY_WORKGROUP
assert result[0].resource_arn == ATHENA_PRIMARY_WORKGROUP_ARN
assert result[0].region == AWS_REGION
assert result[0].region == AWS_REGION_EU_WEST_1
assert result[0].resource_tags == []