OwlCyberSecurity - MANAGER

Edit File: clwpos_collect_information.py

#!/opt/cloudlinux/venv/bin/python3 -sbb

# 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/LICENCE.TXT
#

import json
import argparse
import os
import logging

from clwpos.logsetup import setup_logging
from clwpos.feature_suites import (
    get_allowed_modules,
    get_admin_config_directory,
    get_admin_config_permissions,
)
from clwpos.utils import acquire_lock, user_uid, get_pw, is_wpos_supported
from clwpos.cl_wpos_exceptions import WposError
from clwpos.daemon import whmapi1, WposDaemon
from clcommon.cpapi import cpusers, get_installed_php_versions

_logger = setup_logging(
    caller_name='collect_information',
    file_level=logging.INFO,
    logfile_path='/var/log/clwpos/collect_information.log',
)

def php_get_vhost_versions():
    """
    @return: [
        {
            "account": "rm3",
            "documentroot": "/home/example/public_html",
            "version": "ea-php72",
            "handler": "php-fpm",
            "vhost": "otherchars.rm3.tld"
        }
    ]
    or empty list if command fails somehow
    """
    try:
        return WposDaemon._php_get_vhost_versions()
    except Exception as e:
        raise WposError(e)

if __name__ == '__main__' and is_wpos_supported():
    with acquire_lock(os.path.join('/var/run/collect_information.lock',), attempts=None):
        parser = argparse.ArgumentParser(description="Utility to collect information about user for AccelerateWP")
        parser.add_argument("user", type=str, nargs="?")
        parser.add_argument("--force", default=False, action='store_true')
        args = parser.parse_args()

installed_versions = get_installed_php_versions()
        try:
            vhost_versions = php_get_vhost_versions()
        except WposError as e:
            _logger.exception("No vhosts collected due to: %s", e)
            vhost_versions = list()

for username in cpusers():
            if args.user is not None and username != args.user:
                # per user processing, skipping all other users
                continue

pw = get_pw(username=username)
            if not get_allowed_modules(pw.pw_uid) and not args.force:
                # no features allowed for user, skipping user
                continue

filtered_vhost_versions = [version for version in vhost_versions if version["account"] == username]
            try:
                admin_config_dir = get_admin_config_directory(pw.pw_uid)
                info_json = os.path.join(admin_config_dir, "info.json")

with open(info_json, "w") as f:
                    json.dump(
                        {"installed_versions": installed_versions,
                         "vhost_versions": filtered_vhost_versions},
                        f,
                    )

owner, group, mode = get_admin_config_permissions(pw.pw_gid)
                os.chown(info_json, owner, group)
                os.chmod(info_json, mode)

except Exception as e:
                _logger.exception("Error during collecting information for %s: %s", username, e)
                continue