magic-app/second.py

"""Less flashy version. More hard-coding."""

from hashlib import md5
from json import dumps, loads
from os import environ
from os.path import exists
from pathlib import Path
from random import choice
from shlex import split
from string import ascii_lowercase
from subprocess import run
from typing import Dict

from flask import Flask, render_template, request
from flask_wtf import FlaskForm
from ruamel.yaml import YAML
from wtforms import PasswordField, StringField
from wtforms.validators import URL, DataRequired, Length

app = Flask(__name__)
app.secret_key = b'_5#y2L"F4Q8z\n\xec]/'

yaml = YAML()

APPS_SPEC = {
    "gitea": {
        "url": "https://git.autonomic.zone/compose-stacks/gitea",
        "secrets": [
            "db_passwd",
            "db_root_passwd",
            "internal_token",
            "jwt_secret",
            "secret_key",
        ],
        "configs": ["app_ini"],
    }
}
DATA_DIR = Path("./data")


def get_secret(n: int) -> str:
    """Generate a ASCII secret that is n chars long."""
    return "".join(choice(ascii_lowercase) for _ in range(n))


def clone_app_template(app_name: str) -> None:
    """Git clone an app template repository."""
    clone_path = DATA_DIR / app_name
    clone_url = APPS_SPEC[app_name]["url"]
    run(split(f"git clone {clone_url} {clone_path}"))


def dump_db(db: Dict) -> None:
    """Dump the database."""
    with open(DATA_DIR / "db.json", "w") as handle:
        handle.write(dumps(db))


def load_db() -> Dict:
    """Load the database."""
    db_path = DATA_DIR / "db.json"
    if exists(db_path):
        with open(db_path, "r") as handle:
            return loads(handle.read())
    return {}


def get_hash(value: str) -> str:
    """Hash a value for swarm versioning with good 'ol md5."""
    hasher = md5()
    hasher.update(value.encode())
    return hasher.hexdigest()


def get_secret_names(app_name, form_data, env):
    """Generate versioned names for secrets that swarm will accept"""
    for secret in APPS_SPEC[app_name]["secrets"]:
        if secret in form_data:
            hashed = get_hash(form_data[secret])
            env_key = f"{secret.upper()}_SECRET_VERSION"
            env[env_key] = f"{secret}_{hashed}"
    return env


def get_config_names(app_name, form_data, env):
    """Generate versioned names for configs that swarm will accept"""
    for config in APPS_SPEC[app_name]["configs"]:
        if config in form_data:
            hashed = get_hash(form_data[config])
            env_key = f"{config.upper()}_CONFIG_VERSION"
            env[env_key] = f"{secret}_{hashed}"
    return env


def stack_deploy(app_name, env):
    """Depoy an application to the swarm."""
    compose_yml = DATA_DIR / app_name / "compose.yml"
    command = f"docker stack deploy -c {compose_yml} {app_name}"
    run(split(command), env=env)


def get_loaded_env(app_name, request_form):
    """Load environment from install form for compose.yml injection."""
    environment = environ.copy()
    for key in request_form.keys():
        environment[key.upper()] = request.form[key]
    return environment


class GiteaInstallForm(FlaskForm):
    """Gitea installation form."""

    app_name = StringField("Application name", default="Git with a cup of tea")
    domain = StringField("Domain name", validators=[DataRequired()],)

    db_host = StringField("Database host", default="mariadb:3306")
    db_name = StringField("Database name", default="gitea")
    db_passwd = PasswordField(
        "Database password", validators=[DataRequired(), Length(min=32)],
    )
    db_root_passwd = PasswordField(
        "Database root password", validators=[DataRequired(), Length(min=32)],
    )
    db_type = StringField("Database type", default="mysql")
    db_user = StringField("Database user", default="mysql")
    internal_token = PasswordField(
        "Internal secret token", validators=[DataRequired(), Length(min=105)],
    )
    jwt_secret = PasswordField(
        "JWT secret", validators=[DataRequired(), Length(min=43)],
    )
    secret_key = PasswordField(
        "Secret key", validators=[DataRequired(), Length(min=64)],
    )
    ssh_port = StringField("SSH port", default="2222")


@app.route("/")
def index():
    """Home page for app installation possibilities."""
    return render_template("second/index.html", apps=[app for app in APPS_SPEC])


@app.route("/install/<app_name>")
def install(app_name):
    """Installation page for an app."""
    if app_name == "gitea":
        form = GiteaInstallForm()
    return render_template("second/install.html", app_name=app_name, form=form)


@app.route("/deploy/<app_name>", methods=["POST"])
def deploy(app_name):
    """Deployment end-point for an app."""
    if app_name == "gitea":
        form = GiteaInstallForm(request.form)

    if not form.validate():
        return render_template("second/install.html", app_name=app_name, form=form)

    env = get_loaded_env(app_name, form.data)
    env = get_secret_names(app_name, form.data, env)
    env = get_config_names(app_name, form.data, env)

    app.logger.info(f"env -> {env}")

    dump_db({app_name: form.data})

    stack_deploy(app_name, env)


if __name__ == "__main__":
    app.run(debug=True, host="0.0.0.0")