"""Everything to do with Apps""" import os from sqlalchemy import ForeignKey, Integer, String from sqlalchemy.orm import relationship from database import db import helpers.kubernetes as k8s from .apps import APPS_WITHOUT_OAUTH, APP_NOT_INSTALLED_STATUS 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) def __repr__(self): return f"{self.id} <{self.name}>" def get_kustomization_status(self): """Returns True if the kustomization for this App is ready""" kustomization = k8s.get_kustomization(self.slug) if kustomization is None: return None return kustomization['status'] def get_helmrelease_status(self): """Returns True if the kustomization for this App is ready""" helmrelease = k8s.get_helmrelease(self.slug, self.namespace) if helmrelease is None: return None return helmrelease['status'] def get_status(self): """Returns a string that describes the app state in the cluster""" ks_status = self.get_kustomization_status() if ks_status is not None: ks_ready, ks_message = App.check_condition(ks_status) else: ks_ready = None hr_status = self.get_helmrelease_status() if hr_status is not None: hr_ready, hr_message = App.check_condition(hr_status) else: hr_ready = None if ks_ready is None: return APP_NOT_INSTALLED_STATUS # *Should* not happen, but just in case: if (ks_ready is None and hr_ready is not None) or \ (hr_ready is None and ks_ready is not None): return ("This app is in a strange state. Contact a Stackspin" " administrator if this status stays for longer than 5 minutes") if ks_ready and hr_ready: return "App installed and running" if not hr_ready: return f"App failed installing: {hr_message}" if not ks_ready: return f"App failed installing: {ks_message}" return "App is installing..." 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- kustomization self.__create_kustomization() 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) # Create a secret that contains the oauth variables for Hydra Maester if self.slug not in APPS_WITHOUT_OAUTH: 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) @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") @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' @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 exist, 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 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}")