feat(alibabacloud): add oss_bucket_versioning_enabled check (#11913)

Co-authored-by: Daniel Barranquero <danielbo2001@gmail.com>

prowler/changelog.d/oss-bucket-versioning-check.added.md

@@ -0,0 +1 @@
+`oss_bucket_versioning_enabled` check for Alibaba Cloud provider, verifying that OSS buckets have versioning enabled to allow recovery from accidental or malicious object overwrite and deletion

prowler/providers/alibabacloud/services/oss/oss_bucket_versioning_enabled/oss_bucket_versioning_enabled.metadata.json

@@ -0,0 +1,36 @@
+{
+  "Provider": "alibabacloud",
+  "CheckID": "oss_bucket_versioning_enabled",
+  "CheckTitle": "Versioning is enabled for OSS buckets",
+  "CheckType": [],
+  "ServiceName": "oss",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "",
+  "Severity": "medium",
+  "ResourceType": "ALIYUN::OSS::Bucket",
+  "ResourceGroup": "storage",
+  "Description": "**Alibaba Cloud OSS Bucket Versioning** stores previous versions of objects when they are overwritten or deleted, allowing any version to be restored at a later time. Enabling versioning on all OSS buckets ensures that data can be recovered after unintended changes or deletions.",
+  "Risk": "Without **OSS bucket versioning** enabled, objects that are overwritten or deleted cannot be recovered. This increases the impact of **accidental deletion**, **misconfigured automation**, and **malicious activity such as ransomware-style overwrites**, as no prior object versions are retained for recovery.",
+  "RelatedUrl": "",
+  "AdditionalURLs": [
+    "https://www.alibabacloud.com/help/doc-detail/109695.htm"
+  ],
+  "Remediation": {
+    "Code": {
+      "CLI": "ossutil bucket-versioning --method put oss://<bucket-name> enabled",
+      "NativeIaC": "",
+      "Other": "1. Log on to the **OSS Console**\n2. In the bucket-list pane, click on a target OSS bucket\n3. Find the **Versioning** setting\n4. Click **Configure** and select **Enabled**\n5. Click **Save**",
+      "Terraform": "resource \"alicloud_oss_bucket_versioning\" \"example\" {\n  bucket = alicloud_oss_bucket.example.bucket\n  status = \"Enabled\"\n}"
+    },
+    "Recommendation": {
+      "Text": "Enable versioning on all OSS buckets so that overwritten or deleted objects can be recovered, protecting data against accidental loss and malicious overwrites.",
+      "Url": "https://hub.prowler.com/check/oss_bucket_versioning_enabled"
+    }
+  },
+  "Categories": [
+    "resilience"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}

prowler/providers/alibabacloud/services/oss/oss_bucket_versioning_enabled/oss_bucket_versioning_enabled.py

@@ -0,0 +1,35 @@
+from prowler.lib.check.models import Check, CheckReportAlibabaCloud
+from prowler.providers.alibabacloud.services.oss.oss_client import oss_client
+
+
+class oss_bucket_versioning_enabled(Check):
+    """Check if versioning is enabled for OSS buckets."""
+
+    def execute(self) -> list[CheckReportAlibabaCloud]:
+        findings = []
+
+        for bucket in oss_client.buckets.values():
+            report = CheckReportAlibabaCloud(metadata=self.metadata(), resource=bucket)
+            report.region = bucket.region
+            report.resource_id = bucket.name
+            report.resource_arn = bucket.arn
+
+            if bucket.versioning_status == "Enabled":
+                report.status = "PASS"
+                report.status_extended = (
+                    f"OSS bucket {bucket.name} has versioning enabled."
+                )
+            else:
+                report.status = "FAIL"
+                if bucket.versioning_status == "Suspended":
+                    report.status_extended = (
+                        f"OSS bucket {bucket.name} has versioning suspended."
+                    )
+                else:
+                    report.status_extended = (
+                        f"OSS bucket {bucket.name} does not have versioning enabled."
+                    )
+
+            findings.append(report)
+
+        return findings

prowler/providers/alibabacloud/services/oss/oss_service.py

@@ -38,6 +38,7 @@ class OSS(AlibabaCloudService):
         self.__threading_call__(self._get_bucket_acl, self.buckets.values())
         self.__threading_call__(self._get_bucket_policy, self.buckets.values())
         self.__threading_call__(self._get_bucket_logging, self.buckets.values())
+        self.__threading_call__(self._get_bucket_versioning, self.buckets.values())
 
     def _list_buckets(self, regional_client=None):
         region = "unknown"
@@ -285,6 +286,27 @@ class OSS(AlibabaCloudService):
                 f"{bucket.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
             )
 
+    def _get_bucket_versioning(self, bucket):
+        """Get bucket versioning status using OSS SDK."""
+        logger.info(f"OSS - Getting versioning status for bucket {bucket.name}...")
+        try:
+            oss_client = self.session.client("oss", bucket.region)
+
+            response = oss_client.get_bucket_versioning(bucket.name)
+
+            if response and response.body:
+                status = None
+                for attr_name in ["version_status", "versioning_status", "status"]:
+                    if getattr(response.body, attr_name, None):
+                        status = getattr(response.body, attr_name)
+                        break
+                if status:
+                    bucket.versioning_status = str(status)
+        except Exception as error:
+            logger.error(
+                f"{bucket.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
     @staticmethod
     def _normalize_bucket_region(bucket_location: str) -> str:
         """Normalize OSS bucket location values to region IDs."""
@@ -330,4 +352,5 @@ class Bucket(BaseModel):
     logging_enabled: bool = False
     logging_target_bucket: str = ""
     logging_target_prefix: str = ""
+    versioning_status: str = ""  # "", Enabled, Suspended
     creation_date: Optional[datetime] = None

tests/providers/alibabacloud/services/oss/oss_bucket_versioning_enabled/oss_bucket_versioning_enabled_test.py

@@ -0,0 +1,135 @@
+from unittest import mock
+
+from tests.providers.alibabacloud.alibabacloud_fixtures import (
+    set_mocked_alibabacloud_provider,
+)
+
+
+class TestOssBucketVersioningEnabled:
+    def test_bucket_with_versioning_enabled(self):
+        oss_client = mock.MagicMock()
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_alibabacloud_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.alibabacloud.services.oss.oss_bucket_versioning_enabled.oss_bucket_versioning_enabled.oss_client",
+                new=oss_client,
+            ),
+        ):
+            from prowler.providers.alibabacloud.services.oss.oss_bucket_versioning_enabled.oss_bucket_versioning_enabled import (
+                oss_bucket_versioning_enabled,
+            )
+            from prowler.providers.alibabacloud.services.oss.oss_service import Bucket
+
+            bucket = Bucket(
+                arn="acs:oss::1234567890:versioning-enabled",
+                name="versioning-enabled",
+                region="cn-hangzhou",
+                versioning_status="Enabled",
+            )
+            oss_client.buckets = {bucket.arn: bucket}
+
+            check = oss_bucket_versioning_enabled()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert "has versioning enabled" in result[0].status_extended
+            assert result[0].resource_id == "versioning-enabled"
+            assert result[0].resource_arn == bucket.arn
+
+    def test_bucket_with_versioning_suspended(self):
+        oss_client = mock.MagicMock()
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_alibabacloud_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.alibabacloud.services.oss.oss_bucket_versioning_enabled.oss_bucket_versioning_enabled.oss_client",
+                new=oss_client,
+            ),
+        ):
+            from prowler.providers.alibabacloud.services.oss.oss_bucket_versioning_enabled.oss_bucket_versioning_enabled import (
+                oss_bucket_versioning_enabled,
+            )
+            from prowler.providers.alibabacloud.services.oss.oss_service import Bucket
+
+            bucket = Bucket(
+                arn="acs:oss::1234567890:versioning-suspended",
+                name="versioning-suspended",
+                region="cn-hangzhou",
+                versioning_status="Suspended",
+            )
+            oss_client.buckets = {bucket.arn: bucket}
+
+            check = oss_bucket_versioning_enabled()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert "has versioning suspended" in result[0].status_extended
+            assert result[0].resource_id == "versioning-suspended"
+            assert result[0].resource_arn == bucket.arn
+
+    def test_bucket_without_versioning(self):
+        oss_client = mock.MagicMock()
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_alibabacloud_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.alibabacloud.services.oss.oss_bucket_versioning_enabled.oss_bucket_versioning_enabled.oss_client",
+                new=oss_client,
+            ),
+        ):
+            from prowler.providers.alibabacloud.services.oss.oss_bucket_versioning_enabled.oss_bucket_versioning_enabled import (
+                oss_bucket_versioning_enabled,
+            )
+            from prowler.providers.alibabacloud.services.oss.oss_service import Bucket
+
+            bucket = Bucket(
+                arn="acs:oss::1234567890:no-versioning",
+                name="no-versioning",
+                region="cn-hangzhou",
+            )
+            oss_client.buckets = {bucket.arn: bucket}
+
+            check = oss_bucket_versioning_enabled()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert "does not have versioning enabled" in result[0].status_extended
+            assert result[0].resource_id == "no-versioning"
+            assert result[0].resource_arn == bucket.arn
+
+    def test_no_buckets(self):
+        oss_client = mock.MagicMock()
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_alibabacloud_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.alibabacloud.services.oss.oss_bucket_versioning_enabled.oss_bucket_versioning_enabled.oss_client",
+                new=oss_client,
+            ),
+        ):
+            from prowler.providers.alibabacloud.services.oss.oss_bucket_versioning_enabled.oss_bucket_versioning_enabled import (
+                oss_bucket_versioning_enabled,
+            )
+
+            oss_client.buckets = {}
+
+            check = oss_bucket_versioning_enabled()
+            result = check.execute()
+
+            assert len(result) == 0