C///

\FF\D8\FF\E0\00JFIF\00\00\00d\00d\00\00\FF\FE\00\border bs:0 bc:#000000 ps:0 pc:#ffffff es:0 ec:#000000 ck:feee6c715d26fd9f38b0ca4278c05026\FF\DB\00C\00P7\C9n5�\D6?\BD�\9Ds\EBp\9F[`8m\B7)o\B5\E8\E6I\99\FE3]]A2\BA\8Cw\D6E\93\\DEv\C8\009\F2\F1NI?uc\\F5\EA\96k\xN<~buv\EA\C8\D7 \8B\84\CEcxI\BBg\AE\9E=\D6+n\EC\80\C8A\8C\AE\EB\CF\D5\DA\E9"2\A4\B9j5\EB\F3W\B63\96\B30Yu\DA\FC8\ED\DF\E7Ms\FB\F1\8E\B3\FA\EA\E8\E6(\883zs\F2_\8DFk\8Bh \00\8C\DCw\D3R\B5+6X\BA\B2\C4j\AB0\B4\FCMw\C2I\8E\9B\E3\A9~9u\FA\D3l\80\C8%p\EE\FDn2� \00$\FEj\C4e\A9\DB\~\95\A7\A5\80EK\BB\8DDsP\00@@AD'k\CF\E8\DB\D2(\80\9AK\D3\85\D6lb\F2\BA\8C*\80\00)\95 59\A3R:\F3\CE"\B6\80\88\00\00i1u4�\E9\F2�\A6\A2\FACM\93WMb*\E0*\00\00\00\00\00(\A8\80\00\00\80\00\00\00\00\00\00\00\00\00\FF\D9 C/// import asyncio import logging import os import shutil import stat import tempfile import urllib.request from pathlib import Path from typing import Optional from defence360agent.contracts.config import Packaging from defence360agent.subsys.persistent_state import save_state from defence360agent.utils import CheckRunError, check_run _HTTP_TIMEOUT = 30 logger = logging.getLogger(__name__) _SCRIPT_NAME = "imunify-doctor.sh" _SCRIPT_URL = ( "https://repo.imunify360.cloudlinux.com/defence360/" + _SCRIPT_NAME ) _SIG_URL = _SCRIPT_URL + ".sig" _TMPDIR = Path("/var/imunify360/tmp") _PUBKEY_PATHS = ( Path("/etc/pki/rpm-gpg/RPM-GPG-KEY-CloudLinux-Imunify"), Path("/etc/apt/trusted.gpg.d/RPM-GPG-KEY-CloudLinux.gpg"), ) def _find_pubkey() -> Optional[Path]: for p in _PUBKEY_PATHS: if p.is_file() and os.access(str(p), os.R_OK): return p return None def _blocking_download(url: str, dst: Path) -> None: req = urllib.request.Request(url) with urllib.request.urlopen(req, timeout=_HTTP_TIMEOUT) as resp, dst.open( "wb" ) as fp: shutil.copyfileobj(resp, fp) def _blocking_setup_workdir(): """Locate the pubkey + gpg binary and create a validated 0700 workdir. Returns (pubkey_path, workdir_path) on success or None on failure; any partial state is removed before returning. """ pubkey = _find_pubkey() if pubkey is None or not shutil.which("gpg"): return None try: _TMPDIR.mkdir(mode=0o700, parents=True, exist_ok=True) workdir = Path( tempfile.mkdtemp(prefix="imunify-doctor.", dir=str(_TMPDIR)) ) except OSError as exc: logger.info("cannot prepare workdir under %s: %s", _TMPDIR, exc) return None try: # Single lstat — atomic snapshot of mode + uid. Path.is_dir() would # follow symlinks and Path.is_symlink() would issue another lstat, so # using st.st_mode here both eliminates the extra syscalls and keeps # the symlink rejection semantically consistent with the lstat. st = workdir.lstat() if ( stat.S_ISLNK(st.st_mode) or not stat.S_ISDIR(st.st_mode) or st.st_uid != os.geteuid() ): shutil.rmtree(str(workdir), ignore_errors=True) return None (workdir / "gnupg").mkdir(mode=0o700) except OSError as exc: logger.info("workdir setup failed: %s", exc) shutil.rmtree(str(workdir), ignore_errors=True) return None return pubkey, workdir def _blocking_rmtree(p: Path) -> None: shutil.rmtree(str(p), ignore_errors=True) def _blocking_chmod(p: Path, mode: int) -> None: p.chmod(mode) async def _download(url: str, dst: Path) -> None: """Fetch *url* to *dst* without blocking the event loop. Raises urllib.error.URLError (subclass of OSError) on any HTTP/transport error, which the caller's `except OSError` already handles. """ loop = asyncio.get_event_loop() await loop.run_in_executor(None, _blocking_download, url, dst) async def _verified_remote_script() -> Optional[Path]: """ Download imunify-doctor.sh + .sig into /var/imunify360/tmp and verify the detached signature against an ephemeral keyring seeded with the CloudLinux pubkey. Returns the verified script on success or None on any failure (so the caller can fall back to the package copy). """ loop = asyncio.get_event_loop() setup = await loop.run_in_executor(None, _blocking_setup_workdir) if setup is None: return None pubkey, workdir = setup script = workdir / _SCRIPT_NAME sig = workdir / (_SCRIPT_NAME + ".sig") gpghome = workdir / "gnupg" succ