Back to Checks
alibabacloud

Kubernetes cluster health check has been triggered at least once per week

cs_kubernetes_cluster_check_weekly
Service: csSeverity: medium
OfficialBuilt-inVerified
Categories:cluster-security
Resource Type:ALIYUN::CS::ManagedKubernetesCluster

Alibaba Cloud Kubernetes Engine provides a cluster health check that validates node health and cluster configuration, including kubelet, docker daemon, kernel, and iptables settings. Weekly checks ensure VPC/VSwitch, SLB, and ECS nodes function correctly. Consecutive failures generate diagnostic reports for corrective action.

Risk

Without weekly health checks, node failures, misconfigured network rules, or degraded components may go undetected for extended periods, increasing the risk of cluster instability, service outages, and exploitable security vulnerabilities. Delayed detection can impact the integrity and availability of workloads on the cluster.

Run this check with Prowler

Run in Prowler Cloud
Run
prowler alibabacloud --checks cs_kubernetes_cluster_check_weekly

Recommendation

Trigger a cluster health check at least once per week to ensure all nodes and system components are healthy. Use the Global Check feature in the ACK Console or the aliyun cs CLI to verify and trigger checks.

Remediation

CLI

aliyun cs GET /clusters/<cluster_id>/checks --header 'Content-Type=application/json'

Other

1Log on to the **ACK Console**.
2Select the target cluster and open the **More** pop-menu for advanced options.
3Select **Global Check** and click the **Start** button to trigger the checking.
4Verify the checking time and details in Global Check.
5Trigger cluster checks at least once per week.

Check Code

from datetime import datetime, timedelta, timezone

from prowler.lib.check.models import Check, CheckReportAlibabaCloud
from prowler.providers.alibabacloud.services.cs.cs_client import cs_client


class cs_kubernetes_cluster_check_weekly(Check):
    """Check if Cluster Check is triggered at least once per week for Kubernetes Clusters."""

    def execute(self) -> list[CheckReportAlibabaCloud]:
        findings = []

        # Get configurable max days from audit config (default: 7 days)
        max_cluster_check_days = cs_client.audit_config.get("max_cluster_check_days", 7)

        # Calculate the threshold date
        threshold_date = datetime.now(timezone.utc) - timedelta(
            days=max_cluster_check_days
        )

        for cluster in cs_client.clusters:
            report = CheckReportAlibabaCloud(metadata=self.metadata(), resource=cluster)
            report.region = cluster.region
            report.resource_id = cluster.id
            report.resource_arn = f"acs:cs:{cluster.region}:{cs_client.audited_account}:cluster/{cluster.id}"

            if cluster.last_check_time:
                # Ensure last_check_time is timezone-aware
                last_check = cluster.last_check_time
                if last_check.tzinfo is None:
                    # If naive datetime, assume UTC
                    last_check = last_check.replace(tzinfo=timezone.utc)

                # Calculate days since last check
                days_since_check = (datetime.now(timezone.utc) - last_check).days

                if last_check >= threshold_date:
                    report.status = "PASS"
                    report.status_extended = (
                        f"Kubernetes cluster {cluster.name} has had a successful cluster check "
                        f"within the last {max_cluster_check_days} days "
                        f"(last check: {cluster.last_check_time.strftime('%Y-%m-%d %H:%M:%S UTC')}, "
                        f"{days_since_check} days ago)."
                    )
                else:
                    report.status = "FAIL"
                    report.status_extended = (
                        f"Kubernetes cluster {cluster.name} has not had a successful cluster check "
                        f"within the last {max_cluster_check_days} days "
                        f"(last check: {cluster.last_check_time.strftime('%Y-%m-%d %H:%M:%S UTC')}, "
                        f"{days_since_check} days ago)."
                    )
            else:
                report.status = "FAIL"
                report.status_extended = (
                    f"Kubernetes cluster {cluster.name} has no successful cluster check history. "
                    f"Cluster checks should be triggered at least once every {max_cluster_check_days} days."
                )

            findings.append(report)

        return findings