viper-web

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# This file is part of Viper - https://github.com/viper-framework/viper
# See the file 'LICENSE' for copying permission.

import os
import random
# import logging
import argparse

from viper.common.out import print_info, print_warning, print_success
from viper.core.config import __config__

# log = logging.getLogger("viper-web")
# Viper Global Config
cfg = __config__

# insert path and load Django settings
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "viper.web.settings")

import django  # noqa
django.setup()
from django.conf import settings
from django.contrib.auth.models import User  # noqa
from django.core.exceptions import ObjectDoesNotExist  # noqa
from django.core.management import execute_from_command_line  # noqa
from django.db import connections, DEFAULT_DB_ALIAS  # noqa
from django.db.migrations.executor import MigrationExecutor  # noqa
from rest_framework.authtoken.models import Token  # noqa


def is_database_synchronized(database):
    connection = connections[database]
    connection.prepare_database()
    executor = MigrationExecutor(connection)
    targets = executor.loader.graph.leaf_nodes()
    return False if executor.migration_plan(targets) else True


# Run the Web Server
if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('-H', '--host', help='bind to host (e.g. 127.0.0.1 or 0.0.0.0)', action='store', required=False)
    parser.add_argument('-p', '--port', help='bind to port (e.g. 8080)', action='store', required=False)
    parser.add_argument('--tls', help='enable TLS', action='store_true', required=False)
    parser.add_argument('-c', '--certificate', help='path to .crt file', action='store', required=False)
    parser.add_argument('-k', '--key', help='path to .key file', action='store', required=False)
    args = parser.parse_args()

    if args.host:
        web_host = args.host
    elif cfg.get("web").host:
        web_host = cfg.get("web").host
    else:
        web_host = "127.0.0.1"

    if args.port:
        web_port = args.port
    elif cfg.get("web").port:
        web_port = cfg.get("web").port
    else:
        web_port = 8080

    if args.tls:
        web_tls = args.tls
    elif cfg.get("web").tls:
        web_tls = cfg.get("web").host
    else:
        web_tls = False

    if args.certificate:
        web_certificate = args.certificate
    elif cfg.get("web").certificate:
        web_certificate = cfg.get("web").certificate
    else:
        web_certificate = None

    if args.key:
        web_key = args.key
    elif cfg.get("web").key:
        web_key = cfg.get("web").key
    else:
        web_key = None

    if web_tls:
        if not (web_certificate and web_key):
            raise Exception("please provide --certificate and --key parameter for TLS")

    # check for outstanding migrations
    if is_database_synchronized(DEFAULT_DB_ALIAS):
        # All migrations have been applied.
        print_info("No outstanding Django DB migrations".format())
    else:
        # Unapplied migrations found.
        print_warning("There are outstanding Django DB migrations".format())
        execute_from_command_line(["web/manage.py", "migrate", "--verbosity", "0"])
        if not is_database_synchronized(DEFAULT_DB_ALIAS):
            raise Exception("there was a problem applying the Django DB migrations")
        print_success("Applied outstanding migrations".format())

    # check for admin user
    if cfg.get("web").admin_username:
        admin_username = cfg.get("web").admin_username
    else:
        admin_username = "admin"

    try:
        admin = User.objects.filter(username=admin_username).get()
        print_info("Found Admin with username: {}".format(admin_username))
    except ObjectDoesNotExist as err:
        if cfg.get("web").admin_password:
            pw = cfg.get("web").admin_password
        else:
            pw = ''.join(random.choice('abcdefghjklmnpqrstuvwxABCDEFGHJKLMNPQRSTUVWX23456789') for _ in range(10))
        admin = User.objects.create_superuser(admin_username, 'admin@example.com', pw)
        print_success("Created \"{}\" with initial password: {}".format(admin_username, pw))

    try:
        admin_auth_token_key = admin.auth_token.key
        # log.debug("Found existing API Auth Token for \"{}\": {}".format(admin_username, admin_auth_token_key))
    except ObjectDoesNotExist as err:
        Token.objects.create(user=admin)
        # print_success("Created new API Auth Token for \"{}\": {}".format(admin_username, admin.auth_token.key))

    # setup static files (if DEBUG=True served by Django; if DEBUG=False served by Whitenoise
    execute_from_command_line(["manage.py", "collectstatic", "--no-input", "--verbosity", "0"])

    # run Django built-in Web Server
    print_info("Starting Web Server on {}:{}".format(web_host, web_port))
    # log.info("Starting Web Server on {}:{}".format(web_host, web_port))

    if settings.DEBUG:
        print_warning("Running with DEBUG enabled - use this for debugging/development only!")
        if web_tls:
            execute_from_command_line(["manage.py", "runsslserver",
                                       "--nostatic",
                                       "--certificate", web_certificate, "--key", web_key,
                                       "{}:{}".format(web_host, web_port)])
        else:
            execute_from_command_line(["manage.py", "runserver", "{}:{}".format(web_host, web_port)])

    else:
        if web_tls:
            execute_from_command_line(["manage.py", "runsslserver", "--noreload",
                                       "--nostatic",
                                       "--certificate", web_certificate, "--key", web_key,
                                       "{}:{}".format(web_host, web_port)])
        else:
            execute_from_command_line(["manage.py", "runserver", "--noreload", "{}:{}".format(web_host, web_port)])