"""Helper functions for WordPress CVE protection incidents.
WordPress incidents are stored in a dedicated wordpress_incident table with
plugin-specific data stored in the extra_info JSON field.
This module provides helper functions to work with WordPress incidents.
Available for both AV and IM360 modes.
"""
import logging
import time
import json
from collections import defaultdict
from contextlib import ExitStack, contextmanager
from datetime import timedelta
from typing import Iterable, Mapping
from peewee import (
EXCLUDED,
SQL,
CharField,
FloatField,
IntegerField,
TextField,
fn as peewee_fn,
)
from playhouse.sqlite_ext import JSONField, fn
from defence360agent.internals import geo
from defence360agent.model import Model, instance
from defence360agent.model.simplification import apply_order_by
from defence360agent.rpc_tools.validate import OrderBy
from defence360agent.utils import CHUNK_SIZE_SQL_QUERY, split_for_chunk
logger = logging.getLogger(__name__)
_UNSET = object()
#: Width of an aggregation window, matching the resident agent's flush interval.
BUCKET_SECONDS = 60
AGGREGATE_KEY = (
"abuser",
"name",
"plugin",
"rule",
"severity",
"domain",
"bucket",
)
INSERT_CHUNK_SIZE = 50
#: Upper bound on stored rows, mirroring the resident agent's incident table.
MAX_STORED_INCIDENTS = 100_000
class WordpressIncident(Model):
"""
WordPress incident model for CVE protection.
Uses dedicated wordpress_incident table created in migration 191.
Repeats of the same attack are aggregated per minute: the unique
constraint on (abuser, name, plugin, rule, severity, domain, bucket)
keeps one row per aggregation window, counted by retries.
"""
id = IntegerField(primary_key=True, null=True)
plugin = CharField(null=True)
rule = CharField(null=True)
timestamp = FloatField(null=True)
retries = IntegerField(null=True)
severity = IntegerField(null=True)
name = CharField(null=True)
description = TextField(null=True)
abuser = CharField(null=True)
country = CharField(null=True, column_name="country_id")
domain = TextField(null=True, default=None)
extra_info = JSONField(null=True)
bucket = IntegerField(null=True)
# occurrences correlation has not acknowledged yet; the periodic task
# re-sends the row until it reaches zero. A counter rather than a flag so
# that occurrences merged into an already-reported row are still reported.
# The DEFAULT is in the schema, not just in peewee, because
# src/rpm-tests/test_wordpress/test_list_incidents.py inserts rows with
# raw SQL that names its columns explicitly.
unsent_retries = IntegerField(
null=False,
default=0,
index=True,
constraints=[SQL("DEFAULT 0")],
)
class Meta:
database = instance.db
db_table = "wordpress_incident"
indexes = ((AGGREGATE_KEY, True),)
def build_extra_info(incident_data: dict, site_info: dict) -> dict:
"""
Build extra_info dict from incident data and site information.
Args:
incident_data: Dict with incident fields from PHP incident file
site_info: Dict with site information (domain, site_path, username, user_id)
Returns:
Dict with all WordPress-specific fields for extra_info JSON column
"""
# Serialize JSON fields
files_json = serialize_json_field(incident_data.get("FILES"))
get_names_json = serialize_json_field(incident_data.get("GET_NAMES"))
post_names_json = serialize_json_field(incident_data.get("POST_NAMES"))
return {
# WordPress plugin-populated fields
"cve": incident_data.get("cve"),
"mode": incident_data.get("mode"),
"target": incident_data.get("target"),
"slug": incident_data.get("slug"),
"version": incident_data.get("version"),
"user_logged_in": incident_data.get("user_logged_in"),
"username": site_info.get("username"),
"user_id": site_info.get("user_id"),
"site_path": site_info.get("site_path"),
# HTTP request details
"request_method": incident_data.get("REQUEST_METHOD"),
"script_filename": incident_data.get("SCRIPT_FILENAME"),
"php_self": incident_data.get("PHP_SELF"),
"path_info": incident_data.get("PATH_INFO"),
"request_uri": incident_data.get("REQUEST_URI"),
"query_string": incident_data.get("QUERY_STRING"),
"http_x_forwarded_for": incident_data.get("HTTP_X_FORWARDED_FOR"),
"http_user_agent": incident_data.get("HTTP_USER_AGENT"),
"http_referer": incident_data.get("HTTP_REFERER"),
# Request data
"files": files_json,
"get_names": get_names_json,
"post_names": post_names_json,
"raw_data": incident_data.get("RAW_DATA"),
}
def build_incident_dict(
incident_data: dict, site_info: dict, geo_reader=_UNSET
) -> dict:
"""
Build complete incident dict ready for database insertion.
This is used for both single incident creation and bulk insertion.
Args:
incident_data: Dict with incident fields from PHP incident file
site_info: Dict with site information (domain, site_path, username, user_id)
geo_reader: An open geo Reader (or None) to resolve the abuser country.
Pass one from country_reader() when building many incidents in a
loop to avoid reopening the mmdb per incident. When omitted, a
short-lived reader is opened for this single call.
Returns:
Dict with all fields ready for Incident.create() or bulk insert
"""
message = incident_data.get("message") or build_message_fallback(
incident_data
)
extra_info = build_extra_info(incident_data, site_info)
abuser_ip = incident_data.get("REMOTE_ADDR") or incident_data.get(
"attacker_ip"
)
if geo_reader is _UNSET:
with country_reader() as reader:
country = _country_code(reader, abuser_ip)
else:
country = _country_code(geo_reader, abuser_ip)
timestamp = float(incident_data.get("ts", 0))
return {
# Standard incident fields
"plugin": "wordpress",
"rule": incident_data.get("rule_id", "unknown"),
"timestamp": timestamp,
"bucket": int(timestamp // BUCKET_SECONDS),
"retries": 1,
"severity": calculate_severity(incident_data.get("mode")),
"name": f"WordPress CVE: {incident_data.get('cve', 'Unknown')}",
"description": message,
"abuser": abuser_ip,
"country": country,
"domain": site_info.get("domain"),
# JSONField automatically handles serialization - just pass the dict
"extra_info": extra_info,
}
@contextmanager
def country_reader():
"""Yield an open geo Reader, or None when the mmdb can't be opened.
Lets bulk callers open the mmap'd reader once instead of per incident,
while keeping enrichment non-blocking when the geo bundle is missing.
"""
with ExitStack() as stack:
try:
reader = stack.enter_context(geo.reader())
except Exception as exc:
logger.debug("GeoIP reader unavailable: %s", exc)
yield None
return
yield reader
def _country_code(reader, ip: str | None) -> str | None:
if reader is None or not ip:
return None
try:
return reader.get_code(ip)
except Exception as exc:
logger.debug("GeoIP lookup failed for %s: %s", ip, exc)
return None
def create_wordpress_incident(
incident_data: dict, site_info: dict
) -> WordpressIncident:
"""
Create a WordPress incident in the wordpress_incident table.
Args:
incident_data: Dict with incident fields from PHP incident file
site_info: Dict with site information (domain, site_path, username)
Returns:
WordpressIncident instance with WordPress fields populated in extra_info
"""
incident_dict = build_incident_dict(incident_data, site_info)
return WordpressIncident.create(**incident_dict)
def aggregate_incident_dicts(incident_dicts: list[dict]) -> list[dict]:
"""Collapse incidents sharing an aggregate key into one counted row."""
# The window comes from the incident's own timestamp, so a backlogged
# file yields the same rows as live collection.
groups: dict[tuple, dict] = {}
for incident_dict in incident_dicts:
key = tuple(incident_dict[field] for field in AGGREGATE_KEY)
group = groups.get(key)
if group is None:
groups[key] = dict(incident_dict)
continue
group["retries"] += incident_dict["retries"]
group["timestamp"] = min(
group["timestamp"], incident_dict["timestamp"]
)
return list(groups.values())
def wordpress_incident_to_dict(incident: WordpressIncident) -> dict:
"""
Convert a WordpressIncident model instance to a dictionary.
Args:
incident: WordpressIncident model instance
Returns:
Dictionary representation of the incident
"""
return {
"id": incident.id,
"plugin": incident.plugin,
"rule": incident.rule,
"timestamp": incident.timestamp,
"retries": incident.retries,
"severity": incident.severity,
"name": incident.name,
"description": incident.description,
"abuser": incident.abuser,
"country": incident.country,
"domain": incident.domain,
"extra_info": incident.extra_info,
}
def get_wordpress_incidents(
limit: int = 1000,
offset: int = 0,
user_id: int | None = None,
by_abuser_ip: str | None = None,
by_country_code: str | None = None,
by_domain: str | None = None,
search: str | None = None,
site_search: str | None = None,
since: int | None = None,
to: int | None = None,
order_by: list | None = None,
include_hidden: bool = False,
):
"""
Get WordPress incidents as dictionaries.
Args:
limit: Maximum number of incidents to return
offset: Offset for pagination
user_id: Filter by user ID (None = all)
by_abuser_ip: Filter by abuser IP address (None = all)
by_country_code: Filter by country code (None = all)
by_domain: Filter by domain (None = all)
search: Search in IP address, name, description, or domain (None = all)
site_search: Filter by site path in extra_info (None = all)
since: Filter by timestamp >= this value (unix timestamp, None = all)
to: Filter by timestamp <= this value (unix timestamp, None = all)
order_by: List of fields to order by (None = default order by timestamp desc).
Can be either strings (e.g., ["timestamp+", "severity-"]) or
OrderBy objects. Strings are automatically converted.
include_hidden: When False (default), exclude incidents whose rule has
the TEST- prefix (internal probe rules that the WordPress
plugin hides from its admin UI).
Returns:
List of incident dictionaries
"""
query = WordpressIncident.select(WordpressIncident).where(
(WordpressIncident.plugin == "wordpress")
)
if not include_hidden:
query = query.where(
WordpressIncident.rule.is_null()
| ~WordpressIncident.rule.startswith("TEST-")
)
if user_id is not None:
query = query.where(
fn.json_extract(WordpressIncident.extra_info, "$.user_id")
== user_id
)
if by_abuser_ip is not None:
query = query.where(WordpressIncident.abuser.contains(by_abuser_ip))
if by_country_code is not None:
query = query.where(WordpressIncident.country == by_country_code)
if by_domain is not None:
query = query.where(WordpressIncident.domain.contains(by_domain))
if search is not None:
query = query.where(
WordpressIncident.name.contains(search)
| WordpressIncident.description.contains(search)
| WordpressIncident.domain.contains(search)
| WordpressIncident.abuser.contains(search)
)
if site_search is not None:
query = query.where(
fn.json_extract(WordpressIncident.extra_info, "$.site_path")
== site_search
)
if since is not None:
query = query.where(WordpressIncident.timestamp.cast("REAL") >= since)
if to is not None:
query = query.where(WordpressIncident.timestamp.cast("REAL") <= to)
# Apply ordering
if order_by is not None:
# Convert string format to OrderBy objects if needed
converted_order_by = []
for item in order_by:
if isinstance(item, str):
converted_order_by.append(OrderBy.fromstring(item))
else:
converted_order_by.append(item)
query = apply_order_by(converted_order_by, WordpressIncident, query)
else:
# Default order by timestamp descending
query = query.order_by(WordpressIncident.timestamp.desc())
query = query.limit(limit)
query = query.offset(offset)
return [wordpress_incident_to_dict(inc) for inc in query.execute()]
def bulk_create_wordpress_incidents(incidents_data: list[dict]) -> int:
"""Store aggregated incidents, merging repeats into the stored row."""
if not incidents_data:
return 0
conflict_target = [
getattr(WordpressIncident, field) for field in AGGREGATE_KEY
]
# every occurrence a fresh row carries is still unreported
rows = [
{**incident, "unsent_retries": incident.get("retries") or 1}
for incident in incidents_data
]
# Every chunk shares one transaction, so a failure leaves the table
# untouched and the batch can be retried without double counting.
with WordpressIncident._meta.database.atomic():
for start in range(0, len(rows), INSERT_CHUNK_SIZE):
chunk = rows[start : start + INSERT_CHUNK_SIZE]
WordpressIncident.insert_many(chunk).on_conflict(
conflict_target=conflict_target,
update={
WordpressIncident.retries: (
WordpressIncident.retries + EXCLUDED.retries
),
# occurrences merged into an already-reported row are
# unreported again, which is what keeps them from being
# swallowed by a row correlation already acknowledged
WordpressIncident.unsent_retries: (
WordpressIncident.unsent_retries + EXCLUDED.retries
),
WordpressIncident.timestamp: peewee_fn.MIN(
WordpressIncident.timestamp, EXCLUDED.timestamp
),
},
).execute()
return len(incidents_data)
def get_unsent_wordpress_incidents(
limit: int,
exclude_ids: Iterable[int] = (),
) -> list[dict]:
"""Get incidents carrying occurrences correlation has not acknowledged
yet, oldest first.
Unlike get_wordpress_incidents() this keeps TEST- rules: they are hidden
from the WordPress admin UI but still belong in correlation.
"""
exclude_ids = set(exclude_ids)
# in-flight rows are skipped in python rather than with a NOT IN: the set
# grows with every unacknowledged batch, and binding thousands of ids per
# cycle costs more than over-fetching by its size
rows = (
WordpressIncident.select(WordpressIncident)
.where(
(WordpressIncident.plugin == "wordpress")
& (WordpressIncident.unsent_retries > 0)
)
.order_by(
WordpressIncident.timestamp.asc(), WordpressIncident.id.asc()
)
.limit(limit + len(exclude_ids))
)
incidents = []
for row in rows:
if row.id in exclude_ids:
continue
# attached here rather than in wordpress_incident_to_dict: the UI
# reads that shape too and has no use for delivery bookkeeping
incidents.append(
{
**wordpress_incident_to_dict(row),
"unsent_retries": row.unsent_retries,
}
)
if len(incidents) == limit:
break
return incidents
def settle_wordpress_incidents_reported(reported: Mapping[int, int]) -> int:
"""Discount the occurrences correlation acknowledged.
Subtracts instead of clearing so that occurrences merged into a row while
its message was in flight stay pending rather than being dropped.
"""
by_amount = defaultdict(list)
for incident_id, amount in reported.items():
if amount > 0:
by_amount[amount].append(incident_id)
settled = 0
# chunked because sqlite caps the variables one query may bind, and the
# acknowledged set is only bounded by how large a batch the sender built
with WordpressIncident._meta.database.atomic():
for amount, incident_ids in by_amount.items():
for chunk in split_for_chunk(
incident_ids, chunk_size=CHUNK_SIZE_SQL_QUERY
):
settled += (
WordpressIncident.update(
unsent_retries=fn.MAX(
0, WordpressIncident.unsent_retries - amount
)
)
.where(WordpressIncident.id.in_(chunk))
.execute()
)
return settled
def delete_old_wordpress_incidents(
days: int, limit: int | None = MAX_STORED_INCIDENTS
):
cutoff_time = time.time() - timedelta(days=days).total_seconds()
is_wordpress = WordpressIncident.plugin == "wordpress"
stale = WordpressIncident.timestamp.cast("REAL") < cutoff_time
# Probing for the oldest row worth keeping costs an indexed lookup; the
# keep-set below matches every row against it, so only run that when the
# probe says the table is actually over the cap.
over_cap = (
limit
and (
WordpressIncident.select(WordpressIncident.timestamp)
.order_by(WordpressIncident.timestamp.desc())
.limit(1)
.offset(limit)
.scalar()
)
is not None
)
if over_cap:
keep = (
WordpressIncident.select(WordpressIncident.id)
.where(~stale)
.order_by(WordpressIncident.timestamp.desc())
.limit(limit)
)
stale |= WordpressIncident.id.not_in(keep)
return WordpressIncident.delete().where(is_wordpress & stale).execute()
def build_message_fallback(incident_data: dict) -> str:
"""Build message if plugin didn't provide one (per spec format)."""
parts = ["IM WP plugin:"]
if incident_data.get("rule_id"):
parts.append(incident_data["rule_id"])
if incident_data.get("cve"):
parts.append(incident_data["cve"])
if incident_data.get("slug"):
parts.append(incident_data["slug"])
if incident_data.get("version"):
parts.append(incident_data["version"])
if incident_data.get("mode"):
parts.append(incident_data["mode"])
return " ".join(parts)
def calculate_severity(mode: str | None) -> int:
"""Calculate severity based on mode."""
if mode == "block":
return 8 # Higher severity for blocked attacks
elif mode == "pass":
return 5 # Medium severity for monitored attacks
else:
return 5 # Default
def serialize_json_field(value) -> str | None:
"""Serialize a value to JSON string if it's not already a string."""
if value is None:
return None
if isinstance(value, str):
return value
return json.dumps(value)