fix(gcp): check next rotation time in KMS keys (#4633)

prowler/providers/gcp/services/kms/kms_key_rotation_enabled/kms_key_rotation_enabled.py

@@ -1,3 +1,5 @@
+import datetime
+
 from prowler.lib.check.models import Check, Check_Report_GCP
 from prowler.providers.gcp.services.kms.kms_client import kms_client
 
@@ -11,18 +13,31 @@ class kms_key_rotation_enabled(Check):
             report.resource_id = key.id
             report.resource_name = key.name
             report.location = key.location
-            report.status = "FAIL"
-            report.status_extended = (
-                f"Key {key.name} is not rotated every 90 days or less."
-            )
+            now = datetime.datetime.now()
+            condition_next_rotation_time = False
+            if key.next_rotation_time:
+                next_rotation_time = datetime.datetime.strptime(
+                    key.next_rotation_time, "%Y-%m-%dT%H:%M:%SZ"
+                )
+                condition_next_rotation_time = (
+                    abs((next_rotation_time - now).days) <= 90
+                )
+            condition_rotation_period = False
             if key.rotation_period:
-                if (
+                condition_rotation_period = (
                     int(key.rotation_period[:-1]) // (24 * 3600) <= 90
-                ):  # Convert seconds to days and check if less or equal than 90
-                    report.status = "PASS"
-                    report.status_extended = (
-                        f"Key {key.name} is rotated every 90 days or less."
-                    )
+                )
+            if condition_rotation_period and condition_next_rotation_time:
+                report.status = "PASS"
+                report.status_extended = f"Key {key.name} is rotated every 90 days or less and the next rotation time is in less than 90 days."
+            else:
+                report.status = "FAIL"
+                if condition_rotation_period:
+                    report.status_extended = f"Key {key.name} is rotated every 90 days or less but the next rotation time is in more than 90 days."
+                elif condition_next_rotation_time:
+                    report.status_extended = f"Key {key.name} is not rotated every 90 days or less but the next rotation time is in less than 90 days."
+                else:
+                    report.status_extended = f"Key {key.name} is not rotated every 90 days or less and the next rotation time is in more than 90 days."
             findings.append(report)
 
         return findings

prowler/providers/gcp/services/kms/kms_service.py

@@ -92,6 +92,7 @@ class KMS(GCPService):
                                 name=key["name"].split("/")[-1],
                                 location=key["name"].split("/")[3],
                                 rotation_period=key.get("rotationPeriod"),
+                                next_rotation_time=key.get("nextRotationTime"),
                                 key_ring=ring.name,
                                 project_id=ring.project_id,
                             )
@@ -144,6 +145,7 @@ class CriptoKey(BaseModel):
     name: str
     location: str
     rotation_period: Optional[str]
+    next_rotation_time: Optional[str]
     key_ring: str
     members: list = []
     project_id: str

tests/providers/gcp/services/kms/kms_key_not_publicly_accessible/kms_key_not_publicly_accessible_gcp_test.py

@@ -68,6 +68,7 @@ class Test_kms_key_not_publicly_accessible_gcp:
                     id="projects/123/locations/us-central1/keyRings/keyring1/cryptoKeys/key1",
                     project_id=GCP_PROJECT_ID,
                     rotation_period="7776000s",
+                    next_rotation_time="2021-01-01T00:00:00Z",
                     key_ring=keyring.name,
                     location=keylocation.name,
                     members=["allUsers"],
@@ -125,6 +126,7 @@ class Test_kms_key_not_publicly_accessible_gcp:
                     id="projects/123/locations/us-central1/keyRings/keyring1/cryptoKeys/key1",
                     project_id=GCP_PROJECT_ID,
                     rotation_period="7776000s",
+                    next_rotation_time="2021-01-01T00:00:00Z",
                     key_ring=keyring.name,
                     location=keylocation.name,
                     members=["user:jane@example.com"],
@@ -182,6 +184,7 @@ class Test_kms_key_not_publicly_accessible_gcp:
                     id="projects/123/locations/us-central1/keyRings/keyring1/cryptoKeys/key1",
                     project_id=GCP_PROJECT_ID,
                     rotation_period="7776000s",
+                    next_rotation_time="2021-01-01T00:00:00Z",
                     key_ring=keyring.name,
                     location=keylocation.name,
                     members=[],

tests/providers/gcp/services/kms/kms_key_rotation_enabled/kms_key_rotation_enabled_test.py

@@ -30,7 +30,7 @@ class Test_kms_key_rotation_enabled:
             result = check.execute()
             assert len(result) == 0
 
-    def test_kms_key_no_rotation_period(self):
+    def test_kms_key_no_next_rotation_time_and_no_rotation_period(self):
         kms_client = mock.MagicMock
 
         with mock.patch(
@@ -79,14 +79,242 @@ class Test_kms_key_rotation_enabled:
             assert result[0].status == "FAIL"
             assert (
                 result[0].status_extended
-                == f"Key {kms_client.crypto_keys[0].name} is not rotated every 90 days or less."
+                == f"Key {kms_client.crypto_keys[0].name} is not rotated every 90 days or less and the next rotation time is in more than 90 days."
             )
             assert result[0].resource_id == kms_client.crypto_keys[0].id
             assert result[0].resource_name == kms_client.crypto_keys[0].name
             assert result[0].location == kms_client.crypto_keys[0].location
             assert result[0].project_id == kms_client.crypto_keys[0].project_id
 
-    def test_kms_key_rotation_period_greater_90_days(self):
+    def test_kms_key_no_next_rotation_time_and_big_rotation_period(self):
+        kms_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.kms.kms_key_rotation_enabled.kms_key_rotation_enabled.kms_client",
+            new=kms_client,
+        ):
+            from prowler.providers.gcp.services.kms.kms_key_rotation_enabled.kms_key_rotation_enabled import (
+                kms_key_rotation_enabled,
+            )
+            from prowler.providers.gcp.services.kms.kms_service import (
+                CriptoKey,
+                KeyLocation,
+                KeyRing,
+            )
+
+            kms_client.project_ids = [GCP_PROJECT_ID]
+            kms_client.region = GCP_US_CENTER1_LOCATION
+
+            keyring = KeyRing(
+                name="projects/123/locations/us-central1/keyRings/keyring1",
+                project_id=GCP_PROJECT_ID,
+            )
+
+            keylocation = KeyLocation(
+                name=GCP_US_CENTER1_LOCATION,
+                project_id=GCP_PROJECT_ID,
+            )
+
+            kms_client.crypto_keys = [
+                CriptoKey(
+                    name="key1",
+                    id="projects/123/locations/us-central1/keyRings/keyring1/cryptoKeys/key1",
+                    project_id=GCP_PROJECT_ID,
+                    key_ring=keyring.name,
+                    location=keylocation.name,
+                    rotation_period="8776000s",
+                    members=["user:jane@example.com"],
+                )
+            ]
+
+            check = kms_key_rotation_enabled()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"Key {kms_client.crypto_keys[0].name} is not rotated every 90 days or less and the next rotation time is in more than 90 days."
+            )
+            assert result[0].resource_id == kms_client.crypto_keys[0].id
+            assert result[0].resource_name == kms_client.crypto_keys[0].name
+            assert result[0].location == kms_client.crypto_keys[0].location
+            assert result[0].project_id == kms_client.crypto_keys[0].project_id
+
+    def test_kms_key_no_next_rotation_time_and_appropriate_rotation_period(self):
+        kms_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.kms.kms_key_rotation_enabled.kms_key_rotation_enabled.kms_client",
+            new=kms_client,
+        ):
+            from prowler.providers.gcp.services.kms.kms_key_rotation_enabled.kms_key_rotation_enabled import (
+                kms_key_rotation_enabled,
+            )
+            from prowler.providers.gcp.services.kms.kms_service import (
+                CriptoKey,
+                KeyLocation,
+                KeyRing,
+            )
+
+            kms_client.project_ids = [GCP_PROJECT_ID]
+            kms_client.region = GCP_US_CENTER1_LOCATION
+
+            keyring = KeyRing(
+                name="projects/123/locations/us-central1/keyRings/keyring1",
+                project_id=GCP_PROJECT_ID,
+            )
+
+            keylocation = KeyLocation(
+                name=GCP_US_CENTER1_LOCATION,
+                project_id=GCP_PROJECT_ID,
+            )
+
+            kms_client.crypto_keys = [
+                CriptoKey(
+                    name="key1",
+                    id="projects/123/locations/us-central1/keyRings/keyring1/cryptoKeys/key1",
+                    project_id=GCP_PROJECT_ID,
+                    key_ring=keyring.name,
+                    location=keylocation.name,
+                    rotation_period="7776000s",
+                    members=["user:jane@example.com"],
+                )
+            ]
+
+            check = kms_key_rotation_enabled()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"Key {kms_client.crypto_keys[0].name} is rotated every 90 days or less but the next rotation time is in more than 90 days."
+            )
+            assert result[0].resource_id == kms_client.crypto_keys[0].id
+            assert result[0].resource_name == kms_client.crypto_keys[0].name
+            assert result[0].location == kms_client.crypto_keys[0].location
+            assert result[0].project_id == kms_client.crypto_keys[0].project_id
+
+    def test_kms_key_no_rotation_period_and_big_next_rotation_time(self):
+        kms_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.kms.kms_key_rotation_enabled.kms_key_rotation_enabled.kms_client",
+            new=kms_client,
+        ):
+            from prowler.providers.gcp.services.kms.kms_key_rotation_enabled.kms_key_rotation_enabled import (
+                kms_key_rotation_enabled,
+            )
+            from prowler.providers.gcp.services.kms.kms_service import (
+                CriptoKey,
+                KeyLocation,
+                KeyRing,
+            )
+
+            kms_client.project_ids = [GCP_PROJECT_ID]
+            kms_client.region = GCP_US_CENTER1_LOCATION
+
+            keyring = KeyRing(
+                name="projects/123/locations/us-central1/keyRings/keyring1",
+                project_id=GCP_PROJECT_ID,
+            )
+
+            keylocation = KeyLocation(
+                name=GCP_US_CENTER1_LOCATION,
+                project_id=GCP_PROJECT_ID,
+            )
+
+            kms_client.crypto_keys = [
+                CriptoKey(
+                    name="key1",
+                    id="projects/123/locations/us-central1/keyRings/keyring1/cryptoKeys/key1",
+                    project_id=GCP_PROJECT_ID,
+                    key_ring=keyring.name,
+                    location=keylocation.name,
+                    next_rotation_time="2025-09-01T00:00:00Z",
+                    members=["user:jane@example.com"],
+                )
+            ]
+
+            check = kms_key_rotation_enabled()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"Key {kms_client.crypto_keys[0].name} is not rotated every 90 days or less and the next rotation time is in more than 90 days."
+            )
+            assert result[0].resource_id == kms_client.crypto_keys[0].id
+            assert result[0].resource_name == kms_client.crypto_keys[0].name
+            assert result[0].location == kms_client.crypto_keys[0].location
+            assert result[0].project_id == kms_client.crypto_keys[0].project_id
+
+    def test_kms_key_no_rotation_period_and_appropriate_next_rotation_time(self):
+        kms_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.kms.kms_key_rotation_enabled.kms_key_rotation_enabled.kms_client",
+            new=kms_client,
+        ):
+            from prowler.providers.gcp.services.kms.kms_key_rotation_enabled.kms_key_rotation_enabled import (
+                kms_key_rotation_enabled,
+            )
+            from prowler.providers.gcp.services.kms.kms_service import (
+                CriptoKey,
+                KeyLocation,
+                KeyRing,
+            )
+
+            kms_client.project_ids = [GCP_PROJECT_ID]
+            kms_client.region = GCP_US_CENTER1_LOCATION
+
+            keyring = KeyRing(
+                name="projects/123/locations/us-central1/keyRings/keyring1",
+                project_id=GCP_PROJECT_ID,
+            )
+
+            keylocation = KeyLocation(
+                name=GCP_US_CENTER1_LOCATION,
+                project_id=GCP_PROJECT_ID,
+            )
+
+            kms_client.crypto_keys = [
+                CriptoKey(
+                    name="key1",
+                    id="projects/123/locations/us-central1/keyRings/keyring1/cryptoKeys/key1",
+                    project_id=GCP_PROJECT_ID,
+                    key_ring=keyring.name,
+                    location=keylocation.name,
+                    next_rotation_time="2024-09-01T00:00:00Z",
+                    members=["user:jane@example.com"],
+                )
+            ]
+
+            check = kms_key_rotation_enabled()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"Key {kms_client.crypto_keys[0].name} is not rotated every 90 days or less but the next rotation time is in less than 90 days."
+            )
+            assert result[0].resource_id == kms_client.crypto_keys[0].id
+            assert result[0].resource_name == kms_client.crypto_keys[0].name
+            assert result[0].location == kms_client.crypto_keys[0].location
+            assert result[0].project_id == kms_client.crypto_keys[0].project_id
+
+    def test_kms_key_rotation_period_greater_90_days_and_big_next_rotation_time(self):
         kms_client = mock.MagicMock
 
         with mock.patch(
@@ -124,6 +352,7 @@ class Test_kms_key_rotation_enabled:
                     id="projects/123/locations/us-central1/keyRings/keyring1/cryptoKeys/key1",
                     project_id=GCP_PROJECT_ID,
                     rotation_period="8776000s",
+                    next_rotation_time="2025-09-01T00:00:00Z",
                     key_ring=keyring.name,
                     location=keylocation.name,
                     members=["user:jane@example.com"],
@@ -136,14 +365,74 @@ class Test_kms_key_rotation_enabled:
             assert result[0].status == "FAIL"
             assert (
                 result[0].status_extended
-                == f"Key {kms_client.crypto_keys[0].name} is not rotated every 90 days or less."
+                == f"Key {kms_client.crypto_keys[0].name} is not rotated every 90 days or less and the next rotation time is in more than 90 days."
             )
             assert result[0].resource_id == kms_client.crypto_keys[0].id
             assert result[0].resource_name == kms_client.crypto_keys[0].name
             assert result[0].location == kms_client.crypto_keys[0].location
             assert result[0].project_id == kms_client.crypto_keys[0].project_id
 
-    def test_kms_key_rotation_period_less_90_days(self):
+    def test_kms_key_rotation_period_greater_90_days_and_appropriate_next_rotation_time(
+        self,
+    ):
+        kms_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.kms.kms_key_rotation_enabled.kms_key_rotation_enabled.kms_client",
+            new=kms_client,
+        ):
+            from prowler.providers.gcp.services.kms.kms_key_rotation_enabled.kms_key_rotation_enabled import (
+                kms_key_rotation_enabled,
+            )
+            from prowler.providers.gcp.services.kms.kms_service import (
+                CriptoKey,
+                KeyLocation,
+                KeyRing,
+            )
+
+            kms_client.project_ids = [GCP_PROJECT_ID]
+            kms_client.region = GCP_US_CENTER1_LOCATION
+
+            keyring = KeyRing(
+                name="projects/123/locations/us-central1/keyRings/keyring1",
+                project_id=GCP_PROJECT_ID,
+            )
+
+            keylocation = KeyLocation(
+                name=GCP_US_CENTER1_LOCATION,
+                project_id=GCP_PROJECT_ID,
+            )
+
+            kms_client.crypto_keys = [
+                CriptoKey(
+                    name="key1",
+                    id="projects/123/locations/us-central1/keyRings/keyring1/cryptoKeys/key1",
+                    project_id=GCP_PROJECT_ID,
+                    rotation_period="8776000s",
+                    next_rotation_time="2024-09-01T00:00:00Z",
+                    key_ring=keyring.name,
+                    location=keylocation.name,
+                    members=["user:jane@example.com"],
+                )
+            ]
+
+            check = kms_key_rotation_enabled()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"Key {kms_client.crypto_keys[0].name} is not rotated every 90 days or less but the next rotation time is in less than 90 days."
+            )
+            assert result[0].resource_id == kms_client.crypto_keys[0].id
+            assert result[0].resource_name == kms_client.crypto_keys[0].name
+            assert result[0].location == kms_client.crypto_keys[0].location
+            assert result[0].project_id == kms_client.crypto_keys[0].project_id
+
+    def test_kms_key_rotation_period_less_90_days_and_big_next_rotation_time(self):
         kms_client = mock.MagicMock
 
         with mock.patch(
@@ -181,6 +470,67 @@ class Test_kms_key_rotation_enabled:
                     id="projects/123/locations/us-central1/keyRings/keyring1/cryptoKeys/key1",
                     project_id=GCP_PROJECT_ID,
                     rotation_period="7776000s",
+                    next_rotation_time="2025-09-01T00:00:00Z",
+                    key_ring=keyring.name,
+                    location=keylocation.name,
+                    members=["user:jane@example.com"],
+                )
+            ]
+
+            check = kms_key_rotation_enabled()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"Key {kms_client.crypto_keys[0].name} is rotated every 90 days or less but the next rotation time is in more than 90 days."
+            )
+            assert result[0].resource_id == kms_client.crypto_keys[0].id
+            assert result[0].resource_name == kms_client.crypto_keys[0].name
+            assert result[0].location == kms_client.crypto_keys[0].location
+            assert result[0].project_id == kms_client.crypto_keys[0].project_id
+
+    def test_kms_key_rotation_period_less_90_days_and_appropriate_next_rotation_time(
+        self,
+    ):
+        kms_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.kms.kms_key_rotation_enabled.kms_key_rotation_enabled.kms_client",
+            new=kms_client,
+        ):
+            from prowler.providers.gcp.services.kms.kms_key_rotation_enabled.kms_key_rotation_enabled import (
+                kms_key_rotation_enabled,
+            )
+            from prowler.providers.gcp.services.kms.kms_service import (
+                CriptoKey,
+                KeyLocation,
+                KeyRing,
+            )
+
+            kms_client.project_ids = [GCP_PROJECT_ID]
+            kms_client.region = GCP_US_CENTER1_LOCATION
+
+            keyring = KeyRing(
+                name="projects/123/locations/us-central1/keyRings/keyring1",
+                project_id=GCP_PROJECT_ID,
+            )
+
+            keylocation = KeyLocation(
+                name=GCP_US_CENTER1_LOCATION,
+                project_id=GCP_PROJECT_ID,
+            )
+
+            kms_client.crypto_keys = [
+                CriptoKey(
+                    name="key1",
+                    id="projects/123/locations/us-central1/keyRings/keyring1/cryptoKeys/key1",
+                    project_id=GCP_PROJECT_ID,
+                    rotation_period="7776000s",
+                    next_rotation_time="2024-09-01T00:00:00Z",
                     key_ring=keyring.name,
                     location=keylocation.name,
                     members=["user:jane@example.com"],
@@ -193,7 +543,7 @@ class Test_kms_key_rotation_enabled:
             assert result[0].status == "PASS"
             assert (
                 result[0].status_extended
-                == f"Key {kms_client.crypto_keys[0].name} is rotated every 90 days or less."
+                == f"Key {kms_client.crypto_keys[0].name} is rotated every 90 days or less and the next rotation time is in less than 90 days."
             )
             assert result[0].resource_id == kms_client.crypto_keys[0].id
             assert result[0].resource_name == kms_client.crypto_keys[0].name