Merge branch 'master' into PRWLR-4985-fix-resource-type-aws-metadata

This commit is contained in:
Rubén De la Torre Vico
2024-10-09 10:15:54 +02:00
32 changed files with 1997 additions and 995 deletions
+5 -6
View File
@@ -2,9 +2,9 @@ FROM python:3.12-alpine
LABEL maintainer="https://github.com/prowler-cloud/prowler"
# Update system dependencies
# Update system dependencies and install essential tools
#hadolint ignore=DL3018
RUN apk --no-cache upgrade && apk --no-cache add curl
RUN apk --no-cache upgrade && apk --no-cache add curl git
# Create nonroot user
RUN mkdir -p /home/prowler && \
@@ -13,18 +13,17 @@ RUN mkdir -p /home/prowler && \
chown -R prowler:prowler /home/prowler
USER prowler
# Copy necessary files
# Copy necessary files
WORKDIR /home/prowler
COPY prowler/ /home/prowler/prowler/
COPY dashboard/ /home/prowler/dashboard/
COPY pyproject.toml /home/prowler
COPY README.md /home/prowler
# Install dependencies
# Install Python dependencies
ENV HOME='/home/prowler'
ENV PATH="$HOME/.local/bin:$PATH"
#hadolint ignore=DL3013
RUN pip install --no-cache-dir --upgrade pip && \
RUN pip install --no-cache-dir --upgrade pip setuptools wheel && \
pip install --no-cache-dir .
# Remove deprecated dash dependencies
Generated
+10 -6
View File
@@ -1302,15 +1302,13 @@ wrapt = ">=1.10,<2"
dev = ["PyTest", "PyTest-Cov", "bump2version (<1)", "sphinx (<2)", "tox"]
[[package]]
name = "detect-secrets"
name = "detect_secrets"
version = "1.5.0"
description = "Tool for detecting secrets in the codebase"
optional = false
python-versions = "*"
files = [
{file = "detect_secrets-1.5.0-py3-none-any.whl", hash = "sha256:e24e7b9b5a35048c313e983f76c4bd09dad89f045ff059e354f9943bf45aa060"},
{file = "detect_secrets-1.5.0.tar.gz", hash = "sha256:6bb46dcc553c10df51475641bb30fd69d25645cc12339e46c824c1e0c388898a"},
]
files = []
develop = false
[package.dependencies]
pyyaml = "*"
@@ -1320,6 +1318,12 @@ requests = "*"
gibberish = ["gibberish-detector"]
word-list = ["pyahocorasick"]
[package.source]
type = "git"
url = "https://github.com/Yelp/detect-secrets.git"
reference = "master"
resolved_reference = "462720710ec337300fab2b4f2290949c7ee141eb"
[[package]]
name = "dill"
version = "0.3.9"
@@ -5068,4 +5072,4 @@ type = ["pytest-mypy"]
[metadata]
lock-version = "2.0"
python-versions = ">=3.9,<3.13"
content-hash = "f99a8859e7618ddc32babfd9b742ee8f8cc4520c274e245b358a0df252d56ee6"
content-hash = "0b367fa80501022efe43dc1beaa7f3da278fb64ffaddece72a0e88b09a0e53a2"
+2 -2
View File
@@ -335,8 +335,8 @@ aws:
elbv2_min_azs: 2
# Known secrets to ignore on detection
# this will include a list of regex patterns to ignore on detection
# AWS Secrets Configuration
# Patterns to ignore in the secrets checks
secrets_ignore_patterns: []
# Azure Configuration
+1 -1
View File
@@ -133,7 +133,7 @@ def detect_secrets_scan(
{"name": "SoftlayerDetector"},
{"name": "SquareOAuthDetector"},
{"name": "StripeDetector"},
# {"name": "TelegramBotTokenDetector"}, https://github.com/Yelp/detect-secrets/pull/878
{"name": "TelegramBotTokenDetector"},
{"name": "TwilioKeyDetector"},
],
"filters_used": [
@@ -10980,6 +10980,7 @@
"ap-east-1",
"ap-northeast-1",
"ap-northeast-2",
"ap-northeast-3",
"ap-south-1",
"ap-southeast-1",
"ap-southeast-2",
@@ -8,7 +8,6 @@ from prowler.lib.scan_filters.scan_filters import is_resource_filtered
from prowler.providers.aws.lib.service.service import AWSService
################################ ECS
class ECS(AWSService):
def __init__(self, provider):
# Call AWSService's __init__
@@ -0,0 +1,32 @@
{
"Provider": "aws",
"CheckID": "guardduty_lambda_protection_enabled",
"CheckTitle": "Check if GuardDuty Lambda Protection is enabled.",
"CheckType": [
"Software and Configuration Checks/AWS Security Best Practices"
],
"ServiceName": "guardduty",
"SubServiceName": "",
"ResourceIdTemplate": "arn:aws:guardduty:region:account-id/detector-id",
"Severity": "high",
"ResourceType": "",
"Description": "GuardDuty Lambda Protection helps you identify potential security threats when an AWS Lambda function gets invoked. After you enable Lambda Protection, GuardDuty starts monitoring Lambda network activity logs associated with the Lambda functions in your AWS account.",
"Risk": "If Lambda Protection is not enabled, GuardDuty will not be able to monitor Lambda network activity logs and may miss potential security threats.",
"RelatedUrl": "https://docs.aws.amazon.com/guardduty/latest/ug/lambda-protection.html",
"Remediation": {
"Code": {
"CLI": "aws guardduty update-detector --detector-id <detector-id> --features Name=LAMBDA_NETWORK_LOGS,Status=ENABLED",
"NativeIaC": "",
"Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/guardduty-controls.html#guardduty-6",
"Terraform": ""
},
"Recommendation": {
"Text": "Enable Lambda Protection in your GuardDuty detector to start monitoring Lambda Network Activity in your account.",
"Url": "https://docs.aws.amazon.com/guardduty/latest/ug/configure-lambda-protection-standalone-acc.html"
}
},
"Categories": [],
"Notes": "",
"DependsOn": [],
"RelatedTo": []
}
@@ -0,0 +1,21 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.guardduty.guardduty_client import guardduty_client
class guardduty_lambda_protection_enabled(Check):
def execute(self):
findings = []
for detector in guardduty_client.detectors:
if detector.status:
report = Check_Report_AWS(self.metadata())
report.region = detector.region
report.resource_id = detector.id
report.resource_arn = detector.arn
report.resource_tags = detector.tags
report.status = "FAIL"
report.status_extended = f"GuardDuty detector {detector.id} does not have Lambda Protection enabled."
if detector.lambda_protection:
report.status = "PASS"
report.status_extended = f"GuardDuty detector {detector.id} has Lambda Protection enabled."
findings.append(report)
return findings
@@ -33,7 +33,10 @@ class GuardDuty(AWSService):
):
self.detectors.append(
Detector(
id=detector, arn=arn, region=regional_client.region
id=detector,
arn=arn,
region=regional_client.region,
enabled_in_account=True,
)
)
if not detectors:
@@ -57,7 +60,7 @@ class GuardDuty(AWSService):
detector_info = self.regional_clients[detector.region].get_detector(
DetectorId=detector.id
)
if "Status" in detector_info and detector_info["Status"] == "ENABLED":
if detector_info.get("Status", "DISABLED") == "ENABLED":
detector.status = True
data_sources = detector_info.get("DataSources", {})
@@ -87,10 +90,16 @@ class GuardDuty(AWSService):
for feat in detector_info.get("Features", []):
if (
feat.get("Name") == "RDS_LOGIN_EVENTS"
feat.get("Name", "") == "RDS_LOGIN_EVENTS"
and feat.get("Status", "DISABLED") == "ENABLED"
):
detector.rds_protection = True
elif (
feat.get("Name", "") == "LAMBDA_NETWORK_LOGS"
and feat.get("Status", "DISABLED") == "ENABLED"
):
detector.lambda_protection = True
except Exception as error:
logger.error(
f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"
@@ -205,7 +214,7 @@ class Detector(BaseModel):
id: str
arn: str
region: str
enabled_in_account: bool = True
enabled_in_account: bool
status: bool = None
findings: list = []
member_accounts: list = []
@@ -214,4 +223,5 @@ class Detector(BaseModel):
s3_protection: bool = False
rds_protection: bool = False
eks_audit_log_protection: bool = False
lambda_protection: bool = False
ec2_malware_protection: bool = False
@@ -0,0 +1,34 @@
{
"Provider": "aws",
"CheckID": "redshift_cluster_in_transit_encryption_enabled",
"CheckTitle": "Check if connections to Amazon Redshift clusters are encrypted in transit.",
"CheckType": [
"Software and Configuration Checks/AWS Security Best Practices"
],
"ServiceName": "redshift",
"SubServiceName": "",
"ResourceIdTemplate": "arn:aws:redshift:region:account-id:cluster/cluster-name",
"Severity": "medium",
"ResourceType": "AwsRedshiftCluster",
"Description": "This control checks whether connections to Amazon Redshift clusters are required to use encryption in transit. The control fails if the Redshift cluster parameter 'require_SSL' isn't set to True.",
"Risk": "Without encryption in transit, connections to the Redshift cluster are vulnerable to eavesdropping or person-in-the-middle attacks, exposing sensitive data to unauthorized access.",
"RelatedUrl": "https://docs.aws.amazon.com/redshift/latest/mgmt/security-encryption-in-transit.html",
"Remediation": {
"Code": {
"CLI": "aws redshift modify-cluster-parameter-group --parameter-group-name <group-name> --parameters ParameterName=require_ssl,ParameterValue=true,ApplyType=static",
"NativeIaC": "",
"Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/redshift-controls.html#redshift-2",
"Terraform": ""
},
"Recommendation": {
"Text": "Ensure that connections to Amazon Redshift clusters use encryption in transit by setting the 'require_ssl' parameter to True.",
"Url": "https://www.trendmicro.com/cloudoneconformity/knowledge-base/aws/Redshift/redshift-parameter-groups-require-ssl.html"
}
},
"Categories": [
"encryption"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,26 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.redshift.redshift_client import redshift_client
class redshift_cluster_in_transit_encryption_enabled(Check):
def execute(self):
findings = []
for cluster in redshift_client.clusters:
report = Check_Report_AWS(self.metadata())
report.region = cluster.region
report.resource_id = cluster.id
report.resource_arn = cluster.arn
report.resource_tags = cluster.tags
report.status = "FAIL"
report.status_extended = (
f"Redshift Cluster {cluster.id} is not encrypted in transit."
)
if cluster.require_ssl:
report.status = "PASS"
report.status_extended = (
f"Redshift Cluster {cluster.id} is encrypted in transit."
)
findings.append(report)
return findings
@@ -13,8 +13,9 @@ class Redshift(AWSService):
super().__init__(__class__.__name__, provider)
self.clusters = []
self.__threading_call__(self._describe_clusters)
self._describe_logging_status(self.regional_clients)
self._describe_cluster_snapshots(self.regional_clients)
self.__threading_call__(self._describe_logging_status, self.clusters)
self.__threading_call__(self._describe_cluster_snapshots, self.clusters)
self.__threading_call__(self._describe_cluster_parameters, self.clusters)
def _describe_clusters(self, regional_client):
logger.info("Redshift - describing clusters...")
@@ -41,6 +42,9 @@ class Redshift(AWSService):
tags=cluster.get("Tags"),
master_username=cluster.get("MasterUsername", ""),
database_name=cluster.get("DBName", ""),
parameter_group_name=cluster.get(
"ClusterParameterGroups", [{}]
)[0].get("ParameterGroupName", ""),
)
self.clusters.append(cluster_to_append)
except Exception as error:
@@ -48,37 +52,52 @@ class Redshift(AWSService):
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
def _describe_logging_status(self, regional_clients):
def _describe_logging_status(self, cluster):
logger.info("Redshift - describing logging status...")
try:
for cluster in self.clusters:
regional_client = regional_clients[cluster.region]
cluster_attributes = regional_client.describe_logging_status(
ClusterIdentifier=cluster.id
)
if (
"LoggingEnabled" in cluster_attributes
and cluster_attributes["LoggingEnabled"]
):
cluster.logging_enabled = True
if "BucketName" in cluster_attributes:
cluster.bucket = cluster_attributes["BucketName"]
regional_client = self.regional_clients[cluster.region]
cluster_attributes = regional_client.describe_logging_status(
ClusterIdentifier=cluster.id
)
if (
"LoggingEnabled" in cluster_attributes
and cluster_attributes["LoggingEnabled"]
):
cluster.logging_enabled = True
if "BucketName" in cluster_attributes:
cluster.bucket = cluster_attributes["BucketName"]
except Exception as error:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
def _describe_cluster_snapshots(self, regional_clients):
def _describe_cluster_snapshots(self, cluster):
logger.info("Redshift - describing logging status...")
try:
for cluster in self.clusters:
regional_client = regional_clients[cluster.region]
cluster_snapshots = regional_client.describe_cluster_snapshots(
ClusterIdentifier=cluster.id
)
if "Snapshots" in cluster_snapshots and cluster_snapshots["Snapshots"]:
cluster.cluster_snapshots = True
regional_client = self.regional_clients[cluster.region]
cluster_snapshots = regional_client.describe_cluster_snapshots(
ClusterIdentifier=cluster.id
)
if "Snapshots" in cluster_snapshots and cluster_snapshots["Snapshots"]:
cluster.cluster_snapshots = True
except Exception as error:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
def _describe_cluster_parameters(self, cluster):
logger.info("Redshift - describing cluster parameter groups...")
try:
regional_client = self.regional_clients[cluster.region]
cluster_parameter_groups = regional_client.describe_cluster_parameters(
ClusterParameterGroupName=cluster.parameter_group_name
)
for parameter_group in cluster_parameter_groups["Parameters"]:
if parameter_group["ParameterName"].lower() == "require_ssl":
if parameter_group["ParameterValue"].lower() == "true":
cluster.require_ssl = True
except Exception as error:
logger.error(
@@ -100,3 +119,5 @@ class Cluster(BaseModel):
bucket: str = None
cluster_snapshots: bool = False
tags: Optional[list] = []
parameter_group_name: str = None
require_ssl: bool = False
+1 -1
View File
@@ -53,7 +53,7 @@ colorama = "0.4.6"
cryptography = "43.0.1"
dash = "2.18.1"
dash-bootstrap-components = "1.6.0"
detect-secrets = "1.5.0"
detect-secrets = {git = "https://github.com/Yelp/detect-secrets.git", rev = "master"}
google-api-python-client = "2.147.0"
google-auth-httplib2 = ">=0.1,<0.3"
jsonschema = "4.23.0"
+2 -2
View File
@@ -333,8 +333,8 @@ aws:
# Minimum number of Availability Zones that an ELBv2 must be in
elbv2_min_azs: 2
# Known secrets to ignore on detection
# this will include a list of regex patterns to ignore on detection
# AWS Secrets Configuration
# Patterns to ignore in the secrets checks
secrets_ignore_patterns: []
# Azure Configuration
@@ -28,6 +28,12 @@ def lambda_handler(event, context):
print("custom log event")
return event
"""
LAMBDA_FUNCTION_CODE_WITH_METADATA_API = """
def lambda_handler(event, context):
metadata_api = "169.254.169.254"
print("custom log event")
return event
"""
def create_lambda_function() -> Function:
@@ -59,6 +65,12 @@ def mock_get_function_codewithout_secrets():
)
def mock_get_function_codewith_metadata_api():
yield create_lambda_function(), get_lambda_code_with_secrets(
LAMBDA_FUNCTION_CODE_WITH_METADATA_API
)
class Test_awslambda_function_no_secrets_in_code:
def test_no_functions(self):
lambda_client = mock.MagicMock
@@ -144,3 +156,36 @@ class Test_awslambda_function_no_secrets_in_code:
== f"No secrets found in Lambda function {LAMBDA_FUNCTION_NAME} code."
)
assert result[0].resource_tags == []
def test_function_code_with_metadata_api(self):
lambda_client = mock.MagicMock
lambda_client.functions = {LAMBDA_FUNCTION_ARN: create_lambda_function()}
lambda_client._get_function_code = mock_get_function_codewith_metadata_api
lambda_client.audit_config = {"secrets_ignore_patterns": []}
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_aws_provider(),
), mock.patch(
"prowler.providers.aws.services.awslambda.awslambda_function_no_secrets_in_code.awslambda_function_no_secrets_in_code.awslambda_client",
new=lambda_client,
):
# Test Check
from prowler.providers.aws.services.awslambda.awslambda_function_no_secrets_in_code.awslambda_function_no_secrets_in_code import (
awslambda_function_no_secrets_in_code,
)
check = awslambda_function_no_secrets_in_code()
result = check.execute()
assert len(result) == 1
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_id == LAMBDA_FUNCTION_NAME
assert result[0].resource_arn == LAMBDA_FUNCTION_ARN
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"No secrets found in Lambda function {LAMBDA_FUNCTION_NAME} code."
)
assert result[0].resource_tags == []
@@ -1,22 +1,139 @@
from unittest import mock
from unittest.mock import patch
from prowler.providers.aws.services.ecs.ecs_service import Service
from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_US_EAST_1
import botocore
from boto3 import client
from moto import mock_aws
SERVICE_ARN = (
f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:service/sample-service"
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_US_EAST_1,
set_mocked_aws_provider,
)
SERVICE_NAME = "sample-service"
orig = botocore.client.BaseClient._make_api_call
def mock_make_api_call(self, operation_name, kwarg):
if operation_name == "DescribeServices":
if kwarg["services"] == [
f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:service/test-cluster/test-latest-linux-service"
]:
return {
"services": [
{
"serviceName": "test-latest-linux-service",
"clusterArn": f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster/test-cluster",
"taskDefinition": "test-task",
"loadBalancers": [],
"serviceArn": f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:service/test-cluster/test-latest-linux-service",
"desiredCount": 1,
"launchType": "FARGATE",
"platformVersion": "1.4.0",
"platformFamily": "Linux",
"networkConfiguration": {
"awsvpcConfiguration": {
"subnets": ["subnet-12345678"],
"securityGroups": ["sg-12345678"],
"assignPublicIp": "DISABLED",
},
},
"tags": [],
},
],
}
elif kwarg["services"] == [
f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:service/test-cluster/test-latest-windows-service"
]:
return {
"services": [
{
"serviceName": "test-latest-windows-service",
"clusterArn": f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster/test-cluster",
"taskDefinition": "test-task",
"loadBalancers": [],
"serviceArn": f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:service/test-cluster/test-latest-windows-service",
"desiredCount": 1,
"launchType": "FARGATE",
"platformVersion": "1.0.0",
"platformFamily": "Windows",
"networkConfiguration": {
"awsvpcConfiguration": {
"subnets": ["subnet-12345678"],
"securityGroups": ["sg-12345678"],
"assignPublicIp": "DISABLED",
},
},
"tags": [],
},
],
}
elif kwarg["services"] == [
f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:service/test-cluster/test-no-latest-linux-service"
]:
return {
"services": [
{
"serviceName": "test-no-latest-linux-service",
"clusterArn": f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster/test-cluster",
"taskDefinition": "test-task",
"loadBalancers": [],
"serviceArn": f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:service/test-cluster/test-no-latest-linux-service",
"desiredCount": 1,
"launchType": "FARGATE",
"platformVersion": "1.2.0",
"platformFamily": "Linux",
"networkConfiguration": {
"awsvpcConfiguration": {
"subnets": ["subnet-12345678"],
"securityGroups": ["sg-12345678"],
"assignPublicIp": "DISABLED",
},
},
"tags": [],
},
],
}
elif kwarg["services"] == [
f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:service/test-cluster/test-no-latest-windows-service"
]:
return {
"services": [
{
"serviceName": "test-no-latest-windows-service",
"clusterArn": f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster/test-cluster",
"taskDefinition": "test-task",
"loadBalancers": [],
"serviceArn": f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:service/test-cluster/test-no-latest-windows-service",
"desiredCount": 1,
"launchType": "FARGATE",
"platformVersion": "0.9.0",
"platformFamily": "Windows",
"networkConfiguration": {
"awsvpcConfiguration": {
"subnets": ["subnet-12345678"],
"securityGroups": ["sg-12345678"],
"assignPublicIp": "DISABLED",
},
},
"tags": [],
},
],
}
return orig(self, operation_name, kwarg)
class Test_ecs_service_fargate_latest_platform_version:
def test_no_services(self):
ecs_client = mock.MagicMock
ecs_client.services = {}
from prowler.providers.aws.services.ecs.ecs_service import ECS
with mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_service_fargate_latest_platform_version.ecs_service_fargate_latest_platform_version.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_service_fargate_latest_platform_version.ecs_service_fargate_latest_platform_version import (
ecs_service_fargate_latest_platform_version,
@@ -26,21 +143,31 @@ class Test_ecs_service_fargate_latest_platform_version:
result = check.execute()
assert len(result) == 0
@mock_aws
def test_service_ec2_type(self):
ecs_client = mock.MagicMock
ecs_client.services = {}
ecs_client.services[SERVICE_ARN] = Service(
name=SERVICE_NAME,
arn=SERVICE_ARN,
region=AWS_REGION_US_EAST_1,
launch_type="EC2",
assign_public_ip=False,
tags=[],
ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1)
ecs_client.create_cluster(clusterName="test-cluster")
ecs_client.create_service(
cluster="test-cluster",
serviceName="test-service",
launchType="EC2",
platformVersion="1.4.0",
desiredCount=1,
clientToken="test-token",
)
with mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
from prowler.providers.aws.services.ecs.ecs_service import ECS
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_service_fargate_latest_platform_version.ecs_service_fargate_latest_platform_version.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_service_fargate_latest_platform_version.ecs_service_fargate_latest_platform_version import (
ecs_service_fargate_latest_platform_version,
@@ -50,27 +177,34 @@ class Test_ecs_service_fargate_latest_platform_version:
result = check.execute()
assert len(result) == 0
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
@mock_aws
def test_service_linux_latest_version(self):
ecs_client = mock.MagicMock
ecs_client.services = {}
ecs_client.services[SERVICE_ARN] = Service(
name=SERVICE_NAME,
arn=SERVICE_ARN,
region=AWS_REGION_US_EAST_1,
launch_type="FARGATE",
platform_family="Linux",
platform_version="1.4.0",
assign_public_ip=False,
tags=[],
ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1)
ecs_client.create_cluster(clusterName="test-cluster")
ecs_client.create_service(
cluster="test-cluster",
serviceName="test-latest-linux-service",
launchType="FARGATE",
platformVersion="1.4.0",
desiredCount=1,
clientToken="test-token",
)
ecs_client.audit_config = {
"fargate_linux_latest_version": "1.4.0",
}
from prowler.providers.aws.services.ecs.ecs_service import ECS
with mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
mocked_ecs_client = ECS(mocked_aws_provider)
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_service_fargate_latest_platform_version.ecs_service_fargate_latest_platform_version.ecs_client",
new=mocked_ecs_client,
):
from prowler.providers.aws.services.ecs.ecs_service_fargate_latest_platform_version.ecs_service_fargate_latest_platform_version import (
ecs_service_fargate_latest_platform_version,
@@ -81,32 +215,46 @@ class Test_ecs_service_fargate_latest_platform_version:
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].status_extended == (
f"ECS Service {SERVICE_NAME} is using latest FARGATE Linux version 1.4.0."
"ECS Service test-latest-linux-service is using latest FARGATE Linux version 1.4.0."
)
assert result[0].resource_id == SERVICE_NAME
assert result[0].resource_arn == SERVICE_ARN
assert result[0].resource_id == "test-latest-linux-service"
assert (
result[0].resource_arn
== f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:service/test-cluster/test-latest-linux-service"
)
assert result[0].resource_tags == []
assert result[0].region == AWS_REGION_US_EAST_1
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
@mock_aws
def test_service_windows_latest_version(self):
ecs_client = mock.MagicMock
ecs_client.services = {}
ecs_client.services[SERVICE_ARN] = Service(
name=SERVICE_NAME,
arn=SERVICE_ARN,
region=AWS_REGION_US_EAST_1,
launch_type="FARGATE",
platform_family="Windows",
platform_version="1.0.0",
assign_public_ip=False,
tags=[],
ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1)
ecs_client.create_cluster(clusterName="test-cluster")
ecs_client.create_service(
cluster="test-cluster",
serviceName="test-latest-windows-service",
launchType="FARGATE",
platformVersion="1.0.0",
desiredCount=1,
clientToken="test-token",
)
ecs_client.audit_config = {
"fargate_windows_latest_version": "1.0.0",
}
with mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
from prowler.providers.aws.services.ecs.ecs_service import ECS
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_service_fargate_latest_platform_version.ecs_service_fargate_latest_platform_version.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_service_fargate_latest_platform_version.ecs_service_fargate_latest_platform_version import (
ecs_service_fargate_latest_platform_version,
@@ -117,32 +265,42 @@ class Test_ecs_service_fargate_latest_platform_version:
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].status_extended == (
f"ECS Service {SERVICE_NAME} is using latest FARGATE Windows version 1.0.0."
"ECS Service test-latest-windows-service is using latest FARGATE Windows version 1.0.0."
)
assert result[0].resource_id == SERVICE_NAME
assert result[0].resource_arn == SERVICE_ARN
assert result[0].resource_id == "test-latest-windows-service"
assert (
result[0].resource_arn
== f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:service/test-cluster/test-latest-windows-service"
)
assert result[0].resource_tags == []
assert result[0].region == AWS_REGION_US_EAST_1
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
@mock_aws
def test_service_linux_no_latest_version(self):
ecs_client = mock.MagicMock
ecs_client.services = {}
ecs_client.services[SERVICE_ARN] = Service(
name=SERVICE_NAME,
arn=SERVICE_ARN,
region=AWS_REGION_US_EAST_1,
launch_type="FARGATE",
platform_family="Linux",
platform_version="1.2.0",
assign_public_ip=False,
tags=[],
ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1)
ecs_client.create_cluster(clusterName="test-cluster")
ecs_client.create_service(
cluster="test-cluster",
serviceName="test-no-latest-linux-service",
launchType="FARGATE",
platformVersion="1.2.0",
desiredCount=1,
clientToken="test-token",
)
ecs_client.audit_config = {
"fargate_linux_latest_version": "1.4.0",
}
from prowler.providers.aws.services.ecs.ecs_service import ECS
with mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_service_fargate_latest_platform_version.ecs_service_fargate_latest_platform_version.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_service_fargate_latest_platform_version.ecs_service_fargate_latest_platform_version import (
ecs_service_fargate_latest_platform_version,
@@ -153,32 +311,42 @@ class Test_ecs_service_fargate_latest_platform_version:
assert len(result) == 1
assert result[0].status == "FAIL"
assert result[0].status_extended == (
f"ECS Service {SERVICE_NAME} is not using latest FARGATE Linux version 1.4.0, currently using 1.2.0."
"ECS Service test-no-latest-linux-service is not using latest FARGATE Linux version 1.4.0, currently using 1.2.0."
)
assert result[0].resource_id == SERVICE_NAME
assert result[0].resource_arn == SERVICE_ARN
assert result[0].resource_id == "test-no-latest-linux-service"
assert (
result[0].resource_arn
== f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:service/test-cluster/test-no-latest-linux-service"
)
assert result[0].resource_tags == []
assert result[0].region == AWS_REGION_US_EAST_1
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
@mock_aws
def test_service_windows_no_latest_version(self):
ecs_client = mock.MagicMock
ecs_client.services = {}
ecs_client.services[SERVICE_ARN] = Service(
name=SERVICE_NAME,
arn=SERVICE_ARN,
region=AWS_REGION_US_EAST_1,
launch_type="FARGATE",
platform_family="Windows",
platform_version="0.9.0",
assign_public_ip=False,
tags=[],
ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1)
ecs_client.create_cluster(clusterName="test-cluster")
ecs_client.create_service(
cluster="test-cluster",
serviceName="test-no-latest-windows-service",
launchType="FARGATE",
platformVersion="0.9.0",
desiredCount=1,
clientToken="test-token",
)
ecs_client.audit_config = {
"fargate_windows_latest_version": "1.0.0",
}
from prowler.providers.aws.services.ecs.ecs_service import ECS
with mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_service_fargate_latest_platform_version.ecs_service_fargate_latest_platform_version.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_service_fargate_latest_platform_version.ecs_service_fargate_latest_platform_version import (
ecs_service_fargate_latest_platform_version,
@@ -189,7 +357,12 @@ class Test_ecs_service_fargate_latest_platform_version:
assert len(result) == 1
assert result[0].status == "FAIL"
assert result[0].status_extended == (
f"ECS Service {SERVICE_NAME} is not using latest FARGATE Windows version 1.0.0, currently using 0.9.0."
"ECS Service test-no-latest-windows-service is not using latest FARGATE Windows version 1.0.0, currently using 0.9.0."
)
assert result[0].resource_id == SERVICE_NAME
assert result[0].resource_arn == SERVICE_ARN
assert result[0].resource_id == "test-no-latest-windows-service"
assert (
result[0].resource_arn
== f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:service/test-cluster/test-no-latest-windows-service"
)
assert result[0].resource_tags == []
assert result[0].region == AWS_REGION_US_EAST_1
@@ -1,22 +1,87 @@
from unittest import mock
from unittest.mock import patch
from prowler.providers.aws.services.ecs.ecs_service import Service
from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_US_EAST_1
import botocore
from boto3 import client
from moto import mock_aws
SERVICE_ARN = (
f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:service/sample-service"
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_US_EAST_1,
set_mocked_aws_provider,
)
SERVICE_NAME = "sample-service"
orig = botocore.client.BaseClient._make_api_call
def mock_make_api_call(self, operation_name, kwarg):
if operation_name == "DescribeServices":
if kwarg["services"] == [
f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:service/sample-cluster/service-with-no-public-ip"
]:
return {
"services": [
{
"serviceName": "test-latest-linux-service",
"clusterArn": f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster/sample-cluster",
"taskDefinition": "test-task",
"loadBalancers": [],
"serviceArn": f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:service/sample-cluster/service-with-no-public-ip",
"desiredCount": 1,
"launchType": "FARGATE",
"platformVersion": "1.4.0",
"platformFamily": "Linux",
"networkConfiguration": {
"awsvpcConfiguration": {
"subnets": ["subnet-12345678"],
"securityGroups": ["sg-12345678"],
"assignPublicIp": "DISABLED",
},
},
"tags": [],
},
],
}
elif kwarg["services"] == [
f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:service/sample-cluster/service-with-public-ip"
]:
return {
"services": [
{
"serviceName": "test-latest-linux-service",
"clusterArn": f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster/sample-cluster",
"taskDefinition": "test-task",
"loadBalancers": [],
"serviceArn": f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:service/sample-cluster/service-with-public-ip",
"desiredCount": 1,
"launchType": "FARGATE",
"platformVersion": "1.4.0",
"platformFamily": "Linux",
"networkConfiguration": {
"awsvpcConfiguration": {
"subnets": ["subnet-12345678"],
"securityGroups": ["sg-12345678"],
"assignPublicIp": "ENABLED",
},
},
"tags": [],
},
],
}
return orig(self, operation_name, kwarg)
class Test_ecs_service_no_assign_public_ip:
def test_no_services(self):
ecs_client = mock.MagicMock
ecs_client.services = {}
from prowler.providers.aws.services.ecs.ecs_service import ECS
with mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_service_no_assign_public_ip.ecs_service_no_assign_public_ip.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_service_no_assign_public_ip.ecs_service_no_assign_public_ip import (
ecs_service_no_assign_public_ip,
@@ -26,20 +91,37 @@ class Test_ecs_service_no_assign_public_ip:
result = check.execute()
assert len(result) == 0
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
@mock_aws
def test_service_with_no_public_ip(self):
ecs_client = mock.MagicMock
ecs_client.services = {}
ecs_client.services[SERVICE_ARN] = Service(
name=SERVICE_NAME,
arn=SERVICE_ARN,
region=AWS_REGION_US_EAST_1,
assign_public_ip=False,
tags=[],
)
ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1)
with mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
ecs_client.create_cluster(clusterName="sample-cluster")
service_arn = ecs_client.create_service(
cluster="sample-cluster",
serviceName="service-with-no-public-ip",
desiredCount=1,
launchType="FARGATE",
networkConfiguration={
"awsvpcConfiguration": {
"subnets": ["subnet-123456"],
"securityGroups": ["sg-123456"],
"assignPublicIp": "DISABLED",
}
},
)["service"]["serviceArn"]
from prowler.providers.aws.services.ecs.ecs_service import ECS
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_service_no_assign_public_ip.ecs_service_no_assign_public_ip.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_service_no_assign_public_ip.ecs_service_no_assign_public_ip import (
ecs_service_no_assign_public_ip,
@@ -51,25 +133,44 @@ class Test_ecs_service_no_assign_public_ip:
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"ECS Service {SERVICE_NAME} does not have automatic public IP assignment."
== "ECS Service service-with-no-public-ip does not have automatic public IP assignment."
)
assert result[0].resource_id == SERVICE_NAME
assert result[0].resource_arn == SERVICE_ARN
assert result[0].resource_id == "service-with-no-public-ip"
assert result[0].resource_arn == service_arn
assert result[0].resource_tags == []
assert result[0].region == AWS_REGION_US_EAST_1
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
@mock_aws
def test_task_definition_no_host_network_mode(self):
ecs_client = mock.MagicMock
ecs_client.services = {}
ecs_client.services[SERVICE_ARN] = Service(
name=SERVICE_NAME,
arn=SERVICE_ARN,
region=AWS_REGION_US_EAST_1,
assign_public_ip=True,
tags=[],
)
ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1)
with mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
ecs_client.create_cluster(clusterName="sample-cluster")
service_arn = ecs_client.create_service(
cluster="sample-cluster",
serviceName="service-with-public-ip",
desiredCount=1,
launchType="FARGATE",
networkConfiguration={
"awsvpcConfiguration": {
"subnets": ["subnet-123456"],
"securityGroups": ["sg-123456"],
"assignPublicIp": "ENABLED",
}
},
)["service"]["serviceArn"]
from prowler.providers.aws.services.ecs.ecs_service import ECS
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_service_no_assign_public_ip.ecs_service_no_assign_public_ip.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_service_no_assign_public_ip.ecs_service_no_assign_public_ip import (
ecs_service_no_assign_public_ip,
@@ -81,7 +182,9 @@ class Test_ecs_service_no_assign_public_ip:
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"ECS Service {SERVICE_NAME} has automatic public IP assignment."
== "ECS Service service-with-public-ip has automatic public IP assignment."
)
assert result[0].resource_id == SERVICE_NAME
assert result[0].resource_arn == SERVICE_ARN
assert result[0].resource_id == "service-with-public-ip"
assert result[0].resource_arn == service_arn
assert result[0].resource_tags == []
assert result[0].region == AWS_REGION_US_EAST_1
@@ -1,10 +1,13 @@
from unittest import mock
from unittest.mock import patch
from prowler.providers.aws.services.ecs.ecs_service import (
ContainerDefinition,
TaskDefinition,
from boto3 import client
from moto import mock_aws
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_US_EAST_1,
set_mocked_aws_provider,
)
from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_US_EAST_1
TASK_NAME = "test-task-readonly"
TASK_REVISION = "1"
@@ -14,12 +17,16 @@ TASK_ARN = f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:task-defini
class Test_ecs_task_definitions_containers_readonly_access:
def test_no_task_definitions(self):
ecs_client = mock.MagicMock
ecs_client.task_definitions = {}
from prowler.providers.aws.services.ecs.ecs_service import ECS
with mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_task_definitions_containers_readonly_access.ecs_task_definitions_containers_readonly_access.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_task_definitions_containers_readonly_access.ecs_task_definitions_containers_readonly_access import (
ecs_task_definitions_containers_readonly_access,
@@ -29,29 +36,35 @@ class Test_ecs_task_definitions_containers_readonly_access:
result = check.execute()
assert len(result) == 0
@mock_aws
def test_task_definition_all_containers_readonly(self):
ecs_client = mock.MagicMock
ecs_client.task_definitions = {}
ecs_client.task_definitions[TASK_ARN] = TaskDefinition(
name=TASK_NAME,
arn=TASK_ARN,
revision=TASK_REVISION,
region=AWS_REGION_US_EAST_1,
network_mode="bridge",
container_definitions=[
ContainerDefinition(
name=CONTAINER_NAME,
readonly_rootfilesystem=True,
privileged=False,
user="appuser",
environment=[],
)
],
)
ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1)
with mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
task_definition_arn = ecs_client.register_task_definition(
family=TASK_NAME,
containerDefinitions=[
{
"name": CONTAINER_NAME,
"image": "ubuntu",
"memory": 128,
"readonlyRootFilesystem": True,
"privileged": False,
"user": "appuser",
"environment": [],
}
],
)["taskDefinition"]["taskDefinitionArn"]
from prowler.providers.aws.services.ecs.ecs_service import ECS
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_task_definitions_containers_readonly_access.ecs_task_definitions_containers_readonly_access.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_task_definitions_containers_readonly_access.ecs_task_definitions_containers_readonly_access import (
ecs_task_definitions_containers_readonly_access,
@@ -65,30 +78,40 @@ class Test_ecs_task_definitions_containers_readonly_access:
result[0].status_extended
== f"ECS task definition {TASK_NAME} with revision {TASK_REVISION} does not have containers with write access to the root filesystems."
)
assert result[0].resource_id == f"{TASK_NAME}:{TASK_REVISION}"
assert result[0].resource_arn == task_definition_arn
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_tags == []
@mock_aws
def test_task_definition_some_containers_not_readonly(self):
ecs_client = mock.MagicMock
ecs_client.task_definitions = {}
ecs_client.task_definitions[TASK_ARN] = TaskDefinition(
name=TASK_NAME,
arn=TASK_ARN,
revision=TASK_REVISION,
region=AWS_REGION_US_EAST_1,
network_mode="bridge",
container_definitions=[
ContainerDefinition(
name=CONTAINER_NAME,
readonly_rootfilesystem=False,
privileged=False,
user="appuser",
environment=[],
)
ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1)
task_definition_arn = ecs_client.register_task_definition(
family=TASK_NAME,
containerDefinitions=[
{
"name": CONTAINER_NAME,
"image": "ubuntu",
"memory": 128,
"readonlyRootFilesystem": False,
"privileged": False,
"user": "appuser",
"environment": [],
}
],
)
)["taskDefinition"]["taskDefinitionArn"]
with mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
from prowler.providers.aws.services.ecs.ecs_service import ECS
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_task_definitions_containers_readonly_access.ecs_task_definitions_containers_readonly_access.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_task_definitions_containers_readonly_access.ecs_task_definitions_containers_readonly_access import (
ecs_task_definitions_containers_readonly_access,
@@ -102,38 +125,49 @@ class Test_ecs_task_definitions_containers_readonly_access:
result[0].status_extended
== f"ECS task definition {TASK_NAME} with revision {TASK_REVISION} has containers with write access to the root filesystem: {CONTAINER_NAME}"
)
assert result[0].resource_id == f"{TASK_NAME}:{TASK_REVISION}"
assert result[0].resource_arn == task_definition_arn
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_tags == []
@mock_aws
def test_task_definition_mixed_containers(self):
ecs_client = mock.MagicMock
ecs_client.task_definitions = {
TASK_ARN: TaskDefinition(
name=TASK_NAME,
arn=TASK_ARN,
revision=TASK_REVISION,
region=AWS_REGION_US_EAST_1,
network_mode="bridge",
container_definitions=[
ContainerDefinition(
name=CONTAINER_NAME,
readonly_rootfilesystem=False,
privileged=False,
user="appuser",
environment=[],
),
ContainerDefinition(
name="readonly-container",
readonly_rootfilesystem=True,
privileged=False,
user="appuser",
environment=[],
),
],
)
}
ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1)
with mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
task_definition_arn = ecs_client.register_task_definition(
family=TASK_NAME,
containerDefinitions=[
{
"name": CONTAINER_NAME,
"image": "ubuntu",
"memory": 128,
"readonlyRootFilesystem": False, # Not readonly
"privileged": False,
"user": "appuser",
"environment": [],
},
{
"name": "readonly-container",
"image": "ubuntu",
"memory": 128,
"readonlyRootFilesystem": True, # Readonly
"privileged": False,
"user": "appuser",
"environment": [],
},
],
)["taskDefinition"]["taskDefinitionArn"]
from prowler.providers.aws.services.ecs.ecs_service import ECS
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_task_definitions_containers_readonly_access.ecs_task_definitions_containers_readonly_access.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_task_definitions_containers_readonly_access.ecs_task_definitions_containers_readonly_access import (
ecs_task_definitions_containers_readonly_access,
@@ -147,3 +181,7 @@ class Test_ecs_task_definitions_containers_readonly_access:
result[0].status_extended
== f"ECS task definition {TASK_NAME} with revision {TASK_REVISION} has containers with write access to the root filesystem: {CONTAINER_NAME}"
)
assert result[0].resource_id == f"{TASK_NAME}:{TASK_REVISION}"
assert result[0].resource_arn == task_definition_arn
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_tags == []
@@ -1,33 +1,28 @@
from unittest import mock
from unittest.mock import patch
from prowler.providers.aws.services.ecs.ecs_service import (
ContainerDefinition,
ContainerEnvVariable,
TaskDefinition,
)
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_US_EAST_1,
set_mocked_aws_provider,
)
from boto3 import client
from moto import mock_aws
task_name = "test-task"
task_revision = "1"
container_name = "test-container"
task_arn = f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:task-definition/{task_name}:{task_revision}"
from tests.providers.aws.utils import AWS_REGION_US_EAST_1, set_mocked_aws_provider
TASK_NAME = "test-task"
TASK_REVISION = "1"
CONTAINER_NAME = "test-container"
class Test_ecs_task_definitions_host_namespace_not_shared:
@mock_aws
def test_no_task_definitions(self):
ecs_client = mock.MagicMock
ecs_client.task_definitions = {}
from prowler.providers.aws.services.ecs.ecs_service import ECS
with mock.patch(
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]),
), mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_task_definitions_host_namespace_not_shared.ecs_task_definitions_host_namespace_not_shared.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_task_definitions_host_namespace_not_shared.ecs_task_definitions_host_namespace_not_shared import (
ecs_task_definitions_host_namespace_not_shared,
@@ -37,37 +32,36 @@ class Test_ecs_task_definitions_host_namespace_not_shared:
result = check.execute()
assert len(result) == 0
@mock_aws
def test_task_definition_no_host_pid_mode(self):
ecs_client = mock.MagicMock
ecs_client.task_definitions = {}
ecs_client.task_definitions[task_arn] = TaskDefinition(
name=task_name,
arn=task_arn,
revision=task_revision,
region=AWS_REGION_US_EAST_1,
network_mode="bridge",
pid_mode="task",
container_definitions=[
ContainerDefinition(
name=container_name,
privileged=False,
user="",
environment=[
ContainerEnvVariable(
name="env_var_name_no_secrets",
value="env_var_value_no_secrets",
)
],
)
],
)
ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1)
with mock.patch(
task_arn = ecs_client.register_task_definition(
family=TASK_NAME,
containerDefinitions=[
{
"name": CONTAINER_NAME,
"image": "ubuntu",
"memory": 128,
"readonlyRootFilesystem": True,
"privileged": False,
"user": "appuser",
"environment": [],
}
],
pidMode="task",
)["taskDefinition"]["taskDefinitionArn"]
from prowler.providers.aws.services.ecs.ecs_service import ECS
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]),
), mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_task_definitions_host_namespace_not_shared.ecs_task_definitions_host_namespace_not_shared.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_task_definitions_host_namespace_not_shared.ecs_task_definitions_host_namespace_not_shared import (
ecs_task_definitions_host_namespace_not_shared,
@@ -79,35 +73,43 @@ class Test_ecs_task_definitions_host_namespace_not_shared:
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"ECS task definition {task_name} with revision {task_revision} does not share a host's process namespace with its containers."
== f"ECS task definition {TASK_NAME} with revision {TASK_REVISION} does not share a host's process namespace with its containers."
)
assert result[0].resource_id == f"{TASK_NAME}:{TASK_REVISION}"
assert result[0].resource_arn == task_arn
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_tags == []
@mock_aws
def test_task_definition_host_pid_mode(self):
ecs_client = mock.MagicMock
ecs_client.task_definitions = {}
ecs_client.task_definitions[task_arn] = TaskDefinition(
name=task_name,
arn=task_arn,
revision=task_revision,
region=AWS_REGION_US_EAST_1,
network_mode="host",
pid_mode="host",
container_definitions=[
ContainerDefinition(
name=container_name,
privileged=False,
user="root",
environment=[],
)
],
)
ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1)
with mock.patch(
task_arn = ecs_client.register_task_definition(
family=TASK_NAME,
containerDefinitions=[
{
"name": CONTAINER_NAME,
"image": "ubuntu",
"memory": 128,
"readonlyRootFilesystem": True,
"privileged": False,
"user": "appuser",
"environment": [],
}
],
pidMode="host",
)["taskDefinition"]["taskDefinitionArn"]
from prowler.providers.aws.services.ecs.ecs_service import ECS
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]),
), mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_task_definitions_host_namespace_not_shared.ecs_task_definitions_host_namespace_not_shared.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_task_definitions_host_namespace_not_shared.ecs_task_definitions_host_namespace_not_shared import (
ecs_task_definitions_host_namespace_not_shared,
@@ -119,35 +121,42 @@ class Test_ecs_task_definitions_host_namespace_not_shared:
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"ECS task definition {task_name} with revision {task_revision} is configured to share a host's process namespace with its containers."
== f"ECS task definition {TASK_NAME} with revision {TASK_REVISION} is configured to share a host's process namespace with its containers."
)
assert result[0].resource_id == f"{TASK_NAME}:{TASK_REVISION}"
assert result[0].resource_arn == task_arn
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_tags == []
@mock_aws
def test_task_definition_no_pid_mode(self):
ecs_client = mock.MagicMock
ecs_client.task_definitions = {}
ecs_client.task_definitions[task_arn] = TaskDefinition(
name=task_name,
arn=f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:task-definition/{task_name}:{task_revision}",
revision=task_revision,
region=AWS_REGION_US_EAST_1,
network_mode="host",
pid_mode="",
container_definitions=[
ContainerDefinition(
name=container_name,
privileged=True,
user="root",
environment=[],
)
],
)
ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1)
with mock.patch(
task_arn = ecs_client.register_task_definition(
family=TASK_NAME,
containerDefinitions=[
{
"name": CONTAINER_NAME,
"image": "ubuntu",
"memory": 128,
"readonlyRootFilesystem": True,
"privileged": False,
"user": "appuser",
"environment": [],
}
],
)["taskDefinition"]["taskDefinitionArn"]
from prowler.providers.aws.services.ecs.ecs_service import ECS
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]),
), mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_task_definitions_host_namespace_not_shared.ecs_task_definitions_host_namespace_not_shared.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_task_definitions_host_namespace_not_shared.ecs_task_definitions_host_namespace_not_shared import (
ecs_task_definitions_host_namespace_not_shared,
@@ -159,5 +168,9 @@ class Test_ecs_task_definitions_host_namespace_not_shared:
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"ECS task definition {task_name} with revision {task_revision} does not share a host's process namespace with its containers."
== f"ECS task definition {TASK_NAME} with revision {TASK_REVISION} does not share a host's process namespace with its containers."
)
assert result[0].resource_id == f"{TASK_NAME}:{TASK_REVISION}"
assert result[0].resource_arn == task_arn
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_tags == []
@@ -1,26 +1,27 @@
from unittest import mock
from unittest.mock import patch
from prowler.providers.aws.services.ecs.ecs_service import (
ContainerDefinition,
ContainerEnvVariable,
TaskDefinition,
)
from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_US_EAST_1
from boto3 import client
from moto import mock_aws
TASK_NAME = "test-task-hostmode"
from tests.providers.aws.utils import AWS_REGION_US_EAST_1, set_mocked_aws_provider
TASK_NAME = "test-task"
TASK_REVISION = "1"
CONTAINER_NAME = "test-container"
TASK_ARN = f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:task-definition/{TASK_NAME}:{TASK_REVISION}"
class Test_ecs_task_definitions_host_networking_mode_users:
def test_no_task_definitions(self):
ecs_client = mock.MagicMock
ecs_client.task_definitions = {}
from prowler.providers.aws.services.ecs.ecs_service import ECS
with mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_task_definitions_host_networking_mode_users.ecs_task_definitions_host_networking_mode_users.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_task_definitions_host_networking_mode_users.ecs_task_definitions_host_networking_mode_users import (
ecs_task_definitions_host_networking_mode_users,
@@ -30,33 +31,36 @@ class Test_ecs_task_definitions_host_networking_mode_users:
result = check.execute()
assert len(result) == 0
@mock_aws
def test_task_definition_no_host_network_mode(self):
ecs_client = mock.MagicMock
ecs_client.task_definitions = {}
ecs_client.task_definitions[TASK_ARN] = TaskDefinition(
name=TASK_NAME,
arn=TASK_ARN,
revision=TASK_REVISION,
region=AWS_REGION_US_EAST_1,
network_mode="bridge",
container_definitions=[
ContainerDefinition(
name=CONTAINER_NAME,
privileged=False,
user="",
environment=[
ContainerEnvVariable(
name="env_var_name_no_secrets",
value="env_var_value_no_secrets",
)
],
)
],
)
ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1)
with mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
task_arn = ecs_client.register_task_definition(
family=TASK_NAME,
networkMode="bridge",
containerDefinitions=[
{
"name": CONTAINER_NAME,
"image": "ubuntu",
"memory": 128,
"readonlyRootFilesystem": True,
"privileged": False,
"user": "appuser",
"environment": [],
}
],
)["taskDefinition"]["taskDefinitionArn"]
from prowler.providers.aws.services.ecs.ecs_service import ECS
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_task_definitions_host_networking_mode_users.ecs_task_definitions_host_networking_mode_users.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_task_definitions_host_networking_mode_users.ecs_task_definitions_host_networking_mode_users import (
ecs_task_definitions_host_networking_mode_users,
@@ -70,29 +74,41 @@ class Test_ecs_task_definitions_host_networking_mode_users:
result[0].status_extended
== f"ECS task definition {TASK_NAME} with revision {TASK_REVISION} does not have host network mode."
)
assert result[0].resource_id == f"{TASK_NAME}:{TASK_REVISION}"
assert result[0].resource_arn == task_arn
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_tags == []
@mock_aws
def test_task_definition_host_mode_container_root_non_privileged(self):
ecs_client = mock.MagicMock
ecs_client.task_definitions = {}
ecs_client.task_definitions[TASK_ARN] = TaskDefinition(
name=TASK_NAME,
arn=TASK_ARN,
revision=TASK_REVISION,
region=AWS_REGION_US_EAST_1,
network_mode="host",
container_definitions=[
ContainerDefinition(
name=CONTAINER_NAME,
privileged=False,
user="root",
environment=[],
)
],
)
ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1)
with mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
task_arn = ecs_client.register_task_definition(
family=TASK_NAME,
networkMode="host",
containerDefinitions=[
{
"name": CONTAINER_NAME,
"image": "ubuntu",
"memory": 128,
"readonlyRootFilesystem": True,
"privileged": False,
"user": "root",
"environment": [],
}
],
)["taskDefinition"]["taskDefinitionArn"]
from prowler.providers.aws.services.ecs.ecs_service import ECS
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_task_definitions_host_networking_mode_users.ecs_task_definitions_host_networking_mode_users.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_task_definitions_host_networking_mode_users.ecs_task_definitions_host_networking_mode_users import (
ecs_task_definitions_host_networking_mode_users,
@@ -106,29 +122,41 @@ class Test_ecs_task_definitions_host_networking_mode_users:
result[0].status_extended
== f"ECS task definition {TASK_NAME} with revision {TASK_REVISION} has containers with host network mode and non-privileged containers running as root or with no user specified: {CONTAINER_NAME}"
)
assert result[0].resource_id == f"{TASK_NAME}:{TASK_REVISION}"
assert result[0].resource_arn == task_arn
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_tags == []
@mock_aws
def test_task_definition_host_mode_container_privileged(self):
ecs_client = mock.MagicMock
ecs_client.task_definitions = {}
ecs_client.task_definitions[TASK_ARN] = TaskDefinition(
name=TASK_NAME,
arn=f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:task-definition/{TASK_NAME}:{TASK_REVISION}",
revision=TASK_REVISION,
region=AWS_REGION_US_EAST_1,
network_mode="host",
container_definitions=[
ContainerDefinition(
name=CONTAINER_NAME,
privileged=True,
user="root",
environment=[],
)
],
)
ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1)
with mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
task_arn = ecs_client.register_task_definition(
family=TASK_NAME,
networkMode="host",
containerDefinitions=[
{
"name": CONTAINER_NAME,
"image": "ubuntu",
"memory": 128,
"readonlyRootFilesystem": True,
"privileged": True,
"user": "root",
"environment": [],
}
],
)["taskDefinition"]["taskDefinitionArn"]
from prowler.providers.aws.services.ecs.ecs_service import ECS
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_task_definitions_host_networking_mode_users.ecs_task_definitions_host_networking_mode_users.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_task_definitions_host_networking_mode_users.ecs_task_definitions_host_networking_mode_users import (
ecs_task_definitions_host_networking_mode_users,
@@ -142,29 +170,41 @@ class Test_ecs_task_definitions_host_networking_mode_users:
result[0].status_extended
== f"ECS task definition {TASK_NAME} with revision {TASK_REVISION} has host network mode but no containers running as root or with no user specified."
)
assert result[0].resource_id == f"{TASK_NAME}:{TASK_REVISION}"
assert result[0].resource_arn == task_arn
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_tags == []
@mock_aws
def test_task_definition_host_mode_container_not_root(self):
ecs_client = mock.MagicMock
ecs_client.task_definitions = {}
ecs_client.task_definitions[TASK_ARN] = TaskDefinition(
name=TASK_NAME,
arn=TASK_ARN,
revision=TASK_REVISION,
region=AWS_REGION_US_EAST_1,
network_mode="host",
container_definitions=[
ContainerDefinition(
name=CONTAINER_NAME,
privileged=False,
user="appuser",
environment=[],
)
],
)
ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1)
with mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
task_arn = ecs_client.register_task_definition(
family=TASK_NAME,
networkMode="host",
containerDefinitions=[
{
"name": CONTAINER_NAME,
"image": "ubuntu",
"memory": 128,
"readonlyRootFilesystem": True,
"privileged": False,
"user": "appuser",
"environment": [],
}
],
)["taskDefinition"]["taskDefinitionArn"]
from prowler.providers.aws.services.ecs.ecs_service import ECS
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_task_definitions_host_networking_mode_users.ecs_task_definitions_host_networking_mode_users.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_task_definitions_host_networking_mode_users.ecs_task_definitions_host_networking_mode_users import (
ecs_task_definitions_host_networking_mode_users,
@@ -178,3 +218,7 @@ class Test_ecs_task_definitions_host_networking_mode_users:
result[0].status_extended
== f"ECS task definition {TASK_NAME} with revision {TASK_REVISION} has host network mode but no containers running as root or with no user specified."
)
assert result[0].resource_id == f"{TASK_NAME}:{TASK_REVISION}"
assert result[0].resource_arn == task_arn
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_tags == []
@@ -1,28 +1,27 @@
from unittest import mock
from unittest.mock import patch
from prowler.providers.aws.services.ecs.ecs_service import (
ContainerDefinition,
TaskDefinition,
)
from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_US_EAST_1
from boto3 import client
from moto import mock_aws
from tests.providers.aws.utils import AWS_REGION_US_EAST_1, set_mocked_aws_provider
TASK_NAME = "test-task"
TASK_REVISION = "1"
CONTAINER_NAME = "test-container"
TASK_ARN = f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:task-definition/{TASK_NAME}:{TASK_REVISION}"
class Test_ecs_task_definitions_logging_enabled:
def test_no_task_definitions(self):
ecs_client = mock.MagicMock
ecs_client.task_definitions = {}
from prowler.providers.aws.services.ecs.ecs_service import ECS
with mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
), mock.patch(
"prowler.providers.aws.services.ecs.ecs_client.ecs_client",
ecs_client,
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_task_definitions_logging_enabled.ecs_task_definitions_logging_enabled.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_task_definitions_logging_enabled.ecs_task_definitions_logging_enabled import (
ecs_task_definitions_logging_enabled,
@@ -32,73 +31,35 @@ class Test_ecs_task_definitions_logging_enabled:
result = check.execute()
assert len(result) == 0
@mock_aws
def test_task_definition_no_logconfiguration(self):
ecs_client = mock.MagicMock
ecs_client.task_definitions = {}
ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1)
ecs_client.task_definitions[TASK_ARN] = TaskDefinition(
name=TASK_NAME,
arn=TASK_ARN,
revision=TASK_REVISION,
region=AWS_REGION_US_EAST_1,
network_mode="host",
container_definitions=[
ContainerDefinition(
name=CONTAINER_NAME,
image="test-image",
privileged=False,
user="user-1",
environment=[{"name": "DB_PASSWORD", "value": "pass-12343"}],
)
task_arn = ecs_client.register_task_definition(
family=TASK_NAME,
containerDefinitions=[
{
"name": CONTAINER_NAME,
"image": "ubuntu",
"memory": 128,
"readonlyRootFilesystem": True,
"privileged": False,
"user": "appuser",
"environment": [],
}
],
)
)["taskDefinition"]["taskDefinitionArn"]
with mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
), mock.patch(
"prowler.providers.aws.services.ecs.ecs_client.ecs_client",
ecs_client,
):
from prowler.providers.aws.services.ecs.ecs_task_definitions_logging_enabled.ecs_task_definitions_logging_enabled import (
ecs_task_definitions_logging_enabled,
)
check = ecs_task_definitions_logging_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"ECS task definition {TASK_NAME} with revision {TASK_REVISION} has containers running with no logging configuration: {CONTAINER_NAME}"
)
def test_task_definition_no_logdriver(self):
ecs_client = mock.MagicMock
ecs_client.task_definitions = {}
ecs_client.task_definitions[TASK_ARN] = TaskDefinition(
name=TASK_NAME,
arn=TASK_ARN,
revision=TASK_REVISION,
region=AWS_REGION_US_EAST_1,
network_mode="host",
container_definitions=[
ContainerDefinition(
name=CONTAINER_NAME,
privileged=True,
user="root",
environment=[],
log_driver="",
)
],
)
with mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
), mock.patch(
"prowler.providers.aws.services.ecs.ecs_client.ecs_client",
ecs_client,
from prowler.providers.aws.services.ecs.ecs_service import ECS
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_task_definitions_logging_enabled.ecs_task_definitions_logging_enabled.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_task_definitions_logging_enabled.ecs_task_definitions_logging_enabled import (
ecs_task_definitions_logging_enabled,
@@ -112,33 +73,41 @@ class Test_ecs_task_definitions_logging_enabled:
result[0].status_extended
== f"ECS task definition {TASK_NAME} with revision {TASK_REVISION} has containers running with no logging configuration: {CONTAINER_NAME}"
)
assert result[0].resource_id == f"{TASK_NAME}:{TASK_REVISION}"
assert result[0].resource_arn == task_arn
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_tags == []
@mock_aws
def test_task_definition_privileged_container(self):
ecs_client = mock.MagicMock
ecs_client.task_definitions = {}
ecs_client.task_definitions[TASK_ARN] = TaskDefinition(
name=TASK_NAME,
arn=TASK_ARN,
revision=TASK_REVISION,
region=AWS_REGION_US_EAST_1,
network_mode="host",
container_definitions=[
ContainerDefinition(
name=CONTAINER_NAME,
privileged=True,
user="root",
environment=[],
log_driver="awslogs",
)
],
)
ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1)
with mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
), mock.patch(
"prowler.providers.aws.services.ecs.ecs_client.ecs_client",
ecs_client,
task_arn = ecs_client.register_task_definition(
family=TASK_NAME,
containerDefinitions=[
{
"name": CONTAINER_NAME,
"image": "ubuntu",
"memory": 128,
"readonlyRootFilesystem": True,
"privileged": True,
"user": "root",
"environment": [],
"logConfiguration": {"logDriver": "awslogs"},
}
],
)["taskDefinition"]["taskDefinitionArn"]
from prowler.providers.aws.services.ecs.ecs_service import ECS
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_task_definitions_logging_enabled.ecs_task_definitions_logging_enabled.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_task_definitions_logging_enabled.ecs_task_definitions_logging_enabled import (
ecs_task_definitions_logging_enabled,
@@ -152,3 +121,7 @@ class Test_ecs_task_definitions_logging_enabled:
result[0].status_extended
== f"ECS task definition {TASK_NAME} with revision {TASK_REVISION} containers have logging configured."
)
assert result[0].resource_id == f"{TASK_NAME}:{TASK_REVISION}"
assert result[0].resource_arn == task_arn
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_tags == []
@@ -1,30 +1,31 @@
from unittest import mock
from unittest.mock import patch
from prowler.providers.aws.services.ecs.ecs_service import (
ContainerDefinition,
ContainerEnvVariable,
TaskDefinition,
)
from boto3 import client
from moto import mock_aws
AWS_REGION = "eu-west-1"
AWS_ACCOUNT_NUMBER = "123456789012"
task_name = "test-task"
task_revision = "1"
task_arn = f"arn:aws:ecs:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:task-definition/{task_name}:{task_revision}"
env_var_name_no_secrets = "host"
env_var_value_no_secrets = "localhost:1234"
env_var_name_with_secrets = "DB_PASSWORD"
env_var_value_with_secrets = "pass-12343"
from tests.providers.aws.utils import AWS_REGION_US_EAST_1, set_mocked_aws_provider
TASK_NAME = "test-task"
TASK_REVISION = "1"
CONTAINER_NAME = "test-container"
ENV_VAR_NAME_NO_SECRETS = "host"
ENV_VAR_VALUE_NO_SECRETS = "localhost:1234"
ENV_VAR_NAME_WITH_SECRETS = "DB_PASSWORD"
ENV_VAR_VALUE_WITH_SECRETS = "pass-12343"
class Test_ecs_task_definitions_no_environment_secrets:
def test_no_task_definitions(self):
ecs_client = mock.MagicMock
ecs_client.task_definitions = {}
from prowler.providers.aws.services.ecs.ecs_service import ECS
with mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_task_definitions_no_environment_secrets.ecs_task_definitions_no_environment_secrets.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_task_definitions_no_environment_secrets.ecs_task_definitions_no_environment_secrets import (
ecs_task_definitions_no_environment_secrets,
@@ -34,31 +35,40 @@ class Test_ecs_task_definitions_no_environment_secrets:
result = check.execute()
assert len(result) == 0
@mock_aws
def test_container_env_var_no_secrets(self):
ecs_client = mock.MagicMock
ecs_client.task_definitions = {}
ecs_client.task_definitions[task_arn] = TaskDefinition(
name=task_name,
arn=f"arn:aws:ecs:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:task-definition/{task_name}:{task_revision}",
revision="1",
region=AWS_REGION,
container_definitions=[
ContainerDefinition(
name="container1",
privileged=False,
user="",
environment=[
ContainerEnvVariable(
name=env_var_name_no_secrets, value=env_var_value_no_secrets
)
],
)
],
)
ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1)
with mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
task_arn = ecs_client.register_task_definition(
family=TASK_NAME,
containerDefinitions=[
{
"name": CONTAINER_NAME,
"image": "ubuntu",
"memory": 128,
"readonlyRootFilesystem": True,
"privileged": False,
"user": "appuser",
"environment": [
{
"name": ENV_VAR_NAME_NO_SECRETS,
"value": ENV_VAR_VALUE_NO_SECRETS,
}
],
}
],
)["taskDefinition"]["taskDefinitionArn"]
from prowler.providers.aws.services.ecs.ecs_service import ECS
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_task_definitions_no_environment_secrets.ecs_task_definitions_no_environment_secrets.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_task_definitions_no_environment_secrets.ecs_task_definitions_no_environment_secrets import (
ecs_task_definitions_no_environment_secrets,
@@ -70,40 +80,47 @@ class Test_ecs_task_definitions_no_environment_secrets:
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"No secrets found in variables of ECS task definition {task_name} with revision {task_revision}."
)
assert result[0].resource_id == f"{task_name}:1"
assert (
result[0].resource_arn
== f"arn:aws:ecs:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:task-definition/{task_name}:{task_revision}"
== f"No secrets found in variables of ECS task definition {TASK_NAME} with revision {TASK_REVISION}."
)
assert result[0].resource_id == f"{TASK_NAME}:{TASK_REVISION}"
assert result[0].resource_arn == task_arn
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_tags == []
@mock_aws
def test_container_env_var_with_secrets(self):
ecs_client = mock.MagicMock
ecs_client.task_definitions = {}
ecs_client.task_definitions[task_arn] = TaskDefinition(
name=task_name,
arn=f"arn:aws:ecs:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:task-definition/{task_name}:{task_revision}",
revision="1",
region=AWS_REGION,
container_definitions=[
ContainerDefinition(
name="container1",
privileged=False,
user="",
environment=[
ContainerEnvVariable(
name=env_var_name_with_secrets,
value=env_var_value_with_secrets,
)
],
)
],
)
ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1)
with mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
task_arn = ecs_client.register_task_definition(
family=TASK_NAME,
containerDefinitions=[
{
"name": CONTAINER_NAME,
"image": "ubuntu",
"memory": 128,
"readonlyRootFilesystem": True,
"privileged": False,
"user": "appuser",
"environment": [
{
"name": ENV_VAR_NAME_WITH_SECRETS,
"value": ENV_VAR_VALUE_WITH_SECRETS,
}
],
}
],
)["taskDefinition"]["taskDefinitionArn"]
from prowler.providers.aws.services.ecs.ecs_service import ECS
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_task_definitions_no_environment_secrets.ecs_task_definitions_no_environment_secrets.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_task_definitions_no_environment_secrets.ecs_task_definitions_no_environment_secrets import (
ecs_task_definitions_no_environment_secrets,
@@ -115,10 +132,9 @@ class Test_ecs_task_definitions_no_environment_secrets:
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Potential secrets found in ECS task definition {task_name} with revision {task_revision}: Secrets in container container1 -> Secret Keyword on line 2."
)
assert result[0].resource_id == f"{task_name}:1"
assert (
result[0].resource_arn
== f"arn:aws:ecs:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:task-definition/{task_name}:{task_revision}"
== f"Potential secrets found in ECS task definition {TASK_NAME} with revision {TASK_REVISION}: Secrets in container test-container -> Secret Keyword on line 2."
)
assert result[0].resource_id == f"{TASK_NAME}:{TASK_REVISION}"
assert result[0].resource_arn == task_arn
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_tags == []
@@ -1,26 +1,27 @@
from unittest import mock
from unittest.mock import patch
from prowler.providers.aws.services.ecs.ecs_service import (
ContainerDefinition,
ContainerEnvVariable,
TaskDefinition,
)
from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_US_EAST_1
from boto3 import client
from moto import mock_aws
from tests.providers.aws.utils import AWS_REGION_US_EAST_1, set_mocked_aws_provider
TASK_NAME = "test-task"
TASK_REVISION = "1"
CONTAINER_NAME = "test-container"
TASK_ARN = f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:task-definition/{TASK_NAME}:{TASK_REVISION}"
class Test_ecs_task_definitions_no_privileged_containers:
def test_no_task_definitions(self):
ecs_client = mock.MagicMock
ecs_client.task_definitions = {}
from prowler.providers.aws.services.ecs.ecs_service import ECS
with mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_task_definitions_no_privileged_containers.ecs_task_definitions_no_privileged_containers.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_task_definitions_no_privileged_containers.ecs_task_definitions_no_privileged_containers import (
ecs_task_definitions_no_privileged_containers,
@@ -30,33 +31,35 @@ class Test_ecs_task_definitions_no_privileged_containers:
result = check.execute()
assert len(result) == 0
@mock_aws
def test_task_definition_no_priviled_container(self):
ecs_client = mock.MagicMock
ecs_client.task_definitions = {}
ecs_client.task_definitions[TASK_ARN] = TaskDefinition(
name=TASK_NAME,
arn=TASK_ARN,
revision=TASK_REVISION,
region=AWS_REGION_US_EAST_1,
network_mode="bridge",
container_definitions=[
ContainerDefinition(
name=CONTAINER_NAME,
privileged=False,
user="",
environment=[
ContainerEnvVariable(
name="env_var_name_no_secrets",
value="env_var_value_no_secrets",
)
],
)
],
)
ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1)
with mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
task_arn = ecs_client.register_task_definition(
family=TASK_NAME,
containerDefinitions=[
{
"name": CONTAINER_NAME,
"image": "ubuntu",
"memory": 128,
"readonlyRootFilesystem": True,
"privileged": False,
"user": "appuser",
"environment": [],
}
],
)["taskDefinition"]["taskDefinitionArn"]
from prowler.providers.aws.services.ecs.ecs_service import ECS
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_task_definitions_no_privileged_containers.ecs_task_definitions_no_privileged_containers.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_task_definitions_no_privileged_containers.ecs_task_definitions_no_privileged_containers import (
ecs_task_definitions_no_privileged_containers,
@@ -70,29 +73,40 @@ class Test_ecs_task_definitions_no_privileged_containers:
result[0].status_extended
== f"ECS task definition {TASK_NAME} with revision {TASK_REVISION} does not have privileged containers."
)
assert result[0].resource_id == f"{TASK_NAME}:{TASK_REVISION}"
assert result[0].resource_arn == task_arn
assert result[0].resource_tags == []
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_task_definition_privileged_container(self):
ecs_client = mock.MagicMock
ecs_client.task_definitions = {}
ecs_client.task_definitions[TASK_ARN] = TaskDefinition(
name=TASK_NAME,
arn=TASK_ARN,
revision=TASK_REVISION,
region=AWS_REGION_US_EAST_1,
network_mode="host",
container_definitions=[
ContainerDefinition(
name=CONTAINER_NAME,
privileged=True,
user="root",
environment=[],
)
],
)
ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1)
with mock.patch(
"prowler.providers.aws.services.ecs.ecs_service.ECS",
ecs_client,
task_arn = ecs_client.register_task_definition(
family=TASK_NAME,
containerDefinitions=[
{
"name": CONTAINER_NAME,
"image": "ubuntu",
"memory": 128,
"readonlyRootFilesystem": True,
"privileged": True,
"user": "root",
"environment": [],
}
],
)["taskDefinition"]["taskDefinitionArn"]
from prowler.providers.aws.services.ecs.ecs_service import ECS
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_task_definitions_no_privileged_containers.ecs_task_definitions_no_privileged_containers.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_task_definitions_no_privileged_containers.ecs_task_definitions_no_privileged_containers import (
ecs_task_definitions_no_privileged_containers,
@@ -106,3 +120,7 @@ class Test_ecs_task_definitions_no_privileged_containers:
result[0].status_extended
== f"ECS task definition {TASK_NAME} with revision {TASK_REVISION} has privileged containers: {CONTAINER_NAME}"
)
assert result[0].resource_id == f"{TASK_NAME}:{TASK_REVISION}"
assert result[0].resource_arn == task_arn
assert result[0].resource_tags == []
assert result[0].region == AWS_REGION_US_EAST_1
@@ -1,46 +1,79 @@
from unittest import mock
from uuid import uuid4
from unittest.mock import patch
from prowler.providers.aws.services.guardduty.guardduty_service import Detector
from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_EU_WEST_1
import botocore
from boto3 import client
from moto import mock_aws
AWS_ACCOUNT_NUMBER_ADMIN = "123456789013"
DETECTOR_ID = str(uuid4())
DETECTOR_ARN = f"arn:aws:guardduty:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:detector/{DETECTOR_ID}"
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_EU_WEST_1,
set_mocked_aws_provider,
)
orig = botocore.client.BaseClient._make_api_call
def mock_make_api_call_admin_enabled(self, operation_name, api_params):
if operation_name == "GetAdministratorAccount":
return {
"Administrator": {
"AccountId": "210987654321",
}
}
return orig(self, operation_name, api_params)
def mock_make_api_call_members_managers(self, operation_name, api_params):
if operation_name == "ListMembers":
return {
"Members": [
{
"AccountId": "210987654321",
"RelationshipStatus": "Enabled",
}
]
}
return orig(self, operation_name, api_params)
class Test_guardduty_centrally_managed:
@mock_aws
def test_no_detectors(self):
guardduty_client = mock.MagicMock
guardduty_client.detectors = []
with mock.patch(
"prowler.providers.aws.services.guardduty.guardduty_service.GuardDuty",
guardduty_client,
aws_provider = set_mocked_aws_provider()
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), patch(
"prowler.providers.aws.services.guardduty.guardduty_centrally_managed.guardduty_centrally_managed.guardduty_client",
new=GuardDuty(aws_provider),
):
from prowler.providers.aws.services.guardduty.guardduty_no_high_severity_findings.guardduty_no_high_severity_findings import (
guardduty_no_high_severity_findings,
from prowler.providers.aws.services.guardduty.guardduty_centrally_managed.guardduty_centrally_managed import (
guardduty_centrally_managed,
)
check = guardduty_no_high_severity_findings()
check = guardduty_centrally_managed()
result = check.execute()
assert len(result) == 0
@mock_aws
def test_detector_no_centralized_managed(self):
guardduty_client = mock.MagicMock
guardduty_client.detectors = []
guardduty_client.detectors.append(
Detector(
id=DETECTOR_ID,
region=AWS_REGION_EU_WEST_1,
arn=DETECTOR_ARN,
status=False,
findings=[str(uuid4())],
)
)
guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
with mock.patch(
"prowler.providers.aws.services.guardduty.guardduty_service.GuardDuty",
guardduty_client,
detector_id = guardduty_client.create_detector(Enable=True)["DetectorId"]
aws_provider = set_mocked_aws_provider()
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), patch(
"prowler.providers.aws.services.guardduty.guardduty_centrally_managed.guardduty_centrally_managed.guardduty_client",
new=GuardDuty(aws_provider),
):
# Test Check
from prowler.providers.aws.services.guardduty.guardduty_centrally_managed.guardduty_centrally_managed import (
@@ -53,54 +86,36 @@ class Test_guardduty_centrally_managed:
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"GuardDuty detector {DETECTOR_ID} is not centrally managed."
== f"GuardDuty detector {detector_id} is not centrally managed."
)
assert result[0].resource_id == DETECTOR_ID
assert result[0].resource_id == detector_id
assert result[0].region == AWS_REGION_EU_WEST_1
assert result[0].resource_arn == DETECTOR_ARN
def test_not_enabled_account_detector(self):
guardduty_client = mock.MagicMock
guardduty_client.detectors = []
guardduty_client.detectors.append(
Detector(
id=AWS_ACCOUNT_NUMBER,
region=AWS_REGION_EU_WEST_1,
arn=DETECTOR_ARN,
enabled_in_account=False,
assert (
result[0].resource_arn
== f"arn:aws:guardduty:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:detector/{detector_id}"
)
)
with mock.patch(
"prowler.providers.aws.services.guardduty.guardduty_service.GuardDuty",
guardduty_client,
):
# Test Check
from prowler.providers.aws.services.guardduty.guardduty_centrally_managed.guardduty_centrally_managed import (
guardduty_centrally_managed,
)
check = guardduty_centrally_managed()
result = check.execute()
assert len(result) == 0
assert result[0].resource_tags == []
@patch(
"botocore.client.BaseClient._make_api_call",
new=mock_make_api_call_admin_enabled,
)
@mock_aws
def test_detector_centralized_managed(self):
guardduty_client = mock.MagicMock
guardduty_client.detectors = []
guardduty_client.detectors.append(
Detector(
id=DETECTOR_ID,
region=AWS_REGION_EU_WEST_1,
arn=DETECTOR_ARN,
status=False,
findings=[str(uuid4())],
administrator_account=AWS_ACCOUNT_NUMBER_ADMIN,
)
)
guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
with mock.patch(
"prowler.providers.aws.services.guardduty.guardduty_service.GuardDuty",
guardduty_client,
detector_id = guardduty_client.create_detector(Enable=True)["DetectorId"]
aws_provider = set_mocked_aws_provider()
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), patch(
"prowler.providers.aws.services.guardduty.guardduty_centrally_managed.guardduty_centrally_managed.guardduty_client",
new=GuardDuty(aws_provider),
):
# Test Check
from prowler.providers.aws.services.guardduty.guardduty_centrally_managed.guardduty_centrally_managed import (
@@ -113,29 +128,35 @@ class Test_guardduty_centrally_managed:
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"GuardDuty detector {DETECTOR_ID} is centrally managed by account {AWS_ACCOUNT_NUMBER_ADMIN}."
== f"GuardDuty detector {detector_id} is centrally managed by account 210987654321."
)
assert result[0].resource_id == DETECTOR_ID
assert result[0].resource_id == detector_id
assert result[0].region == AWS_REGION_EU_WEST_1
assert result[0].resource_arn == DETECTOR_ARN
def test_detector_administrator(self):
guardduty_client = mock.MagicMock
guardduty_client.detectors = []
guardduty_client.detectors.append(
Detector(
id=DETECTOR_ID,
region=AWS_REGION_EU_WEST_1,
arn=DETECTOR_ARN,
status=False,
findings=[str(uuid4())],
member_accounts=[AWS_ACCOUNT_NUMBER_ADMIN],
assert (
result[0].resource_arn
== f"arn:aws:guardduty:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:detector/{detector_id}"
)
)
with mock.patch(
"prowler.providers.aws.services.guardduty.guardduty_service.GuardDuty",
guardduty_client,
@patch(
"botocore.client.BaseClient._make_api_call",
new=mock_make_api_call_members_managers,
)
@mock_aws
def test_detector_members_accounts(self):
guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
detector_id = guardduty_client.create_detector(Enable=True)["DetectorId"]
aws_provider = set_mocked_aws_provider()
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), patch(
"prowler.providers.aws.services.guardduty.guardduty_centrally_managed.guardduty_centrally_managed.guardduty_client",
new=GuardDuty(aws_provider),
):
# Test Check
from prowler.providers.aws.services.guardduty.guardduty_centrally_managed.guardduty_centrally_managed import (
@@ -148,8 +169,11 @@ class Test_guardduty_centrally_managed:
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"GuardDuty detector {DETECTOR_ID} is administrator account with 1 member accounts."
== f"GuardDuty detector {detector_id} is administrator account with 1 member accounts."
)
assert result[0].resource_id == DETECTOR_ID
assert result[0].resource_id == detector_id
assert result[0].region == AWS_REGION_EU_WEST_1
assert result[0].resource_arn == DETECTOR_ARN
assert (
result[0].resource_arn
== f"arn:aws:guardduty:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:detector/{detector_id}"
)
@@ -1,171 +1,199 @@
from unittest import mock
from uuid import uuid4
from unittest.mock import patch
from boto3 import client
from moto import mock_aws
from prowler.providers.aws.services.guardduty.guardduty_service import Detector
from tests.providers.aws.utils import (
AWS_ACCOUNT_ARN,
AWS_ACCOUNT_NUMBER,
AWS_REGION_EU_WEST_1,
set_mocked_aws_provider,
)
DETECTOR_ID = str(uuid4())
DETECTOR_ARN = f"arn:aws:guardduty:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:detector/{DETECTOR_ID}"
class Test_guardduty_is_enabled:
@mock_aws
def test_no_detectors(self):
guardduty_client = mock.MagicMock
guardduty_client.region = AWS_REGION_EU_WEST_1
guardduty_client.detectors = []
guardduty_client.detectors.append(
Detector(
id=AWS_ACCOUNT_NUMBER,
region=AWS_REGION_EU_WEST_1,
arn=AWS_ACCOUNT_ARN,
enabled_in_account=False,
)
)
guardduty_client.audited_account_arn = AWS_ACCOUNT_ARN
with mock.patch(
"prowler.providers.aws.services.guardduty.guardduty_service.GuardDuty",
guardduty_client,
):
aws_provider = set_mocked_aws_provider()
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), patch(
"prowler.providers.aws.services.guardduty.guardduty_is_enabled.guardduty_is_enabled.guardduty_client",
new=GuardDuty(aws_provider),
) as guardduty_client:
from prowler.providers.aws.services.guardduty.guardduty_is_enabled.guardduty_is_enabled import (
guardduty_is_enabled,
)
check = guardduty_is_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert result[0].status_extended == "GuardDuty is not enabled."
assert result[0].resource_id == AWS_ACCOUNT_NUMBER
assert result[0].resource_arn == AWS_ACCOUNT_ARN
assert result[0].region == AWS_REGION_EU_WEST_1
guardduty_client.detectors = []
check = guardduty_is_enabled()
results = check.execute()
assert len(results) == 0
@mock_aws
def test_guardduty_enabled(self):
guardduty_client = mock.MagicMock
guardduty_client.detectors = []
guardduty_client.detectors.append(
Detector(
id=DETECTOR_ID,
region=AWS_REGION_EU_WEST_1,
arn=DETECTOR_ARN,
status=True,
)
)
with mock.patch(
"prowler.providers.aws.services.guardduty.guardduty_service.GuardDuty",
guardduty_client,
guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
detector_id = guardduty_client.create_detector(Enable=True)["DetectorId"]
aws_provider = set_mocked_aws_provider()
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), patch(
"prowler.providers.aws.services.guardduty.guardduty_is_enabled.guardduty_is_enabled.guardduty_client",
new=GuardDuty(aws_provider),
):
from prowler.providers.aws.services.guardduty.guardduty_is_enabled.guardduty_is_enabled import (
guardduty_is_enabled,
)
check = guardduty_is_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"GuardDuty detector {DETECTOR_ID} enabled."
)
assert result[0].resource_id == DETECTOR_ID
assert result[0].resource_arn == DETECTOR_ARN
assert result[0].region == AWS_REGION_EU_WEST_1
results = check.execute()
assert len(results) == 29
for result in results:
if result.region == AWS_REGION_EU_WEST_1:
assert result.status == "PASS"
assert (
result.status_extended
== f"GuardDuty detector {result.resource_id} enabled."
)
assert result.resource_id == detector_id
assert (
result.resource_arn
== f"arn:aws:guardduty:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:detector/{detector_id}"
)
assert result.resource_tags == []
@mock_aws
def test_guardduty_configured_but_suspended(self):
guardduty_client = mock.MagicMock
guardduty_client.region = AWS_REGION_EU_WEST_1
guardduty_client.detectors = []
guardduty_client.detectors.append(
Detector(
id=DETECTOR_ID,
arn=DETECTOR_ARN,
region=AWS_REGION_EU_WEST_1,
status=False,
)
)
with mock.patch(
"prowler.providers.aws.services.guardduty.guardduty_service.GuardDuty",
guardduty_client,
):
guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
detector_id = guardduty_client.create_detector(Enable=False)["DetectorId"]
aws_provider = set_mocked_aws_provider()
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), patch(
"prowler.providers.aws.services.guardduty.guardduty_is_enabled.guardduty_is_enabled.guardduty_client",
new=GuardDuty(aws_provider),
) as mock_guardduty_client:
from prowler.providers.aws.services.guardduty.guardduty_is_enabled.guardduty_is_enabled import (
guardduty_is_enabled,
)
check = guardduty_is_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"GuardDuty detector {DETECTOR_ID} configured but suspended."
)
assert result[0].resource_id == DETECTOR_ID
assert result[0].resource_arn == DETECTOR_ARN
assert result[0].region == AWS_REGION_EU_WEST_1
for detector in mock_guardduty_client.detectors:
if detector.region == AWS_REGION_EU_WEST_1:
detector.status = False
check = guardduty_is_enabled()
results = check.execute()
assert len(results) == 29
for result in results:
if result.region == AWS_REGION_EU_WEST_1:
assert result.status == "FAIL"
assert (
result.status_extended
== f"GuardDuty detector {result.resource_id} configured but suspended."
)
assert result.resource_id == detector_id
assert (
result.resource_arn
== f"arn:aws:guardduty:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:detector/{detector_id}"
)
assert result.resource_tags == []
@mock_aws
def test_guardduty_not_configured(self):
guardduty_client = mock.MagicMock
guardduty_client.detectors = []
guardduty_client.region = AWS_REGION_EU_WEST_1
guardduty_client.detectors.append(
Detector(
id=DETECTOR_ID,
arn=DETECTOR_ARN,
region=AWS_REGION_EU_WEST_1,
)
)
with mock.patch(
"prowler.providers.aws.services.guardduty.guardduty_service.GuardDuty",
guardduty_client,
):
guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
detector_id = guardduty_client.create_detector(Enable=False)["DetectorId"]
aws_provider = set_mocked_aws_provider()
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), patch(
"prowler.providers.aws.services.guardduty.guardduty_is_enabled.guardduty_is_enabled.guardduty_client",
new=GuardDuty(aws_provider),
) as mock_guardduty_client:
from prowler.providers.aws.services.guardduty.guardduty_is_enabled.guardduty_is_enabled import (
guardduty_is_enabled,
)
check = guardduty_is_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"GuardDuty detector {DETECTOR_ID} not configured."
)
assert result[0].resource_id == DETECTOR_ID
assert result[0].resource_arn == DETECTOR_ARN
assert result[0].region == AWS_REGION_EU_WEST_1
for detector in mock_guardduty_client.detectors:
if detector.region == AWS_REGION_EU_WEST_1:
detector.status = None
check = guardduty_is_enabled()
results = check.execute()
assert len(results) == 29
for result in results:
if result.region == AWS_REGION_EU_WEST_1:
assert result.status == "FAIL"
assert (
result.status_extended
== f"GuardDuty detector {result.resource_id} not configured."
)
assert result.resource_id == detector_id
assert (
result.resource_arn
== f"arn:aws:guardduty:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:detector/{detector_id}"
)
assert result.resource_tags == []
@mock_aws
def test_guardduty_not_configured_muted(self):
guardduty_client = mock.MagicMock
guardduty_client.audit_config = {"mute_non_default_regions": True}
guardduty_client.region = "eu-south-2"
guardduty_client.detectors = []
guardduty_client.detectors.append(
Detector(
id=DETECTOR_ID,
arn=DETECTOR_ARN,
region=AWS_REGION_EU_WEST_1,
)
)
with mock.patch(
"prowler.providers.aws.services.guardduty.guardduty_service.GuardDuty",
guardduty_client,
):
guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
detector_id = guardduty_client.create_detector(Enable=False)["DetectorId"]
aws_provider = set_mocked_aws_provider()
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), patch(
"prowler.providers.aws.services.guardduty.guardduty_is_enabled.guardduty_is_enabled.guardduty_client",
new=GuardDuty(aws_provider),
) as mock_guardduty_client:
from prowler.providers.aws.services.guardduty.guardduty_is_enabled.guardduty_is_enabled import (
guardduty_is_enabled,
)
mock_guardduty_client.audit_config = {"mute_non_default_regions": True}
check = guardduty_is_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert result[0].muted
assert (
result[0].status_extended
== f"GuardDuty detector {DETECTOR_ID} not configured."
)
assert result[0].resource_id == DETECTOR_ID
assert result[0].resource_arn == DETECTOR_ARN
assert result[0].region == AWS_REGION_EU_WEST_1
results = check.execute()
assert len(results) == 29
for result in results:
if result.region == AWS_REGION_EU_WEST_1:
assert result.status == "FAIL"
assert result.muted
assert (
result.status_extended
== f"GuardDuty detector {result.resource_id} not configured."
)
assert result.resource_id == detector_id
assert (
result.resource_arn
== f"arn:aws:guardduty:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:detector/{detector_id}"
)
assert result.resource_tags == []
assert result.muted
@@ -0,0 +1,148 @@
from unittest.mock import patch
import botocore
from boto3 import client
from moto import mock_aws
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_EU_WEST_1,
set_mocked_aws_provider,
)
orig = botocore.client.BaseClient._make_api_call
class Test_guardduty_lambda_protection_enabled:
def test_no_detectors(self):
aws_provider = set_mocked_aws_provider()
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), patch(
"prowler.providers.aws.services.guardduty.guardduty_lambda_protection_enabled.guardduty_lambda_protection_enabled.guardduty_client",
new=GuardDuty(aws_provider),
):
# Test Check
from prowler.providers.aws.services.guardduty.guardduty_lambda_protection_enabled.guardduty_lambda_protection_enabled import (
guardduty_lambda_protection_enabled,
)
check = guardduty_lambda_protection_enabled()
result = check.execute()
assert len(result) == 0
@mock_aws
def test_detector_disabled(self):
guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
guardduty_client.create_detector(Enable=False)
aws_provider = set_mocked_aws_provider()
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), patch(
"prowler.providers.aws.services.guardduty.guardduty_lambda_protection_enabled.guardduty_lambda_protection_enabled.guardduty_client",
new=GuardDuty(aws_provider),
):
# Test Check
from prowler.providers.aws.services.guardduty.guardduty_lambda_protection_enabled.guardduty_lambda_protection_enabled import (
guardduty_lambda_protection_enabled,
)
check = guardduty_lambda_protection_enabled()
result = check.execute()
assert len(result) == 0
@mock_aws
def test_detector_lambda_protection_enabled(self):
guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
detector_id = guardduty_client.create_detector(
Enable=True,
Features=[{"Name": "LAMBDA_NETWORK_LOGS", "Status": "ENABLED"}],
)["DetectorId"]
aws_provider = set_mocked_aws_provider()
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), patch(
"prowler.providers.aws.services.guardduty.guardduty_lambda_protection_enabled.guardduty_lambda_protection_enabled.guardduty_client",
new=GuardDuty(aws_provider),
):
# Test Check
from prowler.providers.aws.services.guardduty.guardduty_lambda_protection_enabled.guardduty_lambda_protection_enabled import (
guardduty_lambda_protection_enabled,
)
check = guardduty_lambda_protection_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"GuardDuty detector {detector_id} has Lambda Protection enabled."
)
assert result[0].resource_id == detector_id
assert result[0].region == AWS_REGION_EU_WEST_1
assert (
result[0].resource_arn
== f"arn:aws:guardduty:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:detector/{detector_id}"
)
assert result[0].resource_tags == []
@mock_aws
def test_detector_lambda_protection_disabled(self):
guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
detector_id = guardduty_client.create_detector(
Enable=True,
Features=[{"Name": "LAMBDA_NETWORK_LOGS", "Status": "DISABLED"}],
)["DetectorId"]
aws_provider = set_mocked_aws_provider()
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), patch(
"prowler.providers.aws.services.guardduty.guardduty_lambda_protection_enabled.guardduty_lambda_protection_enabled.guardduty_client",
new=GuardDuty(aws_provider),
):
# Test Check
from prowler.providers.aws.services.guardduty.guardduty_lambda_protection_enabled.guardduty_lambda_protection_enabled import (
guardduty_lambda_protection_enabled,
)
check = guardduty_lambda_protection_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"GuardDuty detector {detector_id} does not have Lambda Protection enabled."
)
assert result[0].resource_id == detector_id
assert result[0].region == AWS_REGION_EU_WEST_1
assert (
result[0].resource_arn
== f"arn:aws:guardduty:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:detector/{detector_id}"
)
assert result[0].resource_tags == []
@@ -1,21 +1,43 @@
from re import search
from unittest import mock
from uuid import uuid4
from unittest.mock import patch
from prowler.providers.aws.services.guardduty.guardduty_service import Detector
from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_EU_WEST_1
import botocore
from boto3 import client
from moto import mock_aws
DETECTOR_ID = str(uuid4())
DETECTOR_ARN = f"arn:aws:guardduty:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:detector/{DETECTOR_ID}"
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_EU_WEST_1,
set_mocked_aws_provider,
)
orig = botocore.client.BaseClient._make_api_call
def mock_make_api_call(self, operation_name, kwarg):
if operation_name == "ListFindings":
return {
"FindingIds": [
"f1",
"f2",
]
}
# If we don't want to patch the API call
return orig(self, operation_name, kwarg)
class Test_guardduty_no_high_severity_findings:
@mock_aws
def test_no_detectors(self):
guardduty_client = mock.MagicMock
guardduty_client.detectors = []
with mock.patch(
"prowler.providers.aws.services.guardduty.guardduty_service.GuardDuty",
guardduty_client,
aws_provider = set_mocked_aws_provider()
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), patch(
"prowler.providers.aws.services.guardduty.guardduty_no_high_severity_findings.guardduty_no_high_severity_findings.guardduty_client",
new=GuardDuty(aws_provider),
):
from prowler.providers.aws.services.guardduty.guardduty_no_high_severity_findings.guardduty_no_high_severity_findings import (
guardduty_no_high_severity_findings,
@@ -25,19 +47,22 @@ class Test_guardduty_no_high_severity_findings:
result = check.execute()
assert len(result) == 0
@mock_aws
def test_no_high_findings(self):
guardduty_client = mock.MagicMock
guardduty_client.detectors = []
guardduty_client.detectors.append(
Detector(
id=DETECTOR_ID,
arn=DETECTOR_ARN,
region=AWS_REGION_EU_WEST_1,
)
)
with mock.patch(
"prowler.providers.aws.services.guardduty.guardduty_service.GuardDuty",
guardduty_client,
guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
detector_id = guardduty_client.create_detector(Enable=True)["DetectorId"]
aws_provider = set_mocked_aws_provider()
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), patch(
"prowler.providers.aws.services.guardduty.guardduty_no_high_severity_findings.guardduty_no_high_severity_findings.guardduty_client",
new=GuardDuty(aws_provider),
):
from prowler.providers.aws.services.guardduty.guardduty_no_high_severity_findings.guardduty_no_high_severity_findings import (
guardduty_no_high_severity_findings,
@@ -47,51 +72,35 @@ class Test_guardduty_no_high_severity_findings:
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert search(
"does not have high severity findings.", result[0].status_extended
assert (
result[0].status_extended
== f"GuardDuty detector {detector_id} does not have high severity findings."
)
assert result[0].resource_id == DETECTOR_ID
assert result[0].resource_arn == DETECTOR_ARN
assert result[0].resource_id == detector_id
assert result[0].region == AWS_REGION_EU_WEST_1
def test_not_enabled_account_detector(self):
guardduty_client = mock.MagicMock
guardduty_client.detectors = []
guardduty_client.detectors.append(
Detector(
id=AWS_ACCOUNT_NUMBER,
arn=DETECTOR_ARN,
region=AWS_REGION_EU_WEST_1,
enabled_in_account=False,
)
)
with mock.patch(
"prowler.providers.aws.services.guardduty.guardduty_service.GuardDuty",
guardduty_client,
):
from prowler.providers.aws.services.guardduty.guardduty_no_high_severity_findings.guardduty_no_high_severity_findings import (
guardduty_no_high_severity_findings,
assert (
result[0].resource_arn
== f"arn:aws:guardduty:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:detector/{detector_id}"
)
assert result[0].resource_tags == []
check = guardduty_no_high_severity_findings()
result = check.execute()
assert len(result) == 0
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
@mock_aws
def test_high_findings(self):
guardduty_client = mock.MagicMock
guardduty_client.detectors = []
guardduty_client.detectors.append(
Detector(
id=DETECTOR_ID,
region=AWS_REGION_EU_WEST_1,
arn=DETECTOR_ARN,
status=False,
findings=[str(uuid4())],
)
)
with mock.patch(
"prowler.providers.aws.services.guardduty.guardduty_service.GuardDuty",
guardduty_client,
guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
detector_id = guardduty_client.create_detector(Enable=True)["DetectorId"]
aws_provider = set_mocked_aws_provider()
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), patch(
"prowler.providers.aws.services.guardduty.guardduty_no_high_severity_findings.guardduty_no_high_severity_findings.guardduty_client",
new=GuardDuty(aws_provider),
):
from prowler.providers.aws.services.guardduty.guardduty_no_high_severity_findings.guardduty_no_high_severity_findings import (
guardduty_no_high_severity_findings,
@@ -101,7 +110,14 @@ class Test_guardduty_no_high_severity_findings:
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert search("has 1 high severity findings", result[0].status_extended)
assert result[0].resource_id == DETECTOR_ID
assert result[0].resource_arn == DETECTOR_ARN
assert (
result[0].status_extended
== f"GuardDuty detector {detector_id} has 2 high severity findings."
)
assert result[0].resource_id == detector_id
assert result[0].region == AWS_REGION_EU_WEST_1
assert (
result[0].resource_arn
== f"arn:aws:guardduty:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:detector/{detector_id}"
)
assert result[0].resource_tags == []
@@ -115,6 +115,7 @@ class Test_GuardDuty_Service:
"S3Logs": {"Enable": True},
"Kubernetes": {"AuditLogs": {"Enable": True}},
},
Features=[{"Name": "LAMBDA_NETWORK_LOGS", "Status": "ENABLED"}],
)
aws_provider = set_mocked_aws_provider()
@@ -133,6 +134,7 @@ class Test_GuardDuty_Service:
assert guardduty.detectors[0].s3_protection
assert not guardduty.detectors[0].rds_protection
assert guardduty.detectors[0].eks_audit_log_protection
assert guardduty.detectors[0].lambda_protection
assert not guardduty.detectors[0].ec2_malware_protection
assert guardduty.detectors[0].region == AWS_REGION_EU_WEST_1
assert guardduty.detectors[0].tags == [{"test": "test"}]
@@ -0,0 +1,160 @@
from unittest import mock
from uuid import uuid4
import botocore
from boto3 import client
from moto import mock_aws
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_EU_WEST_1,
set_mocked_aws_provider,
)
CLUSTER_ID = str(uuid4())
CLUSTER_ARN = (
f"arn:aws:redshift:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:cluster:{CLUSTER_ID}"
)
make_api_call = botocore.client.BaseClient._make_api_call
def mock_make_api_call(self, operation_name, kwarg):
if operation_name == "DescribeClusterParameters":
return {
"Parameters": [
{
"ParameterName": "require_ssl",
"ParameterValue": "true",
"Description": "Require SSL for connections",
"Source": "user",
"DataType": "boolean",
"AllowedValues": "true, false",
"IsModifiable": True,
"MinimumEngineVersion": "1.0",
},
]
}
return make_api_call(self, operation_name, kwarg)
class Test_redshift_cluster_in_transit_encryption_enabled:
def test_no_clusters(self):
from prowler.providers.aws.services.redshift.redshift_service import Redshift
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.redshift.redshift_cluster_in_transit_encryption_enabled.redshift_cluster_in_transit_encryption_enabled.redshift_client",
new=Redshift(aws_provider),
):
from prowler.providers.aws.services.redshift.redshift_cluster_in_transit_encryption_enabled.redshift_cluster_in_transit_encryption_enabled import (
redshift_cluster_in_transit_encryption_enabled,
)
check = redshift_cluster_in_transit_encryption_enabled()
result = check.execute()
assert len(result) == 0
@mock_aws
def test_cluster_not_encrypted_in_transit(self):
redshift_client = client("redshift", region_name=AWS_REGION_EU_WEST_1)
redshift_client.create_cluster(
DBName="test",
ClusterIdentifier=CLUSTER_ID,
ClusterType="single-node",
NodeType="ds2.xlarge",
MasterUsername="awsuser",
MasterUserPassword="password",
PubliclyAccessible=True,
Tags=[
{"Key": "test", "Value": "test"},
],
Port=9439,
Encrypted=False,
)
from prowler.providers.aws.services.redshift.redshift_service import Redshift
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.redshift.redshift_cluster_in_transit_encryption_enabled.redshift_cluster_in_transit_encryption_enabled.redshift_client",
new=Redshift(aws_provider),
):
from prowler.providers.aws.services.redshift.redshift_cluster_in_transit_encryption_enabled.redshift_cluster_in_transit_encryption_enabled import (
redshift_cluster_in_transit_encryption_enabled,
)
check = redshift_cluster_in_transit_encryption_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert result[0].status_extended == (
f"Redshift Cluster {CLUSTER_ID} is not encrypted in transit."
)
assert result[0].resource_id == CLUSTER_ID
assert result[0].resource_arn == CLUSTER_ARN
assert result[0].region == AWS_REGION_EU_WEST_1
assert result[0].resource_tags == [{"Key": "test", "Value": "test"}]
@mock_aws
def test_cluster_encrypted_in_transit(self):
with mock.patch(
"botocore.client.BaseClient._make_api_call", new=mock_make_api_call
):
redshift_client = client("redshift", region_name=AWS_REGION_EU_WEST_1)
redshift_client.create_cluster(
DBName="test",
ClusterIdentifier=CLUSTER_ID,
ClusterType="single-node",
NodeType="ds2.xlarge",
MasterUsername="user",
MasterUserPassword="password",
PubliclyAccessible=True,
Tags=[
{"Key": "test", "Value": "test"},
],
Port=9439,
Encrypted=True,
)
from prowler.providers.aws.services.redshift.redshift_service import (
Redshift,
)
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.redshift.redshift_cluster_in_transit_encryption_enabled.redshift_cluster_in_transit_encryption_enabled.redshift_client",
new=Redshift(aws_provider),
):
from prowler.providers.aws.services.redshift.redshift_cluster_in_transit_encryption_enabled.redshift_cluster_in_transit_encryption_enabled import (
redshift_cluster_in_transit_encryption_enabled,
)
check = redshift_cluster_in_transit_encryption_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].status_extended == (
f"Redshift Cluster {CLUSTER_ID} is encrypted in transit."
)
assert result[0].resource_id == CLUSTER_ID
assert result[0].resource_arn == CLUSTER_ARN
assert result[0].region == AWS_REGION_EU_WEST_1
assert result[0].resource_tags == [{"Key": "test", "Value": "test"}]
@@ -43,6 +43,22 @@ def mock_make_api_call(self, operation_name, kwarg):
},
]
}
if operation_name == "DescribeClusterParameters":
return {
"Parameters": [
{
"ParameterName": "require_ssl",
"ParameterValue": "true",
"Description": "Require SSL for connections",
"Source": "user",
"DataType": "boolean",
"AllowedValues": "true, false",
"IsModifiable": True,
"MinimumEngineVersion": "1.0",
},
]
}
return make_api_call(self, operation_name, kwarg)
@@ -94,6 +110,7 @@ class Test_Redshift_Service:
Tags=[
{"Key": "test", "Value": "test"},
],
ClusterParameterGroupName="default.redshift-1.0",
)
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
redshift = Redshift(aws_provider)
@@ -113,6 +130,7 @@ class Test_Redshift_Service:
assert redshift.clusters[0].tags == [
{"Key": "test", "Value": "test"},
]
assert redshift.clusters[0].parameter_group_name == "default.redshift-1.0"
assert redshift.clusters[0].encrypted
assert redshift.clusters[0].master_username == "user"
assert redshift.clusters[0].database_name == "test"
@@ -177,3 +195,39 @@ class Test_Redshift_Service:
assert redshift.clusters[0].logging_enabled
assert redshift.clusters[0].bucket == test_bucket_name
assert redshift.clusters[0].cluster_snapshots
@mock_aws
def test_describe_cluster_parameter_groups(self):
redshift_client = client("redshift", region_name=AWS_REGION_EU_WEST_1)
response = redshift_client.create_cluster(
DBName="test",
ClusterIdentifier=cluster_id,
ClusterType="single-node",
NodeType="ds2.xlarge",
MasterUsername="user",
MasterUserPassword="password",
PubliclyAccessible=True,
Tags=[
{"Key": "test", "Value": "test"},
],
)
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
redshift = Redshift(aws_provider)
assert len(redshift.clusters) == 1
assert redshift.clusters[0].id == cluster_id
assert redshift.clusters[0].region == AWS_REGION_EU_WEST_1
assert redshift.clusters[0].public_access
assert (
redshift.clusters[0].endpoint_address
== response["Cluster"]["Endpoint"]["Address"]
)
assert (
redshift.clusters[0].allow_version_upgrade
== response["Cluster"]["AllowVersionUpgrade"]
)
assert redshift.clusters[0].tags == [
{"Key": "test", "Value": "test"},
]
assert redshift.clusters[0].parameter_group_name == "default.redshift-1.0"
assert redshift.clusters[0].require_ssl is True