I Want To Break Free

Editando: lvemanager-service

#!/opt/cloudlinux/venv/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'
DISABLE_CSP_FLAG='/var/lve/disable_csp.flag'
SECURITY_HEADERS = {
    "X-Content-Type-Options": "nosniff",
    "X-Frame-Options": "DENY",
    "Content-Security-Policy": "default-src 'self' 'unsafe-inline' *.cloudlinux.com *.googleapis.com;object-src 'none';font-src *;script-src 'self' 'unsafe-eval' 'unsafe-inline' *.cloudlinux.com;img-src 'self' data:"
}

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 'mockJson' in request_data:
        data['mockJson'] = request_data.get('mockJson')
    if 'lang' in request_data:
        data['lang'] = request_data.get('lang')
    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,
        }

cli_file_path = CLOUDLINUX_USER_CLI if user_data['type'] not in [TYPE_ADMIN, TYPE_RESELLER] else CLOUDLINUX_CLI
    # Show unavailable page for end user if CLOUDLINUX_CLI missed
    if not os.path.isfile(cli_file_path):
        return sendError({
            'code': 503,
            'context': {'pluginName': plugin_title },
            'error_id': 'ERROR.not_available_plugin',
            'icon': 'disabled',
            'result': ''}, 1)
    cli_comand = [cli_file_path, '--data={}'.format(base64_encode(json.dumps(data)
                                            .encode('utf8').strip())[0].decode('utf-8'))]
    (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:
        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))

app.on_response_prepare.append(set_security_headers)
    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')

async def set_security_headers(request, response):
    """
    Add security headers
    """
    if os.path.isfile(DISABLE_CSP_FLAG):
        return
    for key, value in SECURITY_HEADERS.items():
        response.headers[key] = value

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()
    )

Cancelar