backend/src/appenlight/scripts/initialize_db.py

# -*- coding: utf-8 -*-

# Copyright 2010 - 2017 RhodeCode GmbH and the AppEnlight project authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import argparse
import getpass
import logging

from pyramid.paster import setup_logging, bootstrap
from pyramid.threadlocal import get_current_request
from ziggurat_foundations.models.services.user import UserService


from appenlight.forms import UserRegisterForm
from appenlight.lib.ext_json import json
from appenlight.models import (
    DBSession,
    Group,
    GroupPermission,
    User,
    AuthToken
)
from appenlight.models.services.group import GroupService

log = logging.getLogger(__name__)

_ = str


def is_yes(input_data):
    return input_data in ['y', 'yes']


def is_no(input_data):
    return input_data in ['n', 'no']


def main():
    parser = argparse.ArgumentParser(
        description='Populate AppEnlight database',
        add_help=False)
    parser.add_argument('-c', '--config', required=True,
                        help='Configuration ini file of application')
    parser.add_argument('--username', default=None,
                        help='User  to create')
    parser.add_argument('--password', default=None,
                        help='Password for created user')
    parser.add_argument('--email', default=None,
                        help='Email for created user')
    parser.add_argument('--auth-token', default=None,
                        help='Auth token for created user')
    args = parser.parse_args()
    config_uri = args.config

    setup_logging(config_uri)
    env = bootstrap(config_uri)
    request = env['request']
    with get_current_request().tm:
        group = GroupService.by_id(1)
        if not group:
            group = Group(id=1, group_name='Administrators',
                           description="Top level permission owners")
            DBSession.add(group)
            permission = GroupPermission(perm_name='root_administration')
            group.permissions.append(permission)

    create_user = True if args.username else None
    while create_user is None:
        response = input(
            'Do you want to create a new admin? (n)\n').lower()

        if is_yes(response or 'n'):
            create_user = True
        elif is_no(response or 'n'):
            create_user = False

    if create_user:
        csrf_token = request.session.get_csrf_token()
        user_name = args.username
        print('*********************************************************')
        while user_name is None:
            response = input('What is the username of new admin?\n')
            form = UserRegisterForm(
                user_name=response, csrf_token=csrf_token,
                csrf_context=request)
            form.validate()
            if form.user_name.errors:
                print(form.user_name.errors[0])
            else:
                user_name = response
                print('The admin username is "{}"\n'.format(user_name))
        print('*********************************************************')
        email = args.email
        while email is None:
            response = input('What is the email of admin account?\n')
            form = UserRegisterForm(
                email=response, csrf_token=csrf_token,
                csrf_context=request)
            form.validate()
            if form.email.errors:
                print(form.email.errors[0])
            else:
                email = response
                print('The admin email is "{}"\n'.format(email))
        print('*********************************************************')
        user_password = args.password
        confirmed_password = args.password
        while user_password is None or confirmed_password is None:
            response = getpass.getpass(
                'What is the password for admin account?\n')
            form = UserRegisterForm(
                user_password=response, csrf_token=csrf_token,
                csrf_context=request)
            form.validate()
            if form.user_password.errors:
                print(form.user_password.errors[0])
            else:
                user_password = response

            response = getpass.getpass('Please confirm the password.\n')
            if user_password == response:
                confirmed_password = response
            else:
                print('Passwords do not match. Please try again')
        print('*********************************************************')

    with get_current_request().tm:
        if create_user:
            group = GroupService.by_id(1)
            user = User(user_name=user_name, email=email, status=1)
            UserService.regenerate_security_code(user)
            UserService.set_password(user, user_password)
            DBSession.add(user)
            token = AuthToken(description="Uptime monitoring token")
            if args.auth_token:
                token.token = args.auth_token
            user.auth_tokens.append(token)
            group.users.append(user)
            print('USER CREATED')
            print(json.dumps(user.get_dict()))
            print('*********************************************************')
            print('AUTH TOKEN')
            print(json.dumps(user.auth_tokens[0].get_dict()))