# contact with: nfvlabs@tid.es
##
from typing import Union
+from shlex import quote
import os
import yaml
)
command = "{} --kubeconfig={} get namespaces -o=yaml".format(
- self.kubectl_command, paths["kube_config"]
+ self.kubectl_command, quote(paths["kube_config"])
)
output, _rc = await self._local_async_exec(
command=command, raise_exception_on_error=True, env=env
)
command = "{} --kubeconfig={} create namespace {}".format(
- self.kubectl_command, paths["kube_config"], namespace
+ self.kubectl_command, quote(paths["kube_config"]), quote(namespace)
)
_, _rc = await self._local_async_exec(
command=command, raise_exception_on_error=True, env=env
)
command1 = "env KUBECONFIG={} {} get manifest {} --namespace={}".format(
- kubeconfig, self._helm_command, kdu_instance, namespace
+ kubeconfig, self._helm_command, quote(kdu_instance), quote(namespace)
+ )
+ command2 = "{} get --namespace={} -f -".format(
+ self.kubectl_command, quote(namespace)
)
- command2 = "{} get --namespace={} -f -".format(self.kubectl_command, namespace)
output, _rc = await self._local_async_exec_pipe(
command1, command2, env=env, raise_exception_on_error=True
)
"""
inspect_command = "{} show {} {}{} {}".format(
- self._helm_command, show_command, kdu_model, repo_str, version
+ self._helm_command, show_command, quote(kdu_model), repo_str, version
)
return inspect_command
):
get_command = (
"env KUBECONFIG={} {} get {} {} --namespace={} --output yaml".format(
- kubeconfig, self._helm_command, get_command, kdu_instance, namespace
+ kubeconfig,
+ self._helm_command,
+ get_command,
+ quote(kdu_instance),
+ quote(namespace),
)
)
return get_command
cluster_name=cluster_id, create_if_not_exist=True
)
command = "env KUBECONFIG={} {} status {} --namespace={} --output yaml".format(
- paths["kube_config"], self._helm_command, kdu_instance, namespace
+ paths["kube_config"],
+ self._helm_command,
+ quote(kdu_instance),
+ quote(namespace),
)
output, rc = await self._local_async_exec(
# namespace
namespace_str = ""
if namespace:
- namespace_str = "--namespace {}".format(namespace)
+ namespace_str = "--namespace {}".format(quote(namespace))
# version
version_str = ""
"{params} {timeout} {ns} {model} {ver}".format(
kubeconfig=kubeconfig,
helm=self._helm_command,
- name=kdu_instance,
+ name=quote(kdu_instance),
atomic=atomic_str,
params=params_str,
timeout=timeout_str,
ns=namespace_str,
- model=kdu_model,
+ model=quote(kdu_model),
ver=version_str,
)
)
# version
version_str = ""
if version:
- version_str = "--version {}".format(version)
+ version_str = "--version {}".format(quote(version))
# namespace
namespace_str = ""
if namespace:
- namespace_str = "--namespace {}".format(namespace)
+ namespace_str = "--namespace {}".format(quote(namespace))
command = (
"env KUBECONFIG={kubeconfig} {helm} upgrade {name} {model} {namespace} {atomic} {force}"
).format(
kubeconfig=kubeconfig,
helm=self._helm_command,
- name=kdu_instance,
+ name=quote(kdu_instance),
namespace=namespace_str,
atomic=atomic_str,
force=force_str,
params=params_str,
timeout=timeout_str,
- model=kdu_model,
+ model=quote(kdu_model),
ver=version_str,
)
return command
self, kdu_instance: str, namespace: str, revision: float, kubeconfig: str
) -> str:
return "env KUBECONFIG={} {} rollback {} {} --namespace={} --wait".format(
- kubeconfig, self._helm_command, kdu_instance, revision, namespace
+ kubeconfig,
+ self._helm_command,
+ quote(kdu_instance),
+ revision,
+ quote(namespace),
)
def _get_uninstall_command(
self, kdu_instance: str, namespace: str, kubeconfig: str
) -> str:
return "env KUBECONFIG={} {} uninstall {} --namespace={}".format(
- kubeconfig, self._helm_command, kdu_instance, namespace
+ kubeconfig, self._helm_command, quote(kdu_instance), quote(namespace)
)
def _get_helm_chart_repos_ids(self, cluster_uuid) -> list: