feat(ACM): Add check and service for ACM (#1365)

This commit is contained in:
Sergio Garcia
2022-10-20 17:17:12 +02:00
committed by GitHub
parent 5c78e6b171
commit bd6eb723dd
14 changed files with 536 additions and 110 deletions

View File

View File

@@ -0,0 +1,35 @@
{
"Provider": "aws",
"CheckID": "acm_certificates_expiration_check",
"CheckTitle": "Check if ACM Certificates are about to expire in specific days or less",
"CheckType": ["Data Protection"],
"ServiceName": "acm",
"SubServiceName": "",
"ResourceIdTemplate": "arn:partition:acm:region:account-id:certificate/resource-id",
"Severity": "high",
"ResourceType": "AwsCertificateManagerCertificate",
"Description": "Check if ACM Certificates are about to expire in specific days or less",
"Risk": "Expired certificates can impact service availability.",
"RelatedUrl": "https://docs.aws.amazon.com/config/latest/developerguide/acm-certificate-expiration-check.html",
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "",
"Terraform": ""
},
"Recommendation": {
"Text": "Monitor certificate expiration and take automated action to renew; replace or remove. Having shorter TTL for any security artifact is a general recommendation; but requires additional automation in place. If not longer required delete certificate. Use AWS config using the managed rule: acm-certificate-expiration-check.",
"Url": "https://docs.aws.amazon.com/config/latest/developerguide/acm-certificate-expiration-check.html"
}
},
"Categories": [],
"Tags": {
"Tag1Key": "value",
"Tag2Key": "value"
},
"DependsOn": [],
"RelatedTo": [],
"Notes": "",
"Compliance": []
}

View File

@@ -0,0 +1,25 @@
from lib.check.models import Check, Check_Report
from providers.aws.services.acm.acm_client import acm_client
DAYS_TO_EXPIRE_THRESHOLD = 7
class acm_certificates_expiration_check(Check):
def execute(self):
findings = []
for certificate in acm_client.certificates:
report = Check_Report(self.metadata)
report.region = certificate.region
if certificate.expiration_days > DAYS_TO_EXPIRE_THRESHOLD:
report.status = "PASS"
report.status_extended = f"ACM Certificate for {certificate.name} expires in {certificate.expiration_days} days."
report.resource_id = certificate.name
report.resource_arn = certificate.arn
else:
report.status = "FAIL"
report.status_extended = f"ACM Certificate for {certificate.name} is about to expire in {DAYS_TO_EXPIRE_THRESHOLD} days."
report.resource_id = certificate.name
report.resource_arn = certificate.arn
findings.append(report)
return findings

View File

@@ -0,0 +1,87 @@
from unittest import mock
from boto3 import client
from moto import mock_acm
AWS_REGION = "us-east-1"
class Test_acm_certificates_expiration_check:
@mock_acm
def test_acm_certificate_expirated(self):
# Generate ACM Client
acm_client = client("acm", region_name=AWS_REGION)
# Request ACM certificate
certificate = acm_client.request_certificate(
DomainName="test.com",
)
from providers.aws.lib.audit_info.audit_info import current_audit_info
from providers.aws.services.acm.acm_service import ACM
current_audit_info.audited_partition = "aws"
with mock.patch(
"providers.aws.services.acm.acm_certificates_expiration_check.acm_certificates_expiration_check.acm_client",
new=ACM(current_audit_info),
) as service_client:
# Test Check
from providers.aws.services.acm.acm_certificates_expiration_check.acm_certificates_expiration_check import (
acm_certificates_expiration_check,
)
service_client.certificates[0].expiration_days = 5
check = acm_certificates_expiration_check()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert result[0].resource_id == "test.com"
assert result[0].resource_arn == certificate["CertificateArn"]
@mock_acm
def test_acm_certificate_not_expirated(self):
# Generate ACM Client
acm_client = client("acm", region_name=AWS_REGION)
# Request ACM certificate
certificate = acm_client.request_certificate(
DomainName="test.com",
)
from providers.aws.lib.audit_info.audit_info import current_audit_info
from providers.aws.services.acm.acm_service import ACM
current_audit_info.audited_partition = "aws"
with mock.patch(
"providers.aws.services.acm.acm_certificates_expiration_check.acm_certificates_expiration_check.acm_client",
new=ACM(current_audit_info),
):
# Test Check
from providers.aws.services.acm.acm_certificates_expiration_check.acm_certificates_expiration_check import (
acm_certificates_expiration_check,
)
check = acm_certificates_expiration_check()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].resource_id == "test.com"
assert result[0].resource_arn == certificate["CertificateArn"]
@mock_acm
def test_bad_response(self):
mock_client = mock.MagicMock()
with mock.patch(
"providers.aws.services.acm.acm_certificates_expiration_check.acm_certificates_expiration_check.acm_client",
new=mock_client,
):
# Test Check
from providers.aws.services.acm.acm_certificates_expiration_check.acm_certificates_expiration_check import (
acm_certificates_expiration_check,
)
check = acm_certificates_expiration_check()
result = check.execute()
assert len(result) == 0

View File

@@ -0,0 +1,35 @@
{
"Provider": "aws",
"CheckID": "acm_certificates_transparency_logs_enabled",
"CheckTitle": "Check if ACM certificates have Certificate Transparency logging enabled",
"CheckType": ["Logging and Monitoring"],
"ServiceName": "acm",
"SubServiceName": "",
"ResourceIdTemplate": "arn:partition:acm:region:account-id:certificate/resource-id",
"Severity": "medium",
"ResourceType": "AwsCertificateManagerCertificate",
"Description": "Check if ACM certificates have Certificate Transparency logging enabled",
"Risk": "Domain owners can search the log to identify unexpected certificates, whether issued by mistake or malice. Domain owners can also identify Certificate Authorities (CAs) that are improperly issuing certificates.",
"RelatedUrl": "https://aws.amazon.com/blogs/security/how-to-get-ready-for-certificate-transparency/",
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "",
"Terraform": ""
},
"Recommendation": {
"Text": "Make sure you are logging information about Lambda operations. Create a lifecycle and use cases for each trail.",
"Url": "https://aws.amazon.com/blogs/security/how-to-get-ready-for-certificate-transparency/"
}
},
"Categories": [],
"Tags": {
"Tag1Key": "value",
"Tag2Key": "value"
},
"DependsOn": [],
"RelatedTo": [],
"Notes": "",
"Compliance": []
}

View File

@@ -0,0 +1,30 @@
from lib.check.models import Check, Check_Report
from providers.aws.services.acm.acm_client import acm_client
class acm_certificates_transparency_logs_enabled(Check):
def execute(self):
findings = []
for certificate in acm_client.certificates:
report = Check_Report(self.metadata)
report.region = certificate.region
if certificate.type == "IMPORTED":
report.status = "PASS"
report.status_extended = (
f"ACM Certificate for {certificate.name} is imported."
)
report.resource_id = certificate.name
report.resource_arn = certificate.arn
else:
if not certificate.transparency_logging:
report.status = "FAIL"
report.status_extended = f"ACM Certificate for {certificate.name} has Certificate Transparency logging disabled."
report.resource_id = certificate.name
report.resource_arn = certificate.arn
else:
report.status = "PASS"
report.status_extended = f"ACM Certificate for {certificate.name} has Certificate Transparency logging enabled."
report.resource_id = certificate.name
report.resource_arn = certificate.arn
findings.append(report)
return findings

View File

@@ -0,0 +1,131 @@
from unittest import mock
from boto3 import client
from moto import mock_acm
AWS_REGION = "us-east-1"
class Test_acm_certificates_transparency_logs_enabled:
@mock_acm
def test_acm_certificate_with_logging(self):
# Generate ACM Client
acm_client = client("acm", region_name=AWS_REGION)
# Request ACM certificate
certificate = acm_client.request_certificate(
DomainName="test.com",
Options={"CertificateTransparencyLoggingPreference": "ENABLED"},
)
from providers.aws.lib.audit_info.audit_info import current_audit_info
from providers.aws.services.acm.acm_service import ACM
current_audit_info.audited_partition = "aws"
with mock.patch(
"providers.aws.services.acm.acm_certificates_transparency_logs_enabled.acm_certificates_transparency_logs_enabled.acm_client",
new=ACM(current_audit_info),
):
# Test Check
from providers.aws.services.acm.acm_certificates_transparency_logs_enabled.acm_certificates_transparency_logs_enabled import (
acm_certificates_transparency_logs_enabled,
)
check = acm_certificates_transparency_logs_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "ACM Certificate for test.com has Certificate Transparency logging enabled."
)
assert result[0].resource_id == "test.com"
assert result[0].resource_arn == certificate["CertificateArn"]
@mock_acm
def test_acm_certificate_without_logging(self):
# Generate ACM Client
acm_client = client("acm", region_name=AWS_REGION)
# Request ACM certificate
certificate = acm_client.request_certificate(
DomainName="test.com",
Options={"CertificateTransparencyLoggingPreference": "ENABLED"},
)
from providers.aws.lib.audit_info.audit_info import current_audit_info
from providers.aws.services.acm.acm_service import ACM
current_audit_info.audited_partition = "aws"
with mock.patch(
"providers.aws.services.acm.acm_certificates_transparency_logs_enabled.acm_certificates_transparency_logs_enabled.acm_client",
new=ACM(current_audit_info),
) as service_client:
# Test Check
from providers.aws.services.acm.acm_certificates_transparency_logs_enabled.acm_certificates_transparency_logs_enabled import (
acm_certificates_transparency_logs_enabled,
)
service_client.certificates[0].transparency_logging = False
check = acm_certificates_transparency_logs_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "ACM Certificate for test.com has Certificate Transparency logging disabled."
)
assert result[0].resource_id == "test.com"
assert result[0].resource_arn == certificate["CertificateArn"]
@mock_acm
def test_acm_default_certificate(self):
# Generate ACM Client
acm_client = client("acm", region_name=AWS_REGION)
# Request ACM certificate
certificate = acm_client.request_certificate(
DomainName="test.com",
)
from providers.aws.lib.audit_info.audit_info import current_audit_info
from providers.aws.services.acm.acm_service import ACM
current_audit_info.audited_partition = "aws"
with mock.patch(
"providers.aws.services.acm.acm_certificates_transparency_logs_enabled.acm_certificates_transparency_logs_enabled.acm_client",
new=ACM(current_audit_info),
):
# Test Check
from providers.aws.services.acm.acm_certificates_transparency_logs_enabled.acm_certificates_transparency_logs_enabled import (
acm_certificates_transparency_logs_enabled,
)
check = acm_certificates_transparency_logs_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "ACM Certificate for test.com has Certificate Transparency logging enabled."
)
assert result[0].resource_id == "test.com"
assert result[0].resource_arn == certificate["CertificateArn"]
@mock_acm
def test_bad_response(self):
mock_client = mock.MagicMock()
with mock.patch(
"providers.aws.services.acm.acm_certificates_transparency_logs_enabled.acm_certificates_transparency_logs_enabled.acm_client",
new=mock_client,
):
# Test Check
from providers.aws.services.acm.acm_certificates_transparency_logs_enabled.acm_certificates_transparency_logs_enabled import (
acm_certificates_transparency_logs_enabled,
)
check = acm_certificates_transparency_logs_enabled()
result = check.execute()
assert len(result) == 0

View File

@@ -0,0 +1,4 @@
from providers.aws.lib.audit_info.audit_info import current_audit_info
from providers.aws.services.acm.acm_service import ACM
acm_client = ACM(current_audit_info)

View File

@@ -0,0 +1,95 @@
import threading
from dataclasses import dataclass
from config.config import timestamp_utc
from lib.logger import logger
from providers.aws.aws_provider import generate_regional_clients
################## ACM
class ACM:
def __init__(self, audit_info):
self.service = "acm"
self.session = audit_info.audit_session
self.audited_account = audit_info.audited_account
self.regional_clients = generate_regional_clients(self.service, audit_info)
self.certificates = []
self.__threading_call__(self.__list_certificates__)
self.__describe_certificates__()
def __get_session__(self):
return self.session
def __threading_call__(self, call):
threads = []
for regional_client in self.regional_clients.values():
threads.append(threading.Thread(target=call, args=(regional_client,)))
for t in threads:
t.start()
for t in threads:
t.join()
def __list_certificates__(self, regional_client):
logger.info("ACM - Listing Certificates...")
try:
list_certificates_paginator = regional_client.get_paginator(
"list_certificates"
)
for page in list_certificates_paginator.paginate():
for analyzer in page["CertificateSummaryList"]:
self.certificates.append(
Certificate(
analyzer["CertificateArn"],
analyzer["DomainName"],
False,
regional_client.region,
)
)
except Exception as error:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}: {error}"
)
def __describe_certificates__(self):
logger.info("ACM - Describing Certificates...")
try:
for certificate in self.certificates:
regional_client = self.regional_clients[certificate.region]
response = regional_client.describe_certificate(
CertificateArn=certificate.arn
)["Certificate"]
certificate.type = response["Type"]
certificate.expiration_days = (
response["NotAfter"] - timestamp_utc
).days
if (
response["Options"]["CertificateTransparencyLoggingPreference"]
== "ENABLED"
):
certificate.transparency_logging = True
except Exception as error:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}: {error}"
)
@dataclass
class Certificate:
arn: str
name: str
type: str
expiration_days: int
transparency_logging: bool
region: str
def __init__(
self,
arn,
name,
transparency_logging,
region,
):
self.arn = arn
self.name = name
self.transparency_logging = transparency_logging
self.region = region

View File

@@ -0,0 +1,94 @@
from boto3 import client, session
from moto import mock_acm
from providers.aws.lib.audit_info.models import AWS_Audit_Info
from providers.aws.services.acm.acm_service import ACM
AWS_ACCOUNT_NUMBER = 123456789012
AWS_REGION = "us-east-1"
class Test_ACM_Service:
# Mocked Audit Info
def set_mocked_audit_info(self):
audit_info = AWS_Audit_Info(
original_session=None,
audit_session=session.Session(
profile_name=None,
botocore_session=None,
),
audited_account=AWS_ACCOUNT_NUMBER,
audited_user_id=None,
audited_partition="aws",
audited_identity_arn=None,
profile=None,
profile_region=None,
credentials=None,
assumed_role_info=None,
audited_regions=None,
organizations_metadata=None,
)
return audit_info
# Test ACM Service
@mock_acm
def test_service(self):
# ACM client for this test class
audit_info = self.set_mocked_audit_info()
acm = ACM(audit_info)
assert acm.service == "acm"
# Test ACM Client
@mock_acm
def test_client(self):
# ACM client for this test class
audit_info = self.set_mocked_audit_info()
acm = ACM(audit_info)
for client in acm.regional_clients.values():
assert client.__class__.__name__ == "ACM"
# Test ACM Session
@mock_acm
def test__get_session__(self):
# ACM client for this test class
audit_info = self.set_mocked_audit_info()
acm = ACM(audit_info)
assert acm.session.__class__.__name__ == "Session"
# Test ACM Session
@mock_acm
def test_audited_account(self):
# ACM client for this test class
audit_info = self.set_mocked_audit_info()
acm = ACM(audit_info)
assert acm.audited_account == AWS_ACCOUNT_NUMBER
# Test ACM List Certificates
@mock_acm
def test__list_certificates__(self):
# Generate ACM Client
acm_client = client("acm", region_name=AWS_REGION)
# Request ACM certificate
certificate = acm_client.request_certificate(
DomainName="test.com",
)
# ACM client for this test class
audit_info = self.set_mocked_audit_info()
acm = ACM(audit_info)
assert len(acm.certificates) == 1
assert acm.certificates[0].arn == certificate["CertificateArn"]
# Test ACM Describe Certificates
@mock_acm
def test__describe_certificates__(self):
# Generate ACM Client
acm_client = client("acm", region_name=AWS_REGION)
# Request ACM certificate
certificate = acm_client.request_certificate(
DomainName="test.com",
)
# ACM client for this test class
audit_info = self.set_mocked_audit_info()
acm = ACM(audit_info)
assert acm.certificates[0].type == "AMAZON_ISSUED"
assert acm.certificates[0].arn == certificate["CertificateArn"]

View File

@@ -1,50 +0,0 @@
#!/usr/bin/env bash
# Prowler - the handy cloud security tool (copyright 2018) by Toni de la Fuente
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not
# use this file except in compliance with the License. You may obtain a copy
# of the License at http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software distributed
# under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
# CONDITIONS OF ANY KIND, either express or implied. See the License for the
# specific language governing permissions and limitations under the License.
CHECK_ID_extra724="7.24"
CHECK_TITLE_extra724="[extra724] Check if ACM certificates have Certificate Transparency logging enabled"
CHECK_SCORED_extra724="NOT_SCORED"
CHECK_CIS_LEVEL_extra724="EXTRA"
CHECK_SEVERITY_extra724="Medium"
CHECK_ASFF_RESOURCE_TYPE_extra724="AwsCertificateManagerCertificate"
CHECK_ALTERNATE_check724="extra724"
CHECK_SERVICENAME_extra724="acm"
CHECK_RISK_extra724='Domain owners can search the log to identify unexpected certificates; whether issued by mistake or malice. Domain owners can also identify Certificate Authorities (CAs) that are improperly issuing certificates.'
CHECK_REMEDIATION_extra724='Make sure you are logging information about Lambda operations. Create a lifecycle and use cases for each trail.'
CHECK_DOC_extra724='https://aws.amazon.com/blogs/security/how-to-get-ready-for-certificate-transparency/'
CHECK_CAF_EPIC_extra724='Logging and Monitoring'
extra724(){
# "Check if ACM certificates have Certificate Transparency logging enabled "
for regx in $REGIONS; do
LIST_OF_CERTS=$($AWSCLI acm list-certificates $PROFILE_OPT --region $regx --query CertificateSummaryList[].CertificateArn --output text)
if [[ $LIST_OF_CERTS ]];then
for cert_arn in $LIST_OF_CERTS;do
CT_ENABLED=$($AWSCLI acm describe-certificate $PROFILE_OPT --region $regx --certificate-arn $cert_arn --query Certificate.Options.CertificateTransparencyLoggingPreference --output text)
CERT_DOMAIN_NAME=$(aws acm describe-certificate $PROFILE_OPT --region $regx --certificate-arn $cert_arn --query Certificate.DomainName --output text)
CERT_TYPE=$(aws acm describe-certificate $PROFILE_OPT --region $regx --certificate-arn $cert_arn --query Certificate.Type --output text)
if [[ $CERT_TYPE == "IMPORTED" ]];then
# Ignore imported certificate
textInfo "$regx: ACM Certificate $CERT_DOMAIN_NAME is imported." "$regx" "$CERT_DOMAIN_NAME"
else
if [[ $CT_ENABLED == "ENABLED" ]];then
textPass "$regx: ACM Certificate $CERT_DOMAIN_NAME has Certificate Transparency logging enabled!" "$regx" "$CERT_DOMAIN_NAME"
else
textFail "$regx: ACM Certificate $CERT_DOMAIN_NAME has Certificate Transparency logging disabled!" "$regx" "$CERT_DOMAIN_NAME"
fi
fi
done
else
textInfo "$regx: No ACM Certificates found" "$regx"
fi
done
}

View File

@@ -1,60 +0,0 @@
#!/usr/bin/env bash
# Prowler - the handy cloud security tool (copyright 2018) by Toni de la Fuente
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not
# use this file except in compliance with the License. You may obtain a copy
# of the License at http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software distributed
# under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
# CONDITIONS OF ANY KIND, either express or implied. See the License for the
# specific language governing permissions and limitations under the License.
DAYS_TO_EXPIRE_THRESHOLD="7"
CHECK_ID_extra730="7.30"
CHECK_TITLE_extra730="[extra730] Check if ACM Certificates are about to expire in $DAYS_TO_EXPIRE_THRESHOLD days or less"
CHECK_SCORED_extra730="NOT_SCORED"
CHECK_CIS_LEVEL_extra730="EXTRA"
CHECK_SEVERITY_extra730="High"
CHECK_ASFF_RESOURCE_TYPE_extra730="AwsCertificateManagerCertificate"
CHECK_ALTERNATE_check730="extra730"
CHECK_SERVICENAME_extra730="acm"
CHECK_RISK_extra730='Expired certificates can impact service availability.'
CHECK_REMEDIATION_extra730='Monitor certificate expiration and take automated action to renew; replace or remove. Having shorter TTL for any security artifact is a general recommendation; but requires additional automation in place. If not longer required delete certificate. Use AWS config using the managed rule: acm-certificate-expiration-check.'
CHECK_DOC_extra730='https://docs.aws.amazon.com/config/latest/developerguide/acm-certificate-expiration-check.html'
CHECK_CAF_EPIC_extra730='Data Protection'
extra730(){
# Only RSA key types, needed to recover Amazon Issued, Imported and Private PKIs
local ACM_KEY_TYPES="RSA_1024,RSA_2048,RSA_3072,RSA_4096"
local ACM_CERTIFICATE_STATUSES="ISSUED"
# "Check if ACM Certificates are about to expire in $DAYS_TO_EXPIRE_THRESHOLD days or less"
for regx in $REGIONS; do
LIST_OF_ACM_CERTS=$("${AWSCLI}" acm list-certificates ${PROFILE_OPT} --region "${regx}" --include keyTypes="${ACM_KEY_TYPES}" --certificate-statuses "${ACM_CERTIFICATE_STATUSES}" --query 'CertificateSummaryList[].CertificateArn' --output text)
if [[ $LIST_OF_ACM_CERTS ]]; then
for cert in $LIST_OF_ACM_CERTS; do
CERT_DATA=$("${AWSCLI}" acm describe-certificate ${PROFILE_OPT} --region "${regx}" --certificate-arn "${cert}" --query 'Certificate.[DomainName,NotAfter]' --output text)
# Format: domain.test.com YYYY-MM-DDTHH:MM:SS
echo "$CERT_DATA" | while read -r FQDN NOTAFTER; do
EXPIRES_DATE=$(timestamp_to_date "${NOTAFTER}")
if [[ "${EXPIRES_DATE}" == "" ]]
then
textInfo "${regx}: Certificate for ${FQDN} has an incorrect timestamp format: ${NOTAFTER}" "${regx}" "${FQDN}"
else
COUNTER_DAYS=$(how_many_days_from_today "${EXPIRES_DATE}")
if [[ $COUNTER_DAYS -le $DAYS_TO_EXPIRE_THRESHOLD ]]; then
textFail "${regx}: Certificate for ${FQDN} is about to expire in ${COUNTER_DAYS} days!" "${regx}" "${FQDN}"
else
textPass "${regx}: Certificate for ${FQDN} expires in ${COUNTER_DAYS} days" "${regx}" "{$FQDN}"
fi
fi
done
done
else
textInfo "${regx}: No certificates found" "${regx}"
fi
done
}