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-27 02:38
📄 __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 PKG_VERSION = "@VERSION@"	231 B	2025-10-20 12:20
📁 __pycache__	-	2026-03-21 15:47
📁 common	-	2026-03-21 15:47
📄 db.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 os import contextlib import sqlite3 from datetime import datetime, timedelta from sqlalchemy import ( Column, Boolean, DateTime, Integer, String, create_engine, event ) from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.engine.reflection import Inspector from sqlalchemy.orm import Session from wmt.common.const import WMT_DB, RETENTION_TIME_DAYS Base = declarative_base() class ScrapeResult(Base): """ created_date is saved in local TZ format """ __tablename__ = 'scrape_result' id = Column(Integer, primary_key=True) website = Column(String) create_date = Column(DateTime, default=lambda: datetime.now(), index=True) is_finished = Column(Boolean) response_code = Column(Integer, nullable=True) response_time_ms = Column(Integer, nullable=True) class DomainAlerts(Base): """ alert_time is saved in local TZ format """ __tablename__ = 'domain_alerts' id = Column(Integer, primary_key=True) website = Column(String, index=True) alert_time = Column(DateTime, default=lambda: datetime.now(), index=True) is_resolved = Column(Boolean, default=False) def cleanup_old_data(engine): n_days_ago = datetime.today() - timedelta(days=RETENTION_TIME_DAYS) with session_scope(engine) as session: session.query(ScrapeResult)\ .filter(ScrapeResult.create_date < n_days_ago)\ .delete() session.query(DomainAlerts)\ .filter(DomainAlerts.alert_time < n_days_ago)\ .delete() def create_db_if_not_exist(engine): if not is_db_present(engine): Base.metadata.create_all(engine) def is_db_present(engine): if not os.path.isfile(WMT_DB): return False database_inspection = Inspector.from_engine(engine) tables = [table for table in database_inspection.get_table_names()] return len(tables) > 0 def setup_wal_mode(dbapi_con, con_record): dbapi_con.execute('PRAGMA journal_mode = WAL') def setup_database(readonly=False): connection_string = f'file:{WMT_DB}' if readonly: connection_string = f'{connection_string}?mode=ro' creator = lambda: sqlite3.connect(connection_string, uri=True) engine = create_engine( 'sqlite:////', creator=creator, echo=False, ) event.listen(engine, 'connect', setup_wal_mode) create_db_if_not_exist(engine) return engine @contextlib.contextmanager def session_scope(engine) -> Session: """Provide a transactional scope around a series of operations.""" session = Session(bind=engine) try: yield session session.commit() except: session.rollback() raise finally: session.close()	2.92 KB	2025-10-20 12:20
📄 main.py #!/opt/cloudlinux/venv/bin/python3 -bb import asyncio import os import sys import aiohttp import signal import time import concurrent.futures from datetime import datetime, timedelta from typing import List, Optional from dataclasses import dataclass from sqlalchemy import or_ from wmt.db import ( ScrapeResult, DomainAlerts, setup_database, session_scope, cleanup_old_data ) from wmt.common.utils import get_domains, setup_logger, save_pid_and_lock, intersect from clsentry import init_sentry_client from clsentry.utils import get_pkg_version from wmt.common.const import ( PING_TIMEOUT_STATUS_CODE, SENTRY_DNS, ERROR_DOMAINS_PING_RETRY_INTERVAL, ERROR_DOMAINS_ALERT_INTERVAL, WMT_LOCK_FILE, PING_CONNECTIONS, LICENSE_EXPIRED_FAREWELL_LETTER_MARKER, LICENSE_CHECK_PAUSE ) from wmt.common.notification import Notifier, SupportedNotificationTypes from wmt.common.report import ErrorReport from wmt.common import cfg from cllicense import CloudlinuxLicenseLib logger = setup_logger('wmt_scanner') def reload_conf(sig_number, frame): cfg.reload() logger.info('Reloading config: %s', str(cfg.to_dict())) def shutdown(sig_number, frame): """ Shutdown to call finally block to close all fds, remove lock and file see: save_pid_and_lock() """ sys.exit(0) @dataclass class ScrapeCoroResult: url: str response_code: Optional[int] = None response_time_ms: Optional[int] = None def get_connection_limit() -> int: value = getattr(cfg.cfg, 'ping_connections', PING_CONNECTIONS) try: value = int(value) except (TypeError, ValueError): value = PING_CONNECTIONS return max(1, value) async def ping(url, ping_timeout, session, semaphore): """ Main 'pinger' 1. Requests domains - if domain responded - keep status code - if no response for timeout - keep Timeout status code - if unreachable (ConnectionError or so) - keep 523 status code (same logic as go implementation) """ async with semaphore: async def _fetch(target_url): start = time.time() try: async with session.get(target_url, timeout=ping_timeout) as resp: # Use the final URL after redirects to preserve actual scheme (http/https) final_url = str(resp.url) if hasattr(resp, 'url') else target_url return ScrapeCoroResult( final_url, response_code=resp.status, response_time_ms=int(1000 * (time.time() - start)) ) except concurrent.futures.TimeoutError: return ScrapeCoroResult(target_url, response_code=PING_TIMEOUT_STATUS_CODE, response_time_ms=ping_timeout * 1000) except aiohttp.client_exceptions.ClientError: # 523 is code for unreachable resource # same logic as in go implementation return ScrapeCoroResult(target_url, response_code=523) # Try HTTPS first, then fallback to HTTP if url.startswith('https://'): https_url = url http_url = 'http://' + url[len('https://'):] elif url.startswith('http://'): https_url = 'https://' + url[len('http://'):] http_url = url else: https_url = f'https://{url}' http_url = f'http://{url}' https_result = await _fetch(https_url) if https_result.response_code == 200: return https_result http_result = await _fetch(http_url) return http_result def executors(ping_timeout, session, semaphore, ping_target_domains=None): """ ping_timeout: specified in config timeout time (s) for request semaphore: semaphore obj to handle asyncio tasks ping_target_domains: mostly needed for re-pinging error domains """ domains = get_domains() if ping_target_domains is not None: domains &= set(ping_target_domains) logger.debug('Those domains will be pinged: %s', str(domains)) for domain in domains: if not cfg.is_domain_ignored(domain): yield ping(domain, ping_timeout, session, semaphore) async def scrape_sites(ping_site_timeout, ping_interval, session, semaphore, ping_target_domains=None) -> List[ScrapeCoroResult]: coroutines = list(executors(ping_site_timeout, session, semaphore, ping_target_domains)) if len(coroutines) == 0: return [] tasks = [asyncio.create_task(coro) for coro in coroutines] try: return await asyncio.wait_for(asyncio.gather(tasks), timeout=ping_interval) except asyncio.TimeoutError: results: List[ScrapeCoroResult] = [] for task in tasks: if not task.done(): task.cancel() for task in tasks: try: results.append(await task) except asyncio.CancelledError: continue return results def manage_ping_results(engine, pinged, ping_target_domains=None): """ - obtains all scrape coro results from asyncio tasks - saves ping results to ScrapeResult table - updates 'is_resolved' field in DomainAlerts table, in case error domain`s status code was changed to 200 - returns domains with non-200 status code code """ finished_domains = set() errors_domains = {} resolved = [] with session_scope(engine) as session: for result in pinged: session.add(ScrapeResult( website=result.url, is_finished=True, response_code=result.response_code, response_time_ms=result.response_time_ms )) finished_domains.add(result.url) if result.response_code != 200: errors_domains[result.url] = result.response_code else: resolved.append(result.url) # mark no-more failing domains as resolved session.query(DomainAlerts) \ .filter(DomainAlerts.website.in_(resolved)) \ .update(dict(is_resolved=True), synchronize_session=False) domains = get_domains() if ping_target_domains is not None: domains &= set(ping_target_domains) unfinished_domains = set(domains) - finished_domains with session_scope(engine) as session: for unfinished in unfinished_domains: session.add(ScrapeResult( website=unfinished, is_finished=False )) return errors_domains def get_recent_alerts(engine, alert_domains): """ - gets websites that must NOT be included in alert email: less than ERROR_DOMAINS_ALERT_INTERVAL passed or is_resolved marker was not changed from last alerting """ repeat_interval = datetime.now() - timedelta(hours=ERROR_DOMAINS_ALERT_INTERVAL) with session_scope(engine) as session: recently_alerted = session.query(DomainAlerts.website) \ .filter(DomainAlerts.website.in_(list(alert_domains.keys())), or_(DomainAlerts.alert_time > repeat_interval, DomainAlerts.is_resolved == False)) return [row.website for row in recently_alerted] def alert(domains_data): """ prepares needed error report object with error domains to be alerted and sends this mail returns alerted domains """ logger.info('Alerts will be sent for %s', str(list(domains_data.keys()))) error_report = [ ErrorReport( url=domain, code=', '.join(map(str, set(codes))), count_errors=len(codes) ) for domain, codes in domains_data.items() ] Notifier( target_email=cfg.target_email, from_email=cfg.from_email, report={ 'error_report': error_report }, notification_type=SupportedNotificationTypes.ALERT).notify() return domains_data def flush_alerts(engine, alert_domains): """ - gets recently alerted domains (those that must not be alerted again) and does not include them for alerting - calls alerting for left domains - updates DomainAlerts table: if website was not alerted -> adds new record if website was alerted before -> updates alert time and is_resolved marker """ recently_alerted = get_recent_alerts(engine, alert_domains) domains_to_alert = {k: v for k, v in alert_domains.items() if k not in recently_alerted} if not domains_to_alert: logger.info('All domains "%s" were alerted or still not resolved in last %d hours', str(list(alert_domains.keys())), ERROR_DOMAINS_ALERT_INTERVAL) return alert(domains_to_alert) now = datetime.now() with session_scope(engine) as session: websites = session.query(DomainAlerts) \ .with_entities(DomainAlerts.website) \ .all() urls = [row.website for row in websites] for domain in domains_to_alert: if domain in urls: session.query(DomainAlerts) \ .filter(DomainAlerts.website == domain) \ .update(dict(alert_time=now, is_resolved=False)) else: session.add(DomainAlerts(website=domain, alert_time=now)) def should_be_repinged(error_domains): if error_domains and cfg.cfg.alert_notifications_enabled: return True return False def cleanup_farewell_letter_marker(): if os.path.exists(LICENSE_EXPIRED_FAREWELL_LETTER_MARKER): logger.info('CloudLinux license was updated') os.remove(LICENSE_EXPIRED_FAREWELL_LETTER_MARKER) def manage_license_farewell(): """ Sends farewell letter once (if it was not sent before) """ try: if not os.path.exists(LICENSE_EXPIRED_FAREWELL_LETTER_MARKER): logger.warning('Going to send last email about expired license!') Notifier( target_email=cfg.target_email, from_email=cfg.from_email, report={}, notification_type=SupportedNotificationTypes.FAREWELL).notify() except Exception: logger.exception('Error while managing farewell letter') async def scrape_iteration(previously_errored, engine): """ Scanner logic: 1. Scrapes domains and obtains ping results; 2. Manage ping results (e.g: saving to DB) 3. In case error domains found -> start re-pinging Re-pinging: - in min(ping_interval, 5 mins) - flush alerts if needed 4. Sleep for ping_interval until next ping iteration """ start = time.time() # ping interval parameter stored in minutes in config ping_interval_seconds = cfg.cfg.ping_interval 60 try: cleanup_farewell_letter_marker() cleanup_old_data(engine) connections_limit = get_connection_limit() connector = aiohttp.TCPConnector(limit=connections_limit) async with aiohttp.ClientSession(connector=connector) as session: semaphore = asyncio.Semaphore(connections_limit) ping_result = await scrape_sites( cfg.cfg.ping_timeout, ping_interval_seconds, session, semaphore ) error_domains = manage_ping_results(engine, ping_result) # re-ping if should_be_repinged(error_domains): logger.info('Those domains are unsuccessful: %s \n Try to re-ping them', str(error_domains)) # let`s re-ping in ERROR_DOMAINS_PING_RETRY_INTERVAL time while True: ping_interval_seconds = cfg.cfg.ping_interval * 60 if ping_interval_seconds > ERROR_DOMAINS_PING_RETRY_INTERVAL: elapsed_for_ping = time.time() - start await asyncio.sleep( min(max(ERROR_DOMAINS_PING_RETRY_INTERVAL - elapsed_for_ping, 0), 10) ) if elapsed_for_ping > ERROR_DOMAINS_PING_RETRY_INTERVAL: # re-ping during current ping iteration ping_retry_result = await scrape_sites( cfg.cfg.ping_timeout, ping_interval_seconds, session, semaphore, ping_target_domains=error_domains.keys() ) retry_errors = manage_ping_results( engine, ping_retry_result, error_domains.keys() ) alert_domains = intersect(error_domains, retry_errors) break else: # error domains will be re-pinged together with other domains alert_domains = intersect(previously_errored, error_domains) previously_errored = error_domains break if alert_domains: logger.info('Domains with unsuccessful status code found: "%s"', str(list(alert_domains.keys()))) flush_alerts(engine, alert_domains) else: # clean up, no error domains during current iteration previously_errored = [] except Exception: logger.exception('Error during ping iteration!') finally: while True: ping_interval_seconds = cfg.cfg.ping_interval * 60 # it can be modified with reload elapsed = time.time() - start sleep_time = min(max(ping_interval_seconds - elapsed, 0), 10) # "10" to check "reload" await asyncio.sleep(sleep_time) if elapsed > ping_interval_seconds: break return previously_errored async def scrape_loop(): """ Main loop for wmt_scanner_solo service each 'while: True' iteration returns errored domains (domains that responded with non-200 status code) """ engine = setup_database() previously_errored = [] license_attempt = 0 while True: if CloudlinuxLicenseLib().get_license_status(): license_attempt = 0 previously_errored = await scrape_iteration(previously_errored, engine) else: license_attempt += 1 if license_attempt == 1: logger.warning('Seems your CloudLinux license is expired!') # let`s do several attempts to be really # sure it is not false-positives from CLN or similar if license_attempt >= 5: manage_license_farewell() license_attempt = 0 ping_interval_seconds = cfg.cfg.ping_interval * 60 sleep_time = min(ping_interval_seconds, LICENSE_CHECK_PAUSE) await asyncio.sleep(sleep_time) if __name__ == '__main__': pid = str(os.getpid()) logger.info("PID: %s", pid) with save_pid_and_lock(WMT_LOCK_FILE, pid): init_sentry_client('web-monitoring-tool', get_pkg_version('cl-web-monitoring-tool'), SENTRY_DNS) signal.signal(signal.SIGUSR1, reload_conf) signal.signal(signal.SIGTERM, shutdown) loop = asyncio.get_event_loop() loop.run_until_complete(scrape_loop())	15.17 KB	2025-10-20 12:20
📄 wmt-api.py #!/opt/cloudlinux/venv/bin/python3 import json import fcntl import os import time import sys import psutil import signal from argparse import ArgumentParser from datetime import datetime, timedelta from wmt.common.const import ( WMT_SCANNER_SERVICE, WMT_LOCK_FILE, CONFIG_PATH, WMT_DB ) from wmt.common.report import generate_report, report_dict from wmt.db import setup_database from wmt.common.service import set_service_state from wmt.common.notification import Notifier, SupportedNotificationTypes from wmt.common import cfg from cllicense import CloudlinuxLicenseLib from wmt.common.utils import send_report_to_clickhouse, manage_crons def print_result_and_exit(result='success', exit_code=0, **extra): message = { 'result': result, 'timestamp': time.time() } message.update(extra) print(json.dumps(message, indent=2, sort_keys=True)) sys.exit(exit_code) def get_status(): if not os.path.exists(WMT_LOCK_FILE): return 'stopped' with open(WMT_LOCK_FILE) as f: try: fcntl.flock(f.fileno(), fcntl.LOCK_EX \| fcntl.LOCK_NB) # locked except OSError: return 'started' return 'stopped' def get_scanner_pid(): if not os.path.exists(WMT_LOCK_FILE): return with open(WMT_LOCK_FILE) as f: pid = int(f.read().strip()) return pid if psutil.pid_exists(pid) else None def get_config(): return cfg.to_dict() def change_config(new_json_config): config = cfg.modify(new_json_config) scanner_pid = get_scanner_pid() if scanner_pid is not None: os.kill(scanner_pid, signal.SIGUSR1) return config def run(): args = ArgumentParser() args.add_argument('--config-get', action='store_true') args.add_argument('--config-change', type=str) args.add_argument('--report-get', action='store_true') args.add_argument('--send-clickhouse', action='store_true') # just for 24h report by mail executed by cron args.add_argument('--send-email', action='store_true') args.add_argument('--status', action='store_true') args.add_argument('--start', action='store_true') args.add_argument('--stop', action='store_true') opts = args.parse_args() if opts.config_get: config = get_config() config['default_report_email'] = cfg.default_report_email if config.get("ignore_list"): tmp_val = ",".join(config.get("ignore_list")) config['ignore_list'] = tmp_val print_result_and_exit(config=config) elif opts.config_change: config = change_config(opts.config_change) config['default_report_email'] = cfg.default_report_email print_result_and_exit(config=config) elif opts.report_get: engine = setup_database(readonly=os.path.exists(WMT_DB)) # 24h back from NOW start, end = datetime.now() - timedelta(days=1), datetime.now() report = report_dict(generate_report(engine, start, end)) print_result_and_exit(report=report, date=datetime.now().strftime('%Y-%m-%d %H:%M')) elif opts.send_clickhouse: engine = setup_database(readonly=True) # for the day before start, end = datetime.now().date() - timedelta(days=1), datetime.now().date() send_report_to_clickhouse(report_dict(generate_report(engine, start, end))) print_result_and_exit() elif opts.send_email: is_scanner_running = get_status() == 'started' if is_scanner_running and cfg.cfg.summary_notification_enabled: if not CloudlinuxLicenseLib().get_license_status(): print_result_and_exit('CloudLinux license is expired. ' 'You may buy new license here: https://lp.cloudlinux.com/cloudlinux-os-solo', exit_code=1) engine = setup_database() start, end = datetime.now().date() - timedelta(days=1), datetime.now().date() report = generate_report(engine, start, end) Notifier( target_email=cfg.target_email, from_email=cfg.from_email, report=report, notification_type=SupportedNotificationTypes.REPORT, ).notify() print_result_and_exit() else: print_result_and_exit('Summary report email will not be sent! ' f'Please, ensure "{WMT_SCANNER_SERVICE}" service is running and ' f'alert_notifications is enabled in "{CONFIG_PATH}"', exit_code=1) elif opts.status: status = get_status() print_result_and_exit(status=status) elif opts.start: set_service_state(WMT_SCANNER_SERVICE, 'start') manage_crons(status=True) print_result_and_exit() elif opts.stop: set_service_state(WMT_SCANNER_SERVICE, 'stop') manage_crons(status=False) print_result_and_exit() else: args.print_help() def main(): try: run() except Exception as e: print_result_and_exit(result='error', exit_code=1, context=str(e)) if __name__ == '__main__': main()	5.11 KB	2025-10-20 12:20

Terminal