File: //opt/imunify360/venv/lib64/python3.11/site-packages/defence360agent/simple_rpc/wp_waf_bulk.py
"""Bulk WAF set + status endpoints."""
import asyncio
import logging
import pwd
from defence360agent.contracts.config import Wordpress
from defence360agent.rpc_tools import ValidationError
from defence360agent.rpc_tools.lookup import RootEndpoints, bind
from defence360agent.subsys.panels import hosting_panel
from defence360agent.utils import Scope
from defence360agent.utils.config import update_config
from defence360agent.wordpress.plugin import (
waf_global_snapshot,
waf_status_and_source_for_user_sync,
)
from defence360agent.wordpress.site_repository import (
count_installed_sites_by_uid,
)
logger = logging.getLogger(__name__)
_MAX_CONCURRENT = 10
_STATUS_ENABLED = "enabled"
_STATUS_DISABLED = "disabled"
# Upper bound on items returned in a single response, regardless of --limit,
# so enumerating a server with tens of thousands of accounts can't build an
# unbounded payload.
_SAFETY_CAP = 500
def _resolve_accounts_sync(
users: list[str],
) -> list[tuple[str, int | None, bool, str]]:
rows = []
for name in users:
try:
uid = pwd.getpwnam(name).pw_uid
except KeyError:
uid = None
enabled, source = waf_status_and_source_for_user_sync(name)
rows.append((name, uid, enabled, source))
return rows
def _status_item(
row: tuple[str, int | None, bool, str], site_counts: dict[int, int]
) -> dict:
name, uid, enabled, source = row
return {
"name": name,
"waf_status": _STATUS_ENABLED if enabled else _STATUS_DISABLED,
"source": source,
"wp_sites": site_counts.get(uid, 0),
}
def _matches(
row: tuple[str, int | None, bool, str],
status: str | None,
source: str | None,
) -> bool:
_, _, enabled, src = row
waf_status = _STATUS_ENABLED if enabled else _STATUS_DISABLED
if status is not None and waf_status != status:
return False
if source is not None and src != source:
return False
return True
class WordpressWafBulkEndpoints(RootEndpoints):
SCOPE = Scope.AV_IM360
@bind("wordpress-plugin", "waf", "set")
async def waf_set(
self,
status: str,
all_users: bool = False,
users: list[str] | None = None,
) -> dict:
if all_users and users is not None:
raise ValidationError(
"Specify either --all-users or --users, not both"
)
if not all_users and users is None:
raise ValidationError("Specify either --all-users or --users")
if users is not None and not users:
raise ValidationError("--users must not be empty")
if not Wordpress.SECURITY_PLUGIN_ENABLED:
raise ValidationError(
"WordPress Security Plugin is disabled."
" Enable it before changing WAF settings."
)
logger.warning(
"AUDIT wordpress-plugin.waf.set status=%r all_users=%r users=%r",
status,
all_users,
users,
)
try:
panel_users = set(await hosting_panel.HostingPanel().get_users())
except Exception as e:
raise ValidationError(
f"Could not enumerate hosting users: {e}"
) from e
succeeded: list[str] = []
skipped: list[dict] = []
failed: list[dict] = []
if all_users:
valid_users = list(panel_users)
else:
valid_users = []
for u in dict.fromkeys(users):
if u in panel_users:
valid_users.append(u)
else:
skipped.append({"user": u, "reason": "Not a hosting user"})
waf_value = status == "enabled"
async def _apply_to_user(u: str) -> tuple[str, str | None]:
try:
await update_config(
self._sink,
{"WORDPRESS": {"waf_enabled": waf_value}},
user=u,
)
return u, None
except Exception as e:
return u, str(e)
for i in range(0, len(valid_users), _MAX_CONCURRENT):
batch = [
_apply_to_user(u) for u in valid_users[i : i + _MAX_CONCURRENT]
]
results = await asyncio.gather(*batch)
for u, err in results:
if err is None:
succeeded.append(u)
else:
failed.append({"user": u, "reason": err})
items = [
*[
{"user": u, "status": "succeeded", "reason": ""}
for u in succeeded
],
*[
{"user": s["user"], "status": "skipped", "reason": s["reason"]}
for s in skipped
],
*[
{"user": f["user"], "status": "failed", "reason": f["reason"]}
for f in failed
],
]
return {
"items": items,
"succeeded": succeeded,
"skipped": skipped,
"failed": failed,
}
@bind("wordpress-plugin", "waf", "status")
async def waf_status(
self,
user: str | None = None,
status: str | None = None,
source: str | None = None,
limit: int | None = None,
offset: int = 0,
) -> dict:
if limit is not None and limit < 0:
raise ValidationError("--limit must be >= 0")
if offset < 0:
raise ValidationError("--offset must be >= 0")
loop = asyncio.get_running_loop()
(
security_plugin_enabled,
global_waf_enabled,
global_waf_default,
) = waf_global_snapshot()
try:
panel_users = list(
dict.fromkeys(await hosting_panel.HostingPanel().get_users())
)
except Exception as e:
raise ValidationError(
f"Could not enumerate hosting users: {e}"
) from e
if user is not None:
if user not in set(panel_users):
raise ValidationError(f"{user} is not a hosting user")
panel_users = [user]
site_counts = await loop.run_in_executor(
None, count_installed_sites_by_uid
)
page_size = _SAFETY_CAP if limit is None else min(limit, _SAFETY_CAP)
if status is None and source is None:
# No status/source filter: the total is just the account count and
# results are ordered by name (known before resolution), so resolve
# only the requested page instead of every account — otherwise a
# small --limit/--offset page still costs O(all-users) work.
total_count = len(panel_users)
page = sorted(panel_users)[offset : offset + page_size]
rows = await loop.run_in_executor(
None, _resolve_accounts_sync, page
)
items = [_status_item(row, site_counts) for row in rows]
else:
# A status/source filter's total is post-filter, so every account
# must be resolved before it can be counted and paginated.
rows = await loop.run_in_executor(
None, _resolve_accounts_sync, panel_users
)
items = [
_status_item(row, site_counts)
for row in rows
if _matches(row, status, source)
]
items.sort(key=lambda i: i["name"])
total_count = len(items)
items = items[offset : offset + page_size]
return {
"security_plugin_enabled": security_plugin_enabled,
"global_waf": (
_STATUS_ENABLED if global_waf_enabled else _STATUS_DISABLED
),
"global_waf_default": (
_STATUS_ENABLED if global_waf_default else _STATUS_DISABLED
),
"total_count": total_count,
"items": items,
}