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

from flask import current_app

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.kustomization, self.helmreleases)

    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_json(self):
        """
        represent this object as a json object. Return JSON object
        """

        return {"id": self.id,
                "name": self.name,
                "slug": self.slug,
                "external": self.external,
                "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 kustomization_status: The status of the Kustomization of this app:
    :type kustomization_status: str
    :param helmrelease_status: The status of the helmreleases of this app
    :type helmrelease_status: str[]
    """
    def __init__(self, kustomization, helmreleases):
        self.helmreleases = {}
        if kustomization is not None and "status" in kustomization:
            ks_ready, ks_message = AppStatus.check_condition(kustomization['status'])
            self.installed = True
        else:
            ks_ready = None
            ks_message = "Kustomization does not exist"
            self.installed = False
            self.ready = False
            self.message = "Not installed"
            return

        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
        if ks_ready:
            self.ready = True
            self.message = "Installed"
        else:
            self.ready = False
            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"