feat(cloudfront): add cloudfront_distributions_origin_traffic_encrypted check to ensure traffic encryption to custom origins (#4958)

Co-authored-by: Sergio <sergio@prowler.com>
This commit is contained in:
Hugo Pereira Brito
2024-09-16 15:12:37 +02:00
committed by GitHub
parent f54b64f1f8
commit 7e16702b2f
5 changed files with 303 additions and 1 deletions
@@ -0,0 +1,30 @@
{
"Provider": "aws",
"CheckID": "cloudfront_distributions_origin_traffic_encrypted",
"CheckTitle": "Check if CloudFront distributions encrypt traffic to custom origins.",
"CheckType": [],
"ServiceName": "cloudfront",
"SubServiceName": "",
"ResourceIdTemplate": "arn:partition:cloudfront:region:account-id:distribution/resource-id",
"Severity": "medium",
"ResourceType": "AWSCloudFrontDistribution",
"Description": "Check if CloudFront distributions encrypt traffic to custom origins.",
"Risk": "Allowing unencrypted HTTP traffic between CloudFront and custom origins can expose data to potential eavesdropping and manipulation, compromising data security and integrity.",
"RelatedUrl": "https://docs.aws.amazon.com/whitepapers/latest/secure-content-delivery-amazon-cloudfront/custom-origin-with-cloudfront.html",
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/cloudfront-controls.html#cloudfront-9",
"Terraform": "https://www.trendmicro.com/cloudoneconformity/knowledge-base/aws/CloudFront/cloudfront-traffic-to-origin-unencrypted.html"
},
"Recommendation": {
"Text": "Configure your CloudFront distributions to require HTTPS (TLS) for traffic to custom origins, ensuring all data transmitted between CloudFront and the origin is encrypted and protected from unauthorized access.",
"Url": "https://docs.aws.amazon.com/AmazonCloudFront/latest/DeveloperGuide/using-https-cloudfront-to-custom-origin.html"
}
},
"Categories": [],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,36 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.cloudfront.cloudfront_client import (
cloudfront_client,
)
class cloudfront_distributions_origin_traffic_encrypted(Check):
def execute(self):
findings = []
for distribution in cloudfront_client.distributions.values():
report = Check_Report_AWS(self.metadata())
report.region = distribution.region
report.resource_arn = distribution.arn
report.resource_id = distribution.id
report.resource_tags = distribution.tags
report.status = "PASS"
report.status_extended = f"CloudFront Distribution {distribution.id} does encrypt traffic to custom origins."
unencrypted_origins = []
for origin in distribution.origins:
if (
origin.origin_protocol_policy == ""
or origin.origin_protocol_policy == "http-only"
) or (
origin.origin_protocol_policy == "match-viewer"
and distribution.viewer_protocol_policy == "allow-all"
):
unencrypted_origins.append(origin.id)
if unencrypted_origins:
report.status = "FAIL"
report.status_extended = f"CloudFront Distribution {distribution.id} does not encrypt traffic to custom origins {', '.join(unencrypted_origins)}."
findings.append(report)
return findings
@@ -84,6 +84,7 @@ class CloudFront(AWSService):
try:
for distribution_id in distributions.keys():
distribution_config = client.get_distribution_config(Id=distribution_id)
# Global Config
distributions[distribution_id].logging_enabled = distribution_config[
"DistributionConfig"
@@ -99,7 +100,14 @@ class CloudFront(AWSService):
"DistributionConfig"
]["WebACLId"]
distributions[distribution_id].default_root_object = (
distribution_config["DistributionConfig"].get("DefaultRootObject")
distribution_config["DistributionConfig"].get(
"DefaultRootObject", ""
)
)
distributions[distribution_id].viewer_protocol_policy = (
distribution_config["DistributionConfig"][
"DefaultCacheBehavior"
].get("ViewerProtocolPolicy", "")
)
# Default Cache Config
@@ -198,6 +206,7 @@ class Distribution(BaseModel):
web_acl_id: str = ""
default_certificate: Optional[bool]
default_root_object: Optional[str]
viewer_protocol_policy: Optional[str]
tags: Optional[list] = []
ssl_support_method: Optional[SSLSupportMethod]
certificate: Optional[str]
@@ -0,0 +1,227 @@
from unittest import mock
from prowler.providers.aws.services.cloudfront.cloudfront_service import (
DefaultCacheConfigBehaviour,
Distribution,
Origin,
ViewerProtocolPolicy,
)
from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER
DISTRIBUTION_ID = "E27LVI50CSW06W"
DISTRIBUTION_ARN = (
f"arn:aws:cloudfront::{AWS_ACCOUNT_NUMBER}:distribution/{DISTRIBUTION_ID}"
)
REGION = "eu-west-1"
class Test_cloudfront_distributions_origin_traffic_encrypted:
def test_no_distributions(self):
cloudfront_client = mock.MagicMock
cloudfront_client.distributions = {}
with mock.patch(
"prowler.providers.aws.services.cloudfront.cloudfront_service.CloudFront",
new=cloudfront_client,
):
# Test Check
from prowler.providers.aws.services.cloudfront.cloudfront_distributions_origin_traffic_encrypted.cloudfront_distributions_origin_traffic_encrypted import (
cloudfront_distributions_origin_traffic_encrypted,
)
check = cloudfront_distributions_origin_traffic_encrypted()
result = check.execute()
assert len(result) == 0
def test_distribution_no_traffic_encryption(self):
cloudfront_client = mock.MagicMock
id = "origin1"
cloudfront_client.distributions = {
"DISTRIBUTION_ID": Distribution(
arn=DISTRIBUTION_ARN,
id=DISTRIBUTION_ID,
region=REGION,
origins=[
Origin(
id=id,
domain_name="asdf.s3.us-east-1.amazonaws.com",
origin_protocol_policy="",
origin_ssl_protocols=[],
)
],
default_cache_config=DefaultCacheConfigBehaviour(
realtime_log_config_arn="",
viewer_protocol_policy=ViewerProtocolPolicy.allow_all,
field_level_encryption_id="",
),
default_root_object="",
viewer_protocol_policy="",
)
}
with mock.patch(
"prowler.providers.aws.services.cloudfront.cloudfront_service.CloudFront",
new=cloudfront_client,
):
# Test Check
from prowler.providers.aws.services.cloudfront.cloudfront_distributions_origin_traffic_encrypted.cloudfront_distributions_origin_traffic_encrypted import (
cloudfront_distributions_origin_traffic_encrypted,
)
check = cloudfront_distributions_origin_traffic_encrypted()
result = check.execute()
assert len(result) == 1
assert result[0].region == REGION
assert result[0].resource_arn == DISTRIBUTION_ARN
assert result[0].resource_id == DISTRIBUTION_ID
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"CloudFront Distribution {DISTRIBUTION_ID} does not encrypt traffic to custom origins {id}."
)
assert result[0].resource_tags == []
def test_distribution_http_only(self):
cloudfront_client = mock.MagicMock
id = "origin1"
cloudfront_client.distributions = {
"DISTRIBUTION_ID": Distribution(
arn=DISTRIBUTION_ARN,
id=DISTRIBUTION_ID,
region=REGION,
origins=[
Origin(
id=id,
domain_name="asdf.s3.us-east-1.amazonaws.com",
origin_protocol_policy="http-only",
origin_ssl_protocols=[],
)
],
default_cache_config=DefaultCacheConfigBehaviour(
realtime_log_config_arn="",
viewer_protocol_policy=ViewerProtocolPolicy.allow_all,
field_level_encryption_id="",
),
default_root_object="index.html",
)
}
with mock.patch(
"prowler.providers.aws.services.cloudfront.cloudfront_service.CloudFront",
new=cloudfront_client,
):
# Test Check
from prowler.providers.aws.services.cloudfront.cloudfront_distributions_origin_traffic_encrypted.cloudfront_distributions_origin_traffic_encrypted import (
cloudfront_distributions_origin_traffic_encrypted,
)
check = cloudfront_distributions_origin_traffic_encrypted()
result = check.execute()
assert len(result) == 1
assert result[0].region == REGION
assert result[0].resource_arn == DISTRIBUTION_ARN
assert result[0].resource_id == DISTRIBUTION_ID
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"CloudFront Distribution {DISTRIBUTION_ID} does not encrypt traffic to custom origins {id}."
)
assert result[0].resource_tags == []
def test_distribution_match_viewer_allow_all(self):
cloudfront_client = mock.MagicMock
id = "origin1"
cloudfront_client.distributions = {
"DISTRIBUTION_ID": Distribution(
arn=DISTRIBUTION_ARN,
id=DISTRIBUTION_ID,
region=REGION,
origins=[
Origin(
id=id,
domain_name="asdf.s3.us-east-1.amazonaws.com",
origin_protocol_policy="match-viewer",
origin_ssl_protocols=[],
)
],
default_cache_config=DefaultCacheConfigBehaviour(
realtime_log_config_arn="",
viewer_protocol_policy=ViewerProtocolPolicy.allow_all,
field_level_encryption_id="",
),
default_root_object="index.html",
viewer_protocol_policy="allow-all",
)
}
with mock.patch(
"prowler.providers.aws.services.cloudfront.cloudfront_service.CloudFront",
new=cloudfront_client,
):
# Test Check
from prowler.providers.aws.services.cloudfront.cloudfront_distributions_origin_traffic_encrypted.cloudfront_distributions_origin_traffic_encrypted import (
cloudfront_distributions_origin_traffic_encrypted,
)
check = cloudfront_distributions_origin_traffic_encrypted()
result = check.execute()
assert len(result) == 1
assert result[0].region == REGION
assert result[0].resource_arn == DISTRIBUTION_ARN
assert result[0].resource_id == DISTRIBUTION_ID
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"CloudFront Distribution {DISTRIBUTION_ID} does not encrypt traffic to custom origins {id}."
)
assert result[0].resource_tags == []
def test_distribution_traffic_encrypted(self):
cloudfront_client = mock.MagicMock
cloudfront_client.distributions = {
"DISTRIBUTION_ID": Distribution(
arn=DISTRIBUTION_ARN,
id=DISTRIBUTION_ID,
region=REGION,
origins=[
Origin(
id="origin1",
domain_name="asdf.s3.us-east-1.amazonaws.com",
origin_protocol_policy="https-only",
origin_ssl_protocols=[],
)
],
default_cache_config=DefaultCacheConfigBehaviour(
realtime_log_config_arn="",
viewer_protocol_policy=ViewerProtocolPolicy.allow_all,
field_level_encryption_id="",
),
default_root_object="index.html",
)
}
with mock.patch(
"prowler.providers.aws.services.cloudfront.cloudfront_service.CloudFront",
new=cloudfront_client,
):
# Test Check
from prowler.providers.aws.services.cloudfront.cloudfront_distributions_origin_traffic_encrypted.cloudfront_distributions_origin_traffic_encrypted import (
cloudfront_distributions_origin_traffic_encrypted,
)
check = cloudfront_distributions_origin_traffic_encrypted()
result = check.execute()
assert len(result) == 1
assert result[0].region == REGION
assert result[0].resource_arn == DISTRIBUTION_ARN
assert result[0].resource_id == DISTRIBUTION_ID
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"CloudFront Distribution {DISTRIBUTION_ID} does encrypt traffic to custom origins."
)
assert result[0].resource_tags == []