feat(aws): Update regions for AWS services (#12472 )

Co-authored-by: prowler-bot <179230569+prowler-bot@users.noreply.github.com>
feat(stackit): add ske_cluster_no_public_endpoint check (#11943 )
2026-08-21 05:13:00 +00:00 · 2026-08-20 16:02:56 +01:00 · 2026-08-20 13:01:01 +01:00 · 2026-08-20 10:52:13 +01:00 · 2026-08-20 10:48:16 +02:00
26 changed files with 1623 additions and 83 deletions
@@ -61,6 +61,7 @@ jobs:
            api.github.com:443
            github.com:443
            objects.githubusercontent.com:443
+            release-assets.githubusercontent.com:443
            pypi.org:443
            files.pythonhosted.org:443

@@ -35,6 +35,7 @@ jobs:
            api.github.com:443
            github.com:443
            objects.githubusercontent.com:443
+            release-assets.githubusercontent.com:443
            pypi.org:443
            files.pythonhosted.org:443

@@ -57,6 +57,7 @@ jobs:
          egress-policy: block
          allowed-endpoints: >
            github.com:443
+            release-assets.githubusercontent.com:443
            pypi.org:443
            files.pythonhosted.org:443

@@ -13,7 +13,7 @@ StackIT support in Prowler is community-maintained. For commercial support or to
 Before running Prowler with the StackIT provider, ensure you have:

 1. A StackIT account with at least one project
-2. A StackIT service account key file with permissions on the project (`iaas.viewer` is enough for the currently shipped IaaS checks; `project.owner` works for any future service). See the [Authentication guide](/user-guide/providers/stackit/authentication) for the full setup.
+2. A StackIT service account key file with project permissions for the services to scan. `iaas.viewer` is sufficient for the currently shipped IaaS checks. For SKE, assign the project-scoped `Reader` role as a broad read-only convenience, or create a custom role with `resource-manager.project.get`, `ske.cluster.list`, `ske.cluster.get`, and `ske.version.list`. The `resource-manager.project.get` permission is optional for scanning; it only lets Prowler verify and display the project name. See the [Authentication guide](/user-guide/providers/stackit/authentication) for the full setup.
 3. Access to Prowler CLI (see [Installation](/getting-started/installation/prowler-cli))

 ## Prowler CLI
@@ -124,7 +124,9 @@ Prowler outputs findings to the console and writes reports to the `output/` dire

 | Service | StackIT API | Description | Example Checks |
 |---------|-------------|-------------|----------------|
-| **IaaS** | `iaas` | Virtual machines, network interfaces, security groups | `iaas_security_group_ssh_unrestricted`, `iaas_security_group_rdp_unrestricted`, `iaas_security_group_database_unrestricted`, `iaas_security_group_all_traffic_unrestricted` |
+| **IaaS** | `iaas` | Virtual machines, network interfaces, security groups | `iaas_security_group_ssh_unrestricted`, `iaas_security_group_rdp_unrestricted`, `iaas_security_group_database_unrestricted`, `iaas_security_group_all_traffic_unrestricted`, `iaas_server_public_ip_attached` |
+| **Object Storage** | `objectstorage` | Buckets, object lock, retention policies, access keys | `objectstorage_bucket_object_lock_enabled`, `objectstorage_bucket_retention_policy`, `objectstorage_access_key_expiration` |
+| **Kubernetes Engine** | `ske` | Managed Kubernetes clusters | `ske_cluster_no_public_endpoint` |

 Additional services will be added in future releases. Track progress in the [Prowler release notes](https://github.com/prowler-cloud/prowler/releases).

@@ -136,6 +138,8 @@ If the scan fails with a 401 error, the service account key is no longer valid (

 ### Permission Errors

-If checks fail with a 403 error, the service account is missing the required role on the project. Re-check the role assignment in the StackIT portal (`iaas.viewer` is the minimum for the shipped IaaS checks).
+If the connection probe warns that Resource Manager access could not be verified with a 403 response, Prowler continues to service discovery. Grant the optional `resource-manager.project.get` permission to verify and display the project name.
+
+If a service call fails with a 403 response, the service account is missing a permission required by that service. `iaas.viewer` is sufficient for the shipped IaaS checks. The `ske_cluster_no_public_endpoint` check lists clusters and therefore requires `ske.cluster.list`; STACKIT least-privilege custom-role examples also include `ske.cluster.get` and `ske.version.list`. Service API calls remain authoritative and stop the scan when access is denied.

 For detailed setup steps, see the [Authentication guide](/user-guide/providers/stackit/authentication).
@@ -0,0 +1 @@
+`ske_cluster_no_public_endpoint` check for STACKIT provider, flagging SKE clusters whose Kubernetes API endpoint is reachable from the whole internet because the ACL extension is disabled or its allowed CIDR list contains `0.0.0.0/0` or `::/0`
@@ -0,0 +1 @@
+`vpc_security_group_open_egress` check for Huawei Cloud provider: VPC security groups do not allow open egress to the internet
@@ -3847,6 +3847,7 @@
        "aws": [
          "af-south-1",
          "ap-east-1",
+          "ap-east-2",
          "ap-northeast-1",
          "ap-northeast-2",
          "ap-northeast-3",
@@ -3856,7 +3857,9 @@
          "ap-southeast-2",
          "ap-southeast-3",
          "ap-southeast-4",
+          "ap-southeast-5",
          "ap-southeast-6",
+          "ap-southeast-7",
          "ca-central-1",
          "ca-west-1",
          "eu-central-1",
@@ -3870,6 +3873,7 @@
          "il-central-1",
          "me-central-1",
          "me-south-1",
+          "mx-central-1",
          "sa-east-1",
          "us-east-1",
          "us-east-2",
@@ -3952,6 +3956,7 @@
          "ca-west-1",
          "eu-central-1",
          "eu-north-1",
+          "eu-south-2",
          "eu-west-1",
          "eu-west-2",
          "eu-west-3",
@@ -4634,6 +4639,7 @@
      "regions": {
        "aws": [
          "ap-south-1",
+          "ap-southeast-2",
          "eu-west-1",
          "us-east-1",
          "us-west-2"
@@ -4778,7 +4784,9 @@
          "cn-north-1",
          "cn-northwest-1"
        ],
-        "aws-eusc": [],
+        "aws-eusc": [
+          "eusc-de-east-1"
+        ],
        "aws-us-gov": [
          "us-gov-east-1",
          "us-gov-west-1"
@@ -4963,6 +4971,7 @@
      "regions": {
        "aws": [
          "ap-northeast-1",
+          "ap-northeast-2",
          "ap-south-1",
          "ap-south-2",
          "ap-southeast-1",
@@ -4971,6 +4980,8 @@
          "ca-central-1",
          "ca-west-1",
          "eu-central-1",
+          "eu-central-2",
+          "eu-north-1",
          "eu-south-1",
          "eu-west-1",
          "eu-west-2",
@@ -6231,6 +6242,16 @@
        "aws-us-gov": []
      }
    },
+    "invoicing": {
+      "regions": {
+        "aws": [],
+        "aws-cn": [],
+        "aws-eusc": [
+          "eusc-de-east-1"
+        ],
+        "aws-us-gov": []
+      }
+    },
    "iot": {
      "regions": {
        "aws": [
@@ -6441,56 +6462,6 @@
        ]
      }
    },
-    "iotevents": {
-      "regions": {
-        "aws": [
-          "ap-northeast-1",
-          "ap-northeast-2",
-          "ap-south-1",
-          "ap-southeast-1",
-          "ap-southeast-2",
-          "ca-central-1",
-          "eu-central-1",
-          "eu-west-1",
-          "eu-west-2",
-          "us-east-1",
-          "us-east-2",
-          "us-west-2"
-        ],
-        "aws-cn": [
-          "cn-north-1"
-        ],
-        "aws-eusc": [],
-        "aws-us-gov": [
-          "us-gov-west-1"
-        ]
-      }
-    },
-    "iotevents-data": {
-      "regions": {
-        "aws": [
-          "ap-northeast-1",
-          "ap-northeast-2",
-          "ap-south-1",
-          "ap-southeast-1",
-          "ap-southeast-2",
-          "ca-central-1",
-          "eu-central-1",
-          "eu-west-1",
-          "eu-west-2",
-          "us-east-1",
-          "us-east-2",
-          "us-west-2"
-        ],
-        "aws-cn": [
-          "cn-north-1"
-        ],
-        "aws-eusc": [],
-        "aws-us-gov": [
-          "us-gov-west-1"
-        ]
-      }
-    },
    "iotfleetwise": {
      "regions": {
        "aws": [
@@ -7760,6 +7731,7 @@
          "ap-southeast-5",
          "ap-southeast-7",
          "ca-central-1",
+          "ca-west-1",
          "eu-central-1",
          "eu-north-1",
          "eu-west-1",
@@ -8793,7 +8765,10 @@
        ],
        "aws-cn": [],
        "aws-eusc": [],
-        "aws-us-gov": []
+        "aws-us-gov": [
+          "us-gov-east-1",
+          "us-gov-west-1"
+        ]
      }
    },
    "odb": {
@@ -8826,6 +8801,7 @@
    "omics": {
      "regions": {
        "aws": [
+          "ap-northeast-1",
          "ap-northeast-2",
          "ap-southeast-1",
          "eu-central-1",
@@ -8833,6 +8809,7 @@
          "eu-west-2",
          "il-central-1",
          "us-east-1",
+          "us-east-2",
          "us-west-2"
        ],
        "aws-cn": [],
@@ -9026,6 +9003,7 @@
          "eu-south-2",
          "eu-west-1",
          "eu-west-2",
+          "eu-west-3",
          "sa-east-1",
          "us-east-1",
          "us-east-2",
@@ -9034,7 +9012,10 @@
        ],
        "aws-cn": [],
        "aws-eusc": [],
-        "aws-us-gov": []
+        "aws-us-gov": [
+          "us-gov-east-1",
+          "us-gov-west-1"
+        ]
      }
    },
    "outposts": {
@@ -9075,21 +9056,6 @@
        ]
      }
    },
-    "panorama": {
-      "regions": {
-        "aws": [
-          "ap-southeast-1",
-          "ap-southeast-2",
-          "ca-central-1",
-          "eu-west-1",
-          "us-east-1",
-          "us-west-2"
-        ],
-        "aws-cn": [],
-        "aws-eusc": [],
-        "aws-us-gov": []
-      }
-    },
    "partnercentral-account": {
      "regions": {
        "aws": [
@@ -9239,6 +9205,7 @@
        "aws": [
          "af-south-1",
          "ap-northeast-1",
+          "ap-northeast-2",
          "ap-northeast-3",
          "ap-south-1",
          "ap-southeast-1",
@@ -10001,7 +9968,9 @@
          "cn-north-1",
          "cn-northwest-1"
        ],
-        "aws-eusc": [],
+        "aws-eusc": [
+          "eusc-de-east-1"
+        ],
        "aws-us-gov": [
          "us-gov-east-1",
          "us-gov-west-1"
@@ -10123,7 +10092,10 @@
          "us-west-1",
          "us-west-2"
        ],
-        "aws-cn": [],
+        "aws-cn": [
+          "cn-north-1",
+          "cn-northwest-1"
+        ],
        "aws-eusc": [],
        "aws-us-gov": [
          "us-gov-east-1",
@@ -10814,7 +10786,9 @@
          "cn-north-1",
          "cn-northwest-1"
        ],
-        "aws-eusc": [],
+        "aws-eusc": [
+          "eusc-de-east-1"
+        ],
        "aws-us-gov": [
          "us-gov-east-1",
          "us-gov-west-1"
@@ -12580,7 +12554,9 @@
          "us-east-1"
        ],
        "aws-cn": [],
-        "aws-eusc": [],
+        "aws-eusc": [
+          "eusc-de-east-1"
+        ],
        "aws-us-gov": []
      }
    },
@@ -13156,6 +13132,7 @@
          "ap-southeast-2",
          "ap-southeast-3",
          "ap-southeast-4",
+          "ap-southeast-5",
          "ca-central-1",
          "ca-west-1",
          "eu-central-1",
@@ -0,0 +1,36 @@
+{
+  "Provider": "huaweicloud",
+  "CheckID": "vpc_security_group_open_egress",
+  "CheckTitle": "VPC security groups do not allow open egress to the internet",
+  "CheckType": [],
+  "ServiceName": "vpc",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "",
+  "Severity": "medium",
+  "ResourceType": "HUAWEICLOUD::VPC::SecurityGroup",
+  "ResourceGroup": "network",
+  "Description": "Security groups should not allow unrestricted egress to all destinations. Huawei Cloud represents unrestricted destinations as 0.0.0.0/0, ::/0, or an empty remote IP prefix without a remote security group or address group. Open egress allows instances to reach any destination on the internet, increasing the risk of data exfiltration and command-and-control communication.",
+  "Risk": "Unrestricted egress rules allow instances to communicate with any external destination. This increases the risk of data exfiltration, unauthorized outbound connections, and command-and-control channel establishment by compromised instances.",
+  "RelatedUrl": "",
+  "AdditionalURLs": [
+    "https://support.huaweicloud.com/intl/en-us/usermanual-vpc/vpc_SecurityGroup_0001.html"
+  ],
+  "Remediation": {
+    "Code": {
+      "CLI": "hcloud VPC UpdateSecurityGroupRule --security_group_rule_id <rule_id> --remote_ip_prefix <restricted_cidr>",
+      "NativeIaC": "",
+      "Other": "1. Log on to the Huawei Cloud console.\n2. Choose VPC.\n3. Click Security Groups.\n4. Select the security group.\n5. Edit each allow egress rule whose destination is 0.0.0.0/0, ::/0, or empty and unscoped.\n6. Restrict the destination to required IP ranges or an appropriate security group or address group.\n7. Click OK.",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Restrict allow egress rules to required destination IP ranges, security groups, or address groups instead of allowing all IPv4 or IPv6 destinations.",
+      "Url": "https://hub.prowler.com/check/vpc_security_group_open_egress"
+    }
+  },
+  "Categories": [
+    "trust-boundaries"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}
@@ -0,0 +1,56 @@
+from prowler.lib.check.models import Check, CheckReportHuaweiCloud
+from prowler.providers.huaweicloud.services.vpc.vpc_client import vpc_client
+
+
+class vpc_security_group_open_egress(Check):
+    """Check if VPC security groups allow unrestricted egress to the internet."""
+
+    def execute(self) -> list[CheckReportHuaweiCloud]:
+        """Execute the unrestricted egress check.
+
+        Returns:
+            list[CheckReportHuaweiCloud]: Reports for the evaluated security groups.
+        """
+        findings = []
+
+        for sg in vpc_client.security_groups.values():
+            report = CheckReportHuaweiCloud(metadata=self.metadata(), resource=sg)
+            report.region = sg.region
+            report.resource_id = sg.id
+            report.resource_arn = (
+                f"huaweicloud:vpc:{sg.region}:"
+                f"{vpc_client.audited_account}:security-group/{sg.id}"
+            )
+
+            open_egress_destination = None
+            for rule in sg.rules:
+                if (
+                    rule.direction != "egress"
+                    or rule.action != "allow"
+                    or rule.remote_group_id
+                    or rule.remote_address_group_id
+                ):
+                    continue
+                if rule.remote_ip_prefix in ("0.0.0.0/0", "::/0"):
+                    open_egress_destination = rule.remote_ip_prefix
+                    break
+                if not rule.remote_ip_prefix:
+                    open_egress_destination = "all destinations"
+                    break
+
+            if open_egress_destination:
+                report.status = "FAIL"
+                report.status_extended = (
+                    f"Security group {sg.name} ({sg.id}) allows open egress "
+                    f"({open_egress_destination}) to the internet."
+                )
+            else:
+                report.status = "PASS"
+                report.status_extended = (
+                    f"Security group {sg.name} ({sg.id}) does not allow open egress "
+                    f"to the internet."
+                )
+
+            findings.append(report)
+
+        return findings
@@ -91,6 +91,8 @@ class VPC(HuaweiCloudService):
                                        id=getattr(rule_data, "id", None) or "",
                                        direction=getattr(rule_data, "direction", None)
                                        or "",
+                                        action=getattr(rule_data, "action", None)
+                                        or "allow",
                                        protocol=getattr(rule_data, "protocol", None)
                                        or "",
                                        ethertype=getattr(rule_data, "ethertype", None)
@@ -109,6 +111,12 @@ class VPC(HuaweiCloudService):
                                            rule_data, "remote_group_id", None
                                        )
                                        or "",
+                                        remote_address_group_id=getattr(
+                                            rule_data,
+                                            "remote_address_group_id",
+                                            None,
+                                        )
+                                        or "",
                                        description=getattr(
                                            rule_data, "description", None
                                        )
@@ -148,12 +156,14 @@ class SecurityGroupRule(HuaweiCloudBaseModel):

    id: str
    direction: str
+    action: str = "allow"
    protocol: str
    ethertype: str
    port_range_min: Optional[int] = None
    port_range_max: Optional[int] = None
    remote_ip_prefix: str = ""
    remote_group_id: str = ""
+    remote_address_group_id: str = ""
    description: str = ""


@@ -181,13 +191,16 @@ def rule_source_is_open(rule: SecurityGroupRule) -> bool:

    Huawei Cloud represents "any source" in two ways: an explicit ``0.0.0.0/0``
    (or ``::/0``) in ``remote_ip_prefix``, or leaving both ``remote_ip_prefix``
-    and ``remote_group_id`` empty. Rules that reference another security group
-    via ``remote_group_id`` are NOT open even when ``remote_ip_prefix`` is
-    empty.
+    and both group identifiers empty. Rules that reference a security group or
+    address group are NOT open even when ``remote_ip_prefix`` is empty.
    """
    if rule.remote_ip_prefix in ("0.0.0.0/0", "::/0"):
        return True
-    return not rule.remote_ip_prefix and not rule.remote_group_id
+    return (
+        not rule.remote_ip_prefix
+        and not rule.remote_group_id
+        and not rule.remote_address_group_id
+    )


 def rule_covers_all_ports(rule: SecurityGroupRule) -> bool:
@@ -0,0 +1,4 @@
+from prowler.providers.common.provider import Provider
+from prowler.providers.stackit.services.ske.ske_service import SKEService
+
+ske_client = SKEService(Provider.get_global_provider())
@@ -0,0 +1,38 @@
+{
+  "Provider": "stackit",
+  "CheckID": "ske_cluster_no_public_endpoint",
+  "CheckTitle": "SKE clusters do not expose their Kubernetes API endpoint to the internet",
+  "CheckType": [],
+  "ServiceName": "ske",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "",
+  "Severity": "high",
+  "ResourceType": "NotDefined",
+  "ResourceGroup": "container",
+  "Description": "The **Kubernetes API server** of an SKE cluster should not be reachable from the entire internet. Enable the **ACL extension** to restrict access to a known set of source CIDRs, or run the control plane inside a **STACKIT Network Area (SNA)** so that it is not published on the internet at all.",
+  "Risk": "A publicly reachable **Kubernetes API server** lets anyone on the internet attempt authentication, exploit unpatched control plane vulnerabilities, or brute-force credentials. Combined with leaked or weak credentials, this can lead to **full cluster takeover**, workload tampering, and data exfiltration.",
+  "RelatedUrl": "",
+  "AdditionalURLs": [
+    "https://docs.stackit.cloud/products/runtime/kubernetes-engine/how-tos/enhance-the-security-of-your-cluster/",
+    "https://docs.stackit.cloud/products/runtime/kubernetes-engine/how-tos/enable-private-clusters/",
+    "https://docs.stackit.cloud/products/runtime/kubernetes-engine/"
+  ],
+  "Remediation": {
+    "Code": {
+      "CLI": "",
+      "NativeIaC": "",
+      "Other": "1. In the STACKIT Portal open Kubernetes Engine (SKE) and select the affected cluster. 2. Open the cluster configuration and enable the ACL extension. 3. Add only the source CIDRs that must reach the Kubernetes API (for example your office or VPN egress ranges) and make sure 0.0.0.0/0 and ::/0 are not present. 4. Save the cluster configuration and re-run Prowler to confirm the finding is resolved.",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "**Restrict access to the Kubernetes API.** Enable the **SKE ACL extension** and allowlist only the source CIDRs that legitimately need to reach the API server, never `0.0.0.0/0` or `::/0`. Where available, place the control plane in a **STACKIT Network Area** so the endpoint is not exposed to the internet.",
+      "Url": "https://hub.prowler.com/check/ske_cluster_no_public_endpoint"
+    }
+  },
+  "Categories": [
+    "internet-exposed"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": "A cluster passes when its control plane access scope is SNA, or when the ACL extension is enabled and its allowed CIDR list contains no unrestricted range. A cluster with the ACL extension disabled is reported as FAIL because the Kubernetes API accepts connections from any source address. The SNA private control plane is not generally available yet, so most clusters are expected to rely on the ACL extension."
+}
@@ -0,0 +1,61 @@
+from prowler.lib.check.models import Check, CheckReportStackIT
+from prowler.providers.stackit.services.ske.ske_client import ske_client
+
+
+class ske_cluster_no_public_endpoint(Check):
+    """
+    Check if SKE clusters expose their Kubernetes API endpoint to the internet.
+
+    A cluster passes when its control plane is confined to a STACKIT Network
+    Area, or when the ACL extension restricts the Kubernetes API to a set of
+    source CIDRs that does not include an unrestricted range.
+    """
+
+    def execute(self) -> list[CheckReportStackIT]:
+        """
+        Execute the check for all SKE clusters in the StackIT project.
+
+        Returns:
+            list: A list of CheckReportStackIT findings
+        """
+        findings = []
+
+        for cluster in ske_client.clusters:
+            report = CheckReportStackIT(
+                metadata=self.metadata(),
+                resource=cluster,
+            )
+
+            if cluster.has_public_endpoint():
+                report.status = "FAIL"
+                unrestricted_cidrs = cluster.unrestricted_cidrs()
+                if unrestricted_cidrs:
+                    report.status_extended = (
+                        f"SKE cluster {cluster.name} exposes its Kubernetes API "
+                        f"endpoint to the internet because its ACL allows "
+                        f"unrestricted access from {', '.join(unrestricted_cidrs)}."
+                    )
+                else:
+                    report.status_extended = (
+                        f"SKE cluster {cluster.name} exposes its Kubernetes API "
+                        f"endpoint to the internet because the ACL extension is "
+                        f"not enabled."
+                    )
+            else:
+                report.status = "PASS"
+                if cluster.has_private_control_plane():
+                    report.status_extended = (
+                        f"SKE cluster {cluster.name} has a private control plane "
+                        f"and its Kubernetes API endpoint is not reachable from "
+                        f"the internet."
+                    )
+                else:
+                    report.status_extended = (
+                        f"SKE cluster {cluster.name} restricts access to its "
+                        f"Kubernetes API endpoint to "
+                        f"{len(cluster.allowed_cidrs)} allowed CIDR(s)."
+                    )
+
+            findings.append(report)
+
+        return findings
@@ -0,0 +1,305 @@
+import json
+from typing import Optional
+
+from pydantic.v1 import BaseModel
+
+from prowler.lib.logger import logger
+from prowler.providers.stackit.stackit_provider import StackitProvider, suppress_stderr
+
+# Source ranges that place no restriction on who may reach the Kubernetes API.
+UNRESTRICTED_CIDRS = frozenset({"0.0.0.0/0", "::/0"})
+
+# Control plane access scope that keeps the Kubernetes API inside a STACKIT
+# Network Area instead of publishing it on the internet.
+PRIVATE_ACCESS_SCOPE = "SNA"
+
+
+class SKEService:
+    """
+    StackIT Kubernetes Engine (SKE) service class to handle cluster operations.
+
+    This service uses the StackIT Python SDK to access SKE resources.
+    Authentication is delegated to the SDK, which signs the RSA challenge
+    in the configured service account key and refreshes access tokens
+    internally for the life of the scan.
+    """
+
+    def __init__(self, provider: StackitProvider):
+        """
+        Initialize the SKE service.
+
+        Args:
+            provider: The StackIT provider instance
+        """
+        self.provider = provider
+        self.project_id = provider.identity.project_id
+        self.service_account_key_path = provider.session.get("service_account_key_path")
+
+        # Generate regional clients (AWS pattern)
+        self.regional_clients = provider.generate_regional_clients("ske")
+        self.audited_regions = provider.identity.audited_regions
+
+        # Initialize cluster list
+        self.clusters: list[Cluster] = []
+
+        # Fetch resources from all regions
+        self._fetch_all_regions()
+
+    def _fetch_all_regions(self):
+        """Fetch SKE clusters from all audited regions.
+
+        A project is not necessarily provisioned in every StackIT region. A
+        region where the project does not exist answers the SKE endpoints with
+        HTTP 404 (``resource not found: project``). That is expected, so the
+        region is skipped and the scan continues with the remaining regions
+        instead of aborting.
+
+        A project that enabled SKE in some regions but not others answers the
+        remaining ones with HTTP 403 ``Service not enabled``; that region is
+        skipped too. Genuine credential and permission failures still propagate
+        via ``handle_api_error`` so a misconfigured account fails loudly.
+        """
+        for region, client in self.regional_clients.items():
+            try:
+                self._list_clusters(client, region)
+            except Exception as error:
+                if getattr(error, "status", None) == 404:
+                    logger.info(
+                        f"StackIT project {self.project_id} has no SKE presence "
+                        f"in region {region} (404 resource not found); skipping "
+                        f"this region."
+                    )
+                    continue
+                if self._is_service_not_enabled(error):
+                    logger.info(
+                        f"StackIT project {self.project_id} has SKE disabled in "
+                        f"region {region} (403 service not enabled); skipping "
+                        f"this region."
+                    )
+                    continue
+                raise
+
+    @staticmethod
+    def _extract_items(response, endpoint_name: str) -> list:
+        """Extract the items list from a StackIT SDK response.
+
+        Handles three response shapes safely:
+            - SDK model exposing an ``items`` attribute (not the ``dict.items`` method)
+            - Raw ``dict`` with an ``"items"`` key
+            - Plain ``list``
+
+        ``isinstance(response, dict)`` is checked first because ``dict`` has an
+        ``items`` *method*; ``hasattr(response, "items")`` is otherwise True for
+        plain dicts and silently returns the bound method.
+        """
+        if isinstance(response, dict):
+            return response.get("items", [])
+        if isinstance(response, list):
+            return response
+        items_attr = getattr(response, "items", None)
+        if items_attr is not None and not callable(items_attr):
+            return items_attr
+        logger.warning(
+            f"Unexpected response type from {endpoint_name}: {type(response)}"
+        )
+        return []
+
+    @staticmethod
+    def _get_field(item, *keys, default=None):
+        """Read a field from an SDK model (attribute) or a raw ``dict`` (key).
+
+        ``_extract_items`` yields either SDK models or raw dicts, so the nested
+        cluster parsing must read fields from both shapes. Multiple key aliases
+        are accepted so snake_case SDK attributes and camelCase API/dict keys
+        are both supported (e.g. ``allowed_cidrs`` / ``allowedCidrs``). ``None``
+        items return ``default`` so nested lookups can be chained safely.
+        Returns the first non-None match, otherwise ``default``.
+        """
+        if item is None:
+            return default
+        if isinstance(item, dict):
+            for key in keys:
+                value = item.get(key)
+                if value is not None:
+                    return value
+            return default
+        for key in keys:
+            value = getattr(item, key, None)
+            if value is not None:
+                return value
+        return default
+
+    @staticmethod
+    def _is_service_not_enabled(error) -> bool:
+        """Return True for the 403 SKE returns in a region that never enabled it.
+
+        Status alone cannot separate this from a missing IAM role -- both are
+        403 -- so the body message is the only discriminator.
+        """
+        if getattr(error, "status", None) != 403:
+            return False
+        body = getattr(error, "body", None)
+        if body is None:
+            return False
+        if isinstance(body, bytes):
+            body = body.decode("utf-8", errors="replace")
+        message = ""
+        if isinstance(body, str):
+            try:
+                message = json.loads(body).get("message", "")
+            except (ValueError, AttributeError):
+                message = body
+        elif isinstance(body, dict):
+            message = body.get("message", "")
+        return "service not enabled" in str(message).lower()
+
+    def _handle_api_call(self, api_function, *args, **kwargs):
+        """
+        Centralized API call handler with authentication error detection.
+
+        Args:
+            api_function: The API function to call
+            *args: Positional arguments to pass to the API function
+            **kwargs: Keyword arguments to pass to the API function
+
+        Returns:
+            The API response
+
+        Raises:
+            StackITInvalidTokenError: If authentication fails (401)
+        """
+        try:
+            # Suppress StackIT SDK stderr messages during API calls
+            with suppress_stderr():
+                return api_function(*args, **kwargs)
+        except Exception as e:
+            # A region that never enabled SKE also answers 403; handing it to
+            # handle_api_error would abort the scan as a credentials failure
+            # instead of letting _fetch_all_regions skip the region.
+            if self._is_service_not_enabled(e):
+                raise
+            # Use centralized error handler from provider
+            self.provider.handle_api_error(e)
+            raise
+
+    @classmethod
+    def _parse_access_scope(cls, cluster_data) -> Optional[str]:
+        """Return the control plane access scope of a cluster, if declared.
+
+        The scope lives at ``network.controlPlane.accessScope``. Every level of
+        that chain is optional because the private (``SNA``) control plane is an
+        opt-in preview feature, so most clusters omit it entirely. ``None`` is
+        returned in that case and the caller falls back to the ACL extension.
+        """
+        network = cls._get_field(cluster_data, "network")
+        control_plane = cls._get_field(network, "control_plane", "controlPlane")
+        access_scope = cls._get_field(control_plane, "access_scope", "accessScope")
+        if access_scope is None:
+            return None
+        # ``AccessScope`` is a ``str`` Enum, whose ``str()`` renders as
+        # "AccessScope.SNA" rather than the "SNA" wire value.
+        return str(getattr(access_scope, "value", access_scope))
+
+    @classmethod
+    def _parse_acl(cls, cluster_data) -> tuple[bool, list[str]]:
+        """Return the ``(enabled, allowed_cidrs)`` pair of the cluster ACL extension.
+
+        The ACL extension restricts which source CIDRs may reach the Kubernetes
+        API and lives at ``extensions.acl``. When the extension is absent the
+        API server accepts connections from any address, which is reported here
+        as ``(False, [])``.
+        """
+        extensions = cls._get_field(cluster_data, "extensions")
+        acl = cls._get_field(extensions, "acl")
+        enabled = bool(cls._get_field(acl, "enabled", default=False))
+        allowed_cidrs = (
+            cls._get_field(acl, "allowed_cidrs", "allowedCidrs", default=[]) or []
+        )
+        return enabled, [str(cidr) for cidr in allowed_cidrs]
+
+    def _list_clusters(self, client, region: str):
+        """
+        List all SKE clusters in the StackIT project for a single region.
+
+        Populates ``self.clusters`` with :class:`Cluster` objects describing the
+        control plane access scope and the ACL extension of each cluster.
+        """
+        if not client:
+            logger.warning(
+                f"Cannot list SKE clusters in {region}: StackIT SKE client not available"
+            )
+            return
+
+        response = self._handle_api_call(
+            client.list_clusters, project_id=self.project_id, region=region
+        )
+
+        clusters_list = self._extract_items(response, "list_clusters")
+
+        for cluster_data in clusters_list:
+            try:
+                cluster_name = str(self._get_field(cluster_data, "name") or "")
+                acl_enabled, allowed_cidrs = self._parse_acl(cluster_data)
+                cluster = Cluster(
+                    # SKE addresses a cluster by its name; the API exposes no
+                    # separate identifier, so the name doubles as the id.
+                    id=cluster_name,
+                    name=cluster_name,
+                    project_id=self.project_id,
+                    region=region,
+                    access_scope=self._parse_access_scope(cluster_data),
+                    acl_enabled=acl_enabled,
+                    allowed_cidrs=allowed_cidrs,
+                )
+                self.clusters.append(cluster)
+            except Exception as e:
+                logger.error(f"Error processing SKE cluster: {e}")
+                continue
+
+        logger.info(
+            f"Successfully listed {len(clusters_list)} SKE clusters in {region}"
+        )
+
+
+class Cluster(BaseModel):
+    """
+    Represents a StackIT SKE Cluster.
+
+    Attributes:
+        id: The unique identifier of the cluster (SKE uses the cluster name)
+        name: The name of the cluster
+        project_id: The StackIT project ID containing the cluster
+        region: The region where the cluster runs
+        access_scope: Control plane access scope ("PUBLIC"/"SNA"), None when unset
+        acl_enabled: Whether the ACL extension restricting API access is enabled
+        allowed_cidrs: Source CIDRs allowed to reach the Kubernetes API
+    """
+
+    id: str
+    name: str
+    project_id: str
+    region: str
+    access_scope: Optional[str] = None
+    acl_enabled: bool = False
+    allowed_cidrs: list[str] = []
+
+    def has_private_control_plane(self) -> bool:
+        """Check whether the control plane is confined to a STACKIT Network Area."""
+        return self.access_scope == PRIVATE_ACCESS_SCOPE
+
+    def unrestricted_cidrs(self) -> list[str]:
+        """Return the allowed CIDRs that permit access from any source address."""
+        return [cidr for cidr in self.allowed_cidrs if cidr in UNRESTRICTED_CIDRS]
+
+    def has_public_endpoint(self) -> bool:
+        """Check whether the Kubernetes API endpoint is reachable from the internet.
+
+        A cluster is publicly reachable when its control plane is not confined to
+        a STACKIT Network Area and either the ACL extension is disabled (no source
+        restriction at all) or the allowlist itself contains an unrestricted range.
+        """
+        if self.has_private_control_plane():
+            return False
+        if not self.acl_enabled:
+            return True
+        return bool(self.unrestricted_cidrs())
@@ -17,6 +17,7 @@ from stackit.core.configuration import Configuration
 from stackit.iaas import DefaultApi as IaasDefaultApi
 from stackit.objectstorage import DefaultApi as ObjectStorageDefaultApi
 from stackit.resourcemanager import DefaultApi as ResourceManagerDefaultApi
+from stackit.ske import DefaultApi as SkeDefaultApi

 from prowler.config.config import (
    default_config_file_path,
@@ -228,6 +229,7 @@ class StackitProvider(Provider):
    _SERVICE_API_CLASS = {
        "iaas": IaasDefaultApi,
        "objectstorage": ObjectStorageDefaultApi,
+        "ske": SkeDefaultApi,
    }

    def generate_regional_clients(self, service: str = "iaas") -> dict:
@@ -557,10 +559,11 @@ class StackitProvider(Provider):
        """
        Test connection to StackIT by validating credentials.

-        This method validates the service account credentials and project ID
-        by making a Resource Manager ``get_project`` call. Pass either the
-        key file path or the inline key content; the SDK signs the RSA
-        challenge and mints a short-lived access token internally.
+        This method probes Resource Manager with a ``get_project`` call. A 403
+        response is non-fatal because service-specific permissions are checked
+        during discovery. Pass either the key file path or the inline key
+        content; the SDK signs the RSA challenge and mints a short-lived access
+        token internally.

        Args:
            project_id (str): StackIT project ID
@@ -600,6 +603,13 @@ class StackitProvider(Provider):
                raise error
            return Connection(error=error)
        except Exception as test_error:
+            if getattr(test_error, "status", None) == 403:
+                logger.warning(
+                    "StackIT test_connection: Resource Manager access could not be "
+                    "verified (403). Service permissions will be checked during "
+                    "discovery."
+                )
+                return Connection(is_connected=True)
            try:
                StackitProvider.handle_api_error(test_error)
            except StackITInvalidTokenError as auth_error:
@@ -11,6 +11,12 @@
        "eu01",
        "eu02"
      ]
+    },
+    "ske": {
+      "regions": [
+        "eu01",
+        "eu02"
+      ]
    }
  }
 }
@@ -102,6 +102,7 @@ dependencies = [
  "stackit-iaas==1.4.0",
  "stackit-objectstorage==1.4.0",
  "stackit-resourcemanager==0.8.0",
+  "stackit-ske==1.12.0",
  "tabulate==0.9.0",
  "tzlocal==5.3.1",
  "uuid6==2024.7.10",
@@ -6,6 +6,7 @@ from prowler.providers.huaweicloud.services.vpc.vpc_service import (
    SecurityGroupRule,
    SecurityGroups,
    VPCs,
+    rule_source_is_open,
 )
 from tests.providers.huaweicloud.huaweicloud_fixtures import (
    set_mocked_huaweicloud_provider,
@@ -24,6 +25,18 @@ def _provider_with_client(regional_client):


 class TestVPCService:
+    def test_rule_with_remote_address_group_is_not_open(self):
+        rule = SecurityGroupRule(
+            id="rule-1",
+            direction="ingress",
+            protocol="tcp",
+            ethertype="IPv4",
+            remote_ip_prefix="",
+            remote_address_group_id="address-group-1",
+        )
+
+        assert rule_source_is_open(rule) is False
+
    def test_list_vpcs_and_security_groups_parses(self):
        vpc = SimpleNamespace(
            id="vpc-1",
@@ -36,12 +49,14 @@ class TestVPCService:
        rule = SimpleNamespace(
            id="rule-1",
            direction="ingress",
+            action="deny",
            protocol="tcp",
            ethertype="IPv4",
            port_range_min=22,
            port_range_max=22,
            remote_ip_prefix="0.0.0.0/0",
            remote_group_id="",
+            remote_address_group_id="address-group-1",
            description="ssh open",
        )
        sg = SimpleNamespace(
@@ -82,8 +97,10 @@ class TestVPCService:
        parsed_rule = parsed_sg.rules[0]
        assert isinstance(parsed_rule, SecurityGroupRule)
        assert parsed_rule.direction == "ingress"
+        assert parsed_rule.action == "deny"
        assert parsed_rule.protocol == "tcp"
        assert parsed_rule.remote_ip_prefix == "0.0.0.0/0"
+        assert parsed_rule.remote_address_group_id == "address-group-1"
        assert parsed_rule.port_range_min == 22
        assert parsed_rule.port_range_max == 22

@@ -94,12 +111,14 @@ class TestVPCService:
        rule = SimpleNamespace(
            id="rule-1",
            direction=None,
+            action=None,
            protocol=None,
            ethertype=None,
            port_range_min=None,
            port_range_max=None,
            remote_ip_prefix=None,
            remote_group_id=None,
+            remote_address_group_id=None,
            description=None,
        )
        sg = SimpleNamespace(
@@ -134,8 +153,10 @@ class TestVPCService:
        assert parsed_sg.name == "sg-1"  # falls back to id
        assert parsed_sg.vpc_id == ""
        parsed_rule = parsed_sg.rules[0]
+        assert parsed_rule.action == "allow"
        assert parsed_rule.protocol == ""
        assert parsed_rule.remote_ip_prefix == ""
+        assert parsed_rule.remote_address_group_id == ""
        assert parsed_rule.description == ""
        assert parsed_rule.direction == ""

@@ -0,0 +1,277 @@
+from unittest import mock
+
+from tests.providers.huaweicloud.huaweicloud_fixtures import (
+    set_mocked_huaweicloud_provider,
+)
+
+
+class TestVpcSecurityGroupOpenEgress:
+    def test_no_open_egress_passes(self):
+        vpc_client = mock.MagicMock()
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_huaweicloud_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.huaweicloud.services.vpc.vpc_security_group_open_egress.vpc_security_group_open_egress.vpc_client",
+                new=vpc_client,
+            ),
+        ):
+            from prowler.providers.huaweicloud.services.vpc.vpc_security_group_open_egress.vpc_security_group_open_egress import (
+                vpc_security_group_open_egress,
+            )
+            from prowler.providers.huaweicloud.services.vpc.vpc_service import (
+                SecurityGroupRule,
+                SecurityGroups,
+            )
+
+            sg = SecurityGroups(
+                id="sg-1",
+                name="safe-sg",
+                region="la-south-2",
+                vpc_id="vpc-1",
+                rules=[
+                    SecurityGroupRule(
+                        id="rule-1",
+                        direction="egress",
+                        protocol="tcp",
+                        ethertype="IPv4",
+                        remote_ip_prefix="10.0.0.0/24",
+                        port_range_min=443,
+                        port_range_max=443,
+                    ),
+                    SecurityGroupRule(
+                        id="deny-rule",
+                        direction="egress",
+                        action="deny",
+                        protocol="",
+                        ethertype="IPv4",
+                        remote_ip_prefix="",
+                    ),
+                    SecurityGroupRule(
+                        id="address-group-rule",
+                        direction="egress",
+                        action="allow",
+                        protocol="",
+                        ethertype="IPv4",
+                        remote_ip_prefix="",
+                        remote_address_group_id="address-group-1",
+                    ),
+                ],
+            )
+            vpc_client.security_groups = {"sg-1": sg}
+            vpc_client.audited_account = "123456789012"
+
+            check = vpc_security_group_open_egress()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert result[0].status_extended == (
+                "Security group safe-sg (sg-1) does not allow open egress to "
+                "the internet."
+            )
+            assert result[0].resource_id == "sg-1"
+            assert result[0].resource_name == "safe-sg"
+            assert result[0].resource_arn == (
+                "huaweicloud:vpc:la-south-2:123456789012:security-group/sg-1"
+            )
+            assert result[0].region == "la-south-2"
+
+    def test_open_egress_ipv4_fails(self):
+        vpc_client = mock.MagicMock()
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_huaweicloud_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.huaweicloud.services.vpc.vpc_security_group_open_egress.vpc_security_group_open_egress.vpc_client",
+                new=vpc_client,
+            ),
+        ):
+            from prowler.providers.huaweicloud.services.vpc.vpc_security_group_open_egress.vpc_security_group_open_egress import (
+                vpc_security_group_open_egress,
+            )
+            from prowler.providers.huaweicloud.services.vpc.vpc_service import (
+                SecurityGroupRule,
+                SecurityGroups,
+            )
+
+            sg = SecurityGroups(
+                id="sg-1",
+                name="open-egress-sg",
+                region="la-south-2",
+                vpc_id="vpc-1",
+                rules=[
+                    SecurityGroupRule(
+                        id="rule-1",
+                        direction="egress",
+                        protocol="tcp",
+                        ethertype="IPv4",
+                        remote_ip_prefix="0.0.0.0/0",
+                        port_range_min=80,
+                        port_range_max=80,
+                    ),
+                ],
+            )
+            vpc_client.security_groups = {"sg-1": sg}
+            vpc_client.audited_account = "123456789012"
+
+            check = vpc_security_group_open_egress()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert result[0].status_extended == (
+                "Security group open-egress-sg (sg-1) allows open egress "
+                "(0.0.0.0/0) to the internet."
+            )
+            assert result[0].resource_id == "sg-1"
+            assert result[0].resource_name == "open-egress-sg"
+            assert result[0].resource_arn == (
+                "huaweicloud:vpc:la-south-2:123456789012:security-group/sg-1"
+            )
+            assert result[0].region == "la-south-2"
+
+    def test_open_egress_ipv6_fails(self):
+        vpc_client = mock.MagicMock()
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_huaweicloud_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.huaweicloud.services.vpc.vpc_security_group_open_egress.vpc_security_group_open_egress.vpc_client",
+                new=vpc_client,
+            ),
+        ):
+            from prowler.providers.huaweicloud.services.vpc.vpc_security_group_open_egress.vpc_security_group_open_egress import (
+                vpc_security_group_open_egress,
+            )
+            from prowler.providers.huaweicloud.services.vpc.vpc_service import (
+                SecurityGroupRule,
+                SecurityGroups,
+            )
+
+            sg = SecurityGroups(
+                id="sg-1",
+                name="open-egress-sg-ipv6",
+                region="la-south-2",
+                vpc_id="vpc-1",
+                rules=[
+                    SecurityGroupRule(
+                        id="rule-1",
+                        direction="egress",
+                        protocol="tcp",
+                        ethertype="IPv6",
+                        remote_ip_prefix="::/0",
+                        port_range_min=443,
+                        port_range_max=443,
+                    ),
+                ],
+            )
+            vpc_client.security_groups = {"sg-1": sg}
+            vpc_client.audited_account = "123456789012"
+
+            check = vpc_security_group_open_egress()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert result[0].status_extended == (
+                "Security group open-egress-sg-ipv6 (sg-1) allows open egress "
+                "(::/0) to the internet."
+            )
+            assert result[0].resource_id == "sg-1"
+            assert result[0].resource_name == "open-egress-sg-ipv6"
+            assert result[0].resource_arn == (
+                "huaweicloud:vpc:la-south-2:123456789012:security-group/sg-1"
+            )
+            assert result[0].region == "la-south-2"
+
+    def test_open_egress_with_empty_destination_fails(self):
+        vpc_client = mock.MagicMock()
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_huaweicloud_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.huaweicloud.services.vpc.vpc_security_group_open_egress.vpc_security_group_open_egress.vpc_client",
+                new=vpc_client,
+            ),
+        ):
+            from prowler.providers.huaweicloud.services.vpc.vpc_security_group_open_egress.vpc_security_group_open_egress import (
+                vpc_security_group_open_egress,
+            )
+            from prowler.providers.huaweicloud.services.vpc.vpc_service import (
+                SecurityGroupRule,
+                SecurityGroups,
+            )
+
+            sg = SecurityGroups(
+                id="sg-1",
+                name="open-egress-sg-empty-destination",
+                region="la-south-2",
+                vpc_id="vpc-1",
+                rules=[
+                    SecurityGroupRule(
+                        id="rule-1",
+                        direction="egress",
+                        action="allow",
+                        protocol="",
+                        ethertype="IPv4",
+                        remote_ip_prefix="",
+                        remote_address_group_id="",
+                    ),
+                ],
+            )
+            vpc_client.security_groups = {"sg-1": sg}
+            vpc_client.audited_account = "123456789012"
+
+            check = vpc_security_group_open_egress()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert result[0].status_extended == (
+                "Security group open-egress-sg-empty-destination (sg-1) allows "
+                "open egress (all destinations) to the internet."
+            )
+            assert result[0].resource_id == "sg-1"
+            assert result[0].resource_name == "open-egress-sg-empty-destination"
+            assert result[0].resource_arn == (
+                "huaweicloud:vpc:la-south-2:123456789012:security-group/sg-1"
+            )
+            assert result[0].region == "la-south-2"
+
+    def test_no_security_groups(self):
+        vpc_client = mock.MagicMock()
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_huaweicloud_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.huaweicloud.services.vpc.vpc_security_group_open_egress.vpc_security_group_open_egress.vpc_client",
+                new=vpc_client,
+            ),
+        ):
+            from prowler.providers.huaweicloud.services.vpc.vpc_security_group_open_egress.vpc_security_group_open_egress import (
+                vpc_security_group_open_egress,
+            )
+
+            vpc_client.security_groups = {}
+            vpc_client.audited_account = "123456789012"
+
+            check = vpc_security_group_open_egress()
+            result = check.execute()
+
+            assert len(result) == 0
@@ -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",
+        ]
@@ -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
+        )
@@ -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
    ):
@@ -3845,6 +3845,7 @@ dependencies = [
    { name = "stackit-iaas" },
    { name = "stackit-objectstorage" },
    { name = "stackit-resourcemanager" },
+    { name = "stackit-ske" },
    { name = "tabulate" },
    { name = "tzlocal" },
    { name = "uuid6" },
@@ -3966,6 +3967,7 @@ requires-dist = [
    { name = "stackit-iaas", specifier = "==1.4.0" },
    { name = "stackit-objectstorage", specifier = "==1.4.0" },
    { name = "stackit-resourcemanager", specifier = "==0.8.0" },
+    { name = "stackit-ske", specifier = "==1.12.0" },
    { name = "tabulate", specifier = "==0.9.0" },
    { name = "tzlocal", specifier = "==5.3.1" },
    { name = "uuid6", specifier = "==2024.7.10" },
@@ -5121,6 +5123,21 @@ wheels = [
    { url = "https://files.pythonhosted.org/packages/c7/9c/38a74d0f7a89b4320f6d2366fb660638bda8860daa08748b12c713d84381/stackit_resourcemanager-0.8.0-py3-none-any.whl", hash = "sha256:dd04bb8353d041a137c4dcba190beabded7acfaff1bc98b218fce20a99389ebc", size = 81288, upload-time = "2026-05-13T09:43:07.81Z" },
 ]

+[[package]]
+name = "stackit-ske"
+version = "1.12.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "pydantic" },
+    { name = "python-dateutil" },
+    { name = "requests" },
+    { name = "stackit-core" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/cd/9e/df3ad585cb96d028354f4253568e9879d81bb9395d5ebfa268fa9350e2df/stackit_ske-1.12.0.tar.gz", hash = "sha256:62814279f3b7fb2387648f92d14453a8905ad60115c07579f2741ddb7d1fcc94", size = 37239, upload-time = "2026-06-30T11:18:49.39Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/00/37/dc54fb7185a2d4da37308322ea1a7b992312030b2e37262de4eb4003f5c7/stackit_ske-1.12.0-py3-none-any.whl", hash = "sha256:45bd8084d87f14f818b3d7e824450248c8784ed204ca1b2dc108f491dcbdb1a3", size = 93142, upload-time = "2026-06-30T11:18:48.233Z" },
+]
+
 [[package]]
 name = "std-uritemplate"
 version = "2.0.8"
Author	SHA1	Message	Date
Prowler Botandprowler-bot	db25484ccb	feat(aws): Update regions for AWS services (#12472 ) Co-authored-by: prowler-bot <179230569+prowler-bot@users.noreply.github.com>	2026-08-20 16:02:56 +01:00
Johannes EnglerandHugo P.Brito	3da4209ee7	feat(stackit): add ske_cluster_no_public_endpoint check (#11943 ) Co-authored-by: Hugo P.Brito <hugopbrit@gmail.com>	2026-08-20 13:01:01 +01:00
Tomás Sebastián Tobio tomitobio Hugo P.Brito	acb6ff0425	feat(providers/huaweicloud): add vpc_security_group_open_egress check (#12209 ) Co-authored-by: tomitobio <tomitobio@users.noreply.github.com> Co-authored-by: Hugo P.Brito <hugopbrit@gmail.com>	2026-08-20 10:52:13 +01:00
Pablo Fernandez Guerra (PFE)	ba564af4f4	fix(ci): unblock the Python runtime download in blocked-egress jobs (#12490 )	2026-08-20 10:48:16 +02:00
				`@@ -0,0 +1 @@`
				`ske_cluster_no_public_endpoint` check for STACKIT provider, flagging SKE clusters whose Kubernetes API endpoint is reachable from the whole internet because the ACL extension is disabled or its allowed CIDR list contains `0.0.0.0/0` or `::/0`
				`@@ -0,0 +1 @@`
				`vpc_security_group_open_egress` check for Huawei Cloud provider: VPC security groups do not allow open egress to the internet