Skip to content

Commit

Permalink
Add ServiceTimeout rule
Browse files Browse the repository at this point in the history
The ServiceTimeout rule emits a warning if the ServiceTimeout property is not specified for a CloudFormation Custom Resource.
  • Loading branch information
ConnorKirk committed Dec 2, 2024
1 parent a1c8868 commit e96161f
Show file tree
Hide file tree
Showing 4 changed files with 118 additions and 0 deletions.
56 changes: 56 additions & 0 deletions src/cfnlint/rules/resources/cloudformation/ServiceTimeout.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,56 @@
"""
Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
SPDX-License-Identifier: MIT-0
"""

from cfnlint.rules import CloudFormationLintRule, RuleMatch


class ServiceTimeout(CloudFormationLintRule):
"""Check a ServiceTimeout property is specified for custom resources"""

id = "W3046"
shortdesc = "Ensure a service timeout is specified"
description = """
Custom resources should have a ServiceTimeout property specified.
The default service timeout is 60 minutes.
Specify a short timeout to reduce waiting if the custom resource fails
"""
source_url = "https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-cloudformation-customresource.html#cfn-cloudformation-customresource-servicetimeout"
tags = [
"resources",
"cloudformation",
"custom resource",
]

def match(self, cfn):
"""Basic Rule Matching"""

matches = []

resources = self._get_custom_resources(cfn)

for resource_name, attributes in resources.items():
properties = attributes.get("Properties", {})
resource_type = attributes.get("Type", None)

if "ServiceTimeout" not in properties:
message = f"Missing ServiceTimeout property in {resource_type}"
matches.append(RuleMatch(["Resources", resource_name], message))

return matches

def _get_custom_resources(self, cfn):
resources = cfn.template.get("Resources", {})
if not isinstance(resources, dict):
return {}

results = {}
for k, v in resources.items():
if isinstance(v, dict):
if (v.get("Type", None) == "AWS::CloudFormation::CustomResource") or (
v.get("Type", None).startswith("Custom::")
):
results[k] = v

return results
Original file line number Diff line number Diff line change
@@ -0,0 +1,10 @@
Resources:
CustomResource:
Type: AWS::CloudFormation::CustomResource
Properties:
ServiceToken: "arn::aws::fake"

CustomResource2:
Type: Custom::CustomResource
Properties:
ServiceToken: "arn::aws::fake"
Original file line number Diff line number Diff line change
@@ -0,0 +1,12 @@
Resources:
CustomResource:
Type: AWS::CloudFormation::CustomResource
Properties:
ServiceToken: "arn::aws::fake"
ServiceTimeout: 60

CustomResource2:
Type: Custom::CustomResource
Properties:
ServiceToken: "arn::aws::fake"
ServiceTimeout: 90
40 changes: 40 additions & 0 deletions test/unit/rules/resources/cloudformation/test_service_timeout.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,40 @@
"""
Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
SPDX-License-Identifier: MIT-0
"""

import logging
from test.unit.rules import BaseRuleTestCase

from cfnlint.rules.resources.cloudformation.ServiceTimeout import ServiceTimeout


class TestServiceTimeout(BaseRuleTestCase):
"""Test CloudFormation Nested stack parameters"""

def tearDown(self) -> None:
super().tearDown()
logger = logging.getLogger("cfnlint.decode.decode")
logger.disabled = False

def setUp(self):
"""Setup"""
super(ServiceTimeout, self).setUp()
self.collection.register(ServiceTimeout())
logger = logging.getLogger("cfnlint.decode.decode")
logger.disabled = True
self.success_templates = [
"test/fixtures/templates/good/resources/cloudformation/service_timeout.yaml"
]

def test_file_positive(self):
"""Test Positive"""
self.helper_file_positive()

def test_file_negative(self):
"""Test failure"""
err_count = 2
self.helper_file_negative(
"test/fixtures/templates/bad/resources/cloudformation/service_timeout.yaml",
err_count,
)

0 comments on commit e96161f

Please sign in to comment.