import errno
import ipaddress
import json
import logging
import os
import pwd
import re
import shlex
import subprocess
import time

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,
    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,
    )


@cache
def _iplist_cls():
    """The im360 IPList model, or None on an AV-only install.

    The model binds to core's own ``instance.db``; the agent attaches the
    resident database to that connection at startup, so the model works
    from core without any extra connection."""
    return importer.get(
        module="im360.model.firewall", name="IPList", 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"


def normalize_docroot(docroot: str) -> str:
    """Strip trailing slashes so panel- and WPSite-derived docroots compare equal.

    Panels may report ``document_root`` with a trailing slash; WPSite paths
    never have one. ``"/"`` keeps its slash instead of collapsing to ``""``.
    """
    stripped = docroot.rstrip("/")
    if stripped:
        return stripped
    return "/" if docroot else ""


@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.

    Keys are normalized with normalize_docroot, so callers must normalize
    the path they look up.
    """
    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[normalize_docroot(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(normalize_docroot(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_ai_bot_settings(username: str) -> dict:
    """Per-account effective WORDPRESS settings, without the license read.

    Split from prepare_plugin_config so callers that only need these values
    (e.g. stats collection) don't trigger a license-token read.
    """
    # Local import: plugin imports this module at load time, so the
    # per-account resolver can only be reached lazily from here.
    from defence360agent.wordpress.plugin import (
        _is_ai_bot_protection_enabled_for_user_sync,
        ai_bot_preset_for_user_sync,
    )

    return {
        "ai_bot_protection": _is_ai_bot_protection_enabled_for_user_sync(
            username
        ),
        "preset": ai_bot_preset_for_user_sync(username),
    }


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 — per-account effective status: the
              global kill switch forces off, otherwise the account
              override, otherwise the new-account default.
            - preset: str — per-account effective preset (account override
              else the global preset), normalised to one of
              "balanced"/"strict"/"monitor".
            - license_type: str | None — server license edition
              (imunify360 / imunify360Trial / imunifyAV / imunifyAVPlus),
              or None when it can't be determined; the plugin treats None
              as "do not gate".
    """
    settings = _prepare_ai_bot_settings(username)
    settings["license_type"] = LicenseCLN.get_license_type()
    return settings


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 fetch_manual_whitelist_entries() -> list[tuple[str, int]] | None:
    """Manual, non-expired WHITE ``iplist`` rows as ``(ip, expiration)``.

    Call on the event-loop thread, like every other agent DB read: the
    ``iplist`` table lives in the attached ``resident`` schema, and an
    ATTACH is per connection, so an executor thread's fresh connection
    would not see it. The query is one indexed SELECT (sub-millisecond
    on a 300k-row table).

    Returns None when the IPList model is unavailable (AV-only install),
    otherwise a list in unspecified order. A NULL expiration is mapped to 0
    ("never").

    Only ``manual`` rows are exported: panel-login and CAPTCHA
    auto-whitelisting churn constantly and would cause fleet-wide writes.
    Group-synced entries are manual too and are included.
    """
    IPList = _iplist_cls()
    if IPList is None:
        return None
    # No ORDER BY: sorting on ``ip`` makes SQLite satisfy it through the ip
    # index and scan the whole table (blocklist included) instead of seeking
    # on listname, which stalls this per-tick loop-thread read. Callers do not
    # need order — the grouping keeps the longest-lived value per key and the
    # renderer sorts.
    query = IPList.fetch_non_expired_query(IPList.WHITE).where(
        IPList.manual == True  # noqa: E712 — peewee expression
    )
    return [(row["ip"], row["expiration"] or 0) for row in query.dicts()]


_CIDR_KEY = re.compile(r"[0-9a-f.:/]+")


def bucket_ip_whitelist_entries(
    entries, *, now: float | None = None
) -> tuple[dict[int, dict[str, int]], dict[str, int]]:
    """Validate, normalise and group whitelist entries for ip_whitelist.php.

    Mirrors the plugin's ``SignatureRefresher::bucketRanges()``: a network
    with a prefix of 8 or more goes into the group named by the first byte
    of its network address; anything broader goes into ``ranges_broad``.
    Keys are always ``network/prefix`` (a plain IP becomes /32 or /128),
    rendered from the parsed address and rejected unless they consist of
    CIDR characters only, so no free text can reach PHP source.
    Already-expired entries are dropped; an invalid entry is logged and
    skipped. ``now`` is injectable for tests.

    Returns ``(by_octet, broad)`` with unsorted dicts; the renderer sorts.
    """
    now = int(time.time() if now is None else now)
    by_octet: dict[int, dict[str, int]] = defaultdict(dict)
    broad: dict[str, int] = {}
    for ip, expiration in entries:
        expiration = int(expiration or 0)
        if 0 < expiration <= now:
            continue
        try:
            network = ipaddress.ip_network(str(ip).strip(), strict=False)
        except ValueError:
            logger.warning("Skipping invalid whitelist entry %r", ip)
            continue
        if network.version == 6:
            mapped = network.network_address.ipv4_mapped
            if mapped is not None and network.prefixlen >= 96:
                # The plugin's CidrMatcher demaps an IPv4-mapped client before
                # picking a bucket, so an entry left in the IPv6 (first byte
                # 0x00) bucket would never be searched. Demap it into the
                # client's IPv4 bucket, matching a plain-IPv4 entry.
                network = ipaddress.ip_network(
                    f"{mapped}/{network.prefixlen - 96}", strict=False
                )
        key = f"{network.network_address}/{network.prefixlen}"
        if not _CIDR_KEY.fullmatch(key):
            # ipaddress keeps IPv6 scope ids ("fe80::1%eth0"), which may
            # contain a quote; nothing but CIDR characters may reach PHP.
            logger.warning("Skipping invalid whitelist entry %r", ip)
            continue
        group = (
            broad
            if network.prefixlen < 8
            else by_octet[network.network_address.packed[0]]
        )
        previous = group.get(key)
        if previous is None or (
            previous != 0 and (expiration == 0 or expiration > previous)
        ):
            group[key] = expiration
    return dict(by_octet), broad


def _format_php_ip_whitelist_group(group: dict[str, int], indent: str) -> str:
    quoted = {f"'{key}'": value for key, value in group.items()}
    width = max(len(k) for k in quoted)
    return "".join(
        f"{indent}{key:<{width}} => {quoted[key]},\n" for key in sorted(quoted)
    )


def format_php_ip_whitelist(
    by_octet: dict[int, dict[str, int]], broad: dict[str, int]
) -> str:
    """Render grouped whitelist entries as the ip_whitelist.php literal.

    A plain PHP ``return array(...)`` — compiled once and served from
    OPcache, unlike the embedded-JSON files — with both top-level keys
    always present. Groups are sorted numerically and keys as text, and
    there is no metadata (timestamp, version, count): identical input
    gives identical bytes, which is what lets ``atomic_rewrite`` skip the
    write on every unchanged refresh.
    """
    lines = [
        "<?php\n",
        "if ( ! defined( 'WPINC' ) ) {\n",
        "\texit;\n",
        "}\n",
        "// Auto-generated by imunify360-agent. Do not edit by hand.\n",
        "\n",
        "return array(\n",
    ]
    if by_octet:
        lines.append("\t'ranges_by_octet' => array(\n")
        for octet in sorted(by_octet):
            lines.append(f"\t\t{octet:<3} => array(\n")
            lines.append(
                _format_php_ip_whitelist_group(by_octet[octet], "\t\t\t")
            )
            lines.append("\t\t),\n")
        lines.append("\t),\n")
    else:
        lines.append("\t'ranges_by_octet' => array(),\n")
    if broad:
        lines.append("\t'ranges_broad'    => array(\n")
        lines.append(_format_php_ip_whitelist_group(broad, "\t\t"))
        lines.append("\t),\n")
    else:
        lines.append("\t'ranges_broad'    => array(),\n")
    lines.append(");\n")
    return "".join(lines)


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
) -> Optional[Path]:
    """Create only the ``imunify-security`` leaf inside the site's own
    ``wp-content`` (never the docroot), returning None without creating
    anything when the parent is missing or not owned by the site user."""
    from defence360agent.wordpress import cli

    data_dir = await cli.get_data_dir(site)
    parent_dir = data_dir.parent

    try:
        parent_fd = open_dir_no_symlinks(parent_dir)
    except OSError as exc:
        if exc.errno == errno.ENOENT:
            logger.info(
                "Parent directory %s is missing; skipping data directory"
                " creation for site %s",
                parent_dir,
                site.docroot,
            )
            return None
        if exc.errno == errno.ENOTDIR:
            raise Exception(
                f"Data directory parent {parent_dir} is a symlink or not"
                " a directory."
            ) from exc
        raise

    try:
        # get_data_dir honours the owner-controlled WP_CONTENT_DIR, so the
        # parent can point into another tenant's account. Refuse to create
        # or chown anything unless wp-content is owned by this site's user;
        # checked on the fd (TOCTOU-safe) like _data_dir_belongs_to_user.
        parent_uid = os.fstat(parent_fd).st_uid
        if parent_uid != site.uid:
            logger.warning(
                "wp-content %s for site %s is owned by uid %s, not the"
                " site's uid %s; skipping data-dir creation to avoid"
                " acting in another tenant's account.",
                parent_dir,
                site.docroot,
                parent_uid,
                site.uid,
            )
            return None

        try:
            os.mkdir(data_dir.name, 0o750, dir_fd=parent_fd)
        except FileExistsError:
            pass

        try:
            dir_fd = os.open(
                data_dir.name,
                os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
                dir_fd=parent_fd,
            )
        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
    finally:
        os.close(parent_fd)

    try:
        # Re-assert ownership/mode every call (not only on creation): a
        # prior run may have created the leaf as root and died before the
        # chown/chmod, leaving a root-owned dir the site user cannot enter
        # that later runs would skip. fd is O_NOFOLLOW-opened, so safe.
        os.chown(dir_fd, site.uid, user_info.pw_gid)
        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