dashboard/areas/apps/models.py

"""Everything to do with Apps"""

import os
import base64

from sqlalchemy import ForeignKey, Integer, String, Boolean
from sqlalchemy.orm import relationship

from database import db
import helpers.kubernetes as k8s


DEFAULT_APP_SUBDOMAINS = {
    "nextcloud": "files",
    "wordpress": "www",
    "monitoring": "grafana",
}

class App(db.Model):
    """
    The App object, interact with the App database object. Data is stored in
    the local database.
    """

    id = db.Column(Integer, primary_key=True)
    name = db.Column(String(length=64))
    slug = db.Column(String(length=64), unique=True)
    external = db.Column(Boolean, unique=False, nullable=False, default=True, server_default='0')
    # The URL is only stored in the DB for external applications; otherwise the
    # URL is stored in a configmap (see get_url)
    url = db.Column(String(length=128), unique=False)

    def __repr__(self):
        return f"{self.id} <{self.name}>"

    def get_url(self):
        """
        Returns the URL where this application is running

        For external applications: the URL is stored in the database

        For internal applications: the URL is stored in a configmap named
        `stackspin-{self.slug}-kustomization-variables under
        `{self.slug_domain}`. This function reads that configmap. If the
        configmap does not contain a URL for the application (which is
        possible, if the app is not installed yet, for example), we return a
        default URL.
        """

        if self.external:
            return self.url

        # Get domain name from configmap
        ks_config_map = k8s.get_kubernetes_config_map_data(
                f"stackspin-{self.slug}-kustomization-variables",
                "flux-system")
        domain_key = f"{self.slug}_domain"

        # If config map found with this domain name for this service, return
        # that URL
        if ks_config_map and domain_key in ks_config_map.keys():
            return f"https://{ks_config_map[domain_key]}"

        domain_secret = k8s.get_kubernetes_secret_data(
                "stackspin-cluster-variables",
                "flux-system")
        domain = base64.b64decode(domain_secret['domain']).decode()

        # See if there is another default subdomain for this app than just
        # "slug.{domain}"
        if self.slug in DEFAULT_APP_SUBDOMAINS:
            return f"https://{DEFAULT_APP_SUBDOMAINS[self.slug]}.{domain}"

        # No default known
        return f"https://{self.slug}.{domain}"

    def get_status(self):
        """Returns an AppStatus object that describes the current cluster state"""
        return AppStatus(self)

    def install(self):
        """Creates a Kustomization in the Kubernetes cluster that installs this application"""
        # Generate the necessary passwords, etc. from a template
        self.__generate_secrets()
        # Create add-<app> kustomization
        self.__create_kustomization()

    def uninstall(self):
        """
        Delete the app kustomization.

        In our case, this triggers a deletion of the app's PVCs (so deletes all
        data), as well as any other Kustomizations and HelmReleases related to
        the app. It also triggers a deletion of the OAuth2Client object, but
        does not delete the secrets generated by the `install` command. It also
        does not remove the TLS secret generated by cert-manager.
        """
        self.__delete_kustomization()

    def delete(self):
        """
        Fully deletes an application

        This includes user roles, all kubernetes objects and also PVCs, so your
        data will be *gone*
        """
        # Delete all roles first
        for role in self.roles:
            db.session.delete(role)
        db.session.commit()

        db.session.delete(self)
        return db.session.commit()

    def __generate_secrets(self):
        """Generates passwords for app installation"""
        # Create app variables secret
        if self.variables_template_filepath:
            k8s.create_variables_secret(self.slug, self.variables_template_filepath)

        k8s.create_variables_secret(
            self.slug,
            os.path.join(
                self.__get_templates_dir(),
                "stackspin-oauth-variables.yaml.jinja"
            )
        )

    def __create_kustomization(self):
        """Creates the `add-{app_slug}` kustomization in the Kubernetes cluster"""
        kustomization_template_filepath = \
            os.path.join(self.__get_templates_dir(),
                "add-app-kustomization.yaml.jinja")
        k8s.store_kustomization(kustomization_template_filepath, self.slug)

    def __delete_kustomization(self):
        """Deletes kustomization for this app"""
        k8s.delete_kustomization(f"add-{self.slug}")


    @property
    def variables_template_filepath(self):
        """Path to the variables template used to generate secrets the app needs"""
        variables_template_filepath = os.path.join(self.__get_templates_dir(),
                f"stackspin-{self.slug}-variables.yaml.jinja")
        if os.path.exists(variables_template_filepath):
            return variables_template_filepath
        return None

    @property
    def namespace(self):
        """
        Returns the Kubernetes namespace of this app

        FIXME: This should probably become a database field.
        """
        if self.slug in ['nextcloud', 'wordpress', 'wekan', 'zulip']:
            return 'stackspin-apps'
        return 'stackspin'

    @property
    def roles(self):
        """
        All roles that are linked to this app
        """
        return AppRole.query.filter_by(
            app_id=self.id
        ).all()

    @property
    def kustomization(self):
        """Returns the kustomization object for this app"""
        return k8s.get_kustomization(self.slug)

    def to_dict(self):
        """
        represent this object as a dict, compatible for JSON output
        """

        return {"id": self.id,
                "name": self.name,
                "slug": self.slug,
                "external": self.external,
                "status": self.get_status().to_dict(),
                "url": self.get_url()}

    @property
    def helmreleases(self):
        """Returns the helmreleases associated with the kustomization for this app"""
        return k8s.get_all_helmreleases(self.namespace,
                f"kustomize.toolkit.fluxcd.io/name={self.slug}")

    @staticmethod
    def __get_templates_dir():
        """Returns directory that contains the Jinja templates used to create app secrets."""
        return os.path.join(os.path.dirname(os.path.realpath(__file__)), "templates")


class AppRole(db.Model):  # pylint: disable=too-few-public-methods
    """
    The AppRole object, stores the roles Users have on Apps
    """

    user_id = db.Column(String(length=64), primary_key=True)
    app_id = db.Column(Integer, ForeignKey("app.id"), primary_key=True)
    role_id = db.Column(Integer, ForeignKey("role.id"))

    role = relationship("Role")

    def __repr__(self):
        return (f"role_id: {self.role_id}, user_id: {self.user_id},"
                f" app_id: {self.app_id}, role: {self.role}")

class AppStatus():  # pylint: disable=too-few-public-methods
    """
    Represents the status of an app in the Kubernetes cluster.

    This class can answer a few questions, like "is the app installed?", but
    can also return raw status messages from Kustomizations and HelmReleases

    This constructor sets three variables:

    self.installed (bool): Whether the app should be installed
    self.ready (bool): Whether the app is installed correctly
    self.message (str): Information about the status

    :param app: An app of which the kustomization and helmreleases
        property will be used.
    :type app: App
    """
    def __init__(self, app):
        kustomization = app.kustomization
        if kustomization is not None and "status" in kustomization:
            ks_ready, ks_message = AppStatus.check_condition(kustomization['status'])
            self.installed = True
            if ks_ready:
                self.ready = ks_ready
                self.message = "Installed"
                return
        else:
            ks_ready = None
            ks_message = "Kustomization does not exist"
            self.installed = False
            self.ready = False
            self.message = "Not installed"
            return

        helmreleases = app.helmreleases
        for helmrelease in helmreleases:
            hr_status = helmrelease['status']
            hr_ready, hr_message = AppStatus.check_condition(hr_status)

            # For now, only show the message of the first HR that isn't ready
            if not hr_ready:
                self.ready = False
                self.message = f"HelmRelease {helmrelease['metadata']['name']} status: {hr_message}"
                return

        # If we end up here, all HRs are ready, but the kustomization is not
        self.ready = ks_ready
        self.message = f"App Kustomization status: {ks_message}"

    def __repr__(self):
        return f"Installed: {self.installed}\tReady: {self.ready}\tMessage: {self.message}"

    @staticmethod
    def check_condition(status):
        """
        Returns a tuple that has true/false for readiness and a message

        Ready, in this case means that the condition's type == "Ready" and its
        status == "True". If the condition type "Ready" does not occur, the
        status is interpreted as not ready.

        The message that is returned is the message that comes with the
        condition with type "Ready"

        :param status: Kubernetes resource's "status" object.
        :type status: dict
        """
        for condition in status["conditions"]:
            if condition["type"] == "Ready":
                return condition["status"] == "True", condition["message"]
        return False, "Condition with type 'Ready' not found"

    def to_dict(self):
        """Represents this app status as a dict"""
        return {
            "installed": self.installed,
            "ready": self.ready,
            "message": self.message,
        }