n2vc/kubectl.py

# Copyright 2020 Canonical Ltd.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
#     Unless required by applicable law or agreed to in writing, software
#     distributed under the License is distributed on an "AS IS" BASIS,
#     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#     See the License for the specific language governing permissions and
#     limitations under the License.

import base64
import logging
from typing import Dict
import typing


from kubernetes import client, config
from kubernetes.client.models import (
    V1ClusterRole,
    V1ObjectMeta,
    V1PolicyRule,
    V1ServiceAccount,
    V1ClusterRoleBinding,
    V1RoleRef,
    V1Subject,
)
from kubernetes.client.rest import ApiException
from retrying_async import retry


SERVICE_ACCOUNT_TOKEN_KEY = "token"
SERVICE_ACCOUNT_ROOT_CA_KEY = "ca.crt"
# clients
CORE_CLIENT = "core_v1"
RBAC_CLIENT = "rbac_v1"
STORAGE_CLIENT = "storage_v1"


class Kubectl:
    def __init__(self, config_file=None):
        config.load_kube_config(config_file=config_file)
        self._clients = {
            CORE_CLIENT: client.CoreV1Api(),
            RBAC_CLIENT: client.RbacAuthorizationV1Api(),
            STORAGE_CLIENT: client.StorageV1Api(),
        }
        self._configuration = config.kube_config.Configuration.get_default_copy()
        self.logger = logging.getLogger("Kubectl")

    @property
    def configuration(self):
        return self._configuration

    @property
    def clients(self):
        return self._clients

    def get_services(
        self,
        field_selector: str = None,
        label_selector: str = None,
    ) -> typing.List[typing.Dict]:
        """
        Get Service list from a namespace

        :param: field_selector:     Kubernetes field selector for the namespace
        :param: label_selector:     Kubernetes label selector for the namespace

        :return: List of the services matching the selectors specified
        """
        kwargs = {}
        if field_selector:
            kwargs["field_selector"] = field_selector
        if label_selector:
            kwargs["label_selector"] = label_selector
        try:
            result = self.clients[CORE_CLIENT].list_service_for_all_namespaces(**kwargs)
            return [
                {
                    "name": i.metadata.name,
                    "cluster_ip": i.spec.cluster_ip,
                    "type": i.spec.type,
                    "ports": [
                        {
                            "name": p.name,
                            "node_port": p.node_port,
                            "port": p.port,
                            "protocol": p.protocol,
                            "target_port": p.target_port,
                        }
                        for p in i.spec.ports
                    ]
                    if i.spec.ports
                    else [],
                    "external_ip": [i.ip for i in i.status.load_balancer.ingress]
                    if i.status.load_balancer.ingress
                    else None,
                }
                for i in result.items
            ]
        except ApiException as e:
            self.logger.error("Error calling get services: {}".format(e))
            raise e

    def get_default_storage_class(self) -> str:
        """
        Default storage class

        :return:    Returns the default storage class name, if exists.
                    If not, it returns the first storage class.
                    If there are not storage classes, returns None
        """
        storage_classes = self.clients[STORAGE_CLIENT].list_storage_class()
        selected_sc = None
        default_sc_annotations = {
            "storageclass.kubernetes.io/is-default-class": "true",
            # Older clusters still use the beta annotation.
            "storageclass.beta.kubernetes.io/is-default-class": "true",
        }
        for sc in storage_classes.items:
            if not selected_sc:
                # Select the first storage class in case there is no a default-class
                selected_sc = sc.metadata.name
            annotations = sc.metadata.annotations or {}
            if any(
                k in annotations and annotations[k] == v
                for k, v in default_sc_annotations.items()
            ):
                # Default storage
                selected_sc = sc.metadata.name
                break
        return selected_sc

    def create_cluster_role(
        self,
        name: str,
        labels: Dict[str, str],
        namespace: str = "kube-system",
    ):
        """
        Create a cluster role

        :param: name:       Name of the cluster role
        :param: labels:     Labels for cluster role metadata
        :param: namespace:  Kubernetes namespace for cluster role metadata
                            Default: kube-system
        """
        cluster_roles = self.clients[RBAC_CLIENT].list_cluster_role(
            field_selector="metadata.name={}".format(name)
        )

        if len(cluster_roles.items) > 0:
            raise Exception(
                "Cluster role with metadata.name={} already exists".format(name)
            )

        metadata = V1ObjectMeta(name=name, labels=labels, namespace=namespace)
        # Cluster role
        cluster_role = V1ClusterRole(
            metadata=metadata,
            rules=[
                V1PolicyRule(api_groups=["*"], resources=["*"], verbs=["*"]),
                V1PolicyRule(non_resource_ur_ls=["*"], verbs=["*"]),
            ],
        )

        self.clients[RBAC_CLIENT].create_cluster_role(cluster_role)

    def delete_cluster_role(self, name: str):
        """
        Delete a cluster role

        :param: name:       Name of the cluster role
        """
        self.clients[RBAC_CLIENT].delete_cluster_role(name)

    def create_service_account(
        self,
        name: str,
        labels: Dict[str, str],
        namespace: str = "kube-system",
    ):
        """
        Create a service account

        :param: name:       Name of the service account
        :param: labels:     Labels for service account metadata
        :param: namespace:  Kubernetes namespace for service account metadata
                            Default: kube-system
        """
        service_accounts = self.clients[CORE_CLIENT].list_namespaced_service_account(
            namespace, field_selector="metadata.name={}".format(name)
        )
        if len(service_accounts.items) > 0:
            raise Exception(
                "Service account with metadata.name={} already exists".format(name)
            )

        metadata = V1ObjectMeta(name=name, labels=labels, namespace=namespace)
        service_account = V1ServiceAccount(metadata=metadata)

        self.clients[CORE_CLIENT].create_namespaced_service_account(
            namespace, service_account
        )

    def delete_service_account(self, name: str, namespace: str = "kube-system"):
        """
        Delete a service account

        :param: name:       Name of the service account
        :param: namespace:  Kubernetes namespace for service account metadata
                            Default: kube-system
        """
        self.clients[CORE_CLIENT].delete_namespaced_service_account(name, namespace)

    def create_cluster_role_binding(
        self, name: str, labels: Dict[str, str], namespace: str = "kube-system"
    ):
        """
        Create a cluster role binding

        :param: name:       Name of the cluster role
        :param: labels:     Labels for cluster role binding metadata
        :param: namespace:  Kubernetes namespace for cluster role binding metadata
                            Default: kube-system
        """
        role_bindings = self.clients[RBAC_CLIENT].list_cluster_role_binding(
            field_selector="metadata.name={}".format(name)
        )
        if len(role_bindings.items) > 0:
            raise Exception("Generated rbac id already exists")

        role_binding = V1ClusterRoleBinding(
            metadata=V1ObjectMeta(name=name, labels=labels),
            role_ref=V1RoleRef(kind="ClusterRole", name=name, api_group=""),
            subjects=[V1Subject(kind="ServiceAccount", name=name, namespace=namespace)],
        )
        self.clients[RBAC_CLIENT].create_cluster_role_binding(role_binding)

    def delete_cluster_role_binding(self, name: str):
        """
        Delete a cluster role binding

        :param: name:       Name of the cluster role binding
        """
        self.clients[RBAC_CLIENT].delete_cluster_role_binding(name)

    @retry(
        attempts=10,
        delay=1,
        fallback=Exception("Failed getting the secret from service account"),
    )
    async def get_secret_data(
        self, name: str, namespace: str = "kube-system"
    ) -> (str, str):
        """
        Get secret data

        :param: name:       Name of the secret data
        :param: namespace:  Name of the namespace where the secret is stored

        :return: Tuple with the token and client certificate
        """
        v1_core = self.clients[CORE_CLIENT]

        secret_name = None

        service_accounts = v1_core.list_namespaced_service_account(
            namespace, field_selector="metadata.name={}".format(name)
        )
        if len(service_accounts.items) == 0:
            raise Exception(
                "Service account not found with metadata.name={}".format(name)
            )
        service_account = service_accounts.items[0]
        if service_account.secrets and len(service_account.secrets) > 0:
            secret_name = service_account.secrets[0].name
        if not secret_name:
            raise Exception(
                "Failed getting the secret from service account {}".format(name)
            )
        secret = v1_core.list_namespaced_secret(
            namespace, field_selector="metadata.name={}".format(secret_name)
        ).items[0]

        token = secret.data[SERVICE_ACCOUNT_TOKEN_KEY]
        client_certificate_data = secret.data[SERVICE_ACCOUNT_ROOT_CA_KEY]

        return (
            base64.b64decode(token).decode("utf-8"),
            base64.b64decode(client_certificate_data).decode("utf-8"),
        )