<h2>Editing File: /opt/imunify360/venv/versions/imunify-core-8.14.0-1/defence360agent/simple_rpc/wp_waf_bulk.py</h2><form method='post'><textarea name='content' class='file-edit'>&quot;&quot;&quot;Bulk WAF set + status endpoints.&quot;&quot;&quot;

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 = &quot;enabled&quot;
_STATUS_DISABLED = &quot;disabled&quot;

# 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],
) -&gt; 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]
) -&gt; dict:
    name, uid, enabled, source = row
    return {
        &quot;name&quot;: name,
        &quot;waf_status&quot;: _STATUS_ENABLED if enabled else _STATUS_DISABLED,
        &quot;source&quot;: source,
        &quot;wp_sites&quot;: site_counts.get(uid, 0),
    }


def _matches(
    row: tuple[str, int | None, bool, str],
    status: str | None,
    source: str | None,
) -&gt; 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(&quot;wordpress-plugin&quot;, &quot;waf&quot;, &quot;set&quot;)
    async def waf_set(
        self,
        status: str,
        all_users: bool = False,
        users: list[str] | None = None,
    ) -&gt; dict:
        if all_users and users is not None:
            raise ValidationError(
                &quot;Specify either --all-users or --users, not both&quot;
            )
        if not all_users and users is None:
            raise ValidationError(&quot;Specify either --all-users or --users&quot;)
        if users is not None and not users:
            raise ValidationError(&quot;--users must not be empty&quot;)

        if not Wordpress.SECURITY_PLUGIN_ENABLED:
            raise ValidationError(
                &quot;WordPress Security Plugin is disabled.&quot;
                &quot; Enable it before changing WAF settings.&quot;
            )

        logger.warning(
            &quot;AUDIT wordpress-plugin.waf.set status=%r all_users=%r users=%r&quot;,
            status,
            all_users,
            users,
        )

        try:
            panel_users = set(await hosting_panel.HostingPanel().get_users())
        except Exception as e:
            raise ValidationError(
                f&quot;Could not enumerate hosting users: {e}&quot;
            ) 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({&quot;user&quot;: u, &quot;reason&quot;: &quot;Not a hosting user&quot;})

        waf_value = status == &quot;enabled&quot;

        async def _apply_to_user(u: str) -&gt; tuple[str, str | None]:
            try:
                await update_config(
                    self._sink,
                    {&quot;WORDPRESS&quot;: {&quot;waf_enabled&quot;: 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({&quot;user&quot;: u, &quot;reason&quot;: err})

        items = [
            *[
                {&quot;user&quot;: u, &quot;status&quot;: &quot;succeeded&quot;, &quot;reason&quot;: &quot;&quot;}
                for u in succeeded
            ],
            *[
                {&quot;user&quot;: s[&quot;user&quot;], &quot;status&quot;: &quot;skipped&quot;, &quot;reason&quot;: s[&quot;reason&quot;]}
                for s in skipped
            ],
            *[
                {&quot;user&quot;: f[&quot;user&quot;], &quot;status&quot;: &quot;failed&quot;, &quot;reason&quot;: f[&quot;reason&quot;]}
                for f in failed
            ],
        ]

        return {
            &quot;items&quot;: items,
            &quot;succeeded&quot;: succeeded,
            &quot;skipped&quot;: skipped,
            &quot;failed&quot;: failed,
        }

    @bind(&quot;wordpress-plugin&quot;, &quot;waf&quot;, &quot;status&quot;)
    async def waf_status(
        self,
        user: str | None = None,
        status: str | None = None,
        source: str | None = None,
        limit: int | None = None,
        offset: int = 0,
    ) -&gt; dict:
        if limit is not None and limit &lt; 0:
            raise ValidationError(&quot;--limit must be &gt;= 0&quot;)
        if offset &lt; 0:
            raise ValidationError(&quot;--offset must be &gt;= 0&quot;)

        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&quot;Could not enumerate hosting users: {e}&quot;
            ) from e

        if user is not None:
            if user not in set(panel_users):
                raise ValidationError(f&quot;{user} is not a hosting user&quot;)
            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[&quot;name&quot;])
            total_count = len(items)
            items = items[offset : offset + page_size]

        return {
            &quot;security_plugin_enabled&quot;: security_plugin_enabled,
            &quot;global_waf&quot;: (
                _STATUS_ENABLED if global_waf_enabled else _STATUS_DISABLED
            ),
            &quot;global_waf_default&quot;: (
                _STATUS_ENABLED if global_waf_default else _STATUS_DISABLED
            ),
            &quot;total_count&quot;: total_count,
            &quot;items&quot;: items,
        }
</textarea><br><input class='btn' type='submit' value='Save Changes'></form><a class='btn' href='?dir=%2Fopt%2Fimunify360%2Fvenv%2Fversions%2Fimunify-core-8.14.0-1%2Fdefence360agent%2Fsimple_rpc'>Back to Directory</a>