#!/opt/cloudlinux/venv/bin/python3 -sbb
# -*- coding: utf-8 -*-
#
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2026 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENCE.TXT
#
"""Trust-boundary validation for panel-supplied document root strings.

The docroot value originates from the hosting panel (cPanel /
DirectAdmin / Plesk) and is consumed by privileged code that writes
jail.c mount configuration files read by root. The jail.c mount syntax
is whitespace-delimited (MountEntry.render in mount_types.py joins
source/target/options with spaces) and section headers are bracketed
(`[<docroot>]` in jail_config.MountConfig.render), so any whitespace,
control character, newline, or bracket inside the docroot corrupts the
parser. A sibling module already rejects newlines/carriage-returns on
the analogous crontab write path (crontab/libhooks.py and
crontab/parser.py); this module is the equivalent guard for the jail
mount config write path.

Validation is centralized at the trust boundary - call sites are
`enable_website_isolation` (where a tenant-owned domain is resolved to
a docroot for the first time) and `write_jail_mounts_config` (where
the docroot map is re-read from the panel for every regeneration). The
helper raises ValueError on rejection, matching the sibling crontab
pattern.
"""

from __future__ import annotations

import errno
import os
import re
import stat

# Strict allowlist: alnum, `_`, `-`, `.`, `/`. This excludes whitespace
# (which would split mount-line tokens), brackets (which would close the
# jail section header), quotes, and any control character. Production
# docroots in shared-hosting deployments are conventionally of the form
# `/home/<user>/<subdir>` and well inside this allowlist; anything
# outside it is either a panel misconfiguration or an injection attempt.
# `\Z` (not `$`) anchors the match at the true end of the string.
# Python's default `$` also matches immediately before a terminating
# `\n`, so `^/[A-Za-z0-9_./-]*$` accepts a trailing newline even though
# `\n` is not in the character class - a docroot like
# `/home/u/public_html\n` would slip past this filter and later split
# the `[<docroot>]` section header written by MountConfig.render. `\Z`
# closes that gap without loosening the character class.
_DOCROOT_ALLOWED_RE = re.compile(r"^/[A-Za-z0-9_./-]*\Z")

# Hard cap to bound the size of strings that flow into mount lines and
# section headers. A 4 KiB ceiling is well above any realistic docroot
# (Linux PATH_MAX is 4096) and well below pathological mount-line
# sizes.
_DOCROOT_MAX_LEN = 4096


def validate_docroot(docroot: str) -> str:
    """Validate a panel-supplied document root before it reaches the
    jail mount config writer.

    Args:
        docroot: Document root string returned by the panel (e.g. from
            ``clcommon.cpapi.docroot`` or ``clcommon.cpapi.userdomains``).

    Returns:
        The validated docroot string, unchanged.

    Raises:
        ValueError: If the docroot is empty, not an absolute path, too
            long, contains a path-traversal segment, or contains any
            character outside the strict allowlist (alnum, `_`, `-`,
            `.`, `/`).
    """
    if not isinstance(docroot, str):
        raise ValueError(f"Invalid docroot (not a string): {docroot!r}")
    if not docroot:
        raise ValueError("Invalid docroot: empty string")
    if len(docroot) > _DOCROOT_MAX_LEN:
        raise ValueError(
            f"Invalid docroot (length {len(docroot)} exceeds {_DOCROOT_MAX_LEN}): {docroot!r}"
        )
    if not docroot.startswith("/"):
        raise ValueError(f"Invalid docroot (not absolute): {docroot!r}")
    if not _DOCROOT_ALLOWED_RE.match(docroot):
        raise ValueError(f"Invalid docroot (disallowed characters): {docroot!r}")
    # Reject empty path segments (``//``). The character-class allowlist
    # permits ``/`` freely, so ``/home/user//evil`` passes the regex. But
    # empty segments break the O_NOFOLLOW tail walk in
    # ``_reject_symlinks_in_tail``: ``tail.split('/')`` yields an empty-
    # string component, and ``os.open('', dir_fd=...)`` returns ENOENT,
    # which the walk treats as a safe short-circuit — the leaf symlink
    # is then never O_NOFOLLOWed. Reject at the entry so downstream code
    # never sees a docroot with empty segments.
    if "//" in docroot:
        raise ValueError(f"Invalid docroot (empty path segment): {docroot!r}")
    # Reject traversal segments. The allowlist permits `.` and `..` as
    # path components even though it forbids most metacharacters, so
    # screen explicitly: `..` lets a tenant's owned-domain docroot point
    # at a sibling tenant's tree once it is bind-mounted as the jail
    # source.
    for segment in docroot.split("/"):
        if segment == "..":
            raise ValueError(f"Invalid docroot (parent traversal): {docroot!r}")
    return docroot


def validate_docroot_no_symlinks(docroot: str, allowed_prefix: str) -> str:
    """Reject a docroot whose on-disk path contains an attacker-plantable
    symlink under ``allowed_prefix``.

    The lexical ``validate_docroot`` above pins the *string shape* of a
    panel-supplied docroot; this function is the on-disk companion. From
    the resolved ``allowed_prefix`` downward, each remaining component is
    walked with ``O_NOFOLLOW`` under a ``dir_fd``, so any symlink planted
    in the tenant-writable tree is rejected up front — before the value
    ever reaches ``isolates.mounts``.

    A missing component (``ENOENT``) short-circuits the walk as safe. The
    authoritative no-follow / missing-source guarantee lives in the C
    mount consumer (``lve-kmod/usrc/src/jail.c``), which opens every
    source through ``openat2(RESOLVE_NO_SYMLINKS)`` with an
    ``O_PATH|O_NOFOLLOW`` fallback, binds by fd (``mount_fd`` in
    ``safe_move_bind_mount``, not by pathname), and skips mount entries
    whose source stat's as ``ENOENT``. A tenant symlink planted between
    this walk and the root-run bind is refused at C-side open time
    regardless of what the Python walk saw, so pre-rejecting a still-
    missing leaf here buys nothing and breaks legitimate lifecycles
    (overlay / ``.clwpos`` bind sources that the C consumer materialises
    via its ``mkdir`` mount opt; a tenant home whose leaf was renamed
    between the panel read and this walk). See SECURITY-EXCEPTIONS.md,
    entry ``docroot_validation.py — bind-source symlink guarantee lives
    at the C sink``.

    After a successful walk, the terminal fd's inode/dev is compared to
    a fresh lstat of the docroot string; a mismatch (attacker swapped
    the leaf between the last openat and this stat) is rejected. This
    is early-fail hygiene layered on top of the C-sink guarantee, not
    the primary defence.

    Args:
        docroot: Document root string returned by the panel. Must
            already have passed the lexical ``validate_docroot``
            check (so it is absolute, ``..``-free, and within the
            character allowlist) - this function does *not* re-run
            those checks.
        allowed_prefix: Absolute path the docroot must be inside.
            Typically the user's home directory; passing the
            unresolved value is fine because the function
            canonicalises it once.

    Returns:
        The docroot string, unchanged.

    Raises:
        ValueError: If the docroot is not under ``allowed_prefix``,
            if any component under the resolved prefix is a symlink
            or a non-directory, if the resolved prefix cannot be
            opened, or if the leaf inode observed at return time
            differs from the one the walk validated (post-walk swap).
    """
    try:
        resolved_prefix = os.path.realpath(allowed_prefix, strict=False)
    except OSError as exc:
        raise ValueError(
            f"Invalid docroot (cannot resolve prefix "
            f"{allowed_prefix!r}: {exc})"
        ) from exc

    # Docroot may equal the home itself (some panels emit that), or
    # start with either the raw allowed_prefix (operator-symlink form
    # ``/home/u/...``) or the resolved form (``/home2/u/...``). Anything
    # else is outside the tenant's home tree lexically - reject.
    if docroot == resolved_prefix or docroot == allowed_prefix:
        return docroot
    for prefix in (resolved_prefix, allowed_prefix):
        marker = prefix.rstrip("/") + "/"
        if docroot.startswith(marker):
            relative = docroot[len(marker):]
            break
    else:
        raise ValueError(
            f"Invalid docroot (resolves outside allowed prefix "
            f"{resolved_prefix!r}): {docroot!r}"
        )

    components = [c for c in relative.split("/") if c]
    if not components:
        return docroot

    # Open the resolved prefix as an O_DIRECTORY anchor for the walk.
    # No O_NOFOLLOW here: the operator layer above the tenant tree
    # may legitimately be a symlink and we have already canonicalised
    # through it via realpath().
    try:
        prefix_fd = os.open(
            resolved_prefix,
            os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC,
        )
    except FileNotFoundError:
        # Prefix (tenant home) itself does not exist. Safe short-circuit
        # — jail.c refuses symlinks on bind and skips ENOENT sources on
        # its own, matching the strict=False realpath contract that a
        # not-yet-materialised tree is not an error.
        return docroot
    except OSError as exc:
        raise ValueError(
            f"Invalid docroot (cannot open resolved prefix "
            f"{resolved_prefix!r}: {exc}): {docroot!r}"
        ) from exc
    try:
        leaf_fd = _walk_no_symlinks(
            prefix_fd, components, docroot, resolved_prefix,
        )
        # F-11 (CLOS-5949) DiD: compare the terminal fd's inode/dev to
        # a fresh lstat of the docroot string. If they differ, the leaf
        # was replaced between the O_NOFOLLOW walk and this check
        # (attacker won the race between openat and return). Reject.
        # A ``None`` leaf_fd means the walk short-circuited on an
        # absent component; no post-walk verify is possible there and
        # none is needed — the C-side mount consumer skips ENOENT
        # sources and materialises ``mkdir``-opt leaves at bind time.
        if leaf_fd is not None:
            try:
                walked_st = os.fstat(leaf_fd)
                try:
                    current_st = os.stat(docroot, follow_symlinks=False)
                except FileNotFoundError:
                    raise ValueError(
                        f"Invalid docroot (leaf disappeared after walk): {docroot!r}"
                    )
                if (walked_st.st_ino != current_st.st_ino
                        or walked_st.st_dev != current_st.st_dev):
                    raise ValueError(
                        f"Invalid docroot (leaf inode changed after walk): {docroot!r}"
                    )
            finally:
                os.close(leaf_fd)
    finally:
        os.close(prefix_fd)
    return docroot


def _walk_no_symlinks(
    base_fd: int,
    components: list[str],
    docroot: str,
    resolved_prefix: str,
) -> int | None:
    """Walk ``components`` under ``base_fd`` rejecting any symlink or
    non-directory. Uses an lstat-then-openat-with-O_NOFOLLOW sequence:
    the lstat classifies the inode without following, and the follow-up
    open (with O_NOFOLLOW as belt-and-suspenders) captures a fd bound
    to the classified directory before descending. ``O_DIRECTORY`` and
    ``O_NOFOLLOW`` together return ``ENOTDIR`` (not ``ELOOP``) on a
    trailing symlink, which is why the type check runs before the open.

    A missing component (``ENOENT``) short-circuits the walk as safe.
    The C mount consumer in ``lve-kmod/usrc/src/jail.c`` opens every
    source with ``openat2(RESOLVE_NO_SYMLINKS)`` (``O_PATH|O_NOFOLLOW``
    fallback), binds by fd (``safe_move_bind_mount`` / ``mount_fd``,
    never by pathname), and skips mount entries whose source stat's as
    ``ENOENT`` — so a tenant symlink materialised into a still-missing
    component between this walk and the root-run bind is refused at C
    open time regardless of what we saw here.

    F-11 (CLOS-5949) DiD — fd lifecycle:
      - ``base_fd`` is the caller's; NEVER closed here.
      - Intermediate opens are appended to ``opened`` as they succeed.
      - On a successful full walk: every opened fd except the terminal
        one is closed here, the terminal is handed off (owned by
        caller — the caller must close it).
      - On the ENOENT short-circuit or any incomplete walk: every
        opened fd is closed via the ``finally`` and ``None`` is
        returned — no fd leaks past the return.
      - On any exception: same ``finally`` closes every opened fd
        before the exception propagates.
    """
    current_fd = base_fd
    opened: list[int] = []
    walk_complete = False
    try:
        for comp in components:
            try:
                st = os.stat(comp, dir_fd=current_fd, follow_symlinks=False)
            except OSError as exc:
                if exc.errno == errno.ENOENT:
                    # Safe short-circuit: jail.c refuses to follow
                    # symlinks on bind (openat2 RESOLVE_NO_SYMLINKS +
                    # mount_fd by fd) and skips ENOENT sources on its
                    # own, so a component we can't stat here cannot be
                    # weaponised at bind time.
                    return None
                raise ValueError(
                    f"Invalid docroot (cannot walk component {comp!r}: "
                    f"{exc}): {docroot!r}"
                ) from exc
            if stat.S_ISLNK(st.st_mode):
                raise ValueError(
                    f"Invalid docroot (symlink at component {comp!r} "
                    f"resolves outside allowed prefix "
                    f"{resolved_prefix!r}): {docroot!r}"
                )
            if not stat.S_ISDIR(st.st_mode):
                raise ValueError(
                    f"Invalid docroot (non-directory at component "
                    f"{comp!r}): {docroot!r}"
                )
            try:
                fd = os.open(
                    comp,
                    os.O_RDONLY | os.O_NOFOLLOW | os.O_DIRECTORY | os.O_CLOEXEC,
                    dir_fd=current_fd,
                )
            except OSError as exc:
                # Race: comp was swapped to a symlink between lstat and
                # open. O_NOFOLLOW surfaces this as ELOOP; treat it as
                # the symlink-rejection path above.
                if exc.errno == errno.ELOOP:
                    raise ValueError(
                        f"Invalid docroot (symlink at component {comp!r} "
                        f"resolves outside allowed prefix "
                        f"{resolved_prefix!r}): {docroot!r}"
                    ) from exc
                raise ValueError(
                    f"Invalid docroot (cannot walk component {comp!r}: "
                    f"{exc}): {docroot!r}"
                ) from exc
            opened.append(fd)
            current_fd = fd
        walk_complete = True
    finally:
        # On successful full walk, keep the terminal fd open and hand
        # it to the caller — close only the intermediates. On any
        # incomplete walk (ENOENT return, exception), close every
        # fd we opened. ``base_fd`` is never in ``opened`` and is
        # never closed here.
        if walk_complete and opened:
            keep = opened[-1]
            close_list = opened[:-1]
        else:
            keep = None
            close_list = list(opened)
        for fd in close_list:
            try:
                os.close(fd)
            except OSError:
                # Best-effort cleanup; do not mask the original
                # in-flight exception (if any) with a close error.
                pass
        # (When walk_complete, `keep` is deliberately left open
        # for the caller.)
    if not opened:
        return None
    return opened[-1]