EC2: fix interface endpoint ENI IP, security groups and MAC format

moto/ec2/models/vpcs.py

@@ -32,7 +32,6 @@
     create_dns_entries,
     generic_filter,
     random_ipv6_cidr,
-    random_private_ip,
     random_vpc_cidr_association_id,
     random_vpc_ep_id,
     random_vpc_id,
@@ -992,7 +991,15 @@ def create_vpc_endpoint(
             network_interface_ids = []
             for subnet_id in subnet_ids or []:
                 self.get_subnet(subnet_id)  # type: ignore[attr-defined]
-                eni = self.create_network_interface(subnet_id, random_private_ip())  # type: ignore[attr-defined]
+                eni = self.create_network_interface(  # type: ignore[attr-defined]
+                    subnet_id,
+                    private_ip_address=None,
+                    group_ids=security_group_ids,
+                    description=f"VPC Endpoint Interface {vpc_endpoint_id}",
+                )
+                eni.interface_type = "vpc_endpoint"
+                eni.requester_managed = True
+                eni.status = "in-use"
                 network_interface_ids.append(eni.id)
 
             dns_entries = create_dns_entries(service_name, vpc_endpoint_id)

moto/ec2/utils.py

@@ -286,7 +286,7 @@ def generate_dns_from_ip(ip: Any, dns_type: str = "internal") -> str:
 
 
 def random_mac_address() -> str:
-    return f"02:00:00:{random.randint(0, 255)}02x:{random.randint(0, 255)}02x:{random.randint(0, 255)}02x"
+    return f"02:00:00:{random.randint(0, 255):02x}:{random.randint(0, 255):02x}:{random.randint(0, 255):02x}"
 
 
 def randor_ipv4_cidr() -> str:

tests/test_ec2/test_elastic_network_interfaces.py

@@ -38,6 +38,7 @@ def test_elastic_network_interfaces():
     assert eni["RequesterManaged"] is False
     assert eni["PrivateIpAddresses"][0]["PrivateIpAddress"].startswith("10.") is True
     assert "Association" not in eni
+    assert len(eni["MacAddress"]) == 17
 
     with pytest.raises(ClientError) as ex:
         ec2client.delete_network_interface(NetworkInterfaceId=eni_id, DryRun=True)

tests/test_ec2/test_vpcs.py

@@ -1,5 +1,7 @@
+import ipaddress
 import json
 import random
+from time import sleep
 from unittest import SkipTest
 from uuid import uuid4
 
@@ -1036,6 +1038,61 @@ def test_create_vpc_endpoint__policy():
     assert vpc_end_point["PolicyDocument"] == "my policy document"
 
 
+@ec2_aws_verified(create_vpc=True, create_subnet=True, create_sg=True)
+@pytest.mark.aws_verified
+def test_create_interface_vpc_endpoint_network_interfaces(
+    ec2_client=None, vpc_id=None, subnet_id=None, sg_id=None
+):
+    vpc_end_point_id = ec2_client.create_vpc_endpoint(
+        VpcId=vpc_id,
+        ServiceName="com.amazonaws.us-east-1.execute-api",
+        VpcEndpointType="Interface",
+        PrivateDnsEnabled=False,
+        SubnetIds=[subnet_id],
+        SecurityGroupIds=[sg_id],
+    )["VpcEndpoint"]["VpcEndpointId"]
+
+    try:
+        vpc_end_point = wait_for_vpc_endpoint_available(ec2_client, vpc_end_point_id)
+
+        eni = ec2_client.describe_network_interfaces(
+            NetworkInterfaceIds=vpc_end_point["NetworkInterfaceIds"]
+        )["NetworkInterfaces"][0]
+        assert ipaddress.ip_address(eni["PrivateIpAddress"]) in ipaddress.ip_network(
+            "10.0.0.0/24"
+        )
+        assert [group["GroupId"] for group in eni["Groups"]] == [sg_id]
+        assert eni["InterfaceType"] == "vpc_endpoint"
+        assert eni["RequesterManaged"] is True
+        assert eni["Status"] == "in-use"
+        assert eni["Description"] == f"VPC Endpoint Interface {vpc_end_point_id}"
+    finally:
+        ec2_client.delete_vpc_endpoints(VpcEndpointIds=[vpc_end_point_id])
+        # the subnet cannot be deleted while the endpoint's network interface exists
+        wait_for_subnet_network_interfaces_deleted(ec2_client, subnet_id)
+
+
+def wait_for_vpc_endpoint_available(ec2_client, vpc_end_point_id):
+    for _ in range(60):
+        vpc_end_point = ec2_client.describe_vpc_endpoints(
+            VpcEndpointIds=[vpc_end_point_id]
+        )["VpcEndpoints"][0]
+        if vpc_end_point["State"].lower() == "available":
+            return vpc_end_point
+        sleep(5)
+    raise AssertionError(f"VPC endpoint {vpc_end_point_id} never became available")
+
+
+def wait_for_subnet_network_interfaces_deleted(ec2_client, subnet_id):
+    for _ in range(60):
+        interfaces = ec2_client.describe_network_interfaces(
+            Filters=[{"Name": "subnet-id", "Values": [subnet_id]}]
+        )["NetworkInterfaces"]
+        if not interfaces:
+            return
+        sleep(5)
+
+
 @mock_aws
 def test_describe_vpc_gateway_end_points():
     ec2 = boto3.client("ec2", region_name="us-west-1")