"""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