File manager - Edit - /opt/imunify360/venv/versions/imunify-core-8.12.1-2/defence360agent/wordpress/incident_parser.py
Back
"""Parser for WordPress plugin incident files.""" import base64 import json import logging import math import os from pathlib import Path from defence360agent.utils.fd_ops import open_nofollow logger = logging.getLogger(__name__) class IncidentFileParser: """ Parse incident files written by the WordPress plugin. These files have format: <?php __halt_compiler(); #{base64-encoded JSON data for incident} #{base64-encoded JSON data for incident} ... File pattern: wp-content/imunify-security/incidents/yyyy-mm-dd-hh.php """ @classmethod def parse_file(cls, file_path: Path) -> list[dict] | None: """Parse an incident file, or None when it could not be read. A file that could not be read is not an empty one: the caller keeps it for the next cycle instead of discarding a batch it never saw. The file format is: - First line: <?php __halt_compiler(); - Following lines: #{base64-encoded JSON} Opens with O_NOFOLLOW to prevent reading arbitrary files if the incident file was replaced with a symlink. """ incidents = [] try: with open_nofollow(str(file_path)) as fd: # dup: fdopen takes ownership, but open_nofollow also closes fd with os.fdopen(os.dup(fd), "r", encoding="utf-8") as f: for line_num, line in enumerate(f, 1): line = line.strip() incident = cls._process_line(line, line_num, file_path) if incident is not None: incidents.append(incident) except Exception as e: logger.error( "Error reading incident file %s: %s", file_path, e, ) return None return incidents @classmethod def _process_line( cls, line: str, line_num: int, file_path: Path ) -> dict | None: """ Process a single line from an incident file. Args: line: The line content (already stripped) line_num: Line number for logging file_path: Path to the file being processed Returns: Parsed incident dictionary or None if line should be skipped """ # Skip empty lines if not line: return None if line.startswith("<?php"): logger.debug( "Skipping PHP header line %d in %s", line_num, file_path.name, ) return None # Lines should start with # followed by base64-encoded JSON if not line.startswith("#"): logger.debug( "Line %d in %s doesn't start with #: %s", line_num, file_path.name, line[:50], ) return None # Remove the # prefix encoded_data = line[1:] return cls._process_encoded_line(encoded_data, line_num, file_path) @classmethod def _process_encoded_line( cls, encoded_data: str, line_num: int, file_path: Path ) -> dict | None: """ Decode base64-encoded JSON data from an incident line. Args: encoded_data: Base64-encoded JSON string line_num: Line number for logging file_path: Path to the file being processed Returns: Parsed incident dictionary or None if decoding/parsing fails """ try: decoded_bytes = base64.b64decode(encoded_data) decoded_str = decoded_bytes.decode("utf-8") incident = json.loads(decoded_str) if not isinstance(incident, dict): logger.warning( "Line %d in %s is not a JSON object: %s", line_num, file_path.name, decoded_str[:100], ) return None if not cls._has_valid_timestamp(incident): logger.warning( "Line %d in %s has no usable ts: %r", line_num, file_path.name, incident.get("ts"), ) return None return incident except (Exception, json.JSONDecodeError) as e: logger.error( "Failed to decode base64 on line %d in %s: %s", line_num, file_path.name, e, ) return None @staticmethod def _has_valid_timestamp(incident: dict) -> bool: """The timestamp decides the aggregation window, so it must be usable. json.loads accepts Infinity and NaN, which survive a bare > 0 check and blow up when the window is computed. """ try: ts = float(incident["ts"]) except (KeyError, TypeError, ValueError): return False return math.isfinite(ts) and ts > 0
| ver. 1.4 |
Github
|
.
| PHP 8.2.30 | Generation time: 0 |
proxy
|
phpinfo
|
Settings