1#@!#!123s

��JFIF�� 1#@!#!123s
: / opt / sharedrads / check_software_mods /
Filename : wordpress.py
back
"""WordPress module for check_software"""

from pathlib import Path
import re
from typing import Union
from packaging import version as pkg_version
import requests
import pymysql
from pymysql.cursors import Cursor
from check_software_mods.template import ModTemplate


class Module(ModTemplate):
    """WordPress module"""

    @classmethod
    @property
    def config_file(cls):
        return 'wp-config.php'

    @classmethod
    @property
    def cms_name(cls):
        return 'WordPress'

    @staticmethod
    def is_config(path: Path) -> bool:
        """if the filename is wp-config.php, we assume yes"""
        return path.name == 'wp-config.php'

    def scan_install(self, conf_path: Path):
        site_path = conf_path.parent

        # Read wp-config.php database configuration
        try:
            prefix, db_conf = self.parse_config(conf_path)
        except (OSError, KeyError) as exc:
            self.red(f"{type(exc).__name__}: {exc}")
            self.red(f'Site at {site_path} cannot be scanned')
            return
        # Read version.php site version
        try:
            site_version = self.get_version(site_path)
        except (OSError, ValueError) as exc:
            self.red(f"{type(exc).__name__}: {exc}")
            site_version = ''

        # Connect to the database and collect variables
        try:
            with pymysql.connect(host='localhost', **db_conf) as db_conn:
                with db_conn.cursor() as cur:
                    db_data = self.get_database_data(cur, prefix)
        except pymysql.Error as exc:
            self.red(f"{type(exc).__name__}: {exc}")
            self.red(f'Site at {site_path} cannot be scanned')
            return

        self.print_site_config(db_data, site_version, site_path)
        if not db_data['plugin_paths']:
            self.blue('No active plugins')
            return
        self.blue('List of active plugins')
        for plugin in db_data['plugin_paths']:
            self.show_plugin(site_path, plugin)

    def print_site_config(
        self, db_data: dict, site_version: str, site_path: Path
    ):
        """Print general information on a WordPress site"""
        pad = 12
        self.green('Name:'.ljust(pad), end='')
        self.bold(db_data.get('blogname', '?'))

        self.green('URL:'.ljust(pad), end='')
        self.bold(db_data.get('siteurl', '?'))

        self.green('Path:'.ljust(pad), end='')
        self.bold(str(site_path))

        self.green('Version:'.ljust(pad), end='')
        if not LATEST_WP or not site_version:
            self.bold(site_version or '?')
        else:
            if pkg_version.parse(LATEST_WP) > pkg_version.parse(site_version):
                self.red(f'{site_version} ({LATEST_WP} available)', end='')
                if self.args.style == 'str':
                    how = self.urlize(
                        self.kb_urls["wp_update"], 'How to update WordPress'
                    )
                    self.print(f' - {how}')
                else:
                    self.print('')
            else:
                self.green(site_version)

        self.green('Theme:'.ljust(pad), end='')
        self.bold(db_data.get('current_theme', '?'))

        self.green('Cache:'.ljust(pad), end='')
        self.check_wp_cache(site_path, db_data['plugin_paths'])

        self.green('Comments:'.ljust(pad), end='')
        if db_data.get('num_comments', 0) > 10000:
            self.red(str(db_data['num_comments']), end='')
            if self.args.style == 'str':
                how = self.urlize(
                    self.kb_urls['wp_spam'], 'How to moderate comments'
                )
                self.print(f'- {how}')
            else:
                self.print('')
        else:
            self.bold(str(db_data.get('num_comments', '?')))

        if 'multisites' in db_data:
            self.green('Multisite:'.ljust(pad), end='')
            self.red(str(db_data['multisites']))

        self.green('Plugins:'.ljust(pad), end='')
        if len(db_data['plugin_paths']) > 20:
            self.red(str(len(db_data['plugin_paths'])))
        else:
            self.bold(str(len(db_data['plugin_paths'])))

        self.green('List of Admin Users:')
        for user in db_data['admin_users']:
            self.bold(f"  {user}")

    @staticmethod
    def get_version(site_path: Path) -> str:
        """Obtain WordPress version"""
        site_version = None
        version_path = site_path / 'wp-includes/version.php'
        with open(version_path, encoding='utf-8') as version_file:
            data = version_file.read().splitlines()
        for line in data:
            if '$wp_version = ' in line:
                site_version = re.sub('[^\\w|\\.]', '', line.split(' = ')[1])
        if site_version is None:
            raise ValueError(f'version not found in {version_path}')
        return site_version

    def get_database_data(self, cur: Cursor, prefix: str) -> dict:
        """Collect WordPress database variables"""
        tbl = lambda x: f"{prefix}{x}".replace('`', '``')
        cur.execute(
            f"SELECT `option_name`, `option_value` FROM `{tbl('options')}` "
            "WHERE `option_name` IN "
            "('active_plugins', 'siteurl', 'blogname', 'current_theme')"
        )
        ret = dict(cur.fetchall())
        if active_plugins := ret.pop('active_plugins', ''):
            ret['plugin_paths'] = active_plugins.split('"')[1::2]
        else:
            ret['plugin_paths'] = []
        cur.execute(f"SELECT COUNT(*) FROM `{tbl('comments')}`")
        if row := cur.fetchone():
            try:
                ret['num_comments'] = int(row[0])
            except ValueError:
                pass
        try:
            cur.execute(
                f"SELECT `meta_value` FROM `{tbl('sitemeta')}` "
                "WHERE `meta_key` = 'blog_count'"
            )
            if row := cur.fetchone():
                ret['multisites'] = row[0]
        except pymysql.ProgrammingError:
            pass  # sitemeta table doesn't exist
        # supply {prefix}capabilities as a query arg instead of using `tbl()`
        # because it's a literal value, not a MySQL identifier
        cur.execute(
            f"SELECT u.user_login FROM `{tbl('users')}` AS u "
            f"LEFT JOIN `{tbl('usermeta')}` AS um ON u.ID = um.user_id "
            "WHERE um.meta_key = %s AND um.meta_value LIKE %s",
            (f"{prefix}capabilities", r'%admin%'),
        )
        ret['admin_users'] = [x[0] for x in cur.fetchall()]
        return ret

    def parse_config(self, path: Path) -> tuple[str, dict[str, str]]:
        """Parse database variables from a wp-config.php file"""
        db_re = re.compile(
            r"[^\#]*define\(\s*\'DB\_([A-Z]+)\'\s*\,\s*\'([^\']+)\'"
        )
        prefix_re = re.compile(r"[^#]*\$table_prefix\s*=\s*\'([^\']+)\'")
        db_info = {}
        prefix = None
        with open(path, encoding='utf-8') as conf_file:
            conf = conf_file.read().splitlines()
        for line in conf:
            if db_match := db_re.match(line):  # database setting found
                db_info[db_match.group(1)] = db_match.group(2)
            if prefix_match := prefix_re.match(line):  # table prefix found
                prefix = prefix_match.group(1)
        if not prefix:
            raise KeyError(f"Could not find $table_prefix in {path}")
        try:
            if self.args.use_root:
                conn_kwargs = {
                    'user': 'root',
                    'read_default_file': '/root/.my.cnf',
                    'database': db_info['NAME'],
                }
                return prefix, conn_kwargs
            conn_kwargs = {
                'user': db_info['USER'],
                'password': db_info['PASSWORD'],
                'database': db_info['NAME'],
            }
            return prefix, conn_kwargs
        except KeyError as exc:
            raise KeyError(f"Could not find DB_{exc} in {path}") from exc

    def show_plugin(self, site_path: Path, plugin_path: str) -> None:
        path = site_path / 'wp-content/plugins' / plugin_path
        try:
            with open(path, encoding='utf-8', errors='replace') as conf_file:
                conf = reversed(conf_file.read().splitlines())
        except OSError:
            self.yellow(f"Found active but missing plugin at {path}")
            return
        slug = slug_from_path(plugin_path)
        latest = latest_plugin_version(slug)
        version, uri, name = '', '', ''
        for line in conf:
            if 'Plugin Name:' in line:
                line = line.split(':')
                name = line[1].strip()
            elif 'Version:' in line:
                line = line.split(':')
                version = line[1].strip()
            elif 'Plugin URI:' in line:
                line = line.split(':')
                del line[0]
                uri = ':'.join(line).strip()
        if latest and version:
            outdated = pkg_version.parse(version) < pkg_version.parse(latest)
        else:  # missing data; cannot determine if out of date
            outdated = False
        if slug in GOOD_PLUGINS:
            comment = GOOD_PLUGINS[slug]
            self.green(f"  {name}", end='')
        elif slug in BAD_PLUGINS:
            comment = BAD_PLUGINS[slug]
            self.red(f"  {name}", end='')
        else:
            comment = ''
            self.bold(f"  {name}", end='')
        self.print(' - ', end='')
        self.yellow(version, end=' ')
        if outdated:
            self.print(' - ', end='')
            self.red(f"{latest} available", end='')
            if self.args.style == 'str' and slug not in BAD_PLUGINS:
                how = self.urlize(
                    self.kb_urls["wp_pluginup"], 'How to update plugins'
                )
                self.print(f' - {how}', end='')
        if comment:
            self.print(' - ', end='')
            self.yellow(comment, end='')
        self.print('')
        if self.args.style != 'str':
            self.print(f"    {uri}")

    def check_wp_cache(self, site_path: Path, plugins: list[str]) -> None:
        w3tc_enabled = 'w3-total-cache/w3-total-cache.php' in plugins
        wpsc_enabled = 'wp-super-cache/wp-cache.php' in plugins
        w3tc_browser_cache, w3tc_page_cache, wpsc_rewrites = False, None, False
        try:
            with open(site_path / '.htaccess', encoding='utf-8') as htaccess:
                for line in htaccess:
                    if 'BEGIN W3TC Browser Cache' in line:
                        w3tc_browser_cache = True
                    if 'BEGIN WPSuperCache' in line:
                        wpsc_rewrites = True
        except OSError:
            pass
        master_path = site_path / 'wp-content/w3tc-config/master.php'
        try:
            with open(master_path, encoding='utf-8') as php:
                for line in php:
                    if "pgcache.enabled" in line:
                        line = line.lower()
                        if 'true' in line:
                            w3tc_page_cache = True
                        elif 'false' in line:
                            w3tc_page_cache = False
                        break
        except OSError:
            pass
        if w3tc_enabled and wpsc_enabled:
            self.red(
                'W3 Total Cache AND WP Super Cache are enabled. '
                'They are incompatible'
            )
            return
        if w3tc_enabled:
            if wpsc_rewrites:
                self.red(
                    'WP Super Cache rewrites found, but W3 Total Cache '
                    'is enabled'
                )
                return
            self.green('W3TC enabled:', end=' ')
            show_kb = False
            if w3tc_browser_cache:
                self.green('W3TC Browser Cache rewrites found', end=' ')
            else:
                self.red('W3TC Browser Cache rewrites disabled', end=' ')
                show_kb = True
            if w3tc_page_cache:
                self.green('and W3TC Page Cache enabled', end='')
            elif w3tc_page_cache is None:
                self.red('but could not detect W3TC Page Cache state', end='')
            else:
                self.red('and W3TC Page Cache disabled', end='')
                show_kb = True
            if show_kb and self.args.style == 'str':
                how = self.urlize(
                    self.kb_urls['w3_total'],
                    'Setting up W3 Total Cache',
                )
                self.print(f" - {how}")
            else:
                self.print('')
            return
        if wpsc_enabled:
            self.green("WP Super Cache enabled", end=' ')
            if w3tc_page_cache or w3tc_browser_cache:
                self.red("but W3TC rewrites were found")
                return
            if wpsc_rewrites:
                self.green('and rewrites enabled')
            else:
                self.red('but missing rewrites', end='')
                if self.args.style == 'str':
                    how = self.urlize(
                        self.kb_urls['wp_super'], 'Setting up WP Super Cache'
                    )
                    self.print(f' - {how}')
                else:
                    self.print('')
            return
        self.red("No approved caching enabled", end='')
        if self.args.style == 'str':
            how = self.urlize(
                self.kb_urls['w3_total'], 'Setting up W3 Total Cache'
            )
            self.print(f' - {how}')
        else:
            self.print('')
        if self.args.guide_bool:
            self.args.wp_found = True
            self.red("    **See below for caching instructions")


def latest_wp_ver():
    """Determine the latest WordPress version"""
    try:
        api_response = requests.get(
            'https://api.wordpress.org/core/version-check/1.7/',
            timeout=10.0,
        ).json()
    except (ValueError, TypeError, requests.exceptions.RequestException):
        return None
    try:
        return str(
            max(pkg_version.parse(x['current']) for x in api_response['offers'])
        )
    except (KeyError, TypeError, ValueError):
        return None


def slug_from_path(plugin_path: str) -> str:
    """From a plugin path, obtain a WordPress 'slug'
    http://codex.wordpress.org/Glossary#Slug"""
    # Expected format of plugin_path:
    #  Complex plugins will be in their own directory, and file_path
    #   will point to their entry script, like
    #   'w3-total-cache/w3-total-cache.php'. For these, we want the
    #   'w3-total-cache' part in the beginning
    #  Simple plugins will be directly in wp-content/plugins, such
    #   as 'hello.php' (hello dolly). For these, we want the first
    #   'hello' part in the beginning.
    if plugin_path.endswith('.php') and len(plugin_path) > 4:
        slug = plugin_path[:-4]
    else:
        slug = plugin_path
    slug = slug.lstrip('/')
    if '/' in slug:
        slug = slug[: slug.index('/')]
    return slug


def latest_plugin_version(slug: str) -> Union[str, None]:
    """Determine the latest version of a plugin"""
    try:
        return LATEST_PLUGINS[slug]
    except KeyError:
        pass
    try:
        ver = requests.get(
            f'https://api.wordpress.org/plugins/info/1.0/{slug}.json',
            timeout=10.0,
        ).json()['version']
        if ver:
            LATEST_PLUGINS[slug] = ver
        return ver
    except (
        ValueError,
        TypeError,
        KeyError,
        requests.exceptions.RequestException,
    ):
        return None


def load_plugin_data() -> tuple[dict[str, str], dict[str, str]]:
    data = requests.get(
        'http://repo.imhadmin.net/open/control/wordpress.json',
        timeout=10.0,
    ).json()
    return data['good'], data['bad']


LATEST_WP = latest_wp_ver()
if not LATEST_WP:
    print('Warning: could not determine latest WordPress version')
LATEST_PLUGINS: dict[str, str] = {}
GOOD_PLUGINS, BAD_PLUGINS = load_plugin_data()