fix: reject invalid WebFunction and NetworkConnector inputs; Ref the WebFunction for revision and endpoint FunctionName

bin/sam-translate.py

@@ -123,8 +123,7 @@ def transform_template(input_file_path: Path, output_file_path: Path, stdout: bo
     except InvalidDocumentException as e:
         error_message = reduce(lambda message, error: message + " " + error.message, e.causes, e.message)
         LOG.error(error_message)
-        errors = (cause.message for cause in e.causes)
-        LOG.error(errors)
+        sys.exit(1)
 
 
 def deploy(template_file: Path) -> None:

samtranslator/internal/schema_source/sam-docs.json

@@ -978,7 +978,7 @@
       "EntryPoint": "The file that the runtime runs, relative to the root of the deployment package. AWS SAM passes it to the runtime in the reserved `AWS_LAMBDA_NODEJS_ENTRYPOINT` environment variable.  \nYou can't use this property when `EnvironmentVariables` is an intrinsic function.  \n*Type*: String  \n*Required*: No  \n*Default*: `index.js`  \n*CloudFormation compatibility*: This property is unique to AWS SAM and doesn't have an CloudFormation equivalent.",
       "EnvironmentVariables": "Key-value pairs to set as environment variables for the revision.  \n+ Keys must start with a letter and contain only letters, numbers, and underscores.\n+ Values must be non-empty and at most 4,096 characters.\n+ AWS SAM converts number and Boolean values to strings.\n*Type*: Map  \n*Required*: No  \n*CloudFormation compatibility*: This property is passed directly to the [`ServiceConfig.EnvironmentVariables`](https://docs.aws.amazon.com/AWSCloudFormation/latest/TemplateReference/aws-properties-lambda-webfunctionrevision-serviceconfig.html#cfn-lambda-webfunctionrevision-serviceconfig-environmentvariables) property of an `AWS::Lambda::WebFunctionRevision` resource.",
       "ExecutionRoleArn": "The ARN of the IAM role that the function assumes when it runs. If you don't specify a role, AWS SAM creates one with the `AWSLambdaBasicExecutionRole` managed policy and a logical ID of `<resource-logical-id>Role`.  \n*Type*: String  \n*Required*: No  \n*CloudFormation compatibility*: This property is passed directly to the [`ServiceConfig.ExecutionRoleArn`](https://docs.aws.amazon.com/AWSCloudFormation/latest/TemplateReference/aws-properties-lambda-webfunctionrevision-serviceconfig.html#cfn-lambda-webfunctionrevision-serviceconfig-executionrolearn) property of an `AWS::Lambda::WebFunctionRevision` resource.",
-      "FunctionName": "The name of the function. Function names must be unique within an account across all Regions. If you don't specify a name, AWS SAM generates one from the logical ID and the stack ID, at most 64 characters. Changing the name replaces the function, its revision, and its endpoint.  \n*Type*: String  \n*Required*: No  \n*CloudFormation compatibility*: This property is passed directly to the [`FunctionName`](https://docs.aws.amazon.com/AWSCloudFormation/latest/TemplateReference/aws-resource-lambda-webfunction.html#cfn-lambda-webfunction-functionname) property of the `AWS::Lambda::WebFunction`, `AWS::Lambda::WebFunctionRevision`, and `AWS::Lambda::WebFunctionEndpoint` resources.",
+      "FunctionName": "The name of the function. Function names must be unique within an account across all Regions. If you don't specify a name, AWS SAM generates one from the logical ID and the stack ID, at most 64 characters. Changing the name replaces the function, its revision, and its endpoint.  \n*Type*: String  \n*Required*: No  \n*CloudFormation compatibility*: This property is passed directly to the [`FunctionName`](https://docs.aws.amazon.com/AWSCloudFormation/latest/TemplateReference/aws-resource-lambda-webfunction.html#cfn-lambda-webfunction-functionname) property of an `AWS::Lambda::WebFunction` resource. The generated `AWS::Lambda::WebFunctionRevision` and `AWS::Lambda::WebFunctionEndpoint` resources reference the function with `Ref`.",
       "InlineCode": "Not supported for `AWS::Serverless::WebFunction`. Use [CodeUri](https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/sam-resource-webfunction.html#sam-webfunction-codeuri).  \n*Type*: String  \n*Required*: No  \n*CloudFormation compatibility*: This property is unique to AWS SAM and doesn't have an CloudFormation equivalent.",
       "KmsKeyArn": "The ARN of the AWS KMS key used to encrypt the revision's environment variables.  \n*Type*: String  \n*Required*: No  \n*CloudFormation compatibility*: This property is passed directly to the [`KmsKeyArn`](https://docs.aws.amazon.com/AWSCloudFormation/latest/TemplateReference/aws-resource-lambda-webfunctionrevision.html#cfn-lambda-webfunctionrevision-kmskeyarn) property of an `AWS::Lambda::WebFunctionRevision` resource.",
       "LoggingConfig": "The logging configuration for the revision, such as the log group and log levels.  \n*Type*: Map  \n*Required*: No  \n*CloudFormation compatibility*: This property is passed directly to the [`ServiceConfig.TelemetryConfig.LoggingConfig`](https://docs.aws.amazon.com/AWSCloudFormation/latest/TemplateReference/aws-properties-lambda-webfunctionrevision-telemetryconfig.html#cfn-lambda-webfunctionrevision-telemetryconfig-loggingconfig) property of an `AWS::Lambda::WebFunctionRevision` resource.",

samtranslator/model/__init__.py

@@ -243,6 +243,8 @@ def from_dict(cls, logical_id: str, resource_dict: dict[str, Any], relative_id:
             sam_plugins.act(LifeCycleEvents.before_transform_resource, logical_id, cls.resource_type, properties)
 
         for name, value in properties.items():
+            if not isinstance(name, str):
+                raise InvalidResourceException(logical_id, f"Property names must be strings, got {name!r}.")
             setattr(resource, name, value)
 
         if "DependsOn" in resource_dict:

samtranslator/model/intrinsics.py

@@ -147,8 +147,8 @@ def is_intrinsic(_input: Any) -> bool:
     """
 
     if _input is not None and isinstance(_input, dict) and len(_input) == 1:
-        key: str = next(iter(_input.keys()))
-        return key in {"Ref", "Condition"} or key.startswith("Fn::")
+        key = next(iter(_input.keys()))
+        return isinstance(key, str) and (key in {"Ref", "Condition"} or key.startswith("Fn::"))
 
     return False
 

samtranslator/model/sam_resources.py

@@ -1654,12 +1654,14 @@ class SamNetworkConnector(SamResourceMacro):
     __validation_rules__: list[Any] = []
 
     _VALID_COMPUTE_TYPES = ("MicroVm", "WebFunction")
+    _VALID_NETWORK_PROTOCOLS = ("IPv4", "DualStack")
 
     @cw_timer
     def to_cloudformation(self, **kwargs: Any) -> list[Resource]:
         # Run the ComputeTypes check first so its actionable message wins at transform time; the schema
         # (validate_before_transform / cfn-lint) enforces the same rules for `sam validate --lint`.
         self._validate_compute_types()
+        self._validate_network_protocol()
         self.validate_before_transform(
             schema_class=aws_serverless_networkconnector.Properties,
             collect_all_errors=True,
@@ -1693,12 +1695,26 @@ def _validate_compute_types(self) -> None:
                 f"'ComputeTypes' must contain exactly one value, got {len(self.ComputeTypes)}.",
             )
         for compute_type in self.ComputeTypes:
-            if isinstance(compute_type, str) and compute_type not in self._VALID_COMPUTE_TYPES:
+            if is_intrinsic(compute_type):
+                continue
+            if not isinstance(compute_type, str) or compute_type not in self._VALID_COMPUTE_TYPES:
                 raise InvalidResourceException(
                     self.logical_id,
-                    f"'ComputeTypes' values must be one of {list(self._VALID_COMPUTE_TYPES)}, got '{compute_type}'.",
+                    f"'ComputeTypes' values must be one of {list(self._VALID_COMPUTE_TYPES)}, got {compute_type!r}.",
                 )
 
+    def _validate_network_protocol(self) -> None:
+        # Name the allowed values; the schema's intrinsic-or-literal union reports "not a valid dict" instead.
+        network_protocol = self.VpcConfig.get("NetworkProtocol") if isinstance(self.VpcConfig, dict) else None
+        if network_protocol is None or is_intrinsic(network_protocol):
+            return
+        if network_protocol not in self._VALID_NETWORK_PROTOCOLS:
+            raise InvalidResourceException(
+                self.logical_id,
+                f"'VpcConfig.NetworkProtocol' must be one of {list(self._VALID_NETWORK_PROTOCOLS)}, "
+                f"got {network_protocol!r}.",
+            )
+
 
 class SamMicroVMImage(SamResourceMacro):
     """SAM MicrovmImage resource transformer"""
@@ -1776,6 +1792,24 @@ def to_cloudformation(self, **kwargs: Any) -> list[Resource]:
         return resources
 
 
+_NETWORK_CONNECTOR_TYPES = ("AWS::Serverless::NetworkConnector", "AWS::Lambda::NetworkConnector")
+
+
+def _parse_ref_or_get_att(value: Any) -> tuple[str, str | None] | None:
+    """Return (logical ID, attribute) for a Ref (attribute None) or Fn::GetAtt, otherwise None."""
+    if not is_intrinsic(value):
+        return None
+    if "Ref" in value:
+        target = value["Ref"]
+        return (target, None) if isinstance(target, str) else None
+    get_att = value.get("Fn::GetAtt")
+    if isinstance(get_att, str):
+        get_att = get_att.split(".", 1)
+    if isinstance(get_att, list) and len(get_att) == 2 and all(isinstance(x, str) for x in get_att):  # noqa: PLR2004
+        return get_att[0], get_att[1]
+    return None
+
+
 class SamWebFunction(SamResourceMacro):
     """
     SAM resource for AWS::Serverless::WebFunction.
@@ -1839,6 +1873,7 @@ def to_cloudformation(self, **kwargs: Any) -> list[Any]:
         """Transforms this SAM WebFunction into CloudFormation resources."""
         intrinsics_resolver: IntrinsicsResolver = kwargs["intrinsics_resolver"]
         self._validate(intrinsics_resolver)
+        self._validate_egress_network_connector_reference(kwargs.get("resource_resolver"))
 
         resources: list[Any] = []
 
@@ -1861,7 +1896,7 @@ def to_cloudformation(self, **kwargs: Any) -> list[Any]:
 
         return resources
 
-    def _validate(self, intrinsics_resolver: IntrinsicsResolver) -> None:  # noqa: PLR0912
+    def _validate(self, intrinsics_resolver: IntrinsicsResolver) -> None:  # noqa: PLR0912, PLR0915
         valid_auto_deploy_modes = ("LatestRevision", "Disabled")
 
         # AutoDeploymentMode is a SAM-only property enforced entirely at transform time and never passed
@@ -1957,6 +1992,8 @@ def _validate(self, intrinsics_resolver: IntrinsicsResolver) -> None:  # noqa: P
                     'revision IDs to weights (e.g., {"rev-id": 100}). Intrinsic functions may be used for '
                     "individual weight values.",
                 )
+            if not all(isinstance(revision_id, str) for revision_id in self.RevisionWeights):
+                raise InvalidResourceException(self.logical_id, "'RevisionWeights' keys must be revision ID strings.")
             _MIN_WEIGHT = 1
             _MAX_WEIGHT = 100
             weights = list(self.RevisionWeights.values())
@@ -1994,7 +2031,7 @@ def _validate(self, intrinsics_resolver: IntrinsicsResolver) -> None:  # noqa: P
             for key, value in self.EnvironmentVariables.items():
                 # Match the service key pattern exactly (leading letter required, no leading underscore) so
                 # invalid keys fail here with a clear message instead of at stack creation.
-                if not re.match(r"^[a-zA-Z][a-zA-Z0-9_]*$", key):
+                if not isinstance(key, str) or not re.match(r"^[a-zA-Z][a-zA-Z0-9_]*$", key):
                     raise InvalidResourceException(
                         self.logical_id,
                         f"Environment variable key '{key}' must match pattern '^[a-zA-Z][a-zA-Z0-9_]*$'.",
@@ -2018,25 +2055,44 @@ def _validate_egress_network_connector_arn(self) -> None:
         # value -- a network-connector ARN with a non-empty trailing segment -- without pinning the connector
         # name or qualifier grammar (SAM shouldn't guess at the service's naming/versioning charset, or it
         # would hard-fail otherwise-valid ARNs). Intrinsics (e.g. a Fn::GetAtt on the connector's
-        # LatestVersionArn) are passed through and validated by the service.
+        # LatestVersionArn) are passed through; references to connectors in the same template are checked in
+        # _validate_egress_network_connector_reference.
+        value = self.EgressNetworkConnectorArn
+        if value is None or is_intrinsic(value):
+            return
         network_connector_arn_pattern = r"^arn:[^:]+:lambda:[^:]*:[^:]*:network-connector:[^:]+:[^:]+$"
-        if isinstance(self.EgressNetworkConnectorArn, str) and not re.match(
-            network_connector_arn_pattern, self.EgressNetworkConnectorArn
-        ):
+        if not isinstance(value, str) or not re.match(network_connector_arn_pattern, value):
             raise InvalidResourceException(
                 self.logical_id,
                 "'EgressNetworkConnectorArn' must be a version-qualified network-connector ARN "
-                "(ending in ':<version>', e.g. the connector's LatestVersionArn).",
+                "('arn:<partition>:lambda:<region>:<account>:network-connector:<name>:<version>', e.g. the "
+                f"connector's LatestVersionArn), got {value!r}.",
             )
 
-    def _construct_function_name(self) -> Any:
-        """Return the function name shared by the WebFunction, its Revision, and its Endpoint.
+    def _validate_egress_network_connector_reference(self, resource_resolver: ResourceResolver | None) -> None:
+        """Check a Ref or Fn::GetAtt on a network connector defined in the same template.
 
-        Ref on AWS::Lambda::WebFunction resolves to the function ARN, not the name, so the Revision and
-        Endpoint must reuse this exact value rather than Ref the function. Referencing it with Ref would set
-        their FunctionName to the ARN, which the service rejects (the endpoint/revision then cannot be
-        deleted: the stored ARN does not match the name parsed from it).
+        Ref and the Arn attribute return the unversioned connector ARN, which the service rejects, so only
+        LatestVersionArn is allowed.
         """
+        reference = _parse_ref_or_get_att(self.EgressNetworkConnectorArn)
+        if resource_resolver is None or reference is None:
+            return
+        target, attribute = reference
+        connector = resource_resolver.get_resource_by_logical_id(target)
+        if not isinstance(connector, dict) or connector.get("Type") not in _NETWORK_CONNECTOR_TYPES:
+            return
+
+        if attribute != "LatestVersionArn":
+            used = f"!Ref {target}" if attribute is None else f"!GetAtt {target}.{attribute}"
+            raise InvalidResourceException(
+                self.logical_id,
+                f"'EgressNetworkConnectorArn' must be a version-qualified network-connector ARN; {used} "
+                f"doesn't return one. Use !GetAtt {target}.LatestVersionArn instead.",
+            )
+
+    def _construct_function_name(self) -> Any:
+        """Return the FunctionName for the AWS::Lambda::WebFunction: the user value or a generated name."""
         if self.FunctionName:
             return self.FunctionName
         # Generate a name bounded by construction to the 64-char AWS::Lambda::WebFunction FunctionName
@@ -2081,7 +2137,9 @@ def _construct_revision(self, execution_role_arn: Any) -> LambdaWebFunctionRevis
         revision = LambdaWebFunctionRevision(
             f"{self.logical_id}Revision", attributes=self.get_passthrough_resource_attributes()
         )
-        revision.FunctionName = self._construct_function_name()
+        # Ref the function (its ARN) instead of repeating the name, like AWS::Lambda::Version does, so a
+        # FunctionName change only has to be made on the AWS::Lambda::WebFunction.
+        revision.FunctionName = ref(self.logical_id)
         revision.Description = self.RevisionDescription
         revision.KmsKeyArn = self.KmsKeyArn
         revision.BuildConfig = self._construct_build_config()
@@ -2092,7 +2150,7 @@ def _construct_endpoint(self, revision: LambdaWebFunctionRevision) -> LambdaWebF
         endpoint = LambdaWebFunctionEndpoint(
             f"{self.logical_id}Endpoint", attributes=self.get_passthrough_resource_attributes()
         )
-        endpoint.FunctionName = self._construct_function_name()
+        endpoint.FunctionName = ref(self.logical_id)
         endpoint.EndpointName = self.EndpointName or "prod"
         endpoint.EndpointType = self.EndpointType or "HomeRegion"
         # AuthType is required and has no default (validated in _validate); pass the customer's value through.
@@ -2196,7 +2254,7 @@ def _construct_region_list(self) -> list[Intrinsicable[str]] | None:
         if unknown_keys:
             raise InvalidResourceException(
                 self.logical_id,
-                f"'ReplicaRegions' contains unrecognized keys: {sorted(unknown_keys)}. Valid keys are: IncludeRegions, ExcludeRegions.",
+                f"'ReplicaRegions' contains unrecognized keys: {sorted(map(str, unknown_keys))}. Valid keys are: IncludeRegions, ExcludeRegions.",
             )
 
         # ExcludeRegions is only meaningful as a filter over an explicit inclusion set. Without IncludeRegions
@@ -2215,6 +2273,10 @@ def _construct_region_list(self) -> list[Intrinsicable[str]] | None:
                 "'IncludeRegions' and 'ExcludeRegions' do not support top-level intrinsic functions.",
             )
 
+        for key, value in (("IncludeRegions", include_regions), ("ExcludeRegions", exclude_regions)):
+            if not isinstance(value, list):
+                raise InvalidResourceException(self.logical_id, f"'ReplicaRegions.{key}' must be a list.")
+
         # Separate string entries from intrinsics in both lists
         include_strings = [r for r in include_regions if isinstance(r, str)]
         include_intrinsics = [r for r in include_regions if not isinstance(r, str)]

samtranslator/parser/parser.py

@@ -26,7 +26,8 @@ def parse(self, sam_template: dict[str, Any], parameter_values: dict[str, Any],
     def validate_datatypes(sam_template):  # type: ignore[no-untyped-def]
         """Validates the datatype within the template"""
         if (
-            "Resources" not in sam_template
+            not isinstance(sam_template, dict)
+            or "Resources" not in sam_template
             or not isinstance(sam_template["Resources"], dict)
             or not sam_template["Resources"]
         ):

samtranslator/plugins/globals/globals.py

@@ -314,6 +314,12 @@ def _parse(self, globals_dict):  # type: ignore[no-untyped-def]
             raise InvalidGlobalsSectionException(self._KEYWORD, "It must be a non-empty dictionary")
 
         for section_name, properties in globals_dict.items():
+            if not isinstance(section_name, str):
+                raise InvalidGlobalsSectionException(
+                    self._KEYWORD,
+                    f"'{section_name}' is not supported. "
+                    f"Must be one of the following values - {self.supported_resource_section_names}",
+                )
             resource_type = self._make_resource_type(section_name)  # type: ignore[no-untyped-call]
 
             if resource_type not in self.supported_properties: