[ PARENT ]

EDIT :: cloudlinux_statsnotifier.py

# 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 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

SAVE CANCEL