Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2026-03-21 15:47
📄 __init__.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 from wmt.common.config import ConfigManager cfg = ConfigManager()	270 B	2025-10-20 12:20
📁 __pycache__	-	2026-03-21 15:47
📄 config.py #!/opt/cloudlinux/venv/bin/python3 import logging import os import fnmatch import json from dataclasses import dataclass, asdict, field from wmt.common.const import CONFIG_PATH, PING_CONNECTIONS from wmt.common.exceptions import WmtConfigException from wmt.common.url_parser import parse from clcommon.cpapi import get_admin_email from socket import gethostname from typing import List @dataclass class Cfg: """ Default values, in case config has not been specified yet """ ping_interval: int = 30 ping_timeout: int = 10 ping_connections: int = PING_CONNECTIONS report_email: str = None report_top: int = 4 ignore_list: List[str] = field(default_factory=list) summary_notification_enabled: bool = True alert_notifications_enabled: bool = False class ConfigManager: def __init__(self): self.allowed_params = Cfg.__dataclass_fields__.keys() self.from_email = f'web-monitoring-tool@{gethostname()}' self.default_report_email = get_admin_email() self.cfg = self._init_cfg() # self.ignored_domains is used as cache for checking domains if it is ignored or not self._ignored_domains = self.generate_ignored_domains() self.target_email = self._get_target_email() def _get_target_email(self): """ This function checks to see which email address to use for TO: field of smtp. If report_email has been defined by user then report_email will be used. By default (in case not defined by user) default_report_email will be used """ return self.cfg.report_email if self.cfg.report_email else self.default_report_email def to_dict(self): return asdict(self.cfg) def _init_cfg(self) -> Cfg: # if not present - use defaults if not self.is_present(): return Cfg() data = self.read() cfg = Cfg() for key, value in data.items(): if key not in self.allowed_params: logging.warning( f'unsupported parameter "{key}", please ensure config contains' f' only allowed parameters: {list(self.allowed_params)}' ) else: setattr(cfg, key, value) return cfg @staticmethod def is_present(): return os.path.isfile(CONFIG_PATH) def modify(self, new_json: str): """ Changes configuration of wmt Returns: self.to_dict() Raises: WmtConfigException Example: wmt-api-solo --config-change {'key': 'val'} """ try: new_config = json.loads(new_json) except json.JSONDecodeError as e: raise WmtConfigException(str(e)) if not set(new_config.keys()).issubset(self.allowed_params): raise WmtConfigException(f'some of passed params are unsupported, ' f'only allowed parameters: {list(self.allowed_params)}') config = { self.to_dict(), new_config } if config.get('ignore_list') and isinstance(config.get('ignore_list'), str): config['ignore_list'] = config['ignore_list'].split(',') # Write config to /etc/wmt/config.json file with open(CONFIG_PATH, 'w') as f: json.dump(config, f, indent=4) self.cfg = Cfg(*config) return self.to_dict() @staticmethod def read(): try: with open(CONFIG_PATH) as f: data = json.load(f) except json.JSONDecodeError as e: raise WmtConfigException(str(e)) return data def reload(self): self.cfg = self._init_cfg() def is_domain_ignored(self, domain) -> bool: """ Check if domain is in ignored list. If record contain " then it will be processed as wildcard. Else as substring """ for pattern in self._ignored_domains: if '' in pattern: # set on the first position if it not set to allow filter without scheme (e.g. domain.com) if not pattern.startswith(''): pattern = f"{pattern}" if fnmatch.fnmatch(domain, pattern): return True else: if pattern in domain: return True return False def generate_ignored_domains(self) -> set: """ Generates ignored domains patterns from self.ignore_list and returns it for using as cache in self.ignored_domains set(). """ patterns = set() for pattern in self.cfg.ignore_list: patterns.add(pattern) return patterns	4.66 KB	2025-10-20 12:20
📄 const.py #!/opt/cloudlinux/venv/bin/python3 import os BASE_DIR = '/etc/wmt' CONFIG_PATH = os.path.join(BASE_DIR, 'config.json') WMT_DB = '/var/lve/wmt.db' LICENSE_EXPIRED_FAREWELL_LETTER_MARKER = '/var/lve/wmt/wmt_farewell_letter.done' LICENSE_CHECK_PAUSE = 300 # sec JWT_TOKEN = '/etc/sysconfig/rhn/jwt.token' WMT_LOCK_FILE = '/var/run/wmt_scanner.lock' RETENTION_TIME_DAYS = 2 UNKNOWN_RHN_ID = 'unknown' CLICKHOUSE_WMT_ENDPOINT = 'https://stat-api.cloudlinux.com/api/clos-wsmt' PING_TIMEOUT_STATUS_CODE = 408 # userland project SENTRY_DNS = 'https://9e8970e72aa44adb892b3f651e2f0860@cl.sentry.cloudlinux.com/33' WMT_SCANNER_SERVICE = 'cl_wmt_scanner' WMT_SCANNER_SERVICE_SOLO = 'cl_wmt_scanner_solo' WMT_SCANNER_SERVICE_SOLO_FILE = f'{WMT_SCANNER_SERVICE_SOLO}.service' WMT_TEMPLATES_DIR = '/usr/share/web-monitoring-tool/templates' ERROR_DOMAINS_PING_RETRY_INTERVAL = 300 # 5min ERROR_DOMAINS_ALERT_INTERVAL = 6 # hours PING_CONNECTIONS = 10 # concurrent requests SERVICE_BIN: str = '/sbin/service' CHKCONFIG_BIN: str = '/sbin/chkconfig' SYSTEMCTL_BIN: str = '/usr/bin/systemctl'	1.06 KB	2025-10-20 12:20
📄 exceptions.py #!/opt/cloudlinux/venv/bin/python3 from clcommon import FormattedException class WmtServiceControlException(FormattedException): def __init__(self, service_name, reason): super().__init__({ 'message': 'Error while handling service %(service_name)s: %(reason)s', 'context': { 'service_name': service_name, 'reason': reason } }) class WmtConfigException(FormattedException): def __init__(self, reason): super().__init__({ 'message': 'Error while handling config file: %(reason)s', 'context': { 'reason': reason } }) class WmtEmailException(FormattedException): def __init__(self, reason): super().__init__({ 'message': 'Error managing email notification: %(reason)s', 'context': { 'reason': reason } })	942 B	2025-10-20 12:20
📄 notification.py #!/opt/cloudlinux/venv/bin/python3 import html import os import base64 import time from dataclasses import dataclass from datetime import timedelta, datetime from email.header import Header from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from prettytable import PrettyTable, ALL from wmt.common.const import WMT_TEMPLATES_DIR, LICENSE_EXPIRED_FAREWELL_LETTER_MARKER, JWT_TOKEN from wmt.common.exceptions import WmtEmailException from clcommon import clemail from clcommon.mail_helper import MailHelper from clcommon.lib.network import get_hostname, get_ip_addr HEADERS_REPORT_MAPPING = { 'summary_report': ['All', 'Successful', 'Failed', 'Undone', 'Average time (ms)'], 'error_report': ['URL', 'Errors', 'Error codes'], 'duration_report': ['URL', 'Average time (ms)'] } class SupportedNotificationTypes: ALERT = 'alert' REPORT = 'report' FAREWELL = 'farewell' @dataclass class Email: subject: str text_sample: str html_sample: str @dataclass class EmailReport: text_option: str html_option: str @dataclass class EmailTemplate: target_mail: str from_mail: str summary_report: EmailReport = None error_report: EmailReport = None duration_report: EmailReport = None target_name: str = 'Administrator' locale: str = 'en_US' @staticmethod def _get_hostname(): return get_hostname() or 'UNKNOWN' @staticmethod def get_ip_addr(hostname): if not hostname: return 'UNKNOWN' return get_ip_addr(hostname) or 'UNKNOWN' @staticmethod def _get_logo(): logo_path = os.path.join(WMT_TEMPLATES_DIR, 'logo.png') with open(logo_path, 'rb') as logo: logo_img_encoded = base64.b64encode(logo.read()).decode('utf-8') return logo_img_encoded def to_text_template(self, date, notify_type): """ Convert to txt template keys see wmt_notify.txt """ hostname = self._get_hostname() template = { 'TONAME': self.target_name, 'DATE': date, 'HOSTNAME': hostname, 'IP_ADDR': self.get_ip_addr(hostname) } if notify_type in [SupportedNotificationTypes.REPORT, SupportedNotificationTypes.ALERT]: template.update({ 'ERROR_REPORT': self.error_report.text_option }) if notify_type == SupportedNotificationTypes.REPORT: template.update({ 'SUMMARY_REPORT': self.summary_report.text_option, 'DURATION_REPORT': self.duration_report.text_option, }) return template def to_html_template(self, date, notify_type): """ Convert to html template keys see wmt_notify.html """ hostname = self._get_hostname() template = { 'TONAME': self.target_name, 'DATE': date, 'LOGO': self._get_logo(), 'HOSTNAME': hostname, 'IP_ADDR': self.get_ip_addr(hostname) } if notify_type in [SupportedNotificationTypes.REPORT, SupportedNotificationTypes.ALERT]: template.update({ 'ERROR_HTML_REPORT': self.error_report.html_option }) if notify_type == SupportedNotificationTypes.REPORT: template.update({ 'SUMMARY_HTML_REPORT': self.summary_report.html_option, 'DURATION_HTML_REPORT': self.duration_report.html_option }) return template @dataclass class Notifier: target_email: str from_email: str report: dict notification_type: str @property def period(self) -> str: if self.notification_type == SupportedNotificationTypes.REPORT: return datetime.strftime(datetime.now() - timedelta(days=1), '%Y-%m-%d') elif self.notification_type == SupportedNotificationTypes.ALERT: return datetime.strftime(datetime.now(), '%Y-%m-%d %H:%M:%S') elif self.notification_type == SupportedNotificationTypes.FAREWELL: if not os.path.exists(JWT_TOKEN): return '' seconds = time.time() - os.path.getmtime(JWT_TOKEN) days = int(seconds // (3600 * 24)) if days > 0: return f'{str(days)} day(s) ago' else: hours = int(seconds // 3600) return f'{str(hours)} hour(s) ago' else: raise NotImplementedError(f'unexpected notifier type: {self.notification_type}') @staticmethod def create_farewell_letter_marker(): open(LICENSE_EXPIRED_FAREWELL_LETTER_MARKER, 'w').close() def notify(self): mail_manager = MailHelper() template = self._generate_template_data() message = self._generate_final_message(template) mail_manager.sendmail(template.from_mail, [template.target_mail], message) if self.notification_type == SupportedNotificationTypes.FAREWELL: self.create_farewell_letter_marker() def _get_templates(self): if self.notification_type == SupportedNotificationTypes.REPORT: return os.path.join(WMT_TEMPLATES_DIR, 'wmt_notify.txt'), \ os.path.join(WMT_TEMPLATES_DIR, 'wmt_notify.html') elif self.notification_type == SupportedNotificationTypes.ALERT: return os.path.join(WMT_TEMPLATES_DIR, 'wmt_alert.txt'), \ os.path.join(WMT_TEMPLATES_DIR, 'wmt_alert.html') elif self.notification_type == SupportedNotificationTypes.FAREWELL: return os.path.join(WMT_TEMPLATES_DIR, 'wmt_farewell_solo.txt'), \ os.path.join(WMT_TEMPLATES_DIR, 'wmt_farewell_solo.html') else: raise NotImplementedError(f'unexpected notifier type: {self.notification_type}') @staticmethod def _get_table_headers(section): return HEADERS_REPORT_MAPPING[section] def _generate_tables(self, section) -> EmailReport: no_domains_message = f'No domains for {self.period}' text_table = self._get_table('text', section) html_table = self._get_table('html', section) text_final_str = text_table.get_string() if text_table else no_domains_message html_final_str = html_table.get_html_string(format=True, border=True, hrules=ALL, vrules=ALL) if html_table else no_domains_message return EmailReport( text_option=text_final_str, html_option=html.unescape(html_final_str.replace('text-align: center', 'text-align: left')) ) def _get_table(self, alternative, section): table_data = self.report[section] if not table_data: return None table = PrettyTable(self._get_table_headers(section)) table.align = 'l' if not isinstance(table_data, list): table_data = [table_data] for row in table_data: table.add_row(row.to_template(alternative)) return table def _generate_template_data(self) -> EmailTemplate: template = { 'target_mail': self.target_email, 'from_mail': self.from_email } if self.notification_type in [SupportedNotificationTypes.REPORT, SupportedNotificationTypes.ALERT]: template.update({ 'error_report': self._generate_tables('error_report'), }) if self.notification_type == SupportedNotificationTypes.REPORT: template.update({ 'summary_report': self._generate_tables('summary_report'), 'duration_report': self._generate_tables('duration_report') }) return EmailTemplate(**template) def _generate_email(self, report: EmailTemplate): text_template, html_template = self._get_templates() if not os.path.exists(text_template) or not os.path.exists(html_template): raise WmtEmailException(f'unable to find email templates: {text_template}, {html_template}. ' 'Try to reinstall package') try: period = self.period subject = f'Web Monitoring Tool {self.notification_type} for {period} from server {get_hostname()}' if self.notification_type == SupportedNotificationTypes.ALERT: subject = f'[{self.notification_type.upper()}] {subject}' if self.notification_type == SupportedNotificationTypes.FAREWELL: subject = f'[EXPIRED CloudLinux LICENSE] Web Monitoring Tool {self.notification_type}' subject, text_body = clemail.ClEmail.generate_mail_jinja2( text_template, templ_data=report.to_text_template(period, self.notification_type), subject=subject) _, html_body = clemail.ClEmail.generate_mail_jinja2( html_template, templ_data=report.to_html_template(period, self.notification_type), subject=subject) except clemail.jinja2.exceptions.TemplateError as e: raise WmtEmailException(f'cannot generate email, reason: {e}') return Email(subject=subject, text_sample=text_body, html_sample=html_body) def _generate_final_message(self, template_data: EmailTemplate): email = self._generate_email(template_data) text_body = email.text_sample.encode('utf-8', 'xmlcharrefreplace').decode('utf-8') html_body = email.html_sample.encode('utf-8', 'xmlcharrefreplace').decode('utf-8') message = MIMEMultipart('alternative') message.attach(MIMEText(text_body, 'plain', 'utf-8')) message.attach(MIMEText(html_body, 'html', 'utf-8')) message['Subject'] = Header(email.subject, 'utf-8').encode() message['From'] = template_data.from_mail message['To'] = template_data.target_mail return message	9.96 KB	2025-10-20 12:20
📄 report.py #!/opt/cloudlinux/venv/bin/python3 import re from urllib.parse import urlparse from dataclasses import dataclass, asdict from sqlalchemy import func from sqlalchemy.sql.expression import literal_column from wmt.common import cfg from wmt.common.utils import get_domains from wmt.db import ScrapeResult, session_scope @dataclass class SummaryReport: count_all: int count_successful: int count_failed: int count_undone: int average_time: float def to_template(self, args): return [self.count_all, self.count_successful, self.count_failed, self.count_undone, # emails need time in ms int(self.average_time / 103)] @dataclass class ErrorReport: code: str count_errors: int url: str def to_template(self, alternative): url = url_to_domain(self.url) if alternative == 'html': url = f'<a href="{self.url}">{url}</a>' return [url, self.count_errors, self.code] @dataclass class DurationReport: url: str average_time: float def to_template(self, alternative): url = url_to_domain(self.url) if alternative == 'html': url = f'<a href="{self.url}">{url}</a>' return [url, # emails need time in ms int(self.average_time / 103)] def url_to_domain(url): pattern = r'http(s)?://' return re.sub(pattern, '', url) def generate_report(engine, start_date, end_date): with session_scope(engine) as session: # gets counter per status code per website -> group key: website: status_code pair # e.g (test.com 404 3), (test.com 500 2) subquery = session.query(ScrapeResult.response_code, ScrapeResult.website, func.count().label('err_count')).\ filter(ScrapeResult.create_date >= start_date, ScrapeResult.create_date <= end_date, ScrapeResult.response_code != 200, ScrapeResult.is_finished == True)\ .group_by(ScrapeResult.response_code, ScrapeResult.website)\ .subquery() # group previous subquery by website # code count website # [('451,500', 3, 'http://www.flightradar24.com'), # ('404', 2, 'http://broken.com')] error_stats = session.query(func.group_concat(subquery.c.response_code), func.sum(subquery.c.err_count), subquery.c.website)\ .group_by(subquery.c.website)\ .all() # website avg ms count # [('http://www.stackoverflow.com', 538.0816599732262, 2241), # ('http://www.suser.com', 66.53859883980365, 2241)] success_stats = session.query(ScrapeResult.website, func.avg(ScrapeResult.response_time_ms).label('average_time'), func.count())\ .filter(ScrapeResult.create_date >= start_date, ScrapeResult.create_date <= end_date, ScrapeResult.response_code == 200)\ .group_by(ScrapeResult.website)\ .order_by(literal_column('average_time').desc()) \ .all() count_unsuccessful = session.query(ScrapeResult)\ .filter(ScrapeResult.create_date >= start_date, ScrapeResult.create_date <= end_date, ScrapeResult.is_finished == False)\ .count() # filter out not present domains using host-only comparison and tolerating www/non-www def extract_hostname_only(url_value: str): parsed = urlparse(url_value, 'http') hostname = parsed.netloc or parsed.path return hostname.lower().strip('/') allowed_hosts = set() for domain in get_domains(): hostname = extract_hostname_only(domain) allowed_hosts.add(hostname) # include both www and non-www forms for matching if hostname.startswith('www.'): allowed_hosts.add(hostname[4:]) else: allowed_hosts.add('www.' + hostname) success_stats = [ (url, average_time, count) for url, average_time, count in success_stats if not cfg.is_domain_ignored(url) and extract_hostname_only(url) in allowed_hosts ] error_stats = [ (code, count, url) for code, count, url in error_stats if not cfg.is_domain_ignored(url) and extract_hostname_only(url) in allowed_hosts ] error_report = [ErrorReport(code=code, count_errors=count_errors, url=url) for code, count_errors, url in error_stats] # Collapse by host: prefer https entry if it exists for the host; otherwise keep the first # implemented because in report we do not want to show both http and https entries for the same host host_choice = {} for url, average_time, _ in success_stats: host = extract_hostname_only(url) is_https = str(url).lower().startswith('https://') chosen = host_choice.get(host) if chosen is None: host_choice[host] = (url, average_time, is_https) else: # upgrade to https if available if not chosen[2] and is_https: host_choice[host] = (url, average_time, True) duration_report = [DurationReport(url=item[0], average_time=int(round(item[1] 1000))) for item in host_choice.values()] successful_requests_count = sum(success_count for url, _, success_count in success_stats) error_requests_count = sum(errors_count for _, errors_count, url in error_stats) averages = [item[1] for item in success_stats] average_count = 0 if not averages else int(round(1000 * sum(averages) / len(averages))) summary_report = SummaryReport(count_all=successful_requests_count + error_requests_count + count_unsuccessful, count_successful=successful_requests_count, count_failed=error_requests_count, count_undone=count_unsuccessful, average_time=average_count) return { 'summary_report': summary_report, 'error_report': error_report, 'duration_report': duration_report } def report_dict(report): return { 'summary_report': asdict(report['summary_report']), 'error_report': [asdict(item) for item in report['error_report']], 'duration_report': [asdict(item) for item in report['duration_report']] }	6.61 KB	2025-10-20 12:20
📄 service.py #!/opt/cloudlinux/venv/bin/python3 -bb # coding=utf-8 # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2020 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENCE.TXT # import platform from wmt.common.exceptions import WmtServiceControlException from wmt.common.utils import disable_wmt_daemon, enable_wmt_daemon def set_service_state(service_name, action): _is_cl6 = 'el6' in platform.release() if not _is_cl6: service_name = service_name + '.service' try: if action == 'start': enable_wmt_daemon(service_name, _is_cl6) elif action == 'stop': disable_wmt_daemon(service_name, _is_cl6) except Exception as e: raise WmtServiceControlException(service_name, str(e))	807 B	2025-10-20 12:20
📄 url_parser.py from urllib.parse import urlparse, ParseResult def parse(domain: str, scheme: str = 'http') -> str: """ Convert domain name to 'http://www.domain.com' format """ # https://stackoverflow.com/questions/21659044/how-can-i-prepend-http-to-a-url-if-it-doesnt-begin-with-http p = urlparse(domain, 'http') netloc = p.netloc or p.path path = p.path if p.netloc else '' p = ParseResult(p.scheme, netloc, path, *p[3:]) return p.geturl()	465 B	2025-10-20 12:20
📄 utils.py #!/opt/cloudlinux/venv/bin/python3 -bb # coding=utf-8 # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2020 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENCE.TXT # import contextlib import logging import fcntl import os import time import requests import subprocess from clcommon import cpapi from clcommon.utils import get_rhn_systemid_value from wmt.common.url_parser import parse from wmt.common.const import ( CLICKHOUSE_WMT_ENDPOINT, UNKNOWN_RHN_ID, SERVICE_BIN, CHKCONFIG_BIN, SYSTEMCTL_BIN ) # logic copied from get_domains script, that is used by go def get_domains(): domains = set() if cpapi.CP_NAME == cpapi.PLESK_NAME: users = [_cpinfo[0] for _cpinfo in cpapi.cpinfo(keyls=('cplogin',))] else: users = cpapi.cpusers() if not users: return domains try: suspended_users = cpapi.suspended_users_list() except Exception: logging.exception('Cannot obtain list of suspended users') suspended_users = [] # dirty hack to reload domains in runtime # TODO: replace with more suitable solution if cpapi.CP_NAME == 'cPanel': cpapi.plugins.cpanel._user_to_domains_map_cpanel = dict() for user in users: if user in suspended_users: logging.warning('User: %s is will not be pinged, because account is suspended', user) continue for domain, _ in cpapi.userdomains(user): domains.add(parse(domain)) return domains def setup_logger(logger_name): app_logger = logging.getLogger(logger_name) app_logger.setLevel(logging.DEBUG) try: fh = logging.FileHandler('/var/log/cl_wmt.log') except IOError: pass else: fh.formatter = logging.Formatter('[%(levelname)s \| %(asctime)s]: %(message)s') app_logger.addHandler(fh) return app_logger @contextlib.contextmanager def save_pid_and_lock(file: str, pid: str): mode = 'w+' if not os.path.exists(file) else 'r+' f = open(file, mode) try: fcntl.flock(f.fileno(), fcntl.LOCK_EX \| fcntl.LOCK_NB) except OSError: raise OSError('Process %s already running!' % pid) try: f.write(pid) f.flush() os.fsync(f) yield finally: fcntl.flock(f.fileno(), fcntl.LOCK_UN) f.close() os.remove(file) def intersect(d1, d2): result = {} for key in d1: if key in d2: result[key] = d1[key], d2[key] return result def send_report(url, report): error = requests.RequestException('Error while sending report to ClickHouse') for i in range(5): try: response = requests.post(url, json=report, timeout=60) response.raise_for_status() except requests.RequestException as err: error = err time.sleep(4 i) else: break else: raise error def send_report_to_clickhouse(report): systemd_id = get_rhn_systemid_value('system_id') or UNKNOWN_RHN_ID # ID-XXXXXXXX -> XXXXXXXX report['server_id'] = systemd_id.replace('ID-', '') summary = report.pop('summary_report') send_report(CLICKHOUSE_WMT_ENDPOINT, {report, **summary}) def enable_wmt_daemon(daemon_name, is_cl6): """ Enable cl_wmt_scanner service """ if is_cl6: # CL6 # Start wmt service subprocess.run('{} {} start >/dev/null 2>&1'.format(SERVICE_BIN, daemon_name), shell=True, executable='/bin/bash', capture_output=True) subprocess.run('{} --add {} >/dev/null 2>&1'.format(CHKCONFIG_BIN, daemon_name), shell=True, executable='/bin/bash', capture_output=True) else: # CL7, CL8 subprocess.run('{} daemon-reload'.format(SYSTEMCTL_BIN), shell=True, executable='/bin/bash', capture_output=True) # Start wmt daemon subprocess.run('{} enable {}'.format(SYSTEMCTL_BIN, daemon_name), shell=True, executable='/bin/bash', capture_output=True) subprocess.run('{} start {}'.format(SYSTEMCTL_BIN, daemon_name), shell=True, executable='/bin/bash', capture_output=True) def disable_wmt_daemon(daemon_name, is_cl6): """ Disable WMT daemon :return: """ if is_cl6: # CL6 subprocess.run('{} {} stop >/dev/null 2>&1'.format(SERVICE_BIN, daemon_name), shell=True, executable='/bin/bash', capture_output=True) subprocess.run('{} --del {} >/dev/null 2>&1'.format(CHKCONFIG_BIN, daemon_name), shell=True, executable='/bin/bash', capture_output=True) else: # CL7, CL8 subprocess.run('{} kill {} >/dev/null 2>&1'.format(SYSTEMCTL_BIN, daemon_name), shell=True, executable='/bin/bash', capture_output=True) subprocess.run('{} disable {} >/dev/null 2>&1'.format(SYSTEMCTL_BIN, daemon_name), shell=True, executable='/bin/bash', capture_output=True) def manage_crons(status=True): cron_tool = '/usr/share/web-monitoring-tool/cron_control.py' if status: command = [cron_tool, '-i'] else: command = [cron_tool, '-d'] subprocess.run(command, capture_output=True, text=True)	5.24 KB	2025-10-20 12:20

Terminal