Da

✏️ Editing: clwpos_lib.py

# -*- coding: utf-8 -*-

# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT
#

import json
import re
import os
import subprocess
import logging
from pathlib import Path
from typing import Union, Iterable, Optional, Tuple, List
from enum import Enum

from clcommon.clpwd import ClPwd
from clcommon.utils import get_rhn_systemid_value

class PluginType(Enum):
    """
    Plugin types that are currently detected
    """
    OBJECT_CACHE = 'object-cache'
    ADVANCED_CACHE = 'advanced-cache'

class WpPlugins(Enum):
    """
    Static WP plugin names, that are not detected
    dynamically from drop-in files, dir names, etc
    """
    UNKNOWN = 'Unknown'
    WP_ROCKET = 'WP Rocket'
    ACCELERATE_WP = 'AccelerateWP'

def clean_comment(line: str, is_multiline_comment: bool) -> Tuple[str, bool]:
    """
    Yep, this bicycle is needed to handle different comment types in .php file
    https://www.php.net/manual/en/language.basic-syntax.comments.php
    and ensure that needed line is not under comment
    """
    if is_multiline_comment:
        if '*/' not in line:
            return '', True
        else:
            pos = line.find('*/')
            part1, _ = clean_comment(line[:pos], True)
            part2, is_multiline_comment = clean_comment(line[pos + 2:], False)
            return part1 + part2, is_multiline_comment

if '//' in line:
        pos = line.find('//')
        return line[:pos], False

if '#' in line:
        pos = line.find('#')
        return line[:pos], False

if '/*' in line:
        pos = line.find('/*')
        part1, _ = clean_comment(line[:pos], False)
        part2, is_multiline_comment = clean_comment(line[pos + 2:], True)
        return part1 + part2, is_multiline_comment

return line, False

def _is_real_file(file: str) -> bool:
    realpath_file = os.path.realpath(file)
    return os.path.isfile(realpath_file)

def _check_wp_config_php(abs_path: Union[str, Path]) -> bool:
    """
    WordPress looks for wp-config.php file in the
    (1) WordPress root and (2) one directory above the root.

Check that there is no wp-settings.php file in the second case.
    This check helps when there is a nested installation, e.g
    / is WordPress and /wp_path/ is WordPress.
    """
    try:
        wp_config_php = os.path.join(abs_path, 'wp-config.php')
        if os.path.exists(wp_config_php) and _is_real_file(wp_config_php):
            return True
    except OSError:
        pass

abs_path_level_up = os.path.join(abs_path, os.pardir)
    wp_config_php_level_up = os.path.join(abs_path_level_up, 'wp-config.php')
    wp_settings_php = os.path.join(abs_path_level_up, 'wp-settings.php')
    wp_settings_php_exists = os.path.exists(wp_settings_php) and _is_real_file(wp_settings_php)

return os.path.exists(wp_config_php_level_up) and \
           not wp_settings_php_exists and \
           _is_real_file(wp_config_php_level_up)

def _is_real_dir(dir: str) -> bool:
    realpath_dir = os.path.realpath(dir)
    return os.path.isdir(realpath_dir)

def _check_wp_includes(abs_path: Union[str, Path]) -> bool:
    """
    Check wp-includes exists and is dir.
    """
    wp_includes = os.path.join(abs_path, 'wp-includes')
    return 'wp-includes' in os.listdir(abs_path) and _is_real_dir(wp_includes)

def is_wp_path(abs_path: Union[str, Path]) -> bool:
    """
    Checks whether passed directory is a wordpress directory
    by checking presence of wp-includes folder and wp-config.php file.
    """
    try:
        if not os.path.exists(abs_path):
            return False
    # skip paths that can't be read (wrong permissions etc)
    except OSError:
        return False

try:
        return _check_wp_config_php(abs_path) and _check_wp_includes(abs_path)
    except OSError:
        pass

return False

def find_wp_paths(doc_root: str, excludes: Optional[List[str]] = None) -> Iterable[str]:
    """
    Returns folder with wordpress
    Empty string is wp is in docroot dir

:param doc_root:
        root path to start search from
    :param excludes:
        list of paths that must be excluded from search, e.g. subdomains
    """
    if not os.path.exists(doc_root):
        return

if is_wp_path(doc_root):
        yield ''

for path in Path(doc_root).iterdir():
        if not path.is_dir():
            continue

if excludes and str(path) in excludes:
            continue

if is_wp_path(path):
            yield path.name

def _is_php_define_var_found(var, path):
    """
    Looks for defined php variable with true value
    """
    r = re.compile(fr'^\s*define\s*$\s*((\'{var}\')|(\"{var}\"))\s*,\s*true\s*$\s*;')
    # let`s find needed setting by reading line by line
    with open(path, encoding='utf-8', errors='ignore') as f:
        is_multiline_comment = False
        while True:
            line = f.readline()
            if not line:
                break
            cleaned_line, is_multiline_comment = clean_comment(line, is_multiline_comment)
            if r.match(cleaned_line):
                return True
    return False

def is_advanced_cache_enabled(wordpress_path: Path):
    """
    Detects whether plugin is really enabled,
    cause not all plugins are enabled 'on load'
    # https://kevdees.com/what-are-wordpress-drop-in-plugins/
    """
    wp_config = wordpress_path.joinpath('wp-config.php')
    # really strange when main wordpress config is absent
    if not os.path.exists(wp_config):
        return False

return _is_php_define_var_found('WP_CACHE', wp_config)

def wp_rocket_plugin(drop_in_path):
    """
    They are advising to check whether WP_ROCKET_ADVANCED_CACHE is defined
    to ensure plugin is working
    https://docs.wp-rocket.me/article/134-advanced-cache-error-message
    """
    if accelerate_wp_plugin(drop_in_path) is None and \
            _is_php_define_var_found('WP_ROCKET_ADVANCED_CACHE', drop_in_path):
        return WpPlugins.WP_ROCKET.value
    return None

def accelerate_wp_plugin(drop_in_path):
    """
    Checking if the plugin folder name exists in the drop-in
    """
    with open(drop_in_path, encoding='utf-8', errors='ignore') as f:
        if '/clsop' in f.read():
            return WpPlugins.ACCELERATE_WP.value
    return None

def get_wp_cache_plugin(wordpress_path: Path, plugin_type: str):
    """
    Looking for object-cache.php or advanced-cache.php in wordpress folder
    If found - tries to find 'plugin-owner' of <-cache>.php by
    content comparison
    If cannot be found -> tries to read <-cache>.php headers looking for Plugin name: <Plugin>
    """
    wp_content_dir = wordpress_path.joinpath("wp-content")
    activated_cache = wp_content_dir.joinpath(f'{plugin_type}.php')
    if not os.path.exists(activated_cache):
        return None
    if plugin_type == PluginType.ADVANCED_CACHE.value and not is_advanced_cache_enabled(wordpress_path):
        return None
    plugins_dir = wp_content_dir.joinpath("plugins")
    plugin_name = get_wp_cache_plugin_by_scanning_dirs(activated_cache, plugins_dir) \
                  or get_wp_cache_plugin_by_header(activated_cache) \
                  or accelerate_wp_plugin(activated_cache) \
                  or wp_rocket_plugin(activated_cache) \
                  or WpPlugins.UNKNOWN.value

return plugin_name

def get_wp_cache_plugin_by_scanning_dirs(activated_plugin: Path, plugins_dir: Path) -> Optional[str]:
    """
    Scanning plugins/* dir and looking for similar <object/advanced_cache>.php
    """
    activated_plugin_bytes = activated_plugin.read_bytes()
    if not os.path.exists(plugins_dir):
        return None
    for plugin in plugins_dir.iterdir():
        for root, dirs, files in os.walk(plugin):
            if activated_plugin.name in files:
                plugin_object_cache_path = Path(root) / activated_plugin.name
                if plugin_object_cache_path.read_bytes() == activated_plugin_bytes:
                    return plugin.name
    return None

def get_wp_cache_plugin_by_header(activated_plugin: Path) -> Optional[str]:
    """
    Looking for Plugin name: <Some name> in <object/advanced.php>
    headers
    """
    if not os.path.exists(activated_plugin):
        return None

# must be enough to loop through headers
    max_top_lines_count = 30
    r = re.compile(r'^.*plugin name:\s*(?P<plugin_name>[\w ]+)\s*$', re.IGNORECASE)
    with open(activated_plugin, encoding='utf-8', errors='ignore') as f:
        for _ in range(max_top_lines_count):
            line = f.readline()
            match = r.search(line)
            if match is not None:
                return match.group('plugin_name')
    return None

def get_wp_paths_with_enabled_module(user, user_wp_paths, plugin='object_cache'):
    """
    Filter user`s wp paths with paths with enabled module
    """
    paths_with_enabled_module = []

try:
        home = ClPwd().get_homedir(user)
    except ClPwd.NoSuchUserException:
        return []

config = os.path.join(home, '.clwpos', 'clwpos_config.json')
    if not os.path.exists(config):
        return []

try:
        with open(config, encoding='utf-8', errors='ignore') as f:
            conf = f.read()
        data = json.loads(conf)
    except Exception:
        return []

modules_data = data.get('docroots', {}).get('public_html', {})
    for wp_path in user_wp_paths:
        if wp_path in modules_data and plugin in modules_data[wp_path]:
            paths_with_enabled_module.append(wp_path)
    return paths_with_enabled_module

def install_accelerate_wp():
    packages = []

if not os.path.exists('/usr/bin/cloudlinux-awp-admin'):
        packages.append('accelerate-wp')
    if not os.path.exists('/usr/sbin/cloudlinux-ssa-manager'):
        packages.append('alt-php-ssa')
    if not os.path.exists('/usr/sbin/cloudlinux-xray-agent'):
        packages.append('alt-php-xray')

if packages:
        install_command = ["yum", "install", "-y"] + packages
        proc = subprocess.run(install_command, capture_output=True, text=True, check=False)
        logging.debug('Installing AccelerateWP packages captured out: %s, err: %s', proc.stdout, proc.stderr)

if not os.path.exists('/usr/share/clos_ssa/ssa_enabled'):
        proc = subprocess.run(['/usr/sbin/cloudlinux-ssa-manager', 'enable-ssa'],
                              capture_output=True, text=True, check=False)
        logging.debug('Activation SSA captured out: %s, err: %s', proc.stdout, proc.stderr)
    proc = subprocess.run(
        ['/usr/sbin/cloudlinux-autotracing', 'enable', '--all'],
        capture_output=True,
        text=True,
        check=False,
    )
    logging.debug('Activation autotracing captured out: %s, err: %s', proc.stdout, proc.stderr)

system_id = get_rhn_systemid_value('system_id')
    if system_id:
        proc = subprocess.run(
            ['/usr/sbin/cloudlinux-xray-manager', 'enable-user-agent', '--system_id', system_id.replace('ID-', '')],
            capture_output=True,
            text=True,
            check=False,
        )
        logging.debug('Activation xray manager captured out: %s, err: %s', proc.stdout, proc.stderr)

def configure_accelerate_wp(async_set_suite: bool = False, source: str = None):
    """
    1. Installs needed packages
    2. Enables autotracing
    3. Allows AccelerateWP Free for all
    """
    install_accelerate_wp()

if os.path.exists('/var/clwpos/admin/allowed_for_all_site_optimization.flag'):
        return

command = ['/usr/bin/cloudlinux-awp-admin',
               'set-suite',
               '--allowed-for-all',
               '--suites',
               'accelerate_wp']

if source == 'BILLING_OVERRIDE':
        command.extend(['--source', 'BILLING_OVERRIDE'])

# For WHMCS command
    if async_set_suite:
        subprocess.Popen(  # pylint: disable=consider-using-with
            command,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )
    else:
        proc = subprocess.run(command, check=True, capture_output=True, text=True)
        logging.debug('Activation AccelerateWP Free captured out: %s, err: %s', proc.stdout, proc.stderr)

def configure_accelerate_wp_premium(source: str = None):
    install_accelerate_wp()
    # just in case server already allowed for all
    if os.path.exists('/var/clwpos/admin/allowed_for_all_object_cache.flag'):
        return

command = ['/usr/bin/cloudlinux-awp-admin',
               'set-suite',
               '--visible-for-all',
               '--suites',
               'accelerate_wp_premium']

# If this is a request from WHMCS, we must pass --source to correctly save the suite status
    # and not overwrite the settings of users
    if source == 'BILLING_OVERRIDE':
        command.extend(['--source', 'BILLING_OVERRIDE', '--preserve-user-settings'])

# set AccelerateWP visible for all users
    proc = subprocess.run(command, check=True, capture_output=True, text=True)
    logging.debug('Activation AccelerateWP Premium captured out: %s, err: %s', proc.stdout, proc.stderr)

def configure_accelerate_wp_cdn(source: str = None):
    install_accelerate_wp()
    # just in case server already allowed for all
    if os.path.exists('/var/clwpos/admin/allowed_for_all_cdn.flag'):
        return

# set AccelerateWP visible for all users
    command_cdn = ['/usr/bin/cloudlinux-awp-admin',
                   'set-suite',
                   '--allowed-for-all',
                   '--suites',
                   'accelerate_wp_cdn']

# If this is a request from WHMCS, we must pass --source to correctly save the suite status
    if source == 'BILLING_OVERRIDE':
        command_cdn.extend(['--source', 'BILLING_OVERRIDE'])

proc = subprocess.run(command_cdn, check=True, capture_output=True, text=True)
    logging.debug('Activation AccelerateWP CDN captured out: %s, err: %s', proc.stdout, proc.stderr)

# set AccelerateWP CDN PRO visible for all users
    command_cdn_pro = ['/usr/bin/cloudlinux-awp-admin',
                       'set-suite',
                       '--visible-for-all',
                       '--suites',
                       'accelerate_wp_cdn_pro']

# If this is a request from WHMCS, we must pass --source to correctly save the suite status
    # and not overwrite the settings of users
    if source == 'BILLING_OVERRIDE':
        command_cdn_pro.extend(['--source', 'BILLING_OVERRIDE', '--preserve-user-settings'])

proc = subprocess.run(command_cdn_pro, check=True, capture_output=True, text=True)
    logging.debug('Activation AccelerateWP CDN PRO captured out: %s, err: %s', proc.stdout, proc.stderr)

def configure_upgrade_url(upgrade_url):
    if not upgrade_url:
        return

options_json = subprocess.run(['/usr/bin/cloudlinux-awp-admin', 'get-options'],
                             check=True,
                             capture_output=True,
                             text=True).stdout
    options = json.loads(options_json)

# nothing to do, upgrade url already set to same value
    if upgrade_url == options.get('upgrade_url'):
        return

proc = subprocess.run(['/usr/bin/cloudlinux-awp-admin',
                           'set-options',
                           '--upgrade-url',
                           upgrade_url,
                           '--suite',
                           'accelerate_wp_premium'],
                          check=True,
                          capture_output=True,
                          text=True)
    logging.debug('Setting AccelerateWP Premium upgrade url captured out: %s, err: %s',
                  proc.stdout, proc.stderr)

📁 Sid Gifari File Manager