Merge branch 'develop' into lmi-integration-tests

Author: vicheey

samtranslator/__init__.py

@@ -1 +1 @@
-__version__ = "1.105.0"
+__version__ = "1.106.0"

samtranslator/model/sam_resources.py

@@ -293,7 +293,7 @@ def resources_to_link(self, resources: Dict[str, Any]) -> Dict[str, Any]:
             raise InvalidResourceException(self.logical_id, e.message) from e
 
     @cw_timer
-    def to_cloudformation(self, **kwargs):  # type: ignore[no-untyped-def] # noqa: PLR0915
+    def to_cloudformation(self, **kwargs):  # type: ignore[no-untyped-def] # noqa: PLR0915, PLR0912
         """Returns the Lambda function, role, and event resources to which this SAM Function corresponds.
 
         :param dict kwargs: already-converted resources that may need to be modified when converting this \
@@ -388,16 +388,25 @@ def to_cloudformation(self, **kwargs):  # type: ignore[no-untyped-def] # noqa: P
         managed_policy_map = kwargs.get("managed_policy_map", {})
         get_managed_policy_map = kwargs.get("get_managed_policy_map")
 
-        execution_role = None
+        execution_role = self._construct_role(
+            managed_policy_map,
+            event_invoke_policies,
+            intrinsics_resolver,
+            get_managed_policy_map,
+        )
+
         if lambda_function.Role is None:
-            execution_role = self._construct_role(
-                managed_policy_map,
-                event_invoke_policies,
-                intrinsics_resolver,
-                get_managed_policy_map,
-            )
             lambda_function.Role = execution_role.get_runtime_attr("arn")
             resources.append(execution_role)
+        elif is_intrinsic_if(lambda_function.Role):
+            role_changes = self._make_lambda_role(lambda_function, intrinsics_resolver, execution_role)
+
+            if role_changes["lambda_role_value"] is not None:
+                lambda_function.Role = role_changes["lambda_role_value"]
+                resources.append(role_changes["iam_role_resource"])
+
+            if role_changes["new_condition"] is not None:
+                conditions.update(role_changes["new_condition"])
 
         try:
             resources += self._generate_event_resources(
@@ -415,6 +424,54 @@ def to_cloudformation(self, **kwargs):  # type: ignore[no-untyped-def] # noqa: P
 
         return resources
 
+    def _make_lambda_role(
+        self,
+        lambda_function: LambdaFunction,
+        intrinsics_resolver: IntrinsicsResolver,
+        execution_role: IAMRole,
+    ) -> Dict[str, Any]:
+        """
+        Analyzes lambda role requirements and returns the changes needed.
+
+        Returns:
+            Dict containing:
+                - 'lambda_role_value': Any - value to set for lambda_function.Role
+                - 'new_condition': Dict|None - new condition to add to conditions dict
+                - 'iam_role_resource' : IAMRole - IAM Role used for Lambda execution
+        """
+        lambda_role = lambda_function.Role
+        execution_role_arn = execution_role.get_runtime_attr("arn")
+
+        lambda_role_value = None
+        new_condition = None
+
+        # We need to create and if else condition here
+        role_resolved_value = intrinsics_resolver.resolve_parameter_refs(lambda_role)
+        role_condition, role_if, role_else = role_resolved_value.get("Fn::If")
+
+        if is_intrinsic_no_value(role_if) and is_intrinsic_no_value(role_else):
+            lambda_role_value = execution_role_arn
+
+        # first value is none so we should create condition ? create : [2]
+        # create a condition for IAM role to only create on if case
+        elif is_intrinsic_no_value(role_if):
+            lambda_role_value = make_conditional(role_condition, execution_role_arn, role_else)
+            execution_role.set_resource_attribute("Condition", role_condition)
+
+        # second value is none so we should create condition ? [1] : create
+        # create a condition for IAM role to only create on else case
+        # with top level condition that negates the condition passed
+        elif is_intrinsic_no_value(role_else):
+            lambda_role_value = make_conditional(role_condition, role_if, execution_role_arn)
+            execution_role.set_resource_attribute("Condition", f"NOT{role_condition}")
+            new_condition = {f"NOT{role_condition}": make_not_conditional(role_condition)}
+
+        return {
+            "lambda_role_value": lambda_role_value,
+            "new_condition": new_condition,
+            "iam_role_resource": execution_role,
+        }
+
     def _construct_event_invoke_config(  # noqa: PLR0913
         self,
         function_name: str,

tests/model/test_sam_resources.py

@@ -741,6 +741,113 @@ def test_function_datasource_set_with_none():
     assert none_datasource
 
 
+class TestSamFunctionRoleResolver(TestCase):
+    """
+    Tests for resolving IAM role property values in SamFunction
+    """
+
+    def setUp(self):
+        self.function = SamFunction("foo")
+        self.function.CodeUri = "s3://foobar/foo.zip"
+        self.function.Runtime = "foo"
+        self.function.Handler = "bar"
+        self.kwargs = {
+            "intrinsics_resolver": IntrinsicsResolver({}),
+            "event_resources": [],
+            "managed_policy_map": {},
+            "resource_resolver": ResourceResolver({}),
+            "conditions": {"Conditions": {}},
+        }
+
+    def test_role_none_creates_execution_role(self):
+        self.function.Role = None
+        cfn_resources = self.function.to_cloudformation(**self.kwargs)
+        generated_roles = [x for x in cfn_resources if isinstance(x, IAMRole)]
+
+        self.assertEqual(len(generated_roles), 1)  # Should create execution role
+
+    def test_role_explicit_arn_no_execution_role(self):
+        test_role = "arn:aws:iam::123456789012:role/existing-role"
+        self.function.Role = test_role
+
+        cfn_resources = self.function.to_cloudformation(**self.kwargs)
+        generated_roles = [x for x in cfn_resources if isinstance(x, IAMRole)]
+        lambda_function = next(r for r in cfn_resources if r.resource_type == "AWS::Lambda::Function")
+
+        self.assertEqual(len(generated_roles), 0)  # Should not create execution role
+        self.assertEqual(lambda_function.Role, test_role)
+
+    def test_role_fn_if_no_aws_no_value_keeps_original(self):
+        role_conditional = {
+            "Fn::If": ["Condition", "arn:aws:iam::123456789012:role/existing-role", {"Ref": "iamRoleArn"}]
+        }
+        self.function.Role = role_conditional
+
+        kwargs = dict(self.kwargs)
+        kwargs["conditions"] = {"Condition": True}
+
+        cfn_resources = self.function.to_cloudformation(**self.kwargs)
+        generated_roles = [x for x in cfn_resources if isinstance(x, IAMRole)]
+        lambda_function = next(r for r in cfn_resources if r.resource_type == "AWS::Lambda::Function")
+
+        # Should not create a role if a role is passed in for both cases
+        self.assertEqual(len(generated_roles), 0)
+        self.assertEqual(lambda_function.Role, role_conditional)
+
+    def test_role_fn_if_both_no_value_creates_execution_role(self):
+        role_conditional = {"Fn::If": ["Condition", {"Ref": "AWS::NoValue"}, {"Ref": "AWS::NoValue"}]}
+        self.function.Role = role_conditional
+
+        kwargs = dict(self.kwargs)
+        kwargs["conditions"] = {"Condition": True}
+
+        cfn_resources = self.function.to_cloudformation(**self.kwargs)
+        generated_roles = [x for x in cfn_resources if isinstance(x, IAMRole)]
+
+        self.assertEqual(len(generated_roles), 1)
+
+    def test_role_fn_if_first_no_value_creates_conditional_role(self):
+        role_conditional = {"Fn::If": ["Condition", {"Ref": "AWS::NoValue"}, {"Ref": "iamRoleArn"}]}
+        self.function.Role = role_conditional
+
+        kwargs = dict(self.kwargs)
+        kwargs["conditions"] = {"Condition": True}
+
+        cfn_resources = self.function.to_cloudformation(**self.kwargs)
+        generated_roles = [x for x in cfn_resources if isinstance(x, IAMRole)]
+        lambda_function = next(r for r in cfn_resources if r.resource_type == "AWS::Lambda::Function")
+
+        self.assertEqual(len(generated_roles), 1)
+        self.assertEqual(
+            lambda_function.Role, {"Fn::If": ["Condition", {"Fn::GetAtt": ["fooRole", "Arn"]}, {"Ref": "iamRoleArn"}]}
+        )
+
+    def test_role_fn_if_second_no_value_creates_conditional_role(self):
+        role_conditional = {"Fn::If": ["Condition", {"Ref": "iamRoleArn"}, {"Ref": "AWS::NoValue"}]}
+        self.function.Role = role_conditional
+
+        kwargs = dict(self.kwargs)
+        kwargs["conditions"] = {"Condition": True}
+
+        cfn_resources = self.function.to_cloudformation(**self.kwargs)
+        generated_roles = [x for x in cfn_resources if isinstance(x, IAMRole)]
+        lambda_function = next(r for r in cfn_resources if r.resource_type == "AWS::Lambda::Function")
+
+        self.assertEqual(len(generated_roles), 1)
+        self.assertEqual(
+            lambda_function.Role, {"Fn::If": ["Condition", {"Ref": "iamRoleArn"}, {"Fn::GetAtt": ["fooRole", "Arn"]}]}
+        )
+
+    def test_role_get_att_no_execution_role(self):
+        role_get_att = {"Fn::GetAtt": ["MyCustomRole", "Arn"]}
+        self.function.Role = role_get_att
+
+        cfn_resources = self.function.to_cloudformation(**self.kwargs)
+        lambda_function = next(r for r in cfn_resources if r.resource_type == "AWS::Lambda::Function")
+
+        self.assertEqual(lambda_function.Role, role_get_att)
+
+
 class TestSamCapacityProvider(TestCase):
     """Tests for SamCapacityProvider"""
 

tests/translator/input/error_function_invalid_iam_role_type.yaml

@@ -0,0 +1,8 @@
+Resources:
+  MinimalFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      CodeUri: s3://sam-demo-bucket/hello.zip
+      Handler: hello.handler
+      Runtime: python3.10
+      Role: 2

tests/translator/input/function_with_conditional_iam_role_else.yaml

@@ -0,0 +1,20 @@
+Parameters:
+  iamRoleArn:
+    Type: String
+    Description: The ARN of an IAM role to use as this function's execution role.
+      If a role isn't specified, one is created for you with a logical ID of <function-logical-id>Role.
+
+Conditions:
+  CreateRole: !Not [!Equals ['', !Ref iamRoleArn]]
+
+Resources:
+  MinimalFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      CodeUri: s3://sam-demo-bucket/hello.zip
+      Handler: hello.handler
+      Runtime: python3.10
+      Role: !If
+      - CreateRole
+      - !Ref "AWS::NoValue"
+      - !Ref "iamRoleArn"

tests/translator/input/function_with_conditional_iam_role_if.yaml

@@ -0,0 +1,20 @@
+Parameters:
+  iamRoleArn:
+    Type: String
+    Description: The ARN of an IAM role to use as this function's execution role.
+      If a role isn't specified, one is created for you with a logical ID of <function-logical-id>Role.
+
+Conditions:
+  RoleExists: !Not [!Equals ['', !Ref iamRoleArn]]
+
+Resources:
+  MinimalFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      CodeUri: s3://sam-demo-bucket/hello.zip
+      Handler: hello.handler
+      Runtime: python3.10
+      Role: !If
+      - RoleExists
+      - !Ref "iamRoleArn"
+      - !Ref "AWS::NoValue"

tests/translator/output/aws-cn/error_function_invalid_iam_role_type.json

@@ -0,0 +1,14 @@
+{
+  "_autoGeneratedBreakdownErrorMessage": [
+    "Invalid Serverless Application Specification document. ",
+    "Number of errors found: 1. ",
+    "Resource with id [MinimalFunction] is invalid. ",
+    "Property 'Role' should be a string."
+  ],
+  "errorMessage": "Invalid Serverless Application Specification document. Number of errors found: 1. Resource with id [MinimalFunction] is invalid. Property 'Role' should be a string.",
+  "errors": [
+    {
+      "errorMessage": "Resource with id [MinimalFunction] is invalid. Property 'Role' should be a string."
+    }
+  ]
+}

tests/translator/output/aws-cn/function_with_conditional_iam_role_else.json

@@ -0,0 +1,86 @@
+{
+  "Conditions": {
+    "CreateRole": {
+      "Fn::Not": [
+        {
+          "Fn::Equals": [
+            "",
+            {
+              "Ref": "iamRoleArn"
+            }
+          ]
+        }
+      ]
+    }
+  },
+  "Parameters": {
+    "iamRoleArn": {
+      "Description": "The ARN of an IAM role to use as this function's execution role. If a role isn't specified, one is created for you with a logical ID of <function-logical-id>Role.",
+      "Type": "String"
+    }
+  },
+  "Resources": {
+    "MinimalFunction": {
+      "Properties": {
+        "Code": {
+          "S3Bucket": "sam-demo-bucket",
+          "S3Key": "hello.zip"
+        },
+        "Handler": "hello.handler",
+        "Role": {
+          "Fn::If": [
+            "CreateRole",
+            {
+              "Fn::GetAtt": [
+                "MinimalFunctionRole",
+                "Arn"
+              ]
+            },
+            {
+              "Ref": "iamRoleArn"
+            }
+          ]
+        },
+        "Runtime": "python3.10",
+        "Tags": [
+          {
+            "Key": "lambda:createdBy",
+            "Value": "SAM"
+          }
+        ]
+      },
+      "Type": "AWS::Lambda::Function"
+    },
+    "MinimalFunctionRole": {
+      "Condition": "CreateRole",
+      "Properties": {
+        "AssumeRolePolicyDocument": {
+          "Statement": [
+            {
+              "Action": [
+                "sts:AssumeRole"
+              ],
+              "Effect": "Allow",
+              "Principal": {
+                "Service": [
+                  "lambda.amazonaws.com"
+                ]
+              }
+            }
+          ],
+          "Version": "2012-10-17"
+        },
+        "ManagedPolicyArns": [
+          "arn:aws-cn:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole"
+        ],
+        "Tags": [
+          {
+            "Key": "lambda:createdBy",
+            "Value": "SAM"
+          }
+        ]
+      },
+      "Type": "AWS::IAM::Role"
+    }
+  }
+}