D7net
Home
Console
Upload
information
Create File
Create Folder
About
Tools
:
/
proc
/
3
/
task
/
3
/
root
/
proc
/
3
/
root
/
opt
/
dedrads
/
check_software_mods
/
Filename :
wordpress.py
back
Copy
"""WordPress module for check_software""" from pathlib import Path import re 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 def config_file(cls): return 'wp-config.php' @classmethod 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) -> 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()