Viên ngọc phương Đông

import os
import shlex
from typing import List

BART_WRAPPER_BASE = "/opt/bart/bin"

WRAPPED_SUDO_COMMANDS = {
    "rsync": f"{BART_WRAPPER_BASE}/bart_rsync_wrapper.py",
    "mysqldump": f"{BART_WRAPPER_BASE}/bart_mysqldump_wrapper.py",
    "mysql": f"{BART_WRAPPER_BASE}/bart_mysql_wrapper.py",
    "chown": f"{BART_WRAPPER_BASE}/bart_chown_wrapper.py",
    "chmod": f"{BART_WRAPPER_BASE}/bart_chmod_wrapper.py",
    "rm": f"{BART_WRAPPER_BASE}/bart_rm_wrapper.py",
    "cpanel_rebuild_dbmap": f"{BART_WRAPPER_BASE}/bart_rebuild_dbmap_wrapper.py",
    "cpanel_dbstoregrants": f"{BART_WRAPPER_BASE}/bart_dbstoregrants_wrapper.py",
    "cpanel_uapi": f"{BART_WRAPPER_BASE}/bart_uapi_wrapper.py",
    "restorepkg": f"{BART_WRAPPER_BASE}/bart_restorepkg_wrapper.py",
    "pkgacct": f"{BART_WRAPPER_BASE}/bart_pkgacct_wrapper.py",
    # "lsattr": f"{BART_WRAPPER_BASE}/bart_lsattr_wrapper.py",
    # "find": f"{BART_WRAPPER_BASE}/bart_find_wrapper.py",
    # "xargs": f"{BART_WRAPPER_BASE}/bart_xargs_wrapper.py",
}

def _get_log_printer():
    """
    Lazy import of LogPrinter to avoid circular dependency:
    - utility_functions imports sudo_wrapper
    - sudo_wrapper imports LogPrinter *only when needed*
    """
    from app.utils.utility_functions import LogPrinter
    return LogPrinter

def apply_sudo_wrapper_list(cmd: List[str]) -> List[str]:

log_printer = _get_log_printer()

if not cmd:
        return cmd

if _wrapping_disabled():
        log_printer.debug(f"BART_DISABLE_WRAPPERS=1, not wrapping command: {cmd}")
        return cmd

orig = cmd[0]
    wrapper_cmd = WRAPPED_SUDO_COMMANDS.get(orig)

if wrapper_cmd:
        log_printer.debug(f"Wrapping sudo command {orig!r} -> {wrapper_cmd!r}")
        return [wrapper_cmd] + cmd[1:]

return cmd

def apply_sudo_wrapper_str(cmd_str: str) -> str:
    log_printer = _get_log_printer()

if _wrapping_disabled():
        log_printer.debug(f"BART_DISABLE_WRAPPERS=1, not wrapping string command: {cmd_str}")
        return cmd_str

try:
        parts = shlex.split(cmd_str)
    except ValueError as e:
        log_printer.warn(f"Failed to shlex.parse command string for wrapping: {e}; cmd={cmd_str!r}")
        return cmd_str

if not parts:
        return cmd_str

wrapped_parts = apply_sudo_wrapper_list(parts)

try:
        joined = shlex.join(wrapped_parts)  # Python 3.8+
    except AttributeError:
        joined = " ".join(shlex.quote(p) for p in wrapped_parts)

log_printer.debug(f"Wrapped string sudo command: {cmd_str!r} -> {joined!r}")
    return joined

def _wrapping_disabled() -> bool:
    return os.environ.get("BART_DISABLE_WRAPPERS") == "1"

Register