Gerenciador & Editor PHP

📝 Editando: incident_sender.py

"""Send WordPress incidents to the correlation server."""

import json
import logging
from datetime import datetime
from typing import Any

from defence360agent.contracts.messages import SensorWordpressIncidentList
from defence360agent.contracts.plugins import MessageSink

logger = logging.getLogger(__name__)

class IncidentSender:
    """
    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
    automatically handled by SendToServer/SendToServerFGW plugins.
    """

def _prepare_incident_for_correlation(
        self, incident: dict
    ) -> dict[str, Any]:
        """
        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
        """
        logger.info("Preparing incident for correlation: %s", incident)
        # JSONField automatically deserializes to dict, fallback to empty dict if None
        extra: dict = incident.get("extra_info") or {}

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

return {
            "timestamp": timestamp,
            "dt": dt,
            "plugin_id": incident.get("plugin"),
            "rule": incident.get("rule") or "unknown",
            "name": incident.get("name"),
            "message": incident.get("description"),
            "severity": incident.get("severity"),
            "attackers_ip": incident.get("abuser") or "",
            "domain": incident.get("domain") or "",
            "retries": incident.get("retries") or 1,
            "uri": extra.get("request_uri") or "",
            "user_agent": extra.get("http_user_agent") or "",
            "http_method": extra.get("request_method") or "",
            "user_logged_in": extra.get("user_logged_in") == "true"
            if extra.get("user_logged_in")
            else None,
            "file_path": extra.get("site_path") or "",
            "user": extra.get("username") or "",
            "tag": self._build_tags(extra),
            # Include WordPress-specific fields
            "target": extra.get("target") or "",
            "slug": extra.get("slug") or "",
            "version": extra.get("version") or "",
            "mode": extra.get("mode") or "",
            "details": extra,
        }

def _build_tags(self, extra: dict) -> list[str]:
        tags = ["wordpress", "cve"]

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

return tags

async def send_incidents(
        self, sink: MessageSink | None, incidents: list[dict]
    ) -> int:
        """
        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
        """
        if sink is None:
            logger.warning("No sink provided, skipping incident sending")
            return 0

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

logger.info(
            "Sending %d incidents to correlation server", len(incidents)
        )

correlation_batch = [
            self._prepare_incident_for_correlation(incident)
            for incident in incidents
        ]

await self._send_batch(sink, correlation_batch)

return len(correlation_batch)

async def _send_batch(
        self, sink: MessageSink, correlation_batch: list[dict]
    ):
        """
        Send a batch of incidents to correlation server.

Uses SensorIncidentList Reportable message which is automatically
        sent to correlation via SendToServer/SendToServerFGW plugins.

Args:
            sink: MessageSink to send the batch to
            correlation_batch: Incidents formatted for correlation server
        """
        logger.info(
            "Sending batch of %d incidents to correlation server",
            len(correlation_batch),
        )
        logger.info(
            "Correlation batch json: %s",
            json.dumps(correlation_batch, indent=2),
        )
        try:
            # Send as a Reportable message
            # The SendToServer/SendToServerFGW plugin will handle actual delivery
            await sink.process_message(
                SensorWordpressIncidentList(correlation_batch)
            )

logger.info(
                "Queued %d wordpress incident(s) for correlation server",
                len(correlation_batch),
            )

except Exception as e:
            logger.error(
                "Failed to queue incident batch: %s",
                e,
            )
            raise

[ Cancelar ]

Nome	Permissões	Ações
[ .. Voltar ]
📄 __init__.py	0644	[ EDITAR/VER ]
📁 __pycache__/	0755
📄 changelog_processor.py	0644	[ EDITAR/VER ]
📄 cli.py	0644	[ EDITAR/VER ]
📄 constants.py	0644	[ EDITAR/VER ]
📄 exception.py	0644	[ EDITAR/VER ]
📄 incident_collector.py	0644	[ EDITAR/VER ]
📄 incident_parser.py	0644	[ EDITAR/VER ]
📄 incident_sender.py	0644	[ EDITAR/VER ]
📄 plugin.py	0644	[ EDITAR/VER ]
📄 proxy_auth.py	0644	[ EDITAR/VER ]
📄 site_repository.py	0644	[ EDITAR/VER ]
📄 telemetry.py	0644	[ EDITAR/VER ]
📄 utils.py	0644	[ EDITAR/VER ]
📄 wp_rules.py	0644	[ EDITAR/VER ]

📍 /opt/imunify360/venv/lib/python3.11/site-packages/defence360agent/wordpress

📝 Editando: incident_sender.py