localstack · k-a-il · Mar 13, 2025 · Mar 5, 2025 · Mar 5, 2025 · Mar 6, 2025
diff --git a/localstack-core/localstack/services/kms/models.py b/localstack-core/localstack/services/kms/models.py
@@ -693,6 +693,9 @@ def _get_key_usage(self, request_key_usage: str, key_spec: str) -> str:
         else:
             return request_key_usage or "ENCRYPT_DECRYPT"
 
+    def rotate_key_on_demand(self):
+        self.crypto_key = KmsCryptoKey(KeySpec.SYMMETRIC_DEFAULT)
+
 
 class KmsGrant:
     # AWS documentation doesn't seem to mention any metadata object for grants like it does mention KeyMetadata for

diff --git a/localstack-core/localstack/services/kms/provider.py b/localstack-core/localstack/services/kms/provider.py
@@ -92,6 +92,8 @@
     ReEncryptResponse,
     ReplicateKeyRequest,
     ReplicateKeyResponse,
+    RotateKeyOnDemandRequest,
+    RotateKeyOnDemandResponse,
     ScheduleKeyDeletionRequest,
     ScheduleKeyDeletionResponse,
     SignRequest,
@@ -1254,12 +1256,16 @@ def get_key_rotation_status(
         # We do not model that here, though.
         account_id, region_name, key_id = self._parse_key_id(request["KeyId"], context)
         key = self._get_kms_key(account_id, region_name, key_id, any_key_state_allowed=True)
-        return GetKeyRotationStatusResponse(
+
+        response = GetKeyRotationStatusResponse(
             KeyId=key_id,
             KeyRotationEnabled=key.is_key_rotation_enabled,
             NextRotationDate=key.next_rotation_date,
-            RotationPeriodInDays=key.rotation_period_in_days,
         )
+        if key.is_key_rotation_enabled:
+            response["RotationPeriodInDays"] = key.rotation_period_in_days
+
+        return response
 
     @handler("DisableKeyRotation", expand=False)
     def disable_key_rotation(
@@ -1334,6 +1340,27 @@ def list_resource_tags(
         kwargs = {"NextMarker": next_token, "Truncated": True} if next_token else {}
         return ListResourceTagsResponse(Tags=page, **kwargs)
 
+    @handler("RotateKeyOnDemand", expand=False)
+    # TODO: keep trak of key rotations as AWS does and return them in the ListKeyRotations operation
+    def rotate_key_on_demand(
+        self, context: RequestContext, request: RotateKeyOnDemandRequest
+    ) -> RotateKeyOnDemandResponse:
+        account_id, region_name, key_id = self._parse_key_id(request["KeyId"], context)
+        key = self._get_kms_key(account_id, region_name, key_id)
+
+        if key.metadata["KeySpec"] != KeySpec.SYMMETRIC_DEFAULT:
+            raise UnsupportedOperationException()
+        if key.metadata["Origin"] == OriginType.EXTERNAL:
+            raise UnsupportedOperationException(
+                f"{key.metadata['Arn']} origin is EXTERNAL which is not valid for this operation."
+            )
+
+        key.rotate_key_on_demand()
+
+        return RotateKeyOnDemandResponse(
+            KeyId=key_id,
+        )
+
     @handler("TagResource", expand=False)
     def tag_resource(self, context: RequestContext, request: TagResourceRequest) -> None:
         key = self._get_kms_key(

diff --git a/tests/aws/services/kms/test_kms.py b/tests/aws/services/kms/test_kms.py
@@ -19,6 +19,7 @@
 from localstack.testing.pytest import markers
 from localstack.utils.crypto import encrypt
 from localstack.utils.strings import short_uid, to_str
+from localstack.utils.sync import poll_condition
 
 
 def create_tags(**kwargs):
@@ -1098,6 +1099,123 @@ def test_key_rotation_status(self, kms_key, aws_client):
         aws_client.kms.disable_key_rotation(KeyId=key_id)
         assert aws_client.kms.get_key_rotation_status(KeyId=key_id)["KeyRotationEnabled"] is False
 
+    @markers.aws.validated
+    def test_rotate_key_on_demand_modifies_key_material(self, kms_create_key, aws_client, snapshot):
+        key_id = kms_create_key(KeyUsage="ENCRYPT_DECRYPT", KeySpec="SYMMETRIC_DEFAULT")["KeyId"]
+        message = b"test message 123 !%$@ 1234567890"
+
+        ciphertext_before = aws_client.kms.encrypt(
+            KeyId=key_id,
+            Plaintext=base64.b64encode(message),
+            EncryptionAlgorithm="SYMMETRIC_DEFAULT",
+        )["CiphertextBlob"]
+
+        rotate_on_demand_response = aws_client.kms.rotate_key_on_demand(KeyId=key_id)
+        snapshot.match("rotate-on-demand-response", rotate_on_demand_response)
+
+        ciphertext_after = aws_client.kms.encrypt(
+            KeyId=key_id,
+            Plaintext=base64.b64encode(message),
+            EncryptionAlgorithm="SYMMETRIC_DEFAULT",
+        )["CiphertextBlob"]
+
+        assert ciphertext_before != ciphertext_after
+
+    @markers.aws.validated
+    def test_rotate_key_on_demand_with_symmetric_key_and_automatic_rotation_disabled(
+        self, kms_key, aws_client, snapshot
+    ):
+        key_id = kms_key["KeyId"]
+
+        rotate_on_demand_response = aws_client.kms.rotate_key_on_demand(KeyId=key_id)
+        snapshot.match("rotate-on-demand-response", rotate_on_demand_response)
+
+        def _assert_on_demand_rotation_start_date_not_present():
+            response = aws_client.kms.get_key_rotation_status(KeyId=key_id)
+            return "OnDemandRotationStartDate" not in response
+
+        assert poll_condition(
+            condition=_assert_on_demand_rotation_start_date_not_present, timeout=10, interval=1
+        )
+
+        rotation_status_response = aws_client.kms.get_key_rotation_status(KeyId=key_id)
+        snapshot.match("rotation-status-response-after-rotation", rotation_status_response)
+
+    @markers.aws.validated
+    def test_rotate_key_on_demand_with_symmetric_key_and_automatic_rotation_enabled(
+        self, kms_key, aws_client, snapshot
+    ):
+        key_id = kms_key["KeyId"]
+
+        aws_client.kms.enable_key_rotation(KeyId=key_id)
+        rotation_status_response_before = aws_client.kms.get_key_rotation_status(KeyId=key_id)
+
+        rotate_on_demand_response = aws_client.kms.rotate_key_on_demand(KeyId=key_id)
+        snapshot.match("rotate-on-demand-response", rotate_on_demand_response)
+
+        rotation_status_response_after = aws_client.kms.get_key_rotation_status(KeyId=key_id)
+        assert (
+            rotation_status_response_after["NextRotationDate"]
+            == rotation_status_response_before["NextRotationDate"]
+        )
+
+        def _assert_on_demand_rotation_start_date_not_present():
+            response = aws_client.kms.get_key_rotation_status(KeyId=key_id)
+            return "OnDemandRotationStartDate" not in response
+
+        assert poll_condition(
+            condition=_assert_on_demand_rotation_start_date_not_present, timeout=10, interval=1
+        )
+
+        rotation_status_response = aws_client.kms.get_key_rotation_status(KeyId=key_id)
+        snapshot.match("rotation-status-response-after-rotation", rotation_status_response)
+
+    @markers.aws.validated
+    def test_rotate_key_on_demand_raises_error_given_key_is_disabled(
+        self, kms_create_key, aws_client, snapshot
+    ):
+        key_id = kms_create_key(KeyUsage="ENCRYPT_DECRYPT", KeySpec="RSA_4096")["KeyId"]
+        aws_client.kms.disable_key(KeyId=key_id)
+
+        with pytest.raises(ClientError) as e:
+            aws_client.kms.rotate_key_on_demand(KeyId=key_id)
+        snapshot.match("error-response", e.value.response)
+
+    @markers.aws.validated
+    def test_rotate_key_on_demand_raises_error_given_key_that_does_not_exist(
+        self, aws_client, snapshot
+    ):
+        key_id = "1234abcd-12ab-34cd-56ef-1234567890ab"
+
+        with pytest.raises(ClientError) as e:
+            aws_client.kms.rotate_key_on_demand(KeyId=key_id)
+        snapshot.match("error-response", e.value.response)
+
+    @markers.aws.validated
+    @markers.snapshot.skip_snapshot_verify(
+        paths=[
+            "$..message",
+        ],
+    )
+    def test_rotate_key_on_demand_raises_error_given_non_symmetric_key(
+        self, kms_create_key, aws_client, snapshot
+    ):
+        key_id = kms_create_key(KeyUsage="ENCRYPT_DECRYPT", KeySpec="RSA_4096")["KeyId"]
+
+        with pytest.raises(ClientError) as e:
+            aws_client.kms.rotate_key_on_demand(KeyId=key_id)
+        snapshot.match("error-response", e.value.response)
+
+    @markers.aws.validated
+    def test_rotate_key_on_demand_raises_error_given_key_with_imported_key_material(
+        self, kms_create_key, aws_client, snapshot
+    ):
+        key_id = kms_create_key(Origin="EXTERNAL")["KeyId"]
+
+        with pytest.raises(ClientError) as e:
+            aws_client.kms.rotate_key_on_demand(KeyId=key_id)
+        snapshot.match("error-response", e.value.response)
+
     @markers.aws.validated
     @pytest.mark.parametrize("rotation_period_in_days", [90, 180])
     def test_key_enable_rotation_status(

diff --git a/tests/aws/services/kms/test_kms.snapshot.json b/tests/aws/services/kms/test_kms.snapshot.json
@@ -2035,5 +2035,122 @@
         }
       }
     }
+  },
+  "tests/aws/services/kms/test_kms.py::TestKMS::test_rotate_key_on_demand_modifies_key_material": {
+    "recorded-date": "08-03-2025, 09:24:16",
+    "recorded-content": {
+      "rotate-on-demand-response": {
+        "KeyId": "<key-id:1>",
+        "ResponseMetadata": {
+          "HTTPHeaders": {},
+          "HTTPStatusCode": 200
+        }
+      }
+    }
+  },
+  "tests/aws/services/kms/test_kms.py::TestKMS::test_rotate_key_on_demand_with_symmetric_key_and_automatic_rotation_disabled": {
+    "recorded-date": "12-03-2025, 19:05:50",
+    "recorded-content": {
+      "rotate-on-demand-response": {
+        "KeyId": "<key-id:1>",
+        "ResponseMetadata": {
+          "HTTPHeaders": {},
+          "HTTPStatusCode": 200
+        }
+      },
+      "rotation-status-response-after-rotation": {
+        "KeyId": "<key-id:1>",
+        "KeyRotationEnabled": false,
+        "ResponseMetadata": {
+          "HTTPHeaders": {},
+          "HTTPStatusCode": 200
+        }
+      }
+    }
+  },
+  "tests/aws/services/kms/test_kms.py::TestKMS::test_rotate_key_on_demand_with_symmetric_key_and_automatic_rotation_enabled": {
+    "recorded-date": "12-03-2025, 19:07:01",
+    "recorded-content": {
+      "rotate-on-demand-response": {
+        "KeyId": "<key-id:1>",
+        "ResponseMetadata": {
+          "HTTPHeaders": {},
+          "HTTPStatusCode": 200
+        }
+      },
+      "rotation-status-response-after-rotation": {
+        "KeyId": "<key-id:1>",
+        "KeyRotationEnabled": true,
+        "NextRotationDate": "datetime",
+        "RotationPeriodInDays": 365,
+        "ResponseMetadata": {
+          "HTTPHeaders": {},
+          "HTTPStatusCode": 200
+        }
+      }
+    }
+  },
+  "tests/aws/services/kms/test_kms.py::TestKMS::test_rotate_key_on_demand_raises_error_given_key_is_disabled": {
+    "recorded-date": "08-03-2025, 09:26:50",
+    "recorded-content": {
+      "error-response": {
+        "Error": {
+          "Code": "DisabledException",
+          "Message": "<key-arn> is disabled."
+        },
+        "message": "<key-arn> is disabled.",
+        "ResponseMetadata": {
+          "HTTPHeaders": {},
+          "HTTPStatusCode": 400
+        }
+      }
+    }
+  },
+  "tests/aws/services/kms/test_kms.py::TestKMS::test_rotate_key_on_demand_raises_error_given_key_that_does_not_exist": {
+    "recorded-date": "08-03-2025, 09:27:10",
+    "recorded-content": {
+      "error-response": {
+        "Error": {
+          "Code": "NotFoundException",
+          "Message": "Key '<key-arn>' does not exist"
+        },
+        "message": "Key '<key-arn>' does not exist",
+        "ResponseMetadata": {
+          "HTTPHeaders": {},
+          "HTTPStatusCode": 400
+        }
+      }
+    }
+  },
+  "tests/aws/services/kms/test_kms.py::TestKMS::test_rotate_key_on_demand_raises_error_given_non_symmetric_key": {
+    "recorded-date": "08-03-2025, 09:27:44",
+    "recorded-content": {
+      "error-response": {
+        "Error": {
+          "Code": "UnsupportedOperationException",
+          "Message": ""
+        },
+        "ResponseMetadata": {
+          "HTTPHeaders": {},
+          "HTTPStatusCode": 400
+        }
+      }
+    }
+  },
+  "tests/aws/services/kms/test_kms.py::TestKMS::test_rotate_key_on_demand_raises_error_given_key_with_imported_key_material": {
+    "recorded-date": "08-03-2025, 09:28:13",
+    "recorded-content": {
+      "error-response": {
+        "Error": {
+          "Code": "UnsupportedOperationException",
+          "Message": "<key-arn> origin is EXTERNAL which is not valid for this operation."
+        },
+        "message": "<key-arn> origin is EXTERNAL which is not valid for this operation.",
+        "ResponseMetadata": {
+          "HTTPHeaders": {},
+          "HTTPStatusCode": 400
+        }
+      }
+    }
   }
 }
diff --git a/tests/aws/services/kms/test_kms.validation.json b/tests/aws/services/kms/test_kms.validation.json
@@ -203,6 +203,27 @@
   "tests/aws/services/kms/test_kms.py::TestKMS::test_revoke_grant": {
     "last_validated_date": "2024-04-11T15:52:44+00:00"
   },
+  "tests/aws/services/kms/test_kms.py::TestKMS::test_rotate_key_on_demand_modifies_key_material": {
+    "last_validated_date": "2025-03-08T09:24:15+00:00"
+  },
+  "tests/aws/services/kms/test_kms.py::TestKMS::test_rotate_key_on_demand_raises_error_given_key_is_disabled": {
+    "last_validated_date": "2025-03-08T09:26:50+00:00"
+  },
+  "tests/aws/services/kms/test_kms.py::TestKMS::test_rotate_key_on_demand_raises_error_given_key_that_does_not_exist": {
+    "last_validated_date": "2025-03-08T09:27:10+00:00"
+  },
+  "tests/aws/services/kms/test_kms.py::TestKMS::test_rotate_key_on_demand_raises_error_given_key_with_imported_key_material": {
+    "last_validated_date": "2025-03-08T09:28:13+00:00"
+  },
+  "tests/aws/services/kms/test_kms.py::TestKMS::test_rotate_key_on_demand_raises_error_given_non_symmetric_key": {
+    "last_validated_date": "2025-03-08T09:27:44+00:00"
+  },
+  "tests/aws/services/kms/test_kms.py::TestKMS::test_rotate_key_on_demand_with_symmetric_key_and_automatic_rotation_disabled": {
+    "last_validated_date": "2025-03-12T19:05:50+00:00"
+  },
+  "tests/aws/services/kms/test_kms.py::TestKMS::test_rotate_key_on_demand_with_symmetric_key_and_automatic_rotation_enabled": {
+    "last_validated_date": "2025-03-12T19:07:01+00:00"
+  },
   "tests/aws/services/kms/test_kms.py::TestKMS::test_schedule_and_cancel_key_deletion": {
     "last_validated_date": "2024-04-11T15:52:36+00:00"
   },