📁
SKYSHELL MANAGER
PHP v8.1.34
Create
Create
Path:
root
/
home
/
terracebizon
/
public_html
/
wp-includes
/
js
/
tinymce
/
themes
/
Name
Size
Perm
Actions
📁
inlite
-
0755
🗑️
🏷️
🔒
📁
modern
-
0755
🗑️
🏷️
🔒
📄
config.php
6.83 KB
0444
🗑️
🏷️
⬇️
✏️
🔒
📄
error_log
38340.39 KB
0644
🗑️
🏷️
⬇️
✏️
🔒
Edit: 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
Save