[Networking] Prevent duplicate security group rules in shared storage security group.

The deduplication is needed because when duplicate rules are requested,
the CFN handler responsible for their creation goes into a path
that is more susceptible to eventual consistency issue.
When such issue occur, cluster creation may fail.

As part of this change, we also scoped down the SG rules that allow
traffic to EFS and FSx by applying to head and compute nodes
the same network access we had for login nodes.

Applying the scope down made the deduplication change easier
and it is also beneficial in terms of security posture.

Author: gmarciani

CHANGELOG.md

@@ -8,12 +8,14 @@ CHANGELOG
 - Add validator that warns against the downsides of disabling in-place updates on compute and login nodes through DevSettings.
 - Upgrade jmespath to ~=1.0 (from ~=0.10).
 - Upgrade tabulate to <=0.9.0 (from <=0.8.10).
+- Scope down storage security group ingress rules to specific ports for all node types.
 
 **BUG FIXES**
 - Add validation to block updates that change tag order. Blocking such change prevents update failures.
 - Fix LoginNodes NLB not being publicly accessible when using public subnet with implicit main route table association. 
   See https://github.com/aws/aws-parallelcluster/issues/7173
 - Fix cluster creation failure without Internet access when GPU instances and DCV are used.
+- Fix cluster creation failure caused by eventual consistency issues when head, compute and login nodes have the same security group.
 
 3.14.1
 ------

cli/src/pcluster/templates/cluster_stack.py

@@ -795,19 +795,26 @@ def _add_storage_security_group(self, storage_cfn_id, storage):
             "Storage": [storage_security_group.ref],
         }
 
+        # Deduplicate SGs to avoid creating identical rules for the same security group.
+        # This prevents EC2 duplicate rule errors which are susceptible to eventual consistency issues.
+        seen_sgs = set()
         for sg_type, sg_refs in target_security_groups.items():
             for sg_ref_id, sg_ref in enumerate(sg_refs):
-                # TODO Scope down ingress rules to allow only traffic on the strictly necessary ports.
-                #      Currently scoped down only on Login nodes to limit blast radius.
+                # This conversion is required because sg_ref could either be a string or a CDK token reference.
+                # A CDK token must be converted to a string to make it comparable within the set.
+                sg_key = str(sg_ref)
+                if sg_key in seen_sgs:
+                    continue
+                seen_sgs.add(sg_key)
+                # Scope down ingress rules to allow only traffic on the strictly necessary ports.
                 ingress_protocol = "-1"
                 ingress_port = ALL_PORTS_RANGE
-                if sg_type == "Login":
-                    if storage_type == SharedStorageType.EFS:
-                        ingress_protocol = "tcp"
-                        ingress_port = EFS_PORT
-                    elif storage_type == SharedStorageType.FSX:
-                        ingress_protocol = "tcp"
-                        ingress_port = FSX_PORTS[LUSTRE]["tcp"][0]
+                if storage_type == SharedStorageType.EFS:
+                    ingress_protocol = "tcp"
+                    ingress_port = EFS_PORT
+                elif storage_type == SharedStorageType.FSX:
+                    ingress_protocol = "tcp"
+                    ingress_port = FSX_PORTS[LUSTRE]["tcp"][0]
                 ingress_rule = self._allow_all_ingress(
                     description=f"{storage_cfn_id}SecurityGroup{sg_type}Ingress{sg_ref_id}",
                     source_security_group_id=sg_ref,

cli/tests/pcluster/templates/test_cluster_stack.py

@@ -1374,3 +1374,51 @@ def test_resource_combination_name(
         hash_length=hash_length,
     )
     assert_that(combination_name).is_equal_to(expected_combination_name)
+
+
+def test_storage_security_group_deduplication(mocker, test_datadir):
+    """
+    Test that storage security group rules are deduplicated when head, compute, and login nodes share the same SG.
+
+    When head node, compute nodes, and login nodes all use the same security group (sg-12345678),
+    only one set of ingress/egress rules should be created for that security group, not three separate sets.
+    Additionally, ingress rules should be scoped down to the specific ports required by the storage type.
+    """
+    mock_aws_api(mocker)
+    mock_bucket(mocker)
+    mock_bucket_object_utils(mocker)
+
+    input_yaml = load_yaml_dict(test_datadir / "config-shared-sg.yaml")
+    cluster_config = ClusterSchema(cluster_name="clustername").load(input_yaml)
+
+    generated_template, _ = CDKTemplateBuilder().build_cluster_template(
+        cluster_config=cluster_config, bucket=dummy_cluster_bucket(), stack_name="clustername"
+    )
+
+    # Test both EFS and FSx storage types
+    for storage_type, expected_port in [("EFS", 2049), ("FSX", 988)]:
+        storage_sg_ingress_rules = [
+            (name, resource)
+            for name, resource in generated_template["Resources"].items()
+            if resource["Type"] == "AWS::EC2::SecurityGroupIngress"
+            and name.startswith(storage_type)
+            and "SecurityGroup" in name
+        ]
+
+        # Verify deduplication: only 2 rules instead of 4 (Head, Compute, Login share same SG + Storage SG)
+        assert_that(len(storage_sg_ingress_rules)).is_equal_to(2)
+
+        # Verify ingress rules are scoped down to the expected protocol and port
+        for _name, rule in storage_sg_ingress_rules:
+            props = rule["Properties"]
+            assert_that(props["IpProtocol"]).is_equal_to("tcp")
+            assert_that(props["FromPort"]).is_equal_to(expected_port)
+            assert_that(props["ToPort"]).is_equal_to(expected_port)
+
+        # Verify each unique source security group has exactly one ingress rule
+        source_sgs = {
+            str(rule["Properties"].get("SourceSecurityGroupId"))
+            for _, rule in storage_sg_ingress_rules
+            if rule["Properties"].get("SourceSecurityGroupId")
+        }
+        assert_that(len(storage_sg_ingress_rules)).is_equal_to(len(source_sgs))

cli/tests/pcluster/templates/test_cluster_stack/test_storage_security_group_deduplication/config-shared-sg.yaml

@@ -0,0 +1,46 @@
+Image:
+  Os: alinux2
+HeadNode:
+  InstanceType: t3.micro
+  Ssh:
+    KeyName: ec2-key-name
+  Networking:
+    SubnetId: subnet-12345678
+    SecurityGroups:
+      - sg-12345678
+Scheduling:
+  Scheduler: slurm
+  SlurmQueues:
+    - Name: queue1
+      Networking:
+        SubnetIds:
+          - subnet-12345678
+        SecurityGroups:
+          - sg-12345678
+      ComputeResources:
+        - Name: compute_resource1
+          InstanceType: t3.micro
+          MinCount: 0
+          MaxCount: 10
+LoginNodes:
+  Pools:
+    - Name: login
+      InstanceType: t3.micro
+      Count: 1
+      Networking:
+        SubnetIds:
+          - subnet-12345678
+        SecurityGroups:
+          - sg-12345678
+SharedStorage:
+  - Name: efs
+    StorageType: Efs
+    MountDir: /efs
+    EfsSettings:
+      ThroughputMode: bursting
+  - Name: fsx
+    StorageType: FsxLustre
+    MountDir: /fsx
+    FsxLustreSettings:
+      StorageCapacity: 1200
+      DeploymentType: SCRATCH_2

cli/tests/pcluster/templates/test_shared_storage.py

@@ -107,8 +107,9 @@ def test_shared_storage_efs(mocker, test_datadir, config_file_name, storage_name
         else f"{cluster_config.login_nodes.pools[0].name}LoginNodesSecurityGroup"
     )
     for sg in ["HeadNodeSecurityGroup", "ComputeSecurityGroup", login_nodes_sg_name, mount_target_sg_name]:
-        ingress_ip_protocol = "tcp" if sg == login_nodes_sg_name else "-1"
-        ingress_port_range = [2049, 2049] if sg == login_nodes_sg_name else [0, 65535]
+        # All node types now have scoped-down ingress rules (tcp/2049 for EFS)
+        ingress_ip_protocol = "tcp"
+        ingress_port_range = [2049, 2049]
         rule_deletion_policy = deletion_policy if sg == mount_target_sg_name else None
         assert_sg_rule(
             generated_template,
@@ -173,8 +174,9 @@ def test_shared_storage_fsx(mocker, test_datadir, config_file_name, storage_name
         else f"{cluster_config.login_nodes.pools[0].name}LoginNodesSecurityGroup"
     )
     for sg in ["HeadNodeSecurityGroup", "ComputeSecurityGroup", login_nodes_sg_name, file_system_sg_name]:
-        ingress_ip_protocol = "tcp" if sg == login_nodes_sg_name else "-1"
-        ingress_port_range = [988, 988] if sg == login_nodes_sg_name else [0, 65535]
+        # All node types now have scoped-down ingress rules (tcp/988 for FSx Lustre)
+        ingress_ip_protocol = "tcp"
+        ingress_port_range = [988, 988]
         rule_deletion_policy = deletion_policy if sg == file_system_sg_name else None
         assert_sg_rule(
             generated_template,