# -*- 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 stat
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, is_ubuntu
VENDOR_UI_CONFIG_PATH = '/opt/cpvendor/config/cl-manager-ui-settings.json'
# Drop-in / config files we read here live under per-user, tenant-writable
# paths (docroot wp-content, ~/.clwpos) but are consumed by root. Cap the read
# so a planted huge or /dev/zero target cannot grow RSS unbounded.
_MAX_DROP_IN_BYTES = 1024 * 1024
def _safe_read_bytes(path: Union[str, Path]) -> Optional[bytes]:
"""
Read at most _MAX_DROP_IN_BYTES from an untrusted per-user path, refusing
symlinks (O_NOFOLLOW) and non-regular files (FIFO/device fstat guard).
O_NONBLOCK keeps the open of a planted FIFO/device from blocking.
Returns the bytes read, or None if the path is a symlink, not a regular
file, or cannot be opened/read.
"""
try:
fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK | os.O_CLOEXEC)
except OSError:
return None
try:
if not stat.S_ISREG(os.fstat(fd).st_mode):
return None
return os.read(fd, _MAX_DROP_IN_BYTES)
except OSError:
return None
finally:
os.close(fd)
def _safe_read_text(path: Union[str, Path]) -> Optional[str]:
"""
Same guarantees as _safe_read_bytes, decoded as utf-8 (errors ignored to
match the callers' historic open(..., errors='ignore')). Returns None on
refusal so callers fall through to their not-found path.
"""
data = _safe_read_bytes(path)
if data is None:
return None
return data.decode('utf-8', errors='ignore')
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
if Path(abs_path).name.startswith('.wp-toolkit'):
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
content = _safe_read_text(path)
if content is None:
return False
is_multiline_comment = False
for line in content.splitlines():
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
"""
content = _safe_read_text(drop_in_path)
if content is not None and '/clsop' in content:
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:
"""
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