<h2>Editing File: /opt/imunify360/venv/versions/imunify-core-8.14.0-1/defence360agent/wordpress/incident_sender.py</h2><form method='post'><textarea name='content' class='file-edit'>&quot;&quot;&quot;Send WordPress incidents to the correlation server.&quot;&quot;&quot;

import hashlib
import itertools
import json
import logging
import uuid
from datetime import datetime
from functools import partial
from time import monotonic
from types import MappingProxyType
from typing import Any, Mapping

from defence360agent.contracts.messages import SensorWordpressIncidentList
from defence360agent.contracts.plugins import MessageSink
from defence360agent.internals import delivery_ack
from defence360agent.model.wordpress_incident import (
    get_unsent_wordpress_incidents,
    settle_wordpress_incidents_reported,
)

logger = logging.getLogger(__name__)


class IncidentSender:
    &quot;&quot;&quot;
    Send WordPress incidents to the correlation server.

    WordPress incidents are already in the Incident table (visible to UI).
    This class sends them to correlation via Reportable messages, which are
    handled by the SendToServer/SendToServerNATS/SendToServerFGW plugins.

    Those plugins queue a message rather than deliver it, and a send round
    can lose its batch or be force-cancelled mid-publish while the agent
    shuts down. Each incident therefore keeps a count of the occurrences the
    transport has not acknowledged, and every collection cycle sends whatever
    is still outstanding.
    &quot;&quot;&quot;

    # rows one cycle takes on, so a backlog is paged instead of read whole;
    # the byte budget, not this, is what bounds a single message
    MAX_INCIDENTS_PER_CYCLE = 1000
    # how long to wait for an acknowledgement before assuming the round was
    # lost; also paces retries, since a batch nobody acknowledges is re-sent
    # at most once per timeout
    ACK_TIMEOUT = 300

    def __init__(self) -&gt; None:
        # message_id -&gt; ({incident id: occurrences it reported}, deadline)
        self._inflight: dict[str, tuple[dict[int, int], float]] = {}

    def _prepare_incident_for_correlation(
        self, incident: dict
    ) -&gt; dict[str, Any]:
        &quot;&quot;&quot;
        Prepare an incident for sending to the correlation server.

        WordPress incidents use extra_info JSON field to store plugin-specific data.

        Args:
            incident: WordpressIncident dictionary (with extra_info populated)

        Returns:
            Dictionary formatted for correlation server
        &quot;&quot;&quot;
        logger.info(&quot;Preparing incident for correlation: %s&quot;, incident)
        # JSONField automatically deserializes to dict, fallback to empty dict if None
        extra: dict = incident.get(&quot;extra_info&quot;) or {}

        # Convert timestamp to int and format date
        timestamp_value: float = float(incident.get(&quot;timestamp&quot;) or 0)
        timestamp = int(timestamp_value)
        dt = (
            datetime.fromtimestamp(timestamp_value).strftime(&quot;%Y-%m-%d&quot;)
            if timestamp_value
            else &quot;&quot;
        )

        return {
            &quot;timestamp&quot;: timestamp,
            &quot;dt&quot;: dt,
            &quot;plugin_id&quot;: incident.get(&quot;plugin&quot;),
            &quot;rule&quot;: incident.get(&quot;rule&quot;) or &quot;unknown&quot;,
            &quot;name&quot;: incident.get(&quot;name&quot;),
            &quot;message&quot;: incident.get(&quot;description&quot;),
            &quot;severity&quot;: incident.get(&quot;severity&quot;),
            &quot;attackers_ip&quot;: incident.get(&quot;abuser&quot;) or &quot;&quot;,
            &quot;domain&quot;: incident.get(&quot;domain&quot;) or &quot;&quot;,
            # the occurrences this message reports, not the row's running
            # total: re-reporting occurrences correlation already counted
            # inflates the heuristics that consume this field
            &quot;retries&quot;: incident.get(&quot;unsent_retries&quot;) or 1,
            &quot;uri&quot;: extra.get(&quot;request_uri&quot;) or &quot;&quot;,
            &quot;user_agent&quot;: extra.get(&quot;http_user_agent&quot;) or &quot;&quot;,
            &quot;http_method&quot;: extra.get(&quot;request_method&quot;) or &quot;&quot;,
            &quot;user_logged_in&quot;: extra.get(&quot;user_logged_in&quot;) == &quot;true&quot;
            if extra.get(&quot;user_logged_in&quot;)
            else None,
            &quot;file_path&quot;: extra.get(&quot;site_path&quot;) or &quot;&quot;,
            &quot;user&quot;: extra.get(&quot;username&quot;) or &quot;&quot;,
            &quot;tag&quot;: self._build_tags(extra),
            # Include WordPress-specific fields
            &quot;target&quot;: extra.get(&quot;target&quot;) or &quot;&quot;,
            &quot;slug&quot;: extra.get(&quot;slug&quot;) or &quot;&quot;,
            &quot;version&quot;: extra.get(&quot;version&quot;) or &quot;&quot;,
            &quot;mode&quot;: extra.get(&quot;mode&quot;) or &quot;&quot;,
            &quot;details&quot;: extra,
        }

    def _build_tags(self, extra: dict) -&gt; list[str]:
        tags = [&quot;wordpress&quot;, &quot;cve&quot;]

        if extra.get(&quot;cve&quot;):
            tags.append(extra[&quot;cve&quot;])
        if extra.get(&quot;target&quot;):
            tags.append(f&quot;target_{extra['target']}&quot;)
        if extra.get(&quot;mode&quot;):
            tags.append(f&quot;mode_{extra['mode']}&quot;)

        return tags

    async def send_pending_incidents(self, sink: MessageSink | None) -&gt; int:
        &quot;&quot;&quot;Send every incident the server has not acknowledged.

        Runs once per collection cycle, after the freshly collected
        incidents were stored, so one pass covers both the new rows and the
        ones whose earlier message never left the agent.
        &quot;&quot;&quot;
        if sink is None:
            logger.warning(&quot;No sink provided, skipping incident sending&quot;)
            return 0

        self._expire_inflight()

        incidents = get_unsent_wordpress_incidents(
            limit=self.MAX_INCIDENTS_PER_CYCLE,
            exclude_ids=self._inflight_ids(),
        )

        return await self.send_incidents(sink, incidents)

    async def send_incidents(
        self, sink: MessageSink | None, incidents: list[dict]
    ) -&gt; int:
        &quot;&quot;&quot;
        Send WordPress incidents to the correlation server.

        Since incidents are already in the WordpressIncident table (visible to UI),
        we just need to send them to correlation.

        Args:
            incidents: List of incidents to send

        Returns:
            Number of incidents sent
        &quot;&quot;&quot;
        if sink is None:
            logger.warning(&quot;No sink provided, skipping incident sending&quot;)
            return 0

        if len(incidents) == 0:
            logger.debug(&quot;No incidents to send, skipping&quot;)
            return 0

        logger.info(
            &quot;Sending %d incidents to correlation server&quot;, len(incidents)
        )

        correlation_batch = [
            self._prepare_incident_for_correlation(incident)
            for incident in incidents
        ]
        pending = iter(
            [
                (incident.get(&quot;id&quot;), incident.get(&quot;unsent_retries&quot;) or 1)
                for incident in incidents
            ]
        )
        # minted outside _send_batch so all siblings of one split share it
        group_id = uuid.uuid4().hex
        for batch in SensorWordpressIncidentList.batched(correlation_batch):
            await self._send_batch(
                sink,
                batch,
                {
                    incident_id: occurrences
                    for incident_id, occurrences in itertools.islice(
                        pending, len(batch)
                    )
                    if incident_id is not None
                },
                group_id=group_id,
            )

        return len(correlation_batch)

    async def _send_batch(
        self,
        sink: MessageSink,
        correlation_batch: list[dict],
        reported: Mapping[int, int] = MappingProxyType({}),
        group_id: str | None = None,
    ):
        &quot;&quot;&quot;Send a batch of incidents to correlation server.

        Uses SensorIncidentList Reportable message which is sent to
        correlation via the SendToServer/SendToServerNATS/SendToServerFGW
        plugins.
        &quot;&quot;&quot;
        logger.info(
            &quot;Sending batch of %d incidents to correlation server&quot;,
            len(correlation_batch),
        )
        logger.info(
            &quot;Correlation batch json: %s&quot;,
            json.dumps(correlation_batch, indent=2),
        )
        message = SensorWordpressIncidentList(correlation_batch)
        message.group_id = group_id
        # Pin the id the send path would otherwise generate itself, so its
        # acknowledgement can be tied back to these rows. Deriving it from the
        # rows also makes a re-send carry the id of the message it repeats,
        # which lets the transport's de-duplication window collapse the two.
        message_id = hashlib.sha1(
            json.dumps(
                [sorted(reported.items()), correlation_batch], sort_keys=True
            ).encode()
        ).hexdigest()
        message[&quot;message_id&quot;] = message_id
        self._watch(message_id, reported)
        try:
            await sink.process_message(message)

            logger.info(
                &quot;Queued %d wordpress incident(s) for correlation server&quot;
                &quot; (message %s)&quot;,
                len(correlation_batch),
                message_id,
            )

        except Exception as e:
            # the message never made it into the queue, so stop waiting for
            # an acknowledgement and let the next cycle send it again
            self._unwatch(message_id)
            logger.error(
                &quot;Failed to queue incident batch: %s&quot;,
                e,
            )
            raise

    def _watch(self, message_id: str, reported: Mapping[int, int]) -&gt; None:
        if not reported:
            return
        reported = dict(reported)
        self._inflight[message_id] = (
            reported,
            monotonic() + self.ACK_TIMEOUT,
        )
        delivery_ack.registry.watch(
            message_id,
            partial(self._on_delivered, message_id, reported),
        )

    def _unwatch(self, message_id: str) -&gt; None:
        self._inflight.pop(message_id, None)
        delivery_ack.registry.unwatch(message_id)

    def _on_delivered(self, message_id: str, reported: dict[int, int]) -&gt; None:
        # written here rather than buffered for the next cycle because a
        # send round usually lands during shutdown, and a count kept in
        # memory until then would not survive the restart
        self._inflight.pop(message_id, None)
        settled = settle_wordpress_incidents_reported(reported)
        logger.info(
            &quot;Discounted %d wordpress incident(s) delivered to correlation&quot;,
            settled,
        )

    def _expire_inflight(self) -&gt; None:
        &quot;&quot;&quot;Give up on batches the transport never acknowledged, so their
        incidents become eligible to be sent again.&quot;&quot;&quot;
        now = monotonic()
        expired = [
            message_id
            for message_id, (_, deadline) in self._inflight.items()
            if deadline &lt;= now
        ]
        for message_id in expired:
            reported, _ = self._inflight.pop(message_id)
            delivery_ack.registry.unwatch(message_id)
            logger.warning(
                &quot;No delivery confirmation for %d wordpress incident(s)&quot;
                &quot; (message %s) in %ds, sending them again&quot;,
                len(reported),
                message_id,
                self.ACK_TIMEOUT,
            )

    def _inflight_ids(self) -&gt; set[int]:
        return {
            incident_id
            for reported, _ in self._inflight.values()
            for incident_id in reported
        }
</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%2Fwordpress'>Back to Directory</a>