Yu1709$!%yuyu!$%wutu!5=126+

                
____  ___       .__                 
\   \/  /  _____|  |   ______  _  __
 \     /  /  ___/  |  /  _ \ \/ \/ /
 /     \  \___ \|  |_(  <_> )     / 
/___/\  \/____  >____/\____/ \/\_/  
      \_/     \/    @ligmakokz

Editing: server.py

#!/opt/alt/python37/bin/python3 -bb
# coding=utf-8
#
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT
#
import base64
import json
import os
import uuid
from encodings.base64_codec import base64_encode
from pwd import getpwnam

import aiohttp_jinja2
import jinja2
import pam
from aiohttp import web
from aiohttp.web_request import Request
from aiohttp.web_response import Response
from aiohttp_session import setup as setup_session
from aiohttp_security import is_anonymous, remember, forget, \
    setup as setup_security, SessionIdentityPolicy, authorized_userid, check_permission
from aiohttp_security.abc import AbstractAuthorizationPolicy
from aiohttp_session.cookie_storage import EncryptedCookieStorage
from cryptography import fernet

from constants import *
from utils import parse_params, run_cmd_pw, get_user_data, get_service_port,\
    get_ssl_context, get_user_type, get_user_plugins

# ToDo(dpoleev): Use PKG_VERSION from lvemanager package after migrate to python 3.7
PKG_VERSION = open('/usr/share/l.v.e-manager/version').read().strip()

SERVER_PATH = os.path.dirname(os.path.realpath(__file__))
STATIC_FILE_PATH = os.path.join(SERVER_PATH, '../../commons/spa-resources/')
CLOUDLINUX_CLI = '/usr/share/l.v.e-manager/utils/cloudlinux-cli.py'
CLOUDLINUX_USER_CLI = '/usr/share/l.v.e-manager/utils/cloudlinux-cli-user.py'

class AuthorizationPolicy(AbstractAuthorizationPolicy):
    """
    get information about user by identity
    """
    async def authorized_userid(self, identity):
        pw = getpwnam(identity)
        result = {
            'pw': pw,
            'type': get_user_type(pw.pw_name)
        }
        return result

async def permits(self, identity, permission, context=None):
        return bool(identity)

async def handler_root(request):
    """
    Main page
    """
    is_logged = not await is_anonymous(request)
    if is_logged:
        user_data = await authorized_userid(request)
        if user_data['type'] == TYPE_ADMIN:
            return await admin_page(request)
        elif user_data['type'] == TYPE_RESELLER:
            return await reseller_page(request)
        else:
            return await user_page(request)
    else:
        return await login_page(request)

@aiohttp_jinja2.template('login.html')
async def login_page(request: Request):
    """
    Login page
    """
    return {
        'login_error': request.query_string == 'incorrect'
    }

@aiohttp_jinja2.template('user.html')
async def user_page(request):
    """
    List applications page
    """
    return {
        'userdata': await authorized_userid(request),
        'apps': get_user_plugins()
        }

async def admin_page(request):
    """
    Open admin LveManager
    """
    response = await show_app('main', await authorized_userid(request), request)
    set_csrf_token(response)
    return response

async def reseller_page(request):
    """
    Open Reseller LveManager
    """
    response = await show_app('main', await authorized_userid(request), request)
    set_csrf_token(response)
    return response

async def handler_login(request: Request):
    """
    Check authorization and show error message
    """
    redirect_response = web.HTTPFound('/')
    data = await request.post()
    pam_object = pam.pam()
    if pam_object.authenticate(data.get('username'), data.get('password'), service='system-auth'):
        await remember(request, redirect_response, data.get('username'))
        return redirect_response
    else:
        return web.HTTPFound('/?incorrect')

async def handler_app(request: Request):
    """
    Show app handler
    """
    plugin_name = request.match_info['plugin_name']
    userdata = await authorized_userid(request)
    response = await show_app(plugin_name, userdata, request)
    set_csrf_token(response)
    return response

@aiohttp_jinja2.template('app.html')
async def show_app(plugin_name, userdata, request: Request):
    """
    Show certain SPA application
    :param plugin_name: plugin name show what bundle and title should be used
    :param userdata: information about user
    :param request:Request for rendering jinja template
    :return:
    """
    plugin_title = PLUGINS.get(plugin_name).get('title')
    panel_data = await get_user_data(userdata)
    return {
        'plugin_title': plugin_title,
        'plugin_name': plugin_name,
        'panel_data': panel_data,
        'pluginVersion': PKG_VERSION
    }

async def handler_logout(request):
    """
    Logout handler: remove cookies information
    """
    redirect_response = web.HTTPFound('/')
    await forget(request, redirect_response)
    return redirect_response

async def handler_request(request):
    """
    Middleware for providing requests to cloudlinux-cli level
    """
    check_csrf_token(request)
    plugin_name = request.match_info['plugin_name']
    plugin_title = PLUGINS.get(plugin_name).get('title')
    await check_permission(request, plugin_name)
    user_data = await authorized_userid(request)
    request_data = parse_params(await request.post())

data = {
        'plugin_name': plugin_name,
        'owner': user_data['type'],
        'command': request_data.get('command'),
        'params': request_data.get('params') or {},
    }
    if 'method' in request_data:
        data['method'] = request_data.get('method')

if user_data['pw'].pw_uid > 0:
        data['user_info'] = {
            'username': user_data['pw'].pw_name,
            'lve-id': user_data['pw'].pw_uid,
        }

do_not_use_cagefs = user_data['type'] in [TYPE_ADMIN, TYPE_RESELLER] or \
                        (data['params'] and data['params'].get('interpreter') == 'php') or \
       data['command'] in [
        'cloudlinux-statistics',
        'cloudlinux-quota',
        'cloudlinux-top',
        'cloudlinux-snapshots',
        'cloudlinux-statsnotifier',
        'clwpos-user',
        'cloudlinux-charts'
    ]
    cli_comand = [CLOUDLINUX_USER_CLI if user_data['type'] not in [TYPE_ADMIN, TYPE_RESELLER] else CLOUDLINUX_CLI,
                  '--data={}'.format(base64_encode(json.dumps(data).encode('utf8').strip())[0].decode('utf-8'))
                  ]
    if not do_not_use_cagefs:
        cli_comand.insert(0, '/bin/cagefs_enter.proxied')
    (retcode, stdout, stderr) = await run_cmd_pw(user_data['pw'], cli_comand)
    # If decode_json is catched an exeption, send error header with backtrace
    try:
        json_data = json.loads(stdout)
    except:
        if stderr.startswith('cagefs_enter'):
            return sendError({
                'code': 503,
                'context': {'pluginName': plugin_title },
                'error_id': 'ERROR.not_available_plugin',
                'icon': 'disabled',
                'result': ''}, 1)
        return sendError('ERROR.wrong_received_data', 0, 0, stdout + stderr)
    if json_data.get('result') not in ['success', 'rollback']:
        return sendError(json_data, 1)
    if stdout == '':
        return sendError('RESPONSE OF COMMAND IS EMPTY');
    return web.Response(body=stdout + stderr, content_type='application/json')

def sendError(error_message, is_json = False, logout_signal = False, details = ''):
    if is_json:
        return web.json_response(error_message, status=503)
    else:
        response = json.dumps({
            'result': error_message,
            'logoutSignal': 1 if logout_signal else 0,
            'details': details
        })
        return web.Response(status=503, body=response, content_type='application/json')

def make_app():
    """
    Prepare web server
    """
    app = web.Application()

# add the routes
    app.add_routes([
        web.get('/', handler_root),
        web.post('/login', handler_login),
        web.get('/app/{plugin_name}', handler_app),
        web.get('/logout', handler_logout),
        web.post('/send-request/{plugin_name}', handler_request),
        web.static('/assets', STATIC_FILE_PATH)
    ])

# set up policies
    policy = SessionIdentityPolicy()
    setup_security(app, policy, AuthorizationPolicy())
    # we need to initialize aiohttp_session
    fernet_key = fernet.Fernet.generate_key()
    secret_key = base64.urlsafe_b64decode(fernet_key)
    setup_session(app, EncryptedCookieStorage(secret_key))

return app

def set_csrf_token(response: Response):
    """
    Generate random csrf token and set to cookie
    """
    response.set_cookie('csrftoken', str(uuid.uuid4()))

def check_csrf_token(request: Request):
    """
    Check csrf token
    """
    if not request.cookies.get('csrftoken') or request.cookies.get('csrftoken') != request.headers.get('X-CSRFToken'):
        raise web.HTTPForbidden(body='BAD FORGERY PROTECTION TOKEN')

if __name__ == '__main__':
    app = make_app()
    env = aiohttp_jinja2.setup(
        app,
        loader=jinja2.FileSystemLoader(
            os.path.join(SERVER_PATH, 'templates')),
        context_processors=[],
        autoescape=True,
    )
    web.run_app(
        app,
        ssl_context=get_ssl_context(),
        port=get_service_port()
    )

Name	Type	Actions
templates	Directory
constants.py	File
requirements.txt	File
server.py	File
utils.py	File