Repository: getmoto/moto
Pull request #6904 (2023-10-12 06:53:46)

EC2: Add support for ED25519 SSH key pairs
This PR adds support for ED25519 SSH key pairs in EC2.

It is now possible to use the `KeyType` field in `CreateKeyPair`, as well as the ability to `ImportKeyPair` ED25519 keys.

See:
- https://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_CreateKeyPair.html
- https://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_ImportKeyPair.html
- https://github.com/localstack/localstack/issues/9237


Patch:
diff --git a/moto/batch/models.py b/moto/batch/models.py
index ae9e6449f..887f6f48c 100644
--- a/moto/batch/models.py
+++ b/moto/batch/models.py
@@ -1,13 +1,14 @@
-import re
-from itertools import cycle
-from time import sleep
-from typing import Any, Dict, List, Tuple, Optional, Set
 import datetime
-import time
+import dateutil.parser
 import logging
+import re
 import threading
-import dateutil.parser
+import time
+
 from sys import platform
+from itertools import cycle
+from time import sleep
+from typing import Any, Dict, List, Tuple, Optional, Set
 
 from moto.core import BaseBackend, BackendDict, BaseModel, CloudFormationModel
 from moto.iam.models import iam_backends, IAMBackend
@@ -482,6 +483,8 @@ class Job(threading.Thread, BaseModel, DockerModel, ManagedState):
         depends_on: Optional[List[Dict[str, str]]],
         all_jobs: Dict[str, "Job"],
         timeout: Optional[Dict[str, int]],
+        array_properties: Dict[str, Any],
+        provided_job_id: Optional[str] = None,
     ):
         threading.Thread.__init__(self)
         DockerModel.__init__(self)
@@ -492,7 +495,7 @@ class Job(threading.Thread, BaseModel, DockerModel, ManagedState):
         )
 
         self.job_name = name
-        self.job_id = str(mock_random.uuid4())
+        self.job_id = provided_job_id or str(mock_random.uuid4())
         self.job_definition = job_def
         self.container_overrides: Dict[str, Any] = container_overrides or {}
         self.job_queue = job_queue
@@ -505,6 +508,7 @@ class Job(threading.Thread, BaseModel, DockerModel, ManagedState):
         self.depends_on = depends_on
         self.timeout = timeout
         self.all_jobs = all_jobs
+        self.array_properties: Dict[str, Any] = array_properties
 
         self.arn = make_arn_for_job(
             job_def.backend.account_id, self.job_id, job_def._region
@@ -514,6 +518,7 @@ class Job(threading.Thread, BaseModel, DockerModel, ManagedState):
         self.exit_code: Optional[int] = None
 
         self.daemon = True
+
         self.name = "MOTO-BATCH-" + self.job_id
 
         self._log_backend = log_backend
@@ -523,6 +528,7 @@ class Job(threading.Thread, BaseModel, DockerModel, ManagedState):
 
         self.attempts: List[Dict[str, Any]] = []
         self.latest_attempt: Optional[Dict[str, Any]] = None
+        self._child_jobs: Optional[List[Job]] = None
 
     def describe_short(self) -> Dict[str, Any]:
         result = {
@@ -560,6 +566,26 @@ class Job(threading.Thread, BaseModel, DockerModel, ManagedState):
         if self.timeout:
             result["timeout"] = self.timeout
         result["attempts"] = self.attempts
+        if self._child_jobs:
+            child_statuses = {
+                "STARTING": 0,
+                "FAILED": 0,
+                "RUNNING": 0,
+                "SUCCEEDED": 0,
+                "RUNNABLE": 0,
+                "SUBMITTED": 0,
+                "PENDING": 0,
+            }
+            for child_job in self._child_jobs:
+                if child_job.status is not None:
+                    child_statuses[child_job.status] += 1
+            result["arrayProperties"] = {
+                "statusSummary": child_statuses,
+                "size": len(self._child_jobs),
+            }
+            if len(self._child_jobs) == child_statuses["SUCCEEDED"]:
+                self.status = "SUCCEEDED"
+                result["status"] = self.status
         return result
 
     def _container_details(self) -> Dict[str, Any]:
@@ -675,7 +701,7 @@ class Job(threading.Thread, BaseModel, DockerModel, ManagedState):
                             )
                             for m in self._get_container_property("mountPoints", [])
                         ],
-                        "name": f"{self.job_name}-{self.job_id}",
+                        "name": f"{self.job_name}-{self.job_id.replace(':', '-')}",
                     }
                 )
             else:
@@ -1704,6 +1730,7 @@ class BatchBackend(BaseBackend):
         job_name: str,
         job_def_id: str,
         job_queue: str,
+        array_properties: Dict[str, int],
         depends_on: Optional[List[Dict[str, str]]] = None,
         container_overrides: Optional[Dict[str, Any]] = None,
         timeout: Optional[Dict[str, int]] = None,
@@ -1732,12 +1759,36 @@ class BatchBackend(BaseBackend):
             depends_on=depends_on,
             all_jobs=self._jobs,
             timeout=timeout,
+            array_properties=array_properties or {},
         )
         self._jobs[job.job_id] = job
 
-        # Here comes the fun
-        job.start()
+        if "size" in array_properties:
+            child_jobs = []
+            for array_index in range(array_properties["size"]):
+                provided_job_id = f"{job.job_id}:{array_index}"
+                child_job = Job(
+                    job_name,
+                    job_def,
+                    queue,
+                    log_backend=self.logs_backend,
+                    container_overrides=container_overrides,
+                    depends_on=depends_on,
+                    all_jobs=self._jobs,
+                    timeout=timeout,
+                    array_properties={"statusSummary": {}, "index": array_index},
+                    provided_job_id=provided_job_id,
+                )
+                child_jobs.append(child_job)
+                self._jobs[child_job.job_id] = child_job
+                child_job.start()
 
+            # The 'parent' job doesn't need to be executed
+            # it just needs to keep track of it's children
+            job._child_jobs = child_jobs
+        else:
+            # Here comes the fun
+            job.start()
         return job_name, job.job_id
 
     def describe_jobs(self, jobs: Optional[List[str]]) -> List[Dict[str, Any]]:
diff --git a/moto/batch/responses.py b/moto/batch/responses.py
index 844c28ce7..ef933f3e4 100644
--- a/moto/batch/responses.py
+++ b/moto/batch/responses.py
@@ -210,6 +210,7 @@ class BatchResponse(BaseResponse):
         job_name = self._get_param("jobName")
         job_queue = self._get_param("jobQueue")
         timeout = self._get_param("timeout")
+        array_properties = self._get_param("arrayProperties", {})
 
         name, job_id = self.batch_backend.submit_job(
             job_name,
@@ -218,6 +219,7 @@ class BatchResponse(BaseResponse):
             depends_on=depends_on,
             container_overrides=container_overrides,
             timeout=timeout,
+            array_properties=array_properties,
         )
 
         result = {"jobId": job_id, "jobName": name}
diff --git a/moto/batch_simple/models.py b/moto/batch_simple/models.py
index f9761172e..6d8079ced 100644
--- a/moto/batch_simple/models.py
+++ b/moto/batch_simple/models.py
@@ -1,9 +1,9 @@
 from ..batch.models import (
     batch_backends,
     BaseBackend,
-    Job,
-    ClientException,
     BatchBackend,
+    ClientException,
+    Job,
 )
 from ..core import BackendDict
 
@@ -42,7 +42,7 @@ class BatchSimpleBackend(BaseBackend):
             "url_bases",
         ]:
             return object.__getattribute__(self, name)
-        if name in ["submit_job"]:
+        if name in ["submit_job", "_mark_job_as_finished"]:
 
             def newfunc(*args: Any, **kwargs: Any) -> Any:
                 attr = object.__getattribute__(self, name)
@@ -57,6 +57,7 @@ class BatchSimpleBackend(BaseBackend):
         job_name: str,
         job_def_id: str,
         job_queue: str,
+        array_properties: Dict[str, Any],
         depends_on: Optional[List[Dict[str, str]]] = None,
         container_overrides: Optional[Dict[str, Any]] = None,
         timeout: Optional[Dict[str, int]] = None,
@@ -79,13 +80,40 @@ class BatchSimpleBackend(BaseBackend):
             depends_on=depends_on,
             all_jobs=self._jobs,
             timeout=timeout,
+            array_properties=array_properties,
         )
-        self.backend._jobs[job.job_id] = job
 
+        if "size" in array_properties:
+            child_jobs: List[Job] = []
+            for array_index in range(array_properties.get("size", 0)):
+                provided_job_id = f"{job.job_id}:{array_index}"
+                child_job = Job(
+                    job_name,
+                    job_def,
+                    queue,
+                    log_backend=self.logs_backend,
+                    container_overrides=container_overrides,
+                    depends_on=depends_on,
+                    all_jobs=self._jobs,
+                    timeout=timeout,
+                    array_properties={"statusSummary": {}, "index": array_index},
+                    provided_job_id=provided_job_id,
+                )
+                self._mark_job_as_finished(include_start_attempt=True, job=child_job)
+                child_jobs.append(child_job)
+            self._mark_job_as_finished(include_start_attempt=False, job=job)
+            job._child_jobs = child_jobs
+        else:
+            self._mark_job_as_finished(include_start_attempt=True, job=job)
+
+        return job_name, job.job_id
+
+    def _mark_job_as_finished(self, include_start_attempt: bool, job: Job) -> None:
+        self.backend._jobs[job.job_id] = job
         job.job_started_at = datetime.datetime.now()
         job.log_stream_name = job._stream_name
-        job._start_attempt()
-
+        if include_start_attempt:
+            job._start_attempt()
         # We don't want to actually run the job - just mark it as succeeded or failed
         # depending on whether env var MOTO_SIMPLE_BATCH_FAIL_AFTER is set
         # if MOTO_SIMPLE_BATCH_FAIL_AFTER is set to an integer then batch will
@@ -104,7 +132,5 @@ class BatchSimpleBackend(BaseBackend):
         else:
             job._mark_stopped(success=True)
 
-        return job_name, job.job_id
-
 
 batch_simple_backends = BackendDict(BatchSimpleBackend, "batch")
diff --git a/moto/ec2/models/key_pairs.py b/moto/ec2/models/key_pairs.py
index 7d29018ff..5ab46f529 100644
--- a/moto/ec2/models/key_pairs.py
+++ b/moto/ec2/models/key_pairs.py
@@ -8,11 +8,12 @@ from ..exceptions import (
     InvalidKeyPairFormatError,
 )
 from ..utils import (
-    random_key_pair,
-    rsa_public_key_fingerprint,
-    rsa_public_key_parse,
     generic_filter,
+    public_key_fingerprint,
+    public_key_parse,
     random_key_pair_id,
+    random_ed25519_key_pair,
+    random_rsa_key_pair,
 )
 from moto.core.utils import iso_8601_datetime_with_milliseconds, utcnow
 
@@ -42,10 +43,14 @@ class KeyPairBackend:
     def __init__(self) -> None:
         self.keypairs: Dict[str, KeyPair] = {}
 
-    def create_key_pair(self, name: str) -> KeyPair:
+    def create_key_pair(self, name: str, key_type: str = "rsa") -> KeyPair:
         if name in self.keypairs:
             raise InvalidKeyPairDuplicateError(name)
-        keypair = KeyPair(name, **random_key_pair())
+        if key_type == "ed25519":
+            keypair = KeyPair(name, **random_ed25519_key_pair())
+        else:
+            keypair = KeyPair(name, **random_rsa_key_pair())
+
         self.keypairs[name] = keypair
         return keypair
 
@@ -77,11 +82,11 @@ class KeyPairBackend:
             raise InvalidKeyPairDuplicateError(key_name)
 
         try:
-            rsa_public_key = rsa_public_key_parse(public_key_material)
+            public_key = public_key_parse(public_key_material)
         except ValueError:
             raise InvalidKeyPairFormatError()
 
-        fingerprint = rsa_public_key_fingerprint(rsa_public_key)
+        fingerprint = public_key_fingerprint(public_key)
         keypair = KeyPair(
             key_name, material=public_key_material, fingerprint=fingerprint
         )
diff --git a/moto/ec2/responses/key_pairs.py b/moto/ec2/responses/key_pairs.py
index 65f6d2842..b0b80f01c 100644
--- a/moto/ec2/responses/key_pairs.py
+++ b/moto/ec2/responses/key_pairs.py
@@ -4,9 +4,9 @@ from ._base_response import EC2BaseResponse
 class KeyPairs(EC2BaseResponse):
     def create_key_pair(self) -> str:
         name = self._get_param("KeyName")
+        key_type = self._get_param("KeyType")
         self.error_on_dryrun()
-
-        keypair = self.ec2_backend.create_key_pair(name)
+        keypair = self.ec2_backend.create_key_pair(name, key_type)
         return self.response_template(CREATE_KEY_PAIR_RESPONSE).render(keypair=keypair)
 
     def delete_key_pair(self) -> str:
diff --git a/moto/ec2/utils.py b/moto/ec2/utils.py
index c768f026c..2ce0b03df 100644
--- a/moto/ec2/utils.py
+++ b/moto/ec2/utils.py
@@ -6,6 +6,11 @@ import ipaddress
 from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.hazmat.primitives.asymmetric.ed25519 import (
+    Ed25519PublicKey,
+    Ed25519PrivateKey,
+)
+from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicKey
 from typing import Any, Dict, List, Set, TypeVar, Tuple, Optional, Union
 
 from moto.core.utils import utcnow
@@ -552,7 +557,22 @@ def simple_aws_filter_to_re(filter_string: str) -> str:
     return tmp_filter
 
 
-def random_key_pair() -> Dict[str, str]:
+def random_ed25519_key_pair() -> Dict[str, str]:
+    private_key = Ed25519PrivateKey.generate()
+    private_key_material = private_key.private_bytes(
+        encoding=serialization.Encoding.PEM,
+        format=serialization.PrivateFormat.OpenSSH,
+        encryption_algorithm=serialization.NoEncryption(),
+    )
+    fingerprint = public_key_fingerprint(private_key.public_key())
+
+    return {
+        "fingerprint": fingerprint,
+        "material": private_key_material.decode("ascii"),
+    }
+
+
+def random_rsa_key_pair() -> Dict[str, str]:
     private_key = rsa.generate_private_key(
         public_exponent=65537, key_size=2048, backend=default_backend()
     )
@@ -561,10 +581,10 @@ def random_key_pair() -> Dict[str, str]:
         format=serialization.PrivateFormat.TraditionalOpenSSL,
         encryption_algorithm=serialization.NoEncryption(),
     )
-    public_key_fingerprint = rsa_public_key_fingerprint(private_key.public_key())
+    fingerprint = public_key_fingerprint(private_key.public_key())
 
     return {
-        "fingerprint": public_key_fingerprint,
+        "fingerprint": fingerprint,
         "material": private_key_material.decode("ascii"),
     }
 
@@ -645,7 +665,9 @@ def generate_instance_identity_document(instance: Any) -> Dict[str, Any]:
     return document
 
 
-def rsa_public_key_parse(key_material: Any) -> Any:
+def public_key_parse(
+    key_material: Union[str, bytes]
+) -> Union[RSAPublicKey, Ed25519PublicKey]:
     # These imports take ~.5s; let's keep them local
     import sshpubkeys.exceptions
     from sshpubkeys.keys import SSHKey
@@ -654,19 +676,26 @@ def rsa_public_key_parse(key_material: Any) -> Any:
         if not isinstance(key_material, bytes):
             key_material = key_material.encode("ascii")
 
-        decoded_key = base64.b64decode(key_material).decode("ascii")
-        public_key = SSHKey(decoded_key)
+        decoded_key = base64.b64decode(key_material)
+        public_key = SSHKey(decoded_key.decode("ascii"))
     except (sshpubkeys.exceptions.InvalidKeyException, UnicodeDecodeError):
         raise ValueError("bad key")
 
-    if not public_key.rsa:
-        raise ValueError("bad key")
+    if public_key.rsa:
+        return public_key.rsa
+
+    # `cryptography` currently does not support RSA RFC4716/SSH2 format, otherwise we could get rid of `sshpubkeys` and
+    # simply use `load_ssh_public_key()`
+    if public_key.key_type == b"ssh-ed25519":
+        return serialization.load_ssh_public_key(decoded_key)  # type: ignore[return-value]
 
-    return public_key.rsa
+    raise ValueError("bad key")
 
 
-def rsa_public_key_fingerprint(rsa_public_key: Any) -> str:
-    key_data = rsa_public_key.public_bytes(
+def public_key_fingerprint(public_key: Union[RSAPublicKey, Ed25519PublicKey]) -> str:
+    # TODO: Use different fingerprint calculation methods based on key type and source
+    # see https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/verify-keys.html#how-ec2-key-fingerprints-are-calculated
+    key_data = public_key.public_bytes(
         encoding=serialization.Encoding.DER,
         format=serialization.PublicFormat.SubjectPublicKeyInfo,
     )


Test patch:
diff --git a/tests/test_batch/test_batch_jobs.py b/tests/test_batch/test_batch_jobs.py
index 01d2d51a1..402dbc3c3 100644
--- a/tests/test_batch/test_batch_jobs.py
+++ b/tests/test_batch/test_batch_jobs.py
@@ -86,6 +86,50 @@ def test_submit_job_by_name():
     assert resp_jobs["jobs"][0]["jobDefinition"] == job_definition_arn
 
 
+@mock_ec2
+@mock_ecs
+@mock_iam
+@mock_batch
+def test_submit_job_array_size():
+    # Setup
+    job_definition_name = f"sleep10_{str(uuid4())[0:6]}"
+    ec2_client, iam_client, _, _, batch_client = _get_clients()
+    commands = ["echo", "hello"]
+    _, _, _, iam_arn = _setup(ec2_client, iam_client)
+    _, queue_arn = prepare_job(batch_client, commands, iam_arn, job_definition_name)
+
+    # Execute
+    resp = batch_client.submit_job(
+        jobName="test1",
+        jobQueue=queue_arn,
+        jobDefinition=job_definition_name,
+        arrayProperties={"size": 2},
+    )
+
+    # Verify
+    job_id = resp["jobId"]
+    child_job_1_id = f"{job_id}:0"
+
+    job = batch_client.describe_jobs(jobs=[job_id])["jobs"][0]
+
+    assert job["arrayProperties"]["size"] == 2
+    assert job["attempts"] == []
+
+    _wait_for_job_status(batch_client, job_id, "SUCCEEDED")
+
+    job = batch_client.describe_jobs(jobs=[job_id])["jobs"][0]
+    # If the main job is successful, that means that all child jobs are successful
+    assert job["arrayProperties"]["size"] == 2
+    assert job["arrayProperties"]["statusSummary"]["SUCCEEDED"] == 2
+    # Main job still has no attempts - because only the child jobs are executed
+    assert job["attempts"] == []
+
+    child_job_1 = batch_client.describe_jobs(jobs=[child_job_1_id])["jobs"][0]
+    assert child_job_1["status"] == "SUCCEEDED"
+    # Child job was executed
+    assert len(child_job_1["attempts"]) == 1
+
+
 # SLOW TESTS
 
 
diff --git a/tests/test_batch_simple/test_batch_jobs.py b/tests/test_batch_simple/test_batch_jobs.py
index ebed5aacb..2d2c940ff 100644
--- a/tests/test_batch_simple/test_batch_jobs.py
+++ b/tests/test_batch_simple/test_batch_jobs.py
@@ -23,7 +23,9 @@ def test_submit_job_by_name():
     )
 
     resp = batch_client.submit_job(
-        jobName="test1", jobQueue=queue_arn, jobDefinition=job_definition_name
+        jobName="test1",
+        jobQueue=queue_arn,
+        jobDefinition=job_definition_name,
     )
     job_id = resp["jobId"]
 
@@ -41,6 +43,44 @@ def test_submit_job_by_name():
     assert "logStreamName" in job["container"]
 
 
+@mock_ec2
+@mock_ecs
+@mock_iam
+@mock_batch_simple
+def test_submit_job_array_size():
+    # Setup
+    job_definition_name = f"sleep10_{str(uuid4())[0:6]}"
+    batch_client, _, queue_arn = setup_common_batch_simple(job_definition_name)
+
+    # Execute
+    resp = batch_client.submit_job(
+        jobName="test1",
+        jobQueue=queue_arn,
+        jobDefinition=job_definition_name,
+        arrayProperties={"size": 2},
+    )
+
+    # Verify
+    job_id = resp["jobId"]
+    child_job_1_id = f"{job_id}:0"
+
+    job = batch_client.describe_jobs(jobs=[job_id])["jobs"][0]
+
+    assert job["arrayProperties"]["size"] == 2
+    assert job["attempts"] == []
+
+    # If the main job is successful, that means that all child jobs are successful
+    assert job["arrayProperties"]["size"] == 2
+    assert job["arrayProperties"]["statusSummary"]["SUCCEEDED"] == 2
+    # Main job still has no attempts - because only the child jobs are executed
+    assert job["attempts"] == []
+
+    child_job_1 = batch_client.describe_jobs(jobs=[child_job_1_id])["jobs"][0]
+    assert child_job_1["status"] == "SUCCEEDED"
+    # Child job was executed
+    assert len(child_job_1["attempts"]) == 1
+
+
 @mock_batch_simple
 def test_update_job_definition():
     _, _, _, _, batch_client = _get_clients()
diff --git a/tests/test_ec2/helpers.py b/tests/test_ec2/helpers.py
index 9d231d19a..00a3b9efb 100644
--- a/tests/test_ec2/helpers.py
+++ b/tests/test_ec2/helpers.py
@@ -1,14 +1,23 @@
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import serialization
-from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.hazmat.primitives.asymmetric import rsa, ed25519
 
 
-def rsa_check_private_key(private_key_material):
+def check_private_key(private_key_material, key_type):
     assert isinstance(private_key_material, str)
 
-    private_key = serialization.load_pem_private_key(
-        data=private_key_material.encode("ascii"),
-        backend=default_backend(),
-        password=None,
-    )
-    assert isinstance(private_key, rsa.RSAPrivateKey)
+    if key_type == "rsa":
+        private_key = serialization.load_pem_private_key(
+            data=private_key_material.encode("ascii"),
+            backend=default_backend(),
+            password=None,
+        )
+        assert isinstance(private_key, rsa.RSAPrivateKey)
+    elif key_type == "ed25519":
+        private_key = serialization.load_ssh_private_key(
+            data=private_key_material.encode("ascii"),
+            password=None,
+        )
+        assert isinstance(private_key, ed25519.Ed25519PrivateKey)
+    else:
+        raise AssertionError("Bad private key")
diff --git a/tests/test_ec2/test_key_pairs.py b/tests/test_ec2/test_key_pairs.py
index b0e6573df..45c659cc9 100644
--- a/tests/test_ec2/test_key_pairs.py
+++ b/tests/test_ec2/test_key_pairs.py
@@ -7,9 +7,16 @@ from moto import mock_ec2, settings
 from uuid import uuid4
 from unittest import SkipTest
 
-from .helpers import rsa_check_private_key
+from .helpers import check_private_key
 
 
+ED25519_PUBLIC_KEY_OPENSSH = b"""\
+ssh-ed25519 \
+AAAAC3NzaC1lZDI1NTE5AAAAIEwsSB9HbTeKCdkSlMZeTq9jZggaPJUwAsUi/7wakB+B \
+moto@getmoto"""
+
+ED25519_PUBLIC_KEY_FINGERPRINT = "6c:d9:cf:90:d7:f7:bc:46:83:9e:f5:56:aa:e1:13:38"
+
 RSA_PUBLIC_KEY_OPENSSH = b"""\
 ssh-rsa \
 AAAAB3NzaC1yc2EAAAADAQABAAABAQDusXfgTE4eBP50NglSzCSEGnIL6+cr6m3H\
@@ -77,17 +84,18 @@ def test_key_pairs_create_dryrun_boto3():
 
 
 @mock_ec2
-def test_key_pairs_create_boto3():
+@pytest.mark.parametrize("key_type", ["rsa", "ed25519"])
+def test_key_pairs_create_boto3(key_type):
     ec2 = boto3.resource("ec2", "us-west-1")
     client = boto3.client("ec2", "us-west-1")
 
     key_name = str(uuid4())[0:6]
-    kp = ec2.create_key_pair(KeyName=key_name)
-    rsa_check_private_key(kp.key_material)
+    kp = ec2.create_key_pair(KeyName=key_name, KeyType=key_type)
+    check_private_key(kp.key_material, key_type)
     # Verify the client can create a key_pair as well - should behave the same
     key_name2 = str(uuid4())
-    kp2 = client.create_key_pair(KeyName=key_name2)
-    rsa_check_private_key(kp2["KeyMaterial"])
+    kp2 = client.create_key_pair(KeyName=key_name2, KeyType=key_type)
+    check_private_key(kp2["KeyMaterial"], key_type)
 
     assert kp.key_material != kp2["KeyMaterial"]
 
@@ -145,13 +153,22 @@ def test_key_pairs_delete_exist_boto3():
 
 
 @mock_ec2
-def test_key_pairs_import_boto3():
+@pytest.mark.parametrize(
+    "public_key,fingerprint",
+    [
+        (RSA_PUBLIC_KEY_OPENSSH, RSA_PUBLIC_KEY_FINGERPRINT),
+        (RSA_PUBLIC_KEY_RFC4716, RSA_PUBLIC_KEY_FINGERPRINT),
+        (ED25519_PUBLIC_KEY_OPENSSH, ED25519_PUBLIC_KEY_FINGERPRINT),
+    ],
+    ids=["rsa-openssh", "rsa-rfc4716", "ed25519"],
+)
+def test_key_pairs_import_boto3(public_key, fingerprint):
     client = boto3.client("ec2", "us-west-1")
 
     key_name = str(uuid4())[0:6]
     with pytest.raises(ClientError) as ex:
         client.import_key_pair(
-            KeyName=key_name, PublicKeyMaterial=RSA_PUBLIC_KEY_OPENSSH, DryRun=True
+            KeyName=key_name, PublicKeyMaterial=public_key, DryRun=True
         )
     assert ex.value.response["Error"]["Code"] == "DryRunOperation"
     assert ex.value.response["ResponseMetadata"]["HTTPStatusCode"] == 412
@@ -160,26 +177,15 @@ def test_key_pairs_import_boto3():
         == "An error occurred (DryRunOperation) when calling the ImportKeyPair operation: Request would have succeeded, but DryRun flag is set"
     )
 
-    kp1 = client.import_key_pair(
-        KeyName=key_name, PublicKeyMaterial=RSA_PUBLIC_KEY_OPENSSH
-    )
+    kp1 = client.import_key_pair(KeyName=key_name, PublicKeyMaterial=public_key)
 
     assert "KeyPairId" in kp1
     assert kp1["KeyName"] == key_name
-    assert kp1["KeyFingerprint"] == RSA_PUBLIC_KEY_FINGERPRINT
-
-    key_name2 = str(uuid4())
-    kp2 = client.import_key_pair(
-        KeyName=key_name2, PublicKeyMaterial=RSA_PUBLIC_KEY_RFC4716
-    )
-    assert "KeyPairId" in kp2
-    assert kp2["KeyName"] == key_name2
-    assert kp2["KeyFingerprint"] == RSA_PUBLIC_KEY_FINGERPRINT
+    assert kp1["KeyFingerprint"] == fingerprint
 
     all_kps = client.describe_key_pairs()["KeyPairs"]
     all_names = [kp["KeyName"] for kp in all_kps]
     assert kp1["KeyName"] in all_names
-    assert kp2["KeyName"] in all_names
 
 
 @mock_ec2
diff --git a/tests/test_ec2/test_utils.py b/tests/test_ec2/test_utils.py
index 3061467e3..9211b59fa 100644
--- a/tests/test_ec2/test_utils.py
+++ b/tests/test_ec2/test_utils.py
@@ -5,17 +5,20 @@ from pytest import raises
 
 from moto.ec2 import utils
 
-from .helpers import rsa_check_private_key
+from .helpers import check_private_key
 
 
 def test_random_key_pair():
-    key_pair = utils.random_key_pair()
-    rsa_check_private_key(key_pair["material"])
+    key_pair = utils.random_rsa_key_pair()
+    check_private_key(key_pair["material"], "rsa")
 
     # AWS uses MD5 fingerprints, which are 47 characters long, *not* SHA1
     # fingerprints with 59 characters.
     assert len(key_pair["fingerprint"]) == 47
 
+    key_pair = utils.random_ed25519_key_pair()
+    check_private_key(key_pair["material"], "ed25519")
+
 
 def test_random_ipv6_cidr():
     def mocked_random_resource_id(chars: int):