ruạṛ
import json import shutil import subprocess import traceback import time import os import tarfile from typing import List, Optional, Union from app.config import bart_jobs_folder from app.utils.log_helper import get_job_id_for_context_logging from app.utils.sudo_wrapper import apply_sudo_wrapper_list, apply_sudo_wrapper_str debug_mode = True class LogPrinter: # ANSI escape codes for colors GREEN = '\033[92m' YELLOW = '\033[33m' ORANGE = '\033[38;5;208m' RED = '\033[91m' RESET = '\033[0m' BLUE = "\033[94m" GREY = '\033[90m' # Decide whether to log or print based on context _use_logger = False from loguru import logger as _logger @classmethod def enable_logging(cls): cls._use_logger = True @classmethod def _output(cls, level: str, message: str, color: str): if cls._use_logger: job_id = get_job_id_for_context_logging() bound_logger = cls._logger.bind(log_type="application", job_id=job_id) bound_logger.opt(depth=2).log(level.upper(), message) # depth=2 means log caller's caller function else: print(f"{color}{message}{cls.RESET}") @classmethod def error(cls, message): cls._output("error", f"ERROR: {message}", cls.RED) @classmethod def warn(cls, message): cls._output("warning", f"WARNING: {message}", cls.ORANGE) @classmethod def note(cls, message): cls._output("info", f"NOTE: {message}", cls.YELLOW) @classmethod def success(cls, message): cls._output("info", f"SUCCESS: {message}", cls.GREEN) @classmethod def info(cls, message): cls._output("info", f"INFO: {message}", cls.BLUE) @classmethod def debug(cls, message): if debug_mode: cls._output("debug", f"DEBUG: {message}", cls.GREY) def measure_time(func): def wrapper(*args, **kwargs): start_time = time.time() result = func(*args, **kwargs) end_time = time.time() elapsed_time = end_time - start_time LogPrinter.info(f"==> {func.__name__} took {elapsed_time:.3f} seconds") return result return wrapper def handle_errors(func): def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except Exception as e: LogPrinter.error(f"{e}") traceback.print_exc() raise # Reraise so FastAPI handles it gracefully return wrapper def _is_within_directory(base: str, target: str) -> bool: base_real = os.path.realpath(base) target_real = os.path.realpath(target) try: common = os.path.commonpath([base_real, target_real]) except ValueError: return False # Different drives / invalid paths return common == base_real def _is_under_jobs_folder(path: str, jobs_folder: str = None) -> bool: """ Check if a path is under the jobs folder, handling symlinks correctly. This function handles the case where jobs_folder might be a symlink. It checks if the path is under either: 1. The jobs_folder path (as given, e.g., /opt/bart/jobs) 2. The realpath of jobs_folder (e.g., /home1/nf_staging/bart_jobs) Args: path: The path to check jobs_folder: The jobs folder path (defaults to bart_jobs_folder from config) Returns: True if path is under the jobs folder, False otherwise """ if jobs_folder is None: jobs_folder = bart_jobs_folder # Get real paths for comparison path_real = os.path.realpath(path) jobs_folder_real = os.path.realpath(jobs_folder) # Check if path is under the real jobs folder path try: common = os.path.commonpath([path_real, jobs_folder_real]) if common == jobs_folder_real: return True except ValueError: pass # Different drives / invalid paths # Also check if the original path (before realpath) starts with jobs_folder # This handles cases where the path is given as the symlink path path_abs = os.path.abspath(path) jobs_folder_abs = os.path.abspath(jobs_folder) if path_abs.startswith(jobs_folder_abs + os.sep) or path_abs == jobs_folder_abs: return True return False def _validate_tar_member(member: tarfile.TarInfo, extract_root: str, username: Optional[str] = None): name = member.name # 1) No absolute paths if name.startswith("/") or name.startswith("\\"): raise ValueError(f"Unsafe tar member (absolute path): {name!r}") # 2) No parent directory traversal parts = [p for p in name.split("/") if p not in ("", ".")] if any(p == ".." for p in parts): raise ValueError(f"Unsafe tar member (path traversal): {name!r}") # 3) Final destination must stay under extract_root dest_path = os.path.join(extract_root, name) if not _is_within_directory(extract_root, dest_path): raise ValueError(f"Unsafe tar member (escapes root): {name!r}") # 4) Disallow device files if member.isdev(): raise ValueError(f"Unsafe tar member (device file): {name!r}") # 5) Symlinks / hardlinks must also stay within extract_root if member.issym() or member.islnk(): link_target = member.linkname or "" if username and link_target == f"/usr/local/apache/domlogs/{username}": return # Absolute link targets are not allowed if os.path.isabs(link_target): raise ValueError(f"Unsafe tar member (absolute link target): {name!r} -> {link_target!r}") link_dest = os.path.join(extract_root, link_target) if not _is_within_directory(extract_root, link_dest): raise ValueError( f"Unsafe tar member (link target escapes root): {name!r} -> {link_target!r}" ) @measure_time def extract_tar(tarfile_path: str, extract_to: str, username: Optional[str] = None) -> str: """ Safely extract a gzipped tar archive into extract_to. Security: - Only extracts into directories under bart_jobs_folder. - Validates all members for path traversal, devices, and unsafe links. - Uses tarfile's 'data' extraction filter when available. - No sudo fallback: if we can't extract, we fail fast. """ if not os.path.exists(tarfile_path): raise FileNotFoundError(f"Tar file {tarfile_path} does not exist.") # Only allow extraction under the jobs folder (same pattern as cleanup_backup_folder) if not _is_under_jobs_folder(extract_to, bart_jobs_folder): msg = f"Refusing to extract outside {bart_jobs_folder}: {extract_to}" LogPrinter.error(msg) raise ValueError(msg) os.makedirs(extract_to, exist_ok=True) members_names: List[str] = [] try: with tarfile.open(tarfile_path, "r:gz") as tar: all_members = tar.getmembers() for m in all_members: _validate_tar_member(m, extract_to, username) members_names.append(m.name) LogPrinter.note(f"Validated {len(all_members)} members in {tarfile_path}") # Use the safer 'data' extraction filter when available (Py 3.11+) try: tar.extractall(path=extract_to, members=all_members, filter="data") except Exception: # Older Python fallback (no filter param) tar.extractall(path=extract_to, members=all_members) LogPrinter.success(f"Extracted tar file {tarfile_path} to {extract_to} using standard tarfile module") except Exception as e: LogPrinter.error(f"Failed to extract tar file {tarfile_path}: {e}") raise # Compute top-level folder name for caller convenience top_level_folders = { member.split('/')[0] for member in members_names if member and not member.endswith('.tar.gz') } LogPrinter.note(f"top_level_folders: {list(top_level_folders)}") if len(top_level_folders) == 1: extracted_root = os.path.join(extract_to, top_level_folders.pop()) else: extracted_root = extract_to # fallback if multiple top-level dirs or flat structure LogPrinter.success(f"Extracted tar file {tarfile_path} to {extract_to}") return extracted_root def is_valid_gzipped_tar(file_path): result = run_command(["tar", "-tzf", file_path]) return result is not None def read_file_contents(filename): try: with open(filename, 'r') as original_file: contents = original_file.read() return contents except Exception as e: LogPrinter.warn(f"Standard file read failed: {e}, attempting sudo fallback") try: result = run_sudo_command(["cat", filename]) return result except subprocess.CalledProcessError as e2: LogPrinter.error(f"Failed to read file with sudo: {e2.stderr.strip()}") raise def read_file_contents_sudo(filename): try: result = run_sudo_command(["cat", filename]) return result except subprocess.CalledProcessError as e2: LogPrinter.error(f"Failed to read file with sudo: {e2.stderr.strip()}") raise def read_json_file(file_path): try: contents = read_file_contents(file_path) return json.loads(contents) except json.JSONDecodeError as je: raise ValueError(f"Failed to parse JSON from file {file_path}: {je}") def cleanup_backup_folder(backup_folder: str): if not _is_under_jobs_folder(backup_folder, bart_jobs_folder): LogPrinter.error(f"Refusing to delete folder outside {bart_jobs_folder}: {backup_folder}") raise ValueError(f"Unsafe cleanup path: {backup_folder}") if os.path.exists(backup_folder): try: shutil.rmtree(backup_folder) LogPrinter.success(f"Cleaned up backup folder at {backup_folder}") except Exception as e: LogPrinter.warn(f"Standard cleanup failed: {e}, attempting sudo fallback") try: run_sudo_command(["rm", "-rf", backup_folder]) LogPrinter.success(f"Cleaned up backup folder with sudo: {backup_folder}") except subprocess.CalledProcessError as e2: LogPrinter.error(f"Sudo cleanup failed for {backup_folder}: {e2}") raise def create_directory(path: str): try: os.makedirs(path, exist_ok=True) except PermissionError: LogPrinter.warn(f"Permission denied while creating {path}, retrying with sudo...") result = run_sudo_command(["mkdir", "-p", path]) if result is None: LogPrinter.error(f"Failed to create directory {path} even with sudo.") raise PermissionError(f"Could not create directory {path}") def create_directory_as_user(run_as_user, path: str): result = run_sudo_as_user(run_as_user, ["mkdir", "-p", path]) if result is None: LogPrinter.error(f"Failed to create directory {path} as user {run_as_user}.") raise PermissionError(f"Could not create directory {path} as user {run_as_user}.") def create_backup_tar(backup_folder: str) -> str: tarfile_path = f"{backup_folder}.tar.gz" with tarfile.open(tarfile_path, "w:gz") as tar: tar.add(backup_folder, arcname=os.path.basename(backup_folder)) LogPrinter.success(f"Created tar file: {tarfile_path}") return tarfile_path def create_backup_tar_cli(backup_folder: str) -> str: tarfile_path = f"{backup_folder}.tar.gz" folder_name = os.path.basename(backup_folder) parent_dir = os.path.dirname(backup_folder) cmd = [ "tar", '--use-compress-program=pigz -p 20', "-cf", tarfile_path, "-C", parent_dir, folder_name ] try: run_sudo_command(cmd) LogPrinter.success(f"Created tar file: {tarfile_path}") except Exception as e: LogPrinter.error(f"Failed to create tar: {e}") raise return tarfile_path def read_homedir(account_username): contents = read_file_contents_sudo('/etc/userdatadomains').splitlines() for line in contents: sub_line = line.split(': ') sub_line_content = sub_line[1].split('==') username = sub_line_content[0] if username == account_username: account_homedir = "/".join(sub_line_content[4].split("/")[:3]) return account_homedir return None def read_file_as_owner(path: str, owner: str) -> Optional[str]: try: file_content = run_sudo_as_user(owner, ["/bin/cat", path]) return file_content except Exception as e: return None # Cap command error text so enormous stdout/stderr does not swamp logs/UI _MAX_CMD_ERR_DETAIL_CHARS = 8192 # TODO : remove subprocess.CalledProcessError handling in other places def _run_command_internal( cmd: Union[str, List[str]], shell_mode: bool, input_text: Optional[str] = None ) -> Optional[str]: try: log_cmd = cmd if isinstance(cmd, str) else ' '.join(cmd) if input_text is not None: LogPrinter.note(f"Running command: {log_cmd} [stdin bytes: {len(input_text.encode('utf-8'))}]") else: LogPrinter.note(f"Running command: {log_cmd}") result = subprocess.run( cmd, shell=shell_mode, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, input=input_text ) LogPrinter.success("Command executed successfully.") return result.stdout except subprocess.CalledProcessError as e: stderr = (e.stderr or "").strip() stdout = (e.stdout or "").strip() parts = [] if stderr: parts.append(stderr) if stdout: parts.append(stdout) detailed = "\n".join(parts) if parts else "Command failed" if len(detailed) > _MAX_CMD_ERR_DETAIL_CHARS: detailed = detailed[:_MAX_CMD_ERR_DETAIL_CHARS] + "\n...[truncated]" if "a password is required" in detailed.lower(): LogPrinter.error("Sudo Access Denied: password is required.") raise Exception("Sudo Access Denied: password is required.") LogPrinter.error( f"Command failed with exit code {e.returncode}. Output:\n{detailed}" ) raise Exception(detailed) def run_sudo_command(cmd: List[str], input_text: Optional[str] = None) -> Optional[str]: wrapped_cmd = apply_sudo_wrapper_list(cmd) return _run_command_internal(["sudo", "-n"] + wrapped_cmd, shell_mode=False, input_text=input_text) def run_sudo_command_str(cmd_str: str, input_text: Optional[str] = None) -> Optional[str]: wrapped_cmd_str = apply_sudo_wrapper_str(cmd_str) return _run_command_internal(f"sudo -n {wrapped_cmd_str}", shell_mode=True, input_text=input_text) def run_command(cmd: List[str], input_text: Optional[str] = None) -> Optional[str]: return _run_command_internal(cmd, shell_mode=False, input_text=input_text) def run_command_str(cmd_str: str, input_text: Optional[str] = None) -> Optional[str]: return _run_command_internal(cmd_str, shell_mode=True, input_text=input_text) def run_sudo_as_user(run_as: str, cmd: List[str], input_text: Optional[str] = None) -> Optional[str]: return _run_command_internal(["sudo", "-n", "-u", run_as] + cmd, shell_mode=False, input_text=input_text) def does_path_exists(path: str): try: run_sudo_command(["test", "-e", path]) return True except Exception: return False def is_directory(path: str) -> bool: try: run_sudo_command(["test", "-d", path]) return True except Exception: return False def get_file_size_in_mb(path: str): if os.path.exists(path): size_bytes = os.path.getsize(path) if size_bytes < 1024*1024: size_mb = round(size_bytes / (1024 * 1024), 4) else: size_mb = round(size_bytes / (1024 * 1024), 2) return size_mb return None def is_cpanel_server() -> bool: return does_path_exists("/usr/local/cpanel")
cải xoăn