fix: surface user-facing errors for malformed Fn::If and malformed CodeUri

Three customer templates were crashing the SAM transform with unhandled
exceptions that CloudFormation surfaced as the opaque message
"Transform AWS::Serverless-2016-10-31 failed with: Internal transform
failure.", leaving the customer with no indication of which resource or
property was wrong.

1) AWS::Serverless::Function.Role supports Fn::If so a user can switch
   between an existing IAM role and a SAM-generated one. _make_lambda_role
   went straight from `is_intrinsic_if(role)` to a 3-tuple unpack:
       role_condition, role_if, role_else = role_resolved_value.get("Fn::If")
   A malformed 2- or 4-element list therefore raised
       ValueError: not enough values to unpack (expected 3, got 2)
   Wrap with validate_intrinsic_if_items (already used in
   resource_policies.py and preferences/deployment_preference_collection.py)
   and raise InvalidResourceException(self.logical_id, ...).

2) EventInvokeConfig.DestinationConfig.{OnSuccess,OnFailure}.Destination
   also supports Fn::If. _get_or_make_condition indexed dest_list[1] /
   dest_list[2] without arity validation, so a 2-element list raised
       IndexError: list index out of range
   Same guard using validate_intrinsic_if_items is applied here.

3) parse_s3_uri calls urllib.parse.urlparse, which has validated
   bracketed-host segments against the IPv6 grammar since the
   CVE-2024-11168 fix (Python 3.9.17+/3.10.12+/3.11.4+/3.12/3.13/3.14).
   Unresolved CDK tokens like "s3://[TOKEN.25]/key" now raise:
       ValueError: "TOKEN.25" does not appear to be an IPv4 or IPv6 address
   Catch ValueError and return None so construct_s3_location_object raises
   the existing friendly "CodeUri is not a valid S3 Uri..."
   InvalidResourceException with the logical id.

Tests:
  - tests/translator/input/error_function_role_fn_if_malformed.{yaml,json}
  - tests/translator/input/error_function_codeuri_unresolved_token.{yaml,json}
  - tests/translator/input/error_function_event_destination_fn_if_malformed.{yaml,json}
    (auto-discovered by test_transform_invalid_document)
  - tests/model/s3_utils/test_uri_parser.py: parse_s3_uri returns None on
    malformed bracketed netloc; construct_s3_location_object raises
    InvalidResourceException with the resource logical id
  - tests/model/test_sam_resources.py: _make_lambda_role raises
    InvalidResourceException on too-few and too-many Fn::If items

Full suite: 4503 passed, 0 failed.

Author: roger-zhangg

samtranslator/model/s3_utils/uri_parser.py

@@ -14,7 +14,17 @@ def parse_s3_uri(uri: Any) -> dict[str, Any] | None:
     if not isinstance(uri, str):
         return None
 
-    url = urlparse(uri)
+    try:
+        url = urlparse(uri)
+    except ValueError:
+        # Python's urllib validates bracketed host segments ("[...]") against
+        # RFC 3986 IPv6/IPv4 grammars since the CVE-2024-11168 fix. Unresolved
+        # CDK tokens (for example "s3://[TOKEN.25]/key") or other malformed
+        # URIs therefore raise ValueError here. Treating the input as "not a
+        # valid S3 URI" lets the caller raise the existing, user-friendly
+        # InvalidResourceException with the resource logical id and property
+        # name, instead of surfacing an opaque "Internal transform failure".
+        return None
     query = parse_qs(url.query)
 
     if url.scheme == "s3" and url.netloc and url.path:

samtranslator/model/sam_resources.py

@@ -102,6 +102,7 @@
     make_conditional,
     make_not_conditional,
     ref,
+    validate_intrinsic_if_items,
 )
 from samtranslator.model.lambda_ import (
     LAMBDA_TRACING_CONFIG_DISABLED,
@@ -452,7 +453,17 @@ def _make_lambda_role(
 
         # 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_items = role_resolved_value.get("Fn::If")
+        try:
+            validate_intrinsic_if_items(if_items)
+        except ValueError as e:
+            # Surface a user-facing error instead of letting a ValueError propagate
+            # all the way out of the transform macro as "Internal transform failure".
+            raise InvalidResourceException(
+                self.logical_id,
+                f"Malformed 'Role' property: {e!s}.",
+            ) from e
+        role_condition, role_if, role_else = if_items
 
         if is_intrinsic_no_value(role_if) and is_intrinsic_no_value(role_else):
             lambda_role_value = execution_role_arn
@@ -627,6 +638,15 @@ def _get_or_make_condition(self, destination: Any, logical_id: str, conditions:
             return None, None
         if is_intrinsic_if(destination):
             dest_list = destination.get("Fn::If")
+            try:
+                validate_intrinsic_if_items(dest_list)
+            except ValueError as e:
+                # Surface a user-facing error instead of letting an IndexError
+                # propagate out of the transform macro as "Internal transform failure".
+                raise InvalidResourceException(
+                    logical_id,
+                    f"Malformed 'Destination' property: {e!s}.",
+                ) from e
             if is_intrinsic_no_value(dest_list[1]) and is_intrinsic_no_value(dest_list[2]):
                 return None, None
             if is_intrinsic_no_value(dest_list[1]):

tests/model/s3_utils/__init__.py

@@ -0,0 +1,49 @@
+"""Unit tests for :mod:`samtranslator.model.s3_utils.uri_parser`."""
+
+from unittest import TestCase
+
+from samtranslator.model.exceptions import InvalidResourceException
+from samtranslator.model.s3_utils.uri_parser import construct_s3_location_object, parse_s3_uri
+
+
+class TestParseS3Uri(TestCase):
+    def test_valid_s3_uri(self):
+        self.assertEqual(
+            parse_s3_uri("s3://bucket/key"),
+            {"Bucket": "bucket", "Key": "key"},
+        )
+
+    def test_valid_s3_uri_with_version(self):
+        self.assertEqual(
+            parse_s3_uri("s3://bucket/key?versionId=abcdef"),
+            {"Bucket": "bucket", "Key": "key", "Version": "abcdef"},
+        )
+
+    def test_non_s3_scheme_returns_none(self):
+        self.assertIsNone(parse_s3_uri("https://example.com/key"))
+
+    def test_non_string_returns_none(self):
+        self.assertIsNone(parse_s3_uri({"Bucket": "b", "Key": "k"}))
+        self.assertIsNone(parse_s3_uri(None))
+
+    def test_unresolved_cdk_token_returns_none(self):
+        """Bracketed host segments that are not valid IPv4/IPv6 raise ValueError
+        from urllib (see CVE-2024-11168); parse_s3_uri should treat the input as
+        "not a valid S3 URI" and return None so callers can raise a friendly
+        InvalidResourceException instead of crashing the transform.
+        """
+        self.assertIsNone(parse_s3_uri("s3://[TOKEN.25]/my/key"))
+        self.assertIsNone(parse_s3_uri("https://[TOKEN.25]/path"))
+        self.assertIsNone(parse_s3_uri("s3://bucket-[TOKEN.25]/key"))
+
+
+class TestConstructS3LocationObjectWithMalformedUri(TestCase):
+    """Verify that the top-level helper raises InvalidResourceException with the
+    logical id instead of letting the underlying urllib ValueError propagate.
+    """
+
+    def test_unresolved_cdk_token_raises_invalid_resource_exception(self):
+        with self.assertRaises(InvalidResourceException) as ctx:
+            construct_s3_location_object("s3://[TOKEN.25]/my/key", "MyFunction", "CodeUri")
+        self.assertIn("MyFunction", str(ctx.exception))
+        self.assertIn("'CodeUri' is not a valid S3 Uri", str(ctx.exception))

tests/model/s3_utils/test_uri_parser.py

@@ -2,6 +2,7 @@
 from unittest.mock import patch
 
 import pytest
+from parameterized import parameterized
 from samtranslator.intrinsics.resolver import IntrinsicsResolver
 from samtranslator.model import InvalidResourceException, ResourceResolver
 from samtranslator.model.apigateway import ApiGatewayDeployment, ApiGatewayRestApi, ApiGatewayStage
@@ -847,6 +848,51 @@ def test_role_get_att_no_execution_role(self):
 
         self.assertEqual(lambda_function.Role, role_get_att)
 
+    @parameterized.expand(
+        [
+            # 2-arg Fn::If (missing false value)
+            ({"Fn::If": ["Condition", "arn:aws:iam::123456789012:role/existing-role"]},),
+            # 4-arg Fn::If (extra value)
+            ({"Fn::If": ["Condition", "role_a", "role_b", "role_c"]},),
+        ]
+    )
+    def test_role_fn_if_invalid_arity_raises_invalid_resource_exception(self, malformed_role):
+        """Fn::If must have exactly 3 items: [Condition, TrueValue, FalseValue].
+        Any other arity used to crash SAM-T with a raw ValueError/IndexError,
+        which CloudFormation surfaced as "Internal transform failure".
+        It must now raise a user-facing InvalidResourceException instead.
+        """
+        self.function.Role = malformed_role
+
+        with pytest.raises(InvalidResourceException) as excinfo:
+            self.function.to_cloudformation(**self.kwargs)
+
+        msg = str(excinfo.value)
+        self.assertIn("foo", msg)  # logical id
+        self.assertIn("Role", msg)
+        self.assertIn("Fn::If", msg)
+
+    @parameterized.expand(
+        [
+            # 2-arg Fn::If (missing false value)
+            ({"Fn::If": ["SomeCondition", "queue-arn"]},),
+            # 4-arg Fn::If (extra value)
+            ({"Fn::If": ["SomeCondition", "a", "b", "c"]},),
+        ]
+    )
+    def test_destination_fn_if_invalid_arity_raises_invalid_resource_exception(self, malformed):
+        """EventInvokeConfig.DestinationConfig.{OnSuccess,OnFailure}.Destination
+        also supports Fn::If. Any list arity other than 3 previously crashed the
+        transform with IndexError in _get_or_make_condition.
+        """
+        with pytest.raises(InvalidResourceException) as excinfo:
+            self.function._get_or_make_condition(malformed, "DestLogicalId", {})
+
+        msg = str(excinfo.value)
+        self.assertIn("DestLogicalId", msg)
+        self.assertIn("Destination", msg)
+        self.assertIn("Fn::If", msg)
+
 
 class TestSamCapacityProvider(TestCase):
     """Tests for SamCapacityProvider"""

tests/model/test_sam_resources.py

@@ -0,0 +1,14 @@
+Resources:
+  MyFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      Runtime: python3.12
+      Handler: index.handler
+      # CDK sometimes synthesizes unresolved Token placeholders like
+      # "${Token[Bucket.Name.25]}" / "[TOKEN.25]" into string properties.
+      # Python's urllib validates bracketed-host segments against the IPv6
+      # grammar since CVE-2024-11168, so urlparse("s3://[TOKEN.25]/...")
+      # raises `ValueError: 'TOKEN.25' does not appear to be an IPv4 or IPv6
+      # address`. Previously that propagated out of the transform macro as
+      # the opaque "Internal transform failure" message.
+      CodeUri: s3://[TOKEN.25]/my/key

tests/translator/input/error_function_codeuri_unresolved_token.yaml

@@ -0,0 +1,31 @@
+Parameters:
+  IsProd:
+    Type: String
+    Default: 'true'
+
+Conditions:
+  IsProdCondition: !Equals [!Ref IsProd, 'true']
+
+Resources:
+  MyQueue:
+    Type: AWS::SQS::Queue
+
+  MyFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      Runtime: python3.12
+      Handler: index.handler
+      InlineCode: |
+        def handler(event, context):
+            return 'hi'
+      EventInvokeConfig:
+        MaximumRetryAttempts: 2
+        DestinationConfig:
+          OnSuccess:
+            Type: SQS
+            # Malformed Fn::If: only 2 elements instead of [Condition, TrueValue, FalseValue].
+            # Previously this crashed the transform with an IndexError in
+            # _get_or_make_condition, which CloudFormation surfaced as
+            # "Internal transform failure". It must now raise a user-facing
+            # InvalidResourceException naming the offending resource.
+            Destination: !If [IsProdCondition, !GetAtt MyQueue.Arn]

tests/translator/input/error_function_event_destination_fn_if_malformed.yaml

@@ -0,0 +1,32 @@
+Parameters:
+  IsProd:
+    Type: String
+    Default: 'true'
+
+Conditions:
+  IsProdCondition: !Equals [!Ref IsProd, 'true']
+
+Resources:
+  ExistingRole:
+    Type: AWS::IAM::Role
+    Properties:
+      AssumeRolePolicyDocument:
+        Version: '2012-10-17'
+        Statement:
+        - Effect: Allow
+          Principal: {Service: lambda.amazonaws.com}
+          Action: sts:AssumeRole
+
+  MyFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      Runtime: python3.12
+      Handler: index.handler
+      InlineCode: |
+        def handler(event, context):
+            return 'hi'
+      # Malformed Fn::If: only 2 elements instead of [Condition, TrueValue, FalseValue].
+      # Previously this crashed the transform with a Python
+      # `ValueError: not enough values to unpack (expected 3, got 2)`,
+      # which CloudFormation surfaced as "Internal transform failure".
+      Role: !If [IsProdCondition, !GetAtt ExistingRole.Arn]

tests/translator/input/error_function_role_fn_if_malformed.yaml

@@ -0,0 +1,14 @@
+{
+  "_autoGeneratedBreakdownErrorMessage": [
+    "Invalid Serverless Application Specification document. ",
+    "Number of errors found: 1. ",
+    "Resource with id [MyFunction] is invalid. ",
+    "'CodeUri' is not a valid S3 Uri of the form 's3://bucket/key' with optional versionId query parameter."
+  ],
+  "errorMessage": "Invalid Serverless Application Specification document. Number of errors found: 1. Resource with id [MyFunction] is invalid. 'CodeUri' is not a valid S3 Uri of the form 's3://bucket/key' with optional versionId query parameter.",
+  "errors": [
+    {
+      "errorMessage": "Resource with id [MyFunction] is invalid. 'CodeUri' is not a valid S3 Uri of the form 's3://bucket/key' with optional versionId query parameter."
+    }
+  ]
+}

tests/translator/output/error_function_codeuri_unresolved_token.json

@@ -0,0 +1,14 @@
+{
+  "_autoGeneratedBreakdownErrorMessage": [
+    "Invalid Serverless Application Specification document. ",
+    "Number of errors found: 1. ",
+    "Resource with id [MyFunctionEventInvokeConfig] is invalid. ",
+    "Malformed 'Destination' property: Fn::If requires 3 arguments."
+  ],
+  "errorMessage": "Invalid Serverless Application Specification document. Number of errors found: 1. Resource with id [MyFunctionEventInvokeConfig] is invalid. Malformed 'Destination' property: Fn::If requires 3 arguments.",
+  "errors": [
+    {
+      "errorMessage": "Resource with id [MyFunctionEventInvokeConfig] is invalid. Malformed 'Destination' property: Fn::If requires 3 arguments."
+    }
+  ]
+}