1#@!#!123s

: / opt / cloudlinux / venv / lib / python3.11 / site-packages / lvestats / lib /
Filename : cloudlinux_statsnotifier.py
back
# coding=utf-8
#
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT

import pwd
import logging
import argparse
import json
import shutil
import os
import subprocess
from typing import Tuple, Dict

from clcommon.utils import run_command
from lvestats.lib import config
from lvestats.lib.config import USER_NOTIFICATIONS_OFF_MARKER
from lvestats.lib.commons.logsetup import setup_logging
from lvestats.lib.commons.func import user_should_be_notified
from lvestats.plugins.generic.statsnotifier import get_stats_notifier_parameters
from lvestats.lib.commons import dateutil


def execute(args: argparse.Namespace, log: logging.Logger) -> Tuple[str, int]:
    """
    Generic method to analyze options and call needed funcs
    """
    if not args.json:
        return 'Only JSON mode supported for now', -1

    current_user = os.getuid()
    if current_user == 0 and not args.id:
        return 'Please, specify user id via --id option', -1

    try:
        if args.notifyUser is not None:
            result = manage_user_notifications(args.id, args.notifyUser)
        else:
            result = get_notifications_settings(args.id)
    except Exception as e:
        result = {
            'result': f'error: {e}',
            'timestamp': dateutil.gm_datetime_to_unixtimestamp(),
        }
        log.error(str(e))
    json_str = json.dumps(result)
    return json_str, 0


def get_notifications_settings(user_id: int) -> Dict:
    """
    Returns notification settings
    """
    result = {'result': 'success', 'timestamp': dateutil.gm_datetime_to_unixtimestamp()}
    current_user = os.getuid()
    target_user = current_user if not current_user == 0 else user_id
    username = pwd.getpwuid(target_user).pw_name
    result['notifyUser'] = user_should_be_notified(username)
    result['notificationsAllowed'] = is_notifications_allowed(username)
    return result


def is_notifications_allowed(user: str) -> bool:
    """
    Checks if notifications allowed by admin/reseller
    """
    if os.getuid() == 0:
        return get_stats_notifier_parameters(user)
    else:
        result = subprocess.run(['/usr/share/lve-stats/lve_stats_configs_reader',
                                 'notification_allowed'],
                                text=True, capture_output=True, check=False)
        if result.returncode != 0:
            raise ValueError(f'Unable to get "notifications_enabled" parameter, reason: {result.stdout} \n '
                             f'{result.stderr}')
        result = json.loads(result.stdout.strip())
        return result.get('notification_allowed') is True


def in_cagefs():
    os.path.isdir('/var/.cagefs')


def handle_users_notifications(user_id: int, should_be_enabled: bool) -> None:
    """
    Turns on/off notifications per user by creating/removing
    notifications marker file
    """
    cagefs_enter_user = '/sbin/cagefs_enter_user'
    cagefs_enter = shutil.which('cagefs_enter')
    user_info = pwd.getpwuid(user_id)
    lvestats_dir = os.path.join(user_info.pw_dir, '.lvestats')
    marker_path = os.path.join(lvestats_dir, USER_NOTIFICATIONS_OFF_MARKER)
    if not should_be_enabled:
        if os.getuid() == 0:
            run_command([cagefs_enter_user, user_info.pw_name, 'mkdir', '-p', lvestats_dir])
            run_command([cagefs_enter_user, user_info.pw_name, 'touch', marker_path])
        else:
            run_command([cagefs_enter, 'mkdir', '-p', lvestats_dir])
            run_command([cagefs_enter, 'touch', marker_path])
    else:
        if os.path.exists(marker_path):
            if os.getuid() == 0:
                run_command([cagefs_enter_user, user_info.pw_name, 'rm', '-f', marker_path])
            else:
                run_command([cagefs_enter, 'rm', '-f', marker_path])


def manage_user_notifications(user_id, should_be_notified: bool) -> Dict:
    """
    Managing user`s notifications by checking input opts,
    validating and calling creation/removing of notifications marker file
    """
    result = {'result': 'success', 'timestamp': dateutil.gm_datetime_to_unixtimestamp()}
    current_user = os.getuid()

    target_user = current_user if not current_user == 0 else user_id
    handle_users_notifications(target_user, should_be_notified)
    return result


def main(args_):
    """
    Main entrypoint
    """
    cnf = config.read_config()
    log = setup_logging(cnf, caller_name="CloudLinuxStatsnotifier",
                        file_level=logging.WARNING,
                        console_level=logging.FATAL)

    parser = argparse.ArgumentParser(
        description='%(prog)s - CLI utility for configuring CloudLinux statistics notifier',
        prog='cloudlinux-statsnotifier'
    )
    parser.add_argument(
        '-j', '--json',
        help='return data in JSON format',
        action='store_true')
    parser.add_argument('--id', help='User id', type=int)
    parser.add_argument('--notifyUser', type=lambda x: (str(x).lower() == 'true'))
    args = parser.parse_args(args=args_)

    json_str, exit_code = execute(args, log=log)
    print(json_str)
    return exit_code