feat(cloudfront): Add new cloudfront_distributions_s3_origin_access_control check to ensure OAC is configured in distributions (#4939)

Co-authored-by: Sergio <sergio@prowler.com>
This commit is contained in:
Hugo Pereira Brito
2024-09-13 15:51:49 +01:00
committed by GitHub
parent 917a2ad0fe
commit 48c31a1616
6 changed files with 232 additions and 2 deletions
@@ -0,0 +1,32 @@
{
"Provider": "aws",
"CheckID": "cloudfront_distributions_s3_origin_access_control",
"CheckTitle": "Check if CloudFront distributions with S3 origin use OAC.",
"CheckType": [
"Data Exposure"
],
"ServiceName": "cloudfront",
"SubServiceName": "",
"ResourceIdTemplate": "arn:partition:cloudfront:region:account-id:distribution/resource-id",
"Severity": "medium",
"ResourceType": "AWSCloudFrontDistribution",
"Description": "Check if CloudFront distributions use origin access control.",
"Risk": "Without OAC, your S3 bucket could be accessed directly, bypassing CloudFront, which could expose your content to unauthorized access. Additionally, relying on Origin Access Identity (OAI) may limit functionality and security features, making your distribution less secure and more difficult to manage.",
"RelatedUrl": "https://docs.aws.amazon.com/AmazonCloudFront/latest/DeveloperGuide/private-content-restricting-access-to-s3.html#migrate-from-oai-to-oac",
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "https://docs.prowler.com/checks/aws/iam-policies/ensure-aws-cloudfromt-distribution-with-s3-have-origin-access-set-to-enabled/",
"Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/cloudfront-controls.html#cloudfront-13",
"Terraform": "https://www.trendmicro.com/cloudoneconformity/knowledge-base/aws/CloudFront/s3-origin.html"
},
"Recommendation": {
"Text": "Configure Origin Access Control (OAC) for CloudFront distributions that use an Amazon S3 origin. This will ensure that the content in your S3 bucket is accessible only through the specified CloudFront distribution, enhancing security by preventing direct access to the bucket.",
"Url": "https://docs.aws.amazon.com/AmazonCloudFront/latest/DeveloperGuide/private-content-restricting-access-to-s3.html"
}
},
"Categories": [],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,35 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.cloudfront.cloudfront_client import (
cloudfront_client,
)
class cloudfront_distributions_s3_origin_access_control(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
if any(origin.s3_origin_config for origin in distribution.origins):
s3_buckets_with_no_oac = []
report.status = "PASS"
report.status_extended = f"CloudFront Distribution {distribution.id} is using origin access control (OAC) for S3 origins."
for origin in distribution.origins:
if (
origin.s3_origin_config != {}
and origin.origin_access_control == ""
):
s3_buckets_with_no_oac.append(origin.id)
if s3_buckets_with_no_oac:
report.status = "FAIL"
report.status_extended = f"CloudFront Distribution {distribution.id} is not using origin access control (OAC) in S3 origins {', '.join(s3_buckets_with_no_oac)}."
findings.append(report)
return findings
@@ -55,6 +55,12 @@ class CloudFront(AWSService):
)
.get("OriginSslProtocols", {})
.get("Items", []),
origin_access_control=origin.get(
"OriginAccessControlId", ""
),
s3_origin_config=origin.get(
"S3OriginConfig", {}
),
)
)
distribution = Distribution(
@@ -175,6 +181,8 @@ class Origin(BaseModel):
domain_name: str
origin_protocol_policy: str
origin_ssl_protocols: list[str]
origin_access_control: Optional[str]
s3_origin_config: Optional[dict]
class Distribution(BaseModel):
@@ -0,0 +1,155 @@
from unittest import mock
from prowler.providers.aws.services.cloudfront.cloudfront_service import (
Distribution,
Origin,
)
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_s3_origin_access_control:
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_s3_origin_access_control.cloudfront_distributions_s3_origin_access_control import (
cloudfront_distributions_s3_origin_access_control,
)
check = cloudfront_distributions_s3_origin_access_control()
result = check.execute()
assert len(result) == 0
def test_no_s3_origin_distributions(self):
cloudfront_client = mock.MagicMock
cloudfront_client.distributions = {
"DISTRIBUTION_ID": Distribution(
arn=DISTRIBUTION_ARN,
id=DISTRIBUTION_ID,
region=REGION,
origins=[
Origin(
id="string",
domain_name="string",
origin_protocol_policy="https-only",
origin_ssl_protocols=["TLSv1", "TLSv1.1"],
s3_origin_config={},
),
],
)
}
with mock.patch(
"prowler.providers.aws.services.cloudfront.cloudfront_service.CloudFront",
new=cloudfront_client,
):
# Test Check
from prowler.providers.aws.services.cloudfront.cloudfront_distributions_s3_origin_access_control.cloudfront_distributions_s3_origin_access_control import (
cloudfront_distributions_s3_origin_access_control,
)
check = cloudfront_distributions_s3_origin_access_control()
result = check.execute()
assert len(result) == 0
def test_distribution_using_origin_access_control(self):
cloudfront_client = mock.MagicMock
cloudfront_client.distributions = {
"DISTRIBUTION_ID": Distribution(
arn=DISTRIBUTION_ARN,
id=DISTRIBUTION_ID,
region=REGION,
origins=[
Origin(
id="string",
domain_name="string",
origin_protocol_policy="https-only",
origin_ssl_protocols=["TLSv1", "TLSv1.1"],
origin_access_control="EXAMPLE-OAC-ID",
s3_origin_config={
"OriginAccessIdentity": "origin-access-identity/cloudfront/EXAMPLE-OAI-ID"
},
),
],
)
}
with mock.patch(
"prowler.providers.aws.services.cloudfront.cloudfront_service.CloudFront",
new=cloudfront_client,
):
# Test Check
from prowler.providers.aws.services.cloudfront.cloudfront_distributions_s3_origin_access_control.cloudfront_distributions_s3_origin_access_control import (
cloudfront_distributions_s3_origin_access_control,
)
check = cloudfront_distributions_s3_origin_access_control()
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} is using origin access control (OAC) for S3 origins."
)
assert result[0].resource_tags == []
def test_distribution_not_using_origin_access_control(self):
cloudfront_client = mock.MagicMock
id = "EXAMPLE-OAC-ID"
cloudfront_client.distributions = {
"DISTRIBUTION_ID": Distribution(
arn=DISTRIBUTION_ARN,
id=DISTRIBUTION_ID,
region=REGION,
origins=[
Origin(
id=id,
domain_name="string",
origin_protocol_policy="https-only",
origin_ssl_protocols=["TLSv1", "TLSv1.1"],
origin_access_control="",
s3_origin_config={
"OriginAccessIdentity": "origin-access-identity/cloudfront/EXAMPLE-OAI-ID"
},
),
],
)
}
with mock.patch(
"prowler.providers.aws.services.cloudfront.cloudfront_service.CloudFront",
new=cloudfront_client,
):
# Test Check
from prowler.providers.aws.services.cloudfront.cloudfront_distributions_s3_origin_access_control.cloudfront_distributions_s3_origin_access_control import (
cloudfront_distributions_s3_origin_access_control,
)
check = cloudfront_distributions_s3_origin_access_control()
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} is not using origin access control (OAC) in S3 origins {id}."
)
assert result[0].resource_tags == []
@@ -27,7 +27,7 @@ def example_distribution_config(ref):
"Id": "origin1",
"DomainName": "asdf.s3.us-east-1.amazonaws.com",
"S3OriginConfig": {"OriginAccessIdentity": ""},
}
},
],
},
"DefaultCacheBehavior": {
@@ -179,7 +179,7 @@ class Test_CloudFront_Service:
assert len(cloudfront.distributions) == 0
def test__list_distributions__complete(self):
def test_list_distributionscomplete(self):
from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER
DISTRIBUTION_ID = "E27LVI50CSW06W"