"""Tell a root RPC peer that came through the DirectAdmin sudo wrapper
apart from root itself (DEF-56083).

On DirectAdmin every account in admin.list may run
``sudo /usr/bin/imunify360-command-wrapper`` (see the sudoers rule in
``subsys/panels/directadmin/panel.py``). The wrapper relays raw JSON to
the root socket and the agent only sees uid 0 on SO_PEERCRED, so such
an account reached the full root command set (DEF-56070). The
classifier below recovers the missing fact -- "this uid-0 peer is the
wrapper, started by sudo for a non-root user" -- from /proc, once per
connection, and the middleware and endpoints consult it through
:data:`caller_delegated_admin_var`.

The check is deliberately conservative: a peer that is not the wrapper is
trusted from its cmdline alone, and a wrapper-shaped peer whose /proc
entries cannot be read is treated as delegated (fail closed).
"""
import logging
import os
from contextvars import ContextVar
from dataclasses import dataclass
from typing import Optional

logger = logging.getLogger(__name__)

# imunify-ui installs the wrapper under the venv and symlinks it into
# /usr/bin; the sudoers rule names the /usr/bin spelling. Both are
# compared by realpath so either one classifies.
WRAPPER_PATHS = (
    "/usr/bin/imunify360-command-wrapper",
    "/opt/imunify360/venv/bin/imunify360-command-wrapper",
)


@dataclass(frozen=True)
class DelegatedAdmin:
    pid: int
    sudo_uid: Optional[str]
    sudo_user: Optional[str]
    reason: str


# None for root proper (CLI, WHM, Plesk sbin, internal callers) and for
# the direct in-process CLI path; set per dispatched request by
# _RpcServerProtocol.data_received next to caller_uid_var.
caller_delegated_admin_var: ContextVar[Optional[DelegatedAdmin]] = ContextVar(
    "rpc_caller_delegated_admin", default=None
)


def current_delegated_admin() -> Optional[DelegatedAdmin]:
    return caller_delegated_admin_var.get()


def _read(proc, pid, name) -> bytes:
    with open(os.path.join(proc, str(pid), name), "rb") as f:
        return f.read()


def _is_wrapper(script: str) -> bool:
    targets = set(WRAPPER_PATHS)
    targets.update(os.path.realpath(p) for p in WRAPPER_PATHS)
    return script in targets or os.path.realpath(script) in targets


def _environ(proc, pid) -> dict:
    env = {}
    for item in _read(proc, pid, "environ").split(b"\0"):
        if item:
            key, _, value = item.partition(b"=")
            env[key.decode(errors="replace")] = value.decode(errors="replace")
    return env


def _parent_comm(proc, pid) -> str:
    stat_line = _read(proc, pid, "stat").decode(errors="replace")
    # "pid (comm) state ppid ...": comm may itself contain spaces and
    # parentheses, so the fixed fields start after the LAST ')'.
    fields = stat_line.rsplit(")", 1)[1].split()
    ppid = int(fields[1])
    return _read(proc, ppid, "comm").decode(errors="replace").strip()


def classify_peer(pid: int, proc: str = "/proc") -> Optional[DelegatedAdmin]:
    """Classify the uid-0 peer ``pid`` of a root-socket connection.

    Returns a :class:`DelegatedAdmin` when ``pid`` is the command wrapper
    and either its environment carries ``SUDO_UID`` of a non-root user
    (sudo always sets it for the command it runs) or its parent is
    ``sudo`` (sudo normally forks the command), and ``None`` for every
    other root peer. Anything unreadable about a wrapper-shaped or
    unidentifiable peer classifies as delegated.
    """
    try:
        argv = [
            a.decode(errors="replace")
            for a in _read(proc, pid, "cmdline").split(b"\0")
            if a
        ]
    except OSError as exc:
        return DelegatedAdmin(
            pid, None, None, "cannot read cmdline: {}".format(exc)
        )
    if not argv:
        return DelegatedAdmin(pid, None, None, "empty cmdline")
    # The wrapper is a script: argv[0] is the venv interpreter from its
    # shebang and argv[1] the path sudo (or WHM/Plesk) exec'd.
    if len(argv) < 2 or not _is_wrapper(argv[1]):
        return None

    try:
        env = _environ(proc, pid)
    except OSError as exc:
        return DelegatedAdmin(
            pid, None, None, "wrapper with unreadable environ: {}".format(exc)
        )
    sudo_uid = env.get("SUDO_UID")
    sudo_user = env.get("SUDO_USER")
    if sudo_uid is not None and sudo_uid != "0":
        return DelegatedAdmin(
            pid, sudo_uid, sudo_user, "wrapper with SUDO_UID=" + sudo_uid
        )

    try:
        parent = _parent_comm(proc, pid)
    except (OSError, ValueError, IndexError) as exc:
        return DelegatedAdmin(
            pid,
            sudo_uid,
            sudo_user,
            "wrapper with unreadable parent: {}".format(exc),
        )
    if parent == "sudo":
        return DelegatedAdmin(
            pid, sudo_uid, sudo_user, "wrapper forked by sudo"
        )
    return None