haproxy-wi/app/routes/main/routes.py

import os
import sys
import pytz

from flask import render_template, request, session, g, abort
from flask_login import login_required

sys.path.append(os.path.join(sys.path[0], '/var/www/haproxy-wi/app'))

from app import app, cache
from app.routes.main import bp
import app.modules.db.sql as sql
from modules.db.db_model import conn
from middleware import check_services, get_user_params
import app.modules.common.common as common
import app.modules.roxywi.roxy as roxy
import app.modules.roxywi.auth as roxywi_auth
import app.modules.roxywi.nettools as nettools_mod
import app.modules.roxywi.common as roxywi_common
import app.modules.service.common as service_common
import app.modules.service.haproxy as service_haproxy


@app.errorhandler(403)
@get_user_params()
def page_is_forbidden(e):
    kwargs = {
        'user_params': g.user_params,
        'title': e,
        'e': e
    }
    return render_template('error.html', **kwargs), 403


@app.errorhandler(404)
@get_user_params()
def page_not_found(e):
    kwargs = {
        'user_params': g.user_params,
        'title': e,
        'e': e
    }
    return render_template('error.html', **kwargs), 404


@app.errorhandler(405)
@get_user_params()
def method_not_allowed(e):
    kwargs = {
        'user_params': g.user_params,
        'title': e,
        'e': e
    }
    return render_template('error.html', **kwargs), 405


@app.errorhandler(500)
@get_user_params()
def internal_error(e):
    kwargs = {
        'user_params': g.user_params,
        'title': e,
        'e': e
    }
    return render_template('error.html', **kwargs), 500


@app.before_request
def make_session_permanent():
    session.permanent = True


@app.teardown_request
def _db_close(exc):
    if not conn.is_closed():
        conn.close()


@bp.route('/stats/<service>/', defaults={'serv': None})
@bp.route('/stats/<service>/<serv>')
@login_required
@check_services
@get_user_params()
def stats(service, serv):
    kwargs = {
        'autorefresh': 1,
        'serv': serv,
        'service': service,
        'service_desc': sql.select_service(service),
        'lang': g.user_params['lang']
    }
    return render_template('statsview.html', **kwargs)


@bp.route('/stats/view/<service>/<server_ip>')
@login_required
@check_services
def show_stats(service, server_ip):
    server_ip = common.is_ip_or_dns(server_ip)

    if service in ('nginx', 'apache'):
        return service_common.get_stat_page(server_ip, service)
    else:
        return service_haproxy.stat_page_action(server_ip)


@bp.route('/nettools')
@login_required
@get_user_params(1)
def nettools():
    return render_template('nettools.html', lang=g.user_params['lang'])


@bp.post('/nettols/<check>')
@login_required
def nettols_check(check):
    server_from = common.checkAjaxInput(request.form.get('server_from'))
    server_to = common.is_ip_or_dns(request.form.get('server_to'))
    action = common.checkAjaxInput(request.form.get('nettools_action'))
    port_to = common.checkAjaxInput(request.form.get('nettools_telnet_port_to'))
    dns_name = common.checkAjaxInput(request.form.get('nettools_nslookup_name'))
    dns_name = common.is_ip_or_dns(dns_name)
    record_type = common.checkAjaxInput(request.form.get('nettools_nslookup_record_type'))

    if check == 'icmp':
        return nettools_mod.ping_from_server(server_from, server_to, action)
    elif check == 'tcp':
        return nettools_mod.telnet_from_server(server_from, server_to, port_to)
    elif check == 'dns':
        return nettools_mod.nslookup_from_server(server_from, dns_name, record_type)
    else:
        return 'error: Wrong check'


@bp.route('/history/<service>/<server_ip>')
@login_required
@get_user_params()
def service_history(service, server_ip):
    history = ''
    server_ip = common.checkAjaxInput(server_ip)

    if service in ('haproxy', 'nginx', 'keepalived', 'apache', 'cluster'):
        service_desc = sql.select_service(service)
        if not roxywi_auth.is_access_permit_to_service(service_desc.slug):
            abort(403, f'You do not have needed permissions to access to {service_desc.slug.title()} service')
        if service == 'cluster':
            server_id = server_ip
        else:
            server_id = sql.select_server_id_by_ip(server_ip)
        history = sql.select_action_history_by_server_id_and_service(server_id, service_desc.service)
    elif service == 'server':
        if roxywi_common.check_is_server_in_group(server_ip):
            server_id = sql.select_server_id_by_ip(server_ip)
            history = sql.select_action_history_by_server_id(server_id)
    elif service == 'user':
        history = sql.select_action_history_by_user_id(server_ip)
    else:
        abort(404, f'History not found')

    kwargs = {
        'user_subscription': roxywi_common.return_user_subscription(),
        'users': sql.select_users(),
        'serv': server_ip,
        'service': service,
        'history': history
    }

    return render_template('history.html', **kwargs)


@bp.route('/servers')
@login_required
@get_user_params()
def servers():
    roxywi_auth.page_for_admin(level=2)

    user_group = roxywi_common.get_user_group(id=1)
    kwargs = {
        'h2': 1,
        'users': sql.select_users(group=user_group),
        'groups': sql.select_groups(),
        'servers': roxywi_common.get_dick_permit(virt=1, disable=0, only_group=1),
        'roles': sql.select_roles(),
        'sshs': sql.select_ssh(group=user_group),
        'masters': sql.select_servers(get_master_servers=1, uuid=g.user_params['user_uuid']),
        'group': roxywi_common.get_user_group(id=1),
        'services': sql.select_services(),
        'timezones': pytz.all_timezones,
        'guide_me': 1,
        'settings': sql.get_setting('', all=1),
        'backups': sql.select_backups(),
        's3_backups': sql.select_s3_backups(),
        'page': 'servers.py',
        'ldap_enable': sql.get_setting('ldap_enable'),
        'gits': sql.select_gits(),
        'is_needed_tool': common.is_tool('ansible'),
        'user_roles': sql.select_user_roles_by_group(user_group),
        'user_subscription': roxywi_common.return_user_subscription(),
        'lang': g.user_params['lang']
    }

    return render_template('servers.html', **kwargs)


@bp.route('/internal/show_version')
@cache.cached()
def show_roxywi_version():
    return render_template('ajax/check_version.html', versions=roxy.versions())