Repository: getmoto/moto (python)

Issue:
The filter 'kms-key-id' for DescribeSnapshots has not been implemented in Moto yet
Hello

Tried to create the following unit test : 
create a volume
create a snapshot of the volume
copy the snapshot with a new KMS key
then call `describe_snapshots` with this filter `[{"Name": "kms-key-id", "Values": [kms_key]}]`

and got` moto.ec2.exceptions.FilterNotImplementedError: The filter 'kms-key-id' for DescribeSnapshots has not been implemented in Moto yet.`


Pull request description:
EC2: describe_snapshots() now supports the kms-key-id filter
Fixes #7977

Patch:
diff --git a/moto/ec2/models/elastic_block_store.py b/moto/ec2/models/elastic_block_store.py
index 10b87bf26..cee0558d3 100644
--- a/moto/ec2/models/elastic_block_store.py
+++ b/moto/ec2/models/elastic_block_store.py
@@ -222,6 +222,7 @@ class Snapshot(TaggedEC2Resource):
         volume: Any,
         description: str,
         encrypted: bool = False,
+        kms_key_id: Optional[str] = None,
         owner_id: Optional[str] = None,
         from_ami: Optional[str] = None,
     ):
@@ -233,9 +234,10 @@ class Snapshot(TaggedEC2Resource):
         self.create_volume_permission_userids: Set[str] = set()
         self.ec2_backend = ec2_backend
         self.status = "completed"
-        self.encrypted = encrypted
+        self.encrypted = encrypted or (kms_key_id is not None)
         self.owner_id = owner_id or ec2_backend.account_id
         self.from_ami = from_ami
+        self.kms_key_id = kms_key_id
 
     def get_filter_value(
         self, filter_name: str, method_name: Optional[str] = None
@@ -252,6 +254,8 @@ class Snapshot(TaggedEC2Resource):
             return self.volume.size
         elif filter_name == "encrypted":
             return str(self.encrypted).lower()
+        elif filter_name == "kms-key-id":
+            return self.kms_key_id
         elif filter_name == "status":
             return self.status
         elif filter_name == "owner-id":
@@ -416,12 +420,18 @@ class EBSBackend:
     ) -> Snapshot:
         snapshot_id = random_snapshot_id()
         volume = self.get_volume(volume_id)
-        params = [self, snapshot_id, volume, description, volume.encrypted]
+        params = {
+            "ec2_backend": self,
+            "snapshot_id": snapshot_id,
+            "volume": volume,
+            "description": description,
+            "encrypted": volume.encrypted,
+        }
         if owner_id:
-            params.append(owner_id)
+            params["owner_id"] = owner_id
         if from_ami:
-            params.append(from_ami)
-        snapshot = Snapshot(*params)  # type: ignore[arg-type]
+            params["from_ami"] = from_ami
+        snapshot = Snapshot(**params)  # type: ignore[arg-type]
         self.snapshots[snapshot_id] = snapshot
         return snapshot
 
@@ -463,7 +473,11 @@ class EBSBackend:
         return matches
 
     def copy_snapshot(
-        self, source_snapshot_id: str, source_region: str, description: str
+        self,
+        source_snapshot_id: str,
+        source_region: str,
+        description: str,
+        kms_key_id: Optional[str],
     ) -> Snapshot:
         from ..models import ec2_backends
 
@@ -478,6 +492,7 @@ class EBSBackend:
             volume=source_snapshot.volume,
             description=description,
             encrypted=source_snapshot.encrypted,
+            kms_key_id=kms_key_id,
         )
         self.snapshots[snapshot_id] = snapshot
         return snapshot
diff --git a/moto/ec2/responses/elastic_block_store.py b/moto/ec2/responses/elastic_block_store.py
index 6eb57e488..968b4629a 100644
--- a/moto/ec2/responses/elastic_block_store.py
+++ b/moto/ec2/responses/elastic_block_store.py
@@ -19,11 +19,15 @@ class ElasticBlockStore(EC2BaseResponse):
         description = self._get_param("Description")
         tags = self._parse_tag_specification()
         snapshot_tags = tags.get("snapshot", {})
+        kms_key_id = self._get_param("KmsKeyId")
 
         self.error_on_dryrun()
 
         snapshot = self.ec2_backend.copy_snapshot(
-            source_snapshot_id, source_region, description
+            source_snapshot_id=source_snapshot_id,
+            source_region=source_region,
+            description=description,
+            kms_key_id=kms_key_id,
         )
         snapshot.add_tags(snapshot_tags)
         template = self.response_template(COPY_SNAPSHOT_RESPONSE)
@@ -412,6 +416,7 @@ DESCRIBE_SNAPSHOTS_RESPONSE = """<DescribeSnapshotsResponse xmlns="http://ec2.am
             <volumeSize>{{ snapshot.volume.size }}</volumeSize>
              <description>{{ snapshot.description }}</description>
              <encrypted>{{ 'true' if snapshot.encrypted else 'false' }}</encrypted>
+             {% if snapshot.kms_key_id %}<kmsKeyId>{{ snapshot.kms_key_id }}</kmsKeyId>{% endif %}
              <tagSet>
                {% for tag in snapshot.get_tags() %}
                  <item>


Test patch:
diff --git a/tests/test_ec2/test_elastic_block_store.py b/tests/test_ec2/test_elastic_block_store.py
index 2f9b6ae95..1fdd4d75b 100644
--- a/tests/test_ec2/test_elastic_block_store.py
+++ b/tests/test_ec2/test_elastic_block_store.py
@@ -526,6 +526,14 @@ def test_snapshot_filters():
     assert snapshot1.id in [s["SnapshotId"] for s in snapshots]
     assert snapshot2.id in [s["SnapshotId"] for s in snapshots]
     assert snapshot3.id in [s["SnapshotId"] for s in snapshots]
+    #
+    # Create a copy with a KMS key, and filter on that
+    snapshot4 = client.copy_snapshot(
+        SourceSnapshotId=snapshot1.id,
+        SourceRegion="us-east-1",
+        KmsKeyId="newkmskey",
+    )["SnapshotId"]
+    verify_filter("kms-key-id", "newkmskey", expected=[snapshot4])
 
 
 @mock_aws