Skip to content

Commit c2c4828

Browse files
kddejongkddejong
authored
Add 3 new rules to validate ECS configs (#3546)
* Add 3 new rules to validate ECS configs
1 parent 4d2355d commit c2c4828

File tree

6 files changed

+1132
-0
lines changed

6 files changed

+1132
-0
lines changed

‎src/cfnlint/rules/resources/ecs/ServiceFargate.py

+138
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,138 @@
1+
"""
2+
Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
3+
SPDX-License-Identifier: MIT-0
4+
"""
5+
6+
from __future__ import annotations
7+
8+
from collections import deque
9+
from typing import Any, Iterator
10+
11+
from cfnlint.helpers import ensure_list, is_function
12+
from cfnlint.jsonschema import ValidationError, ValidationResult
13+
from cfnlint.jsonschema.protocols import Validator
14+
from cfnlint.rules.helpers import get_resource_by_name, get_value_from_path
15+
from cfnlint.rules.jsonschema.CfnLintKeyword import CfnLintKeyword
16+
17+
18+
class ServiceFargate(CfnLintKeyword):
19+
id = "E3054"
20+
shortdesc = (
21+
"Validate ECS service using Fargate uses TaskDefinition that allows Fargate"
22+
)
23+
description = (
24+
"When using an ECS service with 'LaunchType' of 'FARGATE' "
25+
"the associated task definition must have 'RequiresCompatibilities' "
26+
"specified with 'FARGATE' listed"
27+
)
28+
tags = ["resources", "ecs"]
29+
30+
def __init__(self) -> None:
31+
super().__init__(
32+
keywords=["Resources/AWS::ECS::Service/Properties"],
33+
)
34+
35+
def _filter_resource_name(self, instance: Any) -> str | None:
36+
fn_k, fn_v = is_function(instance)
37+
if fn_k is None:
38+
return None
39+
if fn_k == "Ref":
40+
if isinstance(fn_v, str):
41+
return fn_v
42+
elif fn_k == "Fn::GetAtt":
43+
name = ensure_list(fn_v)[0].split(".")[0]
44+
if isinstance(name, str):
45+
return name
46+
return None
47+
48+
def _get_service_properties(
49+
self, validator: Validator, instance: Any
50+
) -> Iterator[tuple[str, str, Validator]]:
51+
for task_definition_id, task_definition_validator in get_value_from_path(
52+
validator, instance, deque(["TaskDefinition"])
53+
):
54+
task_definition_resource_name = self._filter_resource_name(
55+
task_definition_id
56+
)
57+
if task_definition_resource_name is None:
58+
continue
59+
60+
for (
61+
launch_type,
62+
launch_type_validator,
63+
) in get_value_from_path(
64+
task_definition_validator, instance, deque(["LaunchType"])
65+
):
66+
yield (
67+
task_definition_resource_name,
68+
launch_type,
69+
launch_type_validator,
70+
)
71+
72+
def _get_task_definition_properties(
73+
self, validator: Validator, resource_name: Any
74+
) -> Iterator[tuple[list[Any] | None, Validator]]:
75+
task_definition, task_definition_validator = get_resource_by_name(
76+
validator, resource_name, ["AWS::ECS::TaskDefinition"]
77+
)
78+
if not task_definition:
79+
return
80+
81+
for capabilities, capabilities_validator in get_value_from_path(
82+
task_definition_validator,
83+
task_definition,
84+
path=deque(["Properties", "RequiresCompatibilities"]),
85+
):
86+
if capabilities is None:
87+
yield capabilities, capabilities_validator
88+
continue
89+
if not isinstance(capabilities, list):
90+
continue
91+
for capibility, _ in get_value_from_path(
92+
capabilities_validator,
93+
capabilities,
94+
path=deque(["*"]),
95+
):
96+
if isinstance(capibility, dict) or capibility == "FARGATE":
97+
break
98+
else:
99+
yield capabilities, capabilities_validator
100+
101+
def validate(
102+
self, validator: Validator, _: Any, instance: Any, schema: dict[str, Any]
103+
) -> ValidationResult:
104+
105+
for (
106+
task_definition_resource_name,
107+
launch_type,
108+
service_validator,
109+
) in self._get_service_properties(
110+
validator,
111+
instance,
112+
):
113+
if launch_type != "FARGATE":
114+
continue
115+
for (
116+
capabilities,
117+
capabilities_validator,
118+
) in self._get_task_definition_properties(
119+
service_validator,
120+
task_definition_resource_name,
121+
):
122+
if capabilities is None:
123+
yield ValidationError(
124+
"'RequiresCompatibilities' is a required property",
125+
validator="required",
126+
rule=self,
127+
path_override=deque(
128+
list(capabilities_validator.context.path.path)[:-1]
129+
),
130+
)
131+
continue
132+
133+
yield ValidationError(
134+
f"{capabilities!r} does not contain items matching 'FARGATE'",
135+
validator="contains",
136+
rule=self,
137+
path_override=capabilities_validator.context.path.path,
138+
)

‎src/cfnlint/rules/resources/ecs/ServiceNetworkConfiguration.py

+107
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,107 @@
1+
"""
2+
Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
3+
SPDX-License-Identifier: MIT-0
4+
"""
5+
6+
from __future__ import annotations
7+
8+
from collections import deque
9+
from typing import Any, Iterator
10+
11+
from cfnlint.helpers import ensure_list, is_function
12+
from cfnlint.jsonschema import ValidationError, ValidationResult
13+
from cfnlint.jsonschema.protocols import Validator
14+
from cfnlint.rules.helpers import get_resource_by_name, get_value_from_path
15+
from cfnlint.rules.jsonschema.CfnLintKeyword import CfnLintKeyword
16+
17+
18+
class ServiceNetworkConfiguration(CfnLintKeyword):
19+
id = "E3052"
20+
shortdesc = "Validate ECS service requires NetworkConfiguration"
21+
description = (
22+
"When using an ECS task definition has NetworkMode set to "
23+
"'awsvpc' then 'NetworkConfiguration' is required"
24+
)
25+
tags = ["resources", "ecs"]
26+
27+
def __init__(self) -> None:
28+
super().__init__(
29+
keywords=["Resources/AWS::ECS::Service/Properties"],
30+
)
31+
32+
def _filter_resource_name(self, instance: Any) -> str | None:
33+
fn_k, fn_v = is_function(instance)
34+
if fn_k is None:
35+
return None
36+
if fn_k == "Ref":
37+
if isinstance(fn_v, str):
38+
return fn_v
39+
elif fn_k == "Fn::GetAtt":
40+
name = ensure_list(fn_v)[0].split(".")[0]
41+
if isinstance(name, str):
42+
return name
43+
return None
44+
45+
def _get_service_properties(
46+
self, validator: Validator, instance: Any
47+
) -> Iterator[tuple[str, str, Validator]]:
48+
for task_definition_id, task_definition_validator in get_value_from_path(
49+
validator, instance, deque(["TaskDefinition"])
50+
):
51+
task_definition_resource_name = self._filter_resource_name(
52+
task_definition_id
53+
)
54+
if task_definition_resource_name is None:
55+
continue
56+
57+
for (
58+
network_configuration,
59+
network_configuration_validator,
60+
) in get_value_from_path(
61+
task_definition_validator, instance, deque(["NetworkConfiguration"])
62+
):
63+
yield (
64+
task_definition_resource_name,
65+
network_configuration,
66+
network_configuration_validator,
67+
)
68+
69+
def _get_task_definition_properties(
70+
self, validator: Validator, resource_name: Any
71+
) -> Iterator[tuple[str | int | None, Validator]]:
72+
target_group, target_group_validator = get_resource_by_name(
73+
validator, resource_name, ["AWS::ECS::TaskDefinition"]
74+
)
75+
if not target_group:
76+
return
77+
78+
for network_mode, network_mode_validator in get_value_from_path(
79+
target_group_validator,
80+
target_group,
81+
path=deque(["Properties", "NetworkMode"]),
82+
):
83+
if network_mode == "awsvpc":
84+
yield network_mode, network_mode_validator
85+
86+
def validate(
87+
self, validator: Validator, _: Any, instance: Any, schema: dict[str, Any]
88+
) -> ValidationResult:
89+
90+
for (
91+
task_definition_resource_name,
92+
network_configuration,
93+
task_definition_validator,
94+
) in self._get_service_properties(
95+
validator,
96+
instance,
97+
):
98+
for _, _ in self._get_task_definition_properties(
99+
task_definition_validator,
100+
task_definition_resource_name,
101+
):
102+
if network_configuration is None:
103+
yield ValidationError(
104+
"'NetworkConfiguration' is a required property",
105+
validator="required",
106+
rule=self,
107+
)

‎src/cfnlint/rules/resources/ecs/TaskDefinitionAwsVpc.py

+81
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,81 @@
1+
"""
2+
Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
3+
SPDX-License-Identifier: MIT-0
4+
"""
5+
6+
from __future__ import annotations
7+
8+
from collections import deque
9+
from typing import Any, Iterator
10+
11+
from cfnlint.jsonschema import ValidationError, ValidationResult
12+
from cfnlint.jsonschema.protocols import Validator
13+
from cfnlint.rules.helpers import get_value_from_path
14+
from cfnlint.rules.jsonschema.CfnLintKeyword import CfnLintKeyword
15+
16+
17+
class TaskDefinitionAwsVpc(CfnLintKeyword):
18+
id = "E3053"
19+
shortdesc = "Validate ECS task definition is has correct values for 'HostPort'"
20+
description = (
21+
"The 'HostPort' must either be undefined or equal to "
22+
"the 'ContainerPort' value"
23+
)
24+
tags = ["resources", "ecs"]
25+
26+
def __init__(self) -> None:
27+
super().__init__(
28+
keywords=["Resources/AWS::ECS::TaskDefinition/Properties"],
29+
)
30+
31+
def _get_port_mappings(
32+
self, validator: Validator, instance: Any
33+
) -> Iterator[tuple[str | int | None, str | int | None, Validator]]:
34+
35+
for container_definition, container_definition_validator in get_value_from_path(
36+
validator,
37+
instance,
38+
path=deque(["ContainerDefinitions", "*", "PortMappings", "*"]),
39+
):
40+
for host_port, host_port_validator in get_value_from_path(
41+
container_definition_validator,
42+
container_definition,
43+
path=deque(["HostPort"]),
44+
):
45+
if not isinstance(host_port, (str, int)):
46+
continue
47+
for container_port, _ in get_value_from_path(
48+
host_port_validator,
49+
container_definition,
50+
path=deque(["ContainerPort"]),
51+
):
52+
if not isinstance(container_port, (str, int)):
53+
continue
54+
if str(host_port) != str(container_port):
55+
yield host_port, container_port, host_port_validator
56+
57+
def validate(
58+
self, validator: Validator, _: Any, instance: Any, schema: dict[str, Any]
59+
) -> ValidationResult:
60+
61+
for network_mode, _ in get_value_from_path(
62+
validator,
63+
instance,
64+
path=deque(["NetworkMode"]),
65+
):
66+
if network_mode != "awsvpc":
67+
continue
68+
for (
69+
host_port,
70+
container_port,
71+
port_mapping_validator,
72+
) in self._get_port_mappings(
73+
validator,
74+
instance,
75+
):
76+
yield ValidationError(
77+
f"{host_port!r} does not equal {container_port!r}",
78+
validator="const",
79+
rule=self,
80+
path_override=port_mapping_validator.context.path.path,
81+
)

0 commit comments

Comments
 (0)