feat(alibabacloud): add oss_bucket_server_side_encryption_enabled check (#11981)

Signed-off-by: Alex Chen <l46983284@gmail.com> Co-authored-by: Daniel Barranquero <danielbo2001@gmail.com>
2026-10-04 02:04:06 +00:00 · 2026-08-25 17:41:52 +02:00
parent f299e1d9ac
commit 301ca50541
7 changed files with 390 additions and 0 deletions
@@ -0,0 +1,179 @@
+from unittest import mock
+
+from tests.providers.alibabacloud.alibabacloud_fixtures import (
+    set_mocked_alibabacloud_provider,
+)
+
+
+class TestOssBucketServerSideEncryptionEnabled:
+    def test_bucket_with_aes256_encryption(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_server_side_encryption_enabled.oss_bucket_server_side_encryption_enabled.oss_client",
+                new=oss_client,
+            ),
+        ):
+            from prowler.providers.alibabacloud.services.oss.oss_bucket_server_side_encryption_enabled.oss_bucket_server_side_encryption_enabled import (
+                oss_bucket_server_side_encryption_enabled,
+            )
+            from prowler.providers.alibabacloud.services.oss.oss_service import Bucket
+
+            bucket = Bucket(
+                arn="acs:oss::1234567890:encrypted-aes",
+                name="encrypted-aes",
+                region="cn-hangzhou",
+                encryption_algorithm="AES256",
+            )
+            oss_client.buckets = {bucket.arn: bucket}
+
+            check = oss_bucket_server_side_encryption_enabled()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert "server-side encryption enabled" in result[0].status_extended
+            assert "AES256" in result[0].status_extended
+            assert result[0].resource_id == "encrypted-aes"
+            assert result[0].resource_arn == bucket.arn
+
+    def test_bucket_with_kms_encryption(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_server_side_encryption_enabled.oss_bucket_server_side_encryption_enabled.oss_client",
+                new=oss_client,
+            ),
+        ):
+            from prowler.providers.alibabacloud.services.oss.oss_bucket_server_side_encryption_enabled.oss_bucket_server_side_encryption_enabled import (
+                oss_bucket_server_side_encryption_enabled,
+            )
+            from prowler.providers.alibabacloud.services.oss.oss_service import Bucket
+
+            bucket = Bucket(
+                arn="acs:oss::1234567890:encrypted-kms",
+                name="encrypted-kms",
+                region="cn-hangzhou",
+                encryption_algorithm="KMS",
+                encryption_kms_key_id="9468da86-3509-4f8d-a61e-6eab1eac",
+            )
+            oss_client.buckets = {bucket.arn: bucket}
+
+            check = oss_bucket_server_side_encryption_enabled()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert "KMS" in result[0].status_extended
+            assert "9468da86-3509-4f8d-a61e-6eab1eac" in result[0].status_extended
+            assert result[0].resource_id == "encrypted-kms"
+            assert result[0].resource_arn == bucket.arn
+
+    def test_bucket_with_kms_key_and_sm4_data_encryption(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_server_side_encryption_enabled.oss_bucket_server_side_encryption_enabled.oss_client",
+                new=oss_client,
+            ),
+        ):
+            from prowler.providers.alibabacloud.services.oss.oss_bucket_server_side_encryption_enabled.oss_bucket_server_side_encryption_enabled import (
+                oss_bucket_server_side_encryption_enabled,
+            )
+            from prowler.providers.alibabacloud.services.oss.oss_service import Bucket
+
+            bucket = Bucket(
+                arn="acs:oss::1234567890:encrypted-sm4",
+                name="encrypted-sm4",
+                region="cn-hangzhou",
+                encryption_algorithm="KMS",
+                encryption_kms_key_id="00000000-1111-2222-3333-444444444444",
+                encryption_kms_data_algorithm="SM4",
+            )
+            oss_client.buckets = {bucket.arn: bucket}
+
+            check = oss_bucket_server_side_encryption_enabled()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert "server-side encryption enabled" in result[0].status_extended
+            assert "data encryption SM4" in result[0].status_extended
+            assert result[0].resource_id == "encrypted-sm4"
+            assert result[0].resource_arn == bucket.arn
+
+    def test_bucket_without_encryption(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_server_side_encryption_enabled.oss_bucket_server_side_encryption_enabled.oss_client",
+                new=oss_client,
+            ),
+        ):
+            from prowler.providers.alibabacloud.services.oss.oss_bucket_server_side_encryption_enabled.oss_bucket_server_side_encryption_enabled import (
+                oss_bucket_server_side_encryption_enabled,
+            )
+            from prowler.providers.alibabacloud.services.oss.oss_service import Bucket
+
+            bucket = Bucket(
+                arn="acs:oss::1234567890:no-encryption",
+                name="no-encryption",
+                region="cn-hangzhou",
+            )
+            oss_client.buckets = {bucket.arn: bucket}
+
+            check = oss_bucket_server_side_encryption_enabled()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                "does not have default server-side encryption enabled"
+                in result[0].status_extended
+            )
+            assert result[0].resource_id == "no-encryption"
+            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_server_side_encryption_enabled.oss_bucket_server_side_encryption_enabled.oss_client",
+                new=oss_client,
+            ),
+        ):
+            from prowler.providers.alibabacloud.services.oss.oss_bucket_server_side_encryption_enabled.oss_bucket_server_side_encryption_enabled import (
+                oss_bucket_server_side_encryption_enabled,
+            )
+
+            oss_client.buckets = {}
+
+            check = oss_bucket_server_side_encryption_enabled()
+            result = check.execute()
+
+            assert len(result) == 0
@@ -133,3 +133,100 @@ def test_list_buckets_inventory_is_loaded_once_across_regions():
        oss.__threading_call__(oss._list_buckets)

    assert get_mock.call_count == 1
+
+
+def _build_bucket(name="prowler-test"):
+    from prowler.providers.alibabacloud.services.oss.oss_service import Bucket
+
+    return Bucket(arn=f"acs:oss::1234567890:{name}", name=name, region="ap-southeast-1")
+
+
+def test_get_bucket_encryption_parses_aes256_rule():
+    from alibabacloud_oss20190517 import models as oss_models
+
+    service = _build_oss_service()
+    bucket = _build_bucket()
+    oss_client = MagicMock()
+    oss_client.get_bucket_encryption.return_value = (
+        oss_models.GetBucketEncryptionResponse(
+            body=oss_models.GetBucketEncryptionResponseBody().from_map(
+                {"ApplyServerSideEncryptionByDefault": {"SSEAlgorithm": "AES256"}}
+            )
+        )
+    )
+    service.session.client.return_value = oss_client
+
+    service._get_bucket_encryption(bucket)
+
+    oss_client.get_bucket_encryption.assert_called_once_with(bucket.name)
+    assert bucket.encryption_algorithm == "AES256"
+    assert bucket.encryption_kms_key_id == ""
+    assert bucket.encryption_kms_data_algorithm == ""
+
+
+def test_get_bucket_encryption_parses_kms_rule():
+    from alibabacloud_oss20190517 import models as oss_models
+
+    service = _build_oss_service()
+    bucket = _build_bucket()
+    oss_client = MagicMock()
+    oss_client.get_bucket_encryption.return_value = (
+        oss_models.GetBucketEncryptionResponse(
+            body=oss_models.GetBucketEncryptionResponseBody().from_map(
+                {
+                    "ApplyServerSideEncryptionByDefault": {
+                        "SSEAlgorithm": "KMS",
+                        "KMSMasterKeyID": "00000000-1111-2222-3333-444444444444",
+                        "KMSDataEncryption": "SM4",
+                    }
+                }
+            )
+        )
+    )
+    service.session.client.return_value = oss_client
+
+    service._get_bucket_encryption(bucket)
+
+    assert bucket.encryption_algorithm == "KMS"
+    assert bucket.encryption_kms_key_id == "00000000-1111-2222-3333-444444444444"
+    assert bucket.encryption_kms_data_algorithm == "SM4"
+
+
+def test_get_bucket_encryption_no_rule_is_not_logged_as_error():
+    from Tea.exceptions import TeaException
+
+    service = _build_oss_service()
+    bucket = _build_bucket()
+    oss_client = MagicMock()
+    oss_client.get_bucket_encryption.side_effect = TeaException(
+        {
+            "code": "NoSuchServerSideEncryptionRule",
+            "message": "No encryption rules are configured for this bucket.",
+        }
+    )
+    service.session.client.return_value = oss_client
+
+    with patch(
+        "prowler.providers.alibabacloud.services.oss.oss_service.logger"
+    ) as mock_logger:
+        service._get_bucket_encryption(bucket)
+
+    mock_logger.error.assert_not_called()
+    assert bucket.encryption_algorithm == ""
+    assert bucket.encryption_kms_key_id == ""
+
+
+def test_get_bucket_encryption_unexpected_error_is_logged():
+    service = _build_oss_service()
+    bucket = _build_bucket()
+    oss_client = MagicMock()
+    oss_client.get_bucket_encryption.side_effect = RuntimeError("boom")
+    service.session.client.return_value = oss_client
+
+    with patch(
+        "prowler.providers.alibabacloud.services.oss.oss_service.logger"
+    ) as mock_logger:
+        service._get_bucket_encryption(bucket)
+
+    mock_logger.error.assert_called_once()
+    assert bucket.encryption_algorithm == ""