feat(stackit): add ske_cluster_no_public_endpoint check (#11943)

Co-authored-by: Hugo P.Brito <hugopbrit@gmail.com>

tests/providers/stackit/services/ske/ske_cluster_no_public_endpoint/ske_cluster_no_public_endpoint_test.py

@@ -0,0 +1,149 @@
+from unittest import mock
+
+from prowler.providers.stackit.services.ske.ske_service import Cluster
+from tests.providers.stackit.stackit_fixtures import (
+    STACKIT_PROJECT_ID,
+    set_mocked_stackit_provider,
+)
+
+
+class Test_ske_cluster_no_public_endpoint:
+    def _run_check(self, ske_client):
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_stackit_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.stackit.services.ske.ske_service.SKEService",
+                new=ske_client,
+            ) as service_client,
+            mock.patch(
+                "prowler.providers.stackit.services.ske.ske_client.ske_client",
+                new=service_client,
+            ),
+        ):
+            from prowler.providers.stackit.services.ske.ske_cluster_no_public_endpoint.ske_cluster_no_public_endpoint import (
+                ske_cluster_no_public_endpoint,
+            )
+
+            check = ske_cluster_no_public_endpoint()
+            return check.execute()
+
+    def _cluster(self, name="test-cluster", **kwargs):
+        defaults = {
+            "id": name,
+            "name": name,
+            "project_id": STACKIT_PROJECT_ID,
+            "region": "eu01",
+        }
+        defaults.update(kwargs)
+        return Cluster(**defaults)
+
+    def test_no_clusters(self):
+        ske_client = mock.MagicMock
+        ske_client.clusters = []
+
+        result = self._run_check(ske_client)
+        assert len(result) == 0
+
+    def test_cluster_without_acl_is_public(self):
+        ske_client = mock.MagicMock
+        ske_client.clusters = [self._cluster(name="open-cluster", acl_enabled=False)]
+
+        result = self._run_check(ske_client)
+        assert len(result) == 1
+        assert result[0].status == "FAIL"
+        assert (
+            result[0].status_extended
+            == "SKE cluster open-cluster exposes its Kubernetes API endpoint to the "
+            "internet because the ACL extension is not enabled."
+        )
+        assert result[0].resource_id == "open-cluster"
+        assert result[0].resource_name == "open-cluster"
+        assert result[0].project_id == STACKIT_PROJECT_ID
+        assert result[0].location == "eu01"
+
+    def test_cluster_with_unrestricted_ipv4_cidr_is_public(self):
+        ske_client = mock.MagicMock
+        ske_client.clusters = [
+            self._cluster(
+                name="wide-cluster",
+                acl_enabled=True,
+                allowed_cidrs=["10.0.0.0/8", "0.0.0.0/0"],
+            )
+        ]
+
+        result = self._run_check(ske_client)
+        assert len(result) == 1
+        assert result[0].status == "FAIL"
+        assert (
+            result[0].status_extended
+            == "SKE cluster wide-cluster exposes its Kubernetes API endpoint to the "
+            "internet because its ACL allows unrestricted access from 0.0.0.0/0."
+        )
+
+    def test_cluster_with_unrestricted_ipv6_cidr_is_public(self):
+        ske_client = mock.MagicMock
+        ske_client.clusters = [
+            self._cluster(name="v6-cluster", acl_enabled=True, allowed_cidrs=["::/0"])
+        ]
+
+        result = self._run_check(ske_client)
+        assert len(result) == 1
+        assert result[0].status == "FAIL"
+        assert "unrestricted access from ::/0." in result[0].status_extended
+
+    def test_cluster_with_restricted_acl_passes(self):
+        ske_client = mock.MagicMock
+        ske_client.clusters = [
+            self._cluster(
+                name="locked-cluster",
+                access_scope="PUBLIC",
+                acl_enabled=True,
+                allowed_cidrs=["10.0.0.0/8", "192.0.2.0/24"],
+            )
+        ]
+
+        result = self._run_check(ske_client)
+        assert len(result) == 1
+        assert result[0].status == "PASS"
+        assert (
+            result[0].status_extended
+            == "SKE cluster locked-cluster restricts access to its Kubernetes API "
+            "endpoint to 2 allowed CIDR(s)."
+        )
+
+    def test_cluster_with_private_control_plane_passes_without_acl(self):
+        ske_client = mock.MagicMock
+        ske_client.clusters = [
+            self._cluster(name="sna-cluster", access_scope="SNA", acl_enabled=False)
+        ]
+
+        result = self._run_check(ske_client)
+        assert len(result) == 1
+        assert result[0].status == "PASS"
+        assert (
+            result[0].status_extended
+            == "SKE cluster sna-cluster has a private control plane and its Kubernetes "
+            "API endpoint is not reachable from the internet."
+        )
+
+    def test_mixed_clusters_report_independently(self):
+        ske_client = mock.MagicMock
+        ske_client.clusters = [
+            self._cluster(name="open-cluster", acl_enabled=False),
+            self._cluster(
+                name="locked-cluster", acl_enabled=True, allowed_cidrs=["10.0.0.0/8"]
+            ),
+            self._cluster(name="sna-cluster", access_scope="SNA"),
+        ]
+
+        result = self._run_check(ske_client)
+        assert len(result) == 3
+        assert [report.status for report in result] == ["FAIL", "PASS", "PASS"]
+        assert [report.resource_name for report in result] == [
+            "open-cluster",
+            "locked-cluster",
+            "sna-cluster",
+        ]

tests/providers/stackit/services/ske/ske_service_test.py

@@ -0,0 +1,526 @@
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from prowler.providers.stackit.exceptions.exceptions import StackITInvalidTokenError
+from prowler.providers.stackit.services.ske.ske_service import Cluster, SKEService
+from tests.providers.stackit.stackit_fixtures import (
+    STACKIT_PROJECT_ID,
+    set_mocked_stackit_provider,
+)
+
+
+def mock_ske_fetch_all_regions(_):
+    """Mock the _fetch_all_regions method to avoid real API calls."""
+
+
+@patch(
+    "prowler.providers.stackit.services.ske.ske_service.SKEService._fetch_all_regions",
+    new=mock_ske_fetch_all_regions,
+)
+class Test_SKE_Service:
+    def test_service_initialization(self):
+        """Test that the SKE service initializes correctly."""
+        ske_service = SKEService(set_mocked_stackit_provider())
+
+        assert ske_service.project_id == STACKIT_PROJECT_ID
+        assert ske_service.service_account_key_path is not None
+        assert isinstance(ske_service.clusters, list)
+
+    def test_service_service_account_key_path(self):
+        """Test that the service correctly extracts the SA key path from provider."""
+        custom_path = "/tmp/custom-sa.json"
+        provider = set_mocked_stackit_provider(service_account_key_path=custom_path)
+        ske_service = SKEService(provider)
+        assert ske_service.service_account_key_path == custom_path
+
+    def test_service_requests_ske_regional_clients(self):
+        """Test that the service asks the provider for SKE regional clients."""
+        provider = set_mocked_stackit_provider()
+        SKEService(provider)
+        provider.generate_regional_clients.assert_called_once_with("ske")
+
+
+class Test_SKE_Service_ExtractItems:
+    """Tests for the _extract_items response normalization helper."""
+
+    def test_dict_response(self):
+        assert SKEService._extract_items({"items": [1, 2]}, "list_clusters") == [1, 2]
+
+    def test_dict_response_without_items_key(self):
+        assert SKEService._extract_items({}, "list_clusters") == []
+
+    def test_list_response(self):
+        assert SKEService._extract_items(["cluster"], "list_clusters") == ["cluster"]
+
+    def test_model_with_items_attribute(self):
+        response = MagicMock(spec=["items"])
+        response.items = ["cluster"]
+        assert SKEService._extract_items(response, "list_clusters") == ["cluster"]
+
+    def test_model_with_callable_items_is_rejected(self):
+        # A bare MagicMock exposes ``items`` as a callable, which must not be
+        # mistaken for the items list.
+        assert SKEService._extract_items(MagicMock(), "list_clusters") == []
+
+    def test_unexpected_response_type(self):
+        assert SKEService._extract_items(object(), "list_clusters") == []
+
+
+class Test_SKE_Service_GetField:
+    """Tests for the _get_field dict/model accessor."""
+
+    def test_none_item_returns_default(self):
+        assert SKEService._get_field(None, "enabled", default="fallback") == "fallback"
+
+    def test_dict_alias_key_is_matched(self):
+        item = {"allowedCidrs": ["10.0.0.0/8"]}
+        assert SKEService._get_field(item, "allowed_cidrs", "allowedCidrs") == [
+            "10.0.0.0/8"
+        ]
+
+    def test_dict_missing_keys_returns_default(self):
+        assert SKEService._get_field({}, "enabled", default=[]) == []
+
+    def test_dict_false_value_is_returned_not_treated_as_missing(self):
+        assert (
+            SKEService._get_field({"enabled": False}, "enabled", default=True) is False
+        )
+
+    def test_model_attribute_is_read(self):
+        item = MagicMock(spec=["enabled"])
+        item.enabled = True
+        assert SKEService._get_field(item, "enabled") is True
+
+    def test_model_false_attribute_is_returned(self):
+        item = MagicMock(spec=["enabled"])
+        item.enabled = False
+        assert SKEService._get_field(item, "enabled", default=True) is False
+
+    def test_model_missing_attribute_returns_default(self):
+        item = MagicMock(spec=["other"])
+        assert SKEService._get_field(item, "enabled", default="fallback") == "fallback"
+
+
+class Test_SKE_Service_ParseAccessScope:
+    """Tests for _parse_access_scope."""
+
+    def test_missing_network_returns_none(self):
+        assert SKEService._parse_access_scope({}) is None
+
+    def test_missing_control_plane_returns_none(self):
+        assert SKEService._parse_access_scope({"network": {"id": "net"}}) is None
+
+    def test_missing_access_scope_returns_none(self):
+        assert SKEService._parse_access_scope({"network": {"controlPlane": {}}}) is None
+
+    def test_camel_case_dict_scope(self):
+        cluster_data = {"network": {"controlPlane": {"accessScope": "SNA"}}}
+        assert SKEService._parse_access_scope(cluster_data) == "SNA"
+
+    def test_snake_case_dict_scope(self):
+        cluster_data = {"network": {"control_plane": {"access_scope": "PUBLIC"}}}
+        assert SKEService._parse_access_scope(cluster_data) == "PUBLIC"
+
+    def test_sdk_enum_is_normalized_to_wire_value(self):
+        # ``AccessScope`` is a ``str`` Enum whose ``str()`` renders as
+        # "AccessScope.SNA"; the parser must yield the "SNA" wire value.
+        from stackit.ske.models.access_scope import AccessScope
+
+        cluster_data = {"network": {"controlPlane": {"accessScope": AccessScope.SNA}}}
+        assert SKEService._parse_access_scope(cluster_data) == "SNA"
+
+
+class Test_SKE_Service_ParseAcl:
+    """Tests for _parse_acl."""
+
+    def test_missing_extensions_reports_no_acl(self):
+        assert SKEService._parse_acl({}) == (False, [])
+
+    def test_extensions_without_acl_reports_no_acl(self):
+        assert SKEService._parse_acl({"extensions": {"dns": {}}}) == (False, [])
+
+    def test_disabled_acl_keeps_its_cidrs(self):
+        cluster_data = {
+            "extensions": {"acl": {"enabled": False, "allowedCidrs": ["10.0.0.0/8"]}}
+        }
+        assert SKEService._parse_acl(cluster_data) == (False, ["10.0.0.0/8"])
+
+    def test_enabled_acl_camel_case_cidrs(self):
+        cluster_data = {
+            "extensions": {"acl": {"enabled": True, "allowedCidrs": ["10.0.0.0/8"]}}
+        }
+        assert SKEService._parse_acl(cluster_data) == (True, ["10.0.0.0/8"])
+
+    def test_enabled_acl_snake_case_cidrs(self):
+        cluster_data = {
+            "extensions": {"acl": {"enabled": True, "allowed_cidrs": ["10.0.0.0/8"]}}
+        }
+        assert SKEService._parse_acl(cluster_data) == (True, ["10.0.0.0/8"])
+
+    def test_enabled_acl_without_cidrs(self):
+        cluster_data = {"extensions": {"acl": {"enabled": True}}}
+        assert SKEService._parse_acl(cluster_data) == (True, [])
+
+
+class Test_SKE_Service_HandleApiCall:
+    """Tests for the centralized _handle_api_call wrapper."""
+
+    def _service(self):
+        service = object.__new__(SKEService)
+        service.provider = MagicMock()
+        return service
+
+    def test_returns_api_response(self):
+        service = self._service()
+        api_function = MagicMock(return_value={"items": []})
+
+        assert service._handle_api_call(api_function, project_id="p") == {"items": []}
+        api_function.assert_called_once_with(project_id="p")
+
+    def test_delegates_errors_to_provider_and_reraises(self):
+        service = self._service()
+        error = ValueError("boom")
+        api_function = MagicMock(side_effect=error)
+
+        with pytest.raises(ValueError):
+            service._handle_api_call(api_function)
+
+        service.provider.handle_api_error.assert_called_once_with(error)
+
+
+class Test_SKE_Service_ListClusters:
+    """Tests for _list_clusters."""
+
+    def _service(self):
+        service = object.__new__(SKEService)
+        service.provider = MagicMock()
+        service.project_id = STACKIT_PROJECT_ID
+        service.clusters = []
+        return service
+
+    def test_list_clusters_without_client_is_noop(self):
+        """A missing regional client is logged and skipped, not fatal."""
+        service = self._service()
+
+        service._list_clusters(None, "eu01")
+
+        assert service.clusters == []
+
+    def test_list_clusters_populates_clusters(self):
+        service = self._service()
+        client = MagicMock()
+        client.list_clusters.return_value = {
+            "items": [
+                {"name": "open-cluster"},
+                {
+                    "name": "locked-cluster",
+                    "extensions": {
+                        "acl": {"enabled": True, "allowedCidrs": ["10.0.0.0/8"]}
+                    },
+                    "network": {"controlPlane": {"accessScope": "PUBLIC"}},
+                },
+            ]
+        }
+
+        service._list_clusters(client, "eu01")
+
+        assert [cluster.name for cluster in service.clusters] == [
+            "open-cluster",
+            "locked-cluster",
+        ]
+        # SKE has no separate cluster id; the name doubles as the identifier.
+        assert service.clusters[0].id == "open-cluster"
+        assert service.clusters[0].acl_enabled is False
+        assert service.clusters[0].access_scope is None
+        assert service.clusters[1].acl_enabled is True
+        assert service.clusters[1].allowed_cidrs == ["10.0.0.0/8"]
+        assert service.clusters[1].access_scope == "PUBLIC"
+        assert all(
+            cluster.project_id == STACKIT_PROJECT_ID and cluster.region == "eu01"
+            for cluster in service.clusters
+        )
+
+    def test_cluster_processing_error_is_skipped(self):
+        """A cluster that raises while being read is skipped, not fatal."""
+
+        class MalformedCluster:
+            @property
+            def name(self):
+                raise ValueError("malformed cluster")
+
+        service = self._service()
+        client = MagicMock()
+        client.list_clusters.return_value = {"items": [MalformedCluster()]}
+
+        service._list_clusters(client, "eu01")
+
+        assert service.clusters == []
+
+
+class Test_SKE_Service_FetchAllRegions:
+    """Tests for the region fetch loop and its error semantics."""
+
+    class _NotFound(Exception):
+        status = 404
+
+    class _Unauthorized(Exception):
+        status = 401
+
+    class _ServerError(Exception):
+        status = 500
+
+    class _ServiceNotEnabled(Exception):
+        # Body copied from a live SKE 403 for a region the project never enabled.
+        status = 403
+        body = (
+            '{"timestamp":"2026-08-20T09:51:33Z","status":403,'
+            '"error":"Forbidden","message":"Service not enabled"}'
+        )
+
+    class _Forbidden(Exception):
+        status = 403
+        body = '{"status":403,"error":"Forbidden","message":"Access denied"}'
+
+    def _service(self, regional_clients):
+        from prowler.providers.stackit.stackit_provider import StackitProvider
+
+        service = object.__new__(SKEService)
+        service.provider = MagicMock()
+        # Reuse the real centralized error handler so 401/403/404 semantics
+        # match production.
+        service.provider.handle_api_error = StackitProvider.handle_api_error
+        service.project_id = STACKIT_PROJECT_ID
+        service.regional_clients = regional_clients
+        service.clusters = []
+        return service
+
+    def _good_client(self, cluster_name="cluster-eu01"):
+        client = MagicMock()
+        client.list_clusters.return_value = {"items": [{"name": cluster_name}]}
+        return client
+
+    def _failing_client(self, error):
+        client = MagicMock()
+        client.list_clusters.side_effect = error
+        return client
+
+    def test_skips_region_where_project_is_absent(self):
+        service = self._service(
+            {
+                "eu01": self._good_client(),
+                "eu02": self._failing_client(self._NotFound()),
+            }
+        )
+
+        service._fetch_all_regions()
+
+        # eu01 cluster is collected; the eu02 404 is skipped silently.
+        assert [cluster.name for cluster in service.clusters] == ["cluster-eu01"]
+
+    def test_skips_region_where_ske_is_not_enabled(self):
+        service = self._service(
+            {
+                "eu01": self._good_client(),
+                "eu02": self._failing_client(self._ServiceNotEnabled()),
+            }
+        )
+
+        service._fetch_all_regions()
+
+        # A project may enable SKE per region; the eu02 403 "Service not
+        # enabled" is a skip, not the credentials failure a bare 403 implies.
+        assert [cluster.name for cluster in service.clusters] == ["cluster-eu01"]
+
+    def test_permission_denied_still_aborts_the_scan(self):
+        service = self._service({"eu01": self._failing_client(self._Forbidden())})
+
+        with pytest.raises(StackITInvalidTokenError):
+            service._fetch_all_regions()
+
+    def test_invalid_token_aborts_the_scan(self):
+        service = self._service({"eu01": self._failing_client(self._Unauthorized())})
+
+        with pytest.raises(StackITInvalidTokenError):
+            service._fetch_all_regions()
+
+    def test_unexpected_error_propagates(self):
+        service = self._service({"eu01": self._failing_client(self._ServerError())})
+
+        with pytest.raises(self._ServerError):
+            service._fetch_all_regions()
+
+
+class Test_SKE_Service_SdkModelShapes:
+    """Regression coverage for object-shaped (SDK model) API responses.
+
+    ``_extract_items`` accepts both raw dicts and SDK models, so the cluster
+    parsing has to read either shape. Reading only dict keys would silently drop
+    the ACL of an SDK-model cluster and report a restricted cluster as exposed,
+    or worse, miss the ACL of an exposed one.
+    """
+
+    @staticmethod
+    def _sdk_cluster(name, acl=None, access_scope=None):
+        """Build a real ``stackit.ske`` Cluster model with optional ACL and scope."""
+        from stackit.ske.models.cluster import Cluster as SDKCluster
+        from stackit.ske.models.extension import Extension
+        from stackit.ske.models.image import Image
+        from stackit.ske.models.kubernetes import Kubernetes
+        from stackit.ske.models.machine import Machine
+        from stackit.ske.models.network import Network
+        from stackit.ske.models.nodepool import Nodepool
+        from stackit.ske.models.v2_control_plane_network import V2ControlPlaneNetwork
+        from stackit.ske.models.volume import Volume
+
+        nodepool = Nodepool(
+            name="np",
+            availabilityZones=["eu01-1"],
+            maximum=1,
+            minimum=1,
+            machine=Machine(image=Image(name="flatcar", version="1.0"), type="g1.2"),
+            volume=Volume(size=20),
+        )
+        return SDKCluster(
+            name=name,
+            kubernetes=Kubernetes(version="1.31.0"),
+            nodepools=[nodepool],
+            extensions=Extension(acl=acl) if acl is not None else None,
+            network=(
+                Network(controlPlane=V2ControlPlaneNetwork(accessScope=access_scope))
+                if access_scope is not None
+                else None
+            ),
+        )
+
+    @staticmethod
+    def _sdk_acl(allowed_cidrs, enabled=True):
+        """Build a real ``stackit.ske`` ACL extension model."""
+        from stackit.ske.models.acl import ACL
+
+        return ACL(allowedCidrs=allowed_cidrs, enabled=enabled)
+
+    def test_sdk_model_acl_is_parsed(self):
+        cluster = self._sdk_cluster("locked-clst", acl=self._sdk_acl(["10.0.0.0/8"]))
+
+        assert SKEService._parse_acl(cluster) == (True, ["10.0.0.0/8"])
+
+    def test_sdk_model_without_extensions_reports_no_acl(self):
+        cluster = self._sdk_cluster("open-clst")
+
+        assert SKEService._parse_acl(cluster) == (False, [])
+        assert SKEService._parse_access_scope(cluster) is None
+
+    def test_sdk_model_access_scope_enum_is_normalized(self):
+        cluster = self._sdk_cluster("sna-clst", access_scope="SNA")
+
+        assert SKEService._parse_access_scope(cluster) == "SNA"
+
+    def test_list_clusters_with_sdk_response_and_model_items(self):
+        from stackit.ske.models.list_clusters_response import ListClustersResponse
+
+        service = object.__new__(SKEService)
+        service.provider = MagicMock()
+        service.project_id = STACKIT_PROJECT_ID
+        service.clusters = []
+
+        client = MagicMock()
+        client.list_clusters.return_value = ListClustersResponse(
+            items=[
+                self._sdk_cluster("open-clst"),
+                self._sdk_cluster("locked-clst", acl=self._sdk_acl(["10.0.0.0/8"])),
+            ]
+        )
+
+        service._list_clusters(client, "eu01")
+
+        assert [cluster.name for cluster in service.clusters] == [
+            "open-clst",
+            "locked-clst",
+        ]
+        # An object-shaped ACL must not be dropped: the restricted cluster is
+        # not reported as internet-exposed, and the open one still is.
+        assert service.clusters[0].has_public_endpoint() is True
+        assert service.clusters[1].acl_enabled is True
+        assert service.clusters[1].allowed_cidrs == ["10.0.0.0/8"]
+        assert service.clusters[1].has_public_endpoint() is False
+
+    def test_dict_and_sdk_model_shapes_parse_identically(self):
+        """The same cluster expressed as a dict or an SDK model must agree."""
+        sdk_cluster = self._sdk_cluster(
+            "locked-clst",
+            acl=self._sdk_acl(["10.0.0.0/8"]),
+            access_scope="PUBLIC",
+        )
+        dict_cluster = {
+            "name": "locked-clst",
+            "extensions": {"acl": {"enabled": True, "allowedCidrs": ["10.0.0.0/8"]}},
+            "network": {"controlPlane": {"accessScope": "PUBLIC"}},
+        }
+
+        assert SKEService._parse_acl(sdk_cluster) == SKEService._parse_acl(dict_cluster)
+        assert SKEService._parse_access_scope(
+            sdk_cluster
+        ) == SKEService._parse_access_scope(dict_cluster)
+
+
+class Test_SKE_Cluster_Model:
+    """Tests for the Cluster public-endpoint logic."""
+
+    def _cluster(self, **kwargs):
+        defaults = {
+            "id": "test-cluster",
+            "name": "test-cluster",
+            "project_id": STACKIT_PROJECT_ID,
+            "region": "eu01",
+        }
+        defaults.update(kwargs)
+        return Cluster(**defaults)
+
+    def test_private_control_plane_passes_regardless_of_acl(self):
+        cluster = self._cluster(access_scope="SNA", acl_enabled=False)
+
+        assert cluster.has_private_control_plane() is True
+        assert cluster.has_public_endpoint() is False
+
+    def test_disabled_acl_is_publicly_reachable(self):
+        cluster = self._cluster(acl_enabled=False)
+
+        assert cluster.has_private_control_plane() is False
+        assert cluster.has_public_endpoint() is True
+
+    def test_public_scope_with_restricted_acl_is_not_publicly_reachable(self):
+        cluster = self._cluster(
+            access_scope="PUBLIC", acl_enabled=True, allowed_cidrs=["10.0.0.0/8"]
+        )
+
+        assert cluster.unrestricted_cidrs() == []
+        assert cluster.has_public_endpoint() is False
+
+    @pytest.mark.parametrize("unrestricted_cidr", ["0.0.0.0/0", "::/0"])
+    def test_unrestricted_cidr_in_allowlist_is_publicly_reachable(
+        self, unrestricted_cidr
+    ):
+        cluster = self._cluster(
+            acl_enabled=True, allowed_cidrs=["10.0.0.0/8", unrestricted_cidr]
+        )
+
+        assert cluster.unrestricted_cidrs() == [unrestricted_cidr]
+        assert cluster.has_public_endpoint() is True
+
+    def test_enabled_acl_with_empty_allowlist_is_not_publicly_reachable(self):
+        cluster = self._cluster(acl_enabled=True, allowed_cidrs=[])
+
+        assert cluster.has_public_endpoint() is False
+
+    def test_unset_access_scope_falls_back_to_the_acl(self):
+        assert (
+            self._cluster(
+                access_scope=None, acl_enabled=True, allowed_cidrs=["10.0.0.0/8"]
+            ).has_public_endpoint()
+            is False
+        )
+        assert (
+            self._cluster(access_scope=None, acl_enabled=False).has_public_endpoint()
+            is True
+        )

tests/providers/stackit/stackit_provider_test.py

@@ -276,6 +276,40 @@ class TestStackITProviderTestConnection:
             (None, self.KEY_CONTENT, self.PROJECT_ID)
         ]
 
+    def test_connection_resource_manager_403_does_not_fail_service_discovery(
+        self, fake_stackit_resourcemanager
+    ):
+        class Http403Error(Exception):
+            status = 403
+
+        fake_stackit_resourcemanager.error = Http403Error()
+
+        with patch.object(stackit_provider_module.logger, "warning") as warning:
+            connection = StackitProvider.test_connection(
+                project_id=self.PROJECT_ID,
+                service_account_key_path=self.KEY_PATH,
+            )
+
+        assert connection == Connection(is_connected=True)
+        warning.assert_called_once_with(
+            "StackIT test_connection: Resource Manager access could not be verified "
+            "(403). Service permissions will be checked during discovery."
+        )
+
+    def test_connection_resource_manager_401_still_fails(
+        self, fake_stackit_resourcemanager
+    ):
+        class Http401Error(Exception):
+            status = 401
+
+        fake_stackit_resourcemanager.error = Http401Error()
+
+        with pytest.raises(StackITInvalidTokenError):
+            StackitProvider.test_connection(
+                project_id=self.PROJECT_ID,
+                service_account_key_path=self.KEY_PATH,
+            )
+
     def test_connection_returns_error_when_raise_on_exception_is_false(
         self, fake_stackit_resourcemanager
     ):