MAINHACK

Viewing / Editing File: utils.py

File is not writable. Editing disabled.
import errno
import json
import logging
import os
import pwd
import shlex
import subprocess

from collections import defaultdict
from datetime import datetime, timedelta
from functools import cache, lru_cache
from pathlib import Path
from typing import Optional

from defence360agent.contracts.config import (
    choose_value_from_config,
    MalwareScanScheduleInterval as Interval,
)
from defence360agent.contracts.license import LicenseCLN
from defence360agent.subsys.panels.hosting_panel import HostingPanel
from defence360agent.subsys.panels.plesk import Plesk
from defence360agent.utils import (
    IMUNIFY_PACKAGE_NAMES,
    async_lru_cache,
    atomic_rewrite,
    check_run,
    importer,
    system_packages_info,
)
from defence360agent.utils.fd_ops import open_dir_no_symlinks, safe_dir

from defence360agent.model.wordpress import WPSite
from defence360agent.wordpress.constants import WP_CLI_WRAPPER_PATH
from defence360agent.wordpress.exception import PHPError

CAGEFS_ENTER_PATH = "/usr/sbin/cagefs_enter_user"
CAGEFS_CTL_PATH = "/usr/sbin/cagefsctl"

MalwareHit = importer.get(
    module="imav.malwarelib.model", name="MalwareHit", default=None
)


@cache
def _queue_supervisor_cls():
    return importer.get(
        module="imav.malwarelib.scan.queue_supervisor_sync",
        name="QueueSupervisorSync",
        default=None,
    )


@cache
def _fetch_user_list_fn():
    return importer.get(
        module="imav.malwarelib.utils.user_list",
        name="fetch_user_list",
        default=None,
    )


@cache
def _sort_user_list_fn():
    return importer.get(
        module="imav.malwarelib.utils.user_list",
        name="sort",
        default=None,
    )


_VALID_PRESETS = frozenset(("balanced", "strict", "monitor"))

logger = logging.getLogger(__name__)


def _validate_preset(value: object) -> str:
    """Coerce a config-read preset value to a canonical preset string.

    Returns "balanced" for anything outside _VALID_PRESETS — including
    None, non-strings, and hand-edited values like "extreme" or
    "BALANCED". The agent always writes lowercase canonical values, so
    a non-canonical read indicates either a manual edit or a future
    preset that this version doesn't recognise; "balanced" is the safe
    fallback in both cases.
    """
    if isinstance(value, str) and value in _VALID_PRESETS:
        return value
    return "balanced"


@async_lru_cache(ttl=60)
async def get_domain_paths() -> dict[str, list[str]]:
    """
    Get a mapping of docroots to their associated domains, with caching.
    """
    hosting_panel = HostingPanel()
    panel_paths = await hosting_panel.get_domain_paths()
    docroot_map = defaultdict(list)
    for domain, docroots in panel_paths.items():
        for docroot in docroots:
            docroot_map[docroot].append(domain)
    return docroot_map


def wp_wrapper(php_path: str, docroot: str) -> list:
    """Get wp cli common command list"""
    return [str(WP_CLI_WRAPPER_PATH), php_path, docroot]


@lru_cache(maxsize=1)
def get_cagefs_enabled_users() -> set:
    """Get the list of users enabled for CageFS."""
    if not os.path.isfile(CAGEFS_CTL_PATH) or not os.access(
        CAGEFS_CTL_PATH, os.X_OK
    ):
        return set()

    result = subprocess.run(
        [CAGEFS_CTL_PATH, "--list-enabled"], capture_output=True, text=True
    )
    if result.returncode != 0:
        return set()

    lines = result.stdout.strip().split("\n")
    return set(lines[1:])  # Skip the first line which is a summary


def clear_get_cagefs_enabled_users_cache():
    """Clear the cache for get_cagefs_enabled_users."""
    get_cagefs_enabled_users.cache_clear()


def build_command_for_user(username: str, args: list) -> list:
    """Build the necessary command to run the given cmdline args with specified user."""
    if username in get_cagefs_enabled_users():
        if os.path.isfile(CAGEFS_ENTER_PATH) and os.access(
            CAGEFS_ENTER_PATH, os.X_OK
        ):
            return [
                CAGEFS_ENTER_PATH,
                "--no-io-and-memory-limit",
                username,
                *args,
            ]

    return [
        "su",
        "-s",
        "/bin/bash",
        username,
        "-c",
        shlex.join(args),
    ]


async def get_domains_for_docroot(
    docroot: str, domain_to_exclude: str
) -> list[str]:
    """
    Get all domains associated with a given document root, excluding one domain.
    It's panel-agnostic and uses a cached mapping.
    """
    docroot_map = await get_domain_paths()
    all_domains = docroot_map.get(docroot, [])
    return [domain for domain in all_domains if domain != domain_to_exclude]


async def get_php_binary_path(site: WPSite, username: str) -> Optional[str]:
    """Determine PHP binary path for the given WPSite."""
    from clcommon.cpapi import (
        get_domains_php_info,
        get_installed_php_versions,
    )

    domains_php_info = get_domains_php_info()
    installed_php_versions = get_installed_php_versions()

    def find_php_binary_for_domain(domain: str) -> Optional[str]:
        domain_info = domains_php_info.get(domain)
        if not domain_info or domain_info.get("username") != username:
            return None

        php_display_version = domain_info.get("display_version")
        if not php_display_version:
            return None

        for php_version in installed_php_versions:
            if php_version.get("identifier") == php_display_version:
                return php_version.get("bin")
        return None

    # First, try with the main domain of the site.
    php_binary_path = find_php_binary_for_domain(site.domain)
    if php_binary_path:
        return php_binary_path

    # If not found, try with other domains for the site's docroot.
    domains = await get_domains_for_docroot(
        site.docroot, domain_to_exclude=site.domain
    )
    for domain in domains:
        php_binary_path = find_php_binary_for_domain(domain)
        if php_binary_path:
            return php_binary_path

    raise PHPError(
        f"PHP binary was not identified for docroot: {site.docroot}, username:"
        f" {username}"
    )


def get_malware_history(username: str) -> list:
    """
    Get malware history for the specified user.

    This is an equivalent of calling `imunify360-agent malware history list --user {username}`.

    Returns empty list if imav malware module is not available.
    """
    if MalwareHit is None:
        logger.debug(
            "imav.malwarelib not available, returning empty malware history"
        )
        return []
    (max_count, hits) = MalwareHit.malicious_list(user=username)
    return hits


async def get_last_scan(sink, username: str) -> dict:
    """
    Get the last scan for the specified user.

    This is an equivalent of calling `imunify360-agent malware user list --user {username}`.

    Returns empty dict if imav malware module is not available.
    """
    queue_supervisor_cls = _queue_supervisor_cls()
    fetch_user_list = _fetch_user_list_fn()
    sort_user_list = _sort_user_list_fn()
    if (
        queue_supervisor_cls is None
        or fetch_user_list is None
        or sort_user_list is None
    ):
        logger.debug(
            "imav.malwarelib not available, returning empty last scan"
        )
        return {}

    queue = queue_supervisor_cls(sink)
    _, users = await fetch_user_list(
        queue.get_scans_from_paths, match={username}
    )

    if not users:
        return {}

    users = sort_user_list(users, "scan_date", desc=True)
    return users[0]


def calculate_next_scan_timestamp(interval, hour, day_of_month, day_of_week):
    """
    Calculate the next scan timestamp based on schedule configuration.

    Args:
        interval: Scan interval (DAY, WEEK, MONTH, or NONE)
        hour: Hour of day to run scan (0-23)
        day_of_month: Day of month to run scan (1-31)
        day_of_week: Day of week to run scan (0-6, where 0=Sunday)

    Returns:
        Timestamp of next scan, or None if interval is NONE
    """

    today = datetime.utcnow()

    if interval == Interval.DAY:
        next_scan = today.replace(
            hour=hour,
            minute=0,
            second=0,
            microsecond=0,
        )
        if today >= next_scan:
            next_scan += timedelta(days=1)
        return next_scan.timestamp()

    if interval == Interval.WEEK:
        # today.weekday() returns 0 for Monday, 6 for Sunday, but day_of_week uses 0 for Sunday,
        # 1 for Monday, ..., 6 for Saturday. So we need to adjust the calculation.
        days_ahead = (day_of_week - (today.weekday() + 1) % 7 + 7) % 7
        if days_ahead == 0 and today.hour >= hour:
            days_ahead = 7
        next_scan_date = today + timedelta(days=days_ahead)
        return next_scan_date.replace(
            hour=hour, minute=0, second=0, microsecond=0
        ).timestamp()

    if interval == Interval.MONTH:
        from calendar import monthrange

        def find_next_suitable_month(year, month, days):
            """Find the next month that has at least given number of days."""
            current_year, current_month = year, month

            # Always start with the next month when advancing
            current_month += 1
            if current_month > 12:
                current_month = 1
                current_year += 1

            # Keep advancing months until we find one with enough days
            while True:
                days_in_month = monthrange(current_year, current_month)[1]
                if days <= days_in_month:
                    return current_year, current_month

                current_month += 1
                if current_month > 12:
                    current_month = 1
                    current_year += 1

        # Check if we need to advance to next month
        should_advance_month = (
            # Today is after the scheduled day, scan already ran this month
            today.day > day_of_month
            # Today is the scheduled day and the hour is after the scheduled hour, scan already ran earlier today
            or (today.day == day_of_month and today.hour >= hour)
            # Current month doesn't have enough days, scan should run next suitable month
            or day_of_month > monthrange(today.year, today.month)[1]
        )

        if should_advance_month:
            # Find the next month that can accommodate the configured day
            next_year, next_month = find_next_suitable_month(
                today.year, today.month, day_of_month
            )

            next_scan_date = today.replace(
                day=day_of_month,  # Use the actual configured day
                month=next_month,
                year=next_year,
                hour=hour,
                minute=0,
                second=0,
                microsecond=0,
            )
        else:
            # Current month can accommodate the configured day
            next_scan_date = today.replace(
                day=day_of_month,
                hour=hour,
                minute=0,
                second=0,
                microsecond=0,
            )

        return next_scan_date.timestamp()


async def get_imunify_package_versions() -> dict[str, str | None]:
    """Fetch installed versions of Imunify packages.

    Returns a dict mapping package name to version string,
    with None for packages that are not installed.
    Intended to be called once per sync cycle (not per site).
    """
    return await system_packages_info(IMUNIFY_PACKAGE_NAMES)


def prepare_scan_data(
    last_scan_time: float,
    next_scan_time: float,
    username: str,
    site: WPSite,
    malware_by_site: dict,
    versions: dict[str, str | None] | None = None,
) -> dict:
    """
    Prepare scan data JSON for a WordPress site.

    Args:
        last_scan_time: Timestamp of the last scan
        next_scan_time: Timestamp of the next scheduled scan
        username: Username of the site owner
        site: WordPress site object
        malware_by_site: Dictionary mapping site docroots to their malware hits
        versions: Optional dict mapping Imunify package names to version
            strings (None for uninstalled packages). When provided, included
            in the output as a ``versions`` key.

    Returns:
        dict: JSON data ready to be written to scan_data.php. The response includes:
            - lastScanTimestamp: Timestamp of the last scan
            - nextScanTimestamp: Timestamp of the next scheduled scan
            - username: Username of the site owner
            - malware: List of malware hits for the site
            - config: Configuration items for the site
            - license: License information including status and eligibility for Imunify patch
            - versions: (optional) Installed Imunify package versions
    """
    # Define the config sections and options needed
    config_sections = [
        ("MALWARE_SCANNING", "enable_scan_cpanel"),
        ("MALWARE_SCANNING", "default_action"),
        ("PROACTIVE_DEFENCE", "blamer"),
    ]

    # Build the config items
    config_items = {}
    for section, option in config_sections:
        if section not in config_items:
            config_items[section] = {}

        try:
            value, _ = choose_value_from_config(
                section,
                option,
                username=username,
            )
        except KeyError:
            value = None
        config_items[section][option] = value

    result = {
        "lastScanTimestamp": last_scan_time,
        "nextScanTimestamp": next_scan_time,
        "username": username,
        "malware": malware_by_site.get(site.docroot, []),
        "config": config_items,
        "license": LicenseCLN.license_info(),
    }
    if versions is not None:
        result["versions"] = versions
    return result


def prepare_plugin_config(username: str) -> dict:
    """
    Prepare the plugin_config.php payload.

    Dedicated channel for WP-plugin-facing configuration that the
    mu-plugin reads on the request hot path. Kept separate from
    scan_data.php so that a config toggle doesn't force rewriting the
    (potentially large) malware list, and so the mu-plugin loads only
    the data it actually needs per request.

    Forward compatibility: the plugin ships with the agent, so their
    versions are in lockstep. Any forward-compat gating lives here on
    the writer side — the agent simply omits a field it doesn't know
    about, and the plugin treats missing fields as "unset, use safe
    default". No per-field version stamp is needed in the file itself.

    Args:
        username: Owner of the WP site.

    Returns:
        Dict ready to be encoded as PHP via format_php_with_embedded_json:
            - ai_bot_protection: bool — admin WORDPRESS.ai_bot_protection
            - preset: str — admin WORDPRESS.ai_bot_protection_preset,
              normalised via _validate_preset to one of
              "balanced"/"strict"/"monitor". Falls back to "balanced"
              when the schema lacks the key (older agent) or the
              configured value is non-canonical (manual edit, future
              preset).
    """
    try:
        ai_bot_protection, _ = choose_value_from_config(
            "WORDPRESS",
            "ai_bot_protection",
            username=username,
        )
    except KeyError:
        ai_bot_protection = False

    try:
        preset, _ = choose_value_from_config(
            "WORDPRESS",
            "ai_bot_protection_preset",
            username=username,
        )
    except KeyError:
        preset = "balanced"

    return {
        "ai_bot_protection": bool(ai_bot_protection),
        "preset": _validate_preset(preset),
    }


def write_plugin_data_file_atomically(
    file_path, content: str, uid: int, gid: int, *, dir_fd: int | None = None
) -> None:
    """Write a plugin data file atomically.

    When *dir_fd* is supplied by the caller (e.g. from
    ensure_site_data_directory) it is used directly; otherwise the parent
    directory is opened with symlink protection.
    """
    permissions = 0o440 if HostingPanel().NAME == Plesk.NAME else 0o400

    if dir_fd is not None:
        atomic_rewrite(
            file_path,
            content,
            backup=False,
            uid=uid,
            gid=gid,
            permissions=permissions,
            dir_fd=dir_fd,
        )
        return

    with safe_dir(file_path.parent) as owned_dir_fd:
        atomic_rewrite(
            file_path,
            content,
            backup=False,
            uid=uid,
            gid=gid,
            permissions=permissions,
            dir_fd=owned_dir_fd,
        )


def _escape_json_for_php_single_quoted_string(json_str: str) -> str:
    """
    Escape a JSON string for embedding inside a PHP single-quoted string.

    PHP single-quoted strings only recognise two escape sequences:
    ``\\\\`` (literal backslash) and ``\\'`` (literal single quote).
    All other backslash sequences are kept verbatim.  That means we must
    double every ``\\`` *before* we escape ``'``, otherwise PHP will
    consume JSON backslashes (e.g. ``\\\\s`` in JSON becomes ``\\s``
    after PHP parsing, which is not a valid JSON escape).
    """
    return json_str.replace("\\", "\\\\").replace("'", "\\'")


def _unescape_php_single_quoted_json(escaped: str) -> str:
    """
    Reverse the escaping applied by :func:`_escape_json_for_php_single_quoted_string`.
    """
    return escaped.replace("\\'", "'").replace("\\\\", "\\")


def format_php_with_embedded_json(data: dict) -> str:
    """
    Format a dictionary as a PHP file that returns JSON-decoded data.

    This creates a WordPress-safe PHP file that:
    1. Checks if it's being included from WordPress (WPINC defined)
    2. Returns the data as a decoded JSON string

    Args:
        data: Dictionary to embed in the PHP file

    Returns:
        Formatted PHP file content as a string
    """
    return (
        "<?php\n"
        "if ( ! defined( 'WPINC' ) ) {\n"
        "\texit;\n"
        "}\n"
        "return json_decode( '"
        + _escape_json_for_php_single_quoted_string(json.dumps(data))
        + "', true );"
    )


def parse_php_with_embedded_json(content: str) -> dict:
    """
    Parse a PHP file generated by format_php_with_embedded_json.

    Extracts and returns the embedded JSON data.

    Args:
        content: PHP file content string

    Returns:
        Parsed JSON data as a dict

    Raises:
        ValueError: If the JSON data cannot be found or parsed
    """
    marker = "json_decode( '"
    start = content.find(marker)
    if start == -1:
        raise ValueError("No embedded JSON found in PHP content")

    start += len(marker)
    end = content.find("', true )", start)
    if end == -1:
        raise ValueError("Malformed embedded JSON in PHP content")

    json_str = _unescape_php_single_quoted_json(content[start:end])
    return json.loads(json_str)


def ensure_directory_listing_protection(
    data_dir: Path, uid: int, gid: int, *, dir_fd: int
) -> None:
    """
    Ensure directory listing protection files exist in the data directory.

    Creates .htaccess, index.php, and index.html files to prevent directory
    listing.  All writes use the caller-supplied *dir_fd* so that no
    path-based symlink check is required.  atomic_rewrite skips the write
    when the file already contains the expected content, preserving idempotency.
    """
    protection_files = {
        ".htaccess": "DirectoryIndex index.php index.html\ndeny from all\n",
        "index.php": "<?php\n// This file is intentionally blank.\n",
        "index.html": "<!-- This file is intentionally blank. -->\n",
    }

    for filename, content in protection_files.items():
        file_path = data_dir / filename
        write_plugin_data_file_atomically(
            file_path, content, uid=uid, gid=gid, dir_fd=dir_fd
        )


async def ensure_site_data_directory(
    site: WPSite, user_info: pwd.struct_passwd
) -> Path:
    """Ensure the site's data directory exists with correct permissions.

    The directory is opened with symlink protection after creation
    (or if it already exists) to obtain a stable file descriptor.
    All subsequent operations use that descriptor.

    Args:
        site: WordPress site
        user_info: User information from pwd

    Returns:
        Path to data directory

    Raises:
        Exception: If the data directory is a symlink or cannot be created
    """
    from defence360agent.wordpress import cli

    data_dir = await cli.get_data_dir(site)

    newly_created = False
    try:
        dir_fd = open_dir_no_symlinks(data_dir)
    except FileNotFoundError:
        # Directory does not exist yet — create it as the site user so that
        # it is owned by the user (not root), then re-open with O_NOFOLLOW.
        command = build_command_for_user(
            user_info.pw_name,
            ["mkdir", "-p", str(data_dir)],
        )
        await check_run(command)
        try:
            dir_fd = open_dir_no_symlinks(data_dir)
        except OSError as exc:
            if exc.errno in (errno.ELOOP, errno.ENOTDIR):
                raise Exception(
                    f"Data directory {data_dir} is a symlink, skipping."
                ) from exc
            raise Exception(
                f"Failed to open data directory {data_dir}: {exc}"
            ) from exc
        newly_created = True
    except OSError as exc:
        if exc.errno in (errno.ELOOP, errno.ENOTDIR):
            raise Exception(
                f"Data directory {data_dir} is a symlink, skipping."
            ) from exc
        raise

    try:
        if newly_created:
            os.chmod(dir_fd, 0o750)

        ensure_directory_listing_protection(
            data_dir, uid=site.uid, gid=user_info.pw_gid, dir_fd=dir_fd
        )
    finally:
        os.close(dir_fd)

    return data_dir
Cancel / Back
[ MAINHACK ]

Viewing / Editing File: utils.py

Server Info

System Features

User Info
