libcloud/common/kubernetes.py

# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You 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.

"""
Module which contains common Kubernetes related code.
"""

from typing import Optional

import os
import base64
import warnings

from libcloud.utils.py3 import httplib
from libcloud.utils.py3 import b

from libcloud.common.base import JsonResponse, ConnectionUserAndKey
from libcloud.common.base import KeyCertificateConnection, ConnectionKey
from libcloud.common.types import InvalidCredsError

__all__ = [
    "KubernetesException",
    "KubernetesBasicAuthConnection",
    "KubernetesTLSAuthConnection",
    "KubernetesTokenAuthConnection",
    "KubernetesDriverMixin",
    "VALID_RESPONSE_CODES",
]

VALID_RESPONSE_CODES = [
    httplib.OK,
    httplib.ACCEPTED,
    httplib.CREATED,
    httplib.NO_CONTENT,
]


class KubernetesException(Exception):
    def __init__(self, code, message):
        self.code = code
        self.message = message
        self.args = (code, message)

    def __str__(self):
        return "%s %s" % (self.code, self.message)

    def __repr__(self):
        return "KubernetesException %s %s" % (self.code, self.message)


class KubernetesResponse(JsonResponse):

    valid_response_codes = [
        httplib.OK,
        httplib.ACCEPTED,
        httplib.CREATED,
        httplib.NO_CONTENT,
    ]

    def parse_error(self):
        if self.status == 401:
            raise InvalidCredsError("Invalid credentials")
        return self.body

    def success(self):
        return self.status in self.valid_response_codes


class KubernetesTLSAuthConnection(KeyCertificateConnection):
    responseCls = KubernetesResponse
    timeout = 60

    def __init__(
        self,
        key,
        secure=True,
        host="localhost",
        port="6443",
        key_file=None,
        cert_file=None,
        **kwargs,
    ):

        super(KubernetesTLSAuthConnection, self).__init__(
            key_file=key_file,
            cert_file=cert_file,
            secure=secure,
            host=host,
            port=port,
            url=None,
            proxy_url=None,
            timeout=None,
            backoff=None,
            retry_delay=None,
        )

        if key_file:
            keypath = os.path.expanduser(key_file)
            is_file_path = os.path.exists(keypath) and os.path.isfile(keypath)

            if not is_file_path:
                raise InvalidCredsError(
                    "You need an key PEM file to authenticate "
                    "via tls. For more info please visit:"
                    "https://kubernetes.io/docs/concepts/"
                    "cluster-administration/certificates/"
                )

            self.key_file = key_file

            certpath = os.path.expanduser(cert_file)
            is_file_path = os.path.exists(certpath) and os.path.isfile(certpath)

            if not is_file_path:
                raise InvalidCredsError(
                    "You need an certificate PEM file to authenticate "
                    "via tls. For more info please visit:"
                    "https://kubernetes.io/docs/concepts/"
                    "cluster-administration/certificates/"
                )

            self.cert_file = cert_file

    def add_default_headers(self, headers):
        if "Content-Type" not in headers:
            headers["Content-Type"] = "application/json"
        return headers


class KubernetesTokenAuthConnection(ConnectionKey):
    responseCls = KubernetesResponse
    timeout = 60

    def add_default_headers(self, headers):
        if "Content-Type" not in headers:
            headers["Content-Type"] = "application/json"
        if self.key:
            headers["Authorization"] = "Bearer " + self.key
        else:
            raise ValueError("Please provide a valid token in the key param")
        return headers


class KubernetesBasicAuthConnection(ConnectionUserAndKey):
    responseCls = KubernetesResponse
    timeout = 60

    def add_default_headers(self, headers):
        """
        Add parameters that are necessary for every request
        If user and password are specified, include a base http auth
        header
        """
        if "Content-Type" not in headers:
            headers["Content-Type"] = "application/json"

        if self.user_id and self.key:
            auth_string = b("%s:%s" % (self.user_id, self.key))
            user_b64 = base64.b64encode(auth_string)
            headers["Authorization"] = "Basic %s" % (user_b64.decode("utf-8"))
        return headers


class KubernetesDriverMixin(object):
    """
    Base driver class to be used with various Kubernetes drivers.

    NOTE: This base class can be used in different APIs such as container and
    compute one.
    """

    def __init__(
        self,
        key=None,
        secret=None,
        secure=False,
        host="localhost",
        port=4243,
        key_file=None,
        cert_file=None,
        ca_cert=None,
        ex_token_bearer_auth=False,
    ):
        """
        :param    key: API key or username to be used (required)
        :type     key: ``str``

        :param    secret: Secret password to be used (required)
        :type     secret: ``str``

        :param    secure: Whether to use HTTPS or HTTP. Note: Some providers
                          only support HTTPS, and it is on by default.
        :type     secure: ``bool``

        :param    host: Override hostname used for connections.
        :type     host: ``str``

        :param    port: Override port used for connections.
        :type     port: ``int``

        :param    key_file: Path to the key file used to authenticate (when
                            using key file auth).
        :type     key_file: ``str``

        :param    cert_file: Path to the cert file used to authenticate (when
                             using key file auth).
        :type     cert_file: ``str``

        :param    ex_token_bearer_auth: True to use token bearer auth.
        :type     ex_token_bearer_auth: ``bool``

        :return: ``None``
        """
        if ex_token_bearer_auth:
            self.connectionCls = KubernetesTokenAuthConnection
            if not key:
                msg = 'The token must be a string provided via "key" argument'
                raise ValueError(msg)
            secure = True

        if key_file or cert_file:
            # Certificate based auth is used
            if not (key_file and cert_file):
                raise ValueError("Both key and certificate files are needed")

        if key_file:
            self.connectionCls = KubernetesTLSAuthConnection
            self.key_file = key_file
            self.cert_file = cert_file
            secure = True

        if host and host.startswith("https://"):
            secure = True

        host = self._santize_host(host=host)

        super(KubernetesDriverMixin, self).__init__(
            key=key,
            secret=secret,
            secure=secure,
            host=host,
            port=port,
            key_file=key_file,
            cert_file=cert_file,
        )

        if ca_cert:
            self.connection.connection.ca_cert = ca_cert
        else:
            # do not verify SSL certificate
            warnings.warn(
                "Kubernetes has its own CA, since you didn't supply "
                "a CA certificate be aware that SSL verification "
                "will be disabled for this session."
            )
            self.connection.connection.ca_cert = False

        self.connection.secure = secure
        self.connection.host = host

        if port is not None:
            self.connection.port = port

    def _ex_connection_class_kwargs(self):
        kwargs = {}
        if hasattr(self, "key_file"):
            kwargs["key_file"] = self.key_file
        if hasattr(self, "cert_file"):
            kwargs["cert_file"] = self.cert_file
        return kwargs

    def _santize_host(self, host=None):
        # type: (Optional[str]) -> Optional[str]
        """
        Sanitize "host" argument any remove any protocol prefix (if specified).
        """
        if not host:
            return None

        prefixes = ["http://", "https://"]
        for prefix in prefixes:
            if host.startswith(prefix):
                host = host.lstrip(prefix)

        return host