feat(sqs): add new fixer sqs_queues_not_publicly_accessible_fixer (#5911)

Co-authored-by: Sergio Garcia <hello@mistercloudsec.com>
This commit is contained in:
Daniel BarranqueroandSergio Garcia authored and GitHub committed 2024-12-10 12:26:42 -04:00
1 parent f2f196cfcd
commit b984f0423a
4 files changed
+206 -9

No files matched your search

@@ -0,0 +1,106 @@
from json import dumps
from unittest import mock
from boto3 import client
from moto import mock_aws
from tests.providers.aws.utils import AWS_REGION_EU_WEST_1, set_mocked_aws_provider
class Test_sqs_queues_not_publicly_accessible_fixer:
@mock_aws
def test_queue_public(self):
sqs_client = client("sqs", region_name=AWS_REGION_EU_WEST_1)
queue_url = sqs_client.create_queue(QueueName="test-queue")["QueueUrl"]
sqs_client.set_queue_attributes(
QueueUrl=queue_url,
Attributes={
"Policy": dumps({"Statement": [{"Effect": "Allow", "Principal": "*"}]})
},
)
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
from prowler.providers.aws.services.sqs.sqs_service import SQS
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.sqs.sqs_queues_not_publicly_accessible.sqs_queues_not_publicly_accessible_fixer.sqs_client",
new=SQS(aws_provider),
):
# Test Fixer
from prowler.providers.aws.services.sqs.sqs_queues_not_publicly_accessible.sqs_queues_not_publicly_accessible_fixer import (
fixer,
)
assert fixer(queue_url, AWS_REGION_EU_WEST_1)
@mock_aws
def test_queue_public_with_aws(self):
sqs_client = client("sqs", region_name=AWS_REGION_EU_WEST_1)
queue_url = sqs_client.create_queue(QueueName="test-queue")["QueueUrl"]
sqs_client.set_queue_attributes(
QueueUrl=queue_url,
Attributes={
"Policy": dumps(
{"Statement": [{"Effect": "Allow", "Principal": {"AWS": "*"}}]}
)
},
)
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
from prowler.providers.aws.services.sqs.sqs_service import SQS
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.sqs.sqs_queues_not_publicly_accessible.sqs_queues_not_publicly_accessible_fixer.sqs_client",
new=SQS(aws_provider),
):
# Test Fixer
from prowler.providers.aws.services.sqs.sqs_queues_not_publicly_accessible.sqs_queues_not_publicly_accessible_fixer import (
fixer,
)
assert fixer(queue_url, AWS_REGION_EU_WEST_1)
@mock_aws
def test_queue_public_error(self):
sqs_client = client("sqs", region_name=AWS_REGION_EU_WEST_1)
queue_url = sqs_client.create_queue(QueueName="test-queue")["QueueUrl"]
sqs_client.set_queue_attributes(
QueueUrl=queue_url,
Attributes={
"Policy": dumps(
{"Statement": [{"Effect": "Allow", "Principal": {"AWS": "*"}}]}
)
},
)
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
from prowler.providers.aws.services.sqs.sqs_service import SQS
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.sqs.sqs_queues_not_publicly_accessible.sqs_queues_not_publicly_accessible_fixer.sqs_client",
new=SQS(aws_provider),
):
# Test Fixer
from prowler.providers.aws.services.sqs.sqs_queues_not_publicly_accessible.sqs_queues_not_publicly_accessible_fixer import (
fixer,
)
assert not fixer("queue_url_non_existing", AWS_REGION_EU_WEST_1)
@@ -110,7 +110,10 @@ class Test_sqs_queues_not_publicly_accessible:
sqs_client.queues = []
with mock.patch(
"prowler.providers.aws.services.sqs.sqs_service.SQS",
sqs_client,
new=sqs_client,
), mock.patch(
"prowler.providers.aws.services.sqs.sqs_client.sqs_client",
new=sqs_client,
):
from prowler.providers.aws.services.sqs.sqs_queues_not_publicly_accessible.sqs_queues_not_publicly_accessible import (
sqs_queues_not_publicly_accessible,
@@ -167,7 +170,10 @@ class Test_sqs_queues_not_publicly_accessible:
)
with mock.patch(
"prowler.providers.aws.services.sqs.sqs_service.SQS",
sqs_client,
new=sqs_client,
), mock.patch(
"prowler.providers.aws.services.sqs.sqs_client.sqs_client",
new=sqs_client,
):
from prowler.providers.aws.services.sqs.sqs_queues_not_publicly_accessible.sqs_queues_not_publicly_accessible import (
sqs_queues_not_publicly_accessible,
@@ -201,7 +207,10 @@ class Test_sqs_queues_not_publicly_accessible:
)
with mock.patch(
"prowler.providers.aws.services.sqs.sqs_service.SQS",
sqs_client,
new=sqs_client,
), mock.patch(
"prowler.providers.aws.services.sqs.sqs_client.sqs_client",
new=sqs_client,
):
from prowler.providers.aws.services.sqs.sqs_queues_not_publicly_accessible.sqs_queues_not_publicly_accessible import (
sqs_queues_not_publicly_accessible,
@@ -235,7 +244,10 @@ class Test_sqs_queues_not_publicly_accessible:
)
with mock.patch(
"prowler.providers.aws.services.sqs.sqs_service.SQS",
sqs_client,
new=sqs_client,
), mock.patch(
"prowler.providers.aws.services.sqs.sqs_client.sqs_client",
new=sqs_client,
):
from prowler.providers.aws.services.sqs.sqs_queues_not_publicly_accessible.sqs_queues_not_publicly_accessible import (
sqs_queues_not_publicly_accessible,
@@ -269,7 +281,10 @@ class Test_sqs_queues_not_publicly_accessible:
)
with mock.patch(
"prowler.providers.aws.services.sqs.sqs_service.SQS",
sqs_client,
new=sqs_client,
), mock.patch(
"prowler.providers.aws.services.sqs.sqs_client.sqs_client",
new=sqs_client,
):
from prowler.providers.aws.services.sqs.sqs_queues_not_publicly_accessible.sqs_queues_not_publicly_accessible import (
sqs_queues_not_publicly_accessible,
@@ -303,7 +318,10 @@ class Test_sqs_queues_not_publicly_accessible:
)
with mock.patch(
"prowler.providers.aws.services.sqs.sqs_service.SQS",
sqs_client,
new=sqs_client,
), mock.patch(
"prowler.providers.aws.services.sqs.sqs_client.sqs_client",
new=sqs_client,
):
from prowler.providers.aws.services.sqs.sqs_queues_not_publicly_accessible.sqs_queues_not_publicly_accessible import (
sqs_queues_not_publicly_accessible,
@@ -18,7 +18,10 @@ class Test_sqs_queues_server_side_encryption_enabled:
sqs_client.queues = []
with mock.patch(
"prowler.providers.aws.services.sqs.sqs_service.SQS",
sqs_client,
new=sqs_client,
), mock.patch(
"prowler.providers.aws.services.sqs.sqs_client.sqs_client",
new=sqs_client,
):
from prowler.providers.aws.services.sqs.sqs_queues_server_side_encryption_enabled.sqs_queues_server_side_encryption_enabled import (
sqs_queues_server_side_encryption_enabled,
@@ -42,7 +45,10 @@ class Test_sqs_queues_server_side_encryption_enabled:
)
with mock.patch(
"prowler.providers.aws.services.sqs.sqs_service.SQS",
sqs_client,
new=sqs_client,
), mock.patch(
"prowler.providers.aws.services.sqs.sqs_client.sqs_client",
new=sqs_client,
):
from prowler.providers.aws.services.sqs.sqs_queues_server_side_encryption_enabled.sqs_queues_server_side_encryption_enabled import (
sqs_queues_server_side_encryption_enabled,
@@ -72,7 +78,10 @@ class Test_sqs_queues_server_side_encryption_enabled:
)
with mock.patch(
"prowler.providers.aws.services.sqs.sqs_service.SQS",
sqs_client,
new=sqs_client,
), mock.patch(
"prowler.providers.aws.services.sqs.sqs_client.sqs_client",
new=sqs_client,
):
from prowler.providers.aws.services.sqs.sqs_queues_server_side_encryption_enabled.sqs_queues_server_side_encryption_enabled import (
sqs_queues_server_side_encryption_enabled,