Admin

name	type	size	perms	modified	actions
[parent directory]
__pycache__/	dir	-	0755	2026-09-01 07:52:27
__init__.py	file	1,260 B	0644	2026-09-01 07:52:27	edit
bot_protection.py	file	4,315 B	0644	2026-09-01 07:52:27	edit
changelog_processor.py	file	12,927 B	0644	2026-09-01 07:52:27	edit
cli.py	file	10,968 B	0644	2026-09-01 07:52:27	edit
constants.py	file	346 B	0644	2026-09-01 07:52:27	edit
exception.py	file	94 B	0644	2026-09-01 07:52:27	edit
incident_collector.py	file	21,292 B	0644	2026-09-01 07:52:27	edit
incident_parser.py	file	5,076 B	0644	2026-09-01 07:52:27	edit
incident_sender.py	file	10,841 B	0644	2026-09-01 07:52:27	edit
plugin.py	file	75,849 B	0644	2026-09-01 07:52:27	edit
proxy_auth.py	file	5,192 B	0644	2026-09-01 07:52:27	edit
site_repository.py	file	18,403 B	0644	2026-09-01 07:52:27	edit
telemetry.py	file	541 B	0644	2026-09-01 07:52:27	edit
utils.py	file	22,674 B	0644	2026-09-01 07:52:27	edit
wp_rules.py	file	3,524 B	0644	2026-09-01 07:52:27	edit

editing: incident_collector.py

"""Collector for WordPress CVE protection incidents."""

import logging
import os
import pwd
import stat as stat_module
import time
import re
from pathlib import Path
from collections import defaultdict

from defence360agent.model.wordpress import WPSite
from defence360agent.wordpress.cli import get_data_dir
from defence360agent.wordpress.incident_parser import IncidentFileParser
from defence360agent.model.wordpress_incident import (
    aggregate_incident_dicts,
    bulk_create_wordpress_incidents,
    build_incident_dict,
    country_reader,
)

logger = logging.getLogger(__name__)

#: Marks a file taken aside for processing. The plugin keeps appending to a
#: freshly created file under the original name, so nothing written during the
#: batch is lost, and a file left behind is retried on the next cycle. The
#: .php extension stays last: a webserver that only hands *.php to the
#: interpreter would serve any other extension as readable text.
PROCESSING_SUFFIX = ".processing.php"

#: A batch this old has failed every cycle since it was set aside. Retiring it
#: stops the retry from repeating forever and unblocks the same-hour file it
#: would otherwise keep out of collection.
QUARANTINE_AFTER_SECONDS = 15 * 60

#: Terminal names for a batch the collector must not pick up again. Neither
#: is matched by the pattern, and both keep .php last for the same reason
#: PROCESSING_SUFFIX does. A stored batch says so: its incidents are safe.
FAILED_SUFFIX = ".failed.php"
STORED_SUFFIX = ".stored.php"

class IncidentRateLimiter:
    """
    Rate limiter to prevent DoS attacks via incident flooding.

Implements per-rule-per-IP rate limiting as per spec:
    - Maximum 100 incidents for each rule from the same IP within 15 minutes

Memory-optimized implementation with bounded entry count using LRU eviction.
    """

def __init__(
        self,
        max_incidents_per_rule_per_ip: int = 100,
        time_window_seconds: int = 900,  # 15 minutes
        max_unique_entries: int = 10000,  # Limit total unique (rule_id, IP) combinations
    ):
        """
        Initialize the rate limiter.

Args:
            max_incidents_per_rule_per_ip: Max incidents per rule per IP (default: 100)
            time_window_seconds: Time window in seconds (default: 900 = 15 minutes)
            max_unique_entries: Max unique (rule_id, IP) combinations to track (default: 10000)
        """
        self.max_per_rule_per_ip = max_incidents_per_rule_per_ip
        self.time_window = time_window_seconds
        self.max_unique_entries = max_unique_entries

# Track incident timestamps: {(rule_id, ip): [timestamp1, timestamp2, ...]}
        self.incident_times = defaultdict(list)

self.cleanup_interval = 60  # Clean up old records every minute
        self.last_cleanup = time.time()

def _cleanup_old_records(self):
        """Remove records older than the time window and enforce max entries limit."""
        now = time.time()
        cutoff = now - self.time_window

# Clean expired timestamps from all entries
        keys_to_delete = []
        for key, timestamps in self.incident_times.items():
            # Filter out timestamps older than the window
            recent = [ts for ts in timestamps if ts > cutoff]

if recent:
                self.incident_times[key] = recent
            else:
                keys_to_delete.append(key)

for key in keys_to_delete:
            del self.incident_times[key]

# Enforce max unique entries limit using LRU eviction
        if len(self.incident_times) > self.max_unique_entries:
            # Find oldest entries (those with oldest timestamp)
            entries_by_age = sorted(
                self.incident_times.items(),
                key=lambda x: x[1][0] if x[1] else 0,
            )

# Remove oldest 10% of entries to avoid frequent evictions
            num_to_remove = max(
                1,
                len(self.incident_times) - int(self.max_unique_entries * 0.9),
            )
            for key, _ in entries_by_age[:num_to_remove]:
                del self.incident_times[key]

logger.warning(
                "Rate limiter exceeded max entries (%d), removed %d oldest"
                " entries",
                self.max_unique_entries,
                num_to_remove,
            )

self.last_cleanup = now

def check_rate_limit(
        self, rule_id: str, attacker_ip: str, pending: int = 0
    ) -> tuple[bool, str]:
        """
        Check if adding an incident would exceed rate limits.

Args:
            rule_id: Rule identifier
            attacker_ip: IP address of the attacker
            pending: Incidents already accepted in the current batch but not
                recorded yet, so one file cannot exceed the limit on its own

Returns:
            Tuple of (allowed: bool, reason: str)
        """
        # Periodic cleanup
        if time.time() - self.last_cleanup > self.cleanup_interval:
            self._cleanup_old_records()

now = time.time()
        cutoff = now - self.time_window
        key = (rule_id, attacker_ip)

# Lazy cleanup: remove expired entries on access
        if key in self.incident_times:
            timestamps = self.incident_times[key]
            # Filter out old timestamps
            recent = [ts for ts in timestamps if ts > cutoff]

if recent:
                self.incident_times[key] = recent
                recent_count = len(recent)
            else:
                # All timestamps expired, remove entry
                del self.incident_times[key]
                recent_count = 0
        else:
            recent_count = 0

# Check if limit exceeded
        recent_count += pending
        if recent_count >= self.max_per_rule_per_ip:
            window_minutes = self.time_window // 60
            return (
                False,
                (
                    f"Rate limit exceeded for rule {rule_id} from IP"
                    f" {attacker_ip}:"
                    f" {recent_count}/{self.max_per_rule_per_ip} within"
                    f" {window_minutes} minutes"
                ),
            )

return True, "OK"

def record_incident(self, rule_id: str, attacker_ip: str):
        """
        Record that an incident was added.

Args:
            rule_id: Rule identifier
            attacker_ip: IP address
        """
        now = time.time()
        key = (rule_id, attacker_ip)

# Create list if it doesn't exist, or append to existing
        if key not in self.incident_times:
            self.incident_times[key] = [now]
        else:
            # Limit list size to prevent unbounded growth
            timestamps = self.incident_times[key]
            if len(timestamps) >= self.max_per_rule_per_ip:
                # Remove oldest timestamp when at limit
                timestamps.pop(0)
            timestamps.append(now)

class IncidentCollector:
    """
    Collect and persist WordPress incidents from plugin incident files.
    """

def __init__(self, rate_limiter: IncidentRateLimiter | None = None):
        """
        Initialize the incident collector.

Args:
            rate_limiter: Optional rate limiter (creates default if not provided)
        """
        self.rate_limiter = rate_limiter or IncidentRateLimiter()
        self.parser = IncidentFileParser()
        #: Batches this process read and failed to clear. The agent idles out
        #: after minutes of quiet, so age alone would retire one that was
        #: never retried.
        self._failed: set[str] = set()

async def collect_incidents_for_site(
        self,
        site: WPSite,
        delete_after_processing: bool = True,
    ) -> list:
        """
        Collect incidents from a single WordPress site.

Args:
            site: WordPress site to collect incidents from
            ruleset_version: Version of the ruleset being used
            delete_after_processing: Whether to delete incident files after processing

Returns:
            List of collected Incident objects
        """
        collected_incidents = []

try:
            data_dir = await get_data_dir(site)
            logger.debug("Data directory for site %s: %s", site, data_dir)
            if not data_dir.exists():
                logger.debug("Data directory does not exist for site %s", site)
                return []

incident_files = self._get_incident_files(data_dir)
            logger.debug(
                "Incident files for site %s: %s", site, incident_files
            )
            if not incident_files:
                logger.debug("No incident files found for site %s", site)
                return []

logger.debug(
                "Found %d incident file(s) for site %s",
                len(incident_files),
                site,
            )

username = self._get_site_username(site)

for incident_file in incident_files:
                file_incidents = await self._process_file(
                    incident_file,
                    site,
                    username,
                    delete_after_processing,
                )
                collected_incidents.extend(file_incidents)

except Exception as e:
            logger.error(
                "Error collecting incidents for site %s: %s",
                site,
                e,
            )

logger.info(
            "Collected %d incident(s) for site %s",
            len(collected_incidents),
            site,
        )

return collected_incidents

async def collect_incidents_for_sites(
        self,
        sites: list[WPSite],
        delete_after_processing: bool = True,
    ) -> list:
        """
        Collect incidents from multiple WordPress sites.

Args:
            sites: List of WordPress sites
            delete_after_processing: Whether to delete incident files after processing

Returns:
            List of collected Incident objects
        """
        all_collected_incidents = []

for site in sites:
            site_incidents = await self.collect_incidents_for_site(
                site,
                delete_after_processing,
            )
            all_collected_incidents.extend(site_incidents)

if all_collected_incidents:
            logger.info(
                "Collected %d WordPress incident(s) from %d site(s)",
                len(all_collected_incidents),
                len(sites),
            )

return all_collected_incidents

@classmethod
    def _get_incident_files(cls, data_dir: Path) -> list[Path]:
        """
        Get all incident files in the incidents directory.

Args:
            data_dir: Path to the imunify-security data directory

Returns:
            List of incident file paths
        """
        incidents_dir = data_dir / "incidents"
        logger.debug(
            "Incidents directory for site %s: %s", data_dir, incidents_dir
        )
        if not incidents_dir.exists() or not incidents_dir.is_dir():
            logger.debug(
                "Incidents directory does not exist for site %s", data_dir
            )
            return []

# Use lstat (not Path.is_file) to identify regular files without following symlinks.
        incident_files = []
        for f in incidents_dir.iterdir():
            try:
                st = os.lstat(f)
            except OSError:
                continue
            if stat_module.S_ISREG(st.st_mode) and cls._is_incident_file(f):
                incident_files.append(f)

logger.debug(
            "Incident files for site %s: %s", data_dir, incident_files
        )
        return incident_files

# Pattern for incident files: yyyy-mm-dd-hh.php, optionally taken aside
    _FILE_PATTERN = re.compile(
        r"^\d{4}-\d{2}-\d{2}-\d{2}(?:\.processing)?\.php$"
    )

@classmethod
    def _is_incident_file(cls, file_path: Path) -> bool:
        """
        Check if a file is an incident file based on naming pattern.

Args:
            file_path: Path to the file to check

Returns:
            True if file matches pattern yyyy-mm-dd-hh.php, with or without
            the suffix marking a batch left behind by an earlier cycle
        """
        return bool(cls._FILE_PATTERN.match(file_path.name))

async def _process_file(
        self,
        incident_file,
        site: WPSite,
        username: str | None,
        delete_after_processing: bool,
    ) -> list:
        try:
            if delete_after_processing:
                incident_file = self._take_aside(incident_file)
                if incident_file is None:
                    return []

incidents = self.parser.parse_file(incident_file)

if incidents is None:
                self._failed.add(str(incident_file))
                return []

if not incidents:
                logger.warning(
                    "No valid incidents in file %s",
                    incident_file.name,
                )
                if delete_after_processing:
                    self._discard(incident_file)

return []

logger.debug(
                "Parsed %d incident(s) from %s for site %s",
                len(incidents),
                incident_file.name,
                site,
            )

collected_incidents = self._process_file_incidents(
                incidents,
                site,
                username,
                incident_file.name,
            )

if delete_after_processing:
                self._discard(incident_file)

return collected_incidents

except Exception as e:
            if delete_after_processing:
                self._failed.add(str(incident_file))
            logger.error(
                "Error processing incident file %s for site %s: %s",
                incident_file.name,
                site,
                e,
            )
            return []

def _take_aside(self, incident_file: Path) -> Path | None:
        """Move the file out of the plugin's way before reading it.

Returns None when an earlier batch is still pending under the aside
        name; that batch is processed in its own turn and the fresh file waits
        for the next cycle rather than overwriting it.
        """
        if incident_file.name.endswith(PROCESSING_SUFFIX):
            if self._quarantine(incident_file):
                return None
            return incident_file

aside = incident_file.with_name(
            incident_file.name[: -len(".php")] + PROCESSING_SUFFIX
        )
        if aside.exists() and self._pending(aside):
            if not self._quarantine(aside):
                return None

incident_file.rename(aside)
        # rename keeps the plugin's mtime, so stamp the file to date the
        # attempt rather than the last write to it.
        os.utime(aside, None, follow_symlinks=False)
        return aside

@staticmethod
    def _pending(aside: Path) -> bool:
        """Whether an aside still holds a batch waiting to be stored.

Anything the site put there that is not a regular file is not one,
        and the rename replaces it.
        """
        try:
            st = os.lstat(aside)
        except OSError:
            return False
        return stat_module.S_ISREG(st.st_mode) and st.st_size > 0

def _retire(self, path: Path, suffix: str) -> Path | None:
        """Give a batch a name the collector will not pick up again.

Renaming touches the name, never what it points at, so it stays safe
        in a directory the site owns.
        """
        stem = path.name
        for known in (PROCESSING_SUFFIX, ".php"):
            if stem.endswith(known):
                stem = stem[: -len(known)]
                break

retired = path.with_name(stem + suffix)
        try:
            path.rename(retired)
        except OSError as e:
            logger.error("Failed to retire %s: %s", path.name, e)
            return None

self._failed.discard(str(path))
        return retired

def _quarantine(self, aside: Path) -> bool:
        """Retire an aside this process has read and still failed to clear."""
        if str(aside) not in self._failed:
            return False

try:
            age = time.time() - os.lstat(aside).st_mtime
        except OSError:
            return True

if age < QUARANTINE_AFTER_SECONDS:
            return False

retired = self._retire(aside, FAILED_SUFFIX)
        if retired is None:
            return False

logger.error(
            "Gave up on %s after %d seconds, kept it as %s",
            aside.name,
            age,
            retired.name,
        )
        return True

def _discard(self, incident_file: Path) -> bool:
        """Delete a stored batch, reporting whether it is gone.

Emptying one we cannot delete would mean writing through a path the
        site owns, so it is retired under a name we never collect instead.
        """
        try:
            incident_file.unlink(missing_ok=True)
            self._failed.discard(str(incident_file))
            return True
        except OSError as e:
            logger.error("Failed to delete %s: %s", incident_file.name, e)

# Its incidents are stored: reading the file again would add to their
        # counts, so retire it now rather than leaving it to be picked up.
        retired = self._retire(incident_file, STORED_SUFFIX)
        if retired is None:
            self._failed.add(str(incident_file))
        else:
            logger.warning(
                "Could not delete %s, kept the stored batch as %s",
                incident_file.name,
                retired.name,
            )
        return False

def _get_site_username(self, site: WPSite) -> str | None:
        try:
            user_info = pwd.getpwuid(site.uid)
            return user_info.pw_name
        except Exception as e:
            logger.error(
                "Failed to get username for uid=%d, site %s: %s",
                site.uid,
                site,
                e,
            )
            return None

def _process_file_incidents(
        self,
        incidents: list[dict],
        site: WPSite,
        username: str | None,
        incident_file_name: str,
    ) -> list:
        incidents_to_insert = []
        accepted: defaultdict = defaultdict(int)
        dropped_count = 0

# Prepare all incidents for bulk insertion
        with country_reader() as geo_reader:
            for incident in incidents:
                rule_id = incident.get("rule_id", "unknown")
                attacker_ip = incident.get("REMOTE_ADDR") or incident.get(
                    "attacker_ip", "unknown"
                )

allowed, reason = self.rate_limiter.check_rate_limit(
                    rule_id,
                    attacker_ip,
                    pending=accepted[(rule_id, attacker_ip)],
                )

if not allowed:
                    logger.warning(
                        "Rate limit exceeded for site %s: %s",
                        site,
                        reason,
                    )
                    dropped_count += 1
                    continue

# Prepare incident data for bulk insert
                site_info = {
                    "domain": site.domain,
                    "site_path": site.docroot,
                    "username": username,
                    "user_id": site.uid,
                }
                incident_data = build_incident_dict(
                    incident, site_info, geo_reader=geo_reader
                )

incidents_to_insert.append(incident_data)
                accepted[(rule_id, attacker_ip)] += 1

if not incidents_to_insert:
            logger.info(
                "Processed file %s: 0 stored, 0 aggregated, %d dropped",
                incident_file_name,
                dropped_count,
            )
            return []

aggregated = aggregate_incident_dicts(incidents_to_insert)

try:
            bulk_create_wordpress_incidents(aggregated)
        except Exception:
            logger.error(
                "Failed to store %d incident(s) from %s, keeping the file"
                " for the next cycle",
                len(incidents_to_insert),
                incident_file_name,
                exc_info=True,
            )
            raise

# Only a stored incident spends rate-limit budget; a kept file is
        # retried next cycle and must not be throttled away unstored.
        for (rule_id, attacker_ip), count in accepted.items():
            for _ in range(count):
                self.rate_limiter.record_incident(rule_id, attacker_ip)

logger.info(
            "Processed file %s: %d stored, %d aggregated, %d dropped",
            incident_file_name,
            len(aggregated),
            len(incidents_to_insert) - len(aggregated),
            dropped_count,
        )

return aggregated

cancel