fix(alibabacloud): read OSS bucket sub-resource configs via the SDK execute path (#12546)

prowler/changelog.d/oss-bucket-subresource-parsing.fixed.md

@@ -0,0 +1 @@
+OSS bucket logging, versioning, default encryption and ACL configurations are now read correctly from the Alibaba Cloud SDK, so `oss_bucket_logging_enabled`, `oss_bucket_versioning_enabled`, `oss_bucket_server_side_encryption_enabled` and `oss_bucket_not_publicly_accessible` no longer report every bucket as unconfigured

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

@@ -8,6 +8,8 @@ from threading import Lock
 from typing import Optional
 
 import requests
+from alibabacloud_tea_openapi import models as open_api_models
+from alibabacloud_tea_util import models as util_models
 from defusedxml import ElementTree
 from pydantic.v1 import BaseModel
 
@@ -133,58 +135,59 @@ class OSS(AlibabaCloudService):
             )
             return
 
+    def _get_bucket_subresource(self, bucket, action: str, subresource: str) -> dict:
+        """Call a bucket sub-resource API (GET /?<subresource>) and return its parsed body.
+
+        The generated OSS SDK methods return empty response models for these
+        APIs: the OSS gateway keeps the XML root element when it deserializes
+        the body, while the generated response models expect its children at the
+        top level. Calling the shared ``execute`` path directly and unwrapping the
+        root element preserves the actual configuration.
+
+        Args:
+            bucket: Bucket to query.
+            action: OSS API action name (e.g. ``GetBucketEncryption``).
+            subresource: Sub-resource query string (e.g. ``encryption``).
+
+        Returns:
+            dict: Content of the XML root element, or an empty dict when the
+            response carries no configuration.
+
+        Raises:
+            Exception: Any error raised by the OSS SDK, including ``TeaException``
+            with the OSS error code for 4xx/5xx responses.
+        """
+        oss_client = self.session.client("oss", bucket.region)
+        params = open_api_models.Params(
+            action=action,
+            version="2019-05-17",
+            protocol="HTTPS",
+            pathname=f"/?{subresource}",
+            method="GET",
+            auth_type="AK",
+            style="ROA",
+            req_body_type="xml",
+            body_type="xml",
+        )
+        request = open_api_models.OpenApiRequest(
+            host_map={"bucket": bucket.name}, headers={}
+        )
+        response = oss_client.execute(params, request, util_models.RuntimeOptions())
+        body = response.get("body") if isinstance(response, dict) else None
+        if not isinstance(body, dict):
+            return {}
+        if len(body) == 1:
+            root_content = next(iter(body.values()))
+            return root_content if isinstance(root_content, dict) else {}
+        return body
+
     def _get_bucket_acl(self, bucket):
-        """Get bucket ACL."""
+        """Get bucket ACL (private, public-read or public-read-write)."""
         logger.info(f"OSS - Getting ACL for bucket {bucket.name}...")
         try:
-            # Get OSS client for the bucket's region
-            # OSS bucket operations use regional endpoint: oss-{region}.aliyuncs.com
-            oss_client = self.session.client("oss", bucket.region)
-
-            # Get bucket ACL
-            response = oss_client.get_bucket_acl(bucket.name)
-
-            if response and response.body:
-                # ACL can be retrieved from the response
-                # The ACL value is typically in the response body
-                acl_value = getattr(response.body, "acl", None)
-                if acl_value:
-                    # ACL values: private, public-read, public-read-write
-                    bucket.acl = acl_value
-                else:
-                    # Try to get from access_control_list if available
-                    acl_list = getattr(response.body, "access_control_list", None)
-                    if acl_list:
-                        grant = getattr(acl_list, "grant", None)
-                        if grant:
-                            # Check grants to determine ACL type
-                            if isinstance(grant, list):
-                                # Check if any grant has public access
-                                for g in grant:
-                                    permission = getattr(g, "permission", "")
-                                    if permission in ["READ", "FULL_CONTROL"]:
-                                        if permission == "READ":
-                                            bucket.acl = "public-read"
-                                        else:
-                                            bucket.acl = "public-read-write"
-                                        break
-                                else:
-                                    bucket.acl = "private"
-                            else:
-                                permission = getattr(grant, "permission", "")
-                                if permission == "READ":
-                                    bucket.acl = "public-read"
-                                elif permission == "FULL_CONTROL":
-                                    bucket.acl = "public-read-write"
-                                else:
-                                    bucket.acl = "private"
-                        else:
-                            bucket.acl = "private"
-                    else:
-                        bucket.acl = "private"
-            else:
-                bucket.acl = "private"
-
+            acl_policy = self._get_bucket_subresource(bucket, "GetBucketAcl", "acl")
+            grant = (acl_policy.get("AccessControlList") or {}).get("Grant")
+            bucket.acl = str(grant) if grant else "private"
         except Exception as error:
             logger.error(
                 f"{bucket.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -221,96 +224,46 @@ class OSS(AlibabaCloudService):
                 bucket.policy = {}
 
     def _get_bucket_logging(self, bucket):
-        """Get bucket logging configuration using OSS SDK."""
+        """Get bucket logging configuration."""
         logger.info(f"OSS - Getting logging configuration for bucket {bucket.name}...")
         try:
-            oss_client = self.session.client("oss", bucket.region)
-
-            response = oss_client.get_bucket_logging(bucket.name)
-
-            if response and response.body:
-                logging_enabled = None
-                if hasattr(response.body, "logging_enabled"):
-                    logging_enabled = response.body.logging_enabled
-                elif hasattr(response.body, "loggingenabled"):
-                    logging_enabled = response.body.loggingenabled
-                elif hasattr(response.body, "bucket_logging"):
-                    logging_enabled = response.body.bucket_logging
-
-                if logging_enabled:
-                    target_bucket = None
-                    target_prefix = None
-
-                    for attr_name in [
-                        "target_bucket",
-                        "targetBucket",
-                        "target_bucket_name",
-                        "targetBucketName",
-                    ]:
-                        if hasattr(logging_enabled, attr_name):
-                            target_bucket = getattr(logging_enabled, attr_name)
-                            break
-
-                    for attr_name in [
-                        "target_prefix",
-                        "targetPrefix",
-                        "target_prefix_name",
-                        "targetPrefixName",
-                    ]:
-                        if hasattr(logging_enabled, attr_name):
-                            target_prefix = getattr(logging_enabled, attr_name)
-                            break
-
-                    if target_bucket:
-                        bucket.logging_enabled = True
-                        bucket.logging_target_bucket = (
-                            str(target_bucket) if target_bucket else ""
-                        )
-                        bucket.logging_target_prefix = (
-                            str(target_prefix) if target_prefix else ""
-                        )
-                    else:
-                        bucket.logging_enabled = False
-                        bucket.logging_target_bucket = ""
-                        bucket.logging_target_prefix = ""
-                else:
-                    bucket.logging_enabled = False
-                    bucket.logging_target_bucket = ""
-                    bucket.logging_target_prefix = ""
+            logging_status = self._get_bucket_subresource(
+                bucket, "GetBucketLogging", "logging"
+            )
+            logging_enabled = logging_status.get("LoggingEnabled") or {}
+            target_bucket = logging_enabled.get("TargetBucket")
+            if target_bucket:
+                bucket.logging_enabled = True
+                bucket.logging_target_bucket = str(target_bucket)
+                bucket.logging_target_prefix = str(
+                    logging_enabled.get("TargetPrefix") or ""
+                )
             else:
                 bucket.logging_enabled = False
                 bucket.logging_target_bucket = ""
                 bucket.logging_target_prefix = ""
-
         except Exception as error:
             logger.error(
                 f"{bucket.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
             )
 
     def _get_bucket_encryption(self, bucket):
-        """Get bucket default server-side encryption configuration using OSS SDK."""
+        """Get bucket default server-side encryption configuration."""
         logger.info(
             f"OSS - Getting encryption configuration for bucket {bucket.name}..."
         )
         try:
-            oss_client = self.session.client("oss", bucket.region)
-
-            response = oss_client.get_bucket_encryption(bucket.name)
-
-            if response and response.body:
-                default_rule = getattr(
-                    response.body, "apply_server_side_encryption_by_default", None
-                )
-                if default_rule:
-                    bucket.encryption_algorithm = str(
-                        getattr(default_rule, "ssealgorithm", None) or ""
-                    )
-                    bucket.encryption_kms_key_id = str(
-                        getattr(default_rule, "kmsmaster_key_id", None) or ""
-                    )
-                    bucket.encryption_kms_data_algorithm = str(
-                        getattr(default_rule, "kmsdata_encryption", None) or ""
-                    )
+            encryption_rule = self._get_bucket_subresource(
+                bucket, "GetBucketEncryption", "encryption"
+            )
+            default_rule = (
+                encryption_rule.get("ApplyServerSideEncryptionByDefault") or {}
+            )
+            bucket.encryption_algorithm = str(default_rule.get("SSEAlgorithm") or "")
+            bucket.encryption_kms_key_id = str(default_rule.get("KMSMasterKeyID") or "")
+            bucket.encryption_kms_data_algorithm = str(
+                default_rule.get("KMSDataEncryption") or ""
+            )
         except Exception as error:
             # No encryption rule configured means default encryption is disabled
             error_code = getattr(error, "code", "")
@@ -324,21 +277,13 @@ class OSS(AlibabaCloudService):
                 )
 
     def _get_bucket_versioning(self, bucket):
-        """Get bucket versioning status using OSS SDK."""
+        """Get bucket versioning status (Enabled, Suspended or unset)."""
         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)
+            versioning_configuration = self._get_bucket_subresource(
+                bucket, "GetBucketVersioning", "versioning"
+            )
+            bucket.versioning_status = str(versioning_configuration.get("Status") or "")
         except Exception as error:
             logger.error(
                 f"{bucket.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"

tests/providers/alibabacloud/services/oss/oss_service_test.py

@@ -141,47 +141,80 @@ def _build_bucket(name="prowler-test"):
     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
+def _mock_subresource_response(root_element, content):
+    """Mimic the dict the OSS SDK execute path returns for XML bodies."""
+    return {"headers": {}, "statusCode": 200, "body": {root_element: content}}
 
+
+def test_get_bucket_subresource_calls_execute_and_unwraps_root():
     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"}}
-            )
-        )
+    oss_client.execute.return_value = _mock_subresource_response(
+        "VersioningConfiguration", {"Status": "Enabled"}
+    )
+    service.session.client.return_value = oss_client
+
+    result = service._get_bucket_subresource(
+        bucket, "GetBucketVersioning", "versioning"
+    )
+
+    assert result == {"Status": "Enabled"}
+    service.session.client.assert_called_once_with("oss", bucket.region)
+    params, request, _ = oss_client.execute.call_args.args
+    assert params.action == "GetBucketVersioning"
+    assert params.pathname == "/?versioning"
+    assert params.method == "GET"
+    assert params.style == "ROA"
+    assert params.body_type == "xml"
+    assert request.host_map == {"bucket": bucket.name}
+
+
+def test_get_bucket_subresource_returns_empty_dict_for_empty_root():
+    service = _build_oss_service()
+    bucket = _build_bucket()
+    oss_client = MagicMock()
+    oss_client.execute.return_value = _mock_subresource_response(
+        "VersioningConfiguration", None
+    )
+    service.session.client.return_value = oss_client
+
+    assert (
+        service._get_bucket_subresource(bucket, "GetBucketVersioning", "versioning")
+        == {}
+    )
+
+
+def test_get_bucket_encryption_parses_aes256_rule():
+    service = _build_oss_service()
+    bucket = _build_bucket()
+    oss_client = MagicMock()
+    oss_client.execute.return_value = _mock_subresource_response(
+        "ServerSideEncryptionRule",
+        {"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",
-                    }
-                }
-            )
-        )
+    oss_client.execute.return_value = _mock_subresource_response(
+        "ServerSideEncryptionRule",
+        {
+            "ApplyServerSideEncryptionByDefault": {
+                "SSEAlgorithm": "KMS",
+                "KMSMasterKeyID": "00000000-1111-2222-3333-444444444444",
+                "KMSDataEncryption": "SM4",
+            }
+        },
     )
     service.session.client.return_value = oss_client
 
@@ -198,7 +231,7 @@ def test_get_bucket_encryption_no_rule_is_not_logged_as_error():
     service = _build_oss_service()
     bucket = _build_bucket()
     oss_client = MagicMock()
-    oss_client.get_bucket_encryption.side_effect = TeaException(
+    oss_client.execute.side_effect = TeaException(
         {
             "code": "NoSuchServerSideEncryptionRule",
             "message": "No encryption rules are configured for this bucket.",
@@ -220,7 +253,7 @@ 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")
+    oss_client.execute.side_effect = RuntimeError("boom")
     service.session.client.return_value = oss_client
 
     with patch(
@@ -230,3 +263,143 @@ def test_get_bucket_encryption_unexpected_error_is_logged():
 
     mock_logger.error.assert_called_once()
     assert bucket.encryption_algorithm == ""
+
+
+def test_get_bucket_logging_parses_target():
+    service = _build_oss_service()
+    bucket = _build_bucket()
+    oss_client = MagicMock()
+    oss_client.execute.return_value = _mock_subresource_response(
+        "BucketLoggingStatus",
+        {"LoggingEnabled": {"TargetBucket": "log-bucket", "TargetPrefix": "logs/"}},
+    )
+    service.session.client.return_value = oss_client
+
+    service._get_bucket_logging(bucket)
+
+    assert bucket.logging_enabled is True
+    assert bucket.logging_target_bucket == "log-bucket"
+    assert bucket.logging_target_prefix == "logs/"
+
+
+def test_get_bucket_logging_disabled_when_no_target():
+    service = _build_oss_service()
+    bucket = _build_bucket()
+    oss_client = MagicMock()
+    oss_client.execute.return_value = _mock_subresource_response(
+        "BucketLoggingStatus", None
+    )
+    service.session.client.return_value = oss_client
+
+    service._get_bucket_logging(bucket)
+
+    assert bucket.logging_enabled is False
+    assert bucket.logging_target_bucket == ""
+
+
+def test_get_bucket_versioning_parses_suspended_status():
+    service = _build_oss_service()
+    bucket = _build_bucket()
+    oss_client = MagicMock()
+    oss_client.execute.return_value = _mock_subresource_response(
+        "VersioningConfiguration", {"Status": "Suspended"}
+    )
+    service.session.client.return_value = oss_client
+
+    service._get_bucket_versioning(bucket)
+
+    assert bucket.versioning_status == "Suspended"
+
+
+def test_get_bucket_acl_parses_grant():
+    service = _build_oss_service()
+    bucket = _build_bucket()
+    oss_client = MagicMock()
+    oss_client.execute.return_value = _mock_subresource_response(
+        "AccessControlPolicy",
+        {"Owner": {"ID": "1234567890"}, "AccessControlList": {"Grant": "public-read"}},
+    )
+    service.session.client.return_value = oss_client
+
+    service._get_bucket_acl(bucket)
+
+    assert bucket.acl == "public-read"
+
+
+def test_get_bucket_acl_parses_private_grant_value():
+    service = _build_oss_service()
+    bucket = _build_bucket()
+    oss_client = MagicMock()
+    oss_client.execute.return_value = _mock_subresource_response(
+        "AccessControlPolicy",
+        {"Owner": {"ID": "1234567890"}, "AccessControlList": {"Grant": "private"}},
+    )
+    service.session.client.return_value = oss_client
+
+    service._get_bucket_acl(bucket)
+
+    assert bucket.acl == "private"
+
+
+def test_get_bucket_acl_defaults_to_private_without_grant():
+    service = _build_oss_service()
+    bucket = _build_bucket()
+    oss_client = MagicMock()
+    oss_client.execute.return_value = _mock_subresource_response(
+        "AccessControlPolicy", {"Owner": {"ID": "1234567890"}}
+    )
+    service.session.client.return_value = oss_client
+
+    service._get_bucket_acl(bucket)
+
+    assert bucket.acl == "private"
+
+
+def test_get_bucket_subresource_with_real_sdk_client_unwraps_xml_root():
+    """Regression test against the SDK deserialization the helper works around.
+
+    The generated ``get_bucket_*`` methods return empty response models for
+    XML bodies (root element kept by the gateway, dropped by the models). Drive
+    the real client with only the HTTP call mocked to make sure the helper still
+    returns the configuration after SDK upgrades.
+    """
+    import io
+
+    import darabonba.core as dara_core
+    from alibabacloud_oss20190517.client import Client as OssClient
+    from alibabacloud_tea_openapi import models as open_api_models
+
+    service = _build_oss_service()
+    bucket = _build_bucket()
+    service.session.client.return_value = OssClient(
+        open_api_models.Config(
+            access_key_id="AKID",
+            access_key_secret="SECRET",
+            endpoint="oss-ap-southeast-1.aliyuncs.com",
+            region_id="ap-southeast-1",
+        )
+    )
+    xml = (
+        b'<?xml version="1.0" encoding="UTF-8"?>'
+        b"<ServerSideEncryptionRule><ApplyServerSideEncryptionByDefault>"
+        b"<SSEAlgorithm>KMS</SSEAlgorithm>"
+        b"<KMSMasterKeyID>00000000-1111-2222-3333-444444444444</KMSMasterKeyID>"
+        b"</ApplyServerSideEncryptionByDefault></ServerSideEncryptionRule>"
+    )
+
+    class FakeHttpResponse:
+        status_code = 200
+        headers = {"content-type": "application/xml"}
+        body = io.BytesIO(xml)
+
+    with patch.object(dara_core.DaraCore, "do_action", return_value=FakeHttpResponse()):
+        result = service._get_bucket_subresource(
+            bucket, "GetBucketEncryption", "encryption"
+        )
+
+    assert result == {
+        "ApplyServerSideEncryptionByDefault": {
+            "SSEAlgorithm": "KMS",
+            "KMSMasterKeyID": "00000000-1111-2222-3333-444444444444",
+        }
+    }