Release 1.112.0 (to main)

.cfnlintrc.yaml

@@ -111,6 +111,7 @@ ignore_templates:
   - tests/translator/output/**/function_with_msk_with_schema_registry_config.json # cfnlint is not updated to recognize the SchemaRegistryConfig property
   - tests/translator/output/**/function_with_logging_config.json # cfnlint is not updated to recognize the LoggingConfig property
   - tests/translator/output/aws-*/*capacity_provider*.json # Ignore Capacity Provider test format in non-aws partitions
+  - tests/translator/output/**/capacity_provider_managed_resource_tags.json # cfnlint not updated for CapacityProvider mode (Phase 2)
 
 ignore_checks:
   - E2531 # Deprecated runtime; not relevant for transform tests

integration/single/test_basic_network_connector.py

@@ -48,12 +48,13 @@ def test_basic_network_connector(self):
                 principals.extend(service)
         self.assertIn("lambda.amazonaws.com", principals)
 
-        # Verify inline policy
-        role_policy = iam_client.get_role_policy(RoleName=role_name, PolicyName="NetworkConnectorOperatorPolicy")
-        statements = role_policy["PolicyDocument"]["Statement"]
-        all_actions = [s["Action"] for s in statements]
-        self.assertIn("ec2:CreateNetworkInterface", all_actions)
-        self.assertIn("ec2:CreateTags", all_actions)
+        # Verify managed policy
+        attached = iam_client.list_attached_role_policies(RoleName=role_name)
+        policy_arns = [p["PolicyArn"] for p in attached["AttachedPolicies"]]
+        self.assertTrue(
+            any("AWSLambdaNetworkConnectorOperatorPolicy" in arn for arn in policy_arns),
+            "OperatorRole should have AWSLambdaNetworkConnectorOperatorPolicy attached",
+        )
 
     def test_network_connector_with_custom_role(self):
         """

samtranslator/__init__.py

@@ -1 +1 @@
-__version__ = "1.111.0"
+__version__ = "1.112.0"

samtranslator/internal/schema_source/aws_serverless_capacity_provider.py

@@ -16,11 +16,18 @@
 VPC_CONFIG_STEM = "sam-property-capacityprovider-vpcconfig"
 INSTANCE_REQUIREMENTS_STEM = "sam-property-capacityprovider-instancerequirements"
 SCALING_CONFIG_STEM = "sam-property-capacityprovider-scalingconfig"
+MANAGED_RESOURCE_TAGS_STEM = "sam-property-capacityprovider-managedresourcetags"
 
 properties = get_prop(PROPERTIES_STEM)
 vpcconfig = get_prop(VPC_CONFIG_STEM)
 instancerequirements = get_prop(INSTANCE_REQUIREMENTS_STEM)
 scalingconfig = get_prop(SCALING_CONFIG_STEM)
+managedresourcetags = get_prop(MANAGED_RESOURCE_TAGS_STEM)
+
+
+class ManagedResourceTags(BaseModel):
+    Tags: DictStrAny | None = managedresourcetags("Tags")
+    Propagate: bool | None = managedresourcetags("Propagate")
 
 
 class VpcConfig(BaseModel):
@@ -82,6 +89,8 @@ class Properties(BaseModel):
     # Uses custom ScalingConfig class because SAM renames construct (CapacityProviderScalingConfig→ScalingConfig)
     ScalingConfig: ScalingConfig | None = properties("ScalingConfig")
 
+    ManagedResourceTags: ManagedResourceTags | None = properties("ManagedResourceTags")
+
     KmsKeyArn: PassThroughProp | None = passthrough_prop(
         PROPERTIES_STEM,
         "KmsKeyArn",
@@ -113,6 +122,8 @@ class Globals(BaseModel):
     # Uses custom ScalingConfig class because SAM renames construct (CapacityProviderScalingConfig→ScalingConfig)
     ScalingConfig: ScalingConfig | None = properties("ScalingConfig")
 
+    ManagedResourceTags: ManagedResourceTags | None = properties("ManagedResourceTags")
+
     KmsKeyArn: PassThroughProp | None = passthrough_prop(
         PROPERTIES_STEM,
         "KmsKeyArn",

samtranslator/model/__init__.py

@@ -576,6 +576,7 @@ class ValidationRule(Enum):
 
 
 # Simple tuple-based rules: (rule_type, [property_names])
+# Property names support "Property.Path=Value" syntax for value-conditional presence checks
 PropertyRule = tuple[ValidationRule, list[str]]
 
 
@@ -726,7 +727,7 @@ def validate_before_transform(self, schema_class: type[RT] | None, collect_all_e
         error_messages = []
 
         for rule_type, properties in rules:
-            present = [prop for prop in properties if self._get_property_value(prop, validated_model) is not None]
+            present = [prop for prop in properties if self._is_property_present(prop, validated_model)]
             if rule_type == ValidationRule.MUTUALLY_EXCLUSIVE:
                 # Check if more than one property exists
                 if len(present) > 1:
@@ -779,6 +780,30 @@ def _get_property_value(self, prop: str, validated_model: Any = None) -> Any:
         except Exception:
             return None
 
+    def _is_property_present(self, prop: str, validated_model: Any = None) -> bool:
+        """Check if a property is 'present' for validation purposes.
+
+        Supports 'Property.Path=Value' syntax: property is only considered present
+        when its value matches the specified value. Without '=', checks non-None.
+
+        The '=' is split only on the first occurrence, so values containing '=' or
+        spaces (e.g. 'Prop=Hello World') are handled correctly.
+        """
+        if "=" not in prop:
+            return self._get_property_value(prop, validated_model) is not None
+
+        prop_path, expected_str = prop.split("=", 1)
+        actual_value = self._get_property_value(prop_path, validated_model)
+
+        if actual_value is None:
+            return False
+
+        if expected_str.lower() == "true":
+            return actual_value is True
+        if expected_str.lower() == "false":
+            return actual_value is False
+        return str(actual_value) == expected_str
+
 
 class ResourceTypeResolver:
     """ResourceTypeResolver maps Resource Types to Resource classes, e.g. AWS::Serverless::Function to

samtranslator/model/capacity_provider/generators.py

@@ -45,6 +45,7 @@ def __init__(self, logical_id: str, **kwargs: Any) -> None:
         self.instance_requirements = kwargs.get("instance_requirements") or {}
         self.scaling_config = kwargs.get("scaling_config") or {}
         self.kms_key_arn = kwargs.get("kms_key_arn")
+        self.managed_resource_tags = kwargs.get("managed_resource_tags")
         self.depends_on = kwargs.get("depends_on")
         self.resource_attributes = kwargs.get("resource_attributes")
         self.passthrough_resource_attributes = kwargs.get("passthrough_resource_attributes")
@@ -111,6 +112,10 @@ def _create_capacity_provider(self) -> LambdaCapacityProvider:
         if self.kms_key_arn:
             capacity_provider.KmsKeyArn = self.kms_key_arn
 
+        # Set PropagateTags from ManagedResourceTags if provided
+        if self.managed_resource_tags:
+            capacity_provider.PropagateTags = self._transform_managed_resource_tags()
+
         # Pass through resource attributes
         if self.passthrough_resource_attributes:
             for attr_name, attr_value in self.passthrough_resource_attributes.items():
@@ -214,3 +219,17 @@ def _create_operator_role(self) -> IAMRole:
         operator_role.logical_id = role_logical_id
 
         return operator_role
+
+    def _transform_managed_resource_tags(self) -> dict[str, Any]:
+        """
+        Transform SAM ManagedResourceTags to CFN PropagateTags format.
+        """
+        tags: dict[str, Any] = self.managed_resource_tags or {}
+
+        if "Tags" in tags:
+            return {"Mode": "Explicit", "ExplicitTags": get_tag_list(tags["Tags"])}
+
+        if "Propagate" in tags:
+            return {"Mode": "CapacityProvider" if tags["Propagate"] else "None"}
+
+        return {}

samtranslator/model/capacity_provider/resources.py

@@ -23,6 +23,7 @@ class LambdaCapacityProvider(Resource):
         "InstanceRequirements": GeneratedProperty(),
         "CapacityProviderScalingConfig": GeneratedProperty(),
         "KmsKeyArn": GeneratedProperty(),
+        "PropagateTags": GeneratedProperty(),
     }
 
     CapacityProviderName: Intrinsicable[str] | None
@@ -32,6 +33,7 @@ class LambdaCapacityProvider(Resource):
     InstanceRequirements: dict[str, Any] | None
     CapacityProviderScalingConfig: dict[str, Any] | None
     KmsKeyArn: Intrinsicable[str] | None
+    PropagateTags: dict[str, Any] | None
 
     runtime_attrs = {
         "name": lambda self: ref(self.logical_id),

samtranslator/model/network_connector/generators.py

@@ -5,10 +5,13 @@
 from typing import Any
 
 from samtranslator.model import Resource
-from samtranslator.model.iam import IAMRole, IAMRolePolicies
+from samtranslator.model.iam import IAMRolePolicies
 from samtranslator.model.intrinsics import fnGetAtt
 from samtranslator.model.network_connector.resources import LambdaNetworkConnector
+from samtranslator.model.resource_policies import ResourcePolicies
+from samtranslator.model.role_utils import construct_role_for_resource
 from samtranslator.model.tags.resource_tagging import get_tag_list
+from samtranslator.translator.arn_generator import ArnGenerator
 
 
 class NetworkConnectorGenerator:
@@ -72,65 +75,30 @@ def _create_network_connector(self) -> LambdaNetworkConnector:
 
         return connector
 
-    def _create_operator_role(self) -> IAMRole:
+    def _create_operator_role(self) -> Resource:
         role_logical_id = f"{self.logical_id}OperatorRole"
 
-        assume_role_policy = IAMRolePolicies.construct_assume_role_policy_for_service_principal("lambda.amazonaws.com")
-
-        role = IAMRole(role_logical_id, attributes=self.passthrough_resource_attributes)
-        role.AssumeRolePolicyDocument = assume_role_policy
-        role.Policies = [
-            {
-                "PolicyName": "NetworkConnectorOperatorPolicy",
-                "PolicyDocument": {
-                    "Version": "2012-10-17",
-                    "Statement": [
-                        {
-                            "Sid": "AllowCreateEniInAnySubnet",
-                            "Effect": "Allow",
-                            "Action": "ec2:CreateNetworkInterface",
-                            "Resource": {"Fn::Sub": "arn:${AWS::Partition}:ec2:*:*:subnet/*"},
-                        },
-                        {
-                            "Sid": "AllowCreateEniWithSecurityGroups",
-                            "Effect": "Allow",
-                            "Action": "ec2:CreateNetworkInterface",
-                            "Resource": {"Fn::Sub": "arn:${AWS::Partition}:ec2:*:*:security-group/*"},
-                        },
-                        {
-                            "Sid": "AllowCreateEniWithLambdaTagKeys",
-                            "Effect": "Allow",
-                            "Action": "ec2:CreateNetworkInterface",
-                            "Resource": {"Fn::Sub": "arn:${AWS::Partition}:ec2:*:*:network-interface/*"},
-                            "Condition": {
-                                "ForAllValues:StringEquals": {
-                                    "aws:TagKeys": [
-                                        "aws:lambda:networkConnectorName",
-                                        "aws:lambda:networkConnectorId",
-                                    ]
-                                }
-                            },
-                        },
-                        {
-                            "Sid": "TagENIOnCreate",
-                            "Effect": "Allow",
-                            "Action": "ec2:CreateTags",
-                            "Resource": {"Fn::Sub": "arn:${AWS::Partition}:ec2:*:*:network-interface/*"},
-                            "Condition": {
-                                "StringEquals": {
-                                    "ec2:CreateAction": "CreateNetworkInterface",
-                                    "ec2:ManagedResourceOperator": "network-connectors.lambda.amazonaws.com",
-                                }
-                            },
-                        },
-                    ],
-                },
-            }
-        ]
+        assume_role_policy_document = IAMRolePolicies.construct_assume_role_policy_for_service_principal(
+            "lambda.amazonaws.com"
+        )
+
+        tags = self._transform_tags()
+
+        managed_policy_arns = [ArnGenerator.generate_aws_managed_policy_arn("AWSLambdaNetworkConnectorOperatorPolicy")]
+
+        operator_role = construct_role_for_resource(
+            resource_logical_id=self.logical_id,
+            attributes=self.passthrough_resource_attributes,
+            managed_policy_map=None,
+            assume_role_policy_document=assume_role_policy_document,
+            resource_policies=ResourcePolicies({}),
+            managed_policy_arns=managed_policy_arns,
+            tags=tags,
+        )
 
-        role.Tags = self._transform_tags()
+        operator_role.logical_id = role_logical_id
 
-        return role
+        return operator_role
 
     def _transform_tags(self, tags: dict[str, Any] | None = None) -> list[dict[str, str]]:
         tags_dict = (tags or {}).copy()

samtranslator/model/sam_resources.py

@@ -1556,6 +1556,7 @@ class SamCapacityProvider(SamResourceMacro):
         "InstanceRequirements": Property(False, IS_DICT),
         "ScalingConfig": Property(False, IS_DICT),
         "KmsKeyArn": Property(False, one_of(IS_STR, IS_DICT)),
+        "ManagedResourceTags": Property(False, IS_DICT),
     }
 
     CapacityProviderName: Intrinsicable[str] | None
@@ -1566,13 +1567,18 @@ class SamCapacityProvider(SamResourceMacro):
     InstanceRequirements: dict[str, Any] | None
     ScalingConfig: dict[str, Any] | None
     KmsKeyArn: Intrinsicable[str] | None
+    ManagedResourceTags: dict[str, Any] | None
 
     # Validation rules
     __validation_rules__ = [
         (
             ValidationRule.MUTUALLY_EXCLUSIVE,
             ["InstanceRequirements.AllowedTypes", "InstanceRequirements.ExcludedTypes"],
         ),
+        (
+            ValidationRule.MUTUALLY_EXCLUSIVE,
+            ["ManagedResourceTags.Propagate=True", "ManagedResourceTags.Tags"],
+        ),
     ]
 
     def to_cloudformation(self, **kwargs: Any) -> list[Resource]:
@@ -1600,6 +1606,9 @@ def to_cloudformation(self, **kwargs: Any) -> list[Resource]:
             ),
             scaling_config=model.ScalingConfig.dict(exclude_none=True) if model.ScalingConfig else None,
             kms_key_arn=passthrough_value(model.KmsKeyArn),
+            managed_resource_tags=(
+                model.ManagedResourceTags.dict(exclude_none=True) if model.ManagedResourceTags else None
+            ),
             depends_on=self.depends_on,
             resource_attributes=self.resource_attributes,
             passthrough_resource_attributes=self.get_passthrough_resource_attributes(),

samtranslator/plugins/globals/globals.py

@@ -114,6 +114,7 @@ class Globals:
             "ScalingConfig",
             "KmsKeyArn",
             "PropagateTags",
+            "ManagedResourceTags",
         ],
         SamResourceType.NetworkConnector.value: [
             "OperatorRole",