import asyncio import errno import logging import os import pwd import time from collections import defaultdict from collections.abc import Awaitable, Callable from distutils.version import LooseVersion from pathlib import Path from defence360agent.api import inactivity from defence360agent.contracts.config import ( MalwareScanScheduleInterval as Interval, SystemConfig, ANTIVIRUS_MODE, choose_value_from_config, ) from defence360agent.files import Index, WP_RULES from defence360agent.sentry import log_message from defence360agent.utils.fd_ops import ( open_dir_no_symlinks, open_nofollow, rmtree_fd, safe_dir, ) from defence360agent.contracts.config import Wordpress from defence360agent.wordpress.wp_rules import ( get_wp_rules_data, get_wp_ruleset_version, ) from defence360agent.model.wordpress import WordpressSite, WPSite from defence360agent.model.wp_disabled_rule import WPDisabledRule from defence360agent.wordpress import cli, telemetry from defence360agent.wordpress.constants import PLUGIN_VERSION_FILE from defence360agent.wordpress.utils import ( _validate_preset, calculate_next_scan_timestamp, clear_get_cagefs_enabled_users_cache, ensure_site_data_directory, format_php_with_embedded_json, get_imunify_package_versions, get_last_scan, get_malware_history, prepare_plugin_config, prepare_scan_data, write_plugin_data_file_atomically, ) from defence360agent.wordpress.site_repository import ( clear_manually_deleted_flag, delete_site, get_installed_sites_by_domains, get_outdated_sites, get_sites_for_user, get_sites_to_adopt, get_sites_to_install, get_sites_to_mark_as_manually_deleted, get_installed_sites, insert_installed_sites, mark_site_as_manually_deleted, update_site_identity, update_site_version, ) from defence360agent.wordpress.proxy_auth import setup_site_authentication logger = logging.getLogger(__name__) # Default when waf_enabled is missing from config (old schema). # True keeps WAF enabled — no behavior change on upgrade. _WAF_ENABLED_DEFAULT = True # Default when ai_bot_protection is missing from config (old schema on # a host where sibling packages haven't shipped the field yet). False # because the feature is opt-in — if we can't determine admin intent, # stay off rather than silently activate request-blocking logic. _AI_BOT_PROTECTION_DEFAULT = False _AI_BOT_PROTECTION_PRESET_DEFAULT = "balanced" def _get_global_waf_enabled() -> bool: """Read WORDPRESS.waf_enabled from config, defaulting to True. Returns _WAF_ENABLED_DEFAULT when the config key is missing (old schema without waf_enabled), so WAF stays enabled — no behavior change on upgrade. """ try: return bool(Wordpress.WAF_ENABLED) except KeyError: return _WAF_ENABLED_DEFAULT def _get_global_ai_bot_protection() -> bool: """Read WORDPRESS.ai_bot_protection from config, defaulting to False. Returns _AI_BOT_PROTECTION_DEFAULT when the config key is missing — e.g. the ai_bot_protection field hasn't rolled out to this install's imunify360 yet, or a sibling package is still on an older schema. Keeps the feature off in all ambiguous cases. """ try: return bool(Wordpress.AI_BOT_PROTECTION) except KeyError: return _AI_BOT_PROTECTION_DEFAULT def _get_global_ai_bot_protection_preset() -> str: """Read WORDPRESS.ai_bot_protection_preset from config, defaulting to "balanced". Two layers of safety: KeyError on a missing key (older schema, agent upgrade in progress) and _validate_preset() on the value itself (hand-edited override file, future preset rolled in via a sibling package this version doesn't recognise). Both fall back to the same canonical default so all layers — schema, agent, plugin — agree. """ try: raw = Wordpress.AI_BOT_PROTECTION_PRESET except KeyError: return _AI_BOT_PROTECTION_PRESET_DEFAULT return _validate_preset(raw) def _is_waf_enabled_for_user_sync(username: str) -> bool: """Check if WAF is enabled for a user, respecting global-off precedence.""" if not _get_global_waf_enabled(): return False try: value, _ = choose_value_from_config( "WORDPRESS", "waf_enabled", username=username ) except KeyError: return _WAF_ENABLED_DEFAULT return _WAF_ENABLED_DEFAULT if value is None else bool(value) async def is_waf_enabled_for_user(username: str) -> bool: """Async wrapper — runs config file I/O in executor.""" loop = asyncio.get_event_loop() return await loop.run_in_executor( None, _is_waf_enabled_for_user_sync, username ) COMPONENTS_DB_PATH = Path( "/var/lib/cloudlinux-app-version-detector/components_versions.sqlite3" ) def _get_user_schedule_config(username: str, admin_config: SystemConfig): """ Get user-specific schedule configuration with lazy import fallback. Returns default values if imav.malwarelib is not available. """ try: from imav.malwarelib.plugins.schedule_watcher import ( get_user_schedule_config, ) return get_user_schedule_config(username, admin_config) except ImportError: logger.debug( "imav.malwarelib not available, returning default schedule config" ) return Interval.NONE, 0, 1, 0 def get_updated_wp_rules_data(index: Index) -> dict | None: """ Retrieve WordPress rules with ANTIVIRUS_MODE handling and global disable filtering. In ANTIVIRUS_MODE, all rules are set to monitoring mode ("pass"). Globally disabled rules are filtered out entirely — they should not appear in rules.php. Domain-specific disables are handled separately via disabled-rules.php. Args: index: The Index object used to locate the wp-rules.zip file. Returns: The parsed wp-rules data with mode adjusted for ANTIVIRUS_MODE and globally disabled rules removed, or None if rules cannot be loaded. """ rules_data = get_wp_rules_data(index) if rules_data is None: return None if ANTIVIRUS_MODE: # all rules will be in monitoring mode only for AV and AV+ for cve, params in rules_data.items(): params["mode"] = "pass" # Filter out globally disabled rules — these are excluded from rules.php globally_disabled = set(WPDisabledRule.get_global_disabled()) if globally_disabled: rules_data = { cve: params for cve, params in rules_data.items() if cve not in globally_disabled } return rules_data def clear_caches(): """Clear all WordPress-related caches.""" clear_get_cagefs_enabled_users_cache() cli.clear_get_content_dir_cache() def site_search(items: dict, user_info: pwd.struct_passwd, matcher) -> dict: # Get all WordPress sites for the user (the main site is always last) user_sites = get_sites_for_user(user_info) result = {path: [] for path in user_sites} for item in items: # Find all matching sites for this item matching_sites = [path for path in user_sites if matcher(item, path)] if matching_sites: # Find the most specific (longest) matching path most_specific_site = max(matching_sites, key=len) result[most_specific_site].append(item) return result async def _get_scan_data_for_user( sink, user_info: pwd.struct_passwd, admin_config: SystemConfig ): # Get the last scan data last_scan = await get_last_scan(sink, user_info.pw_name) # Extract the last scan date last_scan_time = last_scan.get("scan_date", None) # Get user-specific schedule configuration interval, hour, day_of_month, day_of_week = _get_user_schedule_config( user_info.pw_name, admin_config ) next_scan_time = None if interval != Interval.NONE: next_scan_time = calculate_next_scan_timestamp( interval, hour, day_of_month, day_of_week ) # Get the malware history for the user malware_history = get_malware_history(user_info.pw_name) # Split malware history by site. This part relies on the main site being the last one in the list. # Without this all malware could be attributed to the main site. malware_by_site = site_search( malware_history, user_info, lambda item, path: item["resource_type"] == "file" and item["file"].startswith(path), ) return last_scan_time, next_scan_time, malware_by_site async def _send_telemetry_task(coro, semaphore: asyncio.Semaphore): async with semaphore: try: await coro except Exception as e: logger.error(f"Telemetry task failed: {e}") async def process_telemetry_tasks(coroutines: list, concurrency=10): """ Process a list of telemetry coroutines with a concurrency limit.s """ if not coroutines: return semaphore = asyncio.Semaphore(concurrency) tasks = [ asyncio.create_task(_send_telemetry_task(coro, semaphore)) for coro in coroutines ] try: await asyncio.gather(*tasks) except Exception as e: logger.error(f"Some telemetry tasks failed: {e}") async def load_wp_rules_php(): """ Load WordPress rules from the index and format them as PHP. Returns: str or None: PHP-formatted rules data, or None if rules could not be loaded. """ try: wp_rules_index = Index(WP_RULES, integrity_check=False) await wp_rules_index.update() wp_rules_data = get_updated_wp_rules_data(wp_rules_index) except Exception as e: logger.warning( "Failed to load wp-rules index: %s, skipping rules installation", e, ) return None if not wp_rules_data: logger.warning( "valid WordPress rules not found, skipping rules installation" ) return None # Get version and create ruleset dict with version and rules wp_rules_version = get_wp_ruleset_version(wp_rules_index) ruleset_dict = { "version": wp_rules_version, "rules": wp_rules_data, } return format_php_with_embedded_json(ruleset_dict) async def install_everywhere(sink): """Install the imunify-security plugin for all sites where it is not installed.""" sites = get_sites_to_install() installer = WordPressSiteInstaller(sink, sites) return await installer.run() async def adopt_found_sites(sink): """ Adopt WordPress sites where the plugin is installed but not tracked in our database or flagged as manually removed. This handles scenarios like: - Sites copied/migrated from another location - Sites migrated from another server - Sites where the manually_deleted flag was incorrectly set (past bugs) - Sites where the user installed the plugin from wordpress.org """ sites = get_sites_to_adopt() processor = WordPressSiteAdopter(sink, sites) return await processor.run() def get_latest_plugin_version() -> str: """Get the latest version of the imunify-security plugin from the version file.""" try: if not PLUGIN_VERSION_FILE.exists(): logger.error( "Plugin version file does not exist: %s", PLUGIN_VERSION_FILE ) return None return PLUGIN_VERSION_FILE.read_text().strip() except Exception as e: logger.error("Failed to read plugin version file: %s", e) return None async def update_everywhere(sink): """Update the imunify-security plugin on all sites where it is installed.""" latest_version = get_latest_plugin_version() if not latest_version: logger.error("Could not determine latest plugin version") return logger.info( "Updating imunify-security wp plugin to the latest version %s", latest_version, ) updated = set() telemetry_coros = [] with inactivity.track.task("wp-plugin-update"): try: # Get sites with outdated versions outdated_sites = get_outdated_sites(latest_version) logger.info(f"Found {len(outdated_sites)} outdated sites") if not outdated_sites: return # Create SystemConfig once for all users admin_config = SystemConfig() versions = await get_imunify_package_versions() # Group sites by user id sites_by_user = defaultdict(list) for site in outdated_sites: sites_by_user[site.uid].append(site) # Process each user's sites for uid, sites in sites_by_user.items(): try: user_info = pwd.getpwuid(uid) username = user_info.pw_name except Exception as error: logger.error( "Failed to get username for uid=%d. error=%s", uid, error, ) continue # Get scan data once for all sites of this user ( last_scan_time, next_scan_time, malware_by_site, ) = await _get_scan_data_for_user( sink, user_info, admin_config ) plugin_config = prepare_plugin_config(username) for site in sites: if await remove_site_if_missing(sink, site): continue try: # Check if site still exists if not await cli.is_wordpress_installed(site): logger.info( "WordPress site no longer exists: %s", site ) continue # Prepare scan data scan_data = prepare_scan_data( last_scan_time, next_scan_time, username, site, malware_by_site, versions=versions, ) # Update the scan data file await update_scan_data_file(site, scan_data) # Keep plugin_config.php fresh alongside scan_data # — covers the case where a ConfigUpdate event # was missed (plugin re-installed after the # toggle, first scan after an upgrade, etc). await update_plugin_config_file(site, plugin_config) # Now update the plugin await cli.plugin_update(site) updated.add(site) # Get the version after update version = await cli.get_plugin_version(site) if version: # Store original version for comparison original_version = site.version # Update the database with the new version update_site_version(site, version) # Create a new WPSite with updated version site = site.build_with_version(version) # Determine if this is a downgrade is_downgrade = LooseVersion( version ) < LooseVersion(original_version) # Prepare telemetry telemetry_coros.append( telemetry.send_event( sink=sink, event=( "downgraded_by_imunify" if is_downgrade else "updated_by_imunify" ), site=site, version=version, ) ) except Exception as error: logger.error( "Failed to update plugin on site=%s error=%s", site, error, ) logger.info( "Updated imunify-security wp plugin on %d sites", len(updated), ) except asyncio.CancelledError: logger.info( "Update of imunify-security wp plugin was cancelled. Plugin" " was updated on %d sites", len(updated), ) except Exception as error: logger.error( "Error occurred during plugin update. error=%s", error ) raise finally: # Send telemetry await process_telemetry_tasks(telemetry_coros) async def delete_plugin_files(site: WPSite): data_dir = await cli.get_data_dir(site) # Open both target and parent dirs with symlink protection before # performing any destructive operations. try: dir_fd = open_dir_no_symlinks(data_dir) except OSError as exc: if exc.errno == errno.ENOENT: return # directory does not exist — nothing to delete if exc.errno in (errno.ELOOP, errno.ENOTDIR): logger.warning( "Skipping rmtree: data directory %s is a symlink", data_dir ) return raise try: with safe_dir(data_dir.parent) as parent_fd: try: await asyncio.to_thread(rmtree_fd, dir_fd) finally: os.close(dir_fd) dir_fd = -1 # Remove the now-empty directory via the parent fd. os.rmdir(data_dir.name, dir_fd=parent_fd) except BaseException: if dir_fd >= 0: os.close(dir_fd) raise async def remove_from_single_site(site: WPSite, sink, telemetry_coros) -> int: """ Remove the imunify-security plugin from a single site, including all cleanup and telemetry. Returns the number of affected sites (should be 1 if deletion was successful). This function is intended to be protected with asyncio.shield to ensure it completes even if the parent task is cancelled. """ try: # Check if site is still installed and accessible using WP CLI is_installed = await cli.is_plugin_installed(site) if not is_installed: # Plugin is no longer installed. It was removed manually by the user. await process_manually_deleted_plugin( site, time.time(), sink, telemetry_coros ) return 0 # Get the version of the plugin (for telemetry data) version = await cli.get_plugin_version(site) # Uninstall the plugin from WordPress site. await cli.plugin_uninstall(site) # Delete the data files from the site. await delete_plugin_files(site) # Delete the site from database. affected = delete_site(site) # Send telemetry for successful uninstall telemetry_coros.append( telemetry.send_event( sink=sink, event="uninstalled_by_imunify", site=site, version=version, ) ) return affected except Exception as error: # Log any error that occurs during the removal process logger.error("Failed to remove plugin from %s %s", site, error) return 0 async def remove_all_installed(sink): """Remove the imunify-security plugin from all sites where it is installed.""" logger.info("Deleting imunify-security wp plugin") telemetry_coros = [] affected = 0 with inactivity.track.task("wp-plugin-removal"): try: clear_caches() to_remove = get_installed_sites() for site in to_remove: try: affected += await asyncio.shield( remove_from_single_site(site, sink, telemetry_coros) ) except asyncio.CancelledError: logger.info( "Deleting imunify-security wp plugin was cancelled." " Plugin was deleted from %d sites (out of %d)", affected, len(to_remove), ) except Exception as error: logger.error("Error occurred during plugin deleting. %s", error) raise finally: logger.info( "Removed imunify-security wp plugin from %s sites", affected, ) # send telemetry await process_telemetry_tasks(telemetry_coros) async def process_manually_deleted_plugin(site, now, sink, telemetry_coros): """ Process the manually deleted plugin for a single site. Args: site: The site to process. now: The current time. sink: The telemetry/event sink. telemetry_coros: The list of telemetry coroutines to add the event to. The process includes: - marking the site as manually deleted in the database - removing plugin data files - sending telemetry for manual removal """ try: # Mark the site as manually deleted in the database mark_site_as_manually_deleted(site, now) # Remove plugin data files await delete_plugin_files(site) # Send telemetry for manual removal telemetry_coros.append( telemetry.send_event( sink=sink, event="removed_by_user", site=site, version=site.version, ) ) except Exception as error: logger.error( "Failed to process manually deleted plugin for site=%s error=%s", site, error, ) async def tidy_up_manually_deleted( sink, freshly_installed_sites: set[WPSite] = None ): """ Tidy up sites that have been manually deleted by the user. Args: sink: The telemetry/event sink. freshly_installed_sites: Optional set of sites that were just installed and should be excluded from being marked as manually deleted to avoid race conditions. """ telemetry_coros = [] try: to_mark_as_manually_removed = get_sites_to_mark_as_manually_deleted( freshly_installed_sites ) if to_mark_as_manually_removed: now = time.time() for site in to_mark_as_manually_removed: await process_manually_deleted_plugin( site, now, sink, telemetry_coros ) except Exception as error: logger.error("Error occurred during site tidy up. %s", error) finally: if telemetry_coros: await process_telemetry_tasks(telemetry_coros) async def update_data_on_sites(sink, sites: list[WPSite]): if not sites: return # Create SystemConfig once for all users admin_config = SystemConfig() versions = await get_imunify_package_versions() # Group sites by user id sites_by_user = defaultdict(list) for site in sites: sites_by_user[site.uid].append(site) # Now iterate over the grouped sites for uid, sites in sites_by_user.items(): try: user_info = pwd.getpwuid(uid) username = user_info.pw_name except Exception as error: logger.error( "Failed to get username for uid=%d. error=%s", uid, error, ) continue ( last_scan_time, next_scan_time, malware_by_site, ) = await _get_scan_data_for_user(sink, user_info, admin_config) plugin_config = prepare_plugin_config(username) for site in sites: if await remove_site_if_missing(sink, site): continue try: # Prepare scan data scan_data = prepare_scan_data( last_scan_time, next_scan_time, username, site, malware_by_site, versions=versions, ) # Update the scan data file await update_scan_data_file(site, scan_data) # Keep plugin_config.php fresh alongside scan_data. await update_plugin_config_file(site, plugin_config) except Exception as error: logger.error( "Failed to update site data on site=%s error=%s", site, error, ) async def _write_json_php_data_file( site: WPSite, filename: str, data: dict ) -> None: """Embed ``data`` as JSON inside a PHP file at ``/``.""" user_info = pwd.getpwuid(site.uid) gid = user_info.pw_gid data_dir = await ensure_site_data_directory(site, user_info) php_content = format_php_with_embedded_json(data) write_plugin_data_file_atomically( data_dir / filename, php_content, uid=site.uid, gid=gid ) async def update_scan_data_file(site: WPSite, scan_data: dict): await _write_json_php_data_file(site, "scan_data.php", scan_data) async def update_plugin_config_file(site: WPSite, plugin_config: dict) -> None: """ Write plugin_config.php for a single WordPress site. Separate file from scan_data.php so a config toggle doesn't force rewriting the malware list, and so the mu-plugin hot path loads only what it needs per request. """ await _write_json_php_data_file(site, "plugin_config.php", plugin_config) async def update_plugin_config_on_sites(sites: list[WPSite]) -> int: """ Rewrite plugin_config.php on every managed site in one pass. Used by the ConfigUpdate handler that reacts to WORDPRESS.ai_bot_protection toggles. Writes only plugin_config.php — scan_data.php is untouched, so a toggle doesn't churn the (potentially large) malware payload or wait on a scan cycle. No sink is needed: unlike update_data_on_sites we emit no telemetry here — the per-site write loop just needs local file I/O plus the process-level logger for errors. Returns the number of sites successfully updated so the caller can decide whether to advance its cached state. """ if not sites: return 0 updated = 0 # Group by uid so we look up username once per user, mirroring # update_data_on_sites' pattern and making per-user error isolation # straightforward. sites_by_user: dict[int, list[WPSite]] = defaultdict(list) for site in sites: sites_by_user[site.uid].append(site) for uid, user_sites in sites_by_user.items(): try: user_info = pwd.getpwuid(uid) username = user_info.pw_name except Exception as error: logger.error( "Failed to get username for uid=%d. error=%s", uid, error, ) continue plugin_config = prepare_plugin_config(username) for site in user_sites: try: await update_plugin_config_file(site, plugin_config) updated += 1 except Exception as error: logger.error( "Failed to update plugin_config.php on site=%s error=%s", site, error, ) return updated async def update_wp_rules_for_site( site: WPSite, user_info: pwd.struct_passwd, wp_rules_php: str, updated: set, failed: set, ) -> None: """ Deploy wp-rules to a single WordPress site and track the result. Args: site: WordPress site to deploy to user_info: User information from pwd wp_rules_php: Formatted PHP rules content updated: Set to add site to if successful failed: Set to add site to if failed """ gid = user_info.pw_gid try: data_dir = await ensure_site_data_directory(site, user_info) rules_path = data_dir / "rules.php" write_plugin_data_file_atomically( rules_path, wp_rules_php, uid=site.uid, gid=gid ) updated.add(site) logger.info("Updated wp-rules for site %s", site.docroot) except Exception as error: failed.add(site) logger.error( "Failed to update wp-rules for site %s: %s", site.docroot, error, ) async def _deploy_to_sites( sites: list[WPSite], make_task: Callable[ [WPSite, pwd.struct_passwd, set, set], Awaitable[None] ], task_name: str, fingerprint: str, skip_waf_disabled: bool = False, sink=None, ) -> None: """ Run a per-site async deployment over a list of WordPress sites. Groups sites by user, resolves UIDs, then runs tasks concurrently in batches. Args: sites: WordPress sites to deploy to make_task: Callable that creates a coroutine for one site. Signature: (site, user_info, updated_set, failed_set) -> awaitable task_name: Human-readable name for logging and inactivity tracking fingerprint: Sentry fingerprint for user-lookup failures sink: Optional telemetry sink for remove_site_if_missing """ updated = set() failed = set() with inactivity.track.task(task_name): try: start_time = time.time() sites_by_user = defaultdict(list) for site in sites: sites_by_user[site.uid].append(site) tasks = [] for uid, user_sites in sites_by_user.items(): try: user_info = pwd.getpwuid(uid) username = user_info.pw_name except Exception as error: log_message( "Skipping {task} update for {count} site(s)" " belonging to user {user} because username" " retrieval failed. Reason: {reason}", format_args={ "task": task_name, "count": len(user_sites), "user": uid, "reason": error, }, level="warning", component="wordpress", fingerprint=fingerprint, ) for site in user_sites: failed.add(site) continue if skip_waf_disabled: try: waf_enabled = await is_waf_enabled_for_user(username) except Exception: logger.warning( "Could not check WAF status for user %s," " proceeding with deployment", username, exc_info=True, ) waf_enabled = True if not waf_enabled: logger.info( "WAF disabled for user %s, skipping %d site(s)", username, len(user_sites), ) continue for site in user_sites: if await remove_site_if_missing(sink, site): continue tasks.append(make_task(site, user_info, updated, failed)) max_concurrent = 10 for i in range(0, len(tasks), max_concurrent): batch = tasks[i : i + max_concurrent] await asyncio.gather(*batch, return_exceptions=True) elapsed = time.time() - start_time logger.info( "%s deployment complete. Updated: %d, Failed: %d," " Duration: %.2fs", task_name, len(updated), len(failed), elapsed, ) except asyncio.CancelledError: logger.info( "%s deployment was cancelled. Updated %d sites", task_name, len(updated), ) except Exception as error: logger.error( "Error occurred during %s deployment. error=%s", task_name, error, ) raise async def _deploy_wp_rules_php(wp_rules_php: str, sink=None) -> None: """Deploy pre-formatted wp-rules PHP content to all active WordPress sites.""" clear_caches() installed_sites = get_installed_sites() if not installed_sites: logger.debug("No active WordPress sites found") return def make_task(site, user_info, updated, failed): return update_wp_rules_for_site( site, user_info, wp_rules_php, updated, failed ) await _deploy_to_sites( installed_sites, make_task, task_name="wp-rules", fingerprint="wp-rules-update-skip-user", skip_waf_disabled=True, sink=sink, ) async def update_wp_rules_on_sites(index: Index, is_updated: bool) -> None: """ Hook that runs when wp-rules files are updated. Extracts wp-rules.yaml from wp-rules.zip and deploys to all active WordPress sites. Args: index: Index object for wp-rules is_updated: Whether files were actually updated """ if not Wordpress.SECURITY_PLUGIN_ENABLED: logger.info( "wordpress security plugin not enabled, skipping wp-rules" " deployment" ) return if not is_updated: logger.info("wp-rules not updated, skipping deployment") return logger.info("Starting wp-rules deployment to WordPress sites") wp_rules_data = get_updated_wp_rules_data(index) if not wp_rules_data: logger.error("No valid wp-rules found, skipping deployment") return # Get version and create ruleset dict with version and rules wp_rules_version = get_wp_ruleset_version(index) ruleset_dict = { "version": wp_rules_version, "rules": wp_rules_data, } wp_rules_php = format_php_with_embedded_json(ruleset_dict) await _deploy_wp_rules_php(wp_rules_php) _redeploy_wp_rules_lock = asyncio.Lock() # Separate flag is needed because lock.locked() is always True inside the # holder's context, so it cannot indicate whether another caller coalesced. _redeploy_wp_rules_pending = False async def redeploy_wp_rules() -> None: """ Re-deploy rules.php to all WordPress sites. Used when globally disabled rules change, requiring rules.php to be regenerated with updated rule filtering. Uses a coalescing lock: if a redeployment is already running, the request is merged into the current run rather than starting a duplicate deployment. """ global _redeploy_wp_rules_pending if not Wordpress.SECURITY_PLUGIN_ENABLED: logger.info( "wordpress security plugin not enabled, skipping wp-rules" " redeployment" ) return if _redeploy_wp_rules_lock.locked(): _redeploy_wp_rules_pending = True logger.info("wp-rules redeployment already in progress, coalescing") return async with _redeploy_wp_rules_lock: while True: _redeploy_wp_rules_pending = False logger.info( "Starting wp-rules redeployment (global disable change)" ) wp_rules_php = await load_wp_rules_php() if not wp_rules_php: logger.warning("Could not load wp-rules for redeployment") return await _deploy_wp_rules_php(wp_rules_php) if not _redeploy_wp_rules_pending: break logger.info("Re-running wp-rules redeployment (coalesced request)") async def _remove_rules_php_from_sites(sites: list[WPSite]) -> None: """Remove rules.php from the given sites.""" loop = asyncio.get_event_loop() for site in sites: try: data_dir = await cli.get_data_dir(site) rules_path = data_dir / "rules.php" if await loop.run_in_executor(None, rules_path.exists): await loop.run_in_executor(None, os.remove, rules_path) logger.info( "Removed rules.php from %s (WAF disabled)", site.docroot, ) except Exception as e: logger.error( "Failed to remove rules.php from %s: %s", site.docroot, e ) async def redeploy_wp_rules_for_user(username: str) -> None: """Redeploy rules.php only for sites belonging to a specific user.""" if not Wordpress.SECURITY_PLUGIN_ENABLED: return loop = asyncio.get_event_loop() try: user_info = await loop.run_in_executor(None, pwd.getpwnam, username) except KeyError: logger.warning( "User %s not found, skipping WAF rules redeploy", username ) return wp_rules_php = await load_wp_rules_php() if not wp_rules_php: logger.warning("Could not load wp-rules for user redeploy") return sites = await loop.run_in_executor(None, get_installed_sites) user_sites = [s for s in sites if s.uid == user_info.pw_uid] if not user_sites: return updated = set() failed = set() for site in user_sites: await update_wp_rules_for_site( site, user_info, wp_rules_php, updated, failed ) logger.info( "Redeployed wp-rules for user %s: %d updated, %d failed", username, len(updated), len(failed), ) async def remove_waf_rules_for_user(username: str) -> None: """Remove rules.php from all sites belonging to a user.""" loop = asyncio.get_event_loop() try: user_info = await loop.run_in_executor(None, pwd.getpwnam, username) except KeyError: logger.warning( "User %s not found, skipping WAF rules removal", username ) return sites = await loop.run_in_executor(None, get_installed_sites) user_sites = [s for s in sites if s.uid == user_info.pw_uid] await _remove_rules_php_from_sites(user_sites) async def remove_waf_rules_for_all_sites() -> None: """Remove rules.php from all installed sites (global WAF disable).""" loop = asyncio.get_event_loop() sites = await loop.run_in_executor(None, get_installed_sites) logger.info( "Global WAF disabled, removing rules.php from %d site(s)", len(sites), ) await _remove_rules_php_from_sites(sites) def generate_disabled_rules_php(domain: str, timestamp: float) -> str: """ Generate the disabled-rules.php content for a specific domain. Only includes domain-specific disabled rules. Globally disabled rules are handled separately by filtering them out of rules.php. Args: domain: The domain to generate disabled rules for timestamp: Unix timestamp to embed in the file Returns: PHP file content string """ disabled_rule_ids = WPDisabledRule.get_domain_disabled( domain, include_global=False ) data = { "ts": timestamp, "rules": sorted(disabled_rule_ids), } return format_php_with_embedded_json(data) async def update_disabled_rules_for_site( site: WPSite, user_info: pwd.struct_passwd, timestamp: float, updated: set, failed: set, ) -> None: """ Deploy disabled-rules.php to a single WordPress site and track the result. Args: site: WordPress site to deploy to user_info: User information from pwd timestamp: Unix timestamp for both file content and DB record updated: Set to add site to if successful failed: Set to add site to if failed """ gid = user_info.pw_gid try: data_dir = await ensure_site_data_directory(site, user_info) disabled_rules_path = data_dir / "disabled-rules.php" php_content = generate_disabled_rules_php(site.domain, timestamp) write_plugin_data_file_atomically( disabled_rules_path, php_content, uid=site.uid, gid=gid ) WordpressSite.update(disabled_rules_sync_ts=timestamp).where( WordpressSite.docroot == site.docroot ).execute() updated.add(site) logger.info("Updated disabled-rules for site %s", site.docroot) except Exception as error: failed.add(site) logger.error( "Failed to update disabled-rules for site %s: %s", site.docroot, error, ) async def update_disabled_rules_on_sites( domains: list[str] | None = None, sink=None, ) -> None: """ Deploy disabled-rules.php to WordPress sites. If domains are specified, only updates sites for those domains. If domains is None, updates all installed sites (e.g., after a global disable/enable). Args: domains: List of domains to update, or None for all sites sink: Optional telemetry sink for remove_site_if_missing """ if not Wordpress.SECURITY_PLUGIN_ENABLED: logger.info( "wordpress security plugin not enabled, skipping disabled-rules" " deployment" ) return logger.info("Starting disabled-rules deployment to WordPress sites") clear_caches() if domains: sites = get_installed_sites_by_domains(domains) else: sites = get_installed_sites() if not sites: logger.info("No WordPress sites found for disabled-rules deployment") return def make_task(site, user_info, updated, failed): return update_disabled_rules_for_site( site, user_info, time.time(), updated, failed ) await _deploy_to_sites( sites, make_task, task_name="disabled-rules", fingerprint="disabled-rules-update-skip-user", skip_waf_disabled=True, sink=sink, ) async def update_auth_everywhere(sink=None): """Update auth.php files for all existing WordPress sites.""" logger.info("Updating auth.php files for existing WordPress sites") updated = set() failed = set() with inactivity.track.task("wp-auth-update"): try: clear_caches() # Get all installed sites from db installed_sites = get_installed_sites() if not installed_sites: logger.info("No installed WordPress sites found") return sites_by_user = defaultdict(list) for site in installed_sites: sites_by_user[site.uid].append(site) # Process users concurrently tasks = [] for uid, sites in sites_by_user.items(): try: user_info = pwd.getpwuid(uid) except Exception as error: log_message( "Skipping auth update for WordPress sites on" " {count} site(s) because they belong to user" " {user} and it is not possible to retrieve" " username for this user. Reason: {reason}", format_args={ "count": len(sites), "user": uid, "reason": error, }, level="warning", component="wordpress", fingerprint="wp-plugin-auth-update-skip-user", ) continue for site in sites: if await remove_site_if_missing(sink, site): continue task = update_site_auth(site, user_info, updated, failed) tasks.append(task) # Run all site updates concurrently with a reasonable limit # Adjust max_concurrent based on your system's I/O capacity max_concurrent = 10 for i in range(0, len(tasks), max_concurrent): batch = tasks[i : i + max_concurrent] await asyncio.gather(*batch, return_exceptions=True) logger.info( "Updated auth.php files for %d WordPress sites, %d failed", len(updated), len(failed), ) except asyncio.CancelledError: logger.info( "Auth update for WordPress sites was cancelled. Auth was" " updated for %d sites", len(updated), ) except Exception as error: logger.error("Error occurred during auth update. error=%s", error) raise async def update_site_auth(site, user_info, updated, failed): """Process authentication setup for a single site.""" try: await setup_site_authentication(site, user_info) updated.add(site) except Exception as error: failed.add(site) logger.error( "Failed to update auth for site=%s error=%s", site, error, ) async def remove_site_if_missing(sink, site: WPSite) -> bool: """ Checks if the site directory exists. If not, removes the site from the local database and sends a 'site_removed' telemetry event only if deletion is successful. Returns True if the site was removed (directory missing), False otherwise. Parameters: sink: The telemetry/event sink. site: The WPSite object to check and potentially remove. Side effect: If the site is missing and successfully deleted from database, a telemetry event will be sent. """ if os.path.isdir(site.docroot): return False # Attempt to delete the site from the database first rows_deleted = delete_site(site) # Only send telemetry if the deletion was successful (at least one row was deleted) if rows_deleted > 0: if sink is not None: await telemetry.send_event( sink=sink, event="site_removed", site=site, version=site.version, ) else: logger.warning( "Failed to delete missing site %s from database, no rows affected", site, ) log_message( "Failed to delete missing site {site} from database", format_args={"site": site}, level="warning", component="wordpress", fingerprint="wp-plugin-site-delete-failed", ) return True async def fix_site_data_file_permissions( site: WPSite, file_permissions: int ) -> bool: """Fix data file permissions for a single WordPress site.""" try: data_dir = await cli.get_data_dir(site) try: dir_fd = open_dir_no_symlinks(data_dir) except OSError as exc: if exc.errno in (errno.ENOENT, errno.ELOOP, errno.ENOTDIR): return False raise try: # Fix directory permissions via fd. current_dir_mode = os.stat(dir_fd).st_mode & 0o777 if current_dir_mode != 0o750: os.chmod(dir_fd, 0o750) for file_name in [ "scan_data.php", "plugin_config.php", "auth.php", "rules.php", "disabled-rules.php", ]: try: with open_nofollow(file_name, dir_fd=dir_fd) as file_fd: st = os.fstat(file_fd) if st.st_mode & 0o777 != file_permissions: os.chmod(file_fd, file_permissions) except FileNotFoundError: continue except OSError as exc: if exc.errno == errno.ELOOP: logger.warning( "Skipping chmod: %s/%s is a symlink", data_dir, file_name, ) continue raise finally: os.close(dir_fd) return True except Exception as error: logger.error( "Failed to fix permissions for site=%s error=%s", site, error, ) return False async def fix_data_file_permissions_everywhere(sink): """ Fix data file permissions for all WordPress sites with imunify-security plugin installed. Args: sink: The telemetry/event sink """ fixed = set() failed = set() with inactivity.track.task("wp-plugin-fix-permissions"): try: clear_caches() # Get all installed sites installed_sites = get_installed_sites() if not installed_sites: return # Determine file permissions based on hosting panel from defence360agent.subsys.panels.hosting_panel import ( HostingPanel, ) from defence360agent.subsys.panels.plesk import Plesk file_permissions = ( 0o440 if HostingPanel().NAME == Plesk.NAME else 0o400 ) # Process sites for site in installed_sites: if await remove_site_if_missing(sink, site): continue success = await fix_site_data_file_permissions( site, file_permissions ) if success: fixed.add(site) else: failed.add(site) logger.info( "Fixed data file permissions for %d WordPress sites, %d" " failed", len(fixed), len(failed), ) except asyncio.CancelledError: logger.info( "Fixing data file permissions was cancelled. Permissions were" " fixed for %d sites", len(fixed), ) except Exception as error: logger.error( "Error occurred during permission fixing. error=%s", error ) class WordPressSiteInstaller: """ Handles installation of imunify-security plugin on WordPress sites. This class processes WordPress sites and installs the imunify-security plugin, including setting up authentication, scan data files, and rules. """ install_plugin = True telemetry_event = "installed_by_imunify" task_name = "wp-plugin-installation" log_fingerprint_skip_user = "wp-plugin-install-skip-user" messages = { "start": "Installing imunify-security wp plugin", "complete": "Installed imunify-security wp plugin on {count} sites", "found": "Found {count} site(s) for installation", "error": "Failed to install plugin to site={site} error={error}", "cancelled": ( "Installation of imunify-security wp plugin was cancelled. " "Plugin was installed for {count} sites" ), "exception": ( "Error occurred during plugin installation. error={error}" ), "skip_user": ( "Skipping installation of WordPress plugin on " "{count} site(s) because they belong to user " "{user} and it is not possible to retrieve " "username for this user. Reason: {reason}" ), } def __init__(self, sink, sites): self.sink = sink self.sites = sites self.processed = set() self.authenticated = set() self.rules_installed = set() self.failed_rules_updates = set() self.disabled_rules_installed = set() self.failed_disabled_rules_updates = set() self.disabled_rules_ts: float | None = None self.failed_auth = set() self._current_site: WPSite | None = None async def is_site_ready(self, site): """ Check if site is ready for processing. Override in subclasses to implement different readiness checks. Args: site: The WordPress site to check. Returns: bool: True if the site is ready for processing, False otherwise. """ is_wordpress_installed = await cli.is_wordpress_installed(site) if not is_wordpress_installed: logger.warning( "WordPress site is not accessible using WP CLI. site=%s", site, ) log_message( "WordPress site is not accessible using WP CLI. site={site}", format_args={"site": site}, level="warning", component="wordpress", fingerprint="wp-plugin-cli-not-accessible", ) return False return True def _record_processed_site(self, site, version): """ Record a successfully processed site and persist it to the database. Each site is inserted immediately so that it is tracked in the DB at all times — even if the overall installation loop is cancelled mid-run. Override in subclasses to implement different recording logic. Args: site: The WordPress site that was processed. version: The plugin version installed on the site. """ self.processed.add(site) insert_installed_sites({site}) self._stamp_disabled_rules_sync_ts(site) async def _revert_in_flight_site(self): """Revert the site that was mid-processing when cancellation occurred. Deletes data files and, if this processor installs plugins, attempts to uninstall the partially-installed plugin. Each step runs independently so one failure doesn't skip the other. """ site = self._current_site self._current_site = None try: await delete_plugin_files(site) except Exception as error: logger.warning( "Failed to delete data files for in-flight site %s: %s", site, error, ) if self.install_plugin: await cli.try_plugin_uninstall(site) def _stamp_disabled_rules_sync_ts(self, site: WPSite) -> None: """ Stamp disabled_rules_sync_ts for a single site after it has been inserted into the DB. Called from ``_record_processed_site`` so the DB row already exists. """ if site not in self.disabled_rules_installed: return if self.disabled_rules_ts is None: return WordpressSite.update( disabled_rules_sync_ts=self.disabled_rules_ts ).where(WordpressSite.docroot == site.docroot).execute() async def run(self): """ Process WordPress sites for imunify-security plugin operations. Returns: set: The set of successfully processed sites. """ logger.info(self.messages["start"]) telemetry_tasks = [] with inactivity.track.task(self.task_name): try: clear_caches() if not self.sites: logger.info("No WordPress sites found, nothing to do") return self.processed logger.info( self.messages["found"].format(count=len(self.sites)) ) # Create SystemConfig once for all users admin_config = SystemConfig() # Create wp rules once for all users wp_rules_php = await load_wp_rules_php() # Always set the timestamp, even when no disabled rules # currently exist: previously disabled-then-enabled rules # require deploying an empty disabled-rules.php. self.disabled_rules_ts = time.time() versions = await get_imunify_package_versions() # Group sites by user id sites_by_user = defaultdict(list) for site in self.sites: sites_by_user[site.uid].append(site) # Now iterate over the grouped sites for uid, sites in sites_by_user.items(): try: user_info = pwd.getpwuid(uid) username = user_info.pw_name except Exception as error: log_message( self.messages["skip_user"], format_args={ "count": len(sites), "user": uid, "reason": error, }, level="warning", component="wordpress", fingerprint=self.log_fingerprint_skip_user, ) continue try: waf_enabled = await is_waf_enabled_for_user(username) except Exception: logger.warning( "Could not check WAF status for user %s," " proceeding with deployment", username, exc_info=True, ) waf_enabled = True if not waf_enabled: logger.info( "WAF disabled for user %s, skipping WAF" " rules deployment for %d site(s)", username, len(sites), ) ( last_scan_time, next_scan_time, malware_by_site, ) = await _get_scan_data_for_user( self.sink, user_info, admin_config ) plugin_config = prepare_plugin_config(username) for site in sites: if await remove_site_if_missing(self.sink, site): continue try: # Check if site is ready for processing (WP CLI accessible + other checks) if not await self.is_site_ready(site): continue self._current_site = site # Prepare scan data scan_data = prepare_scan_data( last_scan_time, next_scan_time, username, site, malware_by_site, versions=versions, ) # Create data files (scan data, plugin config, auth token) await update_scan_data_file(site, scan_data) await update_plugin_config_file( site, plugin_config ) await update_site_auth( site, user_info, self.authenticated, self.failed_auth, ) # Install rules — skip when WAF is disabled for # this user (global off or per-user override). if wp_rules_php and waf_enabled: await update_wp_rules_for_site( site, user_info, wp_rules_php, self.rules_installed, self.failed_rules_updates, ) # Install disabled rules if waf_enabled: await update_disabled_rules_for_site( site, user_info, self.disabled_rules_ts, self.disabled_rules_installed, self.failed_disabled_rules_updates, ) # Install the plugin if self.install_plugin: await cli.plugin_install(site) # Get the version of the plugin version = await cli.get_plugin_version(site) if version: site = WPSite.build_with_version(site, version) # Record the processed site self._record_processed_site(site, version) self._current_site = None telemetry_tasks.append( asyncio.create_task( telemetry.send_event( sink=self.sink, event=self.telemetry_event, site=site, version=version, ) ) ) except Exception as error: self._current_site = None logger.error( self.messages["error"].format( site=site, error=repr(error) ) ) logger.info( self.messages["complete"].format(count=len(self.processed)) ) if self.failed_auth: logger.warning( "Failed to authenticate %d sites", len(self.failed_auth), ) if self.failed_rules_updates: logger.warning( "Failed to install wp-rules on %d sites", len(self.failed_rules_updates), ) if self.failed_disabled_rules_updates: logger.warning( "Failed to install disabled-rules on %d sites", len(self.failed_disabled_rules_updates), ) except asyncio.CancelledError: if self._current_site: await self._revert_in_flight_site() logger.info( self.messages["cancelled"].format( count=len(self.processed) ) ) except Exception as error: logger.error( self.messages["exception"].format(error=repr(error)) ) raise finally: if telemetry_tasks: results = await asyncio.gather( *telemetry_tasks, return_exceptions=True ) for result in results: if isinstance(result, Exception): logger.warning( "Failed to send telemetry: %s", result ) return self.processed class WordPressSiteAdopter(WordPressSiteInstaller): """ Handles adoption of existing WordPress sites with imunify-security plugin. Adoption is a special case of installation where the site already has the plugin installed but is not tracked in our database. """ install_plugin = False telemetry_event = "site_found" task_name = "wp-plugin-adoption" log_fingerprint_skip_user = "wp-plugin-adopt-skip-user" messages = { "start": "Adopting imunify-security wp plugin", "complete": "Adopted imunify-security wp plugin on {count} sites", "found": "Found {count} site(s) for adoption", "error": "Failed to adopt plugin to site={site} error={error}", "cancelled": ( "Adoption of imunify-security wp plugin was cancelled. " "Plugin was adopted for {count} sites" ), "exception": "Error occurred during plugin adoption. error={error}", "skip_user": ( "Skipping adoption of WordPress plugin on " "{count} site(s) because they belong to user " "{user} and it is not possible to retrieve " "username for this user. Reason: {reason}" ), } def __init__(self, sink, sites): super().__init__(sink, sites) # Load existing docroots from database for adoption logic self.existing_docroots = { r.docroot for r in WordpressSite.select(WordpressSite.docroot) } def _record_processed_site(self, site, version): """ Record a successfully adopted site and persist it immediately. For adoption, sites that already exist in the database (flagged as manually deleted) have their flag cleared. New sites are inserted into the database right away. Args: site: The WordPress site that was processed. version: The plugin version installed on the site. """ self.processed.add(site) if site.docroot in self.existing_docroots: # Site exists in DB but is flagged - clear flag clear_manually_deleted_flag(site) update_site_identity(site) if version: update_site_version(site, version) else: insert_installed_sites({site}) self._stamp_disabled_rules_sync_ts(site) async def is_site_ready(self, site): """ Check if site is ready for adoption. Args: site: The WordPress site to check. Returns: bool: True if the site is ready for adoption, False otherwise. """ if not await super().is_site_ready(site): return False # Verify plugin is actually installed is_installed = await cli.is_plugin_installed(site) if not is_installed: logger.warning( "Plugin not installed on site %s, skipping adoption", site, ) return False return True