File: //opt/imunify360/venv/lib/python3.11/site-packages/defence360agent/plugins/wordpress.py
import asyncio
import logging
import pwd
from pathlib import Path
from typing import Coroutine
from defence360agent.contracts.config import ANTIVIRUS_MODE, SystemConfig
from defence360agent.contracts.hook_events import HookEvent
from defence360agent.contracts.messages import MessageType
from defence360agent.contracts.plugins import (
MessageSink,
MessageSource,
expect,
)
from defence360agent.subsys.panels import hosting_panel
from defence360agent.subsys.persistent_state import (
load_state,
register_lock_file,
save_state,
)
from defence360agent.utils import Scope, recurring_check
from defence360agent.utils.check_lock import check_lock
from defence360agent.utils.common import DAY
from defence360agent.wordpress import plugin
from defence360agent.contracts.config import Wordpress
from defence360agent.model.wordpress import WPSite
from defence360agent.wordpress.site_repository import (
get_sites_by_path,
get_sites_for_user,
get_installed_sites,
)
from defence360agent.wordpress.proxy_auth import (
is_secret_expired,
rotate_secret,
)
from defence360agent.wordpress import (
ChangelogProcessor,
IncidentCollector,
IncidentSender,
)
from defence360agent.wordpress.plugin import update_disabled_rules_on_sites
from defence360agent.model.wordpress_incident import (
delete_old_wordpress_incidents,
)
logger = logging.getLogger(__name__)
LOCK_FILE = register_lock_file("wp-gen-auth", Scope.AV_IM360)
SITE_PROCESSING_LOCK_FILE = register_lock_file(
"wp-site-process", Scope.AV_IM360
)
CONFIG_DIR = Path("/etc/sysconfig/imunify360/imunify360.config.d")
FIRST_INSTALL_CONFIG_FILE = Path(
"/opt/imunify360/venv/share/imunify360/11_on_first_install_wp_av.config"
)
FIRST_INSTALL_CONFIG_PATH = CONFIG_DIR / "11_on_first_install_wp_av.config"
FIRST_INSTALL_FLAG = CONFIG_DIR / ".11_on_first_install_wp_av.flag"
def _get_cleaned_malware_hits(started_timestamp: float) -> list:
"""
Get malware hits cleaned since the given timestamp with lazy import fallback.
Returns empty list if imav.malwarelib is not available.
"""
try:
from imav.malwarelib.model import MalwareHit
return MalwareHit.cleaned_since(started_timestamp)
except ImportError:
logger.debug(
"imav.malwarelib not available, returning empty cleaned hits"
)
return []
class ImunifySecurityPlugin(MessageSink, MessageSource):
SCOPE = Scope.AV_IM360
def __init__(self):
self._loop = None
self._sink = None
self.installation_completed = load_state("ImunifySecurityPlugin").get(
"installed"
)
self.last_config_value = (
load_state("ImunifySecurityPlugin").get("enabled")
or Wordpress.SECURITY_PLUGIN_ENABLED
)
self.installation_task: asyncio.Task | None = None
self.deleting_task: asyncio.Task | None = None
self.install_and_update_task: asyncio.Task | None = None
self.freshly_installed_sites: set[WPSite] = set()
# Incident collection and changelog processing components
self.incident_collector = IncidentCollector()
self.incident_sender = IncidentSender()
self.changelog_processor = ChangelogProcessor()
self._site_processing_task: asyncio.Task | None = None
async def create_sink(self, loop):
pass
async def create_source(self, loop, sink):
self._loop = loop
self._sink = sink
self._update_auth_task = self._loop.create_task(
self.refresh_auth_files()
)
self._site_processing_task = self._loop.create_task(
self.process_wordpress_sites()
)
if ANTIVIRUS_MODE:
await self._apply_first_install_config()
else:
FIRST_INSTALL_FLAG.unlink(missing_ok=True)
async def _apply_first_install_config(self):
if not FIRST_INSTALL_FLAG.exists():
return
if await hosting_panel.HostingPanel().users_count() == 1:
_ = FIRST_INSTALL_CONFIG_PATH.write_text(
FIRST_INSTALL_CONFIG_FILE.read_text()
)
FIRST_INSTALL_CONFIG_PATH.chmod(0o600)
FIRST_INSTALL_FLAG.unlink()
async def shutdown(self):
self._update_auth_task.cancel()
# CancelledError is handled by @recurring_check():
await self._update_auth_task
# Cancel site processing (changelogs + incidents) task
if self._site_processing_task:
self._site_processing_task.cancel()
await self._site_processing_task
def _task_in_progress(self, task_attr_name):
if not hasattr(self, task_attr_name):
logger.error("Unknown task '%s'", task_attr_name)
return False
task = getattr(self, task_attr_name)
return task is not None and not task.done() and not task.cancelled()
async def process_installation(self, coro: Coroutine, for_new_sites=False):
if self._task_in_progress("deleting_task"):
if for_new_sites:
coro.close()
return
if self.deleting_task:
self.deleting_task.cancel()
await self.deleting_task
if self._task_in_progress("installation_task"):
logger.warning("Installation is already running")
coro.close()
return
self.installation_task = asyncio.create_task(coro)
async def process_deleting(self, coro):
if self._task_in_progress("installation_task"):
if self.installation_task:
self.installation_task.cancel()
await self.installation_task
if self._task_in_progress("deleting_task"):
logger.warning("Deleting is already running")
return
self.deleting_task = asyncio.create_task(coro)
@recurring_check(
check_lock,
check_period_first=True,
check_lock_period=DAY,
lock_file=LOCK_FILE,
)
async def refresh_auth_files(self):
if is_secret_expired():
await rotate_secret()
await plugin.update_auth_everywhere()
@recurring_check(
check_lock,
check_period_first=True,
check_lock_period=1 * 60, # Run every 1 minute
lock_file=SITE_PROCESSING_LOCK_FILE,
)
async def process_wordpress_sites(self):
"""
Periodic task for WordPress site file processing.
Runs every minute to:
1. Process changelog.php files written by the WordPress plugin (rule disable/enable from WP admin)
2. Collect incident files written by the WordPress plugin
"""
logger.debug(
"Processing rule disable changelogs"
" and collecting WordPress CVE protection incidents"
)
try:
sites = get_installed_sites()
if not sites:
logger.debug(
"No WordPress sites found for periodic processing"
)
return
# Process changelogs (rule disable/enable from WordPress admin)
affected_sites = (
await self.changelog_processor.process_changelogs_for_sites(
sites, self._sink
)
)
if affected_sites:
await update_disabled_rules_on_sites(
domains=[s.domain for s in affected_sites]
)
# Collect incidents
incidents = (
await self.incident_collector.collect_incidents_for_sites(
sites,
delete_after_processing=True,
)
)
await self.incident_sender.send_incidents(self._sink, incidents)
delete_old_wordpress_incidents(days=30)
except Exception as e:
logger.error("Error in WordPress periodic processing: %s", e)
async def _install_on_new_sites(self):
"""Install plugin on new WordPress sites."""
# Clear any previously tracked sites
self.freshly_installed_sites.clear()
async def install_and_track():
installed_sites = await plugin.install_everywhere(sink=self._sink)
if installed_sites:
self.freshly_installed_sites.update(installed_sites)
return installed_sites
await self.process_installation(
install_and_track(),
for_new_sites=True,
)
async def _tidy_up(self):
"""Tidy up sites from which the WordPress plugin was deleted manually."""
await plugin.tidy_up_manually_deleted(
sink=self._sink,
freshly_installed_sites=self.freshly_installed_sites,
)
await plugin.fix_data_file_permissions_everywhere(sink=self._sink)
if not Wordpress.SECURITY_PLUGIN_ENABLED:
# If the feature is disabled, remove the WordPress plugin from all sites.
await self.process_deleting(
plugin.remove_all_installed(sink=self._sink)
)
self.installation_completed = False
async def _adopt_found_sites(self):
"""Adopt sites where plugin is installed but not tracked in our database."""
adopted_sites = await plugin.adopt_found_sites(sink=self._sink)
# Add adopted sites to freshly_installed_sites to prevent them from being
# marked as manually deleted by tidy_up (AVD database may not be updated yet)
if adopted_sites:
self.freshly_installed_sites.update(adopted_sites)
async def _update_existing(self):
"""Update plugin on all sites where it is installed."""
await plugin.update_everywhere(sink=self._sink)
async def _run_install_and_update(self):
"""
Combined operation: install on new sites, adopt found sites, tidy up,
and update existing plugins.
This runs all operations sequentially to avoid race conditions.
"""
# Install plugin on new sites.
await self._install_on_new_sites()
# Wait for installation to complete before proceeding.
if self.installation_task:
await self.installation_task
# Adopt sites where plugin is installed but not in our database.
await self._adopt_found_sites()
# Tidy up and update.
await self._tidy_up()
await self._update_existing()
@expect(MessageType.WordpressPluginAction)
async def manage_plugin_action(self, message):
logger.info(
"ImunifySecurityPlugin received message action: %s method: %s",
message.action,
message.method,
)
# Check if install_and_update is running - it blocks all other actions
if self._task_in_progress("install_and_update_task"):
logger.warning(
"Install-and-update is still running, skipping action %s",
message.action,
)
return
if message.action == "install_on_new_sites":
if not self.installation_completed:
# The installation is not completed yet. We cannot know reliably which sites are new.
return
await self._install_on_new_sites()
return
if self._task_in_progress("installation_task"):
logger.warning(
"Installation is still running, skipping action %s",
message.action,
)
return
if self._task_in_progress("deleting_task"):
logger.warning(
"Uninstallation is already running, skipping action %s",
message.action,
)
return
if message.action == "update_existing":
await self._update_existing()
return
if message.action == "tidy_up":
await self._tidy_up()
return
if message.action == "install_and_update":
if not self.installation_completed:
# The installation is not completed yet. We cannot know reliably which sites are new.
logger.warning(
"Installation is not completed yet, skipping"
" install_and_update"
)
return
# Run install_and_update as a background task to prevent blocking.
# Note: No need to check if already running - the check at the top of this function
# (line 182) already handles that case.
self.install_and_update_task = asyncio.create_task(
self._run_install_and_update()
)
@expect(MessageType.ConfigUpdate)
async def manage_plugin_installation(self, message):
if not isinstance(message["conf"], SystemConfig):
return
current_config_value = Wordpress.SECURITY_PLUGIN_ENABLED
if current_config_value == self.last_config_value:
return
# Update last config value immediately to prevent multiple installations
self.last_config_value = current_config_value
if current_config_value and not self.installation_completed:
# Feature was enabled in config, install the WordPress plugin on all sites.
await self.process_installation(
plugin.install_everywhere(sink=self._sink)
)
self.installation_completed = True
elif not current_config_value and self.installation_completed:
# Feature was disabled in config, remove the WordPress plugin from all sites.
await self.process_deleting(
plugin.remove_all_installed(sink=self._sink)
)
self.installation_completed = False
save_state(
"ImunifySecurityPlugin",
{
"installed": self.installation_completed,
"enabled": current_config_value,
},
)
@expect(HookEvent.MalwareCleanupFinished)
async def handle_malware_cleanup_finished(self, message):
"""
INFO [2025-02-24 12:00:20,384] imav.plugins.wordpress: Malware cleanup finished:
HookEvent.MalwareCleanupFinished(
{
'cleanup_id': 'fa4fe7e48dbf45588f53b24366cd8893',
'started': 1740398411.786418,
'error': None,
'total_files': 3,
'total_cleaned': 3,
'status': 'ok'
}
)
"""
# Skip if plugin is disabled
if not self.last_config_value:
return
# Leave early if status is not ok or the started time is missing.
if message.get("status") != "ok" or not message.get("started"):
return
# load all malware hits cleaned since the cleanup started
hits = _get_cleaned_malware_hits(message["started"])
site_paths = set()
# Collect all site paths that need to be updated.
for hit in hits:
if hit.resource_type == "file":
try:
user_info = pwd.getpwnam(hit.user)
user_sites = get_sites_for_user(user_info)
uid = ( # In None cases there also no user_sites, so it wouldn't be used
user_info.pw_uid if user_info else None
)
for site_path in user_sites:
if hit.orig_file.startswith(site_path):
site_paths.add((site_path, uid))
break
except KeyError:
pass
if not site_paths:
logger.debug("Cleanup finished => no sites found for cleaned hits")
return
logger.info(
"Cleanup finished => %s site(s) need to be updated",
len(site_paths),
)
# Convert paths to WPSite objects with empty domain and update data on the sites that need to be updated.
# We need to work with paths here because sometimes the domain is not set, see https://cloudlinux.atlassian.net/browse/DEF-32238.
wordpress_sites = [
WPSite(docroot=site_path, domain="", uid=uid)
for site_path, uid in site_paths
]
await plugin.update_data_on_sites(self._sink, wordpress_sites)
logger.info("%s site(s) updated after a cleanup", len(wordpress_sites))
@expect(HookEvent.MalwareScanningFinished)
async def handle_malware_scan_finished(self, message):
"""
INFO [2025-02-24 11:57:17,968] imav.plugins.wordpress: Malware scan finished:
HookEvent.MalwareScanningFinished(
{
'scan_id': 'b9bd136aff0a4d87a248c859cfe41c47',
'scan_type': 'user',
'path': '/home/user1'
}
)
INFO [2025-02-24 12:00:10,740] imav.plugins.wordpress: Malware scan finished:
HookEvent.MalwareScanningFinished(
{
'scan_id': 'a74271d2cdd04e0c9bd49ef6de23e0d8',
'scan_type': 'user',
'path': '/home/user4',
'started': 1740398383,
'total_files': 39229,
'total_malicious': 3,
'error': None,
'status': 'ok',
'scan_params': {'intensity_cpu': 2, 'intensity_io': 2, 'intensity_ram': 2048, 'initiator': None, 'file_patterns': None, 'exclude_patterns': None, 'follow_symlinks': False, 'detect_elf': True},
'stats': {'scan_time': 27, 'mem_peak': 28217344, 'smart_time_hs': 0.004, 'scan_time_hs': 1.1751, 'smart_time_preg': 0, 'scan_time_preg': 2.7391, 'finder_time': 13.5896, 'cas_time': 0.7562, 'deobfuscate_time': 0.8998, 'total_files': 39229}
}
)
"""
# Skip if plugin is disabled
if not self.last_config_value:
return
# Leave early if status is not ok or path or stats are missing.
if (
message.get("status") != "ok"
or not message.get("path")
or not message.get("stats")
):
return
# Malware scan is finished, figure out what sites need to be updated based on the path.
path = message["path"]
sites = get_sites_by_path(path)
if not sites:
logger.debug("Scan finished => no sites found for path=%s", path)
return
# Update data on the sites that need to be updated.
logger.info(
"Scan finished => %s site(s) need to be updated", len(sites)
)
await plugin.update_data_on_sites(self._sink, sites)
logger.info("%s site(s) updated after a scan", len(sites))