mirror of
https://github.com/ghndrx/prowler.git
synced 2026-02-10 14:55:00 +00:00
feat(s3): Add S3 KMS encryption check (#2757)
Co-authored-by: Pepe Fagoaga <pepe@verica.io>
This commit is contained in:
@@ -0,0 +1,34 @@
|
|||||||
|
{
|
||||||
|
"Provider": "aws",
|
||||||
|
"CheckID": "s3_bucket_kms_encryption",
|
||||||
|
"CheckTitle": "Check if S3 buckets have KMS encryption enabled.",
|
||||||
|
"CheckType": [
|
||||||
|
"Data Protection"
|
||||||
|
],
|
||||||
|
"ServiceName": "s3",
|
||||||
|
"SubServiceName": "",
|
||||||
|
"ResourceIdTemplate": "arn:partition:s3:::bucket_name",
|
||||||
|
"Severity": "medium",
|
||||||
|
"ResourceType": "AwsS3Bucket",
|
||||||
|
"Description": "Check if S3 buckets have KMS encryption enabled.",
|
||||||
|
"Risk": "Amazon S3 KMS encryption provides a way to set the encryption behavior for an S3 bucket using a managed key. This will ensure data-at-rest is encrypted.",
|
||||||
|
"RelatedUrl": "https://docs.aws.amazon.com/AmazonS3/latest/userguide/UsingKMSEncryption.html",
|
||||||
|
"Remediation": {
|
||||||
|
"Code": {
|
||||||
|
"CLI": "https://www.trendmicro.com/cloudoneconformity-staging/knowledge-base/aws/S3/encrypted-with-kms-customer-master-keys.html",
|
||||||
|
"NativeIaC": "https://www.trendmicro.com/cloudoneconformity-staging/knowledge-base/aws/S3/encrypted-with-kms-customer-master-keys.html",
|
||||||
|
"Other": "",
|
||||||
|
"Terraform": "https://docs.bridgecrew.io/docs/ensure-that-s3-buckets-are-encrypted-with-kms-by-default#terraform"
|
||||||
|
},
|
||||||
|
"Recommendation": {
|
||||||
|
"Text": "Ensure that S3 buckets have encryption at rest enabled using KMS.",
|
||||||
|
"Url": "https://docs.aws.amazon.com/AmazonS3/latest/userguide/UsingKMSEncryption.html"
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"Categories": [
|
||||||
|
"encryption"
|
||||||
|
],
|
||||||
|
"DependsOn": [],
|
||||||
|
"RelatedTo": [],
|
||||||
|
"Notes": ""
|
||||||
|
}
|
||||||
@@ -0,0 +1,22 @@
|
|||||||
|
from prowler.lib.check.models import Check, Check_Report_AWS
|
||||||
|
from prowler.providers.aws.services.s3.s3_client import s3_client
|
||||||
|
|
||||||
|
|
||||||
|
class s3_bucket_kms_encryption(Check):
|
||||||
|
def execute(self):
|
||||||
|
findings = []
|
||||||
|
for bucket in s3_client.buckets:
|
||||||
|
report = Check_Report_AWS(self.metadata())
|
||||||
|
report.region = bucket.region
|
||||||
|
report.resource_id = bucket.name
|
||||||
|
report.resource_arn = bucket.arn
|
||||||
|
report.resource_tags = bucket.tags
|
||||||
|
|
||||||
|
if bucket.encryption == "aws:kms" or bucket.encryption == "aws:kms:dsse":
|
||||||
|
report.status = "PASS"
|
||||||
|
report.status_extended = f"S3 Bucket {bucket.name} has Server Side Encryption with {bucket.encryption}."
|
||||||
|
else:
|
||||||
|
report.status = "FAIL"
|
||||||
|
report.status_extended = f"Server Side Encryption is not configured with kms for S3 Bucket {bucket.name}."
|
||||||
|
findings.append(report)
|
||||||
|
return findings
|
||||||
@@ -0,0 +1,275 @@
|
|||||||
|
from unittest import mock
|
||||||
|
|
||||||
|
from boto3 import client, session
|
||||||
|
from moto import mock_s3
|
||||||
|
|
||||||
|
from prowler.providers.aws.lib.audit_info.models import AWS_Audit_Info
|
||||||
|
from prowler.providers.common.models import Audit_Metadata
|
||||||
|
|
||||||
|
AWS_ACCOUNT_NUMBER = "123456789012"
|
||||||
|
AWS_REGION = "us-east-1"
|
||||||
|
AWS_ACCOUNT_ARN = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root"
|
||||||
|
|
||||||
|
|
||||||
|
class Test_s3_bucket_kms_encryption:
|
||||||
|
# Mocked Audit Info
|
||||||
|
def set_mocked_audit_info(self):
|
||||||
|
audit_info = AWS_Audit_Info(
|
||||||
|
session_config=None,
|
||||||
|
original_session=None,
|
||||||
|
audit_session=session.Session(
|
||||||
|
profile_name=None,
|
||||||
|
botocore_session=None,
|
||||||
|
region_name=AWS_REGION,
|
||||||
|
),
|
||||||
|
audited_account=AWS_ACCOUNT_NUMBER,
|
||||||
|
audited_account_arn=AWS_ACCOUNT_ARN,
|
||||||
|
audited_user_id=None,
|
||||||
|
audited_partition="aws",
|
||||||
|
audited_identity_arn=None,
|
||||||
|
profile=None,
|
||||||
|
profile_region=AWS_REGION,
|
||||||
|
credentials=None,
|
||||||
|
assumed_role_info=None,
|
||||||
|
audited_regions=None,
|
||||||
|
organizations_metadata=None,
|
||||||
|
audit_resources=None,
|
||||||
|
mfa_enabled=False,
|
||||||
|
audit_metadata=Audit_Metadata(
|
||||||
|
services_scanned=0,
|
||||||
|
expected_checks=[],
|
||||||
|
completed_checks=0,
|
||||||
|
audit_progress=0,
|
||||||
|
),
|
||||||
|
)
|
||||||
|
return audit_info
|
||||||
|
|
||||||
|
@mock_s3
|
||||||
|
def test_no_buckets(self):
|
||||||
|
from prowler.providers.aws.services.s3.s3_service import S3
|
||||||
|
|
||||||
|
audit_info = self.set_mocked_audit_info()
|
||||||
|
|
||||||
|
with mock.patch(
|
||||||
|
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
|
||||||
|
new=audit_info,
|
||||||
|
), mock.patch(
|
||||||
|
"prowler.providers.aws.services.s3.s3_bucket_kms_encryption.s3_bucket_kms_encryption.s3_client",
|
||||||
|
new=S3(audit_info),
|
||||||
|
):
|
||||||
|
# Test Check
|
||||||
|
from prowler.providers.aws.services.s3.s3_bucket_kms_encryption.s3_bucket_kms_encryption import (
|
||||||
|
s3_bucket_kms_encryption,
|
||||||
|
)
|
||||||
|
|
||||||
|
check = s3_bucket_kms_encryption()
|
||||||
|
result = check.execute()
|
||||||
|
|
||||||
|
assert len(result) == 0
|
||||||
|
|
||||||
|
@mock_s3
|
||||||
|
def test_bucket_no_encryption(self):
|
||||||
|
s3_client_us_east_1 = client("s3", region_name=AWS_REGION)
|
||||||
|
bucket_name_us = "bucket_test_us"
|
||||||
|
s3_client_us_east_1.create_bucket(Bucket=bucket_name_us)
|
||||||
|
|
||||||
|
from prowler.providers.aws.services.s3.s3_service import S3
|
||||||
|
|
||||||
|
audit_info = self.set_mocked_audit_info()
|
||||||
|
|
||||||
|
with mock.patch(
|
||||||
|
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
|
||||||
|
new=audit_info,
|
||||||
|
), mock.patch(
|
||||||
|
"prowler.providers.aws.services.s3.s3_bucket_kms_encryption.s3_bucket_kms_encryption.s3_client",
|
||||||
|
new=S3(audit_info),
|
||||||
|
):
|
||||||
|
# Test Check
|
||||||
|
from prowler.providers.aws.services.s3.s3_bucket_kms_encryption.s3_bucket_kms_encryption import (
|
||||||
|
s3_bucket_kms_encryption,
|
||||||
|
)
|
||||||
|
|
||||||
|
check = s3_bucket_kms_encryption()
|
||||||
|
result = check.execute()
|
||||||
|
|
||||||
|
assert len(result) == 1
|
||||||
|
assert result[0].status == "FAIL"
|
||||||
|
assert (
|
||||||
|
result[0].status_extended
|
||||||
|
== f"Server Side Encryption is not configured with kms for S3 Bucket {bucket_name_us}."
|
||||||
|
)
|
||||||
|
assert result[0].resource_id == bucket_name_us
|
||||||
|
assert (
|
||||||
|
result[0].resource_arn
|
||||||
|
== f"arn:{audit_info.audited_partition}:s3:::{bucket_name_us}"
|
||||||
|
)
|
||||||
|
assert result[0].resource_tags == []
|
||||||
|
assert result[0].region == AWS_REGION
|
||||||
|
|
||||||
|
@mock_s3
|
||||||
|
def test_bucket_no_kms_encryption(self):
|
||||||
|
s3_client_us_east_1 = client("s3", region_name=AWS_REGION)
|
||||||
|
bucket_name_us = "bucket_test_us"
|
||||||
|
s3_client_us_east_1.create_bucket(
|
||||||
|
Bucket=bucket_name_us, ObjectOwnership="BucketOwnerEnforced"
|
||||||
|
)
|
||||||
|
sse_config = {
|
||||||
|
"Rules": [
|
||||||
|
{
|
||||||
|
"ApplyServerSideEncryptionByDefault": {
|
||||||
|
"SSEAlgorithm": "AES256",
|
||||||
|
}
|
||||||
|
}
|
||||||
|
]
|
||||||
|
}
|
||||||
|
|
||||||
|
s3_client_us_east_1.put_bucket_encryption(
|
||||||
|
Bucket=bucket_name_us, ServerSideEncryptionConfiguration=sse_config
|
||||||
|
)
|
||||||
|
|
||||||
|
from prowler.providers.aws.services.s3.s3_service import S3
|
||||||
|
|
||||||
|
audit_info = self.set_mocked_audit_info()
|
||||||
|
|
||||||
|
with mock.patch(
|
||||||
|
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
|
||||||
|
new=audit_info,
|
||||||
|
), mock.patch(
|
||||||
|
"prowler.providers.aws.services.s3.s3_bucket_kms_encryption.s3_bucket_kms_encryption.s3_client",
|
||||||
|
new=S3(audit_info),
|
||||||
|
):
|
||||||
|
# Test Check
|
||||||
|
from prowler.providers.aws.services.s3.s3_bucket_kms_encryption.s3_bucket_kms_encryption import (
|
||||||
|
s3_bucket_kms_encryption,
|
||||||
|
)
|
||||||
|
|
||||||
|
check = s3_bucket_kms_encryption()
|
||||||
|
result = check.execute()
|
||||||
|
|
||||||
|
assert len(result) == 1
|
||||||
|
assert result[0].status == "FAIL"
|
||||||
|
assert (
|
||||||
|
result[0].status_extended
|
||||||
|
== f"Server Side Encryption is not configured with kms for S3 Bucket {bucket_name_us}."
|
||||||
|
)
|
||||||
|
assert result[0].resource_id == bucket_name_us
|
||||||
|
assert (
|
||||||
|
result[0].resource_arn
|
||||||
|
== f"arn:{audit_info.audited_partition}:s3:::{bucket_name_us}"
|
||||||
|
)
|
||||||
|
assert result[0].resource_tags == []
|
||||||
|
assert result[0].region == AWS_REGION
|
||||||
|
|
||||||
|
@mock_s3
|
||||||
|
def test_bucket_kms_encryption(self):
|
||||||
|
s3_client_us_east_1 = client("s3", region_name=AWS_REGION)
|
||||||
|
bucket_name_us = "bucket_test_us"
|
||||||
|
s3_client_us_east_1.create_bucket(
|
||||||
|
Bucket=bucket_name_us, ObjectOwnership="BucketOwnerEnforced"
|
||||||
|
)
|
||||||
|
kms_encryption = "aws:kms"
|
||||||
|
sse_config = {
|
||||||
|
"Rules": [
|
||||||
|
{
|
||||||
|
"ApplyServerSideEncryptionByDefault": {
|
||||||
|
"SSEAlgorithm": f"{kms_encryption}",
|
||||||
|
"KMSMasterKeyID": "12345678",
|
||||||
|
}
|
||||||
|
}
|
||||||
|
]
|
||||||
|
}
|
||||||
|
|
||||||
|
s3_client_us_east_1.put_bucket_encryption(
|
||||||
|
Bucket=bucket_name_us, ServerSideEncryptionConfiguration=sse_config
|
||||||
|
)
|
||||||
|
|
||||||
|
from prowler.providers.aws.services.s3.s3_service import S3
|
||||||
|
|
||||||
|
audit_info = self.set_mocked_audit_info()
|
||||||
|
|
||||||
|
with mock.patch(
|
||||||
|
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
|
||||||
|
new=audit_info,
|
||||||
|
):
|
||||||
|
with mock.patch(
|
||||||
|
"prowler.providers.aws.services.s3.s3_bucket_kms_encryption.s3_bucket_kms_encryption.s3_client",
|
||||||
|
new=S3(audit_info),
|
||||||
|
):
|
||||||
|
# Test Check
|
||||||
|
from prowler.providers.aws.services.s3.s3_bucket_kms_encryption.s3_bucket_kms_encryption import (
|
||||||
|
s3_bucket_kms_encryption,
|
||||||
|
)
|
||||||
|
|
||||||
|
check = s3_bucket_kms_encryption()
|
||||||
|
result = check.execute()
|
||||||
|
|
||||||
|
assert len(result) == 1
|
||||||
|
assert result[0].status == "PASS"
|
||||||
|
assert (
|
||||||
|
result[0].status_extended
|
||||||
|
== f"S3 Bucket {bucket_name_us} has Server Side Encryption with {kms_encryption}."
|
||||||
|
)
|
||||||
|
assert result[0].resource_id == bucket_name_us
|
||||||
|
assert (
|
||||||
|
result[0].resource_arn
|
||||||
|
== f"arn:{audit_info.audited_partition}:s3:::{bucket_name_us}"
|
||||||
|
)
|
||||||
|
assert result[0].resource_tags == []
|
||||||
|
assert result[0].region == AWS_REGION
|
||||||
|
|
||||||
|
@mock_s3
|
||||||
|
def test_bucket_kms_dsse_encryption(self):
|
||||||
|
s3_client_us_east_1 = client("s3", region_name=AWS_REGION)
|
||||||
|
bucket_name_us = "bucket_test_us"
|
||||||
|
s3_client_us_east_1.create_bucket(
|
||||||
|
Bucket=bucket_name_us, ObjectOwnership="BucketOwnerEnforced"
|
||||||
|
)
|
||||||
|
kms_encryption = "aws:kms:dsse"
|
||||||
|
sse_config = {
|
||||||
|
"Rules": [
|
||||||
|
{
|
||||||
|
"ApplyServerSideEncryptionByDefault": {
|
||||||
|
"SSEAlgorithm": f"{kms_encryption}",
|
||||||
|
"KMSMasterKeyID": "12345678",
|
||||||
|
}
|
||||||
|
}
|
||||||
|
]
|
||||||
|
}
|
||||||
|
|
||||||
|
s3_client_us_east_1.put_bucket_encryption(
|
||||||
|
Bucket=bucket_name_us, ServerSideEncryptionConfiguration=sse_config
|
||||||
|
)
|
||||||
|
|
||||||
|
from prowler.providers.aws.services.s3.s3_service import S3
|
||||||
|
|
||||||
|
audit_info = self.set_mocked_audit_info()
|
||||||
|
|
||||||
|
with mock.patch(
|
||||||
|
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
|
||||||
|
new=audit_info,
|
||||||
|
):
|
||||||
|
with mock.patch(
|
||||||
|
"prowler.providers.aws.services.s3.s3_bucket_kms_encryption.s3_bucket_kms_encryption.s3_client",
|
||||||
|
new=S3(audit_info),
|
||||||
|
):
|
||||||
|
# Test Check
|
||||||
|
from prowler.providers.aws.services.s3.s3_bucket_kms_encryption.s3_bucket_kms_encryption import (
|
||||||
|
s3_bucket_kms_encryption,
|
||||||
|
)
|
||||||
|
|
||||||
|
check = s3_bucket_kms_encryption()
|
||||||
|
result = check.execute()
|
||||||
|
|
||||||
|
assert len(result) == 1
|
||||||
|
assert result[0].status == "PASS"
|
||||||
|
assert (
|
||||||
|
result[0].status_extended
|
||||||
|
== f"S3 Bucket {bucket_name_us} has Server Side Encryption with {kms_encryption}."
|
||||||
|
)
|
||||||
|
assert result[0].resource_id == bucket_name_us
|
||||||
|
assert (
|
||||||
|
result[0].resource_arn
|
||||||
|
== f"arn:{audit_info.audited_partition}:s3:::{bucket_name_us}"
|
||||||
|
)
|
||||||
|
assert result[0].resource_tags == []
|
||||||
|
assert result[0].region == AWS_REGION
|
||||||
Reference in New Issue
Block a user