From 2667a53725e47dfde95f63001b5339c65fdd4043 Mon Sep 17 00:00:00 2001 From: Lewis Gaul Date: Mon, 3 Apr 2023 10:16:31 +0100 Subject: [PATCH] EC2: Support assigning specified private IP addrs (#6147) --- moto/ec2/models/elastic_network_interfaces.py | 16 ++++++- .../responses/elastic_network_interfaces.py | 5 +- .../test_elastic_network_interfaces.py | 48 +++++++++++++------ 3 files changed, 52 insertions(+), 17 deletions(-) diff --git a/moto/ec2/models/elastic_network_interfaces.py b/moto/ec2/models/elastic_network_interfaces.py index c366ac539..5e99038a9 100644 --- a/moto/ec2/models/elastic_network_interfaces.py +++ b/moto/ec2/models/elastic_network_interfaces.py @@ -376,12 +376,22 @@ class NetworkInterfaceBackend: return eni def assign_private_ip_addresses( - self, eni_id: str, secondary_ips_count: Optional[int] = None + self, + eni_id: str, + private_ip_addresses: Optional[List[str]] = None, + secondary_ips_count: Optional[int] = None, ) -> NetworkInterface: eni = self.get_network_interface(eni_id) eni_assigned_ips = [ item.get("PrivateIpAddress") for item in eni.private_ip_addresses ] + if private_ip_addresses: + eni.private_ip_addresses.extend( + {"Primary": False, "PrivateIpAddress": ip} + for ip in private_ip_addresses + if ip not in eni_assigned_ips + ) + return eni while secondary_ips_count: ip = random_private_ip(eni.subnet.cidr_block) if ip not in eni_assigned_ips: @@ -399,7 +409,9 @@ class NetworkInterfaceBackend: ) -> NetworkInterface: eni = self.get_network_interface(eni_id) if ipv6_addresses: - eni.ipv6_addresses.extend(ipv6_addresses) + eni.ipv6_addresses.extend( + (ip for ip in ipv6_addresses if ip not in eni.ipv6_addresses) + ) while ipv6_count: association = list(eni.subnet.ipv6_cidr_block_associations.values())[0] diff --git a/moto/ec2/responses/elastic_network_interfaces.py b/moto/ec2/responses/elastic_network_interfaces.py index af193d4f6..a4dba0fa2 100644 --- a/moto/ec2/responses/elastic_network_interfaces.py +++ b/moto/ec2/responses/elastic_network_interfaces.py @@ -124,7 +124,10 @@ class ElasticNetworkInterfaces(EC2BaseResponse): def assign_private_ip_addresses(self) -> str: eni_id = self._get_param("NetworkInterfaceId") secondary_ips_count = self._get_int_param("SecondaryPrivateIpAddressCount", 0) - eni = self.ec2_backend.assign_private_ip_addresses(eni_id, secondary_ips_count) + private_ip_addresses = self._get_multi_param("PrivateIpAddress") + eni = self.ec2_backend.assign_private_ip_addresses( + eni_id, private_ip_addresses, secondary_ips_count + ) template = self.response_template(ASSIGN_PRIVATE_IP_ADDRESSES) return template.render(eni=eni) diff --git a/tests/test_ec2/test_elastic_network_interfaces.py b/tests/test_ec2/test_elastic_network_interfaces.py index 08a34021e..fbdff0089 100644 --- a/tests/test_ec2/test_elastic_network_interfaces.py +++ b/tests/test_ec2/test_elastic_network_interfaces.py @@ -736,32 +736,52 @@ def test_elastic_network_interfaces_auto_create_securitygroup(): @mock_ec2 -def test_assign_private_ip_addresses(): +def test_assign_private_ip_addresses__by_address(): ec2 = boto3.resource("ec2", region_name="us-east-1") client = boto3.client("ec2", "us-east-1") vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") subnet = ec2.create_subnet(VpcId=vpc.id, CidrBlock="10.0.0.0/18") - private_ip = "54.0.0.1" - eni = ec2.create_network_interface(SubnetId=subnet.id, PrivateIpAddress=private_ip) + primary_ip = "54.0.0.1" + secondary_ip = "80.0.0.1" + eni = ec2.create_network_interface(SubnetId=subnet.id, PrivateIpAddress=primary_ip) resp = client.describe_network_interfaces(NetworkInterfaceIds=[eni.id]) - my_eni = resp["NetworkInterfaces"][0] - my_eni.should.have.key("PrivateIpAddress").equals("54.0.0.1") - my_eni.should.have.key("PrivateIpAddresses").equals( - [{"Primary": True, "PrivateIpAddress": "54.0.0.1"}] + resp_eni = resp["NetworkInterfaces"][0] + resp_eni.should.have.key("PrivateIpAddress").equals(primary_ip) + resp_eni.should.have.key("PrivateIpAddresses").equals( + [{"Primary": True, "PrivateIpAddress": primary_ip}] ) - # Do not pass SecondaryPrivateIpAddressCount-parameter - client.assign_private_ip_addresses(NetworkInterfaceId=eni.id) + # Pass IP address to assign rather than SecondaryPrivateIpAddressCount. + client.assign_private_ip_addresses( + NetworkInterfaceId=eni.id, PrivateIpAddresses=[secondary_ip] + ) - # Verify nothing changes + # Verify secondary IP address is now present. resp = client.describe_network_interfaces(NetworkInterfaceIds=[eni.id]) - my_eni = resp["NetworkInterfaces"][0] - my_eni.should.have.key("PrivateIpAddress").equals("54.0.0.1") - my_eni.should.have.key("PrivateIpAddresses").equals( - [{"Primary": True, "PrivateIpAddress": "54.0.0.1"}] + resp_eni = resp["NetworkInterfaces"][0] + resp_eni.should.have.key("PrivateIpAddress").equals(primary_ip) + resp_eni.should.have.key("PrivateIpAddresses").equals( + [ + {"Primary": True, "PrivateIpAddress": primary_ip}, + {"Primary": False, "PrivateIpAddress": secondary_ip}, + ] + ) + + # Assign the same IP address, this time via the ENI object. + eni.assign_private_ip_addresses(PrivateIpAddresses=[secondary_ip]) + + # Verify nothing changes. + resp = client.describe_network_interfaces(NetworkInterfaceIds=[eni.id]) + resp_eni = resp["NetworkInterfaces"][0] + resp_eni.should.have.key("PrivateIpAddress").equals(primary_ip) + resp_eni.should.have.key("PrivateIpAddresses").equals( + [ + {"Primary": True, "PrivateIpAddress": primary_ip}, + {"Primary": False, "PrivateIpAddress": secondary_ip}, + ] )