Security audit
Security checks across malware telemetry and agentic risk
The skill is mostly a disclosed read-only monitoring tool, but it needs review because it can obtain Kubernetes access material and read TLS Secret objects while handling cloud credentials.
Install only if you are comfortable granting read-only Huawei Cloud and Kubernetes permissions that include cluster access setup, Pod/Service/Ingress reads, and get access to TLS Secrets. Use a dedicated least-privilege IAM user and Kubernetes RBAC role, avoid production-wide credentials, consider setting check_certificates=false unless certificate checks are needed, and verify the kubectl-cce plugin before installing it.
access_key = ak or os.environ.get("HUAWEI_AK") or os.environ.get("HUAWEICLOUD_SDK_AK") or os.environ.get("HW_ACCESS_KEY")
secret_key = sk or os.environ.get("HUAWEI_SK") or os.environ.get("HUAWEICLOUD_SDK_SK") or os.environ.get("HW_SECRET_KEY")
proj_id = project_id or os.environ.get("HUAWEI_PROJECT_ID") or os.environ.get("HUAWEICLOUD_SDK_PROJECT_ID") or os.environ.get("HW_PROJECT_ID")
env = os.environ.copy()
env["CCE_CLUSTER_ID"] = cluster_id
env["CCE_REGION"] = region
env["HW_REGION"] = regionIt also includes CoreDNS, nginx-ingress, and autoscaler summaries. Cloud resources are scoped to the current cluster when an association can be proven: ELB is matched through LoadBalancer Service IP/EIP, NAT Gateway is filtered by the cluster VPC, and EIP is limited to associated ELB/NAT/Service IPs. LoadBalancer Service discovery uses `kubectl` with generated kubeconfig through the cluster EIP when external access is available. If the cluster has no EIP, it uses the `kubectl cce` plugin. If neither path works, aggregation fails. ## Risk Levels
return {"success": True, "cluster": result.get("data") or {}}
def _create_kubeconfig(region: str, cluster_id: str, ak: Optional[str], sk: Optional[str], project_id: Optional[str]) -> Dict[str, Any]:
result = run_hcloud(
"CCE",
"CreateKubernetesClusterCert",)
if not result.get("success"):
return result
kubeconfig = result.get("data") or {}
if not kubeconfig.get("clusters"):
return {"success": False, "error": "CreateKubernetesClusterCert returned no kubeconfig clusters"}
_prefer_external_context(kubeconfig)if not result.get("success"):
return result
kubeconfig = result.get("data") or {}
if not kubeconfig.get("clusters"):
return {"success": False, "error": "CreateKubernetesClusterCert returned no kubeconfig clusters"}
_prefer_external_context(kubeconfig)
return {"success": True, "kubeconfig": kubeconfig}return result
kubeconfig = result.get("data") or {}
if not kubeconfig.get("clusters"):
return {"success": False, "error": "CreateKubernetesClusterCert returned no kubeconfig clusters"}
_prefer_external_context(kubeconfig)
return {"success": True, "kubeconfig": kubeconfig}kubeconfig = result.get("data") or {}
if not kubeconfig.get("clusters"):
return {"success": False, "error": "CreateKubernetesClusterCert returned no kubeconfig clusters"}
_prefer_external_context(kubeconfig)
return {"success": True, "kubeconfig": kubeconfig}if not kubeconfig.get("clusters"):
return {"success": False, "error": "CreateKubernetesClusterCert returned no kubeconfig clusters"}
_prefer_external_context(kubeconfig)
return {"success": True, "kubeconfig": kubeconfig}
def _prefer_external_context(kubeconfig: Dict[str, Any]) -> None:if not kubeconfig.get("clusters"):
return {"success": False, "error": "CreateKubernetesClusterCert returned no kubeconfig clusters"}
_prefer_external_context(kubeconfig)
return {"success": True, "kubeconfig": kubeconfig}
def _prefer_external_context(kubeconfig: Dict[str, Any]) -> None:return {"success": True, "kubeconfig": kubeconfig}
def _prefer_external_context(kubeconfig: Dict[str, Any]) -> None:
external_cluster_name = None
for cluster in kubeconfig.get("clusters", []) or []:
name = cluster.get("name", "")def _prefer_external_context(kubeconfig: Dict[str, Any]) -> None:
external_cluster_name = None
for cluster in kubeconfig.get("clusters", []) or []:
name = cluster.get("name", "")
if "external" in name and "TLS" not in name:
external_cluster_name = namebreak
if not external_cluster_name:
return
for context in kubeconfig.get("contexts", []) or []:
context_data = context.get("context") or {}
if context_data.get("cluster") == external_cluster_name:
kubeconfig["current-context"] = context.get("name")for context in kubeconfig.get("contexts", []) or []:
context_data = context.get("context") or {}
if context_data.get("cluster") == external_cluster_name:
kubeconfig["current-context"] = context.get("name")
returnreturn
def _kubectl_get_with_kubeconfig(region: str, cluster_id: str, resource_args: List[str], ak: Optional[str], sk: Optional[str], project_id: Optional[str]) -> Dict[str, Any]:
cluster_result = _show_cluster(region, cluster_id, ak, sk, project_id)
if not cluster_result.get("success"):
return cluster_resultif not _cluster_has_external_access(cluster_result.get("cluster") or {}):
return {"success": False, "error": "cluster has no bound EIP/external endpoint"}
kubeconfig_result = _create_kubeconfig(region, cluster_id, ak, sk, project_id)
if not kubeconfig_result.get("success"):
return kubeconfig_resultif not _cluster_has_external_access(cluster_result.get("cluster") or {}):
return {"success": False, "error": "cluster has no bound EIP/external endpoint"}
kubeconfig_result = _create_kubeconfig(region, cluster_id, ak, sk, project_id)
if not kubeconfig_result.get("success"):
return kubeconfig_resultreturn {"success": False, "error": "cluster has no bound EIP/external endpoint"}
kubeconfig_result = _create_kubeconfig(region, cluster_id, ak, sk, project_id)
if not kubeconfig_result.get("success"):
return kubeconfig_result
kubeconfig_file = Nonekubeconfig_result = _create_kubeconfig(region, cluster_id, ak, sk, project_id)
if not kubeconfig_result.get("success"):
return kubeconfig_result
kubeconfig_file = None
try:if not kubeconfig_result.get("success"):
return kubeconfig_result
kubeconfig_file = None
try:
with tempfile.NamedTemporaryFile("w", delete=False, suffix=".json") as handle:
json.dump(kubeconfig_result["kubeconfig"], handle)VirusTotal findings are pending for this skill version.
No suspicious patterns detected.