mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-10-30 23:34:27 +00:00
feat(opensearch): Change service adding two new attributes to Domain model for new check and to use getters in describe method, use moto in service tests and assert new atributtes
This commit is contained in:
MarioRgzLpz committed
1 parent
0c149461b3
commit
eca94e2e3b
2 files changed
+179
-169
No files matched your search
@@ -2,6 +2,8 @@ from json import dumps
|
||||
from unittest.mock import patch
|
||||
|
||||
import botocore
|
||||
from boto3 import client
|
||||
from moto import mock_aws
|
||||
|
||||
from prowler.providers.aws.services.opensearch.opensearch_service import (
|
||||
OpenSearchService,
|
||||
@@ -13,7 +15,7 @@ from tests.providers.aws.utils import (
|
||||
)
|
||||
|
||||
test_domain_name = "test"
|
||||
domain_arn = f"arn:aws:es:us-west-2:{AWS_ACCOUNT_NUMBER}:domain/{test_domain_name}"
|
||||
domain_arn = f"arn:aws:es:eu-west-1:{AWS_ACCOUNT_NUMBER}:domain/{test_domain_name}"
|
||||
|
||||
policy_data = {
|
||||
"Version": "2012-10-17",
|
||||
@@ -56,35 +58,6 @@ def mock_make_api_call(self, operation_name, kwarg):
|
||||
},
|
||||
}
|
||||
}
|
||||
if operation_name == "DescribeDomain":
|
||||
return {
|
||||
"DomainStatus": {
|
||||
"ARN": domain_arn,
|
||||
"Endpoints": {
|
||||
"vpc": "vpc-endpoint-h2dsd34efgyghrtguk5gt6j2foh4.us-east-1.es.amazonaws.com"
|
||||
},
|
||||
"EngineVersion": "opensearch-version1",
|
||||
"VPCOptions": {
|
||||
"VPCId": "test-vpc-id",
|
||||
},
|
||||
"CognitoOptions": {"Enabled": True},
|
||||
"EncryptionAtRestOptions": {"Enabled": True},
|
||||
"NodeToNodeEncryptionOptions": {"Enabled": True},
|
||||
"AdvancedOptions": {"string": "string"},
|
||||
"LogPublishingOptions": {
|
||||
"string": {
|
||||
"CloudWatchLogsLogGroupArn": "string",
|
||||
"Enabled": True | False,
|
||||
}
|
||||
},
|
||||
"ServiceSoftwareOptions": {"UpdateAvailable": True},
|
||||
"DomainEndpointOptions": {"EnforceHTTPS": True},
|
||||
"AdvancedSecurityOptions": {
|
||||
"InternalUserDatabaseEnabled": True,
|
||||
"SAMLOptions": {"Enabled": True},
|
||||
},
|
||||
}
|
||||
}
|
||||
if operation_name == "ListTags":
|
||||
return {
|
||||
"TagList": [
|
||||
@@ -132,45 +105,80 @@ class Test_OpenSearchService_Service:
|
||||
aws_provider = set_mocked_aws_provider([])
|
||||
opensearch = OpenSearchService(aws_provider)
|
||||
assert len(opensearch.opensearch_domains) == 1
|
||||
assert opensearch.opensearch_domains[0].name == test_domain_name
|
||||
assert opensearch.opensearch_domains[0].region == AWS_REGION_EU_WEST_1
|
||||
assert opensearch.opensearch_domains[domain_arn].name == test_domain_name
|
||||
assert opensearch.opensearch_domains[domain_arn].region == AWS_REGION_EU_WEST_1
|
||||
|
||||
# Test OpenSearchService describ domain config
|
||||
def test_describe_domain_config(self):
|
||||
aws_provider = set_mocked_aws_provider([])
|
||||
opensearch = OpenSearchService(aws_provider)
|
||||
assert len(opensearch.opensearch_domains) == 1
|
||||
assert opensearch.opensearch_domains[0].name == test_domain_name
|
||||
assert opensearch.opensearch_domains[0].region == AWS_REGION_EU_WEST_1
|
||||
assert opensearch.opensearch_domains[0].access_policy
|
||||
assert opensearch.opensearch_domains[0].logging[0].name == "SEARCH_SLOW_LOGS"
|
||||
assert opensearch.opensearch_domains[0].logging[0].enabled
|
||||
assert opensearch.opensearch_domains[0].logging[1].name == "INDEX_SLOW_LOGS"
|
||||
assert opensearch.opensearch_domains[0].logging[1].enabled
|
||||
assert opensearch.opensearch_domains[0].logging[2].name == "AUDIT_LOGS"
|
||||
assert opensearch.opensearch_domains[0].logging[2].enabled
|
||||
assert opensearch.opensearch_domains[domain_arn].name == test_domain_name
|
||||
assert opensearch.opensearch_domains[domain_arn].region == AWS_REGION_EU_WEST_1
|
||||
assert opensearch.opensearch_domains[domain_arn].access_policy
|
||||
assert (
|
||||
opensearch.opensearch_domains[domain_arn].logging[0].name
|
||||
== "SEARCH_SLOW_LOGS"
|
||||
)
|
||||
assert opensearch.opensearch_domains[domain_arn].logging[0].enabled
|
||||
assert (
|
||||
opensearch.opensearch_domains[domain_arn].logging[1].name
|
||||
== "INDEX_SLOW_LOGS"
|
||||
)
|
||||
assert opensearch.opensearch_domains[domain_arn].logging[1].enabled
|
||||
assert opensearch.opensearch_domains[domain_arn].logging[2].name == "AUDIT_LOGS"
|
||||
assert opensearch.opensearch_domains[domain_arn].logging[2].enabled
|
||||
|
||||
# Test OpenSearchService describ domain
|
||||
@mock_aws
|
||||
def test_describe_domain(self):
|
||||
opensearch_client = client("opensearch", region_name=AWS_REGION_EU_WEST_1)
|
||||
domain = opensearch_client.create_domain(
|
||||
DomainName=test_domain_name,
|
||||
EncryptionAtRestOptions={"Enabled": True},
|
||||
ClusterConfig={"InstanceCount": 1, "ZoneAwarenessEnabled": True},
|
||||
NodeToNodeEncryptionOptions={"Enabled": True},
|
||||
AdvancedSecurityOptions={
|
||||
"Enabled": True,
|
||||
"InternalUserDatabaseEnabled": True,
|
||||
"SAMLOptions": {"Enabled": True},
|
||||
},
|
||||
CognitoOptions={"Enabled": True},
|
||||
VPCOptions={
|
||||
"SubnetIds": ["test-subnet-id"],
|
||||
"SecurityGroupIds": ["test-security-group-id"],
|
||||
},
|
||||
DomainEndpointOptions={"EnforceHTTPS": True},
|
||||
AccessPolicies=policy_json,
|
||||
EngineVersion="opensearch-version1",
|
||||
TagList=[
|
||||
{"Key": "test", "Value": "test"},
|
||||
],
|
||||
)
|
||||
aws_provider = set_mocked_aws_provider([])
|
||||
opensearch = OpenSearchService(aws_provider)
|
||||
domain_arn = domain["DomainStatus"]["ARN"]
|
||||
assert len(opensearch.opensearch_domains) == 1
|
||||
assert opensearch.opensearch_domains[0].name == test_domain_name
|
||||
assert opensearch.opensearch_domains[0].region == AWS_REGION_EU_WEST_1
|
||||
assert opensearch.opensearch_domains[0].arn == domain_arn
|
||||
assert opensearch.opensearch_domains[0].access_policy
|
||||
assert opensearch.opensearch_domains[0].vpc_endpoints == [
|
||||
"vpc-endpoint-h2dsd34efgyghrtguk5gt6j2foh4.us-east-1.es.amazonaws.com"
|
||||
assert opensearch.opensearch_domains[domain_arn].name == test_domain_name
|
||||
assert opensearch.opensearch_domains[domain_arn].region == AWS_REGION_EU_WEST_1
|
||||
assert opensearch.opensearch_domains[domain_arn].arn == domain_arn
|
||||
assert opensearch.opensearch_domains[domain_arn].access_policy
|
||||
assert opensearch.opensearch_domains[domain_arn].vpc_endpoints == [
|
||||
"test.eu-west-1.es.amazonaws.com"
|
||||
]
|
||||
assert opensearch.opensearch_domains[0].vpc_id == "test-vpc-id"
|
||||
assert opensearch.opensearch_domains[0].cognito_options
|
||||
assert opensearch.opensearch_domains[0].encryption_at_rest
|
||||
assert opensearch.opensearch_domains[0].node_to_node_encryption
|
||||
assert opensearch.opensearch_domains[0].enforce_https
|
||||
assert opensearch.opensearch_domains[0].internal_user_database
|
||||
assert opensearch.opensearch_domains[0].saml_enabled
|
||||
assert opensearch.opensearch_domains[0].update_available
|
||||
assert opensearch.opensearch_domains[0].version == "opensearch-version1"
|
||||
assert opensearch.opensearch_domains[0].tags == [
|
||||
assert not opensearch.opensearch_domains[domain_arn].vpc_id
|
||||
assert opensearch.opensearch_domains[domain_arn].cognito_options
|
||||
assert opensearch.opensearch_domains[domain_arn].encryption_at_rest
|
||||
assert opensearch.opensearch_domains[domain_arn].node_to_node_encryption
|
||||
assert opensearch.opensearch_domains[domain_arn].enforce_https
|
||||
assert opensearch.opensearch_domains[domain_arn].internal_user_database
|
||||
assert opensearch.opensearch_domains[domain_arn].saml_enabled
|
||||
assert not opensearch.opensearch_domains[domain_arn].update_available
|
||||
assert (
|
||||
opensearch.opensearch_domains[domain_arn].version == "opensearch-version1"
|
||||
)
|
||||
assert opensearch.opensearch_domains[domain_arn].instance_count == 1
|
||||
assert opensearch.opensearch_domains[domain_arn].zone_awareness_enabled
|
||||
assert opensearch.opensearch_domains[domain_arn].tags == [
|
||||
{"Key": "test", "Value": "test"},
|
||||
]
|
||||
Reference in new issue
Block a user