ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-23 20:42:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(opensearch): add new fixer opensearch_service_domains_not_publicly_accessible_fixer (#5926)

Browse Source
This commit is contained in:
Daniel Barranquero
2024-12-11 16:29:48 +01:00
committed by GitHub
parent a915c04e9e
commit fdb1956b0b
2 changed files with 178 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/providers/aws/services/opensearch/opensearch_service_domains_not_publicly_accessible/opensearch_service_domains_not_publicly_accessible_fixer.py
+43
View File
@@ -0,0 +1,43 @@
from prowler.lib.logger import logger
from prowler.providers.aws.services.opensearch.opensearch_client import (
opensearch_client,
)
def fixer(resource_id: str, region: str) -> bool:
"""
Modify the OpenSearch domain's resource-based policy to remove public access.
Specifically, this fixer update the domain config and add an empty policy to remove the old one.
Requires the es:UpdateDomainConfig permission.
Permissions:
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "es:UpdateDomainConfig",
"Resource": "*"
}
]
}
Args:
resource_id (str): The OpenSearch domain name.
region (str): AWS region where the OpenSearch domain exists.
Returns:
bool: True if the operation is successful (policy updated), False otherwise.
"""
try:
regional_client = opensearch_client.regional_clients[region]
regional_client.update_domain_config(
DomainName=resource_id,
AccessPolicies="",
)
except Exception as error:
logger.error(
f"{region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return False
else:
return True
tests/providers/aws/services/opensearch/opensearch_service_domains_not_publicly_accessible/opensearch_service_domains_not_publicly_accessible_fixer_test.py
+135
View File
@@ -0,0 +1,135 @@
from json import dumps
from unittest import mock
from boto3 import client
from moto import mock_aws
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_US_WEST_2,
set_mocked_aws_provider,
)
domain_name = "test-domain"
policy_data_restricted = {
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {"AWS": [f"{AWS_ACCOUNT_NUMBER}"]},
"Action": ["es:*"],
"Resource": f"arn:aws:es:us-west-2:{AWS_ACCOUNT_NUMBER}:domain/{domain_name}/*",
}
],
}
policy_data_not_restricted = {
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {"AWS": ["*"]},
"Action": ["es:*"],
"Resource": f"arn:aws:es:us-west-2:{AWS_ACCOUNT_NUMBER}:domain/{domain_name}/*",
}
],
}
policy_data_not_restricted_principal = {
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": "*",
"Action": ["es:*"],
"Resource": f"arn:aws:es:us-west-2:{AWS_ACCOUNT_NUMBER}:domain/{domain_name}/*",
}
],
}
class Test_opensearch_service_domains_not_publicly_accessible_fixer:
@mock_aws
def test_policy_data_restricted_error(self):
opensearch_client = client("opensearch", region_name=AWS_REGION_US_WEST_2)
opensearch_client.create_domain(DomainName=domain_name)["DomainStatus"]["ARN"]
opensearch_client.update_domain_config(
DomainName=domain_name,
AccessPolicies=str(policy_data_restricted),
)
aws_provider = set_mocked_aws_provider([AWS_REGION_US_WEST_2])
from prowler.providers.aws.services.opensearch.opensearch_service import (
OpenSearchService,
)
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.opensearch.opensearch_service_domains_not_publicly_accessible.opensearch_service_domains_not_publicly_accessible_fixer.opensearch_client",
new=OpenSearchService(aws_provider),
):
from prowler.providers.aws.services.opensearch.opensearch_service_domains_not_publicly_accessible.opensearch_service_domains_not_publicly_accessible_fixer import (
fixer,
)
assert not fixer("domain_name_non_existing", AWS_REGION_US_WEST_2)
@mock_aws
def test_policy_data_not_restricted_with_principal_AWS(self):
opensearch_client = client("opensearch", region_name=AWS_REGION_US_WEST_2)
opensearch_client.create_domain(DomainName=domain_name)["DomainStatus"]["ARN"]
opensearch_client.update_domain_config(
DomainName=domain_name,
AccessPolicies=dumps(policy_data_not_restricted),
)
aws_provider = set_mocked_aws_provider([AWS_REGION_US_WEST_2])
from prowler.providers.aws.services.opensearch.opensearch_service import (
OpenSearchService,
)
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.opensearch.opensearch_service_domains_not_publicly_accessible.opensearch_service_domains_not_publicly_accessible_fixer.opensearch_client",
new=OpenSearchService(aws_provider),
):
from prowler.providers.aws.services.opensearch.opensearch_service_domains_not_publicly_accessible.opensearch_service_domains_not_publicly_accessible_fixer import (
fixer,
)
assert fixer(domain_name, AWS_REGION_US_WEST_2)
@mock_aws
def test_policy_data_not_restricted_with_principal_no_AWS(self):
opensearch_client = client("opensearch", region_name=AWS_REGION_US_WEST_2)
opensearch_client.create_domain(DomainName=domain_name)["DomainStatus"]["ARN"]
opensearch_client.update_domain_config(
DomainName=domain_name,
AccessPolicies=dumps(policy_data_not_restricted_principal),
)
aws_provider = set_mocked_aws_provider([AWS_REGION_US_WEST_2])
from prowler.providers.aws.services.opensearch.opensearch_service import (
OpenSearchService,
)
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.opensearch.opensearch_service_domains_not_publicly_accessible.opensearch_service_domains_not_publicly_accessible_fixer.opensearch_client",
new=OpenSearchService(aws_provider),
):
from prowler.providers.aws.services.opensearch.opensearch_service_domains_not_publicly_accessible.opensearch_service_domains_not_publicly_accessible_fixer import (
fixer,
)
assert fixer(domain_name, AWS_REGION_US_WEST_2)