From 9756b4554baadb602ad1503372962cabce6378f8 Mon Sep 17 00:00:00 2001 From: nik-localstack Date: Wed, 25 Mar 2026 21:54:56 +0100 Subject: [PATCH] ec2: Add support for security groups from NetworkInterfaces in RunInstances --- moto/ec2/models/instances.py | 10 +++ tests/test_ec2/test_instances.py | 128 ++++++++++++++++++++++++++++++- 2 files changed, 135 insertions(+), 3 deletions(-) diff --git a/moto/ec2/models/instances.py b/moto/ec2/models/instances.py index 1815643af482..93175f0716d6 100644 --- a/moto/ec2/models/instances.py +++ b/moto/ec2/models/instances.py @@ -773,6 +773,7 @@ def run_instances( The KeyPair-parameter can be validated, to see if it is a known key-pair. Enable this validation by setting the environment variable `MOTO_ENABLE_KEYPAIR_VALIDATION=true` """ + template_nics: list[dict[str, Any]] = [] if launch_template := kwargs.get("launch_template"): tmpl = self._get_template_from_args(launch_template).data @@ -795,6 +796,15 @@ def run_instances( template_sgs := tmpl.get("SecurityGroups") ): security_group_names = template_sgs + template_nics = tmpl.get("NetworkInterfaces") or [] + + if not kwargs.get("security_group_ids") and not security_group_names: + if nic_groups := [ + g + for nic in (kwargs.get("nics") or template_nics) + for g in (nic.get("Groups") or []) + ]: + kwargs["security_group_ids"] = nic_groups location_type = "availability-zone" if kwargs.get("placement") else "region" default_region = "us-east-1" diff --git a/tests/test_ec2/test_instances.py b/tests/test_ec2/test_instances.py index 96c47dae1ffd..6ea0df350121 100644 --- a/tests/test_ec2/test_instances.py +++ b/tests/test_ec2/test_instances.py @@ -3124,6 +3124,64 @@ def test_block_device_status_conversion(): assert Instance.get_block_device_status("deleting") == "deleting" +@ec2_aws_verified() +@pytest.mark.aws_verified +def test_run_instances__security_groups_from_request_network_interfaces( + valid_ami, cleanups, ec2_client=None +): + vpc_id = ec2_client.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]["VpcId"] + cleanups.append(lambda: ec2_client.delete_vpc(VpcId=vpc_id)) + + subnet_id = ec2_client.create_subnet( + VpcId=vpc_id, + CidrBlock="10.0.1.0/24", + AvailabilityZone=ec2_client.meta.region_name + "a", + )["Subnet"]["SubnetId"] + cleanups.append(lambda: ec2_client.delete_subnet(SubnetId=subnet_id)) + + sg_id = ec2_client.create_security_group( + GroupName=f"test-sg-{str(uuid4())[0:6]}", + Description="test", + VpcId=vpc_id, + )["GroupId"] + cleanups.append(lambda: ec2_client.delete_security_group(GroupId=sg_id)) + + instance = ec2_client.run_instances( + MinCount=1, + MaxCount=1, + ImageId=valid_ami, + NetworkInterfaces=[ + {"DeviceIndex": 0, "SubnetId": subnet_id, "Groups": [sg_id]} + ], + )["Instances"][0] + instance_id = instance["InstanceId"] + + def terminate_and_wait(): + ec2_client.terminate_instances(InstanceIds=[instance_id]) + ec2_client.get_waiter("instance_terminated").wait(InstanceIds=[instance_id]) + + cleanups.append(terminate_and_wait) + + sg_ids = [g["GroupId"] for g in instance["SecurityGroups"]] + assert sg_id in sg_ids + nic_sg_ids = [g["GroupId"] for g in instance["NetworkInterfaces"][0]["Groups"]] + assert sg_id in nic_sg_ids + + described = ec2_client.describe_instances(InstanceIds=[instance_id]) + sg_ids = [ + g["GroupId"] + for g in described["Reservations"][0]["Instances"][0]["SecurityGroups"] + ] + assert sg_id in sg_ids + nic_sg_ids = [ + g["GroupId"] + for g in described["Reservations"][0]["Instances"][0]["NetworkInterfaces"][0][ + "Groups" + ] + ] + assert sg_id in nic_sg_ids + + class TestCreateInstanceFromLaunchTemplate: _LT_USER_DATA_SCRIPT = b"#!/bin/bash\necho from-template" _LT_USER_DATA_B64 = base64.b64encode(_LT_USER_DATA_SCRIPT).decode() @@ -3383,9 +3441,12 @@ def test_run_instances__security_group_ids_from_launch_template( ec2_client, template_name=lt_name, ImageId=valid_ami, SubnetId=subnet_id ) instance_id = instance["InstanceId"] - cleanups.append( - lambda: ec2_client.terminate_instances(InstanceIds=[instance_id]) - ) + + def terminate_and_wait(): + ec2_client.terminate_instances(InstanceIds=[instance_id]) + ec2_client.get_waiter("instance_terminated").wait(InstanceIds=[instance_id]) + + cleanups.append(terminate_and_wait) instance_sg_ids = [g["GroupId"] for g in instance["SecurityGroups"]] assert sg_id in instance_sg_ids @@ -3405,6 +3466,67 @@ def test_run_instances__security_group_ids_from_launch_template( ] assert sg_id in instance_sg_ids + @ec2_aws_verified() + @pytest.mark.aws_verified + def test_run_instances__security_groups_from_launch_template_network_interfaces( + self, valid_ami, cleanups, ec2_client=None + ): + """ + SecurityGroups specified in NetworkInterfaces[].Groups in a launch template + must appear in DescribeInstances SecurityGroups. + """ + vpc_id = ec2_client.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]["VpcId"] + cleanups.append(lambda: ec2_client.delete_vpc(VpcId=vpc_id)) + + subnet_id = ec2_client.create_subnet( + VpcId=vpc_id, + CidrBlock="10.0.1.0/24", + AvailabilityZone=ec2_client.meta.region_name + "a", + )["Subnet"]["SubnetId"] + cleanups.append(lambda: ec2_client.delete_subnet(SubnetId=subnet_id)) + + sg_id = ec2_client.create_security_group( + GroupName=f"test-sg-{str(uuid4())[0:6]}", + Description="test", + VpcId=vpc_id, + )["GroupId"] + cleanups.append(lambda: ec2_client.delete_security_group(GroupId=sg_id)) + + lt_name = str(uuid4()) + ec2_client.create_launch_template( + LaunchTemplateName=lt_name, + LaunchTemplateData={ + "InstanceType": "t2.micro", + "NetworkInterfaces": [ + {"DeviceIndex": 0, "SubnetId": subnet_id, "Groups": [sg_id]} + ], + }, + ) + cleanups.append( + lambda: ec2_client.delete_launch_template(LaunchTemplateName=lt_name) + ) + + instance = self._run_instance_from_template( + ec2_client, template_name=lt_name, ImageId=valid_ami + ) + instance_id = instance["InstanceId"] + + def terminate_and_wait(): + ec2_client.terminate_instances(InstanceIds=[instance_id]) + ec2_client.get_waiter("instance_terminated").wait(InstanceIds=[instance_id]) + + cleanups.append(terminate_and_wait) + + sg_ids = [g["GroupId"] for g in instance["SecurityGroups"]] + assert sg_id in sg_ids + + described = ec2_client.describe_instances(InstanceIds=[instance_id]) + sg_ids = [ + g["GroupId"] + for g in described["Reservations"][0]["Instances"][0]["SecurityGroups"] + ] + assert sg_id in sg_ids + @ec2_aws_verified() @pytest.mark.aws_verified def test_create_instance_from_launch_template_single_template_version(